"""GitHub App auth: mint short-lived installation tokens on the trusted host. The trusted apply path (:mod:`agent_team.dispatcher`) needs a write-capable GitHub token to push the candidate head branch and trigger the apply/verify ``workflow_dispatch``. Rather than park a long-lived PAT on the operator host, this module mints an **installation access token** from a GitHub App private key: build a short-lived App JWT (signed RS256 with the App key), POST it to ``/app/installations/{installation_id}/access_tokens``, and receive a token that expires within the hour. :class:`TokenProvider` caches the minted token and re-mints just before expiry so callers can ask for a fresh token cheaply. SECRET HYGIENE (BLOCKING): the App JWT, the installation token, and anything derived from them are NEVER logged, NEVER placed in an exception message or ``str()``, and NEVER written to ledger / graph / task state. Mint/HTTP failures fail closed (raise :class:`GitHubAppError` with a scrubbed reason) so the dispatch node parks rather than fabricating success. Prevailing HTTP approach mirrors :mod:`agent_team.ci_fetcher`: ``requests`` is a deferred optional import, and an injectable ``requests``-like client (any object exposing ``post(url, *, json, timeout)`` / a callable for tests, with ``.status_code`` and ``.json()``) makes this unit-testable with no network. ``jwt`` (PyJWT) is likewise a deferred import. """ from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any, Callable __all__ = ["mint_installation_token", "TokenProvider", "GitHubAppError"] GITHUB_API_ROOT = "https://api.github.com" # App JWT lifetime knobs. GitHub rejects an App JWT whose ``exp`` is more than 10 # minutes out and is sensitive to clock skew, so we backdate ``iat`` by 60s and # cap the lifetime well under the 10-minute ceiling. _JWT_BACKDATE_S = 60 _JWT_LIFETIME_S = 540 # 9 minutes (<= 10 min GitHub ceiling) # Conservative default timeout for the single mint POST. A hang must fail (the # dispatch node parks), never wedge the operator host. _DEFAULT_TIMEOUT_S = 15.0 class GitHubAppError(Exception): """Raised on a missing/invalid key, a missing lib, or a mint failure. The message is ALWAYS scrubbed: it never contains the App private key, the App JWT, or the minted installation token. """ def _utc_now() -> datetime: """UTC now as an aware datetime (the default ``_now`` clock).""" return datetime.now(timezone.utc) def mint_installation_token( *, app_id: str, private_key_pem: str, installation_id: str, _http: Callable[..., Any] | None = None, _now: Callable[[], datetime] | None = None, ) -> dict: """Mint a GitHub App installation access token (fail-closed, secret-safe). Builds a short-lived RS256 App JWT from ``private_key_pem`` (``iss=app_id``, ``iat`` backdated 60s, ``exp`` 9 min out), then POSTs it to ``/app/installations/{installation_id}/access_tokens`` and returns ``{"token", "expires_at"}`` from the ``201`` response. ``_http`` injects a callable ``(url, *, headers, timeout) -> response`` (with ``.status_code`` / ``.json()``) for tests; when omitted, ``requests.post`` is used via a deferred import. ``_now`` injects the clock (a zero-arg callable returning an aware UTC datetime) for deterministic JWT claims. Raises :class:`GitHubAppError` (with a scrubbed message — never the key, the JWT, or the token) if PyJWT is unavailable, the key is empty/invalid, or the mint request does not return a ``201`` with a token + expiry. """ now = (_now or _utc_now)() iat = int(now.timestamp()) - _JWT_BACKDATE_S exp = int(now.timestamp()) + _JWT_LIFETIME_S try: import jwt # deferred: optional dependency (PyJWT) payload = {"iss": str(app_id), "iat": iat, "exp": exp} token_jwt = jwt.encode(payload, private_key_pem, algorithm="RS256") except Exception as exc: # noqa: BLE001 - missing lib / empty / invalid key # SECRET HYGIENE: surface only the exception TYPE, never the key or any # partially-built JWT material that an exception payload might carry. raise GitHubAppError(f"could not build app JWT: {type(exc).__name__}") from None url = f"{GITHUB_API_ROOT}/app/installations/{installation_id}/access_tokens" headers = { "Authorization": f"Bearer {token_jwt}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", } if _http is not None: resp = _http(url, headers=headers, timeout=_DEFAULT_TIMEOUT_S) else: import requests # deferred: optional dependency (see module docstring) resp = requests.post(url, headers=headers, timeout=_DEFAULT_TIMEOUT_S) status = getattr(resp, "status_code", None) body = resp.json() if status == 201 else None if status != 201 or not isinstance(body, dict): # NEVER include the JWT or any token in the failure message. Avoid even # the literal substring "tok" so a naive secret scan can't false-positive. raise GitHubAppError(f"installation-access mint failed: status={status}") token = body.get("token") expires_at = body.get("expires_at") if not token or not expires_at: raise GitHubAppError(f"installation-access mint failed: status={status}") return {"token": token, "expires_at": expires_at} class TokenProvider: """Caches a minted installation token, re-minting just before expiry. Construction does NO network and NO key validation — minting is lazy, on the first :meth:`token` call. The cached token is re-minted once ``now`` reaches ``expiry - refresh_margin_s`` so a caller always gets a token with usable headroom. The token is NEVER logged or otherwise exposed. """ def __init__( self, *, app_id: str, private_key_pem: str, installation_id: str, _http: Callable[..., Any] | None = None, _now: Callable[[], datetime] | None = None, refresh_margin_s: int = 300, ) -> None: self._app_id = app_id self._private_key_pem = private_key_pem self._installation_id = installation_id self._http = _http self._now = _now self._refresh_margin_s = refresh_margin_s self._cached_token: str | None = None self._cached_expiry: datetime | None = None def token(self) -> str: """Return a valid installation token, minting/re-minting as needed. Mints on first use and re-mints once within ``refresh_margin_s`` of the cached expiry. Propagates :class:`GitHubAppError` on a mint failure (the caller fails closed). The returned token is NEVER logged. """ now = (self._now or _utc_now)() if ( self._cached_token is None or self._cached_expiry is None or now + timedelta(seconds=self._refresh_margin_s) >= self._cached_expiry ): result = mint_installation_token( app_id=self._app_id, private_key_pem=self._private_key_pem, installation_id=self._installation_id, _http=self._http, _now=self._now, ) # Parse the expiry BEFORE caching the token so a malformed expires_at # raises GitHubAppError (fail closed, scrubbed) rather than leaving a # half-written cache (token set, expiry None) behind a bare ValueError. expiry = _parse_expires_at(result["expires_at"]) self._cached_token = result["token"] self._cached_expiry = expiry return self._cached_token def _parse_expires_at(expires_at: str) -> datetime: """Parse a GitHub ``expires_at`` ISO-8601 ``...Z`` string to aware UTC. GitHub returns e.g. ``2026-06-24T12:00:00Z``; normalise the trailing ``Z`` to a ``+00:00`` offset for :meth:`datetime.fromisoformat`. A malformed value raises :class:`GitHubAppError` (scrubbed — never the token) so the caller fails closed rather than propagating a bare ``ValueError``. """ try: return datetime.fromisoformat(expires_at.replace("Z", "+00:00")) except (ValueError, AttributeError) as exc: # Avoid even the literal substring "tok" so a naive secret scan / a test # asserting the token value is absent cannot false-positive on the word. raise GitHubAppError( f"could not parse installation-access expiry: {type(exc).__name__}" ) from None