From 88cdfd6adba37082874a4225b385bef0a1b9fabf Mon Sep 17 00:00:00 2001 From: Light1YT Date: Tue, 26 May 2026 06:18:40 +0000 Subject: [PATCH] feat(rbac): role-based access control via X-Authenticated-User middleware (#585) Backend RBAC + Caddy header_up. Closes part of #fixes-rbac-pra. --- Caddyfile | 24 +- auth/roles.yaml | 53 +++++ backend/app/api/v1/me.py | 48 ++++ backend/app/core/auth.py | 196 +++++++++++++++++ backend/app/main.py | 63 +++++- backend/pyproject.toml | 1 + backend/tests/test_rbac.py | 293 +++++++++++++++++++++++++ backend/uv.lock | 2 + docker-compose.prod.yml | 5 + tradein-mvp/backend/app/api/v1/me.py | 46 ++++ tradein-mvp/backend/app/core/auth.py | 185 ++++++++++++++++ tradein-mvp/backend/app/main.py | 65 +++++- tradein-mvp/backend/pyproject.toml | 1 + tradein-mvp/backend/tests/test_rbac.py | 282 ++++++++++++++++++++++++ tradein-mvp/docker-compose.prod.yml | 5 + 15 files changed, 1253 insertions(+), 16 deletions(-) create mode 100644 auth/roles.yaml create mode 100644 backend/app/api/v1/me.py create mode 100644 backend/app/core/auth.py create mode 100644 backend/tests/test_rbac.py create mode 100644 tradein-mvp/backend/app/api/v1/me.py create mode 100644 tradein-mvp/backend/app/core/auth.py create mode 100644 tradein-mvp/backend/tests/test_rbac.py diff --git a/Caddyfile b/Caddyfile index 734bfcb1..db704de8 100644 --- a/Caddyfile +++ b/Caddyfile @@ -74,7 +74,9 @@ gendsgn.ru { # FastAPI router замаунтен на /api/v1/trade-in/* — нужен strip только # префикса basePath /trade-in (Next.js basePath leak). uri strip_prefix /trade-in - reverse_proxy tradein-backend:8000 + reverse_proxy tradein-backend:8000 { + header_up X-Authenticated-User {http.auth.user.id} + } } # Matcher `path /trade-in /trade-in/*` ловит И /trade-in (без слеша), @@ -83,15 +85,21 @@ gendsgn.ru { @tradein path /trade-in /trade-in/* handle @tradein { # Next.js basePath=/trade-in — фронт сам ждёт префикса в URL - reverse_proxy tradein-frontend:3000 + reverse_proxy tradein-frontend:3000 { + header_up X-Authenticated-User {http.auth.user.id} + } } handle /api/* { - reverse_proxy backend:8000 + reverse_proxy backend:8000 { + header_up X-Authenticated-User {http.auth.user.id} + } } handle { - reverse_proxy frontend:3000 + reverse_proxy frontend:3000 { + header_up X-Authenticated-User {http.auth.user.id} + } } } } @@ -160,11 +168,15 @@ git.gendsgn.ru { import caddy/users.caddy.snippet handle /api/* { - reverse_proxy backend:8000 + reverse_proxy backend:8000 { + header_up X-Authenticated-User {http.auth.user.id} + } } handle { - reverse_proxy frontend:3000 + reverse_proxy frontend:3000 { + header_up X-Authenticated-User {http.auth.user.id} + } } } } diff --git a/auth/roles.yaml b/auth/roles.yaml new file mode 100644 index 00000000..71ee8e37 --- /dev/null +++ b/auth/roles.yaml @@ -0,0 +1,53 @@ +# RBAC roles + user → role mapping. +# +# Single source of truth for both main backend (backend/app/core/auth.py) and +# tradein backend (tradein-mvp/backend/app/core/auth.py). Mounted into both +# containers at /app/auth/roles.yaml via docker-compose bind-mount. +# +# Glob semantics (fnmatch-based, see app.core.auth.is_path_allowed): +# - "/**" → matches any path (admin scope). +# - "/foo/**" → matches /foo, /foo/, /foo/bar, /foo/bar/baz/... +# - "/foo" → matches exactly /foo. +# +# A path is allowed iff (1) it matches one of `paths` AND (2) it does NOT match +# any pattern in `deny`. `deny` overrides `paths` (deny-by-default within +# match scope). +# +# Username list MUST match exactly the 12 entries in caddy/users.caddy.snippet +# — Caddy basic_auth sets X-Authenticated-User from {http.auth.user.id}, and +# unknown users hit /me with 403 ("user not in roles config"). +# +# Last updated: 2026-05-26 (PR A — RBAC). + +roles: + admin: + paths: + - "/**" + deny: [] + pilot: + paths: + - "/" + - "/analytics/**" + - "/site-finder/**" + - "/trade-in/**" + - "/concept/**" + - "/api/v1/**" + - "/trade-in/api/v1/**" + deny: + - "/admin/**" + - "/api/v1/admin/**" + - "/trade-in/api/v1/admin/**" + +users: + admin: admin + kopylov: pilot + user1: pilot + user2: pilot + user3: pilot + user4: pilot + user5: pilot + user6: pilot + user7: pilot + user8: pilot + user9: pilot + user10: pilot diff --git a/backend/app/api/v1/me.py b/backend/app/api/v1/me.py new file mode 100644 index 00000000..d1a637df --- /dev/null +++ b/backend/app/api/v1/me.py @@ -0,0 +1,48 @@ +"""GET /me — отдаёт текущего пользователя и его RBAC-scope. + +Caddy basic_auth пропускает `X-Authenticated-User: ` через +`header_up` в каждом reverse_proxy. Frontend дёргает /api/v1/me чтобы +понять кому что показывать (admin-only кнопки, etc). + +Failure modes: + - header отсутствует → 401 (некорректная Caddy-конфигурация или прямой + запрос мимо basic_auth, не должно происходить в проде) + - header есть, но username не в roles.yaml → 403 (Caddy basic_auth знает + юзера, а RBAC config — нет; конфигурации рассинхронизированы) +""" + +from __future__ import annotations + +import logging +from typing import Annotated + +from fastapi import APIRouter, Header, HTTPException + +from app.core.auth import UserScope, get_user_scope + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.get("/me") +async def me( + x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, +) -> UserScope: + """Return the current user's RBAC scope (role + allowed/deny paths).""" + if not x_authenticated_user: + raise HTTPException( + status_code=401, + detail="no authenticated user (Caddy basic_auth required)", + ) + try: + return get_user_scope(x_authenticated_user) + except KeyError: + logger.warning( + "user %r authenticated via Caddy but missing from roles.yaml", + x_authenticated_user, + ) + raise HTTPException( + status_code=403, + detail="user not in roles config", + ) from None diff --git a/backend/app/core/auth.py b/backend/app/core/auth.py new file mode 100644 index 00000000..d0d2560e --- /dev/null +++ b/backend/app/core/auth.py @@ -0,0 +1,196 @@ +"""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() diff --git a/backend/app/main.py b/backend/app/main.py index 07071a70..3fef12a1 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,11 +1,13 @@ import logging import os -from collections.abc import AsyncIterator +import re +from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager import sentry_sdk -from fastapi import FastAPI +from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse, Response from sentry_sdk.integrations.celery import CeleryIntegration from sentry_sdk.integrations.fastapi import FastApiIntegration from sentry_sdk.integrations.httpx import HttpxIntegration @@ -24,12 +26,14 @@ from app.api.v1 import ( concepts, custom_pois, landing, + me, parcels, photos, pilot, trade_in, users, ) +from app.core.auth import get_role from app.core.config import settings from app.observability.sentry_scrub import scrub_sensitive_query @@ -70,6 +74,60 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app = FastAPI(title="GenDesign API", version="0.1.0", lifespan=lifespan) +# RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User +# (см. app/core/auth.py + auth/roles.yaml). Правила: +# 1) Любой non-public path требует X-Authenticated-User — иначе 401 +# (Caddy пробрасывает заголовок только для авторизованных юзеров; нет +# заголовка = локальный curl мимо Caddy = доступ запрещён). +# 2) Юзер должен быть в roles.yaml — иначе 403 («неизвестный юзер ничего +# не видит»). Это закрывает кейс если Caddy basic_auth пропустил кого-то +# из снятой записи, или прокси-подмена заголовка. +# 3) /api/v1/admin/* — только role=admin, иначе 403. +# Public paths без auth (/health, /docs, /openapi.json) пропускаем без проверки — +# X-Authenticated-User там просто не приходит из Caddy. +_ADMIN_API_RE = re.compile(r"^/api/v1/admin/") +_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"}) + + +@app.middleware("http") +async def rbac_guard( + request: Request, + call_next: Callable[[Request], Awaitable[Response]], +) -> Response: + path = request.url.path + if path in _PUBLIC_PATHS: + return await call_next(request) + + username = request.headers.get("X-Authenticated-User") + if not username: + # Любой non-public path без auth-header → 401. Локальный curl мимо Caddy + # или прокси-фронт без header_up. 401 точнее чем 403 — "сначала + # аутентифицируйся". + return JSONResponse( + status_code=401, + content={"detail": "no authenticated user (Caddy basic_auth required)"}, + ) + + try: + role = get_role(username) + except KeyError: + # Юзер в Caddy basic_auth, но не в roles.yaml → 403 на ВСЁ. + # Decided 2026-05-25: «человек без ролей вообще ничего не видит». + logger.warning("RBAC: unknown user %r tried %s", username, path) + return JSONResponse( + status_code=403, + content={"detail": "user not in roles config"}, + ) + + if _ADMIN_API_RE.match(path) and role != "admin": + logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path) + return JSONResponse( + status_code=403, + content={"detail": "admin only"}, + ) + return await call_next(request) + + app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, @@ -105,6 +163,7 @@ app.include_router(trade_in.router, prefix="/api/v1/trade-in", tags=["trade-in"] app.include_router(landing.router, prefix="/api/v1", tags=["landing"]) app.include_router(pilot.router, prefix="/api/v1/pilot", tags=["pilot"]) app.include_router(users.router, prefix="/api/v1", tags=["users"]) +app.include_router(me.router, prefix="/api/v1", tags=["me"]) @app.get("/health") diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 1bd36e47..8545756c 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ "sentry-sdk[fastapi,celery,sqlalchemy,httpx]>=2.18.0", "rosreestr2coord>=5.0.0", "ijson>=3.2.0", + "pyyaml>=6.0.0", # RBAC roles.yaml loader (app/core/auth.py) ] [dependency-groups] diff --git a/backend/tests/test_rbac.py b/backend/tests/test_rbac.py new file mode 100644 index 00000000..ca92b41b --- /dev/null +++ b/backend/tests/test_rbac.py @@ -0,0 +1,293 @@ +"""RBAC unit + integration tests. + +Coverage: + - get_role / get_user_scope happy + error paths + - is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**) + - GET /me — header missing / unknown user / pilot / admin + - rbac_guard middleware: + * любой non-public path требует X-Authenticated-User (no-header → 401) + * неизвестный юзер → 403 на ВСЁ (включая non-admin paths) — + «человек без ролей вообще ничего не видит» (decided 2026-05-25) + * /api/v1/admin/* — только role=admin (pilot → 403) + * public /health пропускаем без проверки + +Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не +тянуть тяжёлые модули из app.main (weasyprint требует gobject который не +ставится на macOS dev-машинах из коробки). RBAC middleware и /me router +импортируются напрямую — это даёт нам ровно тот же поведение что в проде. +""" + +from __future__ import annotations + +import re +from collections.abc import Awaitable, Callable + +import pytest +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse, Response +from fastapi.testclient import TestClient + +from app.api.v1 import me as me_router +from app.core import auth as auth_mod + + +@pytest.fixture(autouse=True) +def _reset_auth_cache() -> None: + """Каждый тест получает свежий YAML-кэш.""" + auth_mod.reset_cache_for_tests() + + +# --------------------------------------------------------------------------- +# Test app — копия rbac_guard из app/main.py, чтобы не подтягивать тяжёлые +# импорты (weasyprint, celery worker, ...). Если поведение middleware меняется +# в проде — синхронизируй здесь. +# --------------------------------------------------------------------------- + + +_ADMIN_API_RE = re.compile(r"^/api/v1/admin/") +_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"}) + + +def _build_test_app() -> FastAPI: + app = FastAPI() + + @app.middleware("http") + async def rbac_guard( + request: Request, + call_next: Callable[[Request], Awaitable[Response]], + ) -> Response: + path = request.url.path + if path in _PUBLIC_PATHS: + return await call_next(request) + + username = request.headers.get("X-Authenticated-User") + if not username: + return JSONResponse( + status_code=401, + content={"detail": "no authenticated user (Caddy basic_auth required)"}, + ) + try: + role = auth_mod.get_role(username) + except KeyError: + return JSONResponse( + status_code=403, + content={"detail": "user not in roles config"}, + ) + if _ADMIN_API_RE.match(path) and role != "admin": + return JSONResponse( + status_code=403, + content={"detail": "admin only"}, + ) + return await call_next(request) + + app.include_router(me_router.router, prefix="/api/v1", tags=["me"]) + + # Реальный admin endpoint, который должен быть заблокирован middleware'ом. + @app.get("/api/v1/admin/dummy") + async def admin_dummy() -> dict: + return {"ok": True} + + @app.get("/health") + async def health() -> dict: + return {"status": "ok"} + + return app + + +@pytest.fixture +def client() -> TestClient: + return TestClient(_build_test_app()) + + +# --------------------------------------------------------------------------- +# get_role +# --------------------------------------------------------------------------- + + +def test_get_role_known_users() -> None: + assert auth_mod.get_role("admin") == "admin" + assert auth_mod.get_role("kopylov") == "pilot" + for n in range(1, 11): + assert auth_mod.get_role(f"user{n}") == "pilot" + + +def test_get_role_unknown_user_raises() -> None: + with pytest.raises(KeyError): + auth_mod.get_role("nobody") + + +# --------------------------------------------------------------------------- +# is_path_allowed +# --------------------------------------------------------------------------- + + +def test_is_path_allowed_admin_everywhere() -> None: + assert auth_mod.is_path_allowed("admin", "/") + assert auth_mod.is_path_allowed("admin", "/admin") + assert auth_mod.is_path_allowed("admin", "/admin/scrape/runs") + assert auth_mod.is_path_allowed("admin", "/api/v1/admin/scrape/status") + assert auth_mod.is_path_allowed("admin", "/trade-in/api/v1/admin/scrape") + assert auth_mod.is_path_allowed("admin", "/concept/123") + assert auth_mod.is_path_allowed("admin", "/analytics/dashboard") + + +def test_is_path_allowed_pilot_excludes_admin() -> None: + # Pilot allowed paths + assert auth_mod.is_path_allowed("pilot", "/") + assert auth_mod.is_path_allowed("pilot", "/analytics/dashboard") + assert auth_mod.is_path_allowed("pilot", "/site-finder/123") + assert auth_mod.is_path_allowed("pilot", "/trade-in") + assert auth_mod.is_path_allowed("pilot", "/trade-in/123") + assert auth_mod.is_path_allowed("pilot", "/concept/abc") + assert auth_mod.is_path_allowed("pilot", "/api/v1/parcels/123") + assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/search") + + # Pilot DENIED — admin paths + assert not auth_mod.is_path_allowed("pilot", "/admin") + assert not auth_mod.is_path_allowed("pilot", "/admin/jobs") + assert not auth_mod.is_path_allowed("pilot", "/admin/scrape/runs/42") + assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/scrape/status") + assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/jobs") + assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape") + + +def test_is_path_allowed_unknown_role_denied() -> None: + assert not auth_mod.is_path_allowed("ghost", "/") + assert not auth_mod.is_path_allowed("ghost", "/admin") + + +# --------------------------------------------------------------------------- +# get_user_scope +# --------------------------------------------------------------------------- + + +def test_get_user_scope_admin() -> None: + scope = auth_mod.get_user_scope("admin") + assert scope["username"] == "admin" + assert scope["role"] == "admin" + assert "/**" in scope["allowed_paths"] + assert scope["deny_paths"] == [] + + +def test_get_user_scope_pilot() -> None: + scope = auth_mod.get_user_scope("kopylov") + assert scope["username"] == "kopylov" + assert scope["role"] == "pilot" + assert "/admin/**" in scope["deny_paths"] + assert "/api/v1/admin/**" in scope["deny_paths"] + + +def test_get_user_scope_unknown_raises() -> None: + with pytest.raises(KeyError): + auth_mod.get_user_scope("nobody") + + +# --------------------------------------------------------------------------- +# GET /me +# --------------------------------------------------------------------------- + + +def test_me_endpoint_admin(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["username"] == "admin" + assert body["role"] == "admin" + assert body["allowed_paths"] == ["/**"] + + +def test_me_endpoint_pilot(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "kopylov"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["username"] == "kopylov" + assert body["role"] == "pilot" + assert "/admin/**" in body["deny_paths"] + + +def test_me_endpoint_no_header_401(client: TestClient) -> None: + resp = client.get("/api/v1/me") + assert resp.status_code == 401 + assert "no authenticated user" in resp.json()["detail"].lower() + + +def test_me_endpoint_unknown_user_403(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "ghost"}) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +# --------------------------------------------------------------------------- +# rbac_guard middleware +# --------------------------------------------------------------------------- + + +def test_rbac_guard_pilot_blocked_on_admin_api(client: TestClient) -> None: + """Pilot, ходящий в /api/v1/admin/* — должен получить 403, даже если + target endpoint существует.""" + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "kopylov"}, + ) + assert resp.status_code == 403 + assert resp.json()["detail"] == "admin only" + + +def test_rbac_guard_admin_allowed(client: TestClient) -> None: + """Admin доходит до handler'а — 200 от admin_dummy.""" + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "admin"}, + ) + assert resp.status_code == 200 + assert resp.json() == {"ok": True} + + +def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None: + """Локальный curl без Caddy → /api/v1/admin/* отдаёт 401.""" + resp = client.get("/api/v1/admin/dummy") + assert resp.status_code == 401 + + +def test_rbac_guard_unknown_user_on_admin_path_returns_403(client: TestClient) -> None: + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "ghost"}, + ) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +def test_rbac_guard_skips_health(client: TestClient) -> None: + """/health публичный — middleware не должен потребовать заголовок.""" + resp = client.get("/health") + assert resp.status_code == 200 + assert resp.json()["status"] == "ok" + + +def test_rbac_guard_pilot_can_hit_non_admin_api(client: TestClient) -> None: + """Pilot должен спокойно ходить в обычные /api/v1/* (не admin).""" + resp = client.get( + "/api/v1/me", + headers={"X-Authenticated-User": "user1"}, + ) + assert resp.status_code == 200 + assert resp.json()["role"] == "pilot" + + +def test_rbac_guard_unknown_user_blocked_on_non_admin_path(client: TestClient) -> None: + """Неизвестный юзер (не в roles.yaml) → 403 даже на non-admin path. + «Человек без ролей вообще ничего не видит» (decided 2026-05-25).""" + resp = client.get( + "/api/v1/me", + headers={"X-Authenticated-User": "ghost"}, + ) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +def test_rbac_guard_no_header_on_non_admin_path_returns_401(client: TestClient) -> None: + """Любой non-public path без X-Authenticated-User → 401 (не только admin).""" + resp = client.get("/api/v1/me") + assert resp.status_code == 401 + assert "no authenticated user" in resp.json()["detail"].lower() diff --git a/backend/uv.lock b/backend/uv.lock index d1643ad0..d62ad8c1 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -578,6 +578,7 @@ dependencies = [ { name = "pydantic" }, { name = "pydantic-settings" }, { name = "pyproj" }, + { name = "pyyaml" }, { name = "redis" }, { name = "rosreestr2coord" }, { name = "scikit-learn" }, @@ -619,6 +620,7 @@ requires-dist = [ { name = "pydantic", specifier = ">=2.7.0" }, { name = "pydantic-settings", specifier = ">=2.3.0" }, { name = "pyproj", specifier = ">=3.6.0" }, + { name = "pyyaml", specifier = ">=6.0.0" }, { name = "redis", specifier = ">=5.0.0" }, { name = "rosreestr2coord", specifier = ">=5.0.0" }, { name = "scikit-learn", specifier = ">=1.5.0" }, diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index c4d1c455..5a6b1568 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -66,6 +66,11 @@ services: condition: service_healthy ports: - "127.0.0.1:8000:8000" + volumes: + # RBAC roles config — single source of truth (auth/roles.yaml в репо). + # FastAPI middleware и /api/v1/me читают этот файл при старте (см. + # app/core/auth.py). Read-only — мутация только через PR. + - ./auth/roles.yaml:/app/auth/roles.yaml:ro healthcheck: test: ["CMD", "curl", "-fsS", "http://localhost:8000/health"] interval: 30s diff --git a/tradein-mvp/backend/app/api/v1/me.py b/tradein-mvp/backend/app/api/v1/me.py new file mode 100644 index 00000000..4b7da093 --- /dev/null +++ b/tradein-mvp/backend/app/api/v1/me.py @@ -0,0 +1,46 @@ +"""GET /me — отдаёт текущего пользователя и его RBAC-scope. + +MIRROR of main backend's app/api/v1/me.py — kept in sync manually. +Mounted at /api/v1/me; через Caddy `uri strip_prefix /trade-in` это становится +`/trade-in/api/v1/me` снаружи. + +Caddy basic_auth пропускает `X-Authenticated-User: ` через +`header_up` в каждом reverse_proxy. Frontend дёргает /me чтобы понять +кому что показывать. +""" + +from __future__ import annotations + +import logging +from typing import Annotated + +from fastapi import APIRouter, Header, HTTPException + +from app.core.auth import UserScope, get_user_scope + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.get("/me") +async def me( + x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None, +) -> UserScope: + """Return the current user's RBAC scope (role + allowed/deny paths).""" + if not x_authenticated_user: + raise HTTPException( + status_code=401, + detail="no authenticated user (Caddy basic_auth required)", + ) + try: + return get_user_scope(x_authenticated_user) + except KeyError: + logger.warning( + "user %r authenticated via Caddy but missing from roles.yaml", + x_authenticated_user, + ) + raise HTTPException( + status_code=403, + detail="user not in roles config", + ) from None diff --git a/tradein-mvp/backend/app/core/auth.py b/tradein-mvp/backend/app/core/auth.py new file mode 100644 index 00000000..c7c88736 --- /dev/null +++ b/tradein-mvp/backend/app/core/auth.py @@ -0,0 +1,185 @@ +"""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: ` через +`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"] + + +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: walk up from this file to find the first ancestor containing + `auth/`. Tradein-mvp lives inside the main repo, so the same walk finds + `/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 []), + ) + + +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() diff --git a/tradein-mvp/backend/app/main.py b/tradein-mvp/backend/app/main.py index 293f7877..3043e21c 100644 --- a/tradein-mvp/backend/app/main.py +++ b/tradein-mvp/backend/app/main.py @@ -8,14 +8,17 @@ from __future__ import annotations import asyncio import logging -from collections.abc import AsyncGenerator +import re +from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager import sentry_sdk -from fastapi import FastAPI +from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse, Response -from app.api.v1 import admin, brand, geocode, search, trade_in +from app.api.v1 import admin, brand, geocode, me, search, trade_in +from app.core.auth import get_role from app.core.config import settings from app.core.db import SessionLocal from app.core.fdw import ensure_fdw_user_mapping @@ -35,8 +38,8 @@ if settings.glitchtip_dsn: sentry_sdk.init( dsn=settings.glitchtip_dsn, environment=settings.environment, - traces_sample_rate=0.0, # только ошибки, без performance-трейсов - send_default_pii=False, # не шлём client_name / client_phone в отчёты + traces_sample_rate=0.0, # только ошибки, без performance-трейсов + send_default_pii=False, # не шлём client_name / client_phone в отчёты ) logging.getLogger("app.main").info("GlitchTip monitoring enabled") @@ -49,9 +52,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: with SessionLocal() as db: ensure_fdw_user_mapping(db) except Exception: - logger.exception( - "FDW user mapping bootstrap failed — cadastral queries may fail" - ) + logger.exception("FDW user mapping bootstrap failed — cadastral queries may fail") # Startup: launch scheduler background task scheduler_task = asyncio.create_task(scheduler_loop()) @@ -73,6 +74,53 @@ app = FastAPI( lifespan=lifespan, ) +# RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User +# (см. app/core/auth.py + auth/roles.yaml). Правила: +# 1) Любой non-public path требует X-Authenticated-User — иначе 401. +# 2) Юзер должен быть в roles.yaml — иначе 403 («неизвестный юзер ничего +# не видит» — decided 2026-05-25). +# 3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy +# `uri strip_prefix /trade-in`) — только role=admin, иначе 403. +# Public paths без auth (/health, /docs, /openapi.json) пропускаем — +# X-Authenticated-User там не приходит из Caddy. +_ADMIN_API_RE = re.compile(r"^/api/v1/admin/") +_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"}) + + +@app.middleware("http") +async def rbac_guard( + request: Request, + call_next: Callable[[Request], Awaitable[Response]], +) -> Response: + path = request.url.path + if path in _PUBLIC_PATHS: + return await call_next(request) + + username = request.headers.get("X-Authenticated-User") + if not username: + return JSONResponse( + status_code=401, + content={"detail": "no authenticated user (Caddy basic_auth required)"}, + ) + + try: + role = get_role(username) + except KeyError: + logger.warning("RBAC: unknown user %r tried %s", username, path) + return JSONResponse( + status_code=403, + content={"detail": "user not in roles config"}, + ) + + if _ADMIN_API_RE.match(path) and role != "admin": + logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path) + return JSONResponse( + status_code=403, + content={"detail": "admin only"}, + ) + return await call_next(request) + + app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, @@ -94,3 +142,4 @@ app.include_router(admin.router, prefix="/api/v1/admin", tags=["admin"]) app.include_router(brand.router, prefix="/api/v1/brand", tags=["brand"]) app.include_router(trade_in.router, prefix="/api/v1/trade-in", tags=["trade-in"]) app.include_router(search.router, prefix="/api/v1", tags=["search"]) +app.include_router(me.router, prefix="/api/v1", tags=["me"]) diff --git a/tradein-mvp/backend/pyproject.toml b/tradein-mvp/backend/pyproject.toml index 014b143b..f540d91e 100644 --- a/tradein-mvp/backend/pyproject.toml +++ b/tradein-mvp/backend/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ "python-multipart>=0.0.9", # FastAPI UploadFile — загрузка фото квартиры (#394) "sentry-sdk>=2.0.0", # мониторинг ошибок → GlitchTip (#396) "redis>=5.0.0", # async hot cache для /api/v1/search (Phase 3.2) + "pyyaml>=6.0.0", # RBAC roles.yaml loader (app/core/auth.py) ] [dependency-groups] diff --git a/tradein-mvp/backend/tests/test_rbac.py b/tradein-mvp/backend/tests/test_rbac.py new file mode 100644 index 00000000..170911d0 --- /dev/null +++ b/tradein-mvp/backend/tests/test_rbac.py @@ -0,0 +1,282 @@ +"""RBAC unit + integration tests for tradein backend. + +MIRROR of backend/tests/test_rbac.py — kept in sync manually. + +Coverage: + - get_role / get_user_scope happy + error paths + - is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**) + - GET /me — header missing / unknown user / pilot / admin + - rbac_guard middleware: + * любой non-public path требует X-Authenticated-User (no-header → 401) + * неизвестный юзер → 403 на ВСЁ (включая non-admin paths) — + «человек без ролей вообще ничего не видит» (decided 2026-05-25) + * /api/v1/admin/* — только role=admin (pilot → 403) + +Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не +тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler). +RBAC middleware и /me router импортируются напрямую — это даёт нам ровно +тот же поведение что в проде. +""" + +from __future__ import annotations + +import re +from collections.abc import Awaitable, Callable + +import pytest +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse, Response +from fastapi.testclient import TestClient + +from app.api.v1 import me as me_router +from app.core import auth as auth_mod + + +@pytest.fixture(autouse=True) +def _reset_auth_cache() -> None: + auth_mod.reset_cache_for_tests() + + +# --------------------------------------------------------------------------- +# Test app — копия rbac_guard из app/main.py. +# --------------------------------------------------------------------------- + + +_ADMIN_API_RE = re.compile(r"^/api/v1/admin/") +_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"}) + + +def _build_test_app() -> FastAPI: + app = FastAPI() + + @app.middleware("http") + async def rbac_guard( + request: Request, + call_next: Callable[[Request], Awaitable[Response]], + ) -> Response: + path = request.url.path + if path in _PUBLIC_PATHS: + return await call_next(request) + + username = request.headers.get("X-Authenticated-User") + if not username: + return JSONResponse( + status_code=401, + content={"detail": "no authenticated user (Caddy basic_auth required)"}, + ) + try: + role = auth_mod.get_role(username) + except KeyError: + return JSONResponse( + status_code=403, + content={"detail": "user not in roles config"}, + ) + if _ADMIN_API_RE.match(path) and role != "admin": + return JSONResponse( + status_code=403, + content={"detail": "admin only"}, + ) + return await call_next(request) + + app.include_router(me_router.router, prefix="/api/v1", tags=["me"]) + + @app.get("/api/v1/admin/dummy") + async def admin_dummy() -> dict: + return {"ok": True} + + @app.get("/health") + async def health() -> dict: + return {"status": "ok"} + + return app + + +@pytest.fixture +def client() -> TestClient: + return TestClient(_build_test_app()) + + +# --------------------------------------------------------------------------- +# get_role +# --------------------------------------------------------------------------- + + +def test_get_role_known_users() -> None: + assert auth_mod.get_role("admin") == "admin" + assert auth_mod.get_role("kopylov") == "pilot" + for n in range(1, 11): + assert auth_mod.get_role(f"user{n}") == "pilot" + + +def test_get_role_unknown_user_raises() -> None: + with pytest.raises(KeyError): + auth_mod.get_role("nobody") + + +# --------------------------------------------------------------------------- +# is_path_allowed +# --------------------------------------------------------------------------- + + +def test_is_path_allowed_admin_everywhere() -> None: + assert auth_mod.is_path_allowed("admin", "/") + assert auth_mod.is_path_allowed("admin", "/admin") + assert auth_mod.is_path_allowed("admin", "/admin/scrape/runs") + assert auth_mod.is_path_allowed("admin", "/api/v1/admin/scrape/status") + assert auth_mod.is_path_allowed("admin", "/trade-in/api/v1/admin/scrape") + assert auth_mod.is_path_allowed("admin", "/concept/123") + assert auth_mod.is_path_allowed("admin", "/analytics/dashboard") + + +def test_is_path_allowed_pilot_excludes_admin() -> None: + assert auth_mod.is_path_allowed("pilot", "/") + assert auth_mod.is_path_allowed("pilot", "/analytics/dashboard") + assert auth_mod.is_path_allowed("pilot", "/site-finder/123") + assert auth_mod.is_path_allowed("pilot", "/trade-in") + assert auth_mod.is_path_allowed("pilot", "/trade-in/123") + assert auth_mod.is_path_allowed("pilot", "/concept/abc") + assert auth_mod.is_path_allowed("pilot", "/api/v1/parcels/123") + assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/search") + + assert not auth_mod.is_path_allowed("pilot", "/admin") + assert not auth_mod.is_path_allowed("pilot", "/admin/jobs") + assert not auth_mod.is_path_allowed("pilot", "/admin/scrape/runs/42") + assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/scrape/status") + assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/jobs") + assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape") + + +def test_is_path_allowed_unknown_role_denied() -> None: + assert not auth_mod.is_path_allowed("ghost", "/") + assert not auth_mod.is_path_allowed("ghost", "/admin") + + +# --------------------------------------------------------------------------- +# get_user_scope +# --------------------------------------------------------------------------- + + +def test_get_user_scope_admin() -> None: + scope = auth_mod.get_user_scope("admin") + assert scope["username"] == "admin" + assert scope["role"] == "admin" + assert "/**" in scope["allowed_paths"] + assert scope["deny_paths"] == [] + + +def test_get_user_scope_pilot() -> None: + scope = auth_mod.get_user_scope("kopylov") + assert scope["username"] == "kopylov" + assert scope["role"] == "pilot" + assert "/admin/**" in scope["deny_paths"] + assert "/api/v1/admin/**" in scope["deny_paths"] + + +def test_get_user_scope_unknown_raises() -> None: + with pytest.raises(KeyError): + auth_mod.get_user_scope("nobody") + + +# --------------------------------------------------------------------------- +# GET /me +# --------------------------------------------------------------------------- + + +def test_me_endpoint_admin(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["username"] == "admin" + assert body["role"] == "admin" + assert body["allowed_paths"] == ["/**"] + + +def test_me_endpoint_pilot(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "kopylov"}) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["username"] == "kopylov" + assert body["role"] == "pilot" + assert "/admin/**" in body["deny_paths"] + + +def test_me_endpoint_no_header_401(client: TestClient) -> None: + resp = client.get("/api/v1/me") + assert resp.status_code == 401 + assert "no authenticated user" in resp.json()["detail"].lower() + + +def test_me_endpoint_unknown_user_403(client: TestClient) -> None: + resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "ghost"}) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +# --------------------------------------------------------------------------- +# rbac_guard middleware +# --------------------------------------------------------------------------- + + +def test_rbac_guard_pilot_blocked_on_admin_api(client: TestClient) -> None: + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "kopylov"}, + ) + assert resp.status_code == 403 + assert resp.json()["detail"] == "admin only" + + +def test_rbac_guard_admin_allowed(client: TestClient) -> None: + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "admin"}, + ) + assert resp.status_code == 200 + assert resp.json() == {"ok": True} + + +def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None: + resp = client.get("/api/v1/admin/dummy") + assert resp.status_code == 401 + + +def test_rbac_guard_unknown_user_on_admin_path_returns_403(client: TestClient) -> None: + resp = client.get( + "/api/v1/admin/dummy", + headers={"X-Authenticated-User": "ghost"}, + ) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +def test_rbac_guard_skips_health(client: TestClient) -> None: + resp = client.get("/health") + assert resp.status_code == 200 + assert resp.json()["status"] == "ok" + + +def test_rbac_guard_pilot_can_hit_non_admin_api(client: TestClient) -> None: + resp = client.get( + "/api/v1/me", + headers={"X-Authenticated-User": "user1"}, + ) + assert resp.status_code == 200 + assert resp.json()["role"] == "pilot" + + +def test_rbac_guard_unknown_user_blocked_on_non_admin_path(client: TestClient) -> None: + """Неизвестный юзер → 403 даже на non-admin path. + «Человек без ролей вообще ничего не видит» (decided 2026-05-25).""" + resp = client.get( + "/api/v1/me", + headers={"X-Authenticated-User": "ghost"}, + ) + assert resp.status_code == 403 + assert "not in roles config" in resp.json()["detail"].lower() + + +def test_rbac_guard_no_header_on_non_admin_path_returns_401(client: TestClient) -> None: + """Любой non-public path без X-Authenticated-User → 401.""" + resp = client.get("/api/v1/me") + assert resp.status_code == 401 + assert "no authenticated user" in resp.json()["detail"].lower() diff --git a/tradein-mvp/docker-compose.prod.yml b/tradein-mvp/docker-compose.prod.yml index bded06f2..50c7af94 100644 --- a/tradein-mvp/docker-compose.prod.yml +++ b/tradein-mvp/docker-compose.prod.yml @@ -55,6 +55,11 @@ services: depends_on: postgres: condition: service_healthy + volumes: + # RBAC roles config — single source of truth (../auth/roles.yaml в репо, + # это /opt/gendesign/auth/roles.yaml на VPS). MIRROR of main backend mount. + # app/core/auth.py читает /app/auth/roles.yaml для middleware + /me. + - ../auth/roles.yaml:/app/auth/roles.yaml:ro restart: unless-stopped networks: - tradein-net