"""Team-management API — CRUD сотрудников, квоты, история (#2554, эпик #2549). Mounted at `/api/v1/team`; через Caddy `uri strip_prefix /trade-in` это `/trade-in/api/v1/team/*` снаружи. `app.services.auth_session.DB_ROLE_PATHS` уже закладывает `/api/v1/team/**` в scope роли `manager` (и `admin` через `/**`) для `rbac_guard` (см. `app.core.rbac`) — этот роутер добавляет ВТОРОЙ, более узкий барьер именно на identity: - `current_team_actor` резолвит юзера ТОЛЬКО из session-cookie (`app.services.auth_session.get_session_user`). Legacy `X-Authenticated-User` (Caddy trusted-header, dual-mode) НЕ принимается здесь — team-API новый, не участвует в переходном dual-mode auth. Без валидной cookie — 401, даже если `rbac_guard` пропустил запрос по legacy-заголовку (напр. admin через roles.yaml). - Роль должна быть `admin` или `manager` — иначе 403. Org-изоляция (главный инвариант фичи): manager видит/меняет ТОЛЬКО своих employee (`tradein_users.manager_id = actor.user_id`). Чужой/несуществующий employee_id → 404 (НЕ 403) — не подтверждаем/не опровергаем существование чужого сотрудника перед manager'ом. См. `_authorize_employee`. """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Annotated, Any from urllib.parse import urlparse from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy import text from sqlalchemy.engine import RowMapping from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.auth import get_role from app.core.config import settings from app.core.db import get_db from app.core.password import hash_password from app.schemas.team import ( EmployeeCreateRequest, EmployeeHistoryEntry, EmployeeOut, EmployeeUpdateRequest, QuotaStatusOut, ) from app.services import account_quota from app.services.auth_session import get_session_user, revoke_user_sessions from app.services.user_events import schedule_event logger = logging.getLogger(__name__) router = APIRouter() @dataclass class TeamActor: """Резолвленный из session-cookie актёр team-API — admin или manager.""" user_id: int username: str role: str # "admin" | "manager" async def current_team_actor( request: Request, db: Annotated[Session, Depends(get_db)], ) -> TeamActor: """Dependency: session-only identity, роль admin|manager, иначе 401/403. Намеренно НЕ читает `X-Authenticated-User` — см. модульный docstring. """ token = request.cookies.get(settings.session_cookie_name) if not token: raise HTTPException(status_code=401, detail="valid session required") try: session_user = get_session_user(db, token) except Exception: logger.exception("team: session lookup failed") raise HTTPException(status_code=401, detail="valid session required") from None if session_user is None: raise HTTPException(status_code=401, detail="valid session required") role = session_user["role"] if role not in ("admin", "manager"): raise HTTPException(status_code=403, detail="admin or manager role required") return TeamActor( user_id=session_user["user_id"], username=session_user["username"], role=role, ) def _origin_host_allowed(candidate: str) -> bool: """True если scheme://netloc *candidate* совпадает с одним из `settings.cors_origins`. `cors_origins` уже является источником правды для «какие origin'ы это наш фронт» (см. CORSMiddleware в app/main.py, ENV CORS_ORIGINS) — переиспользуем его вместо нового хардкода.""" try: parsed = urlparse(candidate) except ValueError: return False if not parsed.scheme or not parsed.netloc: return False origin = f"{parsed.scheme}://{parsed.netloc}" return origin in settings.cors_origins def _require_same_origin(request: Request) -> None: """CSRF defense-in-depth (issue #2554 DoD) для state-changing team-роутов (POST/PATCH): `Origin` (или `Referer` как fallback) обязан матчить один из `settings.cors_origins`, иначе 403. Оба заголовка отсутствуют → ПРОПУСКАЕМ (не 403). Причина: это единственный надёжный сигнал non-browser клиента в этом стеке — curl-смоуки внутри контейнера (см. `.claude/rules/tradein.md` "Тестировать HTTP только ВНУТРИ контейнера", `docker exec tradein-backend curl ...`) не шлют ни один из этих заголовков, а реальный браузер (fetch/XHR/form) ВСЕГДА прикладывает Origin на unsafe-методах (POST/PATCH) — так что "оба отсутствуют" практически невозможно для настоящего кросс-сайтового CSRF через браузер. Session-cookie уже стоит на `SameSite=Lax` (см. `app.api.v1.auth.login`) — это первый рубеж против CSRF, Origin-check — второй. """ candidate = request.headers.get("origin") or request.headers.get("referer") if candidate is None: return if not _origin_host_allowed(candidate): logger.warning( "team: Origin/Referer mismatch %r on %s — possible CSRF", candidate, request.url.path ) raise HTTPException(status_code=403, detail="origin not allowed") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fetch_employee_row(db: Session, employee_id: int) -> RowMapping | None: return ( db.execute( text( """ SELECT id, username, display_name, org_name, email, is_active, manager_id, created_at FROM tradein_users WHERE id = :id AND role = 'employee' """ ), {"id": employee_id}, ) .mappings() .fetchone() ) def _authorize_employee(actor: TeamActor, row: RowMapping | None) -> RowMapping: """404 (НЕ 403) если сотрудник не найден ИЛИ принадлежит другому manager'у. Org-изоляция: manager может видеть/менять только `manager_id == actor.user_id`. 404 вместо 403 — не палим существование чужого employee_id. """ if row is None: raise HTTPException(status_code=404, detail="employee not found") if actor.role == "manager" and row["manager_id"] != actor.user_id: raise HTTPException(status_code=404, detail="employee not found") return row def _upsert_quota_override( db: Session, username: str, monthly_limit: int, actor_username: str ) -> None: """Upsert персонального лимита. Явная установка monthly_limit — сигнал "хочу numeric-квоту", поэтому ВСЕГДА сбрасывает `unlimited=false` (иначе лимит может молча не применяться — прежний unlimited-грант выигрывал бы у нового limit). `note` — НЕ затирается, если уже задан (`COALESCE`): не перезаписываем человеко-читаемую причину прошлого гранта (напр. "пилот, грант ...") молча сгенерированной строкой; note проставляется только при первом upsert записи. """ db.execute( text( """ INSERT INTO account_quota_overrides (username, monthly_limit, unlimited, note) VALUES (:username, CAST(:monthly_limit AS integer), false, :note) ON CONFLICT (username) DO UPDATE SET monthly_limit = EXCLUDED.monthly_limit, unlimited = false, note = COALESCE(account_quota_overrides.note, EXCLUDED.note), updated_at = now() """ ), { "username": username, "monthly_limit": monthly_limit, "note": f"team-api: set by {actor_username}", }, ) def _batch_quota_status(db: Session, usernames: list[str]) -> dict[str, dict[str, Any]]: """Батч-версия `account_quota.get_status` для N сотрудников — 2 SQL-запроса вместо 2N (было 2N+3 на GET /employees, HIGH/Medium2 review PR #2563). Семантика ИДЕНТИЧНА `account_quota.is_unlimited`/`user_limit`/`get_status` (follow-up review PR #2563 п.2 — предыдущая версия расходилась: батч ВСЕГДА читал `account_quota_overrides.unlimited`, а `is_unlimited` — ТОЛЬКО для username, присутствующего в roles.yaml): - username НЕ в roles.yaml (`get_role` → KeyError) → unlimited=False ВСЕГДА, `account_quota_overrides.unlimited` даже не проверяется (roles.yaml — источник правды "кто вообще может быть unlimited", override — "у кого именно из известных roles.yaml-юзеров"). Сегодня недостижимо для DB-only сотрудников team-API (`_upsert_quota_override` всегда пишет `unlimited=false`), но станет достижимым при ручном UPDATE `account_quota_overrides` или расширении roles.yaml — расхождение с реальным enforcement (`check_and_raise`/`increment`, тот же `is_unlimited`) было бы честной ложью в списке: "без лимита", который движок всё равно считает. - username в roles.yaml и role == admin → unlimited=True (без похода в БД). - username в roles.yaml, role != admin → unlimited = override.unlimited. limit = override.monthly_limit (читается для ЛЮБОГО username, без gate по roles.yaml — так же ведёт себя `account_quota.user_limit`), иначе глобальный `account_quota.MONTHLY_LIMIT`. """ if not usernames: return {} overrides = ( db.execute( text( """ SELECT username, monthly_limit, unlimited FROM account_quota_overrides WHERE username = ANY(CAST(:usernames AS text[])) """ ), {"usernames": usernames}, ) .mappings() .all() ) override_by_username = {r["username"]: r for r in overrides} period = account_quota.current_period() usage_rows = ( db.execute( text( """ SELECT username, used FROM account_estimate_usage WHERE username = ANY(CAST(:usernames AS text[])) AND period_month = :period """ ), {"usernames": usernames, "period": period}, ) .mappings() .all() ) used_by_username = {r["username"]: r["used"] for r in usage_rows} 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 if role == "admin": unlimited = True elif role is not None: unlimited = bool(override is not None and override["unlimited"]) else: # username не в roles.yaml — is_unlimited() короткое замыкание на # False, override НЕ проверяется (см. докстринг выше). unlimited = False limit = ( int(override["monthly_limit"]) if override is not None and override["monthly_limit"] is not None else account_quota.MONTHLY_LIMIT ) used = used_by_username.get(username, 0) if unlimited: result[username] = { "limit": limit, "used": used, "remaining": limit, "unlimited": True, } else: remaining = max(0, limit - max(0, used)) result[username] = { "limit": limit, "used": used, "remaining": remaining, "unlimited": False, } return result def _employee_out(row: RowMapping, quota: dict[str, Any]) -> EmployeeOut: return EmployeeOut( id=row["id"], username=row["username"], display_name=row["display_name"], org_name=row["org_name"], email=row["email"], is_active=row["is_active"], manager_id=row["manager_id"], created_at=row["created_at"], quota=QuotaStatusOut(**quota), ) # --------------------------------------------------------------------------- # POST /employees # --------------------------------------------------------------------------- @router.post("/employees", response_model=EmployeeOut, status_code=201) async def create_employee( body: EmployeeCreateRequest, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], _origin_check: Annotated[None, Depends(_require_same_origin)], ) -> EmployeeOut: """Создать сотрудника. Роль всегда `employee`. manager_id: для actor.role == manager — принудительно свой id (любое значение из тела ИГНОРИРУЕТСЯ, org-изоляция инвариант #2554). Для actor.role == admin — опционально из тела, валидируется что указанный id существует и role='manager' (иначе 422). """ existing = db.execute( text("SELECT id FROM tradein_users WHERE username = :u"), {"u": body.username}, ).fetchone() if existing is not None: raise HTTPException(status_code=409, detail="username already exists") try: password_hash = hash_password(body.password) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from None manager_id: int | None if actor.role == "manager": # Инвариант org-изоляции: manager не может создать сотрудника под # чужим manager_id — любое значение из тела игнорируется молча. manager_id = actor.user_id else: manager_id = body.manager_id if manager_id is not None: mgr = db.execute( text("SELECT id FROM tradein_users WHERE id = :id AND role = 'manager'"), {"id": manager_id}, ).fetchone() if mgr is None: raise HTTPException( status_code=422, detail="manager_id does not reference an existing manager", ) try: row = ( db.execute( text( """ INSERT INTO tradein_users (username, password_hash, role, manager_id, display_name, org_name, email, is_active) VALUES (:username, :password_hash, 'employee', :manager_id, :display_name, :org_name, :email, true) RETURNING id, username, display_name, org_name, email, is_active, manager_id, created_at """ ), { "username": body.username, "password_hash": password_hash, "manager_id": manager_id, "display_name": body.display_name, "org_name": body.org_name, "email": body.email, }, ) .mappings() .fetchone() ) except IntegrityError: # TOCTOU: два конкурентных POST с одинаковым username между pre-check # выше и этим INSERT — UNIQUE-констрейнт на tradein_users.username ловит. db.rollback() raise HTTPException(status_code=409, detail="username already exists") from None assert row is not None # RETURNING на успешный INSERT всегда отдаёт строку if body.monthly_limit is not None: _upsert_quota_override(db, body.username, body.monthly_limit, actor.username) db.commit() schedule_event( event_type="employee_created", username=actor.username, payload={ "employee_id": row["id"], "employee_username": row["username"], "manager_id": manager_id, }, ) quota = account_quota.get_status(db, body.username) return _employee_out(row, quota) # --------------------------------------------------------------------------- # PATCH /employees/{id} # --------------------------------------------------------------------------- @router.patch("/employees/{employee_id}", response_model=EmployeeOut) async def update_employee( employee_id: int, body: EmployeeUpdateRequest, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], _origin_check: Annotated[None, Depends(_require_same_origin)], ) -> EmployeeOut: """Частичное обновление сотрудника — block/unblock, лимит, профиль, пароль. manager может патчить ТОЛЬКО своих (manager_id == actor.user_id), иначе 404. При is_active=False ИЛИ смене пароля (new_password) — обязательно revoke всех сессий (HIGH, deep-review PR #2563): без этого блокировка/reset не подействуют до истечения TTL текущей сессии сотрудника — хуже того, sliding-refresh (`app.services.auth_session.get_session_user`) продлевает `expires_at` на КАЖДОМ запросе, так что скомпрометированная/чужая сессия живёт неограниченно долго, а не «до TTL». `revoke_user_sessions` сам называет смену пароля своим use-case — см. его докстринг. """ row = _fetch_employee_row(db, employee_id) row = _authorize_employee(actor, row) new_password_hash: str | None = None if body.new_password is not None: try: new_password_hash = hash_password(body.new_password) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from None db.execute( text( """ UPDATE tradein_users SET display_name = COALESCE(:display_name, display_name), org_name = COALESCE(:org_name, org_name), email = COALESCE(:email, email), is_active = COALESCE(CAST(:is_active AS boolean), is_active), password_hash = COALESCE(:password_hash, password_hash), updated_at = now() WHERE id = :id """ ), { "display_name": body.display_name, "org_name": body.org_name, "email": body.email, "is_active": body.is_active, "password_hash": new_password_hash, "id": employee_id, }, ) if body.monthly_limit is not None: _upsert_quota_override(db, row["username"], body.monthly_limit, actor.username) if body.is_active is False or body.new_password is not None: # Обязательно ПОСЛЕ UPDATE, ДО финального commit — revoke_user_sessions # коммитит сам (см. app.services.auth_session), это флашит и наш # предшествующий UPDATE/quota-upsert в той же сессии. Self-lockout # невозможен: _fetch_employee_row фильтрует role='employee', actor # (admin|manager) никогда не может патчить сам себя через этот роут. revoke_user_sessions(db, employee_id) db.commit() changed_profile_fields = [ f for f, v in ( ("display_name", body.display_name), ("org_name", body.org_name), ("email", body.email), ) if v is not None ] if changed_profile_fields: schedule_event( event_type="employee_updated", username=actor.username, payload={ "employee_id": employee_id, "employee_username": row["username"], "fields": changed_profile_fields, }, ) if body.new_password is not None: schedule_event( event_type="employee_password_reset", username=actor.username, payload={"employee_id": employee_id, "employee_username": row["username"]}, ) if body.is_active is not None: schedule_event( event_type="employee_blocked" if body.is_active is False else "employee_unblocked", username=actor.username, payload={"employee_id": employee_id, "employee_username": row["username"]}, ) if body.monthly_limit is not None: schedule_event( event_type="quota_changed", username=actor.username, payload={ "employee_id": employee_id, "employee_username": row["username"], "monthly_limit": body.monthly_limit, }, ) updated_row = _fetch_employee_row(db, employee_id) assert updated_row is not None # только что успешно обновили эту же строку quota = account_quota.get_status(db, updated_row["username"]) return _employee_out(updated_row, quota) # --------------------------------------------------------------------------- # GET /employees # --------------------------------------------------------------------------- # Два статических варианта WHERE (НЕ f-string/динамическая сборка — Medium/ # "заодно" review PR #2563: значения биндятся параметрами и без того безопасны, # но статические ветки не провоцируют будущие правки в сторону конкатенации SQL). # # ORDER BY created_at DESC, id DESC — тай-брейкер по `id` ОБЯЗАТЕЛЕН (follow-up # review PR #2563 п.1): `created_at DEFAULT now()` — время ТРАНЗАКЦИИ, а bulk-seed # (#2557) вставляет много юзеров одной транзакцией → идентичный timestamp у N строк. # Без тай-брейкера порядок между страницами (LIMIT/OFFSET) на PostgreSQL для # строк-«близнецов» не гарантирован — сотрудники пропадали/дублировались бы при # постраничном листании. `id` монотонно растёт (BIGINT IDENTITY) — детерминированный # tie-break без доп. индекса (созданные позже = бОльший id, тот же порядок что и # намерение DESC-сортировки по времени). _LIST_EMPLOYEES_BY_MANAGER_SQL = text( """ SELECT id, username, display_name, org_name, email, is_active, manager_id, created_at FROM tradein_users WHERE role = 'employee' AND manager_id = :manager_id ORDER BY created_at DESC, id DESC LIMIT :limit OFFSET :offset """ ) _LIST_EMPLOYEES_ALL_SQL = text( """ SELECT id, username, display_name, org_name, email, is_active, manager_id, created_at FROM tradein_users WHERE role = 'employee' ORDER BY created_at DESC, id DESC LIMIT :limit OFFSET :offset """ ) @router.get("/employees", response_model=list[EmployeeOut]) async def list_employees( actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], manager_id: Annotated[int | None, Query()] = None, limit: Annotated[int, Query(ge=1, le=200)] = 50, offset: Annotated[int, Query(ge=0)] = 0, ) -> list[EmployeeOut]: """Список сотрудников. manager видит только своих; admin — всех, опц. ?manager_id=. Квота — ОДИН батч-запрос на всю страницу (`_batch_quota_status`), не N+1 (Medium2, review PR #2563: было 2N+3 SQL-запросов на N сотрудников). """ if actor.role == "manager": rows = ( db.execute( _LIST_EMPLOYEES_BY_MANAGER_SQL, {"manager_id": actor.user_id, "limit": limit, "offset": offset}, ) .mappings() .all() ) elif manager_id is not None: rows = ( db.execute( _LIST_EMPLOYEES_BY_MANAGER_SQL, {"manager_id": manager_id, "limit": limit, "offset": offset}, ) .mappings() .all() ) else: rows = ( db.execute(_LIST_EMPLOYEES_ALL_SQL, {"limit": limit, "offset": offset}).mappings().all() ) quota_by_username = _batch_quota_status(db, [row["username"] for row in rows]) return [_employee_out(row, quota_by_username[row["username"]]) for row in rows] # --------------------------------------------------------------------------- # GET /employees/{id}/history # --------------------------------------------------------------------------- @router.get("/employees/{employee_id}/history", response_model=list[EmployeeHistoryEntry]) async def employee_history( employee_id: int, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], limit: Annotated[int, Query(ge=1, le=200)] = 50, offset: Annotated[int, Query(ge=0)] = 0, ) -> list[EmployeeHistoryEntry]: """История оценок сотрудника (адрес/дата/результат) — из `user_events`, LEFT JOIN `trade_in_estimates` за фактическим результатом. Та же org-проверка что и в PATCH: чужой employee_id → 404. """ row = _fetch_employee_row(db, employee_id) row = _authorize_employee(actor, row) rows = ( db.execute( text( """ SELECT CAST(ue.estimate_id AS text) AS estimate_id, ue.payload ->> 'address' AS address, ue.payload ->> 'area_m2' AS area_m2, ue.payload ->> 'rooms' AS rooms, te.median_price, te.confidence, te.n_analogs, ue.created_at FROM user_events ue LEFT JOIN trade_in_estimates te ON te.id = ue.estimate_id WHERE ue.username = :username AND ue.event_type = 'estimate_request' ORDER BY ue.created_at DESC LIMIT :limit OFFSET :offset """ ), {"username": row["username"], "limit": limit, "offset": offset}, ) .mappings() .all() ) return [EmployeeHistoryEntry.model_validate(dict(r)) for r in rows]