ObsiGate/backend/auth/scope.py

# backend/auth/scope.py
# Scoped API/MCP tokens (#199, synchro TikTak — Gitea #2).
#
# A token may carry an endpoint perimeter in its ``scope`` JWT claim:
# - ``"full"`` (default, and the case of every token issued before #199):
#   no restriction — today's behaviour, unchanged ;
# - ``"files"``: file read/write surface only (vault file CRUD, listings,
#   search, backups read/restore). Everything else (admin, config, AI,
#   webhooks, shares, auth management…) answers **403**.
#
# Vault restriction is orthogonal and needs no new machinery: a scoped token
# embeds the allowed vault list in its ``vaults`` JWT claim, so the existing
# ``check_vault_access`` (REST routes) and ``ToolContext.require_vault_access``
# (MCP/agent tools) refuse other vaults with 403 — for REST *and* MCP alike.
"""Endpoint perimeter for scoped API tokens."""

from __future__ import annotations

from typing import Any

SCOPE_FULL = "full"
SCOPE_FILES = "files"
KNOWN_SCOPES = (SCOPE_FULL, SCOPE_FILES)

# (methods, path prefix or exact path). ``methods`` None = any method.
# Keep this list tight: a sync client (TikTak) needs file CRUD, listings,
# search and backup read/restore — nothing else.
_RULES_FOR_FILES: tuple[tuple[frozenset[str] | None, str, bool], ...] = (
    # (methods, path, exact_only)
    (frozenset({"GET"}), "/api/health", False),
    (frozenset({"GET"}), "/api/auth/me", True),
    (frozenset({"GET"}), "/api/vaults", True),
    (None, "/api/browse/", False),
    (None, "/api/file/", False),
    (None, "/api/media/", False),
    (None, "/api/directory/", False),
    (None, "/api/move/", False),
    (None, "/api/vault/", False),  # {vault}/files, {vault}/paths, batch-upload…
    (frozenset({"GET"}), "/api/search", False),
    (frozenset({"GET"}), "/api/tree-search", True),
    (frozenset({"GET"}), "/api/suggest", False),
    (frozenset({"GET"}), "/api/tags", False),
    (frozenset({"GET"}), "/api/graph/", False),
    (frozenset({"GET"}), "/api/backups", False),
    (frozenset({"GET"}), "/docs", False),
    (frozenset({"GET"}), "/redoc", True),
    (frozenset({"GET"}), "/openapi.json", True),
    (None, "/mcp", False),
)


def is_path_allowed_for_scope(scope: str | None, method: str, path: str) -> bool:
    """True when a token with *scope* may call (*method*, *path*).

    - ``None``/``"full"`` → everything allowed (backward compatible).
    - ``"files"`` → allow-list above (API surface only ; static assets and
      public pages carry no data and are handled by the caller — the
      middleware only gates ``/api/*`` and ``/mcp*``).
    - unknown scope → fail-closed (deny).
    """
    if scope in (None, SCOPE_FULL):
        return True
    if scope != SCOPE_FILES:
        return False
    if not (path.startswith("/api/") or path.startswith("/mcp")):
        return True
    method = (method or "GET").upper()
    for methods, rule_path, exact in _RULES_FOR_FILES:
        if methods is not None and method not in methods:
            continue
        if exact:
            if path == rule_path:
                return True
        elif path == rule_path or path.startswith(rule_path):
            return True
    return False


def scope_of_user(user: dict[str, Any] | None) -> str:
    """Endpoint scope attached to an authenticated user (default ``"full"``)."""
    if not user:
        return SCOPE_FULL
    return user.get("_token_scope") or SCOPE_FULL


class TokenScopeMiddleware:
    """Pure-ASGI middleware enforcing the endpoint perimeter of scoped tokens.

    Denies out-of-perimeter calls made with a scoped token (**403**) before
    they reach the routes. Only *adds* restrictions: the ``scope`` claim is
    read unverified here, and the signature is still verified downstream by
    ``get_current_user`` (an invalid token is rejected with 401 a step later
    anyway). Full-scope and scopeless tokens pass through untouched.
    """

    def __init__(self, app: Any):
        self.app = app

    async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
        if scope.get("type") == "http":
            method = scope.get("method", "GET")
            path = scope.get("path", "")
            if path.startswith("/api/") or path.startswith("/mcp"):
                token_scope = _scope_from_asgi(scope)
                if token_scope not in (None, SCOPE_FULL) and not is_path_allowed_for_scope(
                    token_scope, method, path
                ):
                    from starlette.responses import JSONResponse

                    response = JSONResponse(
                        {"detail": "Token à portée limitée : endpoint hors périmètre 'files'"},
                        status_code=403,
                    )
                    await response(scope, receive, send)
                    return
        await self.app(scope, receive, send)


def _scope_from_asgi(asgi_scope: Any) -> str | None:
    """Extract the unverified ``scope`` claim from an ASGI scope's auth."""
    headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in asgi_scope.get("headers", [])}
    token: str | None = None
    auth = headers.get("authorization", "")
    if auth.lower().startswith("bearer "):
        token = auth[7:].strip()
    else:
        for part in headers.get("cookie", "").split(";"):
            name, _, value = part.strip().partition("=")
            if name.strip() == "access_token" and value.strip():
                token = value.strip()
                break
    if not token:
        return None
    try:
        from jose import jwt as _jwt

        claims = _jwt.get_unverified_claims(token)
    except Exception:
        return None
    token_scope = (claims or {}).get("scope")
    return token_scope if isinstance(token_scope, str) and token_scope else None