diff --git a/runpod/api/mutations/apps.py b/runpod/api/mutations/apps.py index 2e3c73a44..1818cce6a 100644 --- a/runpod/api/mutations/apps.py +++ b/runpod/api/mutations/apps.py @@ -42,15 +42,15 @@ } """ -MUTATION_CREATE_SECRET = """ -mutation secretCreate($input: SecretCreateInput!) { - secretCreate(input: $input) { id name } +MUTATION_CREATE_API_KEY = """ +mutation createApiKeyNew($input: CreateApiKeyInput!) { + createApiKeyNew(input: $input) { id rawKey } } """ -MUTATION_DELETE_SECRET = """ -mutation secretDelete($id: ID!) { - secretDelete(id: $id) +MUTATION_DELETE_API_KEY = """ +mutation deleteApiKey($input: DeleteApiKeyInput!) { + deleteApiKey(input: $input) { count } } """ diff --git a/runpod/api/queries/apps.py b/runpod/api/queries/apps.py index 9da1b2cde..431cb608d 100644 --- a/runpod/api/queries/apps.py +++ b/runpod/api/queries/apps.py @@ -25,10 +25,10 @@ } """ -QUERY_SECRETS = """ -query mySecrets { +QUERY_API_KEYS = """ +query myApiKeys { myself { - secrets { id name description createdAt } + apiKeys { id name } } } """ diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 9b557800a..b0dc898cd 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -478,31 +478,60 @@ async def delete_registry_auth(self, auth_id: str) -> bool: ) return True - async def list_secrets(self) -> List[Dict[str, Any]]: - query = app_queries.QUERY_SECRETS - data = await self._execute(query, retry=True) - return data["myself"].get("secrets") or [] + async def list_secrets(self, name: Optional[str] = None) -> List[Dict[str, Any]]: + """account secrets, or only the one matching name (case-insensitive).""" + import asyncio + + attempt = 0 + while True: + try: + data = await run_rest_request_async( + "GET", + "/v2/account/secrets", + api_key=self._api_key, + params={"name": name} if name else None, + ) + return data["secrets"] + except (aiohttp.ClientError, OSError, asyncio.TimeoutError): + if attempt == _TRANSPORT_RETRIES - 1: + raise + await asyncio.sleep(2**attempt) + attempt += 1 async def create_secret( self, name: str, value: str, description: str = "" ) -> Dict[str, Any]: - mutation = app_mutations.MUTATION_CREATE_SECRET - data = await self._execute( - mutation, - { - "input": { - "name": name, - "value": value, - "description": description, - } - }, + body = {"name": name, "value": value} + if description: + body["description"] = description + return await run_rest_request_async( + "POST", "/v2/account/secrets", api_key=self._api_key, json=body ) - return data["secretCreate"] async def delete_secret(self, secret_id: str) -> bool: - mutation = app_mutations.MUTATION_DELETE_SECRET - data = await self._execute(mutation, {"id": secret_id}) - return bool(data.get("secretDelete")) + await run_rest_request_async( + "DELETE", + f"/v2/account/secrets/{_path_segment(secret_id)}", + api_key=self._api_key, + ) + return True + + async def list_api_keys(self) -> List[Dict[str, Any]]: + query = app_queries.QUERY_API_KEYS + data = await self._execute(query, retry=True) + return data["myself"].get("apiKeys") or [] + + async def create_api_key(self, name: str, scopes: List[str]) -> Dict[str, Any]: + """create a scoped api key; the result carries rawKey, shown once.""" + mutation = app_mutations.MUTATION_CREATE_API_KEY + data = await self._execute( + mutation, {"input": {"name": name, "scopes": scopes}} + ) + return data["createApiKeyNew"] + + async def delete_api_key(self, key_id: str) -> None: + mutation = app_mutations.MUTATION_DELETE_API_KEY + await self._execute(mutation, {"input": {"id": key_id}}) async def list_apps(self) -> List[Dict[str, Any]]: """all flash apps with their environments and builds.""" diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py new file mode 100644 index 000000000..29b8512d0 --- /dev/null +++ b/runpod/apps/app_key.py @@ -0,0 +1,180 @@ +"""per-app api keys for workers. + +every resource gets RUNPOD_API_KEY from a platform secret holding an api +key minted for its app and environment, unless its env sets one. the key +value travels only from key creation into the secret; templates carry the +secret reference and the platform decrypts it when a worker boots. + +one key serves an app environment across deploys. dev sessions share a +separate key per app. +""" + +import hashlib +import json +import logging +import re +from typing import Any, Dict, List, Optional + +from ..error import AuthenticationError, QueryError +from .secret import Secret + +APP_KEY_ENV = "RUNPOD_API_KEY" +log = logging.getLogger(__name__) + +# what flash runtime does from a worker: call endpoints and list them +# (serverless:read), launch and terminate task pods (pods), resolve +# task mounts (volumes), and resolve registry credentials and secret +# references for task pods (secrets:read) +APP_KEY_SCOPES = [ + "pods:read", + "pods:write", + "secrets:read", + "serverless:read", + "volumes:read", + "volumes:write", +] + +# api key names are limited to 64 characters +_MAX_NAME = 64 +_UNSAFE = re.compile(r"[^A-Za-z0-9_.-]") + + +def app_key_name(app_name: str, env_name: Optional[str] = None) -> str: + """secret and api key name for an app environment, or for the app's + dev sessions when env_name is None.""" + if env_name is None: + raw = f"flash-dev-key.{app_name}" + else: + raw = f"flash-app-key.{app_name}.{env_name}" + name = _UNSAFE.sub("-", raw).lower() + identity = json.dumps([app_name, env_name], ensure_ascii=True) + digest = hashlib.sha256(identity.encode()).hexdigest()[:24] + return f"{name[: _MAX_NAME - len(digest) - 1]}.{digest}" + + +def needs_app_key(spec) -> bool: + """true when a resource does not supply its own credentials through env.""" + return APP_KEY_ENV not in (spec.env or {}) + + +async def _find_secret(api, name: str) -> Optional[Dict[str, Any]]: + for secret in await api.list_secrets(name=name): + if secret.get("name") == name: + return secret + return None + + +async def _key_ids(api, name: str) -> List[str]: + return [k["id"] for k in await api.list_api_keys() if k.get("name") == name] + + +async def _secret_has_live_key( + api, name: str, secret: Dict[str, Any], exclude_key_id: Optional[str] = None +) -> bool: + key_ids = [ + key_id for key_id in await _key_ids(api, name) if key_id != exclude_key_id + ] + owner = re.match(r"api key (\S+) for ", secret.get("description") or "") + return owner[1] in key_ids if owner else bool(key_ids) + + +async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> Secret: + """the secret holding the app environment's api key, minting the key + and the secret on first use. refresh preserves other same-name keys for + concurrent deployments; teardown revokes every matching key.""" + name = app_key_name(app_name, env_name) + secret = await _find_secret(api, name) + if secret: + if await _secret_has_live_key(api, name, secret): + return Secret(name) + try: + await api.delete_secret(secret["id"]) + except QueryError as exc: + if exc.status_code != 404: + raise + try: + key = await api.create_api_key(name, APP_KEY_SCOPES) + except (AuthenticationError, QueryError) as exc: + if not ( + isinstance(exc, AuthenticationError) + or exc.status_code in (401, 403) + or ( + exc.status_code in (None, 400) + and any( + reason in str(exc).casefold() + for reason in ( + "not grantable", + "permission", + "unauthorized", + "forbidden", + ) + ) + ) + ): + raise + exc.message = ( + f"{exc}. Automatic worker credentials require API key creation " + "permission and permission to grant: " + + ", ".join(APP_KEY_SCOPES) + + ". Use a caller with those permissions, or set RUNPOD_API_KEY " + "in each resource's env to an appropriate scoped key or Secret." + ) + raise + target = f"{app_name}/{env_name}" if env_name else f"{app_name} dev sessions" + description = f"api key {key['id']} for {target} workers" + try: + await api.create_secret(name, key["rawKey"], description=description) + except BaseException as create_error: + try: + secret = await _find_secret(api, name) + except Exception: + log.warning( + "could not reconcile secret %s; retained associated api key for cleanup", + name, + ) + raise create_error from None + owns_secret = secret and secret.get("description") == description + if ( + secret + and not secret.get("description") + and not ( + isinstance(create_error, QueryError) and create_error.status_code == 409 + ) + ): + log.warning( + "secret %s has no ownership metadata; retained associated api key for cleanup", + name, + ) + raise + if not owns_secret: + try: + await api.delete_api_key(key["id"]) + except Exception: + log.warning( + "could not revoke api key for secret %s; retry credential cleanup", + name, + ) + if secret and isinstance(create_error, Exception): + try: + if await _secret_has_live_key( + api, name, secret, None if owns_secret else key["id"] + ): + return Secret(name) + except Exception: + log.warning("could not validate the api key for secret %s", name) + raise create_error from None + raise + return Secret(name) + + +async def delete_app_key(api, app_name: str, env_name: Optional[str] = None) -> bool: + """remove the app environment's secret and revoke its keys. returns + whether anything existed.""" + name = app_key_name(app_name, env_name) + secret = await _find_secret(api, name) + if secret is not None: + await api.delete_secret(secret["id"]) + key_ids = await _key_ids(api, name) + for key_id in key_ids: + await api.delete_api_key(key_id) + return secret is not None or bool(key_ids) diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index a0ef47ca7..61b01063e 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -11,10 +11,9 @@ from pathspec import GitIgnoreSpec -import runpod - from .api import AppsApiClient from .app import App +from .app_key import APP_KEY_ENV, ensure_app_key, needs_app_key from .build import ( DEFAULT_PYTHON_VERSION, MAX_ARTIFACT_MB, @@ -24,6 +23,7 @@ ) from .errors import InvalidResourceError, ScheduleNotSupported from .schedule import SCHEDULES_ENABLED +from .secret import Secret, render_env from .shim import runtime_launcher from .spec import ResourceKind from .utils.events import emit @@ -285,6 +285,7 @@ def _deployed_endpoint_input( python_version: str, *, env_name: Optional[str] = None, + app_key: Optional[Secret] = None, ) -> Dict[str, Any]: """saveEndpoint payload for one deployed queue/api resource. @@ -292,12 +293,13 @@ def _deployed_endpoint_input( resolution matches X-Flash-Endpoint against endpoint.name within the environment. binding via flashEnvironmentId is also what makes hosts deliver the build artifact to this endpoint's workers. + + app_key is referenced as RUNPOD_API_KEY unless the resource's env + sets its own. """ from .datacenter import CPU3_DATACENTERS, CPU5_DATACENTERS, DataCenter from .images import image_for_spec - from .secret import render_env - spec.validate() template_env = { "FLASH_RESOURCE_NAME": spec.name, @@ -307,12 +309,8 @@ def _deployed_endpoint_input( "FLASH_ENVIRONMENT": env_name or app.env, } - # cross-resource calls from inside a worker go through the sentinel, - # which authenticates with the account api key; without it any - # .remote() from worker code would fail - api_key = runpod.api_key - if api_key and "RUNPOD_API_KEY" not in template_env: - template_env["RUNPOD_API_KEY"] = api_key + if app_key is not None and needs_app_key(spec): + template_env[APP_KEY_ENV] = app_key.reference # nested tasks use the same runtime and sdk channels as the endpoint for key in ( @@ -410,6 +408,9 @@ async def reconcile_endpoints( from .volume import VolumeResolver, attach_endpoint_volumes resolver = VolumeResolver(client) + app_key = None + if any(needs_app_key(h.spec) for h in provisionable.values()): + app_key = await ensure_app_key(client, app.name, environment["name"]) endpoints: Dict[str, str] = {} for name, handle in sorted(provisionable.items()): payload = _deployed_endpoint_input( @@ -419,6 +420,7 @@ async def reconcile_endpoints( build_id, python_version, env_name=environment["name"], + app_key=app_key, ) await attach_endpoint_volumes(payload, handle.spec, resolver, app) auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client) diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index c4e95ad1c..39de51080 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -12,8 +12,10 @@ from .api import AppsApiClient from .app import App +from .app_key import APP_KEY_ENV, delete_app_key, ensure_app_key, needs_app_key from .errors import AppError from .handles import ApiHandle, FunctionHandle +from .secret import Secret, render_env from .shim import runtime_launcher from .spec import ResourceKind, ResourceSpec from .targets import LiveTarget @@ -73,23 +75,6 @@ def _image_for(spec: ResourceSpec) -> str: return image_for_spec(spec, python_version=local_python_version()) -def _render_env(env): - from .secret import render_env - - return render_env(env) - - -def _client_api_key() -> str: - from .utils.network import api_key - - try: - return api_key() - except RuntimeError: - # payload construction must not require credentials (tests, - # dry runs); the session itself fails loudly on its first call - return "" - - def _cpu_locations(instance_ids: List[str]) -> str: """locations covering every requested cpu flavor's stock, limited to datacenters with storage and S3 support. the flash images are @@ -107,11 +92,20 @@ def _cpu_locations(instance_ids: List[str]) -> str: GENERATION_ENV = "RUNPOD_DEV_GENERATION" -def _endpoint_input(app: App, spec: ResourceSpec, generation: int = 1) -> Dict: - """build a live endpoint payload with an atomically bound container config.""" +def _endpoint_input( + app: App, + spec: ResourceSpec, + generation: int = 1, + app_key: Optional[Secret] = None, +) -> Dict: + """build a live endpoint payload with an atomically bound container config. + + app_key is referenced as RUNPOD_API_KEY unless the resource's env + sets its own. + """ spec.validate() - resource_env = _render_env(spec.env) + resource_env = render_env(spec.env) resource_env["RUNPOD_DEV_RESOURCE"] = spec.name if spec.kind is ResourceKind.API: resource_env["PORT"] = "80" @@ -131,15 +125,14 @@ def _endpoint_input(app: App, spec: ResourceSpec, generation: int = 1) -> Dict: "dockerArgs": "", "env": [ {"key": GENERATION_ENV, "value": str(generation)}, - # nested .remote() calls from inside a dev worker need - # credentials and the dev-session marker to resolve - # sibling dev endpoints by name + # nested .remote() calls resolve sibling dev endpoints + # by name through the dev-session marker + {"key": "RUNPOD_DEV_APP", "value": app.name}, *( - [{"key": "RUNPOD_API_KEY", "value": key}] - if (key := _client_api_key()) + [{"key": APP_KEY_ENV, "value": app_key.reference}] + if app_key is not None and needs_app_key(spec) else [] ), - {"key": "RUNPOD_DEV_APP", "value": app.name}, *( [ { @@ -215,6 +208,7 @@ def __init__( self._payloads: Dict[str, Dict] = {} # volumes resolve once per session (placement is stable) self._volume_resolver = None + self._app_keys: Dict[str, Secret] = {} def _emit(self, event: str, *args) -> None: emit(self.events, event, *args) @@ -236,6 +230,14 @@ async def _attach_volumes(self, payload: Dict, spec, app) -> None: self._volume_resolver = VolumeResolver(self.api, events=self.events) await attach_endpoint_volumes(payload, spec, self._volume_resolver, app) + async def _app_key_for(self, app: App, spec: ResourceSpec) -> Optional[Secret]: + """the app's dev key secret, resolved once per session.""" + if not needs_app_key(spec): + return None + if app.name not in self._app_keys: + self._app_keys[app.name] = await ensure_app_key(self.api, app.name) + return self._app_keys[app.name] + @property def _endpoint_ids(self) -> List[str]: return list(self._endpoints.values()) @@ -274,7 +276,9 @@ async def start(self) -> None: if found: self._endpoints[name] = found["id"] self._resource_names[name] = spec.name - payload = _endpoint_input(app, spec, self.generation) + payload = _endpoint_input( + app, spec, self.generation, await self._app_key_for(app, spec) + ) await self._attach_volumes(payload, spec, app) hardware = ",".join(spec.cpu or spec.gpu or ["any"]) @@ -351,7 +355,12 @@ async def refresh(self, apps: List[App]) -> None: # update survivors (config + generation bump) and create additions for name, (app, handle) in desired.items(): - payload = _endpoint_input(app, handle.spec, self.generation) + payload = _endpoint_input( + app, + handle.spec, + self.generation, + await self._app_key_for(app, handle.spec), + ) await self._attach_volumes(payload, handle.spec, app) existing_id = self._endpoints.get(name) @@ -394,7 +403,7 @@ async def refresh(self, apps: List[App]) -> None: ) async def stop(self, events: Optional[object] = None) -> None: - """delete every endpoint this session owns. + """delete every endpoint this session owns, then its dev credentials. events, when given, overrides the session sink for teardown rendering: cleanup_started(total), deleting(name), deleted(name), @@ -426,11 +435,32 @@ async def stop(self, events: Optional[object] = None) -> None: if failures: raise AppError("dev cleanup incomplete: " + "; ".join(failures)) + for app_name in list(self._app_keys): + try: + await delete_app_key(self.api, app_name) + except Exception as exc: + failures.append(f"credentials for {app_name} ({type(exc).__name__})") + log.warning( + "failed to delete dev credentials for %s (%s)", + app_name, + type(exc).__name__, + ) + else: + del self._app_keys[app_name] + if failures: + raise AppError("dev cleanup incomplete: " + "; ".join(failures)) + async def __aenter__(self) -> "DevSession": try: await self.start() except BaseException: - await self.stop() + try: + await self.stop() + except BaseException as exc: + log.warning( + "failed to clean up dev session after startup failure (%s)", + type(exc).__name__, + ) raise return self diff --git a/runpod/apps/manage.py b/runpod/apps/manage.py index 7c7145e72..dd9f01e73 100644 --- a/runpod/apps/manage.py +++ b/runpod/apps/manage.py @@ -10,6 +10,7 @@ from typing import Dict, List, Optional from .api import AppsApiClient +from .app_key import delete_app_key from .errors import AppError from .utils.events import emit @@ -75,7 +76,8 @@ async def undeploy_environment( delete_env: bool = True, events: Optional[object] = None, ) -> UndeployResult: - """tear down an environment: endpoints first, then the environment. + """tear down an environment: endpoints first, then its app key + and the environment. events, when given, receives cleanup_started(total), deleting(name), deleted(name), delete_failed(name). @@ -98,6 +100,12 @@ async def undeploy_environment( result.failures.append(f"endpoint {name}: {exc}") emit(events, "delete_failed", name) + if delete_env and not result.failures: + try: + await delete_app_key(client, app_name, env_name) + except Exception as exc: + result.failures.append(f"app key for {env_name}: {exc}") + if delete_env and not result.failures: try: await client.delete_environment(env["id"]) @@ -115,7 +123,8 @@ async def delete_app( api: Optional[AppsApiClient] = None, events: Optional[object] = None, ) -> UndeployResult: - """delete an app after undeploying every environment in it.""" + """delete an app after undeploying its environments and removing its + dev key.""" client = api or AppsApiClient() app = await get_app(app_name, api=client) @@ -127,6 +136,12 @@ async def delete_app( result.endpoints_deleted += env_result.endpoints_deleted result.failures.extend(env_result.failures) + if not result.failures: + try: + await delete_app_key(client, app_name) + except Exception as exc: + result.failures.append(f"dev key for {app_name}: {exc}") + if not result.failures: try: await client.delete_app(app["id"]) diff --git a/tests/test_apps/test_api_client.py b/tests/test_apps/test_api_client.py index d3ea81e9a..b03718501 100644 --- a/tests/test_apps/test_api_client.py +++ b/tests/test_apps/test_api_client.py @@ -6,8 +6,7 @@ import pytest from runpod.apps.api import AppsApiClient -from runpod.error import QueryError - +from runpod.error import AuthenticationError, QueryError def _respond(data): return AsyncMock(return_value={"data": data}) @@ -18,6 +17,65 @@ def _client_with(data): return client, patch("runpod.apps.api.run_graphql_query_async", _respond(data)) +class TestSecretReadRetry: + async def test_retries_transport_failures_with_bounded_backoff(self): + client = AppsApiClient(api_key="test-key") + transport = AsyncMock( + side_effect=[ + aiohttp.ClientConnectionError("reset"), + OSError("dns"), + TimeoutError("timeout"), + {"secrets": [{"id": "s1", "name": "tok"}]}, + ] + ) + with ( + patch("runpod.apps.api.run_rest_request_async", transport), + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + ): + assert await client.list_secrets(name="tok") == [ + {"id": "s1", "name": "tok"} + ] + assert sleep.await_args_list == [call(1), call(2), call(4)] + assert ( + transport.await_args_list + == [ + call( + "GET", + "/v2/account/secrets", + api_key="test-key", + params={"name": "tok"}, + ) + ] + * 4 + ) + + async def test_transport_retries_stop_after_four_attempts(self): + transport = AsyncMock(side_effect=TimeoutError("timeout")) + with ( + patch("runpod.apps.api.run_rest_request_async", transport), + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + pytest.raises(TimeoutError), + ): + await AppsApiClient(api_key="test-key").list_secrets() + assert transport.await_count == 4 + assert sleep.await_args_list == [call(1), call(2), call(4)] + + @pytest.mark.parametrize( + "failure", + [AuthenticationError("denied"), QueryError("forbidden", status_code=403)], + ) + async def test_authorization_failures_do_not_retry(self, failure): + transport = AsyncMock(side_effect=failure) + with ( + patch("runpod.apps.api.run_rest_request_async", transport), + patch("asyncio.sleep", new_callable=AsyncMock) as sleep, + pytest.raises(type(failure)), + ): + await AppsApiClient(api_key="test-key").list_secrets() + transport.assert_awaited_once() + sleep.assert_not_awaited() + + class TestExecuteRetry: async def test_returns_data(self): client, patcher = _client_with({"ok": 1}) @@ -704,17 +762,27 @@ async def test_any_supported_tier_is_eligible(self): class TestVolumesRegistrySecrets: async def test_secret_crud(self): - client, patcher = _client_with( - { - "myself": {"secrets": [{"id": "s1"}]}, - "secretCreate": {"id": "s1", "name": "tok"}, - "secretDelete": True, - } + client = AppsApiClient(api_key="test-key") + rest = AsyncMock( + side_effect=[ + {"secrets": [{"id": "s1"}]}, + {"secrets": [{"id": "s1", "name": "tok"}]}, + {"id": "s1", "name": "tok"}, + None, + ] ) - with patcher: + with patch("runpod.apps.api.run_rest_request_async", rest): assert await client.list_secrets() == [{"id": "s1"}] + assert await client.list_secrets(name="tok") == [{"id": "s1", "name": "tok"}] assert (await client.create_secret("tok", "v"))["id"] == "s1" assert await client.delete_secret("s1") is True + calls = [(c.args, c.kwargs.get("params"), c.kwargs.get("json")) for c in rest.await_args_list] + assert calls == [ + (("GET", "/v2/account/secrets"), None, None), + (("GET", "/v2/account/secrets"), {"name": "tok"}, None), + (("POST", "/v2/account/secrets"), None, {"name": "tok", "value": "v"}), + (("DELETE", "/v2/account/secrets/s1"), None, None), + ] class TestAuthRequests: diff --git a/tests/test_apps/test_dev.py b/tests/test_apps/test_dev.py index 52a82e7b0..60481c7f4 100644 --- a/tests/test_apps/test_dev.py +++ b/tests/test_apps/test_dev.py @@ -76,7 +76,7 @@ def q(): assert "FLASH_RESOURCE_NAME" not in env assert "RUNPOD_RESOURCE_NAME" not in env - def test_api_key_forwarded_when_configured(self, monkeypatch): + def test_account_api_key_never_forwarded(self, monkeypatch): monkeypatch.setenv("RUNPOD_API_KEY", "rpa_test123") app = App("keyed") @@ -86,7 +86,8 @@ def q(): payload = _endpoint_input(app, q.spec) env = {e["key"]: e["value"] for e in payload["template"]["env"]} - assert env["RUNPOD_API_KEY"] == "rpa_test123" + assert "RUNPOD_API_KEY" not in env + assert "rpa_test123" not in str(payload) assert "gpuIds" not in payload def test_gpu_queue_payload(self): diff --git a/tests/test_apps/test_worker_credentials.py b/tests/test_apps/test_worker_credentials.py new file mode 100644 index 000000000..17c8ee288 --- /dev/null +++ b/tests/test_apps/test_worker_credentials.py @@ -0,0 +1,906 @@ +"""worker credentials: workers get an api key minted per app environment, +and templates only ever carry a secret reference to it.""" + +import asyncio +import json +from unittest.mock import AsyncMock, call + +import aiohttp +import pytest + +import runpod +from runpod.apps import App +from runpod.apps.app import _clear_registry +from runpod.apps.app_key import ( + APP_KEY_SCOPES, + app_key_name, + delete_app_key, + ensure_app_key, +) +from runpod.apps.deploy import _deployed_endpoint_input, reconcile_endpoints +from runpod.apps.dev import DevSession, _endpoint_input +from runpod.apps.errors import AppError +from runpod.apps.manage import delete_app, undeploy_environment +from runpod.apps.secret import Secret +from runpod.error import AuthenticationError, QueryError + +ACCOUNT_KEY = "rpa_fakeaccountkey0000000000000000" +APP_KEY = "rpa_fakeappkey00000000000000000000" +PROD_SECRET = "flash-app-key.pipe.prod.ab680dbced7cc40bff0314e6" +DEV_SECRET = "flash-dev-key.pipe.8afde8d00cfa7b4458c20c7d" + + +def _reference(name): + return "{{ RUNPOD_SECRET_" + name + " }}" + + +@pytest.fixture(autouse=True) +def clean_registry(): + _clear_registry() + yield + _clear_registry() + + +@pytest.fixture +def logged_in(monkeypatch): + for name in ("RUNPOD_ENDPOINT_ID", "RUNPOD_POD_ID", "RUNPOD_DEV_APP"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("RUNPOD_API_KEY", ACCOUNT_KEY) + + +def _api(secrets=None, keys=None): + api = AsyncMock() + api.list_secrets.return_value = list(secrets or []) + api.list_api_keys.return_value = list(keys or []) + api.create_api_key.return_value = {"id": "key-new", "rawKey": APP_KEY} + api.list_my_endpoints.return_value = [] + api.save_endpoint.return_value = {"id": "ep-new"} + return api + + +def _pipeline_app(): + app = App("pipe") + + @app.queue(name="leaf", cpu="cpu3c-1-2") + def leaf(x: int): + return x + 1 + + @app.queue(name="root", cpu="cpu3c-1-2") + def root(x: int): + return leaf.remote(x) + + return app, leaf, root + + +def _env(payload): + return {e["key"]: e["value"] for e in payload["template"]["env"]} + + +def _saved(api): + return { + call.args[0]["name"]: call.args[0] for call in api.save_endpoint.await_args_list + } + + +def _no_key_values(*payloads): + text = json.dumps(payloads) + return ACCOUNT_KEY not in text and APP_KEY not in text + + +class TestAppKeyName: + def test_environment_and_dev_names_differ(self): + assert app_key_name("pipe", "prod") == PROD_SECRET + assert app_key_name("pipe") == DEV_SECRET + assert app_key_name("pipe", "dev") != app_key_name("pipe") + + def test_long_names_fit_api_key_limit_and_stay_distinct(self): + a = app_key_name("a" * 80, "prod") + b = app_key_name("a" * 80, "staging") + assert len(a) <= 64 and len(b) <= 64 + assert a != b + Secret(a) + + def test_unsafe_characters_become_valid_secret_names(self): + Secret(app_key_name("my app", "prod/eu")) + + @pytest.mark.parametrize( + "left,right", + [ + (("my app", "prod"), ("my-app", "prod")), + (("a.b", "c"), ("a", "b.c")), + (("MyApp", "prod"), ("myapp", "prod")), + (("pipe", "Prod"), ("pipe", "prod")), + (("MyApp", None), ("myapp", None)), + (("pipe", None), ("pipe", "")), + ], + ) + def test_original_identities_are_distinct_case_insensitively(self, left, right): + first, second = app_key_name(*left), app_key_name(*right) + assert first.casefold() != second.casefold() + assert first == app_key_name(*left) + assert len(first) <= 64 and len(second) <= 64 + Secret(first) + Secret(second) + + +class TestEnsureAppKey: + + @pytest.mark.parametrize( + "outcome", ["absent", "committed", "winner", "unknown", "revoke_failed"] + ) + async def test_cancelled_secret_creation_reconciles_and_propagates( + self, outcome, caplog + ): + api = _api() + entered = asyncio.Event() + blocked = asyncio.Event() + + async def create_secret(name, value, description): + if outcome in ("committed", "winner"): + api.list_secrets.return_value = [ + { + "id": "s1", + "name": name, + "description": ( + description if outcome == "committed" else "another owner" + ), + } + ] + api.list_api_keys.return_value = [{"id": "key-new", "name": name}] + elif outcome == "unknown": + api.list_secrets.side_effect = AuthenticationError(APP_KEY) + elif outcome == "revoke_failed": + api.delete_api_key.side_effect = RuntimeError(APP_KEY) + entered.set() + await blocked.wait() + + api.create_secret.side_effect = create_secret + task = asyncio.create_task(ensure_app_key(api, "pipe", "prod")) + try: + await asyncio.wait_for(entered.wait(), 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 2) + if outcome in ("committed", "unknown"): + api.delete_api_key.assert_not_awaited() + else: + api.delete_api_key.assert_awaited_once_with("key-new") + assert APP_KEY not in caplog.text + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_base_exception_after_key_creation_runs_rollback(self): + api = _api() + failure = BaseException("interrupted") + api.create_secret.side_effect = failure + with pytest.raises(BaseException) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_awaited_once_with("key-new") + + async def test_concurrent_stale_secret_refresh_reconciles_one_live_pair(self): + api = _api() + stale = { + "id": "stale", + "name": PROD_SECRET, + "description": "api key revoked for pipe/prod workers", + } + secret = stale + keys = {} + reads = 0 + both_read = asyncio.Event() + + async def list_secrets(name): + nonlocal reads + if reads < 2: + reads += 1 + if reads == 2: + both_read.set() + await both_read.wait() + return [stale] + return [secret] if secret else [] + + async def list_keys(): + return [{"id": key_id, "name": PROD_SECRET} for key_id in keys] + + async def delete_secret(secret_id): + nonlocal secret + if secret is None or secret["id"] != secret_id: + raise QueryError("not found", status_code=404) + secret = None + + async def create_key(name, scopes): + key_id = f"key-{api.create_api_key.await_count}" + keys[key_id] = APP_KEY + return {"id": key_id, "rawKey": APP_KEY} + + async def create_secret(name, value, description): + nonlocal secret + if secret is not None: + raise QueryError("conflict", status_code=409) + secret = {"id": "fresh", "name": name, "description": description} + + async def delete_key(key_id): + del keys[key_id] + + api.list_secrets.side_effect = list_secrets + api.list_api_keys.side_effect = list_keys + api.delete_secret.side_effect = delete_secret + api.create_api_key.side_effect = create_key + api.create_secret.side_effect = create_secret + api.delete_api_key.side_effect = delete_key + results = await asyncio.wait_for( + asyncio.gather(*(ensure_app_key(api, "pipe", "prod") for _ in range(2))), + 2, + ) + assert [result.name for result in results] == [PROD_SECRET, PROD_SECRET] + assert api.delete_secret.await_count == 2 + assert api.create_api_key.await_count == 2 + assert len(keys) == 1 + assert ( + secret["description"] == f"api key {next(iter(keys))} for pipe/prod workers" + ) + + @pytest.mark.parametrize("status", [403, 500]) + async def test_stale_secret_delete_errors_other_than_missing_propagate( + self, status + ): + api = _api([{"id": "stale", "name": PROD_SECRET}]) + failure = QueryError("delete failed", status_code=status) + api.delete_secret.side_effect = failure + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.create_api_key.assert_not_awaited() + + async def test_refresh_preserves_other_keys_until_teardown(self): + api = _api( + [ + { + "id": "stale", + "name": PROD_SECRET, + "description": "api key revoked for pipe/prod workers", + } + ], + [{"id": "orphan", "name": PROD_SECRET}], + ) + await ensure_app_key(api, "pipe", "prod") + api.delete_api_key.assert_not_awaited() + api.list_secrets.return_value = [{"id": "fresh", "name": PROD_SECRET}] + api.list_api_keys.return_value.append({"id": "key-new", "name": PROD_SECRET}) + await delete_app_key(api, "pipe", "prod") + assert api.delete_api_key.await_args_list == [call("orphan"), call("key-new")] + + @pytest.mark.parametrize( + "message,status", + [ + ("max of 50 API keys are allowed", None), + ("rate limit exceeded", 429), + ("permission service unavailable", 503), + ("upstream error", 500), + ("invalid scope", 400), + ], + ) + async def test_nonpermission_key_errors_are_unchanged(self, message, status): + api = _api() + failure = QueryError(message, status_code=status, errors=["detail"]) + api.create_api_key.side_effect = failure + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + assert str(caught.value) == message + assert caught.value.errors == ["detail"] + api.create_secret.assert_not_awaited() + + @pytest.mark.parametrize( + "message,status", + [("denied", 403), ("unauthorized", 401), ("permission denied", None)], + ) + async def test_permission_query_errors_include_override_guidance( + self, message, status + ): + api = _api() + api.create_api_key.side_effect = QueryError(message, status_code=status) + with pytest.raises(QueryError, match="RUNPOD_API_KEY"): + await ensure_app_key(api, "pipe", "prod") + + async def test_mints_scoped_key_into_secret(self): + api = _api() + secret = await ensure_app_key(api, "pipe", "prod") + assert secret.name == PROD_SECRET + api.create_api_key.assert_awaited_once_with(PROD_SECRET, APP_KEY_SCOPES) + args = api.create_secret.await_args + assert args.args == (PROD_SECRET, APP_KEY) + assert APP_KEY not in args.kwargs["description"] + + async def test_reuses_existing_secret_without_minting(self): + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [{"id": "k1", "name": PROD_SECRET}], + ) + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.create_api_key.assert_not_awaited() + api.create_secret.assert_not_awaited() + + async def test_revoked_key_refreshes_stale_secret(self): + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [{"id": "other", "name": "unrelated"}], + ) + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.delete_secret.assert_awaited_once_with("s1") + api.create_api_key.assert_awaited_once_with(PROD_SECRET, APP_KEY_SCOPES) + api.create_secret.assert_awaited_once() + assert api.mock_calls.index(call.delete_secret("s1")) < api.mock_calls.index( + call.create_api_key(PROD_SECRET, APP_KEY_SCOPES) + ) + + @pytest.mark.parametrize("owner_live", [True, False]) + async def test_existing_secret_validates_recorded_owner_not_retained_orphan( + self, owner_live + ): + keys = [{"id": "orphan", "name": PROD_SECRET}] + if owner_live: + keys.append({"id": "owner", "name": PROD_SECRET}) + api = _api( + [ + { + "id": "s1", + "name": PROD_SECRET, + "description": "api key owner for pipe/prod workers", + } + ], + keys, + ) + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + if owner_live: + api.delete_secret.assert_not_awaited() + api.create_api_key.assert_not_awaited() + else: + api.delete_secret.assert_awaited_once_with("s1") + api.create_api_key.assert_awaited_once_with(PROD_SECRET, APP_KEY_SCOPES) + api.create_secret.assert_awaited_once() + + async def test_legacy_secret_description_uses_same_name_key_check(self): + api = _api( + [ + { + "id": "s1", + "name": PROD_SECRET, + "description": "api key for pipe/prod workers", + } + ], + [{"id": "owner", "name": PROD_SECRET}], + ) + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.delete_secret.assert_not_awaited() + api.create_api_key.assert_not_awaited() + + async def test_stale_secret_lookup_failure_does_not_rotate_credentials(self): + api = _api([{"id": "s1", "name": PROD_SECRET}]) + api.list_api_keys.side_effect = AuthenticationError("denied") + with pytest.raises(AuthenticationError): + await ensure_app_key(api, "pipe", "prod") + api.delete_secret.assert_not_awaited() + api.create_api_key.assert_not_awaited() + + async def test_concurrent_creation_revokes_the_extra_key(self): + api = _api(keys=[{"id": "winner", "name": PROD_SECRET}]) + api.list_secrets.side_effect = [[], [{"id": "s1", "name": PROD_SECRET}]] + api.create_secret.side_effect = QueryError( + "Secret name must be unique.", status_code=409 + ) + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.delete_api_key.assert_awaited_once_with("key-new") + + @pytest.mark.parametrize("owner_live", [True, False]) + async def test_concurrent_winner_requires_its_recorded_live_key(self, owner_live): + keys = [ + {"id": "key-new", "name": PROD_SECRET}, + {"id": "orphan", "name": PROD_SECRET}, + ] + if owner_live: + keys.append({"id": "winner", "name": PROD_SECRET}) + api = _api(keys=keys) + failure = QueryError("conflict", status_code=409) + api.create_secret.side_effect = failure + api.delete_api_key.side_effect = RuntimeError("revocation failed") + api.list_secrets.side_effect = [ + [], + [ + { + "id": "s1", + "name": PROD_SECRET, + "description": "api key winner for pipe/prod workers", + } + ], + ] + if owner_live: + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + else: + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_awaited_once_with("key-new") + api.delete_secret.assert_not_awaited() + + async def test_metadata_less_winner_cannot_use_failed_rollback_key(self): + api = _api(keys=[{"id": "key-new", "name": PROD_SECRET}]) + failure = QueryError("conflict", status_code=409) + api.create_secret.side_effect = failure + api.delete_api_key.side_effect = RuntimeError("revocation failed") + api.list_secrets.side_effect = [[], [{"id": "s1", "name": PROD_SECRET}]] + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + + async def test_concurrent_winner_key_lookup_failure_preserves_original_error(self): + api = _api() + failure = QueryError("conflict", status_code=409) + api.create_secret.side_effect = failure + api.list_secrets.side_effect = [ + [], + [ + { + "id": "s1", + "name": PROD_SECRET, + "description": "api key winner for pipe/prod workers", + } + ], + ] + api.list_api_keys.side_effect = AuthenticationError("read denied") + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_awaited_once_with("key-new") + + @pytest.mark.parametrize( + "failure", + [ + AuthenticationError("denied"), + aiohttp.ClientConnectionError("disconnected"), + TimeoutError("timed out"), + OSError("unreachable"), + ], + ) + async def test_secret_transport_failure_revokes_unstored_key(self, failure): + api = _api() + api.create_secret.side_effect = failure + with pytest.raises(type(failure)) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_awaited_once_with("key-new") + + @pytest.mark.parametrize( + "failure", [TimeoutError("timed out"), QueryError("failed", status_code=500)] + ) + async def test_committed_secret_keeps_its_key_after_lost_response(self, failure): + api = _api(keys=[{"id": "key-new", "name": PROD_SECRET}]) + + async def commit(name, value, description): + api.list_secrets.return_value = [ + {"id": "s1", "name": name, "description": description} + ] + raise failure + + api.create_secret.side_effect = commit + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.delete_api_key.assert_not_awaited() + + async def test_committed_secret_with_revoked_owner_is_not_reused(self): + api = _api(keys=[{"id": "orphan", "name": PROD_SECRET}]) + failure = TimeoutError("post response lost") + api.create_secret.side_effect = failure + api.list_secrets.side_effect = [ + [], + [ + { + "id": "s1", + "name": PROD_SECRET, + "description": "api key key-new for pipe/prod workers", + } + ], + ] + with pytest.raises(TimeoutError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_not_awaited() + + async def test_uncertain_secret_outcome_preserves_key_and_original_error(self): + api = _api() + failure = TimeoutError("post response lost") + api.create_secret.side_effect = failure + api.list_secrets.side_effect = [[], AuthenticationError("read denied")] + with pytest.raises(TimeoutError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_not_awaited() + + async def test_uncertain_secret_metadata_does_not_revoke_its_possible_key(self): + api = _api() + failure = TimeoutError("post response lost") + api.create_secret.side_effect = failure + api.list_secrets.side_effect = [[], [{"id": "s1", "name": PROD_SECRET}]] + with pytest.raises(TimeoutError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_not_awaited() + + @pytest.mark.parametrize("winner", [True, False]) + async def test_revoke_failure_does_not_mask_creation_result(self, winner, caplog): + api = _api(keys=[{"id": "winner", "name": PROD_SECRET}] if winner else []) + failure = QueryError("conflict", status_code=409) + api.create_secret.side_effect = failure + api.delete_api_key.side_effect = RuntimeError(APP_KEY) + api.list_secrets.side_effect = [ + [], + ( + [{"id": "s1", "name": PROD_SECRET, "description": "another key"}] + if winner + else [] + ), + ] + if winner: + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + else: + with pytest.raises(QueryError) as caught: + await ensure_app_key(api, "pipe", "prod") + assert caught.value is failure + api.delete_api_key.assert_awaited_once_with("key-new") + assert "could not revoke api key for secret" in caplog.text + assert "key-new" not in caplog.text + assert APP_KEY not in caplog.text + + @pytest.mark.parametrize("outcome", ["uncertain", "metadata_less", "revoke_failed"]) + async def test_cleanup_logs_exclude_key_response_fields(self, outcome, caplog): + api = _api() + api.create_api_key.return_value = {"id": APP_KEY, "rawKey": APP_KEY} + api.create_secret.side_effect = TimeoutError(APP_KEY) + if outcome == "uncertain": + api.list_secrets.side_effect = [[], AuthenticationError(APP_KEY)] + elif outcome == "metadata_less": + api.list_secrets.side_effect = [[], [{"id": "s1", "name": PROD_SECRET}]] + else: + api.delete_api_key.side_effect = RuntimeError(APP_KEY) + with pytest.raises(TimeoutError): + await ensure_app_key(api, "pipe", "prod") + assert PROD_SECRET in caplog.text + assert APP_KEY not in caplog.text + + async def test_secret_failure_revokes_key_and_propagates(self): + api = _api() + api.create_secret.side_effect = QueryError("forbidden") + with pytest.raises(QueryError): + await ensure_app_key(api, "pipe", "prod") + api.delete_api_key.assert_awaited_once_with("key-new") + + async def test_key_creation_failure_propagates_before_any_secret(self): + api = _api() + api.create_api_key.side_effect = QueryError("max of 50 API keys are allowed") + with pytest.raises(QueryError): + await ensure_app_key(api, "pipe", "prod") + api.create_secret.assert_not_awaited() + + @pytest.mark.parametrize( + "failure", + [ + QueryError("Scope(s) not grantable by the caller"), + AuthenticationError("denied"), + ], + ) + async def test_key_permission_error_explains_explicit_credentials(self, failure): + api = _api() + api.create_api_key.side_effect = failure + with pytest.raises(type(failure), match="RUNPOD_API_KEY") as caught: + await ensure_app_key(api, "pipe", "prod") + assert "each resource's env" in str(caught.value) + assert "API key creation permission" in str(caught.value) + assert all(scope in str(caught.value) for scope in APP_KEY_SCOPES) + api.create_secret.assert_not_awaited() + + +class TestDeleteAppKey: + async def test_removes_secret_and_every_key_with_its_name(self): + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [ + {"id": "k1", "name": PROD_SECRET}, + {"id": "k2", "name": PROD_SECRET}, + {"id": "k3", "name": "mine"}, + ], + ) + assert await delete_app_key(api, "pipe", "prod") + api.delete_secret.assert_awaited_once_with("s1") + assert [c.args[0] for c in api.delete_api_key.await_args_list] == ["k1", "k2"] + + async def test_nothing_to_delete(self): + api = _api() + assert not await delete_app_key(api, "pipe", "prod") + api.delete_secret.assert_not_awaited() + + async def test_environment_deletion_removes_its_key(self): + api = _api([{"id": "s1", "name": PROD_SECRET}]) + api.get_environment_by_name.return_value = {"id": "env-1", "endpoints": []} + result = await undeploy_environment("pipe", "prod", api=api) + assert result.environment_deleted and not result.failures + api.delete_secret.assert_awaited_once_with("s1") + + async def test_kept_environment_keeps_its_key(self): + api = _api([{"id": "s1", "name": PROD_SECRET}]) + api.get_environment_by_name.return_value = {"id": "env-1", "endpoints": []} + await undeploy_environment("pipe", "prod", api=api, delete_env=False) + api.delete_secret.assert_not_awaited() + + async def test_app_deletion_removes_environment_and_dev_keys(self): + api = _api( + [{"id": "s1", "name": PROD_SECRET}, {"id": "s2", "name": DEV_SECRET}] + ) + api.get_app_by_name.return_value = { + "id": "app-1", + "flashEnvironments": [{"name": "prod"}], + } + api.get_environment_by_name.return_value = {"id": "env-1", "endpoints": []} + result = await delete_app("pipe", api=api) + assert result.app_deleted and not result.failures + assert {c.args[0] for c in api.delete_secret.await_args_list} == {"s1", "s2"} + + @pytest.mark.parametrize( + "operation", + ["list_secrets", "delete_secret", "list_api_keys", "delete_api_key"], + ) + async def test_environment_survives_credential_failure_for_retry(self, operation): + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [{"id": "k1", "name": PROD_SECRET}], + ) + api.get_environment_by_name.return_value = { + "id": "env-1", + "endpoints": [{"id": "ep-1", "name": "worker"}], + } + getattr(api, operation).side_effect = RuntimeError("unavailable") + result = await undeploy_environment("pipe", "prod", api=api) + assert result.endpoints_deleted == 1 + assert result.failures and not result.environment_deleted + api.delete_environment.assert_not_awaited() + getattr(api, operation).side_effect = None + result = await undeploy_environment("pipe", "prod", api=api) + assert result.environment_deleted and not result.failures + assert api.mock_calls.index( + call.delete_endpoint("ep-1") + ) < api.mock_calls.index(call.delete_secret("s1")) + assert api.mock_calls.index(call.delete_api_key("k1")) < api.mock_calls.index( + call.delete_environment("env-1") + ) + + @pytest.mark.parametrize( + "operation", + ["list_secrets", "delete_secret", "list_api_keys", "delete_api_key"], + ) + async def test_app_survives_dev_credential_failure_for_retry(self, operation): + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + api.get_app_by_name.return_value = {"id": "app-1", "flashEnvironments": []} + getattr(api, operation).side_effect = RuntimeError("unavailable") + result = await delete_app("pipe", api=api) + assert result.failures and not result.app_deleted + api.delete_app.assert_not_awaited() + getattr(api, operation).side_effect = None + result = await delete_app("pipe", api=api) + assert result.app_deleted and not result.failures + assert api.mock_calls.index(call.delete_api_key("k1")) < api.mock_calls.index( + call.delete_app("app-1") + ) + + +class TestDeployedEndpoints: + def test_account_key_never_lands_in_payload(self, logged_in, monkeypatch): + monkeypatch.setattr(runpod, "api_key", ACCOUNT_KEY) + app, leaf, _ = _pipeline_app() + payload = _deployed_endpoint_input(app, leaf.spec, "env-1", "build-1", "3.12") + assert "RUNPOD_API_KEY" not in _env(payload) + assert _no_key_values(payload) + + async def test_every_resource_references_environment_key(self, logged_in): + app, _, _ = _pipeline_app() + api = _api() + await reconcile_endpoints( + api, app, {"id": "env-1", "name": "prod"}, "build-1", "3.12" + ) + saved = _saved(api) + assert _env(saved["root"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) + assert _env(saved["leaf"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) + assert _no_key_values(*saved.values()) + api.create_api_key.assert_awaited_once() + + async def test_redeploy_reuses_the_environment_key(self, logged_in): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [{"id": "k1", "name": PROD_SECRET}], + ) + await reconcile_endpoints( + api, app, {"id": "env-1", "name": "prod"}, "build-2", "3.12" + ) + assert _env(_saved(api)["root"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) + api.create_api_key.assert_not_awaited() + + async def test_explicit_env_credentials_win(self, logged_in): + app = App("own") + + @app.queue( + name="caller", + cpu="cpu3c-1-2", + env={"RUNPOD_API_KEY": Secret("my-scoped-key")}, + ) + def caller(): + pass + + api = _api() + await reconcile_endpoints( + api, app, {"id": "env-1", "name": "default"}, "build-1", "3.12" + ) + env = _env(_saved(api)["caller"]) + assert env["RUNPOD_API_KEY"] == _reference("my-scoped-key") + api.list_secrets.assert_not_awaited() + api.create_api_key.assert_not_awaited() + + +class TestDevEndpoints: + + @pytest.mark.parametrize("cancelled", [False, True]) + async def test_start_failure_survives_credential_cleanup_failure( + self, cancelled, caplog + ): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + failure = ( + asyncio.CancelledError("interrupted") + if cancelled + else RuntimeError("provisioning failed") + ) + api.save_endpoint.side_effect = failure + api.delete_api_key.side_effect = QueryError(APP_KEY, status_code=503) + session = DevSession([app], api=api) + with pytest.raises(type(failure)) as caught: + async with session: + pytest.fail("startup must fail") + assert caught.value is failure + assert "QueryError" in caplog.text + assert "failed to clean up dev session" in caplog.text + assert APP_KEY not in caplog.text + assert session._app_keys + api.delete_api_key.side_effect = None + await session.stop() + assert not session._app_keys + + async def test_credential_cleanup_reports_type_without_sensitive_message( + self, caplog + ): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + session = DevSession([app], api=api) + await session.start() + api.delete_api_key.side_effect = QueryError(APP_KEY, status_code=503) + with pytest.raises( + AppError, match=r"credentials for pipe \(QueryError\)" + ) as caught: + await session.stop() + assert APP_KEY not in str(caught.value) + assert APP_KEY not in caplog.text + + async def test_stop_deletes_credentials_without_a_registered_app(self): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + api.get_app_by_name.return_value = None + session = DevSession([app], api=api) + await session.start() + await session.stop() + api.delete_secret.assert_awaited_once_with("s1") + api.delete_api_key.assert_awaited_once_with("k1") + api.get_app_by_name.assert_not_awaited() + assert api.mock_calls.index( + call.delete_endpoint("ep-new") + ) < api.mock_calls.index(call.delete_secret("s1")) + await session.stop() + api.delete_api_key.assert_awaited_once() + + async def test_failed_endpoint_teardown_retains_dev_credentials_for_retry(self): + app, _, _ = _pipeline_app() + api = _api() + session = DevSession([app], api=api) + await session.start() + api.delete_endpoint.side_effect = RuntimeError("unavailable") + with pytest.raises(AppError, match="dev cleanup incomplete"): + await session.stop() + api.delete_secret.assert_not_awaited() + api.delete_api_key.assert_not_awaited() + assert session._app_keys + api.delete_endpoint.side_effect = None + await session.stop() + assert not session._app_keys + + async def test_failed_credential_teardown_can_be_retried(self): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + session = DevSession([app], api=api) + await session.start() + await session.refresh([]) + api.delete_api_key.side_effect = RuntimeError("unavailable") + with pytest.raises(AppError, match="credentials for pipe"): + await session.stop() + assert session._app_keys + api.delete_api_key.side_effect = None + await session.stop() + assert not session._app_keys + + async def test_failed_start_cleans_up_credentials(self): + app, _, _ = _pipeline_app() + api = _api( + [{"id": "s1", "name": DEV_SECRET}], + [{"id": "k1", "name": DEV_SECRET}], + ) + api.save_endpoint.side_effect = RuntimeError("provisioning failed") + with pytest.raises(RuntimeError, match="provisioning failed"): + async with DevSession([app], api=api): + pass + api.delete_secret.assert_awaited_once_with("s1") + api.delete_api_key.assert_awaited_once_with("k1") + + def test_payload_needs_no_credentials(self, monkeypatch): + monkeypatch.delenv("RUNPOD_API_KEY", raising=False) + monkeypatch.setattr(runpod, "api_key", None) + app, _, root = _pipeline_app() + assert "RUNPOD_API_KEY" not in _env(_endpoint_input(app, root.spec)) + + async def test_session_references_dev_key(self, logged_in): + app, _, _ = _pipeline_app() + api = _api() + session = DevSession([app], api=api) + await session.start() + await session.refresh([app]) + saved = { + call.args[0]["name"].split("-")[2]: call.args[0] + for call in api.save_endpoint.await_args_list + } + assert _env(saved["root"])["RUNPOD_API_KEY"] == _reference(DEV_SECRET) + assert _env(saved["leaf"])["RUNPOD_API_KEY"] == _reference(DEV_SECRET) + assert _no_key_values([c.args for c in api.save_endpoint.await_args_list]) + api.create_api_key.assert_awaited_once() + + +class TestNestedCallsInWorker: + async def test_worker_authenticates_with_substituted_key( + self, logged_in, monkeypatch + ): + app, leaf, _ = _pipeline_app() + api = _api() + await reconcile_endpoints( + api, app, {"id": "env-1", "name": "prod"}, "build-1", "3.12" + ) + root_env = _env(_saved(api)["root"]) + assert root_env["RUNPOD_API_KEY"] == _reference(PROD_SECRET) + + # the platform decrypts the secret reference when the worker boots + monkeypatch.setenv("RUNPOD_API_KEY", APP_KEY) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-root") + monkeypatch.setenv("FLASH_ENVIRONMENT", root_env["FLASH_ENVIRONMENT"]) + target = await app._resolve(leaf.spec) + headers = target._sentinel_headers() + assert headers["Authorization"] == f"Bearer {APP_KEY}" + assert headers["X-Flash-Endpoint"] == "leaf" + assert headers["X-Flash-Environment"] == "prod"