DRU-761 -- Use the manager's GitHub App for API calls, Git credentials, and account links

Author: czpython

AGENTS.md

@@ -68,6 +68,9 @@ For app-surface changes, inspect the proof app at
 - Route a workflow cadence or pause through its schedule overrides. Do not use a
   settings column for this purpose.
 - Use one canonical name and shape for each contract. Fail loudly on every other shape.
+- A managed service differs from a self-hosted one only in configuration: a URL,
+  a key, or a verifier. Any manager that serves the same routes and publishes a
+  JWKS works.
 - Do not type-switch over a typed stream in app code. When a projection needs
   ordering or anchoring, grow the SDK primitive instead of an `isinstance` chain.
 - Put only identity and facts in a read-side field. Keep UI wording in the app pages.

backend/druks/accounts/jwt.py

@@ -1,5 +1,5 @@
 import re
-from functools import lru_cache
+from functools import cache
 
 from fastmcp.server.auth.providers.jwt import JWTVerifier
 from jsonpointer import JsonPointer
@@ -13,32 +13,39 @@
 _ARRAY_INDEX = re.compile(r"0|[1-9][0-9]*")
 
 
-@lru_cache(maxsize=1)
+@cache
 def _verifier(jwks_uri: str, issuer: str, audience: str) -> JWTVerifier:
     return JWTVerifier(jwks_uri=jwks_uri, issuer=issuer, audience=audience, algorithm=_ALGORITHM)
 
 
+async def verify_claims(token: str, *, jwks_url: str, issuer: str, audience: str) -> dict:
+    """The claims of a token that the keys at ``jwks_url`` signed for ``issuer`` and
+    ``audience``, else InvalidAssertionError."""
+    access = await _verifier(jwks_url, issuer, audience).verify_token(token)
+    # The verifier checks signature, issuer, audience, and expiry-if-present;
+    # our contract additionally requires exp.
+    if access and "exp" in access.claims:
+        return access.claims
+    raise InvalidAssertionError("Assertion rejected.")
+
+
 async def verify_assertion(token: str, settings: Settings) -> str:
     """The verified identity claim of an edge-minted assertion, else
     InvalidAssertionError."""
