POC: experimental Oauth support

acapy_agent/admin/auth_context.py

@@ -0,0 +1,75 @@
+"""Helpers for reading request-scoped auth context safely.
+
+These helpers support both:
+
+1) Preferred request metadata (``AdminRequestContext.metadata``), and
+2) Request-scoped settings fallback for compatibility.
+"""
+
+from collections.abc import Mapping
+from typing import Any, Optional
+
+AUTH_SCOPES_SETTING = "auth.scopes"
+AUTH_SUBJECT_SETTING = "auth.subject"
+AUTH_WALLET_ID_SETTING = "auth.wallet_id"
+
+
+def _get_context_settings(context: Any) -> Mapping[str, Any]:
+    settings = getattr(context, "settings", None)
+    if isinstance(settings, Mapping):
+        return settings
+    return {}
+
+
+def get_auth_metadata(context: Any) -> dict:
+    """Return auth metadata when available, else an empty dict."""
+    metadata = getattr(context, "metadata", None)
+    if isinstance(metadata, dict):
+        return metadata
+    return {}
+
+
+def has_auth_scopes(context: Any) -> bool:
+    """Whether auth scopes are present on the request context."""
+    metadata = get_auth_metadata(context)
+    if "scopes" in metadata:
+        return True
+    settings = _get_context_settings(context)
+    return AUTH_SCOPES_SETTING in settings
+
+
+def get_auth_scopes(context: Any) -> set[str]:
+    """Return normalized auth scopes from metadata or request settings."""
+    metadata = get_auth_metadata(context)
+    raw_scopes = metadata.get("scopes")
+    if raw_scopes is None:
+        raw_scopes = _get_context_settings(context).get(AUTH_SCOPES_SETTING, ())
+
+    if isinstance(raw_scopes, str):
+        return set(raw_scopes.split())
+    if isinstance(raw_scopes, (set, list, tuple, frozenset)):
+        return {scope for scope in raw_scopes if isinstance(scope, str)}
+    return set()
+
+
+def get_auth_subject(context: Any) -> Optional[str]:
+    """Return token subject claim from metadata or request settings."""
+    metadata = get_auth_metadata(context)
+    subject = metadata.get("sub")
+    if subject is None:
+        subject = _get_context_settings(context).get(AUTH_SUBJECT_SETTING)
+    return subject if isinstance(subject, str) else None
+
+
+def get_auth_wallet_id(context: Any) -> Optional[str]:
+    """Return request wallet id from metadata or request settings."""
+    metadata = get_auth_metadata(context)
+    wallet_id = metadata.get("wallet_id")
+    if wallet_id is None:
+        wallet_id = _get_context_settings(context).get(AUTH_WALLET_ID_SETTING)
+    return wallet_id if isinstance(wallet_id, str) else None
+
+
+def has_auth_wallet_id(context: Any) -> bool:
+    """Whether request context carries a wallet_id claim."""
+    return bool(get_auth_wallet_id(context))

acapy_agent/admin/decorators/auth.py

@@ -7,6 +7,7 @@
 from aiohttp import web
 
 from ...utils import general as general_utils
+from ..auth_context import get_auth_scopes, has_auth_scopes
 from ..request_context import AdminRequestContext
 
 
@@ -22,28 +23,38 @@
     async def admin_auth(request):
         context: AdminRequestContext = request["context"]
         profile = context.profile
+
+        if request.method == "OPTIONS":
+            return await handler(request)
+
+        # OAuth path: token was validated by setup_context middleware.
+        # Require acapy:admin scope for admin-only routes.
+        if has_auth_scopes(context):
+            if "acapy:admin" in get_auth_scopes(context):
+                return await handler(request)
+            raise web.HTTPForbidden(
+                reason="acapy:admin scope required",
+                text="acapy:admin scope required",
+            )
+
         header_admin_api_key = request.headers.get("x-api-key")
         valid_key = general_utils.const_compare(
             profile.settings.get("admin.admin_api_key"), header_admin_api_key
         )
         insecure_mode = bool(profile.settings.get("admin.admin_insecure_mode"))
 
-        # We have to allow OPTIONS method access to paths without a key since
-        # browsers performing CORS requests will never include the original
-        # x-api-key header from the method that triggered the preflight
-        # OPTIONS check.
-        if insecure_mode or valid_key or (request.method == "OPTIONS"):
+        if insecure_mode or valid_key:
             return await handler(request)
