fix: redact identity secrets through a public hook and mask issued agm_ tokens

Identity gains a redact field that the executor honours, so the OAuth
resolver no longer needs an executor subclass overriding private
internals. Opaque agm_ tokens issued by our authorization server are now
masked by the secret patterns wherever they appear in content.

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 = {

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."""

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

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
         )
 

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"),

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(
+    identity = Identity(login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,))
-        login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,)
-    )
 
     outcome = await executor.request("GET", "/version", identity)
     await executor.aclose()

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"}

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"}