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,23 +23,33 @@ def admin_authentication(handler):
     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
 
@@ -58,6 +69,15 @@ def tenant_authentication(handler):
     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.
+        if has_auth_scopes(context):
+            _enforce_tenant_scope(get_auth_scopes(context), request.method)
+            return await handler(request)
+
         authorization_header = request.headers.get("Authorization")
         header_admin_api_key = request.headers.get("x-api-key")
         valid_key = general_utils.const_compare(
@@ -73,25 +93,82 @@ async def tenant_auth(request):
             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 _enforce_tenant_scope(scopes: set, method: str) -> None:
+    """Authorize a tenant request by its OAuth scopes, else raise HTTPForbidden.
+
+    ``acapy:tenant`` or ``acapy:admin`` grant tenant-level access;
+    ``acapy:tenant:read`` grants read-only access (safe HTTP methods only).
+    """
+    if scopes & {"acapy:tenant", "acapy:admin"}:
+        return
+    if "acapy:tenant:read" in scopes:
+        if method in ("GET", "HEAD"):
+            return
+        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",
+    )
+
+
 def _base_wallet_route_access(additional_routes: List[str], request_path: str) -> bool:
     """Check if request path matches additional routes."""
     additional_routes_pattern = (