From 8ac9493d57c7ce9bc3ef79e0161c3344335860dc Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:34:46 -0700 Subject: [PATCH 1/6] fix: give workers a scoped per-app key only when they make remote calls --- runpod/api/mutations/apps.py | 18 ++ runpod/api/queries/apps.py | 8 + runpod/apps/api.py | 21 ++ runpod/apps/app.py | 13 +- runpod/apps/app_key.py | 122 ++++++++ runpod/apps/deploy.py | 22 +- runpod/apps/dev.py | 66 ++-- runpod/apps/manage.py | 19 +- runpod/apps/spec.py | 1 + runpod/apps/utils/network.py | 7 + runpod/rp_cli/main.py | 22 ++ tests/e2e/examples/07_pipelines.py | 2 +- tests/test_apps/test_dev.py | 5 +- tests/test_apps/test_worker_credentials.py | 340 +++++++++++++++++++++ 14 files changed, 622 insertions(+), 44 deletions(-) create mode 100644 runpod/apps/app_key.py create mode 100644 tests/test_apps/test_worker_credentials.py diff --git a/runpod/api/mutations/apps.py b/runpod/api/mutations/apps.py index 2e3c73a44..ad864fbd3 100644 --- a/runpod/api/mutations/apps.py +++ b/runpod/api/mutations/apps.py @@ -54,6 +54,24 @@ } """ +MUTATION_UPDATE_SECRET_VALUE = """ +mutation secretValueUpdate($input: SecretValueUpdateInput!) { + secretValueUpdate(input: $input) { id } +} +""" + +MUTATION_CREATE_API_KEY = """ +mutation createApiKeyNew($input: CreateApiKeyInput!) { + createApiKeyNew(input: $input) { id rawKey } +} +""" + +MUTATION_DELETE_API_KEY = """ +mutation deleteApiKey($input: DeleteApiKeyInput!) { + deleteApiKey(input: $input) { count } +} +""" + MUTATION_DELETE_FLASH_APP = """ mutation deleteFlashApp($flashAppId: String!) { deleteFlashApp(flashAppId: $flashAppId) diff --git a/runpod/api/queries/apps.py b/runpod/api/queries/apps.py index 9da1b2cde..9ddc78261 100644 --- a/runpod/api/queries/apps.py +++ b/runpod/api/queries/apps.py @@ -33,6 +33,14 @@ } """ +QUERY_API_KEYS = """ +query myApiKeys { + myself { + apiKeys { id name } + } +} +""" + QUERY_FLASH_APPS = """ query getFlashApps { myself { diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 9b557800a..1ca3af5d5 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -504,6 +504,27 @@ async def delete_secret(self, secret_id: str) -> bool: data = await self._execute(mutation, {"id": secret_id}) return bool(data.get("secretDelete")) + async def update_secret_value(self, secret_id: str, value: str) -> None: + mutation = app_mutations.MUTATION_UPDATE_SECRET_VALUE + await self._execute(mutation, {"input": {"id": secret_id, "value": value}}) + + 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.""" query = app_queries.QUERY_FLASH_APPS diff --git a/runpod/apps/app.py b/runpod/apps/app.py index c2a6a0e5d..144d11b4b 100644 --- a/runpod/apps/app.py +++ b/runpod/apps/app.py @@ -98,8 +98,13 @@ def queue( min_cuda_version: Optional[str] = None, accelerate_downloads: bool = True, container_disk_gb: Optional[int] = None, + remote_calls: bool = False, ) -> Callable[[Callable], FunctionHandle]: - """declare a queue-based serverless endpoint from a function.""" + """declare a queue-based serverless endpoint from a function. + + set remote_calls=True when the function calls other resources with + .remote(); only then do its workers receive api credentials. + """ def decorator(fn: Callable) -> FunctionHandle: spec = ResourceSpec( @@ -126,6 +131,7 @@ def decorator(fn: Callable) -> FunctionHandle: min_cuda_version=min_cuda_version, accelerate_downloads=accelerate_downloads, container_disk_gb=container_disk_gb, + remote_calls=remote_calls, ) handle = FunctionHandle(self, fn, spec) self._register(spec.name, handle) @@ -201,6 +207,7 @@ def api( min_cuda_version: Optional[str] = None, accelerate_downloads: bool = True, container_disk_gb: Optional[int] = None, + remote_calls: bool = False, ) -> Callable[[Any], ApiHandle]: """declare a load-balanced serverless endpoint. @@ -222,6 +229,9 @@ def web(): server = FastAPI() ... return server + + set remote_calls=True when the service calls other resources with + .remote(); only then do its workers receive api credentials. """ def decorator(target: Any) -> ApiHandle: @@ -248,6 +258,7 @@ def decorator(target: Any) -> ApiHandle: min_cuda_version=min_cuda_version, accelerate_downloads=accelerate_downloads, container_disk_gb=container_disk_gb, + remote_calls=remote_calls, ) handle = ApiHandle(self, target, spec) self._register(spec.name, handle) diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py new file mode 100644 index 000000000..baaa6550a --- /dev/null +++ b/runpod/apps/app_key.py @@ -0,0 +1,122 @@ +"""per-app api keys for workers that make outbound calls. + +a resource that declares remote_calls gets RUNPOD_API_KEY from a platform +secret holding an api key minted for its app and environment. 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 re +from typing import Any, Dict, List, Optional + +from ..error import QueryError +from .errors import AppError +from .secret import Secret + +APP_KEY_ENV = "RUNPOD_API_KEY" + +# 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_.-]") + + +class AppKeyError(AppError): + pass + + +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"runpod-dev-key.{app_name}" + else: + raw = f"runpod-app-key.{app_name}.{env_name}" + name = _UNSAFE.sub("-", raw) + if len(name) <= _MAX_NAME: + return name + digest = hashlib.sha256(raw.encode()).hexdigest()[:12] + return f"{name[: _MAX_NAME - len(digest) - 1]}.{digest}" + + +def needs_app_key(spec) -> bool: + """true when a resource makes outbound calls and does not supply its + own credentials through env.""" + return spec.remote_calls and 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(): + 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 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.""" + name = app_key_name(app_name, env_name) + if await _find_secret(api, name): + return Secret(name) + key = await api.create_api_key(name, APP_KEY_SCOPES) + target = f"{app_name}/{env_name}" if env_name else f"{app_name} dev sessions" + try: + await api.create_secret( + name, key["rawKey"], description=f"api key for {target} workers" + ) + except QueryError: + await api.delete_api_key(key["id"]) + # a concurrent deploy may have created it first + if not await _find_secret(api, name): + raise + return Secret(name) + + +async def rotate_app_key(api, app_name: str, env_name: Optional[str] = None) -> None: + """mint a new key into the existing secret and revoke the old ones. + + workers read the secret when they boot, so running workers keep the + revoked key until they restart. + """ + name = app_key_name(app_name, env_name) + secret = await _find_secret(api, name) + if secret is None: + raise AppKeyError(f"no app key '{name}' to rotate") + stale = await _key_ids(api, name) + key = await api.create_api_key(name, APP_KEY_SCOPES) + await api.update_secret_value(secret["id"], key["rawKey"]) + for key_id in stale: + await api.delete_api_key(key_id) + + +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..dff289cc7 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 only for resources that + declare remote_calls; every other worker boots without credentials. """ 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..fa97f2c45 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, 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 only for resources that + declare remote_calls; every other worker boots without credentials. + """ 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,15 @@ 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 and only + when a resource declares remote_calls.""" + 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 +277,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 +356,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) diff --git a/runpod/apps/manage.py b/runpod/apps/manage.py index 7c7145e72..ff3154e89 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 the environment + and its app key. events, when given, receives cleanup_started(total), deleting(name), deleted(name), delete_failed(name). @@ -106,6 +108,12 @@ async def undeploy_environment( except Exception as exc: # noqa: BLE001 - collected for the caller result.failures.append(f"environment {env_name}: {exc}") + if result.environment_deleted: + try: + await delete_app_key(client, app_name, env_name) + except Exception as exc: # noqa: BLE001 - collected for the caller + result.failures.append(f"app key for {env_name}: {exc}") + return result @@ -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 every environment in it, then its + dev key.""" client = api or AppsApiClient() app = await get_app(app_name, api=client) @@ -135,4 +144,10 @@ async def delete_app( except Exception as exc: # noqa: BLE001 - collected for the caller result.failures.append(f"app {app_name}: {exc}") + if result.app_deleted: + try: + await delete_app_key(client, app_name) + except Exception as exc: # noqa: BLE001 - collected for the caller + result.failures.append(f"dev key for {app_name}: {exc}") + return result diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index a6a41373d..327859502 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -190,6 +190,7 @@ class ResourceSpec: container_disk_gb: Optional[int] = None routes: List[RouteSpec] = field(default_factory=list) asgi_factory: Optional[str] = None + remote_calls: bool = False def __post_init__(self) -> None: self.validate() diff --git a/runpod/apps/utils/network.py b/runpod/apps/utils/network.py index d681766ea..54df2fe48 100644 --- a/runpod/apps/utils/network.py +++ b/runpod/apps/utils/network.py @@ -13,6 +13,13 @@ def api_key() -> str: key = runpod.api_key if not key: + from ..context import Context, current_context + + if current_context() is Context.WORKER: + raise RuntimeError( + "no api key in this worker. declare remote_calls=True on the " + "resource that makes .remote() calls." + ) raise RuntimeError( "no api key configured. run `rp login` or set RUNPOD_API_KEY." ) diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 5c4b4bfdc..359c24b78 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -673,6 +673,28 @@ def app_delete(app_name, yes): ) +@app.command(name="rotate-key") +@click.argument("app_name") +@click.option("--env", "-e", "env_name", default="default", show_default=True) +@click.option("--dev", is_flag=True, help="Rotate the key for dev sessions.") +def app_rotate_key(app_name, env_name, dev): + """Rotate the api key that an app's workers use for outbound calls. + + Running workers keep the revoked key until they restart. + """ + from runpod.apps.api import AppsApiClient + from runpod.apps.app_key import rotate_app_key + from runpod.rp_cli import console as ui + + target_env = None if dev else env_name + try: + asyncio.run(rotate_app_key(AppsApiClient(), app_name, target_env)) + except Exception as exc: # noqa: BLE001 - surface engine errors cleanly + raise click.ClickException(str(exc)) from exc + label = f"{app_name} dev sessions" if dev else f"{app_name}/{env_name}" + ui.success(f"rotated the api key for [white]{label}[/white]") + + @cli.group() def registry(): """Manage container registry credentials (private images).""" diff --git a/tests/e2e/examples/07_pipelines.py b/tests/e2e/examples/07_pipelines.py index f01d61f83..bab1306eb 100644 --- a/tests/e2e/examples/07_pipelines.py +++ b/tests/e2e/examples/07_pipelines.py @@ -20,7 +20,7 @@ def count(tokens: list): return len(tokens) -@app.queue(cpu="cpu3c-1-2") +@app.queue(cpu="cpu3c-1-2", remote_calls=True) def pipeline(text: str): # nested .remote() calls run on the sibling resources tokens = tokenize.remote(text) diff --git a/tests/test_apps/test_dev.py b/tests/test_apps/test_dev.py index 52a82e7b0..1edf2941d 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_api_key_not_forwarded_without_remote_calls(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..f8d733cc2 --- /dev/null +++ b/tests/test_apps/test_worker_credentials.py @@ -0,0 +1,340 @@ +"""worker credentials: only resources that declare remote_calls get an api +key, minted per app environment, and templates only ever carry a secret +reference to it.""" + +import json +from unittest.mock import AsyncMock + +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, + AppKeyError, + app_key_name, + delete_app_key, + ensure_app_key, + rotate_app_key, +) +from runpod.apps.deploy import _deployed_endpoint_input, reconcile_endpoints +from runpod.apps.dev import DevSession, _endpoint_input +from runpod.apps.manage import delete_app, undeploy_environment +from runpod.apps.secret import Secret +from runpod.error import QueryError + +ACCOUNT_KEY = "rpa_fakeaccountkey0000000000000000" +APP_KEY = "rpa_fakeappkey00000000000000000000" +PROD_SECRET = "runpod-app-key.pipe.prod" +DEV_SECRET = "runpod-dev-key.pipe" + + +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", remote_calls=True) + 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")) + + +class TestEnsureAppKey: + 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}]) + 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_concurrent_creation_revokes_the_extra_key(self): + api = _api() + api.list_secrets.side_effect = [[], [{"id": "s1", "name": PROD_SECRET}]] + api.create_secret.side_effect = QueryError("Secret name must be unique.") + assert (await ensure_app_key(api, "pipe", "prod")).name == PROD_SECRET + api.delete_api_key.assert_awaited_once_with("key-new") + + 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() + + +class TestRotateAppKey: + async def test_new_key_replaces_secret_value_then_old_keys_are_revoked(self): + api = _api( + [{"id": "s1", "name": PROD_SECRET}], + [{"id": "key-old", "name": PROD_SECRET}, {"id": "other", "name": "mine"}], + ) + await rotate_app_key(api, "pipe", "prod") + api.update_secret_value.assert_awaited_once_with("s1", APP_KEY) + api.delete_api_key.assert_awaited_once_with("key-old") + + async def test_missing_secret_is_an_error(self): + api = _api() + with pytest.raises(AppKeyError): + await rotate_app_key(api, "pipe", "prod") + api.create_api_key.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"} + + +class TestDeployedEndpoints: + def test_undeclared_resource_gets_no_key(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", app_key=Secret(PROD_SECRET) + ) + assert "RUNPOD_API_KEY" not in _env(payload) + assert _no_key_values(payload) + + async def test_declared_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 "RUNPOD_API_KEY" not in _env(saved["leaf"]) + 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}]) + 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_no_declared_resource_never_touches_keys(self, logged_in): + app = App("solo") + + @app.queue(name="only", cpu="cpu3c-1-2") + def only(): + pass + + api = _api() + await reconcile_endpoints( + api, app, {"id": "env-1", "name": "default"}, "build-1", "3.12" + ) + assert "RUNPOD_API_KEY" not in _env(_saved(api)["only"]) + api.list_secrets.assert_not_awaited() + 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", + remote_calls=True, + 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.create_api_key.assert_not_awaited() + + +class TestDevEndpoints: + 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_only_for_declared(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 "RUNPOD_API_KEY" not in _env(saved["leaf"]) + assert _no_key_values([c.args for c in api.save_endpoint.await_args_list]) + api.create_api_key.assert_awaited_once() + + async def test_session_without_declared_never_touches_keys(self, logged_in): + app = App("solo") + + @app.queue(name="only", cpu="cpu3c-1-2") + def only(): + pass + + api = _api() + await DevSession([app], api=api).start() + api.list_secrets.assert_not_awaited() + api.create_api_key.assert_not_awaited() + + +class TestNestedCallsInWorker: + async def test_declared_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" + + async def test_undeclared_worker_fails_with_declaration_hint(self, monkeypatch): + monkeypatch.delenv("RUNPOD_API_KEY", raising=False) + monkeypatch.delenv("RUNPOD_DEV_APP", raising=False) + monkeypatch.setattr(runpod, "api_key", None) + monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-leaf") + app, leaf, _ = _pipeline_app() + target = await app._resolve(leaf.spec) + with pytest.raises(RuntimeError, match="remote_calls=True"): + target._sentinel_headers() From 912be194d05513c5d4ebe5e652b240ad2bf5b400 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:58:43 -0700 Subject: [PATCH 2/6] fix: mint the app key for every worker and store it through rest secrets --- runpod/api/mutations/apps.py | 18 ----- runpod/api/queries/apps.py | 8 -- runpod/apps/api.py | 42 +++++------ runpod/apps/app.py | 13 +--- runpod/apps/app_key.py | 39 ++-------- runpod/apps/deploy.py | 4 +- runpod/apps/dev.py | 7 +- runpod/apps/spec.py | 1 - runpod/apps/utils/network.py | 7 -- runpod/rp_cli/main.py | 22 ------ tests/e2e/examples/07_pipelines.py | 2 +- tests/test_apps/test_api_client.py | 24 ++++-- tests/test_apps/test_dev.py | 2 +- tests/test_apps/test_worker_credentials.py | 85 ++++------------------ 14 files changed, 66 insertions(+), 208 deletions(-) diff --git a/runpod/api/mutations/apps.py b/runpod/api/mutations/apps.py index ad864fbd3..1818cce6a 100644 --- a/runpod/api/mutations/apps.py +++ b/runpod/api/mutations/apps.py @@ -42,24 +42,6 @@ } """ -MUTATION_CREATE_SECRET = """ -mutation secretCreate($input: SecretCreateInput!) { - secretCreate(input: $input) { id name } -} -""" - -MUTATION_DELETE_SECRET = """ -mutation secretDelete($id: ID!) { - secretDelete(id: $id) -} -""" - -MUTATION_UPDATE_SECRET_VALUE = """ -mutation secretValueUpdate($input: SecretValueUpdateInput!) { - secretValueUpdate(input: $input) { id } -} -""" - MUTATION_CREATE_API_KEY = """ mutation createApiKeyNew($input: CreateApiKeyInput!) { createApiKeyNew(input: $input) { id rawKey } diff --git a/runpod/api/queries/apps.py b/runpod/api/queries/apps.py index 9ddc78261..431cb608d 100644 --- a/runpod/api/queries/apps.py +++ b/runpod/api/queries/apps.py @@ -25,14 +25,6 @@ } """ -QUERY_SECRETS = """ -query mySecrets { - myself { - secrets { id name description createdAt } - } -} -""" - QUERY_API_KEYS = """ query myApiKeys { myself { diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 1ca3af5d5..43c079a82 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -478,35 +478,33 @@ 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).""" + 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"] 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")) - - async def update_secret_value(self, secret_id: str, value: str) -> None: - mutation = app_mutations.MUTATION_UPDATE_SECRET_VALUE - await self._execute(mutation, {"input": {"id": secret_id, "value": value}}) + 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 diff --git a/runpod/apps/app.py b/runpod/apps/app.py index 144d11b4b..c2a6a0e5d 100644 --- a/runpod/apps/app.py +++ b/runpod/apps/app.py @@ -98,13 +98,8 @@ def queue( min_cuda_version: Optional[str] = None, accelerate_downloads: bool = True, container_disk_gb: Optional[int] = None, - remote_calls: bool = False, ) -> Callable[[Callable], FunctionHandle]: - """declare a queue-based serverless endpoint from a function. - - set remote_calls=True when the function calls other resources with - .remote(); only then do its workers receive api credentials. - """ + """declare a queue-based serverless endpoint from a function.""" def decorator(fn: Callable) -> FunctionHandle: spec = ResourceSpec( @@ -131,7 +126,6 @@ def decorator(fn: Callable) -> FunctionHandle: min_cuda_version=min_cuda_version, accelerate_downloads=accelerate_downloads, container_disk_gb=container_disk_gb, - remote_calls=remote_calls, ) handle = FunctionHandle(self, fn, spec) self._register(spec.name, handle) @@ -207,7 +201,6 @@ def api( min_cuda_version: Optional[str] = None, accelerate_downloads: bool = True, container_disk_gb: Optional[int] = None, - remote_calls: bool = False, ) -> Callable[[Any], ApiHandle]: """declare a load-balanced serverless endpoint. @@ -229,9 +222,6 @@ def web(): server = FastAPI() ... return server - - set remote_calls=True when the service calls other resources with - .remote(); only then do its workers receive api credentials. """ def decorator(target: Any) -> ApiHandle: @@ -258,7 +248,6 @@ def decorator(target: Any) -> ApiHandle: min_cuda_version=min_cuda_version, accelerate_downloads=accelerate_downloads, container_disk_gb=container_disk_gb, - remote_calls=remote_calls, ) handle = ApiHandle(self, target, spec) self._register(spec.name, handle) diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py index baaa6550a..b285ba16e 100644 --- a/runpod/apps/app_key.py +++ b/runpod/apps/app_key.py @@ -1,7 +1,7 @@ -"""per-app api keys for workers that make outbound calls. +"""per-app api keys for workers. -a resource that declares remote_calls gets RUNPOD_API_KEY from a platform -secret holding an api key minted for its app and environment. the key +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. @@ -14,7 +14,6 @@ from typing import Any, Dict, List, Optional from ..error import QueryError -from .errors import AppError from .secret import Secret APP_KEY_ENV = "RUNPOD_API_KEY" @@ -37,17 +36,13 @@ _UNSAFE = re.compile(r"[^A-Za-z0-9_.-]") -class AppKeyError(AppError): - pass - - 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"runpod-dev-key.{app_name}" + raw = f"flash-dev-key.{app_name}" else: - raw = f"runpod-app-key.{app_name}.{env_name}" + raw = f"flash-app-key.{app_name}.{env_name}" name = _UNSAFE.sub("-", raw) if len(name) <= _MAX_NAME: return name @@ -56,13 +51,12 @@ def app_key_name(app_name: str, env_name: Optional[str] = None) -> str: def needs_app_key(spec) -> bool: - """true when a resource makes outbound calls and does not supply its - own credentials through env.""" - return spec.remote_calls and APP_KEY_ENV not in (spec.env or {}) + """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(): + for secret in await api.list_secrets(name=name): if secret.get("name") == name: return secret return None @@ -92,23 +86,6 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> return Secret(name) -async def rotate_app_key(api, app_name: str, env_name: Optional[str] = None) -> None: - """mint a new key into the existing secret and revoke the old ones. - - workers read the secret when they boot, so running workers keep the - revoked key until they restart. - """ - name = app_key_name(app_name, env_name) - secret = await _find_secret(api, name) - if secret is None: - raise AppKeyError(f"no app key '{name}' to rotate") - stale = await _key_ids(api, name) - key = await api.create_api_key(name, APP_KEY_SCOPES) - await api.update_secret_value(secret["id"], key["rawKey"]) - for key_id in stale: - await api.delete_api_key(key_id) - - 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.""" diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index dff289cc7..61b01063e 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -294,8 +294,8 @@ def _deployed_endpoint_input( 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 only for resources that - declare remote_calls; every other worker boots without credentials. + 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 diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index fa97f2c45..d78c234c4 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -100,8 +100,8 @@ def _endpoint_input( ) -> Dict: """build a live endpoint payload with an atomically bound container config. - app_key is referenced as RUNPOD_API_KEY only for resources that - declare remote_calls; every other worker boots without credentials. + app_key is referenced as RUNPOD_API_KEY unless the resource's env + sets its own. """ spec.validate() @@ -231,8 +231,7 @@ async def _attach_volumes(self, payload: Dict, spec, app) -> None: 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 and only - when a resource declares remote_calls.""" + """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: diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index 327859502..a6a41373d 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -190,7 +190,6 @@ class ResourceSpec: container_disk_gb: Optional[int] = None routes: List[RouteSpec] = field(default_factory=list) asgi_factory: Optional[str] = None - remote_calls: bool = False def __post_init__(self) -> None: self.validate() diff --git a/runpod/apps/utils/network.py b/runpod/apps/utils/network.py index 54df2fe48..d681766ea 100644 --- a/runpod/apps/utils/network.py +++ b/runpod/apps/utils/network.py @@ -13,13 +13,6 @@ def api_key() -> str: key = runpod.api_key if not key: - from ..context import Context, current_context - - if current_context() is Context.WORKER: - raise RuntimeError( - "no api key in this worker. declare remote_calls=True on the " - "resource that makes .remote() calls." - ) raise RuntimeError( "no api key configured. run `rp login` or set RUNPOD_API_KEY." ) diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 359c24b78..5c4b4bfdc 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -673,28 +673,6 @@ def app_delete(app_name, yes): ) -@app.command(name="rotate-key") -@click.argument("app_name") -@click.option("--env", "-e", "env_name", default="default", show_default=True) -@click.option("--dev", is_flag=True, help="Rotate the key for dev sessions.") -def app_rotate_key(app_name, env_name, dev): - """Rotate the api key that an app's workers use for outbound calls. - - Running workers keep the revoked key until they restart. - """ - from runpod.apps.api import AppsApiClient - from runpod.apps.app_key import rotate_app_key - from runpod.rp_cli import console as ui - - target_env = None if dev else env_name - try: - asyncio.run(rotate_app_key(AppsApiClient(), app_name, target_env)) - except Exception as exc: # noqa: BLE001 - surface engine errors cleanly - raise click.ClickException(str(exc)) from exc - label = f"{app_name} dev sessions" if dev else f"{app_name}/{env_name}" - ui.success(f"rotated the api key for [white]{label}[/white]") - - @cli.group() def registry(): """Manage container registry credentials (private images).""" diff --git a/tests/e2e/examples/07_pipelines.py b/tests/e2e/examples/07_pipelines.py index bab1306eb..f01d61f83 100644 --- a/tests/e2e/examples/07_pipelines.py +++ b/tests/e2e/examples/07_pipelines.py @@ -20,7 +20,7 @@ def count(tokens: list): return len(tokens) -@app.queue(cpu="cpu3c-1-2", remote_calls=True) +@app.queue(cpu="cpu3c-1-2") def pipeline(text: str): # nested .remote() calls run on the sibling resources tokens = tokenize.remote(text) diff --git a/tests/test_apps/test_api_client.py b/tests/test_apps/test_api_client.py index d3ea81e9a..51b389e90 100644 --- a/tests/test_apps/test_api_client.py +++ b/tests/test_apps/test_api_client.py @@ -704,17 +704,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 1edf2941d..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_not_forwarded_without_remote_calls(self, monkeypatch): + def test_account_api_key_never_forwarded(self, monkeypatch): monkeypatch.setenv("RUNPOD_API_KEY", "rpa_test123") app = App("keyed") diff --git a/tests/test_apps/test_worker_credentials.py b/tests/test_apps/test_worker_credentials.py index f8d733cc2..8c8dfe29c 100644 --- a/tests/test_apps/test_worker_credentials.py +++ b/tests/test_apps/test_worker_credentials.py @@ -1,6 +1,5 @@ -"""worker credentials: only resources that declare remote_calls get an api -key, minted per app environment, and templates only ever carry a secret -reference to it.""" +"""worker credentials: workers get an api key minted per app environment, +and templates only ever carry a secret reference to it.""" import json from unittest.mock import AsyncMock @@ -12,11 +11,9 @@ from runpod.apps.app import _clear_registry from runpod.apps.app_key import ( APP_KEY_SCOPES, - AppKeyError, app_key_name, delete_app_key, ensure_app_key, - rotate_app_key, ) from runpod.apps.deploy import _deployed_endpoint_input, reconcile_endpoints from runpod.apps.dev import DevSession, _endpoint_input @@ -26,8 +23,8 @@ ACCOUNT_KEY = "rpa_fakeaccountkey0000000000000000" APP_KEY = "rpa_fakeappkey00000000000000000000" -PROD_SECRET = "runpod-app-key.pipe.prod" -DEV_SECRET = "runpod-dev-key.pipe" +PROD_SECRET = "flash-app-key.pipe.prod" +DEV_SECRET = "flash-dev-key.pipe" def _reference(name): @@ -65,7 +62,7 @@ def _pipeline_app(): def leaf(x: int): return x + 1 - @app.queue(name="root", cpu="cpu3c-1-2", remote_calls=True) + @app.queue(name="root", cpu="cpu3c-1-2") def root(x: int): return leaf.remote(x) @@ -142,23 +139,6 @@ async def test_key_creation_failure_propagates_before_any_secret(self): api.create_secret.assert_not_awaited() -class TestRotateAppKey: - async def test_new_key_replaces_secret_value_then_old_keys_are_revoked(self): - api = _api( - [{"id": "s1", "name": PROD_SECRET}], - [{"id": "key-old", "name": PROD_SECRET}, {"id": "other", "name": "mine"}], - ) - await rotate_app_key(api, "pipe", "prod") - api.update_secret_value.assert_awaited_once_with("s1", APP_KEY) - api.delete_api_key.assert_awaited_once_with("key-old") - - async def test_missing_secret_is_an_error(self): - api = _api() - with pytest.raises(AppKeyError): - await rotate_app_key(api, "pipe", "prod") - api.create_api_key.assert_not_awaited() - - class TestDeleteAppKey: async def test_removes_secret_and_every_key_with_its_name(self): api = _api( @@ -206,16 +186,14 @@ async def test_app_deletion_removes_environment_and_dev_keys(self): class TestDeployedEndpoints: - def test_undeclared_resource_gets_no_key(self, logged_in, monkeypatch): + 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", app_key=Secret(PROD_SECRET) - ) + 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_declared_resource_references_environment_key(self, logged_in): + async def test_every_resource_references_environment_key(self, logged_in): app, _, _ = _pipeline_app() api = _api() await reconcile_endpoints( @@ -223,7 +201,7 @@ async def test_declared_resource_references_environment_key(self, logged_in): ) saved = _saved(api) assert _env(saved["root"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) - assert "RUNPOD_API_KEY" not in _env(saved["leaf"]) + assert _env(saved["leaf"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) assert _no_key_values(*saved.values()) api.create_api_key.assert_awaited_once() @@ -236,28 +214,12 @@ async def test_redeploy_reuses_the_environment_key(self, logged_in): assert _env(_saved(api)["root"])["RUNPOD_API_KEY"] == _reference(PROD_SECRET) api.create_api_key.assert_not_awaited() - async def test_no_declared_resource_never_touches_keys(self, logged_in): - app = App("solo") - - @app.queue(name="only", cpu="cpu3c-1-2") - def only(): - pass - - api = _api() - await reconcile_endpoints( - api, app, {"id": "env-1", "name": "default"}, "build-1", "3.12" - ) - assert "RUNPOD_API_KEY" not in _env(_saved(api)["only"]) - api.list_secrets.assert_not_awaited() - 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", - remote_calls=True, env={"RUNPOD_API_KEY": Secret("my-scoped-key")}, ) def caller(): @@ -269,6 +231,7 @@ def caller(): ) 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() @@ -279,7 +242,7 @@ def test_payload_needs_no_credentials(self, monkeypatch): app, _, root = _pipeline_app() assert "RUNPOD_API_KEY" not in _env(_endpoint_input(app, root.spec)) - async def test_session_references_dev_key_only_for_declared(self, logged_in): + async def test_session_references_dev_key(self, logged_in): app, _, _ = _pipeline_app() api = _api() session = DevSession([app], api=api) @@ -290,25 +253,13 @@ async def test_session_references_dev_key_only_for_declared(self, logged_in): for call in api.save_endpoint.await_args_list } assert _env(saved["root"])["RUNPOD_API_KEY"] == _reference(DEV_SECRET) - assert "RUNPOD_API_KEY" not in _env(saved["leaf"]) + 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() - async def test_session_without_declared_never_touches_keys(self, logged_in): - app = App("solo") - - @app.queue(name="only", cpu="cpu3c-1-2") - def only(): - pass - - api = _api() - await DevSession([app], api=api).start() - api.list_secrets.assert_not_awaited() - api.create_api_key.assert_not_awaited() - class TestNestedCallsInWorker: - async def test_declared_worker_authenticates_with_substituted_key( + async def test_worker_authenticates_with_substituted_key( self, logged_in, monkeypatch ): app, leaf, _ = _pipeline_app() @@ -328,13 +279,3 @@ async def test_declared_worker_authenticates_with_substituted_key( assert headers["Authorization"] == f"Bearer {APP_KEY}" assert headers["X-Flash-Endpoint"] == "leaf" assert headers["X-Flash-Environment"] == "prod" - - async def test_undeclared_worker_fails_with_declaration_hint(self, monkeypatch): - monkeypatch.delenv("RUNPOD_API_KEY", raising=False) - monkeypatch.delenv("RUNPOD_DEV_APP", raising=False) - monkeypatch.setattr(runpod, "api_key", None) - monkeypatch.setenv("RUNPOD_ENDPOINT_ID", "ep-leaf") - app, leaf, _ = _pipeline_app() - target = await app._resolve(leaf.spec) - with pytest.raises(RuntimeError, match="remote_calls=True"): - target._sentinel_headers() From 93d3c19b926060f04ca634243b259c2157917cba Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:54:32 -0700 Subject: [PATCH 3/6] fix: harden app credential lifecycle --- runpod/apps/api.py | 21 +- runpod/apps/app_key.py | 92 ++++- runpod/apps/dev.py | 15 +- runpod/apps/manage.py | 30 +- tests/test_apps/test_api_client.py | 62 +++- tests/test_apps/test_worker_credentials.py | 395 ++++++++++++++++++++- 6 files changed, 567 insertions(+), 48 deletions(-) diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 43c079a82..bde8fa7f7 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -480,12 +480,21 @@ async def delete_registry_auth(self, auth_id: str) -> bool: async def list_secrets(self, name: Optional[str] = None) -> List[Dict[str, Any]]: """account secrets, or only the one matching name (case-insensitive).""" - data = await run_rest_request_async( - "GET", - "/v2/account/secrets", - api_key=self._api_key, - params={"name": name} if name else None, - ) + import asyncio + + for attempt in range(_TRANSPORT_RETRIES): + try: + data = await run_rest_request_async( + "GET", + "/v2/account/secrets", + api_key=self._api_key, + params={"name": name} if name else None, + ) + break + except (aiohttp.ClientError, OSError, asyncio.TimeoutError): + if attempt == _TRANSPORT_RETRIES - 1: + raise + await asyncio.sleep(2**attempt) return data["secrets"] async def create_secret( diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py index b285ba16e..1f14b346f 100644 --- a/runpod/apps/app_key.py +++ b/runpod/apps/app_key.py @@ -10,13 +10,16 @@ """ import hashlib +import json +import logging import re from typing import Any, Dict, List, Optional -from ..error import QueryError +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 @@ -43,10 +46,9 @@ def app_key_name(app_name: str, env_name: Optional[str] = None) -> str: raw = f"flash-dev-key.{app_name}" else: raw = f"flash-app-key.{app_name}.{env_name}" - name = _UNSAFE.sub("-", raw) - if len(name) <= _MAX_NAME: - return name - digest = hashlib.sha256(raw.encode()).hexdigest()[:12] + 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}" @@ -66,23 +68,83 @@ 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.""" name = app_key_name(app_name, env_name) - if await _find_secret(api, name): - return Secret(name) - key = await api.create_api_key(name, APP_KEY_SCOPES) - target = f"{app_name}/{env_name}" if env_name else f"{app_name} dev sessions" + secret = await _find_secret(api, name) + if secret: + if await _secret_has_live_key(api, name, secret): + return Secret(name) + await api.delete_secret(secret["id"]) try: - await api.create_secret( - name, key["rawKey"], description=f"api key for {target} workers" + key = await api.create_api_key(name, APP_KEY_SCOPES) + except (AuthenticationError, QueryError) as exc: + 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." ) - except QueryError: - await api.delete_api_key(key["id"]) - # a concurrent deploy may have created it first - if not await _find_secret(api, name): + 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 Exception as create_error: + try: + secret = await _find_secret(api, name) + except Exception: + log.warning( + "could not reconcile secret %s; retained api key %s for cleanup", + name, + key["id"], + ) + 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 api key %s for cleanup", + name, + key["id"], + ) raise + if not owns_secret: + try: + await api.delete_api_key(key["id"]) + except Exception: + log.warning( + "could not revoke api key %s for secret %s; retry credential cleanup", + key["id"], + name, + ) + if secret: + 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) diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index d78c234c4..593e05d0a 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -12,7 +12,7 @@ from .api import AppsApiClient from .app import App -from .app_key import APP_KEY_ENV, ensure_app_key, needs_app_key +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 @@ -403,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), @@ -435,6 +435,17 @@ 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: + failures.append(f"credentials for {app_name}") + log.warning("failed to delete dev credentials for %s", app_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() diff --git a/runpod/apps/manage.py b/runpod/apps/manage.py index ff3154e89..dd9f01e73 100644 --- a/runpod/apps/manage.py +++ b/runpod/apps/manage.py @@ -76,8 +76,8 @@ async def undeploy_environment( delete_env: bool = True, events: Optional[object] = None, ) -> UndeployResult: - """tear down an environment: endpoints first, then the environment - and its app key. + """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). @@ -100,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"]) @@ -108,12 +114,6 @@ async def undeploy_environment( except Exception as exc: # noqa: BLE001 - collected for the caller result.failures.append(f"environment {env_name}: {exc}") - if result.environment_deleted: - try: - await delete_app_key(client, app_name, env_name) - except Exception as exc: # noqa: BLE001 - collected for the caller - result.failures.append(f"app key for {env_name}: {exc}") - return result @@ -123,7 +123,7 @@ async def delete_app( api: Optional[AppsApiClient] = None, events: Optional[object] = None, ) -> UndeployResult: - """delete an app after undeploying every environment in it, then its + """delete an app after undeploying its environments and removing its dev key.""" client = api or AppsApiClient() app = await get_app(app_name, api=client) @@ -136,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"]) @@ -144,10 +150,4 @@ async def delete_app( except Exception as exc: # noqa: BLE001 - collected for the caller result.failures.append(f"app {app_name}: {exc}") - if result.app_deleted: - try: - await delete_app_key(client, app_name) - except Exception as exc: # noqa: BLE001 - collected for the caller - result.failures.append(f"dev key for {app_name}: {exc}") - return result diff --git a/tests/test_apps/test_api_client.py b/tests/test_apps/test_api_client.py index 51b389e90..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}) diff --git a/tests/test_apps/test_worker_credentials.py b/tests/test_apps/test_worker_credentials.py index 8c8dfe29c..1c41b116a 100644 --- a/tests/test_apps/test_worker_credentials.py +++ b/tests/test_apps/test_worker_credentials.py @@ -2,8 +2,9 @@ and templates only ever carry a secret reference to it.""" import json -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, call +import aiohttp import pytest import runpod @@ -17,14 +18,15 @@ ) 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 QueryError +from runpod.error import AuthenticationError, QueryError ACCOUNT_KEY = "rpa_fakeaccountkey0000000000000000" APP_KEY = "rpa_fakeappkey00000000000000000000" -PROD_SECRET = "flash-app-key.pipe.prod" -DEV_SECRET = "flash-dev-key.pipe" +PROD_SECRET = "flash-app-key.pipe.prod.ab680dbced7cc40bff0314e6" +DEV_SECRET = "flash-dev-key.pipe.8afde8d00cfa7b4458c20c7d" def _reference(name): @@ -100,6 +102,25 @@ def test_long_names_fit_api_key_limit_and_stay_distinct(self): 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: async def test_mints_scoped_key_into_secret(self): @@ -112,17 +133,241 @@ async def test_mints_scoped_key_into_secret(self): assert APP_KEY not in args.kwargs["description"] async def test_reuses_existing_secret_without_minting(self): - api = _api([{"id": "s1", "name": PROD_SECRET}]) + 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() + 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}]] - api.create_secret.side_effect = QueryError("Secret name must be unique.") + 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 key-new" in caplog.text + assert APP_KEY not in caplog.text async def test_secret_failure_revokes_key_and_propagates(self): api = _api() @@ -138,6 +383,23 @@ async def test_key_creation_failure_propagates_before_any_secret(self): 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): @@ -184,6 +446,55 @@ async def test_app_deletion_removes_environment_and_dev_keys(self): 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): @@ -207,7 +518,10 @@ async def test_every_resource_references_environment_key(self, logged_in): async def test_redeploy_reuses_the_environment_key(self, logged_in): app, _, _ = _pipeline_app() - api = _api([{"id": "s1", "name": PROD_SECRET}]) + 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" ) @@ -236,6 +550,71 @@ def caller(): class TestDevEndpoints: + + 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) From dca25410e82ef98ae31c262d86d4216136790980 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:14:48 -0700 Subject: [PATCH 4/6] fix: keep credential response fields out of cleanup logs --- runpod/apps/api.py | 3 +-- runpod/apps/app_key.py | 9 +++------ tests/test_apps/test_worker_credentials.py | 19 ++++++++++++++++++- 3 files changed, 22 insertions(+), 9 deletions(-) diff --git a/runpod/apps/api.py b/runpod/apps/api.py index bde8fa7f7..0f20ade25 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -490,12 +490,11 @@ async def list_secrets(self, name: Optional[str] = None) -> List[Dict[str, Any]] api_key=self._api_key, params={"name": name} if name else None, ) - break + return data["secrets"] except (aiohttp.ClientError, OSError, asyncio.TimeoutError): if attempt == _TRANSPORT_RETRIES - 1: raise await asyncio.sleep(2**attempt) - return data["secrets"] async def create_secret( self, name: str, value: str, description: str = "" diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py index 1f14b346f..cae3aa7c7 100644 --- a/runpod/apps/app_key.py +++ b/runpod/apps/app_key.py @@ -107,9 +107,8 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> secret = await _find_secret(api, name) except Exception: log.warning( - "could not reconcile secret %s; retained api key %s for cleanup", + "could not reconcile secret %s; retained associated api key for cleanup", name, - key["id"], ) raise create_error from None owns_secret = secret and secret.get("description") == description @@ -121,9 +120,8 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> ) ): log.warning( - "secret %s has no ownership metadata; retained api key %s for cleanup", + "secret %s has no ownership metadata; retained associated api key for cleanup", name, - key["id"], ) raise if not owns_secret: @@ -131,8 +129,7 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> await api.delete_api_key(key["id"]) except Exception: log.warning( - "could not revoke api key %s for secret %s; retry credential cleanup", - key["id"], + "could not revoke api key for secret %s; retry credential cleanup", name, ) if secret: diff --git a/tests/test_apps/test_worker_credentials.py b/tests/test_apps/test_worker_credentials.py index 1c41b116a..2e1463ff5 100644 --- a/tests/test_apps/test_worker_credentials.py +++ b/tests/test_apps/test_worker_credentials.py @@ -366,7 +366,24 @@ async def test_revoke_failure_does_not_mask_creation_result(self, winner, caplog 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 key-new" in caplog.text + 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): From cb063dea96695470392be7dbe57d8b06b4f4cd9b Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:34:46 -0700 Subject: [PATCH 5/6] fix: make secret retry termination explicit --- runpod/apps/api.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 0f20ade25..b0dc898cd 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -482,7 +482,8 @@ 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 - for attempt in range(_TRANSPORT_RETRIES): + attempt = 0 + while True: try: data = await run_rest_request_async( "GET", @@ -495,6 +496,7 @@ async def list_secrets(self, name: Optional[str] = None) -> List[Dict[str, Any]] if attempt == _TRANSPORT_RETRIES - 1: raise await asyncio.sleep(2**attempt) + attempt += 1 async def create_secret( self, name: str, value: str, description: str = "" From f8a8fd25f0a9534c474c8c17369545a972ce354e Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:35:40 -0700 Subject: [PATCH 6/6] fix: preserve worker credential cleanup on interruption --- runpod/apps/app_key.py | 30 ++- runpod/apps/dev.py | 18 +- tests/test_apps/test_worker_credentials.py | 229 +++++++++++++++++++++ 3 files changed, 269 insertions(+), 8 deletions(-) diff --git a/runpod/apps/app_key.py b/runpod/apps/app_key.py index cae3aa7c7..29b8512d0 100644 --- a/runpod/apps/app_key.py +++ b/runpod/apps/app_key.py @@ -80,16 +80,38 @@ async def _secret_has_live_key( 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.""" + 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) - await api.delete_secret(secret["id"]) + 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: " @@ -102,7 +124,7 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> description = f"api key {key['id']} for {target} workers" try: await api.create_secret(name, key["rawKey"], description=description) - except Exception as create_error: + except BaseException as create_error: try: secret = await _find_secret(api, name) except Exception: @@ -132,7 +154,7 @@ async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> "could not revoke api key for secret %s; retry credential cleanup", name, ) - if secret: + if secret and isinstance(create_error, Exception): try: if await _secret_has_live_key( api, name, secret, None if owns_secret else key["id"] diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index 593e05d0a..39de51080 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -438,9 +438,13 @@ async def stop(self, events: Optional[object] = None) -> None: for app_name in list(self._app_keys): try: await delete_app_key(self.api, app_name) - except Exception: - failures.append(f"credentials for {app_name}") - log.warning("failed to delete dev credentials for %s", 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: @@ -450,7 +454,13 @@ 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/tests/test_apps/test_worker_credentials.py b/tests/test_apps/test_worker_credentials.py index 2e1463ff5..17c8ee288 100644 --- a/tests/test_apps/test_worker_credentials.py +++ b/tests/test_apps/test_worker_credentials.py @@ -1,6 +1,7 @@ """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 @@ -123,6 +124,187 @@ def test_original_identities_are_distinct_case_insensitively(self, left, right): 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") @@ -568,6 +750,53 @@ def caller(): 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(