gendesign/tradein-mvp/backend/app/core/auth.py

"""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.

⚠️ РАСХОЖДЕНИЕ С ЗЕРКАЛОМ (#3316, намеренное — не «синхронизировать» обратно):
здесь `get_role` резолвит роль СНАЧАЛА из реестра людей (`tradein_users.role` /
`auth.users.role`), и только потом из YAML. У основного бэкенда реестра нет,
там копия остаётся YAML-only.

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, cast

import yaml

logger = logging.getLogger(__name__)

# legacy roles.yaml-роли + роли реестра ('admin'|'manager'|'employee', CHECK
# tradein м.192 / auth м.004). Оба набора приходят из одного `get_role` (#3316).
Role = Literal["admin", "pilot", "analyst", "expired", "manager", "employee"]


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
    # #2046 real profile fields для TopNav (фамилия/орг/email). None = фронт
    # фолбэкается на username/role (как раньше) — фолбэк остаётся на фронте.
    display_name: str | None
    org: str | None
    email: str | None


# ---------------------------------------------------------------------------
# Account → brand mapping (#657 brand-by-account)
# ---------------------------------------------------------------------------
#
# White-label бренд «Практика» снят целиком (бизнес-решение 2026-06-19): строка
# удалена из таблицы `brands` (миграция 121), SVG-логотип убран из фронта.
# Маппинг аккаунт→бренд теперь пустой — все юзеры резолвятся в None (generic
# UI/PDF). Механизм сохранён для будущих white-label клиентов: добавить slug
# сюда + строку в `brands` (см. services/brand.py). NB: аккаунты praktika/
# kopylov остаются пилот-логинами (роль `pilot` в roles.yaml) — это РОЛЬ, не
# бренд, не трогаем.
_USERNAME_BRAND: dict[str, str] = {}


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)


# ---------------------------------------------------------------------------
# Account → profile mapping (#2046 real profile fields for TopNav)
# ---------------------------------------------------------------------------
#
# Реальные имя/организация/email пилотов для TopNav-меню (замена фабрикованного
# "username as name"). Известные данные — только display_name для kopylov
# (фамилия «Копылов» из комментария в caddy/users.caddy.snippet). org/email для
# него не задокументированы нигде — умышленно НЕ выдумываем, оставляем None,
# фронт держит текущий фолбэк (username / brand ?? role / "").
_USERNAME_PROFILE: dict[str, dict[str, str]] = {
    "kopylov": {"display_name": "Копылов"},
}


def get_profile_for_user(username: str) -> tuple[str | None, str | None, str | None]:
    """Return (display_name, org, email) bound to *username*, or all-None."""
    profile = _USERNAME_PROFILE.get(username, {})
    return profile.get("display_name"), profile.get("org"), profile.get("email")


# ---------------------------------------------------------------------------
# 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 yaml_role(username: str) -> Role | None:
    """Роль из roles.yaml (без похода в реестр) или None, если юзера там нет.

    Нужна там, где спрашивают именно про legacy-файл, а не про эффективную роль:
    `team.create_employee` (#3316) не даёт занять имя, за которым в YAML уже
    числятся права.
    """
    users: dict[str, Role] = _load_roles_config()["users"]
    return users.get(username)


def _registry_role(username: str) -> str | None:
    """Роль из реестра людей (`tradein_users.role` / `auth.users.role`) или None.

    None означает «реестр про этого юзера ничего не сказал»: строки нет, роль
    пустая, либо реестр вообще недоступен. Во всех трёх случаях решение
    остаётся за roles.yaml — падение БД не имеет права выключить legacy-вход.

    ⚠️ Осознанный компромисс (#3316 review): последняя ветка — недоступный
    реестр — на время сбоя ВОЗВРАЩАЕТ авторитетность roles.yaml, то есть ровно
    то состояние, которое этот фикс и лечит. Сегодня это безопасно: коллизий
    имён между реестром и YAML на проде нет, а новые закрыты 409-гвардом в
    `team.create_employee`. Если коллизия всё же появится (ручной INSERT в
    реестр, расширение roles.yaml) — сбой БД станет окном эскалации, и тогда
    эту ветку надо менять на fail-closed (отказ вместо YAML-роли), а не
    дописывать проверки у вызывающих.

    Имя таблицы берётся из фиксированного словаря `identity_schema()`, значение
    едет bind-параметром: снаружи в SQL не попадает ничего.
    """
    try:
        from sqlalchemy import text

        from app.services.identity_store import identity_schema, identity_session

        schema = identity_schema()
        with identity_session() as db:
            row = db.execute(
                text(f"SELECT role FROM {schema.users_table} WHERE username = :username"),
                {"username": username},
            ).fetchone()
    except Exception:
        logger.exception(
            "registry role lookup failed for %r — fallback to roles.yaml",
            username,
        )
        return None
    if row is None or not row.role:
        return None
    return str(row.role)


def get_role(username: str) -> Role:
    """Эффективная роль *username*: реестр (БД) первый, roles.yaml — fallback.

    Raises KeyError, если юзера нет ни там, ни там.

    #3316: раньше роль резолвилась ТОЛЬКО из roles.yaml, при том что люди
    заводятся в БД (`tradein_users`) — два дефекта разом. Вверх: сотрудник,
    чьё имя совпало с YAML-админом, получал admin (IDOR по чужим оценкам +
    безлимит квоты). Вниз: сотрудник, которого в YAML нет, получал KeyError →
    403 на СОБСТВЕННУЮ оценку. Единственный источник истины теперь один, и он
    здесь — вызывающие (rbac, trade_in, team, account_quota) не меняются.
    """
    db_role = _registry_role(username)
    if db_role is not None:
        return cast(Role, db_role)

    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"].get(role)
    if role_def is None:
        # Роль реестра (employee/manager) — её scope живёт в DB_ROLE_PATHS, а не
        # в roles.yaml (#3316: get_role теперь может вернуть и такую роль).
        from app.services.auth_session import get_db_role_scope

        allowed_paths, deny_paths = get_db_role_scope(role)
    else:
        allowed_paths = list(role_def.get("paths", []) or [])
        deny_paths = list(role_def.get("deny", []) or [])
    display_name, org, email = get_profile_for_user(username)
    return UserScope(
        username=username,
        role=role,
        allowed_paths=allowed_paths,
        deny_paths=deny_paths,
        brand=get_brand_for_user(username),
        display_name=display_name,
        org=org,
        email=email,
    )


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()