214 lines
7.7 KiB
Python
214 lines
7.7 KiB
Python
"""Role-based access control (RBAC) — MIRROR of main backend's
|
||
``app/core/auth.py``.
|
||
|
||
Kept in sync manually; rationale: separate stacks, shared YAML config mounted
|
||
from repo root. We deliberately do NOT share code between repos via
|
||
``sys.path`` tricks — each docker stack ships its own image with its own
|
||
``app/`` tree.
|
||
|
||
When updating one copy, update the other.
|
||
|
||
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
|
||
"""
|
||
|
||
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]
|
||
# #657 white-label: brand slug, привязанный к аккаунту. None = generic UI.
|
||
# Фронт (useBrand.ts) применяет бренд автоматически на login (без ?brand=).
|
||
brand: str | None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Account → brand mapping (#657 brand-by-account)
|
||
# ---------------------------------------------------------------------------
|
||
#
|
||
# Брендирование «Практика» (#657 brand-by-account) применяется ТОЛЬКО под
|
||
# аккаунтом `praktika`. Резолвится в /me как UserScope.brand, а при генерации
|
||
# trade-in PDF — по `created_by` оценки (анти-spoofing #7, бренд не хранится).
|
||
# Остальные юзеры (вкл. `kopylov`) → None (стандартный generic UI/PDF). Slug
|
||
# должен существовать в таблице `brands` (см. services/brand.py) — иначе
|
||
# useBrand fallback на generic.
|
||
#
|
||
# `kopylov` ранее мапился на бренд `praktika`, но это приводило к утечке лого
|
||
# «Практики» в отчёты, созданные под `kopylov`, не тем клиентам (инцидент:
|
||
# отчёт для Брусники с лого Практики) — маппинг удалён. NB: kopylov/praktika
|
||
# остаются в роли `pilot` в roles.yaml — это РОЛЬ, не бренд, не трогаем.
|
||
_USERNAME_BRAND: dict[str, str] = {
|
||
"praktika": "praktika",
|
||
}
|
||
|
||
|
||
def get_brand_for_user(username: str) -> str | None:
|
||
"""Return the brand slug bound to *username*, or None for generic UI."""
|
||
return _USERNAME_BRAND.get(username)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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: walk up from this file to find the first ancestor containing
|
||
`auth/`. Tradein-mvp lives inside the main repo, so the same walk finds
|
||
`<repo>/auth/roles.yaml`.
|
||
"""
|
||
container_path = Path("/app/auth/roles.yaml")
|
||
if container_path.is_file():
|
||
return container_path
|
||
|
||
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'")
|
||
|
||
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:
|
||
`/**` → matches everything (admin scope).
|
||
`/foo/**` → matches `/foo`, `/foo/`, `/foo/bar`, `/foo/bar/baz`.
|
||
`/foo/*` → matches one segment after `/foo/`.
|
||
`/foo` → matches exactly `/foo`.
|
||
"""
|
||
dstar = "\x00DSTAR\x00"
|
||
sstar = "\x00SSTAR\x00"
|
||
|
||
work = pattern.replace("**", dstar).replace("*", sstar)
|
||
regex = re.escape(work)
|
||
regex = regex.replace(re.escape(dstar), ".*")
|
||
regex = regex.replace(re.escape(sstar), "[^/]*")
|
||
|
||
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*.
|
||
|
||
Allowed iff matches `paths` AND not in `deny`. `deny` is final.
|
||
"""
|
||
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 []))
|
||
|
||
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 []),
|
||
brand=get_brand_for_user(username),
|
||
)
|
||
|
||
|
||
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()
|