From a171d14da70fad72b11d7166f49eb931beff72f0 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 15:49:44 +0200 Subject: [PATCH] [F053] _token_for_project: log decryption failure (key rotation) with project slug --- roboco/services/git.py | 20 ++++- .../services/test_git_token_decryption_log.py | 77 +++++++++++++++++++ 2 files changed, 95 insertions(+), 2 deletions(-) create mode 100644 tests/unit/services/test_git_token_decryption_log.py diff --git a/roboco/services/git.py b/roboco/services/git.py index a9302988..157d4111 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -309,13 +309,29 @@ class GitService(BaseService): return result async def _token_for_project(self, project_slug: str) -> str | None: - """Decrypted project token for orchestrator-side remote git ops.""" + """Decrypted project token for orchestrator-side remote git ops. + + A decryption failure (an encryption-key rotation left the stored PAT + encrypted with the old key) is logged loudly with the project slug + before returning None — without this every best-effort workspace git + op (push, PR, clone-with-token) silently looks like 'this project has + no token', indistinguishable from a project that genuinely never set + one, and the operator can't tell which project a key rotation wedged. + The best-effort skip behavior is preserved (callers still get None and + skip the remote op); this only makes the cause diagnosable. + """ from roboco.utils.crypto import EncryptionError project_service = get_project_service(self.session) try: return await project_service.get_decrypted_token_by_slug(project_slug) - except EncryptionError: + except EncryptionError as exc: + self.log.error( + "git token decryption failed — encryption key may have rotated;" + " treating as no-token for this project's remote git ops", + project_slug=project_slug, + error=str(exc), + ) return None async def _token_for_workspace(self, workspace: Path) -> str | None: diff --git a/tests/unit/services/test_git_token_decryption_log.py b/tests/unit/services/test_git_token_decryption_log.py new file mode 100644 index 00000000..1cd46cf8 --- /dev/null +++ b/tests/unit/services/test_git_token_decryption_log.py @@ -0,0 +1,77 @@ +"""F053: _token_for_project must log a Fernet decryption failure with the +project context, not swallow it silently as 'no token'. + +On an encryption-key rotation the stored PAT (encrypted with the old key) can't +be decrypted — ``crypto.decrypt_token`` raises ``EncryptionError``. The +crypto layer logs a generic message, but ``_token_for_project`` catches the +``EncryptionError`` and returns ``None`` with no project context, so every +best-effort workspace git op (push, PR, clone-with-token) silently looks like +'this project has no token' — indistinguishable from a project that genuinely +never set one. The operator can't tell which project is wedged by a key +rotation. Log the failure with the project slug before returning None (the +best-effort skip behavior is preserved — this only makes the cause +diagnosable). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from roboco.services.git import GitService +from roboco.utils.crypto import EncryptionError +from structlog.testing import capture_logs + +_PROJECT_SLUG = "acme-backend" + + +def _svc() -> GitService: + return GitService(MagicMock()) + + +def _patch_project_service_raising(exc: Exception) -> object: + fake_service = MagicMock() + fake_service.get_decrypted_token_by_slug = AsyncMock(side_effect=exc) + return patch("roboco.services.git.get_project_service", return_value=fake_service) + + +@pytest.mark.asyncio +async def test_decryption_failure_logged_with_project_slug() -> None: + """An EncryptionError (key rotation) is logged with the project slug so the + operator can tell WHICH project is wedged — not silently masked as 'no + token'.""" + svc = _svc() + with ( + _patch_project_service_raising( + EncryptionError("Unable to decrypt token - encryption key may have changed") + ), + capture_logs() as logs, + ): + result = await svc._token_for_project(_PROJECT_SLUG) + + # Best-effort behavior preserved: returns None (callers skip remote ops). + assert result is None + # ... but the failure is loud + project-scoped, not silent. + error_logs = [e for e in logs if e["log_level"] == "error"] + assert error_logs, "decryption failure must be logged at error level" + log = error_logs[0] + assert "decrypt" in log["event"].lower() or "token" in log["event"].lower() + assert log.get("project_slug") == _PROJECT_SLUG or _PROJECT_SLUG in str(log) + + +@pytest.mark.asyncio +async def test_missing_token_returns_none_silently() -> None: + """A project that genuinely has no token (None, no exception) returns None + with NO error log — the fix must not over-log the legitimate no-token + case (regression guard).""" + svc = _svc() + fake_service = MagicMock() + fake_service.get_decrypted_token_by_slug = AsyncMock(return_value=None) + with ( + patch("roboco.services.git.get_project_service", return_value=fake_service), + capture_logs() as logs, + ): + result = await svc._token_for_project(_PROJECT_SLUG) + + assert result is None + assert not [e for e in logs if e["log_level"] == "error"]