-        else:
-            raise web.HTTPUnauthorized(
-                reason="API Key invalid or missing",
-                text="API Key invalid or missing",
-            )
+
+        raise web.HTTPUnauthorized(
+            reason="API Key invalid or missing",
+            text="API Key invalid or missing",
+        )
 
     return admin_auth
 
 
 def tenant_authentication(handler):
     """Decorator to enable non-admin authentication.
 
     The decorator will:
@@ -58,6 +69,29 @@
     async def tenant_auth(request):
         context: AdminRequestContext = request["context"]
         profile = context.profile
+
+        if request.method == "OPTIONS":
+            return await handler(request)
+
+        # OAuth path: token was validated by setup_context middleware.
+        # acapy:tenant or acapy:admin both grant tenant-level access.
+        # acapy:tenant:read grants read-only access (safe HTTP methods only).
+        if has_auth_scopes(context):
+            scopes = get_auth_scopes(context)
+            if scopes & {"acapy:tenant", "acapy:admin"}:
+                return await handler(request)
+            if "acapy:tenant:read" in scopes:
+                if request.method in ("GET", "HEAD"):
+                    return await handler(request)
+                raise web.HTTPForbidden(
+                    reason="acapy:tenant:read scope does not permit write operations",
+                    text="acapy:tenant:read scope does not permit write operations",
+                )
+            raise web.HTTPForbidden(
+                reason="acapy:tenant scope required",
+                text="acapy:tenant scope required",
+            )
+
         authorization_header = request.headers.get("Authorization")
         header_admin_api_key = request.headers.get("x-api-key")
         valid_key = general_utils.const_compare(
@@ -73,25 +107,61 @@
             request.path,
         )
 
-        # CORS fix: allow OPTIONS method access to paths without a token
         if (
             (multitenant_enabled and authorization_header)
             or (not multitenant_enabled and valid_key)
             or (multitenant_enabled and valid_key and base_wallet_allowed_route)
             or (insecure_mode and not multitenant_enabled)
-            or request.method == "OPTIONS"
         ):
             return await handler(request)
-        else:
-            auth_mode = "Authorization token" if multitenant_enabled else "API key"
-            raise web.HTTPUnauthorized(
-                reason=f"{auth_mode} missing or invalid",
-                text=f"{auth_mode} missing or invalid",
-            )
+
+        auth_mode = "Authorization token" if multitenant_enabled else "API key"
+        raise web.HTTPUnauthorized(
+            reason=f"{auth_mode} missing or invalid",
+            text=f"{auth_mode} missing or invalid",
+        )
 
     return tenant_auth
 
 
+def require_scope(*required_scopes: str):
+    """Require at least one of the given OAuth2 scopes on the request token.
+
+    No-op when OAuth mode is not enabled (admin.oauth_enabled is not True), so
+    routes decorated with require_scope continue to work with API key / insecure
+    mode without any changes.
+
+    Must be stacked inside tenant_authentication or admin_authentication so that
+    authentication is checked before scope enforcement.
+
+    Example::
+
+        @tenant_authentication
+        @require_scope("acapy:tenant", "acapy:admin")
+        async def my_handler(request): ...
+    """
+
+    def decorator(handler):
+        @functools.wraps(handler)
+        async def scope_check(request):
+            if request.method == "OPTIONS":
+                return await handler(request)
+            context: AdminRequestContext = request["context"]
+            if not context.profile.settings.get("admin.oauth_enabled"):
+                return await handler(request)
+            token_scopes = get_auth_scopes(context)
+            if not token_scopes.intersection(required_scopes):
+                raise web.HTTPForbidden(
+                    reason="Insufficient scope",
+                    text="Insufficient scope",
+                )
+            return await handler(request)
+
+        return scope_check
+
+    return decorator
+
+
 def _base_wallet_route_access(additional_routes: List[str], request_path: str) -> bool:
     """Check if request path matches additional routes."""
     additional_routes_pattern = (

acapy_agent/admin/oauth_validator.py

@@ -0,0 +1,129 @@
+"""OAuth2 token validator for ACA-Py acting as a Resource Server."""
+
+import logging
+from typing import Optional
+
+import aiohttp
+import jwt
+from aiohttp import web
+
+LOGGER = logging.getLogger(__name__)
+
+_SUPPORTED_ALGORITHMS = [
+    "RS256",
+    "RS384",
+    "RS512",
+    "ES256",
+    "ES384",
+    "ES512",
+    "PS256",
+    "PS384",
+    "PS512",
+]
+
+
+class OAuthTokenValidator:
+    """Validate OAuth2 bearer tokens from an external Authorization Server.
+
+    Supports JWT access tokens validated locally via JWKS, with opaque token
+    fallback via RFC 7662 token introspection.
+    """
+
+    def __init__(self, settings):
+        """Initialize the validator from application settings."""
+        self.jwks_uri: Optional[str] = settings.get("oauth.jwks_uri")
+        self.issuer: Optional[str] = settings.get("oauth.issuer")
+        self.audience: Optional[str] = settings.get("oauth.audience")
+        self.introspection_endpoint: Optional[str] = settings.get(
+            "oauth.introspection_endpoint"
+        )
+        self.introspection_client_id: Optional[str] = settings.get(
+            "oauth.introspection_client_id"
+        )
+        self.introspection_client_secret: Optional[str] = settings.get(
+            "oauth.introspection_client_secret"
+        )
+
+        # PyJWKClient handles JWKS fetching and caching internally.
+        self._jwks_client = (
+            jwt.PyJWKClient(self.jwks_uri, cache_keys=True) if self.jwks_uri else None
+        )
+
+    async def validate(self, token: str) -> dict:
+        """Validate an access token and return its claims.
+
+        Tries JWT validation via JWKS first; falls back to introspection if the
+        token is opaque or if JWKS is not configured.
+
+        Raises:
+            web.HTTPUnauthorized: If the token is invalid or inactive.
+
+        Returns:
+            dict: Validated token claims.
+
+        """
+        if self._jwks_client:
+            try:
+                return self._validate_jwt(token)
+            except jwt.exceptions.PyJWKClientError as exc:
+                # JWKS infrastructure error — only fall through if introspection is
+                # configured as a fallback, otherwise the token cannot be validated.
+                if not self.introspection_endpoint:
+                    raise web.HTTPUnauthorized(reason="Token validation failed") from exc
+                LOGGER.debug("JWKS lookup failed, falling back to introspection: %s", exc)
+            except jwt.DecodeError:
+                # Token is not a JWT — only meaningful if introspection is available.
+                if not self.introspection_endpoint:
+                    raise web.HTTPUnauthorized(reason="Invalid token")
+                LOGGER.debug("JWT decode failed, falling back to introspection")
+            except jwt.InvalidTokenError as exc:
+                raise web.HTTPUnauthorized(reason=str(exc)) from exc
+
+        if self.introspection_endpoint:
+            return await self._introspect(token)
+
+        raise web.HTTPUnauthorized(
+            reason="No token validation method available — configure oauth.jwks_uri or "
+            "oauth.introspection_endpoint"
+        )
+
+    def _validate_jwt(self, token: str) -> dict:
+        """Validate a JWT access token using the configured JWKS endpoint."""
+        signing_key = self._jwks_client.get_signing_key_from_jwt(token)
+
+        decode_kwargs = dict(
+            algorithms=_SUPPORTED_ALGORITHMS,
+            options={"require": ["exp", "iss", "sub"]},
+        )
+        if self.issuer:
+            decode_kwargs["issuer"] = self.issuer
+        if self.audience:
+            decode_kwargs["audience"] = self.audience
+
+        return jwt.decode(token, signing_key.key, **decode_kwargs)
+
+    async def _introspect(self, token: str) -> dict:
+        """Validate an opaque token via RFC 7662 introspection."""
+        auth = None
+        if self.introspection_client_id:
+            auth = aiohttp.BasicAuth(
+                self.introspection_client_id,
+                self.introspection_client_secret or "",
+            )
+
+        async with aiohttp.ClientSession() as session:
+            resp = await session.post(
+                self.introspection_endpoint,
+                data={"token": token, "token_type_hint": "access_token"},
+                auth=auth,
+            )
+            if resp.status != 200:
+                raise web.HTTPUnauthorized(
+                    reason=f"Token introspection returned HTTP {resp.status}"
+                )
+            body = await resp.json()
+
+        if not body.get("active"):
+            raise web.HTTPUnauthorized(reason="Token is not active")
+
+        return body