feat(rbac): enforce role-based access via X-Authenticated-User middleware

Backend-side enforcement поверх Caddy basic_auth — defense-in-depth для
12-юзеров на gendsgn.ru (admin + kopylov + 10 пилотных слотов).

- `auth/roles.yaml` — single source of truth: 2 роли (admin, pilot)
  + 12 user→role mappings. Bind-mounted в обоих backend контейнерах.
- `app/core/auth.py` (main + tradein-mvp mirror) — YAML loader через
  `pyyaml`, `get_role`/`get_user_scope`/`is_path_allowed` с fnmatch globs.
  Custom `_glob_to_regex` чтобы `/**` ≠ `/*` (matches multi-segment paths).
- `GET /api/v1/me` (main + tradein) — фронт читает scope при login,
  фильтрует nav по `allowed_paths`/`deny_paths`.
- `rbac_guard` middleware (main + tradein):
  * любой non-public path требует `X-Authenticated-User` → иначе 401
  * юзер должен быть в roles.yaml → иначе 403 («человек без ролей
    вообще ничего не видит» — decided 2026-05-25)
  * /api/v1/admin/* — только role=admin, иначе 403
- `Caddyfile` — `header_up X-Authenticated-User {http.auth.user.id}` в
  6 reverse_proxy блоках (gendsgn.ru main + trade-in + git.gendsgn.ru).
- Tests: 20 unit + integration в обоих стэках. Покрытие:
  no-header / unknown user / pilot / admin × admin-path / non-admin path.
  `_build_test_app()` копирует middleware inline, чтобы обойти heavy
  imports (weasyprint dlopen падает на macOS dev).

Public paths (/health, /docs, /redoc, /openapi.json) пропускаются —
X-Authenticated-User там просто не приходит из Caddy.

Frontend nav-filter + 403 fullscreen — отдельный PR B.
This commit is contained in:
Light1YT 2026-05-26 11:14:47 +05:00
parent fc12b2c160
commit 3651c85f9b
15 changed files with 1253 additions and 16 deletions

View file

@ -74,7 +74,9 @@ gendsgn.ru {
# FastAPI router замаунтен на /api/v1/trade-in/* — нужен strip только # FastAPI router замаунтен на /api/v1/trade-in/* — нужен strip только
# префикса basePath /trade-in (Next.js basePath leak). # префикса basePath /trade-in (Next.js basePath leak).
uri strip_prefix /trade-in 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 (без слеша), # Matcher `path /trade-in /trade-in/*` ловит И /trade-in (без слеша),
@ -83,15 +85,21 @@ gendsgn.ru {
@tradein path /trade-in /trade-in/* @tradein path /trade-in /trade-in/*
handle @tradein { handle @tradein {
# Next.js basePath=/trade-in — фронт сам ждёт префикса в URL # 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/* { handle /api/* {
reverse_proxy backend:8000 reverse_proxy backend:8000 {
header_up X-Authenticated-User {http.auth.user.id}
}
} }
handle { 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 import caddy/users.caddy.snippet
handle /api/* { handle /api/* {
reverse_proxy backend:8000 reverse_proxy backend:8000 {
header_up X-Authenticated-User {http.auth.user.id}
}
} }
handle { handle {
reverse_proxy frontend:3000 reverse_proxy frontend:3000 {
header_up X-Authenticated-User {http.auth.user.id}
}
} }
} }
} }

53
auth/roles.yaml Normal file
View file

@ -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

48
backend/app/api/v1/me.py Normal file
View file

@ -0,0 +1,48 @@
"""GET /me — отдаёт текущего пользователя и его RBAC-scope.
Caddy basic_auth пропускает `X-Authenticated-User: <username>` через
`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

196
backend/app/core/auth.py Normal file
View file

@ -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: <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"]
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()

View file

@ -1,11 +1,13 @@
import logging import logging
import os import os
from collections.abc import AsyncIterator import re
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import sentry_sdk import sentry_sdk
from fastapi import FastAPI from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from sentry_sdk.integrations.celery import CeleryIntegration from sentry_sdk.integrations.celery import CeleryIntegration
from sentry_sdk.integrations.fastapi import FastApiIntegration from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.httpx import HttpxIntegration from sentry_sdk.integrations.httpx import HttpxIntegration
@ -24,12 +26,14 @@ from app.api.v1 import (
concepts, concepts,
custom_pois, custom_pois,
landing, landing,
me,
parcels, parcels,
photos, photos,
pilot, pilot,
trade_in, trade_in,
users, users,
) )
from app.core.auth import get_role
from app.core.config import settings from app.core.config import settings
from app.observability.sentry_scrub import scrub_sensitive_query 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) 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( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins=settings.cors_origins, 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(landing.router, prefix="/api/v1", tags=["landing"])
app.include_router(pilot.router, prefix="/api/v1/pilot", tags=["pilot"]) app.include_router(pilot.router, prefix="/api/v1/pilot", tags=["pilot"])
app.include_router(users.router, prefix="/api/v1", tags=["users"]) app.include_router(users.router, prefix="/api/v1", tags=["users"])
app.include_router(me.router, prefix="/api/v1", tags=["me"])
@app.get("/health") @app.get("/health")

View file

@ -31,6 +31,7 @@ dependencies = [
"sentry-sdk[fastapi,celery,sqlalchemy,httpx]>=2.18.0", "sentry-sdk[fastapi,celery,sqlalchemy,httpx]>=2.18.0",
"rosreestr2coord>=5.0.0", "rosreestr2coord>=5.0.0",
"ijson>=3.2.0", "ijson>=3.2.0",
"pyyaml>=6.0.0", # RBAC roles.yaml loader (app/core/auth.py)
] ]
[dependency-groups] [dependency-groups]

293
backend/tests/test_rbac.py Normal file
View file

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

2
backend/uv.lock generated
View file

@ -578,6 +578,7 @@ dependencies = [
{ name = "pydantic" }, { name = "pydantic" },
{ name = "pydantic-settings" }, { name = "pydantic-settings" },
{ name = "pyproj" }, { name = "pyproj" },
{ name = "pyyaml" },
{ name = "redis" }, { name = "redis" },
{ name = "rosreestr2coord" }, { name = "rosreestr2coord" },
{ name = "scikit-learn" }, { name = "scikit-learn" },
@ -619,6 +620,7 @@ requires-dist = [
{ name = "pydantic", specifier = ">=2.7.0" }, { name = "pydantic", specifier = ">=2.7.0" },
{ name = "pydantic-settings", specifier = ">=2.3.0" }, { name = "pydantic-settings", specifier = ">=2.3.0" },
{ name = "pyproj", specifier = ">=3.6.0" }, { name = "pyproj", specifier = ">=3.6.0" },
{ name = "pyyaml", specifier = ">=6.0.0" },
{ name = "redis", specifier = ">=5.0.0" }, { name = "redis", specifier = ">=5.0.0" },
{ name = "rosreestr2coord", specifier = ">=5.0.0" }, { name = "rosreestr2coord", specifier = ">=5.0.0" },
{ name = "scikit-learn", specifier = ">=1.5.0" }, { name = "scikit-learn", specifier = ">=1.5.0" },

View file

@ -66,6 +66,11 @@ services:
condition: service_healthy condition: service_healthy
ports: ports:
- "127.0.0.1:8000:8000" - "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: healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8000/health"] test: ["CMD", "curl", "-fsS", "http://localhost:8000/health"]
interval: 30s interval: 30s

View file

@ -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: <username>` через
`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

View file

@ -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: <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"]
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
`<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 []),
)
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()

View file

@ -8,14 +8,17 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
from collections.abc import AsyncGenerator import re
from collections.abc import AsyncGenerator, Awaitable, Callable
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
import sentry_sdk import sentry_sdk
from fastapi import FastAPI from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware 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.config import settings
from app.core.db import SessionLocal from app.core.db import SessionLocal
from app.core.fdw import ensure_fdw_user_mapping from app.core.fdw import ensure_fdw_user_mapping
@ -35,8 +38,8 @@ if settings.glitchtip_dsn:
sentry_sdk.init( sentry_sdk.init(
dsn=settings.glitchtip_dsn, dsn=settings.glitchtip_dsn,
environment=settings.environment, environment=settings.environment,
traces_sample_rate=0.0, # только ошибки, без performance-трейсов traces_sample_rate=0.0, # только ошибки, без performance-трейсов
send_default_pii=False, # не шлём client_name / client_phone в отчёты send_default_pii=False, # не шлём client_name / client_phone в отчёты
) )
logging.getLogger("app.main").info("GlitchTip monitoring enabled") logging.getLogger("app.main").info("GlitchTip monitoring enabled")
@ -49,9 +52,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
with SessionLocal() as db: with SessionLocal() as db:
ensure_fdw_user_mapping(db) ensure_fdw_user_mapping(db)
except Exception: except Exception:
logger.exception( logger.exception("FDW user mapping bootstrap failed — cadastral queries may fail")
"FDW user mapping bootstrap failed — cadastral queries may fail"
)
# Startup: launch scheduler background task # Startup: launch scheduler background task
scheduler_task = asyncio.create_task(scheduler_loop()) scheduler_task = asyncio.create_task(scheduler_loop())
@ -73,6 +74,53 @@ app = FastAPI(
lifespan=lifespan, 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( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins=settings.cors_origins, 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(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(trade_in.router, prefix="/api/v1/trade-in", tags=["trade-in"])
app.include_router(search.router, prefix="/api/v1", tags=["search"]) app.include_router(search.router, prefix="/api/v1", tags=["search"])
app.include_router(me.router, prefix="/api/v1", tags=["me"])

View file

@ -22,6 +22,7 @@ dependencies = [
"python-multipart>=0.0.9", # FastAPI UploadFile — загрузка фото квартиры (#394) "python-multipart>=0.0.9", # FastAPI UploadFile — загрузка фото квартиры (#394)
"sentry-sdk>=2.0.0", # мониторинг ошибок → GlitchTip (#396) "sentry-sdk>=2.0.0", # мониторинг ошибок → GlitchTip (#396)
"redis>=5.0.0", # async hot cache для /api/v1/search (Phase 3.2) "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] [dependency-groups]

View file

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

View file

@ -55,6 +55,11 @@ services:
depends_on: depends_on:
postgres: postgres:
condition: service_healthy 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 restart: unless-stopped
networks: networks:
- tradein-net - tradein-net