-    verifier = _verifier(
-        settings.identity.jwks_url,
-        settings.identity.jwt_issuer,
-        settings.identity.jwt_audience,
+    claim = await verify_claims(
+        token,
+        jwks_url=settings.identity.jwks_url,
+        issuer=settings.identity.jwt_issuer,
+        audience=settings.identity.jwt_audience,
     )
-    access = await verifier.verify_token(token)
-    # The verifier checks signature, issuer, audience, and expiry-if-present;
-    # our contract additionally requires exp and a nonblank string identity.
-    if not access or "exp" not in access.claims:
-        raise InvalidAssertionError("Assertion rejected.")
-    claim: object = access.claims
+    # Objects and arrays only: jsonpointer would index a string's characters.
     for part in JsonPointer(settings.identity.jwt_identity_claim).parts:
-        if isinstance(claim, dict) and part in claim:
-            claim = claim[part]
+        if isinstance(claim, dict):
+            claim = claim.get(part)
         elif isinstance(claim, list) and _ARRAY_INDEX.fullmatch(part) and int(part) < len(claim):
             claim = claim[int(part)]
         else:
-            raise InvalidAssertionError("Assertion carries no value at the identity pointer.")
+            claim = None
     if isinstance(claim, str) and claim.strip():
         return claim.strip()
     raise InvalidAssertionError(

backend/druks/core/apis/github.py

@@ -4,14 +4,17 @@
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any, Literal, TypeVar
 
-from githubkit import AppAuthStrategy, AppInstallationAuthStrategy, GitHub
+from githubkit import GitHub, TokenAuthStrategy
+from githubkit.auth import BaseAuthStrategy
 from githubkit.exception import GraphQLFailed, RequestFailed
 
 from druks.core.apis.exceptions import GitHubAppNotInstalledError
 from druks.core.utils.time import ensure_utc
+from druks.redis import get_client
+from druks.services.constants import OAUTH_TOKEN_TTL_SKEW_SECONDS
 
 logger = logging.getLogger(__name__)
 
@@ -63,22 +66,20 @@ async def wrapper(self: "GitHubClient", repo: str, *args: Any, **kwargs: Any) ->
 
 
 class GitHubClient:
+    """``app_auth`` signs the App routes at ``app_url``. Repository calls go to GitHub
+    with the installation tokens those routes mint."""
+
     def __init__(
         self,
         *,
         app_id: str,
-        private_key: str,
-        base_url: str = GITHUB_API_URL,
+        app_auth: BaseAuthStrategy,
+        app_url: str = GITHUB_API_URL,
         slug: str = "",
     ) -> None:
         self._app_id = app_id
-        self._private_key = private_key
-        self._base_url = base_url
         self._slug = slug
-        self._app = GitHub(
-            AppAuthStrategy(app_id, private_key),
-            base_url=base_url,
-        )
+        self._app = GitHub(app_auth, base_url=app_url)
         self._installation_cache: dict[str, int] = {}
         self._repo_gh_cache: dict[int, GitHub] = {}
 
@@ -116,10 +117,8 @@ async def list_repos_for_owner(self, owner: str) -> list[dict[str, Any]]:
         except GitHubAppNotInstalledError:
             return []
 
-        async with GitHub(
-            AppInstallationAuthStrategy(self._app_id, self._private_key, installation_id),
-            base_url=self._base_url,
-        ) as github:
+        github = await self._get_for_installation(installation_id)
+        async with github:
             repos: list[dict[str, Any]] = []
             page = 1
             while True:
@@ -149,10 +148,8 @@ async def create_repo_from_template(self, owner: str, template_repo: str, name:
         installation_id = await self._owner_installation_id(owner)
         template_owner, template_name = template_repo.split("/", 1)
         full_name = f"{owner}/{name}"
-        async with GitHub(
-            AppInstallationAuthStrategy(self._app_id, self._private_key, installation_id),
-            base_url=self._base_url,
-        ) as github:
+        github = await self._get_for_installation(installation_id)
+        async with github:
             try:
                 await github.rest.repos.async_create_using_template(
                     template_owner,
@@ -168,15 +165,16 @@ async def create_repo_from_template(self, owner: str, template_repo: str, name:
                 raise
         return full_name
 
-    async def list_installation_accounts(self) -> tuple[str, ...]:
+    async def list_installation_accounts(self, cached: bool = True) -> tuple[str, ...]:
         """Account logins (orgs/users) this App is installed on — the
         authoritative "where druks may act". Install the App somewhere and
         druks works there; uninstall and it stops. Cached ~10 min per app
-        id, serving the last-known set when GitHub hiccups."""
-        cached = _INSTALLATION_ACCOUNTS_CACHE.get(self._app_id)
+        id, serving the last-known set when GitHub hiccups. ``cached=False``
+        asks now, after an install."""
+        known = _INSTALLATION_ACCOUNTS_CACHE.get(self._app_id)
         now = time.monotonic()
-        if cached and now - cached[0] < _INSTALLATION_ACCOUNTS_TTL_SECONDS:
-            return cached[1]
+        if cached and known and now - known[0] < _INSTALLATION_ACCOUNTS_TTL_SECONDS:
+            return known[1]
         try:
             accounts: list[str] = []
             page = 1
@@ -194,12 +192,12 @@ async def list_installation_accounts(self) -> tuple[str, ...]:
                     break
                 page += 1
         except Exception:
-            if cached:
+            if known:
                 logger.warning(
                     "Could not refresh App installations; serving the last-known set.",
                     exc_info=True,
                 )
-                return cached[1]
+                return known[1]
             raise
         result = tuple(dict.fromkeys(accounts))
         _INSTALLATION_ACCOUNTS_CACHE[self._app_id] = (now, result)
@@ -228,7 +226,7 @@ async def get_bot_git_author(self) -> tuple[str, str]:
         bot_name = f"{await self.get_mention_handle()}[bot]"
         user_id = _BOT_USER_ID_CACHE.get(self._app_id)
         if not user_id:
-            async with GitHub(base_url=self._base_url) as github:
+            async with GitHub(base_url=GITHUB_API_URL) as github:
                 response = await github.rest.users.async_get_by_username(bot_name)
             user_id = response.parsed_data.id
             _BOT_USER_ID_CACHE[self._app_id] = user_id
@@ -238,12 +236,8 @@ async def get_authenticated_app_slug(self) -> str:
         """The slug GitHub reports for these credentials — a live App-JWT call,
         so it proves the App ID matches the PEM. The connect flow's validation;
         mention resolution reads the stored slug instead."""
-        async with GitHub(
-            AppAuthStrategy(self._app_id, self._private_key),
-            base_url=self._base_url,
-        ) as github:
-            response = await github.rest.apps.async_get_authenticated()
-            return str(getattr(response.parsed_data, "slug", None) or "")
+        response = await self._app.rest.apps.async_get_authenticated()
+        return str(getattr(response.parsed_data, "slug", None) or "")
 
     async def _installation_id(self, repo: str) -> int:
         if repo in self._installation_cache:
@@ -259,22 +253,38 @@ async def _installation_id(self, repo: str) -> int:
         self._installation_cache[repo] = installation_id
         return installation_id
 
+    async def _get_installation_token(self, installation_id: int) -> tuple[str, datetime]:
+        """The installation's token and its expiry, kept in Redis until the skew before it
+        expires."""
+        redis = get_client()
+        key = f"github:installation_token:{installation_id}"
+        if token := await redis.get(key):
+            return token.decode(), datetime.now(UTC) + timedelta(seconds=await redis.ttl(key))
+        response = await self._app.rest.apps.async_create_installation_access_token(
+            installation_id,
+        )
+        minted = response.parsed_data
+        expires_at = ensure_utc(datetime.fromisoformat(str(minted.expires_at)))
+        ttl = int((expires_at - datetime.now(UTC)).total_seconds()) - OAUTH_TOKEN_TTL_SKEW_SECONDS
+        if ttl > 0:
+            await redis.set(key, str(minted.token), ex=ttl)
+        return str(minted.token), expires_at
+
+    async def _get_for_installation(self, installation_id: int) -> GitHub:
+        token, _ = await self._get_installation_token(installation_id)
+        return GitHub(TokenAuthStrategy(token), base_url=GITHUB_API_URL)
+
     async def _for_repo(self, repo: str) -> GitHub:
         installation_id = await self._installation_id(repo)
         if installation_id not in self._repo_gh_cache:
-            self._repo_gh_cache[installation_id] = GitHub(
-                AppInstallationAuthStrategy(
-                    self._app_id,
-                    self._private_key,
-                    installation_id,
-                ),
-                base_url=self._base_url,
-            )
+            self._repo_gh_cache[installation_id] = await self._get_for_installation(installation_id)
         return self._repo_gh_cache[installation_id]
 
     async def _invalidate_for_repo(self, repo: str) -> None:
         installation_id = self._installation_cache.pop(repo, None)
         github = self._repo_gh_cache.pop(installation_id, None) if installation_id else None
+        if installation_id:
+            await get_client().delete(f"github:installation_token:{installation_id}")
         if github:
             try:
                 await github.__aexit__(None, None, None)
@@ -288,22 +298,9 @@ async def _invalidate_for_repo(self, repo: str) -> None:
     @_retry_on_401
     async def token_for_repo(self, repo: str) -> tuple[str, datetime]:
         """The installation token for ``repo`` and the expiry GitHub gave it."""
-        # The decorator drops the cached installation client + id on a
-        # 401 and retries once. Important here because git is the
-        # consumer of the minted token — once it's handed to git,
-        # there's no httpx-layer retry hook to recover from a stale
-        # ``installation_id`` (e.g. after the App was reinstalled and
-        # got a new id, leaving every worker's cache pointing at the
-        # dead one). The 401 from this method's own SDK call gives us
-        # the only chance to invalidate before git presents a bad
-        # token to GitHub and produces ``expected flush after ref
-        # listing``.
-        installation_id = await self._installation_id(repo)
-        token_response = await self._app.rest.apps.async_create_installation_access_token(
-            installation_id,
-        )
-        token = token_response.parsed_data
-        return str(token.token), ensure_utc(datetime.fromisoformat(str(token.expires_at)))
+        # git holds the token after this call, so a 401 here is the one chance to drop
+        # a stale installation id (the App was reinstalled) and mint again.
+        return await self._get_installation_token(await self._installation_id(repo))
 
     @_retry_on_401
     async def get_repository(self, repo: str) -> dict[str, Any]:

backend/druks/core/routes.py

@@ -83,11 +83,12 @@ async def github_manifest_callback(slug: str, code: str = "") -> HTMLResponse:
             "webhook_secret": app["webhook_secret"],
         }
     )
-    app_slug = row.identity["slug"]
-    install_url = f"https://github.com/apps/{quote(app_slug, safe='')}/installations/new"
     # druks opened this tab via window.open; the broadcast tells the connect
     # card to refetch, then the tab moves on to the one step GitHub still
     # needs — installing the App on the repositories druks should work in.
     return render_page(
-        "github_manifest_callback.html", slug=app_slug, install_url=install_url, service=slug
+        "github_manifest_callback.html",
+        slug=row.identity["slug"],
+        install_url=await service.get_install_endpoint(),
+        service=slug,
     )

backend/druks/core/services.py

@@ -4,9 +4,10 @@
 from contextlib import suppress
 from datetime import datetime
 from typing import Any
+from urllib.parse import quote
 
 import httpx
-from githubkit import GitHub
+from githubkit import AppAuthStrategy, GitHub, TokenAuthStrategy
 from pydantic import BaseModel, Field, SecretStr
 from slack_sdk.errors import SlackApiError
 from sqlalchemy import select
@@ -96,7 +97,7 @@ def get_manifest(cls, *, endpoint: str, webhook_base: str) -> dict[str, Any]:
     async def verify(cls, settings: Settings) -> dict[str, Any]:
         client = GitHubClient(
             app_id=settings.app_id,
-            private_key=settings.private_key.get_secret_value(),
+            app_auth=AppAuthStrategy(settings.app_id, settings.private_key.get_secret_value()),
         )
         try:
             slug = await client.get_authenticated_app_slug()
@@ -109,15 +110,33 @@ async def verify(cls, settings: Settings) -> dict[str, Any]:
 
     @classmethod
     async def get_client(cls) -> GitHubClient:
-        """The client of the connected App's vault row. PEM plaintext exists only
-        here, feeding the auth strategy."""
+        """The client of the connected App's vault row."""
         row = await cls.get()
+        if cls.is_manager_held():
+            return GitHubClient(
+                app_id=row.identity["app_id"],
+                app_auth=TokenAuthStrategy(load_settings().manager.token.get_secret_value()),
+                app_url=row.identity["url"],
+                slug=row.identity["slug"],
+            )
         return GitHubClient(
             app_id=row.identity["app_id"],
-            private_key=row.secrets["private_key"],
+            app_auth=AppAuthStrategy(row.identity["app_id"], row.secrets["private_key"]),
             slug=row.identity["slug"],
         )
 
+    @classmethod
+    async def get_install_endpoint(cls) -> str:
+        slug = (await cls.get()).identity["slug"]
+        return f"https://github.com/apps/{quote(slug, safe='')}/installations/new"
+
+    @classmethod
+    async def sync_installations(cls) -> None:
+        """The accounts the App is installed on, as the fact ``installations``."""
+        card = await cls.get()
+        accounts = await (await cls.get_client()).list_installation_accounts(cached=False)
+        card.identity = {**card.identity, "installations": list(accounts)}
+
     @classmethod
     async def issue_token(cls, resource: str) -> tuple[str, datetime]:
         """The installation token for the repo, and the expiry GitHub gave it."""