gendesign/backend/app/core/auth.py

"""Role-based access control (RBAC) — load `auth/roles.yaml` and answer
authorisation questions.

Caddy gates the whole site with basic_auth (см. `caddy/users.caddy.snippet`)
и пропускает в backend заголовок `X-Authenticated-User: <username>` через
`header_up X-Authenticated-User {http.auth.user.id}` в каждом reverse_proxy.
Этот модуль читает yaml-конфиг **один раз при импорте** и отдаёт чистые
функции — middleware и endpoint /me потом гоняют их per-request.

Source-of-truth file:
  - container: /app/auth/roles.yaml (bind-mount из репо)
  - repo:     auth/roles.yaml

Если файл не найден — модуль падает на import (без RBAC сервис стартовать
не должен; лучше "не стартовать" чем "молча пропустить всех в /admin").

Mirror lives in tradein-mvp/backend/app/core/auth.py — синхронизация
руками; rationale в комментарии того файла.
"""

from __future__ import annotations

import logging
import re
from functools import lru_cache
from pathlib import Path
from typing import Literal, TypedDict

import yaml

logger = logging.getLogger(__name__)

Role = Literal["admin", "pilot", "analyst"]


class UserScope(TypedDict):
    """Возвращается /me — фронт может использовать allowed_paths для UI gating."""

    username: str
    role: Role
    allowed_paths: list[str]
    deny_paths: list[str]


# ---------------------------------------------------------------------------
# YAML loading
# ---------------------------------------------------------------------------


def _find_roles_yaml() -> Path:
    """Find `roles.yaml` either in the container mount path or in the repo root.

    Container layout (prod): `/app/auth/roles.yaml` (bind-mount).
    Dev layout: `<repo>/auth/roles.yaml` — backend code lives in `<repo>/backend/app`
    so we walk up from this file to find the first ancestor containing `auth/`.
    """
    container_path = Path("/app/auth/roles.yaml")
    if container_path.is_file():
        return container_path

    # Walk up from this file to find <repo>/auth/roles.yaml.
    here = Path(__file__).resolve()
    for ancestor in here.parents:
        candidate = ancestor / "auth" / "roles.yaml"
        if candidate.is_file():
            return candidate

    raise FileNotFoundError(
        "roles.yaml not found in /app/auth/ or in any ancestor of "
        f"{here} — RBAC cannot start without it"
    )


@lru_cache(maxsize=1)
def _load_roles_config() -> dict:
    """Parse `roles.yaml` once and cache the result for the process lifetime."""
    path = _find_roles_yaml()
    try:
        with path.open(encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
    except Exception:
        logger.exception("failed to parse roles.yaml at %s", path)
        raise

    if not isinstance(data, dict) or "roles" not in data or "users" not in data:
        raise ValueError(f"roles.yaml at {path} must have top-level keys 'roles' and 'users'")

    # Cheap sanity-check: each user maps to a role that exists.
    roles = data["roles"]
    users = data["users"]
    for username, role in users.items():
        if role not in roles:
            raise ValueError(
                f"user {username!r} maps to unknown role {role!r} (known: {list(roles)})"
            )

    logger.info(
        "RBAC loaded from %s — %d users, %d roles",
        path,
        len(users),
        len(roles),
    )
    return data


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------


def get_role(username: str) -> Role:
    """Return the role for *username* or raise KeyError if unknown."""
    config = _load_roles_config()
    users: dict[str, Role] = config["users"]
    if username not in users:
        raise KeyError(f"user {username!r} not in roles config")
    return users[username]


def _glob_to_regex(pattern: str) -> re.Pattern[str]:
    """Compile a glob pattern with `**` semantics to a regex.

    Semantics we want:
      `/**`          → matches everything (admin scope).
      `/foo/**`      → matches `/foo`, `/foo/`, `/foo/bar`, `/foo/bar/baz`, ...
      `/foo/*`       → matches one segment after `/foo/` (no slashes).
      `/foo`         → matches exactly `/foo`.

    Python's `fnmatch` doesn't distinguish `*` and `**` — we use a small custom
    translator with placeholders so we can do it ourselves without false sharing.
    """
    # Sentinels chosen to never appear in a real path or in glob metachars.
    dstar = "\x00DSTAR\x00"
    sstar = "\x00SSTAR\x00"

    work = pattern.replace("**", dstar).replace("*", sstar)
    regex = re.escape(work)
    # `**` → match anything (slashes allowed). `/**` at the end also matches the
    # parent dir without a trailing slash, e.g. `/admin/**` matches `/admin`.
    regex = regex.replace(re.escape(dstar), ".*")
    regex = regex.replace(re.escape(sstar), "[^/]*")

    # Special-case `/foo/**` → also match `/foo` (no trailing segment) so
    # `/admin/**` covers a bare `/admin` request — that's what callers expect.
    if regex.endswith("/.*"):
        regex = regex[: -len("/.*")] + r"(?:/.*)?"
    return re.compile(f"^{regex}$")


@lru_cache(maxsize=512)
def _compile_globs(patterns: tuple[str, ...]) -> tuple[re.Pattern[str], ...]:
    return tuple(_glob_to_regex(p) for p in patterns)


def is_path_allowed(role: str, path: str) -> bool:
    """Check whether *role* may access *path*.

    A path is allowed iff it matches at least one entry in `paths` AND does
    not match any entry in `deny`. `deny` is final — admin role uses an empty
    deny list so it always wins.
    """
    config = _load_roles_config()
    roles = config["roles"]
    if role not in roles:
        return False

    role_def = roles[role]
    allow_globs = _compile_globs(tuple(role_def.get("paths", [])))
    deny_globs = _compile_globs(tuple(role_def.get("deny", []) or []))

    # Deny overrides allow.
    if any(g.match(path) for g in deny_globs):
        return False
    return any(g.match(path) for g in allow_globs)


def get_user_scope(username: str) -> UserScope:
    """Return everything a frontend needs to do UI-level gating for *username*.

    Raises KeyError if *username* is unknown.
    """
    config = _load_roles_config()
    role = get_role(username)
    role_def = config["roles"][role]
    return UserScope(
        username=username,
        role=role,
        allowed_paths=list(role_def.get("paths", []) or []),
        deny_paths=list(role_def.get("deny", []) or []),
    )


def reset_cache_for_tests() -> None:
    """Drop cached YAML — only for unit tests that patch the file path."""
    _load_roles_config.cache_clear()
    _compile_globs.cache_clear()