diff --git a/tradein-mvp/backend/app/api/v1/team.py b/tradein-mvp/backend/app/api/v1/team.py index d42bd250..1a422822 100644 --- a/tradein-mvp/backend/app/api/v1/team.py +++ b/tradein-mvp/backend/app/api/v1/team.py @@ -78,7 +78,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from sqlalchemy.sql.elements import TextClause -from app.core.auth import get_role +from app.core.auth import get_role, yaml_role from app.core.config import settings from app.core.db import get_db from app.core.password import hash_password @@ -288,7 +288,9 @@ def _upsert_quota_override( ) -def _batch_quota_status(db: Session, usernames: list[str]) -> dict[str, dict[str, Any]]: +def _batch_quota_status( + db: Session, usernames: list[str], known_roles: dict[str, str] | None = None +) -> dict[str, dict[str, Any]]: """Батч-версия `account_quota.get_status` для N сотрудников — 2 SQL-запроса вместо 2N (было 2N+3 на GET /employees, HIGH/Medium2 review PR #2563). @@ -351,13 +353,21 @@ def _batch_quota_status(db: Session, usernames: list[str]) -> dict[str, dict[str result: dict[str, dict[str, Any]] = {} for username in usernames: override = override_by_username.get(username) - try: - role = get_role(username) - except KeyError: - role = None + # #3316: get_role ходит в реестр, а вызывающий уже прочитал роли этих + # же строк — иначе батч снова стал бы N+1 (ловит + # test_list_employees_query_count_is_not_n_plus_1). Роль реестра — + # ровно то, что вернул бы get_role: он спрашивает реестр первым. + role: str | None + if known_roles is not None and username in known_roles: + role = known_roles[username] + else: + try: + role = get_role(username) + except KeyError: + role = None if role == "admin": unlimited = True - elif role is not None: + elif yaml_role(username) is not None: unlimited = bool(override is not None and override["unlimited"]) else: # username не в roles.yaml — is_unlimited() короткое замыкание на @@ -432,6 +442,20 @@ async def create_employee( `identity_db` — реестр (строка сотрудника), `db` — продуктовая квота; в дефолтном режиме это одна и та же сессия и одна транзакция. """ + # #3316 defense-in-depth: имя, за которым в roles.yaml уже числятся права + # (admin/pilot/analyst), занять нельзя. Роль резолвится из реестра первой + # (app.core.auth.get_role), так что эскалации не было бы и без этой + # проверки — но совпадение имён само по себе означает двух разных людей с + # одним логином, и дешевле отказать на входе, чем разбирать это в логах. + legacy = yaml_role(body.username) + if legacy is not None and legacy != "employee": + logger.warning( + "create_employee: %r refused — username занят в roles.yaml (role=%s)", + body.username, + legacy, + ) + raise HTTPException(status_code=409, detail="username reserved in roles config") + schema = identity_schema() existing = identity_db.execute( text(f"SELECT id FROM {schema.users_table} WHERE username = :u"), @@ -762,7 +786,11 @@ async def list_employees( .all() ) - quota_by_username = _batch_quota_status(db, [row["username"] for row in rows]) + quota_by_username = _batch_quota_status( + db, + [row["username"] for row in rows], + known_roles={row["username"]: row["role"] for row in rows}, + ) return [_employee_out(row, quota_by_username[row["username"]]) for row in rows] diff --git a/tradein-mvp/backend/app/core/auth.py b/tradein-mvp/backend/app/core/auth.py index affe6c4e..512b924a 100644 --- a/tradein-mvp/backend/app/core/auth.py +++ b/tradein-mvp/backend/app/core/auth.py @@ -8,6 +8,11 @@ from repo root. We deliberately do NOT share code between repos via When updating one copy, update the other. +⚠️ РАСХОЖДЕНИЕ С ЗЕРКАЛОМ (#3316, намеренное — не «синхронизировать» обратно): +здесь `get_role` резолвит роль СНАЧАЛА из реестра людей (`tradein_users.role` / +`auth.users.role`), и только потом из YAML. У основного бэкенда реестра нет, +там копия остаётся YAML-only. + Caddy gates the whole site with basic_auth (см. `caddy/users.caddy.snippet`) и пропускает в backend заголовок `X-Authenticated-User: ` через `header_up X-Authenticated-User {http.auth.user.id}` в каждом reverse_proxy. @@ -25,13 +30,15 @@ import logging import re from functools import lru_cache from pathlib import Path -from typing import Literal, TypedDict +from typing import Literal, TypedDict, cast import yaml logger = logging.getLogger(__name__) -Role = Literal["admin", "pilot", "analyst", "expired"] +# legacy roles.yaml-роли + роли реестра ('admin'|'manager'|'employee', CHECK +# tradein м.192 / auth м.004). Оба набора приходят из одного `get_role` (#3316). +Role = Literal["admin", "pilot", "analyst", "expired", "manager", "employee"] class UserScope(TypedDict): @@ -155,8 +162,74 @@ def _load_roles_config() -> dict: # --------------------------------------------------------------------------- +def yaml_role(username: str) -> Role | None: + """Роль из roles.yaml (без похода в реестр) или None, если юзера там нет. + + Нужна там, где спрашивают именно про legacy-файл, а не про эффективную роль: + `team.create_employee` (#3316) не даёт занять имя, за которым в YAML уже + числятся права. + """ + users: dict[str, Role] = _load_roles_config()["users"] + return users.get(username) + + +def _registry_role(username: str) -> str | None: + """Роль из реестра людей (`tradein_users.role` / `auth.users.role`) или None. + + None означает «реестр про этого юзера ничего не сказал»: строки нет, роль + пустая, либо реестр вообще недоступен. Во всех трёх случаях решение + остаётся за roles.yaml — падение БД не имеет права выключить legacy-вход. + + ⚠️ Осознанный компромисс (#3316 review): последняя ветка — недоступный + реестр — на время сбоя ВОЗВРАЩАЕТ авторитетность roles.yaml, то есть ровно + то состояние, которое этот фикс и лечит. Сегодня это безопасно: коллизий + имён между реестром и YAML на проде нет, а новые закрыты 409-гвардом в + `team.create_employee`. Если коллизия всё же появится (ручной INSERT в + реестр, расширение roles.yaml) — сбой БД станет окном эскалации, и тогда + эту ветку надо менять на fail-closed (отказ вместо YAML-роли), а не + дописывать проверки у вызывающих. + + Имя таблицы берётся из фиксированного словаря `identity_schema()`, значение + едет bind-параметром: снаружи в SQL не попадает ничего. + """ + try: + from sqlalchemy import text + + from app.services.identity_store import identity_schema, identity_session + + schema = identity_schema() + with identity_session() as db: + row = db.execute( + text(f"SELECT role FROM {schema.users_table} WHERE username = :username"), + {"username": username}, + ).fetchone() + except Exception: + logger.exception( + "registry role lookup failed for %r — fallback to roles.yaml", + username, + ) + return None + if row is None or not row.role: + return None + return str(row.role) + + def get_role(username: str) -> Role: - """Return the role for *username* or raise KeyError if unknown.""" + """Эффективная роль *username*: реестр (БД) первый, roles.yaml — fallback. + + Raises KeyError, если юзера нет ни там, ни там. + + #3316: раньше роль резолвилась ТОЛЬКО из roles.yaml, при том что люди + заводятся в БД (`tradein_users`) — два дефекта разом. Вверх: сотрудник, + чьё имя совпало с YAML-админом, получал admin (IDOR по чужим оценкам + + безлимит квоты). Вниз: сотрудник, которого в YAML нет, получал KeyError → + 403 на СОБСТВЕННУЮ оценку. Единственный источник истины теперь один, и он + здесь — вызывающие (rbac, trade_in, team, account_quota) не меняются. + """ + db_role = _registry_role(username) + if db_role is not None: + return cast(Role, db_role) + config = _load_roles_config() users: dict[str, Role] = config["users"] if username not in users: @@ -217,13 +290,22 @@ def get_user_scope(username: str) -> UserScope: """ config = _load_roles_config() role = get_role(username) - role_def = config["roles"][role] + role_def = config["roles"].get(role) + if role_def is None: + # Роль реестра (employee/manager) — её scope живёт в DB_ROLE_PATHS, а не + # в roles.yaml (#3316: get_role теперь может вернуть и такую роль). + from app.services.auth_session import get_db_role_scope + + allowed_paths, deny_paths = get_db_role_scope(role) + else: + allowed_paths = list(role_def.get("paths", []) or []) + deny_paths = list(role_def.get("deny", []) or []) display_name, org, email = get_profile_for_user(username) return UserScope( username=username, role=role, - allowed_paths=list(role_def.get("paths", []) or []), - deny_paths=list(role_def.get("deny", []) or []), + allowed_paths=allowed_paths, + deny_paths=deny_paths, brand=get_brand_for_user(username), display_name=display_name, org=org, diff --git a/tradein-mvp/backend/app/core/rbac.py b/tradein-mvp/backend/app/core/rbac.py index 06f14afc..62701225 100644 --- a/tradein-mvp/backend/app/core/rbac.py +++ b/tradein-mvp/backend/app/core/rbac.py @@ -38,7 +38,7 @@ from fastapi.responses import JSONResponse, Response from app.core.auth import get_role, is_path_allowed from app.core.config import settings -from app.services.auth_session import get_db_role_scope, get_session_user +from app.services.auth_session import DB_ROLE_PATHS, get_db_role_scope, get_session_user from app.services.identity_store import identity_session logger = logging.getLogger(__name__) @@ -334,7 +334,11 @@ async def rbac_guard( # scope-narrowed юзер не смог бы получить свою роль вовсе. if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT): external_path = _EXTERNAL_PREFIX + path - if from_session: + # Матчер выбирается по РОДУ роли, а не только по источнику (#3316): + # с DB-first резолвом legacy trusted-header путь тоже может отдать роль + # реестра (employee/manager), а её паттернов в roles.yaml нет — сверка + # с `is_path_allowed` дала бы 403 на всё. + if from_session or role in DB_ROLE_PATHS: allowed = _db_role_path_allowed(role, external_path) else: try: diff --git a/tradein-mvp/backend/app/services/account_quota.py b/tradein-mvp/backend/app/services/account_quota.py index 10d778ca..7f420657 100644 --- a/tradein-mvp/backend/app/services/account_quota.py +++ b/tradein-mvp/backend/app/services/account_quota.py @@ -46,7 +46,7 @@ from fastapi import HTTPException from sqlalchemy import text from sqlalchemy.orm import Session -from app.core.auth import get_role +from app.core.auth import get_role, yaml_role from app.core.config import settings logger = logging.getLogger(__name__) @@ -61,8 +61,7 @@ def limit_exhausted_message(limit: int) -> str: отличаться от глобального MONTHLY_LIMIT для персонального override ИЛИ anon default_limit, см. #b2c-antiabuse-2).""" return ( - f"Лимит из {limit} оценок в этом месяце исчерпан. " - "За полной версией обращайтесь к Копылову." + f"Лимит из {limit} оценок в этом месяце исчерпан. За полной версией обращайтесь к Копылову." ) @@ -97,6 +96,14 @@ def is_unlimited(db: Session, username: str) -> bool: return False if role == "admin": return True + # #3316: get_role резолвит роль из реестра (БД) первой, поэтому сотрудник + # team-API больше не даёт KeyError. Право на ПЕРСОНАЛЬНЫЙ безлимит при этом + # осталось там же, где было — за roles.yaml: фикс убирает эскалацию, а не + # раздаёт новую. Иначе руками проставленный `unlimited` начал бы работать + # для аккаунтов, которым он раньше молча игнорировался (и разъехался бы с + # `_batch_quota_status` в списке «Команды»). + if yaml_role(username) is None: + return False row = db.execute( text( """ diff --git a/tradein-mvp/backend/tests/test_rbac.py b/tradein-mvp/backend/tests/test_rbac.py index ea0837dd..dc3ecc0b 100644 --- a/tradein-mvp/backend/tests/test_rbac.py +++ b/tradein-mvp/backend/tests/test_rbac.py @@ -42,6 +42,23 @@ def _reset_auth_cache() -> None: auth_mod.reset_cache_for_tests() +@pytest.fixture(autouse=True) +def _legacy_yaml_only(monkeypatch: pytest.MonkeyPatch) -> None: + """Реестр в ЭТОМ файле молчит — здесь проверяется legacy-путь roles.yaml. + + #3316 сделал `get_role` DB-first (реестр → YAML-fallback), и без этой + изоляции результат файла зависел бы от ОКРУЖЕНИЯ: локально без БД шла + YAML-ветка и всё было зелено, а в CI, где реестр засеян миграцией 193 + (kopylov=manager, user*=employee), те же ассерты краснели. Тест, который + отвечает по-разному в двух окружениях, не проверяет ничего. + + Здесь закреплена ровно YAML-семантика (разбор файла, globs, поведение + guard'а и /me на trusted-header пути); DB-first, приоритет реестра и + эквивалентность scope employee↔pilot покрыты tests/test_role_single_source.py. + """ + monkeypatch.setattr(auth_mod, "_registry_role", lambda username: None) + + # --------------------------------------------------------------------------- # Test app — использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py), а не копию. # --------------------------------------------------------------------------- diff --git a/tradein-mvp/backend/tests/test_role_single_source.py b/tradein-mvp/backend/tests/test_role_single_source.py new file mode 100644 index 00000000..c60e00a8 --- /dev/null +++ b/tradein-mvp/backend/tests/test_role_single_source.py @@ -0,0 +1,188 @@ +"""#3316 — роль резолвится из ОДНОГО источника: реестр (БД) первый, roles.yaml — fallback. + +Проверяется значение роли, а не факт вызова механизма: + * имя из roles.yaml, заведённое в реестре сотрудником → роль `employee` + (эскалации в admin нет: ни IDOR по чужим оценкам, ни безлимитной квоты); + * сотрудник, которого в roles.yaml НЕТ → роль резолвится, ownership-check + пропускает его к СВОЕЙ оценке и держит на чужой (раньше был KeyError → 403); + * legacy-юзер (есть в YAML, в реестре строки нет) → роль ровно как раньше — + сверяется ВЕСЬ маппинг roles.yaml, а не один аккаунт; + * реестр недоступен → fallback на YAML (падение БД не выключает legacy-вход). + +FastAPI здесь не поднимается: резолвер — чистая функция от (реестр, YAML), +реестр подменяется фейковой сессией. +""" + +from __future__ import annotations + +import os +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from types import SimpleNamespace +from typing import Any + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import pytest + +from app.core import auth +from app.services import identity_store + +_MISSING = object() + + +class _FakeSession: + """Отдаёт одну строку `SELECT role ...` — или ничего, если роли нет.""" + + def __init__(self, role: str | None) -> None: + self.role = role + self.sql: str | None = None + self.params: dict[str, Any] | None = None + + def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any: + self.sql = str(sql) + self.params = params + row = None if self.role is None else SimpleNamespace(role=self.role) + return SimpleNamespace(fetchone=lambda: row) + + +@pytest.fixture +def registry(monkeypatch: pytest.MonkeyPatch) -> Callable[..., _FakeSession | None]: + """`registry(role)` — что реестр отвечает на запрос роли. + + role=None → строки нет (legacy-юзер); role=_MISSING → реестр падает. + """ + + def install(role: str | None | object) -> _FakeSession | None: + if role is _MISSING: + + @contextmanager + def broken_session() -> Iterator[Any]: + raise RuntimeError("registry down") + yield # pragma: no cover — нужен, чтобы функция была генератором + + monkeypatch.setattr(identity_store, "identity_session", broken_session) + return None + + session = _FakeSession(role) # type: ignore[arg-type] + + @contextmanager + def fake_session() -> Iterator[_FakeSession]: + yield session + + monkeypatch.setattr(identity_store, "identity_session", fake_session) + return session + + return install + + +def _yaml_users() -> dict[str, str]: + return dict(auth._load_roles_config()["users"]) + + +def _yaml_admin() -> str: + for username, role in _yaml_users().items(): + if role == "admin": + return username + pytest.skip("в auth/roles.yaml нет ни одного admin — тест неприменим") + + +def test_registry_employee_beats_yaml_admin(registry: Callable[..., Any]) -> None: + """Эскалация закрыта: имя YAML-админа + строка `employee` в реестре = employee.""" + victim_name = _yaml_admin() + session = registry("employee") + + assert auth.get_role(victim_name) == "employee" + # username едет bind-параметром, а не склейкой в SQL. + assert session.params == {"username": victim_name} + assert victim_name not in (session.sql or "") + + +def test_employee_absent_from_yaml_resolves_and_owns_estimate( + registry: Callable[..., Any], +) -> None: + """Сотрудник вне roles.yaml: роль есть, своя оценка читается, чужая — нет.""" + name = "employee_not_in_yaml_3316" + assert name not in _yaml_users() + registry("employee") + + from fastapi import HTTPException + + from app.api.v1.trade_in import _assert_estimate_access + + # Продуктовое поведение проверяется ПЕРВЫМ: до #3316 здесь прилетал 403 + # («user not in roles config») на СОБСТВЕННУЮ оценку сотрудника. + _assert_estimate_access(name, name) + + assert auth.get_role(name) == "employee" + + with pytest.raises(HTTPException) as exc: + _assert_estimate_access("someone_else", name) + assert exc.value.status_code == 404 + + +def test_legacy_yaml_users_keep_their_roles(registry: Callable[..., Any]) -> None: + """В реестре строки нет → роли ВСЕХ YAML-юзеров ровно те же, что и были.""" + registry(None) + users = _yaml_users() + assert users, "roles.yaml без юзеров — сверять нечего" + assert {username: auth.get_role(username) for username in users} == users + + +def _yaml_pilot() -> str: + for username, role in _yaml_users().items(): + if role == "pilot": + return username + pytest.skip("в auth/roles.yaml нет ни одного pilot — тест неприменим") + + +def test_prod_config_pilot_in_yaml_employee_in_registry(registry: Callable[..., Any]) -> None: + """Прод-конфигурация 12 из 13 аккаунтов: строка в реестре ЕСТЬ и роль там иная. + + Реестр главнее (`employee`), а объём прав от этого не меняется: scope + DB-роли `employee` обязан совпадать с вчерашним YAML-scope роли `pilot`. + Списки сверяются целиком — дрейф ЛЮБОГО из двух ловится здесь, а не + тихой потерей/выдачей раздела в проде. + """ + from app.services.auth_session import DB_ROLE_PATHS + + registry("employee") + assert auth.get_role(_yaml_pilot()) == "employee" + + pilot = auth._load_roles_config()["roles"]["pilot"] + allow, deny = DB_ROLE_PATHS["employee"] + assert sorted(allow) == sorted(pilot["paths"]) + assert sorted(deny) == sorted(pilot["deny"] or []) + + +def test_prod_config_pilot_in_yaml_manager_in_registry(registry: Callable[..., Any]) -> None: + """Конфигурация kopylov: YAML pilot + реестр manager → manager. + + Лишних путей на tradein-периметре это не даёт: manager отличается от + employee ровно одним префиксом `/api/v1/team/**` (дашборд «Команды», + ВНЕ `/trade-in/**`), а deny-списки совпадают. + """ + from app.services.auth_session import DB_ROLE_PATHS + + registry("manager") + assert auth.get_role(_yaml_pilot()) == "manager" + + emp_allow, emp_deny = DB_ROLE_PATHS["employee"] + mgr_allow, mgr_deny = DB_ROLE_PATHS["manager"] + extra = set(mgr_allow) - set(emp_allow) + assert extra == {"/api/v1/team/**"} + assert not any(p.startswith("/trade-in") for p in extra) + assert sorted(mgr_deny) == sorted(emp_deny) + + +def test_registry_failure_falls_back_to_yaml(registry: Callable[..., Any]) -> None: + """Реестр недоступен → legacy-вход продолжает работать по YAML.""" + registry(_MISSING) + assert auth.get_role(_yaml_admin()) == "admin" + + +def test_yaml_role_is_yaml_only(registry: Callable[..., Any]) -> None: + """Предикат гварда create_employee смотрит ИМЕННО в YAML, мимо реестра.""" + registry("employee") + assert auth.yaml_role(_yaml_admin()) == "admin" + assert auth.yaml_role("employee_not_in_yaml_3316") is None