orchestrator/agent-team/tests/test_github_app.py

"""Unit tests for agent_team.github_app — App-JWT mint + TokenProvider cache.

These tests generate a throwaway RSA-2048 key with ``cryptography``, sign a real
App JWT, and intercept the mint HTTP call with a fake client so no network and no
real GitHub App is touched. They assert: the JWT claims/alg are correct, a clean
``201`` returns the ``{token, expires_at}`` mapping, the provider caches and
re-mints around the refresh margin, and — critically — that no token or JWT
material ever appears in a raised error's message (secret hygiene).
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

from agent_team.github_app import (
    GitHubAppError,
    TokenProvider,
    mint_installation_token,
)

_APP_ID = "123456"
_INSTALLATION_ID = "987654"

# A fixed clock so JWT iat/exp are deterministic.
_FIXED_NOW = datetime(2026, 6, 24, 12, 0, 0, tzinfo=timezone.utc)


@pytest.fixture
def rsa_keypair():
    """Generate a throwaway RSA-2048 keypair; return (private_pem, public_obj)."""
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    private_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode("ascii")
    return private_pem, private_key.public_key()


class _FakeResponse:
    def __init__(self, status: int, body):
        self.status_code = status
        self._body = body

    def json(self):
        return self._body


class _FakeHttp:
    """A callable (url, *, headers, timeout) mint client recording calls."""

    def __init__(self, response=None, raise_exc=None):
        self._response = response
        self._raise = raise_exc
        self.calls: list[dict] = []

    def __call__(self, url, *, headers=None, timeout=None):
        self.calls.append({"url": url, "headers": headers, "timeout": timeout})
        if self._raise is not None:
            raise self._raise
        return self._response

    @property
    def call_count(self) -> int:
        return len(self.calls)


def _fixed_now():
    return _FIXED_NOW


def _future_iso(seconds: int = 3600) -> str:
    return (_FIXED_NOW + timedelta(seconds=seconds)).strftime("%Y-%m-%dT%H:%M:%SZ")


# --------------------------------------------------------------------------- #
# JWT claims / algorithm                                                       #
# --------------------------------------------------------------------------- #
def test_mint_signs_jwt_with_expected_claims(rsa_keypair):
    private_pem, public_key = rsa_keypair
    http = _FakeHttp(_FakeResponse(201, {"token": "tok", "expires_at": _future_iso()}))

    mint_installation_token(
        app_id=_APP_ID,
        private_key_pem=private_pem,
        installation_id=_INSTALLATION_ID,
        _http=http,
        _now=_fixed_now,
    )

    auth = http.calls[0]["headers"]["Authorization"]
    assert auth.startswith("Bearer ")
    token = auth.split(" ", 1)[1]

    assert jwt.get_unverified_header(token)["alg"] == "RS256"
    # Verify the signature and claims, but not exp/iat against the real wall
    # clock: the JWT is minted from the fixed test clock (_fixed_now), so its
    # short-lived exp is in the past relative to the real time the suite runs.
    claims = jwt.decode(
        token,
        public_key,
        algorithms=["RS256"],
        options={"verify_aud": False, "verify_exp": False, "verify_iat": False},
    )
    assert claims["iss"] == _APP_ID
    now_ts = int(_FIXED_NOW.timestamp())
    assert claims["iat"] <= now_ts
    assert claims["exp"] - claims["iat"] <= 600


def test_mint_targets_installation_token_url(rsa_keypair):
    private_pem, _ = rsa_keypair
    http = _FakeHttp(_FakeResponse(201, {"token": "tok", "expires_at": _future_iso()}))

    mint_installation_token(
        app_id=_APP_ID,
        private_key_pem=private_pem,
        installation_id=_INSTALLATION_ID,
        _http=http,
        _now=_fixed_now,
    )

    assert http.calls[0]["url"].endswith(
        f"/app/installations/{_INSTALLATION_ID}/access_tokens"
    )


# --------------------------------------------------------------------------- #
# Mint success                                                                 #
# --------------------------------------------------------------------------- #
def test_mint_success_returns_token_and_expiry(rsa_keypair):
    private_pem, _ = rsa_keypair
    expires_at = _future_iso()
    http = _FakeHttp(_FakeResponse(201, {"token": "tok", "expires_at": expires_at}))

    result = mint_installation_token(
        app_id=_APP_ID,
        private_key_pem=private_pem,
        installation_id=_INSTALLATION_ID,
        _http=http,
        _now=_fixed_now,
    )

    assert result == {"token": "tok", "expires_at": expires_at}


# --------------------------------------------------------------------------- #
# TokenProvider caching / re-mint                                             #
# --------------------------------------------------------------------------- #
def test_provider_caches_token_across_calls(rsa_keypair):
    private_pem, _ = rsa_keypair
    http = _FakeHttp(
        _FakeResponse(201, {"token": "tok", "expires_at": _future_iso(86400)})
    )
    provider = TokenProvider(
        app_id=_APP_ID,
        private_key_pem=private_pem,
        installation_id=_INSTALLATION_ID,
        _http=http,
        _now=_fixed_now,
    )

    assert provider.token() == "tok"
    assert provider.token() == "tok"
    assert http.call_count == 1


def test_provider_remints_near_expiry(rsa_keypair):
    private_pem, _ = rsa_keypair
    # First mint expires 10 minutes out; with a 5-minute margin the second call
    # (clock advanced past expiry - margin) must re-mint.
    responses = [
        _FakeResponse(201, {"token": "tok1", "expires_at": _future_iso(600)}),
        _FakeResponse(201, {"token": "tok2", "expires_at": _future_iso(1200)}),
    ]

    class _SeqHttp(_FakeHttp):
        def __call__(self, url, *, headers=None, timeout=None):
            self.calls.append({"url": url})
            return responses[len(self.calls) - 1]

    http = _SeqHttp()

    clock = {"now": _FIXED_NOW}

    def _moving_now():
        return clock["now"]

    provider = TokenProvider(
        app_id=_APP_ID,
        private_key_pem=private_pem,
        installation_id=_INSTALLATION_ID,
        _http=http,
        _now=_moving_now,
        refresh_margin_s=300,
    )

    assert provider.token() == "tok1"
    assert http.call_count == 1

    # Advance past (expiry - margin) = +300s -> re-mint.
    clock["now"] = _FIXED_NOW + timedelta(seconds=400)
    assert provider.token() == "tok2"
    assert http.call_count == 2


# --------------------------------------------------------------------------- #
# Secret hygiene                                                               #
# --------------------------------------------------------------------------- #
def test_mint_failure_status_is_scrubbed(rsa_keypair):
    private_pem, _ = rsa_keypair
    # Even on a failure GitHub may echo the (bad) token; ensure nothing leaks.
    http = _FakeHttp(_FakeResponse(401, {"token": "tok", "message": "Bad creds"}))

    with pytest.raises(GitHubAppError) as exc:
        mint_installation_token(
            app_id=_APP_ID,
            private_key_pem=private_pem,
            installation_id=_INSTALLATION_ID,
            _http=http,
            _now=_fixed_now,
        )

    message = str(exc.value)
    assert "tok" not in message
    # No JWT material (RS256 JWTs start with the base64 header "eyJ").
    assert "eyJ" not in message


def test_mint_http_error_is_scrubbed(rsa_keypair):
    private_pem, _ = rsa_keypair
    http = _FakeHttp(raise_exc=RuntimeError("boom"))

    # A raised transport error propagates (fail closed); it must not carry the
    # JWT. We do not catch a specific type here — only assert no JWT leaks if it
    # were ever wrapped.
    with pytest.raises(Exception) as exc:  # noqa: PT011
        mint_installation_token(
            app_id=_APP_ID,
            private_key_pem=private_pem,
            installation_id=_INSTALLATION_ID,
            _http=http,
            _now=_fixed_now,
        )

    assert "eyJ" not in str(exc.value)


def test_mint_invalid_key_is_scrubbed():
    # An empty/invalid PEM must fail closed with a scrubbed message (no key).
    bad_key = "-----BEGIN PRIVATE KEY-----\nnotreallyakey\n-----END PRIVATE KEY-----"
    http = _FakeHttp(_FakeResponse(201, {"token": "tok", "expires_at": _future_iso()}))

    with pytest.raises(GitHubAppError) as exc:
        mint_installation_token(
            app_id=_APP_ID,
            private_key_pem=bad_key,
            installation_id=_INSTALLATION_ID,
            _http=http,
            _now=_fixed_now,
        )

    message = str(exc.value)
    assert "could not build app JWT" in message
    assert "notreallyakey" not in message