diff --git a/src/aegis_gitea_mcp/gitea/executor.py b/src/aegis_gitea_mcp/gitea/executor.py index 23e0d74..48ce832 100644 --- a/src/aegis_gitea_mcp/gitea/executor.py +++ b/src/aegis_gitea_mcp/gitea/executor.py @@ -10,7 +10,7 @@ Security properties (all covered by tests): * path values are validated and percent-encoded (see :mod:`.encoding`); * redirects are never followed (a 3xx is an error of kind ``redirect``); * cookies are never stored, so one identity can not leak into another call; -* the identity token, the configured Gitea token, the MCP bearer token, the +* the identity token and ``Identity.redact``, the configured Gitea token, the MCP bearer token, the OAuth client secret, ``extra_secrets`` and secret values submitted in a request body are replaced with ``[REDACTED]`` in every result, always; * secret-pattern masking/blocking follows ``SECRET_DETECTION_MODE``, except for @@ -412,7 +412,9 @@ class GiteaExecutor: # Sending def _secrets(self, identity: Identity, call: _Call) -> tuple[str, ...]: - return normalize_secrets([identity.token, *self._base_secrets, *call.call_secrets]) + return normalize_secrets( + [identity.token, *identity.redact, *self._base_secrets, *call.call_secrets] + ) async def _send(self, call: _Call, identity: Identity) -> ToolOutcome: headers = { diff --git a/src/aegis_gitea_mcp/identity.py b/src/aegis_gitea_mcp/identity.py index b381ebb..dfe1c69 100644 --- a/src/aegis_gitea_mcp/identity.py +++ b/src/aegis_gitea_mcp/identity.py @@ -23,11 +23,15 @@ class Identity: mode: ``admin-token`` or ``oauth``. token: Gitea token for this identity. Excluded from ``repr`` and never part of :meth:`metadata`. + redact: Further secrets of this identity (for example the OAuth user's + Gitea refresh token) that the executor replaces in every result. + Excluded from ``repr`` and comparisons. """ login: str | None mode: IdentityMode token: str = field(repr=False) + redact: tuple[str, ...] = field(default=(), repr=False, compare=False) def metadata(self, sudo: str | None = None) -> dict[str, Any]: """Return result and audit metadata: ``{"login", "mode", "sudo"}``, never the token.""" diff --git a/src/aegis_gitea_mcp/oauth/components.py b/src/aegis_gitea_mcp/oauth/components.py index 2d01aa2..15258d2 100644 --- a/src/aegis_gitea_mcp/oauth/components.py +++ b/src/aegis_gitea_mcp/oauth/components.py @@ -15,10 +15,11 @@ from starlette.routing import Route from aegis_gitea_mcp.audit import AuditLogger from aegis_gitea_mcp.auth import MCP_SCOPE, AdminIdentityResolver from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.gitea import GiteaExecutor from aegis_gitea_mcp.oauth import OAuthSetupError from aegis_gitea_mcp.oauth.callback import build_browser_routes from aegis_gitea_mcp.oauth.provider import CONSENT_PATH, GiteaOAuthProvider, OAuthTokenVerifier -from aegis_gitea_mcp.oauth.resolver import OAuthGiteaExecutor, OAuthIdentityResolver +from aegis_gitea_mcp.oauth.resolver import OAuthIdentityResolver from aegis_gitea_mcp.oauth.store import OAuthStore, OAuthStoreError from aegis_gitea_mcp.oauth.upstream import CALLBACK_PATH, GiteaUpstream from aegis_gitea_mcp.transports import HttpAuthExtras @@ -88,7 +89,7 @@ def build_oauth_components( upstream = GiteaUpstream(settings, client_factory=client_factory, clock=clock) provider = GiteaOAuthProvider(settings, store, upstream, clock=clock, audit=audit) - executor = OAuthGiteaExecutor( + executor = GiteaExecutor( settings, client=executor_client, extra_secrets=[key.get_secret_value()] ) admin = AdminIdentityResolver(settings, executor) if settings.gitea_token is not None else None diff --git a/src/aegis_gitea_mcp/oauth/resolver.py b/src/aegis_gitea_mcp/oauth/resolver.py index fc52b3a..6b2ae69 100644 --- a/src/aegis_gitea_mcp/oauth/resolver.py +++ b/src/aegis_gitea_mcp/oauth/resolver.py @@ -18,7 +18,6 @@ import dataclasses import logging import time from collections.abc import Callable -from dataclasses import dataclass, field from typing import TYPE_CHECKING from mcp.server.auth.provider import AccessToken @@ -26,14 +25,12 @@ from mcp.server.auth.provider import AccessToken from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger from aegis_gitea_mcp.auth import STATIC_CLIENT_ID, IdentityResolver from aegis_gitea_mcp.errors import ToolError -from aegis_gitea_mcp.gitea import GiteaExecutor -from aegis_gitea_mcp.gitea.redaction import normalize_secrets from aegis_gitea_mcp.identity import Identity from aegis_gitea_mcp.oauth.store import OAuthStore, OAuthStoreError, SessionRecord from aegis_gitea_mcp.oauth.upstream import GiteaUpstream, UpstreamError, UpstreamInvalidGrant if TYPE_CHECKING: - from aegis_gitea_mcp.gitea.executor import _Call + pass logger = logging.getLogger(__name__) @@ -47,26 +44,6 @@ OAUTH_ONLY = ( ) -@dataclass(frozen=True) -class SessionIdentity(Identity): - """An OAuth identity that also names session secrets to redact from results. - - Attributes: - redact: Further secrets of the session (the Gitea refresh token) that must - never appear in a result. Excluded from ``repr`` and comparisons. - """ - - redact: tuple[str, ...] = field(default=(), repr=False, compare=False) - - -class OAuthGiteaExecutor(GiteaExecutor): - """Executor that also redacts the session secrets carried by a :class:`SessionIdentity`.""" - - def _secrets(self, identity: Identity, call: _Call) -> tuple[str, ...]: - extra = identity.redact if isinstance(identity, SessionIdentity) else () - return normalize_secrets([*super()._secrets(identity, call), *extra]) - - class OAuthIdentityResolver: """Maps a verified access token to the identity a tool call runs as. @@ -120,7 +97,8 @@ class OAuthIdentityResolver: if self._needs_refresh(session): session = await self._refresh(session_id) redact = (session.gitea_refresh_token,) if session.gitea_refresh_token else () - return SessionIdentity( + # The executor redacts ``Identity.redact`` (the Gitea refresh token) from results. + return Identity( login=session.login, mode="oauth", token=session.gitea_access_token, redact=redact ) diff --git a/src/aegis_gitea_mcp/security.py b/src/aegis_gitea_mcp/security.py index 2c3161c..a3bdafe 100644 --- a/src/aegis_gitea_mcp/security.py +++ b/src/aegis_gitea_mcp/security.py @@ -28,6 +28,11 @@ _SECRET_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( "github_token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b"), ), + ( + # Opaque OAuth tokens this server issues (``agm_`` + token_urlsafe(32)). + "aegis_oauth_token", + re.compile(r"\bagm_[A-Za-z0-9_-]{32,}"), + ), ( "jwt", re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{4,}\.[A-Za-z0-9_-]{4,}\b"), diff --git a/tests/test_oauth_resolver.py b/tests/test_oauth_resolver.py index 8b7e726..183170f 100644 --- a/tests/test_oauth_resolver.py +++ b/tests/test_oauth_resolver.py @@ -11,13 +11,12 @@ from mcp.server.auth.provider import AccessToken from aegis_gitea_mcp.auth import STATIC_CLIENT_ID from aegis_gitea_mcp.errors import ToolError +from aegis_gitea_mcp.gitea import GiteaExecutor from aegis_gitea_mcp.identity import Identity from aegis_gitea_mcp.oauth.resolver import ( OAUTH_ONLY, SIGN_IN_AGAIN, - OAuthGiteaExecutor, OAuthIdentityResolver, - SessionIdentity, ) from aegis_gitea_mcp.oauth.store import OAuthStore, SessionRecord, TokenRecord from aegis_gitea_mcp.oauth.upstream import UpstreamError, UpstreamInvalidGrant, UpstreamTokens @@ -108,7 +107,6 @@ async def test_oauth_token_resolves_to_the_session_user(tmp_path: Path) -> None: clock = FakeClock() identity = await resolver(make_store(tmp_path, clock), clock=clock).resolve(oauth_access()) - assert isinstance(identity, SessionIdentity) assert (identity.login, identity.mode, identity.token) == ("alice", "oauth", USER_TOKEN) assert identity.redact == (USER_REFRESH,) assert USER_TOKEN not in repr(identity) and USER_REFRESH not in repr(identity) @@ -341,14 +339,12 @@ async def test_audit_failure_does_not_block_ending_a_session( async def test_executor_redacts_session_secrets(httpx_mock: Any) -> None: - executor = OAuthGiteaExecutor(oauth_settings(), extra_secrets=[ENCRYPTION_KEY]) + executor = GiteaExecutor(oauth_settings(), extra_secrets=[ENCRYPTION_KEY]) httpx_mock.add_response( url="https://gitea.example.com/api/v1/version", json={"a": USER_REFRESH, "b": ENCRYPTION_KEY, "c": USER_TOKEN}, ) - identity = SessionIdentity( - login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,) - ) + identity = Identity(login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,)) outcome = await executor.request("GET", "/version", identity) await executor.aclose() diff --git a/tests/test_oauth_startup.py b/tests/test_oauth_startup.py index 5a8f63f..71c95cd 100644 --- a/tests/test_oauth_startup.py +++ b/tests/test_oauth_startup.py @@ -10,10 +10,10 @@ import pytest from pytest_httpx import HTTPXMock from aegis_gitea_mcp import transports +from aegis_gitea_mcp.gitea import GiteaExecutor from aegis_gitea_mcp.oauth import OAuthSetupError from aegis_gitea_mcp.oauth import components as components_module from aegis_gitea_mcp.oauth.components import RATE_LIMITED_PATHS, build_oauth_components -from aegis_gitea_mcp.oauth.resolver import OAuthGiteaExecutor from aegis_gitea_mcp.oauth.store import STORE_FILENAME from aegis_gitea_mcp.transports import AccessLogQueryFilter, http_startup_errors, run_http from tests.test_oauth_helpers import ( @@ -129,7 +129,7 @@ def test_components_wiring() -> None: settings = oauth_settings() extras = build_oauth_components(settings) - assert isinstance(extras.executor, OAuthGiteaExecutor) + assert isinstance(extras.executor, GiteaExecutor) assert extras.rate_limited_paths == RATE_LIMITED_PATHS paths = {route.path for route in extras.extra_routes} # type: ignore[attr-defined] assert paths == {"/oauth/consent", "/oauth/callback", "/.well-known/oauth-protected-resource"} diff --git a/tests/test_security.py b/tests/test_security.py index 8a458bf..cf88fe8 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -9,6 +9,19 @@ def test_detect_secrets_api_key_pattern() -> None: assert findings +def test_issued_oauth_tokens_are_masked() -> None: + """Our own ``agm_`` OAuth tokens are masked wherever they show up in content.""" + issued = "agm_" + "Ab3_-x" * 7 + "Z" # the shape of agm_ + token_urlsafe(32) + text = f"pasted by mistake: {issued}." + + findings = detect_secrets(text) + masked = sanitize_data({"body": text}) + + assert [finding.secret_type for finding in findings] == ["aegis_oauth_token"] + assert issued not in masked["body"] and masked["body"].endswith(".") + assert detect_secrets("agm_short") == [] + + def test_sanitize_data_mask_mode() -> None: """Mask mode should preserve structure while redacting values.""" payload = {"content": "api_key=AKIA1234567890ABCDEF"}