feat(github-app): App credentials, installation tokens, and a Select repo picker (#621)

* feat(github-app): App credentials, installation tokens, and a Select repo picker RoboCo was 100% PAT-based. A singleton Fernet-encrypted github_app_credentials row (migration 077, telegram-credentials pattern) now stores the App id + private key; github_app_auth mints RS256 app JWTs and caches installation tokens until 5 minutes before expiry. Projects can bind an installation (projects.github_installation_id): get_decrypted_token returns a minted installation token for bound projects and falls back to the stored PAT on any minting failure, so all ten token consumers work unchanged. CEO-gated routes expose credentials CRUD plus installation/repo listing, and the New Project dialog gains a Select repo picker (disabled with a HelpTip until the App is configured) that fills the git URL and binds the installation; manual URL + PAT stays the default path. * test(panel): mock the GitHub App credentials card in the settings page test --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
2026-08-03 07:23:24 +02:00 · 2026-07-21 00:54:20 +02:00
parent 2d210ce6ee
commit 7b84162ae9
26 changed files with 2105 additions and 38 deletions
@@ -0,0 +1,208 @@
+"""GitHub App installation-token minting — JWT shape, token cache, pagination.
+
+The App JWT is signed RS256 from the (mocked) stored credentials; every REST
+call goes through a MockTransport-free ``httpx.AsyncClient`` patch mirroring
+``test_git_pr_ci_status.py``'s idiom, since the module owns its own client
+lifecycle (no injection seam, matching the spec's plain-function shape).
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from roboco.services import github_app_auth
+from roboco.services.github_app_auth import (
+    GitHubAppAPIError,
+    GitHubAppNotConfiguredError,
+    list_installation_repositories,
+    list_installations,
+    mint_installation_token,
+)
+from roboco.services.github_app_credentials import GitHubAppCredentialsData
+
+
+def _generate_rsa_keypair() -> str:
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    return private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode()
+
+
+_PRIVATE_KEY_PEM = _generate_rsa_keypair()
+_APP_ID = "998877"
+_SECOND_CALL_COUNT = 2
+
+
+@pytest.fixture(autouse=True)
+def _clear_token_cache() -> None:
+    github_app_auth._token_cache.clear()
+
+
+def _patch_creds(*, configured: bool = True) -> Any:
+    fake_service = MagicMock()
+    creds = (
+        GitHubAppCredentialsData(app_id=_APP_ID, private_key=_PRIVATE_KEY_PEM)
+        if configured
+        else None
+    )
+    fake_service.get_decrypted = AsyncMock(return_value=creds)
+    return patch(
+        "roboco.services.github_app_auth.get_github_app_credentials_service",
+        return_value=fake_service,
+    )
+
+
+def _client(
+    *, get: list[MagicMock] | None = None, post: list[MagicMock] | None = None
+) -> MagicMock:
+    client = MagicMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=False)
+    if get is not None:
+        client.get = AsyncMock(side_effect=get)
+    if post is not None:
+        client.post = AsyncMock(side_effect=post)
+    return client
+
+
+def _resp(status_code: int, json_payload: Any = None, text: str = "") -> MagicMock:
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.json.return_value = json_payload
+    resp.text = text
+    return resp
+
+
+def _token_resp(token: str, *, expires_in_seconds: int = 3600) -> MagicMock:
+    expires_at = (datetime.now(UTC) + timedelta(seconds=expires_in_seconds)).isoformat()
+    return _resp(201, {"token": token, "expires_at": expires_at})
+
+
+@pytest.mark.asyncio
+async def test_missing_credentials_raises_not_configured() -> None:
+    with _patch_creds(configured=False), pytest.raises(GitHubAppNotConfiguredError):
+        await mint_installation_token(MagicMock(), 42)
+
+
+@pytest.mark.asyncio
+async def test_jwt_claims_shape() -> None:
+    """The App JWT sent as the POST Authorization header decodes to
+    iss=app_id, an iat backdated ~60s, and an exp within the 9-minute TTL."""
+    client = _client(post=[_token_resp("tok-1")])
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+    ):
+        before = int(time.time())
+        await mint_installation_token(MagicMock(), 1)
+
+    _, kwargs = client.post.call_args
+    auth_header = kwargs["headers"]["Authorization"]
+    assert auth_header.startswith("Bearer ")
+    app_jwt = auth_header.removeprefix("Bearer ")
+
+    claims = jwt.decode(app_jwt, options={"verify_signature": False})
+    assert claims["iss"] == _APP_ID
+    assert before - claims["iat"] == pytest.approx(60, abs=2)
+    assert claims["exp"] - claims["iat"] == pytest.approx(9 * 60 + 60, abs=2)
+
+
+@pytest.mark.asyncio
+async def test_cache_reuse_skips_a_second_mint() -> None:
+    client = _client(post=[_token_resp("tok-cached", expires_in_seconds=3600)])
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+    ):
+        first = await mint_installation_token(MagicMock(), 7)
+        second = await mint_installation_token(MagicMock(), 7)
+
+    assert first == second == "tok-cached"
+    assert client.post.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_expiry_re_mints_past_the_refresh_margin() -> None:
+    # Expires in 30s — under the 5-minute refresh margin, so the second call
+    # must not trust the cached entry.
+    client = _client(
+        post=[
+            _token_resp("tok-old", expires_in_seconds=30),
+            _token_resp("tok-fresh", expires_in_seconds=3600),
+        ]
+    )
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+    ):
+        first = await mint_installation_token(MagicMock(), 9)
+        second = await mint_installation_token(MagicMock(), 9)
+
+    assert first == "tok-old"
+    assert second == "tok-fresh"
+    assert client.post.call_count == _SECOND_CALL_COUNT
+
+
+@pytest.mark.asyncio
+async def test_mint_failure_raises_api_error() -> None:
+    client = _client(post=[_resp(401, text="Bad credentials")])
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+        pytest.raises(GitHubAppAPIError),
+    ):
+        await mint_installation_token(MagicMock(), 5)
+
+
+@pytest.mark.asyncio
+async def test_list_installations_maps_account_login() -> None:
+    payload = [
+        {"id": 1, "account": {"login": "acme"}},
+        {"id": 2, "account": {"login": "widgets"}},
+    ]
+    client = _client(get=[_resp(200, payload)])
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+    ):
+        installations = await list_installations(MagicMock())
+
+    assert [(i.id, i.account_login) for i in installations] == [
+        (1, "acme"),
+        (2, "widgets"),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_list_installation_repositories_paginates(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(github_app_auth, "_PER_PAGE", 2)
+
+    def _repo(name: str) -> dict[str, Any]:
+        return {
+            "full_name": f"acme/{name}",
+            "clone_url": f"https://github.com/acme/{name}.git",
+            "private": True,
+        }
+
+    page1 = _resp(200, {"repositories": [_repo("a"), _repo("b")]})
+    page2 = _resp(200, {"repositories": [_repo("c")]})
+    client = _client(post=[_token_resp("tok")], get=[page1, page2])
+    with (
+        _patch_creds(),
+        patch("roboco.services.github_app_auth.httpx.AsyncClient", return_value=client),
+    ):
+        repos = await list_installation_repositories(MagicMock(), 11)
+
+    assert [r.full_name for r in repos] == ["acme/a", "acme/b", "acme/c"]
+    assert client.get.call_count == _SECOND_CALL_COUNT
@@ -0,0 +1,137 @@
+"""ProjectService.get_decrypted_token{,_by_slug} — the GitHub App installation-
+token branch + its fall back to the stored PAT.
+
+An installation-bound project with App credentials mints a token; any minting
+failure (App unconfigured, revoked installation, network hiccup) falls back
+to the PAT rather than breaking git operations. Mocks the DB boundary (the
+project lookup) and the two collaborators (``github_app_credentials``,
+``github_app_auth``) directly — no real network/DB involved.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.services.github_app_auth import GitHubAppAPIError
+from roboco.services.project import ProjectService
+from roboco.utils.crypto import encrypt_token
+
+
+def _project(*, installation_id: int | None, pat: str | None) -> MagicMock:
+    p = MagicMock()
+    p.id = uuid4()
+    p.github_installation_id = installation_id
+    p.git_token_encrypted = encrypt_token(pat) if pat else None
+    return p
+
+
+def _svc_with_get(project: MagicMock) -> ProjectService:
+    svc = ProjectService(MagicMock())
+    svc.get = AsyncMock(return_value=project)  # type: ignore[method-assign]
+    svc.get_by_slug = AsyncMock(return_value=project)  # type: ignore[method-assign]
+    return svc
+
+
+@pytest.mark.asyncio
+async def test_no_installation_uses_stored_pat() -> None:
+    svc = _svc_with_get(_project(installation_id=None, pat="ghp_plain"))
+    token = await svc.get_decrypted_token(uuid4())
+    assert token == "ghp_plain"
+
+
+@pytest.mark.asyncio
+async def test_no_project_returns_none() -> None:
+    svc = _svc_with_get(None)  # type: ignore[arg-type]
+    assert await svc.get_decrypted_token(uuid4()) is None
+    assert await svc.get_decrypted_token_by_slug("nope") is None
+
+
+@pytest.mark.asyncio
+async def test_installation_id_without_app_creds_falls_back_to_pat() -> None:
+    svc = _svc_with_get(_project(installation_id=42, pat="ghp_fallback"))
+    fake_creds_svc = MagicMock()
+    fake_creds_svc.has_credentials = AsyncMock(return_value=False)
+    with patch(
+        "roboco.services.project.get_github_app_credentials_service",
+        return_value=fake_creds_svc,
+    ):
+        token = await svc.get_decrypted_token(uuid4())
+    assert token == "ghp_fallback"
+
+
+@pytest.mark.asyncio
+async def test_installation_id_with_app_creds_mints_token() -> None:
+    svc = _svc_with_get(_project(installation_id=42, pat="ghp_unused"))
+    fake_creds_svc = MagicMock()
+    fake_creds_svc.has_credentials = AsyncMock(return_value=True)
+    with (
+        patch(
+            "roboco.services.project.get_github_app_credentials_service",
+            return_value=fake_creds_svc,
+        ),
+        patch(
+            "roboco.services.project.mint_installation_token",
+            AsyncMock(return_value="ghs_minted"),
+        ),
+    ):
+        token = await svc.get_decrypted_token(uuid4())
+    assert token == "ghs_minted"
+
+
+@pytest.mark.asyncio
+async def test_mint_failure_falls_back_to_pat() -> None:
+    svc = _svc_with_get(_project(installation_id=42, pat="ghp_fallback"))
+    fake_creds_svc = MagicMock()
+    fake_creds_svc.has_credentials = AsyncMock(return_value=True)
+    with (
+        patch(
+            "roboco.services.project.get_github_app_credentials_service",
+            return_value=fake_creds_svc,
+        ),
+        patch(
+            "roboco.services.project.mint_installation_token",
+            AsyncMock(side_effect=GitHubAppAPIError("revoked")),
+        ),
+    ):
+        token = await svc.get_decrypted_token(uuid4())
+    assert token == "ghp_fallback"
+
+
+@pytest.mark.asyncio
+async def test_mint_failure_with_no_pat_returns_none() -> None:
+    svc = _svc_with_get(_project(installation_id=42, pat=None))
+    fake_creds_svc = MagicMock()
+    fake_creds_svc.has_credentials = AsyncMock(return_value=True)
+    with (
+        patch(
+            "roboco.services.project.get_github_app_credentials_service",
+            return_value=fake_creds_svc,
+        ),
+        patch(
+            "roboco.services.project.mint_installation_token",
+            AsyncMock(side_effect=GitHubAppAPIError("revoked")),
+        ),
+    ):
+        token = await svc.get_decrypted_token(uuid4())
+    assert token is None
+
+
+@pytest.mark.asyncio
+async def test_by_slug_mints_token_too() -> None:
+    svc = _svc_with_get(_project(installation_id=7, pat=None))
+    fake_creds_svc = MagicMock()
+    fake_creds_svc.has_credentials = AsyncMock(return_value=True)
+    with (
+        patch(
+            "roboco.services.project.get_github_app_credentials_service",
+            return_value=fake_creds_svc,
+        ),
+        patch(
+            "roboco.services.project.mint_installation_token",
+            AsyncMock(return_value="ghs_minted"),
+        ),
+    ):
+        token = await svc.get_decrypted_token_by_slug("acme")
+    assert token == "ghs_minted"