fix: give workers a per-app key stored as a secret

runpod/api/mutations/apps.py

@@ -42,15 +42,15 @@
 }
 """
 
-MUTATION_CREATE_SECRET = """
-mutation secretCreate($input: SecretCreateInput!) {
-    secretCreate(input: $input) { id name }
+MUTATION_CREATE_API_KEY = """
+mutation createApiKeyNew($input: CreateApiKeyInput!) {
+    createApiKeyNew(input: $input) { id rawKey }
 }
 """
 
-MUTATION_DELETE_SECRET = """
-mutation secretDelete($id: ID!) {
-    secretDelete(id: $id)
+MUTATION_DELETE_API_KEY = """
+mutation deleteApiKey($input: DeleteApiKeyInput!) {
+    deleteApiKey(input: $input) { count }
 }
 """
 

runpod/api/queries/apps.py

@@ -25,10 +25,10 @@
 }
 """
 
-QUERY_SECRETS = """
-query mySecrets {
+QUERY_API_KEYS = """
+query myApiKeys {
     myself {
-        secrets { id name description createdAt }
+        apiKeys { id name }
     }
 }
 """

runpod/apps/api.py

@@ -478,31 +478,60 @@ async def delete_registry_auth(self, auth_id: str) -> bool:
         )
         return True
 
-    async def list_secrets(self) -> List[Dict[str, Any]]:
-        query = app_queries.QUERY_SECRETS
-        data = await self._execute(query, retry=True)
-        return data["myself"].get("secrets") or []
+    async def list_secrets(self, name: Optional[str] = None) -> List[Dict[str, Any]]:
+        """account secrets, or only the one matching name (case-insensitive)."""
+        import asyncio
+
+        attempt = 0
+        while True:
+            try:
+                data = await run_rest_request_async(
+                    "GET",
+                    "/v2/account/secrets",
+                    api_key=self._api_key,
+                    params={"name": name} if name else None,
+                )
+                return data["secrets"]
+            except (aiohttp.ClientError, OSError, asyncio.TimeoutError):
+                if attempt == _TRANSPORT_RETRIES - 1:
+                    raise
+                await asyncio.sleep(2**attempt)
+                attempt += 1
 
     async def create_secret(
         self, name: str, value: str, description: str = ""
     ) -> Dict[str, Any]:
-        mutation = app_mutations.MUTATION_CREATE_SECRET
-        data = await self._execute(
-            mutation,
-            {
-                "input": {
-                    "name": name,
-                    "value": value,
-                    "description": description,
-                }
-            },
+        body = {"name": name, "value": value}
+        if description:
+            body["description"] = description
+        return await run_rest_request_async(
+            "POST", "/v2/account/secrets", api_key=self._api_key, json=body
         )
-        return data["secretCreate"]
 
     async def delete_secret(self, secret_id: str) -> bool:
-        mutation = app_mutations.MUTATION_DELETE_SECRET
-        data = await self._execute(mutation, {"id": secret_id})
-        return bool(data.get("secretDelete"))
+        await run_rest_request_async(
+            "DELETE",
+            f"/v2/account/secrets/{_path_segment(secret_id)}",
+            api_key=self._api_key,
+        )
+        return True
+
+    async def list_api_keys(self) -> List[Dict[str, Any]]:
+        query = app_queries.QUERY_API_KEYS
+        data = await self._execute(query, retry=True)
+        return data["myself"].get("apiKeys") or []
+
+    async def create_api_key(self, name: str, scopes: List[str]) -> Dict[str, Any]:
+        """create a scoped api key; the result carries rawKey, shown once."""
+        mutation = app_mutations.MUTATION_CREATE_API_KEY
+        data = await self._execute(
+            mutation, {"input": {"name": name, "scopes": scopes}}
+        )
+        return data["createApiKeyNew"]
+
+    async def delete_api_key(self, key_id: str) -> None:
+        mutation = app_mutations.MUTATION_DELETE_API_KEY
+        await self._execute(mutation, {"input": {"id": key_id}})
 
     async def list_apps(self) -> List[Dict[str, Any]]:
         """all flash apps with their environments and builds."""

runpod/apps/app_key.py

@@ -0,0 +1,180 @@
+"""per-app api keys for workers.
+
+every resource gets RUNPOD_API_KEY from a platform secret holding an api
+key minted for its app and environment, unless its env sets one. the key
+value travels only from key creation into the secret; templates carry the
+secret reference and the platform decrypts it when a worker boots.
+
+one key serves an app environment across deploys. dev sessions share a
+separate key per app.
+"""
+
+import hashlib
+import json
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+from ..error import AuthenticationError, QueryError
+from .secret import Secret
+
+APP_KEY_ENV = "RUNPOD_API_KEY"
+log = logging.getLogger(__name__)
+
+# what flash runtime does from a worker: call endpoints and list them
+# (serverless:read), launch and terminate task pods (pods), resolve
+# task mounts (volumes), and resolve registry credentials and secret
+# references for task pods (secrets:read)
+APP_KEY_SCOPES = [
+    "pods:read",
+    "pods:write",
+    "secrets:read",
+    "serverless:read",
+    "volumes:read",
+    "volumes:write",
+]
+
+# api key names are limited to 64 characters
+_MAX_NAME = 64
+_UNSAFE = re.compile(r"[^A-Za-z0-9_.-]")
+
+
+def app_key_name(app_name: str, env_name: Optional[str] = None) -> str:
+    """secret and api key name for an app environment, or for the app's
+    dev sessions when env_name is None."""
+    if env_name is None:
+        raw = f"flash-dev-key.{app_name}"
+    else:
+        raw = f"flash-app-key.{app_name}.{env_name}"
+    name = _UNSAFE.sub("-", raw).lower()
+    identity = json.dumps([app_name, env_name], ensure_ascii=True)
+    digest = hashlib.sha256(identity.encode()).hexdigest()[:24]
+    return f"{name[: _MAX_NAME - len(digest) - 1]}.{digest}"
+
+
+def needs_app_key(spec) -> bool:
+    """true when a resource does not supply its own credentials through env."""
+    return APP_KEY_ENV not in (spec.env or {})
+
+
+async def _find_secret(api, name: str) -> Optional[Dict[str, Any]]:
+    for secret in await api.list_secrets(name=name):
+        if secret.get("name") == name:
+            return secret
+    return None
+
+
+async def _key_ids(api, name: str) -> List[str]:
+    return [k["id"] for k in await api.list_api_keys() if k.get("name") == name]
+
+
+async def _secret_has_live_key(
+    api, name: str, secret: Dict[str, Any], exclude_key_id: Optional[str] = None
+) -> bool:
+    key_ids = [
+        key_id for key_id in await _key_ids(api, name) if key_id != exclude_key_id
+    ]
+    owner = re.match(r"api key (\S+) for ", secret.get("description") or "")
+    return owner[1] in key_ids if owner else bool(key_ids)
+
+
+async def ensure_app_key(api, app_name: str, env_name: Optional[str] = None) -> Secret:
+    """the secret holding the app environment's api key, minting the key
+    and the secret on first use. refresh preserves other same-name keys for
+    concurrent deployments; teardown revokes every matching key."""
+    name = app_key_name(app_name, env_name)
+    secret = await _find_secret(api, name)
+    if secret:
+        if await _secret_has_live_key(api, name, secret):
+            return Secret(name)
+        try:
+            await api.delete_secret(secret["id"])
+        except QueryError as exc:
+            if exc.status_code != 404:
+                raise
+    try:
+        key = await api.create_api_key(name, APP_KEY_SCOPES)
+    except (AuthenticationError, QueryError) as exc:
+        if not (
+            isinstance(exc, AuthenticationError)
+            or exc.status_code in (401, 403)
+            or (
+                exc.status_code in (None, 400)
+                and any(
+                    reason in str(exc).casefold()
+                    for reason in (
+                        "not grantable",
+                        "permission",
+                        "unauthorized",
+                        "forbidden",
+                    )
+                )
+            )
+        ):
+            raise
+        exc.message = (
+            f"{exc}. Automatic worker credentials require API key creation "
+            "permission and permission to grant: "
+            + ", ".join(APP_KEY_SCOPES)
+            + ". Use a caller with those permissions, or set RUNPOD_API_KEY "
+            "in each resource's env to an appropriate scoped key or Secret."
+        )
+        raise
+    target = f"{app_name}/{env_name}" if env_name else f"{app_name} dev sessions"
+    description = f"api key {key['id']} for {target} workers"
+    try:
+        await api.create_secret(name, key["rawKey"], description=description)
+    except BaseException as create_error:
+        try:
+            secret = await _find_secret(api, name)
+        except Exception:
+            log.warning(
+                "could not reconcile secret %s; retained associated api key for cleanup",
+                name,
+            )
+            raise create_error from None
+        owns_secret = secret and secret.get("description") == description
+        if (
+            secret
+            and not secret.get("description")
+            and not (
+                isinstance(create_error, QueryError) and create_error.status_code == 409
+            )
+        ):
+            log.warning(
+                "secret %s has no ownership metadata; retained associated api key for cleanup",
+                name,
+            )
+            raise
+        if not owns_secret:
+            try:
+                await api.delete_api_key(key["id"])
+            except Exception:
+                log.warning(
+                    "could not revoke api key for secret %s; retry credential cleanup",
+                    name,
+                )
+        if secret and isinstance(create_error, Exception):
+            try:
+                if await _secret_has_live_key(
+                    api, name, secret, None if owns_secret else key["id"]
+                ):
+                    return Secret(name)
+            except Exception:
+                log.warning("could not validate the api key for secret %s", name)
+                raise create_error from None
+        raise
+    return Secret(name)
+
+
+async def delete_app_key(api, app_name: str, env_name: Optional[str] = None) -> bool:
+    """remove the app environment's secret and revoke its keys. returns
+    whether anything existed."""
+    name = app_key_name(app_name, env_name)
+    secret = await _find_secret(api, name)
+    if secret is not None:
+        await api.delete_secret(secret["id"])
+    key_ids = await _key_ids(api, name)
+    for key_id in key_ids:
+        await api.delete_api_key(key_id)
+    return secret is not None or bool(key_ids)

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,19 +285,21 @@ 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.
 
     the endpoint name must equal the resource name: sentinel
     resolution matches X-Flash-Endpoint against endpoint.name within
     the environment. binding via flashEnvironmentId is also what makes
     hosts deliver the build artifact to this endpoint's workers.
+
+    app_key is referenced as RUNPOD_API_KEY unless the resource's env
+    sets its own.
     """
     from .datacenter import CPU3_DATACENTERS, CPU5_DATACENTERS, DataCenter
     from .images import image_for_spec
 
-    from .secret import render_env
-
     spec.validate()
     template_env = {
         "FLASH_RESOURCE_NAME": spec.name,
@@ -307,12 +309,8 @@ def _deployed_endpoint_input(
         "FLASH_ENVIRONMENT": env_name or app.env,
     }
 
-    # cross-resource calls from inside a worker go through the sentinel,
-    # which authenticates with the account api key; without it any
-    # .remote() from worker code would fail
-    api_key = runpod.api_key
-    if api_key and "RUNPOD_API_KEY" not in template_env:
-        template_env["RUNPOD_API_KEY"] = api_key
+    if app_key is not None and needs_app_key(spec):
+        template_env[APP_KEY_ENV] = app_key.reference
 
     # nested tasks use the same runtime and sdk channels as the endpoint
     for key in (
@@ -410,6 +408,9 @@ async def reconcile_endpoints(
     from .volume import VolumeResolver, attach_endpoint_volumes
 
     resolver = VolumeResolver(client)
+    app_key = None
+    if any(needs_app_key(h.spec) for h in provisionable.values()):
+        app_key = await ensure_app_key(client, app.name, environment["name"])
     endpoints: Dict[str, str] = {}
     for name, handle in sorted(provisionable.items()):
         payload = _deployed_endpoint_input(
@@ -419,6 +420,7 @@ async def reconcile_endpoints(
             build_id,
             python_version,
             env_name=environment["name"],
+            app_key=app_key,
         )
         await attach_endpoint_volumes(payload, handle.spec, resolver, app)
         auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client)