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:
+        # #3316: get_role ходит в реестр, а вызывающий уже прочитал роли этих
-            role = get_role(username)
+        # же строк — иначе батч снова стал бы N+1 (ловит
-        except KeyError:
+        # test_list_employees_query_count_is_not_n_plus_1). Роль реестра —
-            role = None
+        # ровно то, что вернул бы 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]
 
 

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: <username>` через
 `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 []),
+        allowed_paths=allowed_paths,
-        deny_paths=list(role_def.get("deny", []) or []),
+        deny_paths=deny_paths,
         brand=get_brand_for_user(username),
         display_name=display_name,
         org=org,

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:

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

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), а не копию.
 # ---------------------------------------------------------------------------

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