v1
V1 Endpoints
Overview
This API provides endpoints for securely storing, retrieving, and deleting Snyk credentials using OpenBao (Vault) and Redis as backends.
All endpoints are versioned under /credentials and /cache.
Authentication
This API does not implement authentication by default.
If you deploy in production, you should secure the API (e.g., with a reverse proxy, network policy, or FastAPI dependencies).
Error Handling
- All endpoints return JSON responses.
- On error, the response will include an
error field and an appropriate HTTP status code (e.g., 400, 404, 500, 503).
Example error response:
{
"error": "Vault is sealed, cannot store credentials."
}
Endpoint Overview
delete_cache_key
Deletes the Snyk auth token for the specified org/client from Redis.
Parameters:
| Name | Type | Description | Default |
org_id | str | | required |
client_id | str | | required |
Returns:
| Name | Type | Description |
JSONResponse | JSONResponse | A confirmation message indicating the auth token was deleted. |
Source code in snykey/api/v1/endpoints.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176 | @router.delete("/cache")
def delete_cache_key(org_id: str, client_id: str) -> JSONResponse:
"""
Deletes the Snyk auth token for the specified org/client from Redis.
Args:
org_id (str): The organization ID.
client_id (str): The client ID.
Returns:
JSONResponse: A confirmation message indicating the auth token was deleted.
"""
response: dict = redis.delete_auth_token(org_id, client_id)
return JSONResponse(content=response)
|
delete_credentials
Delete Snyk credentials for a given org_id and client_id.
Parameters:
| Name | Type | Description | Default |
org_id | str | | required |
client_id | str | | required |
Returns:
| Name | Type | Description |
JSONResponse | JSONResponse | A response indicating success or failure of the deletion. |
Source code in snykey/api/v1/endpoints.py
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158 | @router.delete("/credentials")
def delete_credentials(org_id: str, client_id: str) -> JSONResponse:
"""
Delete Snyk credentials for a given org_id and client_id.
Args:
org_id (str): The organization ID.
client_id (str): The client ID.
Returns:
JSONResponse: A response indicating success or failure of the deletion.
"""
if not org_id or not client_id:
return JSONResponse(
status_code=400, content={"error": "org_id and client_id are required"}
)
# Delete auth token from Redis
logger.info(
"Deleting auth token from Redis for org_id: %s, client_id: %s",
org_id,
client_id,
)
redis.delete_auth_token(org_id, client_id)
openbao.delete_refresh_key(org_id, client_id)
return JSONResponse(content={"message": "Credentials deleted."})
|
get_credentials
Gather Snyk credentials using the provided org_id and client_id.
Parameters:
| Name | Type | Description | Default |
org_id | str | | required |
client_id | str | | required |
Returns:
| Name | Type | Description |
JSONResponse | JSONResponse | A response containing the gathered credentials or an error message. |
Source code in snykey/api/v1/endpoints.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126 | @router.get("/credentials")
def get_credentials(org_id: str, client_id: str, client_secret: str) -> JSONResponse:
"""
Gather Snyk credentials using the provided org_id and client_id.
Args:
org_id (str): The organization ID.
client_id (str): The client ID.
Returns:
JSONResponse: A response containing the gathered credentials or an error message.
"""
# Check if auth token exists in Redis
logger.info(
"Checking Redis for auth token for org_id: %s, client_id: %s",
org_id,
client_id,
)
auth_token: bytes | None = None
try:
auth_token = redis.get_auth_token(org_id, client_id)
except Exception as e:
logger.error("Failed to retrieve auth token from Redis: %s", e)
if auth_token:
logger.info("Found auth token in Redis")
return JSONResponse(content={"access_token": str(auth_token.decode())})
# Get refresh key from OpenBao
logger.info(
"Gathering Snyk credentials for org_id: %s, client_id: %s",
org_id,
client_id,
)
refresh_key: str | None = openbao.get_refresh_key(org_id, client_id)
if not refresh_key:
return JSONResponse(
status_code=404, content={"error": "No refresh key found for org/client"}
)
# Refresh Snyk token
logger.info(
"Refreshing Snyk token for org_id: %s, client_id: %s", org_id, client_id
)
try:
result: dict = snyk.refresh_snyk_token(client_id, client_secret, refresh_key)
logger.info(
"Successfully refreshed Snyk token for org_id: %s, client_id: %s",
org_id,
client_id,
)
openbao.update_refresh_key(org_id, client_id, result["refresh_token"])
except Exception as e:
logger.error("Failed to refresh Snyk token: %s", e)
return JSONResponse(status_code=500, content={"error": str(e)})
try:
redis.store_auth_token(
org_id,
client_id,
str(result["access_token"]),
expiration=settings.REDIS_CACHE_TIME * 60,
)
except Exception as e:
logger.error("Failed to store auth token in Redis: %s", e)
return JSONResponse(status_code=500, content={"error": str(e)})
return JSONResponse(content={"access_token": str(result["access_token"])})
|
store_credentials
Store Snyk credentials in OpenBao.
Parameters:
| Name | Type | Description | Default |
org_id | str | | required |
client_id | str | | required |
client_secret | str | | required |
refresh_key | str | | required |
Returns:
| Name | Type | Description |
JSONResponse | JSONResponse | A response indicating success or failure. |
Source code in snykey/api/v1/endpoints.py
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50 | @router.put("/credentials")
def store_credentials(
org_id: str, client_id: str, client_secret: str, refresh_key: str
) -> JSONResponse:
"""
Store Snyk credentials in OpenBao.
Args:
org_id (str): The organization ID.
client_id (str): The client ID.
client_secret (str): The client secret.
refresh_key (str): The refresh key.
Returns:
JSONResponse: A response indicating success or failure.
"""
if openbao.check_vault_sealed():
return JSONResponse(
status_code=503,
content={"error": "Vault is sealed, cannot store credentials."},
)
logger.info("Refreshing key to ensure no other process can use it.")
try:
result: dict = snyk.refresh_snyk_token(client_id, client_secret, refresh_key)
logger.info(
"Successfully refreshed Snyk token for org_id: %s, client_id: %s",
org_id,
client_id,
)
openbao.store_refresh_key(org_id, client_id, result["refresh_token"])
except Exception as e:
logger.error("Failed to refresh Snyk token: %s", e)
return JSONResponse(status_code=500, content={"error": str(e)})
return JSONResponse(content={"message": "Credentials stored."})
|