Feature: use decorators for admin api authentication (openwallet-foun…

…dation#2860)

* Add authentication decorators for admin API

Signed-off-by: Emiliano Suñé <[email protected]>

---------

Signed-off-by: Emiliano Suñé <[email protected]>

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ repos:
         additional_dependencies: ['@commitlint/config-conventional']
   - repo: https://github.com/psf/black
     # Ensure this is synced with pyproject.toml
-    rev: 24.1.1
+    rev: 24.4.0
     hooks:
       - id: black
         stages: [commit]

aries_cloudagent/admin/decorators/auth.py

@@ -0,0 +1,80 @@
+"""Authentication decorators for the admin API."""
+
+import functools
+
+from aiohttp import web
+
+from ...utils import general as general_utils
+from ..request_context import AdminRequestContext
+
+
+def admin_authentication(handler):
+    """Decorator to require authentication via admin API key.
+
+    The decorator will check for a valid x-api-key header and
+    reject the request if it is missing or invalid.
+    If the agent is running in insecure mode, the request will be allowed without a key.
+    """
+
+    @functools.wraps(handler)
+    async def admin_auth(request):
+        context: AdminRequestContext = request["context"]
+        profile = context.profile
+        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"):
+            return await handler(request)
+        else:
+            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:
+    - check for a valid bearer token in the Autorization header if running
+    in multi-tenant mode
+    - check for a valid x-api-key header if running in single-tenant mode
+    """
+
+    @functools.wraps(handler)
+    async def tenant_auth(request):
+        context: AdminRequestContext = request["context"]
+        profile = context.profile
+        authorization_header = request.headers.get("Authorization")
+        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"))
+        multitenant_enabled = profile.settings.get("multitenant.enabled")
+
+        # 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 insecure_mode
+            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",
+            )
+
+    return tenant_auth

aries_cloudagent/admin/routes.py

@@ -0,0 +1,232 @@
+"""Admin server routes."""
+
+import asyncio
+import re
+
+from aiohttp import web
+from aiohttp_apispec import (
+    docs,
+    response_schema,
+)
+from marshmallow import fields
+
+from ..core.plugin_registry import PluginRegistry
+from ..messaging.models.openapi import OpenAPISchema
+from ..utils.stats import Collector
+from ..version import __version__
+from .decorators.auth import admin_authentication
+
+
+class AdminModulesSchema(OpenAPISchema):
+    """Schema for the modules endpoint."""
+
+    result = fields.List(
+        fields.Str(metadata={"description": "admin module"}),
+        metadata={"description": "List of admin modules"},
+    )
+
+
+class AdminConfigSchema(OpenAPISchema):
+    """Schema for the config endpoint."""
+
+    config = fields.Dict(
+        required=True, metadata={"description": "Configuration settings"}
+    )
+
+
+class AdminStatusSchema(OpenAPISchema):
+    """Schema for the status endpoint."""
+
+    version = fields.Str(metadata={"description": "Version code"})
+    label = fields.Str(allow_none=True, metadata={"description": "Default label"})
+    timing = fields.Dict(required=False, metadata={"description": "Timing results"})
+    conductor = fields.Dict(
+        required=False, metadata={"description": "Conductor statistics"}
+    )
+
+
+class AdminResetSchema(OpenAPISchema):
+    """Schema for the reset endpoint."""
+
+
+class AdminStatusLivelinessSchema(OpenAPISchema):
+    """Schema for the liveliness endpoint."""
+
+    alive = fields.Boolean(
+        metadata={"description": "Liveliness status", "example": True}
+    )
+
+
+class AdminStatusReadinessSchema(OpenAPISchema):
+    """Schema for the readiness endpoint."""
+
+    ready = fields.Boolean(
+        metadata={"description": "Readiness status", "example": True}
+    )
+
+
+class AdminShutdownSchema(OpenAPISchema):
+    """Response schema for admin Module."""
+
+
+@docs(tags=["server"], summary="Fetch the list of loaded plugins")
+@response_schema(AdminModulesSchema(), 200, description="")
+@admin_authentication
+async def plugins_handler(request: web.BaseRequest):
+    """Request handler for the loaded plugins list.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The module list response
+
+    """
+    registry = request.app["context"].inject_or(PluginRegistry)
+    plugins = registry and sorted(registry.plugin_names) or []
+    return web.json_response({"result": plugins})
+
+
+@docs(tags=["server"], summary="Fetch the server configuration")
+@response_schema(AdminConfigSchema(), 200, description="")
+@admin_authentication
+async def config_handler(request: web.BaseRequest):
+    """Request handler for the server configuration.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response
+
+    """
+    config = {
+        k: (
+            request.app["context"].settings[k]
+            if (isinstance(request.app["context"].settings[k], (str, int)))
+            else request.app["context"].settings[k].copy()
+        )
+        for k in request.app["context"].settings
+        if k
+        not in [
+            "admin.admin_api_key",
+            "multitenant.jwt_secret",
+            "wallet.key",
+            "wallet.rekey",
+            "wallet.seed",
+            "wallet.storage_creds",
+        ]
+    }
+    for index in range(len(config.get("admin.webhook_urls", []))):
+        config["admin.webhook_urls"][index] = re.sub(
+            r"#.*",
+            "",
+            config["admin.webhook_urls"][index],
+        )
+
+    return web.json_response({"config": config})
+
+
+@docs(tags=["server"], summary="Fetch the server status")
+@response_schema(AdminStatusSchema(), 200, description="")
+@admin_authentication
+async def status_handler(request: web.BaseRequest):
+    """Request handler for the server status information.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response
+
+    """
+    status = {"version": __version__}
+    status["label"] = request.app["context"].settings.get("default_label")
+    collector = request.app["context"].inject_or(Collector)
+    if collector:
+        status["timing"] = collector.results
+    if request.app["conductor_stats"]:
+        status["conductor"] = await request.app["conductor_stats"]()
+    return web.json_response(status)
+
+
+@docs(tags=["server"], summary="Reset statistics")
+@response_schema(AdminResetSchema(), 200, description="")
+@admin_authentication
+async def status_reset_handler(request: web.BaseRequest):
+    """Request handler for resetting the timing statistics.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response
+
+    """
+    collector = request.app["context"].inject_or(Collector)
+    if collector:
+        collector.reset()
+    return web.json_response({})
+
+
+async def redirect_handler(request: web.BaseRequest):
+    """Perform redirect to documentation."""
+    raise web.HTTPFound("/api/doc")
+
+
+@docs(tags=["server"], summary="Liveliness check")
+@response_schema(AdminStatusLivelinessSchema(), 200, description="")
+async def liveliness_handler(request: web.BaseRequest):
+    """Request handler for liveliness check.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response, always indicating True
+
+    """
+    app_live = request.app._state["alive"]
+    if app_live:
+        return web.json_response({"alive": app_live})
+    else:
+        raise web.HTTPServiceUnavailable(reason="Service not available")
+
+
+@docs(tags=["server"], summary="Readiness check")
+@response_schema(AdminStatusReadinessSchema(), 200, description="")
+async def readiness_handler(request: web.BaseRequest):
+    """Request handler for liveliness check.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response, indicating readiness for further calls
+
+    """
+    app_ready = request.app._state["ready"] and request.app._state["alive"]
+    if app_ready:
+        return web.json_response({"ready": app_ready})
+    else:
+        raise web.HTTPServiceUnavailable(reason="Service not ready")
+
+
+@docs(tags=["server"], summary="Shut down server")
+@response_schema(AdminShutdownSchema(), description="")
+@admin_authentication
+async def shutdown_handler(request: web.BaseRequest):
+    """Request handler for server shutdown.
+
+    Args:
+        request: aiohttp request object
+
+    Returns:
+        The web response (empty production)
+
+    """
+    request.app._state["ready"] = False
+    loop = asyncio.get_event_loop()
+    asyncio.ensure_future(request.app["conductor_stop"](), loop=loop)
+
+    return web.json_response({})