feat: add per-user OAuth mode proxying sign-in to Gitea
Authorization server for MCP clients (SDK provider with dynamic client registration, PKCE, refresh rotation and revocation) that signs users in through Gitea and runs tool calls with the user's own Gitea token. - consent page with CSRF protection and a browser-binding cookie on the Gitea callback (confused-deputy protection for the shared Gitea app) - encrypted token store (Fernet, HKDF-derived key, 0600, atomic writes, symlink and corrupt-file refusal); our tokens stored only as hashes - tokens bound to the MCP endpoint resource and the gitea scope; the static bearer keeps working next to OAuth - code replay and refresh token reuse revoke the grant - automatic upstream refresh with per-session locking; invalid_grant ends the session and never falls back to the admin token - OAuth-only deployments (no GITEA_TOKEN, no bearer) may start - rate limit on the OAuth endpoints, query strings dropped from access logs
This commit is contained in:
1 parent
5e144772b2
commit
566d0b65dd
10 files changed
+1813
-31
No files matched your search
@@ -27,6 +27,9 @@ if TYPE_CHECKING:
|
||||
#: ``client_id`` of the access token the static bearer verifier returns.
|
||||
STATIC_CLIENT_ID = "aegis-static-bearer"
|
||||
|
||||
#: The one OAuth scope this server grants and requires on the MCP endpoint.
|
||||
MCP_SCOPE = "gitea"
|
||||
|
||||
# After a failed ``GET /user`` the admin login lookup is not retried for this long.
|
||||
_LOGIN_RETRY_SECONDS = 30.0
|
||||
|
||||
@@ -45,10 +48,14 @@ class StaticBearerVerifier:
|
||||
|
||||
Args:
|
||||
token: The expected bearer token.
|
||||
resource: Resource (MCP endpoint URL) reported on the access token, so
|
||||
the SDK's RFC 8707 audience check accepts the static bearer when
|
||||
per-user OAuth enables it.
|
||||
"""
|
||||
|
||||
def __init__(self, token: str) -> None:
|
||||
def __init__(self, token: str, *, resource: str | None = None) -> None:
|
||||
self._expected = _digest(token)
|
||||
self._resource = resource
|
||||
|
||||
async def verify_token(self, token: str) -> AccessToken | None:
|
||||
"""Return an admin access token when ``token`` matches, else ``None``."""
|
||||
@@ -57,7 +64,8 @@ class StaticBearerVerifier:
|
||||
return AccessToken(
|
||||
token="static", # noqa: S106 (placeholder, never the real secret)
|
||||
client_id=STATIC_CLIENT_ID,
|
||||
scopes=["gitea"],
|
||||
scopes=[MCP_SCOPE],
|
||||
resource=self._resource,
|
||||
subject="admin",
|
||||
)
|
||||
|
||||
|
||||
Reference in new issue
Block a user