"""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: ` через `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"] 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: `/auth/roles.yaml` — backend code lives in `/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 /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()