fix: /docs и OpenAPI только при DEBUG, CORS из CORS_ORIGINS в проде (OQ-65, OQ-66, #9 п.2)

Приложение собирает `create_app(settings, web_dir=...)`: при DEBUG=false нет
`/docs`, `/redoc`, `/openapi.json` (SPA-заглушка на эти пути отвечает 404, а не
index.html); CORS — `cors_options`: DEBUG — `*` с credentials, иначе
`CORS_ORIGINS` (пусто — middleware не подключается, credentials только для
списка без `*`). Настройка `CORS_ORIGINS` читает JSON-список или строку через
запятую. `.env.example` и README «Конфигурация» дополнены.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev 2026-10-05 13:16:51 +04:00
parent 5e11f0d0bf
commit 0eedb822e2
7 changed files with 209 additions and 35 deletions

View File

@ -144,7 +144,7 @@ curl -s localhost:8000/api/v1/me -H "Authorization: Bearer <JWT>"
``` ```
(идентификаторы в вашей БД будут другими). Токен действует 8 часов. Интерактивная документация API — `/docs` (идентификаторы в вашей БД будут другими). Токен действует 8 часов. Интерактивная документация API — `/docs`
(Swagger UI), схема — `/openapi.json`. (Swagger UI), схема — `/openapi.json`; обе — только при `DEBUG=true` (в проде их нет, 404).
Dev-пользователи (`api/seed/dev/users.yaml`, создаются `make seed-dev`): Dev-пользователи (`api/seed/dev/users.yaml`, создаются `make seed-dev`):
@ -438,7 +438,8 @@ email без учёта регистра проверяет только сер
| Переменная | По умолчанию | Назначение | | Переменная | По умолчанию | Назначение |
|---|---|---| |---|---|---|
| `ENV_FILE` | `.env` | файл окружения (относительно `api/`) | | `ENV_FILE` | `.env` | файл окружения (относительно `api/`) |
| `DEBUG` | `false` | режим разработки. `true` необходим dev-эмитенту токенов; при `false` пакеты РАИСА подгружаются условно | | `DEBUG` | `false` | режим разработки. `true` необходим dev-эмитенту токенов и `make seed-dev`, включает документацию API (`/docs`, `/redoc`, `/openapi.json`) и CORS `*`; при `false` документации нет (404), CORS — из `CORS_ORIGINS`, пакеты РАИСА подгружаются условно |
| `CORS_ORIGINS` | пусто | при `DEBUG=false` — источники, которым разрешены кросс-доменные запросы: строка через запятую (`https://a.example.ru,https://b.example.ru`) или JSON-список. Пусто — кросс-доменные запросы запрещены (middleware CORS не подключается); `allow_credentials` — только для явного списка без `*`. При `DEBUG=true` не используется (всегда `*`) |
| `SERVER__HOST`, `SERVER__PORT` | `0.0.0.0`, `8000` | адрес и порт сервера (`run_server.py`); в `.env.example` — `127.0.0.1` (dev-запуск только на loopback) | | `SERVER__HOST`, `SERVER__PORT` | `0.0.0.0`, `8000` | адрес и порт сервера (`run_server.py`); в `.env.example` — `127.0.0.1` (dev-запуск только на loopback) |
| `ROOT_PATH` | `/` | корневой путь за прокси (должен совпадать с `REACT_APP_ROOT_PATH` фронта) | | `ROOT_PATH` | `/` | корневой путь за прокси (должен совпадать с `REACT_APP_ROOT_PATH` фронта) |
| `VERSION` | `1.0.0` | версия в `/healthcheck` и OpenAPI | | `VERSION` | `1.0.0` | версия в `/healthcheck` и OpenAPI |

View File

@ -30,6 +30,10 @@ OPENBAO__SMB__SMB_PATH=
OPENBAO__SMB__SMB_LOGIN= OPENBAO__SMB__SMB_LOGIN=
OPENBAO__SMB__SMB_PASSWORD= OPENBAO__SMB__SMB_PASSWORD=
# CORS при DEBUG=false: источники через запятую или JSON-список; пусто — кросс-доменные запросы запрещены.
# При DEBUG=true не используется (CORS `*`); документация API (/docs, /redoc, /openapi.json) — тоже только при DEBUG=true
CORS_ORIGINS=
# Аутентификация. Dev-эмитент работает только при AUTH_DEV_ISSUER=true И DEBUG=true. # Аутентификация. Dev-эмитент работает только при AUTH_DEV_ISSUER=true И DEBUG=true.
AUTH_DEV_ISSUER=true AUTH_DEV_ISSUER=true
# ТОЛЬКО ДЛЯ РАЗРАБОТКИ: секрет HS256 dev-эмитента, в проде не использовать # ТОЛЬКО ДЛЯ РАЗРАБОТКИ: секрет HS256 dev-эмитента, в проде не использовать

View File

@ -27,6 +27,9 @@ OPENBAO__SMB__SMB_PATH=
OPENBAO__SMB__SMB_LOGIN= OPENBAO__SMB__SMB_LOGIN=
OPENBAO__SMB__SMB_PASSWORD= OPENBAO__SMB__SMB_PASSWORD=
# CORS при DEBUG=false (в тестах DEBUG=true — не используется), как в .env.example
CORS_ORIGINS=
AUTH_DEV_ISSUER=true AUTH_DEV_ISSUER=true
AUTH_DEV_SECRET=test-only-secret-not-for-production-0123456789 AUTH_DEV_SECRET=test-only-secret-not-for-production-0123456789
AUTH_JWKS_URL= AUTH_JWKS_URL=

View File

@ -3,8 +3,9 @@
Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core
NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `recognition.result` в queue NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `recognition.result` в queue
group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается), group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается),
middleware (gzip, proxy headers, CORS; при DEBUG=False пакет РАИСА лишь условно импортируется, middleware не middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False пакет РАИСА лишь условно импортируется,
подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API
(`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`.
Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает
состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и
вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения
@ -21,6 +22,7 @@ from contextlib import asynccontextmanager
from datetime import UTC, datetime from datetime import UTC, datetime
from functools import partial from functools import partial
from http import HTTPStatus from http import HTTPStatus
from typing import Any
import smbclient import smbclient
from fastapi import FastAPI, Request from fastapi import FastAPI, Request
@ -32,9 +34,9 @@ from miniopy_async import Minio
from starlette.middleware.gzip import GZipMiddleware from starlette.middleware.gzip import GZipMiddleware
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from src.api.router import API_PREFIX, router from src.api.router import API_PREFIX, build_api_router
from src.core.clients import S3Storage from src.core.clients import S3Storage
from src.core.config import get_settings from src.core.config import Settings, get_settings
from src.core.exception_handlers import register_exception_handlers from src.core.exception_handlers import register_exception_handlers
from src.core.nats import MessageHandler, NatsConnection from src.core.nats import MessageHandler, NatsConnection
from src.db.db import async_session_maker, engine from src.db.db import async_session_maker, engine
@ -73,9 +75,6 @@ def _normalize_root_path(value: str) -> str:
return "/" + value.strip("/") return "/" + value.strip("/")
_APP_ROOT_PATH = _normalize_root_path(settings.ROOT_PATH)
async def _connect_s3() -> S3Storage | None: async def _connect_s3() -> S3Storage | None:
"""Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует).""" """Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует)."""
s3_client = Minio( s3_client = Minio(
@ -188,36 +187,61 @@ async def lifespan(app: FastAPI):
await _shutdown(app) await _shutdown(app)
app = FastAPI(title="АРМ КА", version=VERSION, root_path=settings.ROOT_PATH, lifespan=lifespan) def cors_options(app_settings: Settings) -> dict[str, Any] | None:
"""Источники CORS (OQ-65): DEBUG — `*` с credentials, как в шаблоне; иначе — `CORS_ORIGINS`.
Пустой `CORS_ORIGINS` вне DEBUG — `None`: middleware не подключается, браузер кросс-доменные запросы не выполнит.
`allow_credentials` вне DEBUG — только для явного списка без `*` (спецификация CORS запрещает `*` с credentials).
"""
if app_settings.DEBUG:
return {"allow_origins": ["*"], "allow_credentials": True}
origins = list(app_settings.CORS_ORIGINS)
if not origins:
return None
return {"allow_origins": origins, "allow_credentials": "*" not in origins}
@app.get("/healthcheck", status_code=200) #: пути документации API: при DEBUG=false маршрутов нет, а SPA-заглушка отвечает на них 404, а не index.html
async def healthcheck(request: Request): DOC_PATHS = ("/docs", "/docs/oauth2-redirect", "/redoc", "/openapi.json")
"""Живость приложения. ``nats``: ``off`` — NATS выключен, ``ok`` — подключён, ``down`` — включён, но нет связи."""
nats_status = "off"
if settings.NATS_ENABLED:
nc = getattr(request.app.state, "nats", None)
nats_status = "ok" if nc is not None and nc.is_connected else "down"
return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status}
app.include_router(router) def create_app(app_settings: Settings, *, web_dir: str = "web") -> FastAPI:
"""Приложение под настройки: документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66),
CORS — `cors_options`, SPA из `web_dir`. Модульный `app` собран по настройкам процесса; тесты собирают свои."""
docs_enabled = app_settings.DEBUG
application = FastAPI(
title="АРМ КА",
version=VERSION,
root_path=app_settings.ROOT_PATH,
lifespan=lifespan,
docs_url="/docs" if docs_enabled else None,
redoc_url="/redoc" if docs_enabled else None,
openapi_url="/openapi.json" if docs_enabled else None,
)
register_exception_handlers(app) @application.get("/healthcheck", status_code=200)
async def healthcheck(request: Request):
"""Живость приложения. ``nats``: ``off`` — NATS выключен, ``ok`` — подключён, ``down`` — нет связи."""
nats_status = "off"
if app_settings.NATS_ENABLED:
nc = getattr(request.app.state, "nats", None)
nats_status = "ok" if nc is not None and nc.is_connected else "down"
return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status}
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1) application.include_router(build_api_router(app_settings))
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
register_exception_handlers(application)
# --- Раздача собранного SPA (web/) --------------------------------------------------------------- application.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1)
os.makedirs("web", exist_ok=True) application.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])
cors = cors_options(app_settings)
if cors is not None:
application.add_middleware(CORSMiddleware, allow_methods=["*"], allow_headers=["*"], **cors)
# --- Раздача собранного SPA (web/) ---
os.makedirs(web_dir, exist_ok=True)
application.mount("/", SPAStaticFiles(directory=web_dir, html=True), name="web")
return application
def sep_static(path: str, stat_dir: str) -> str: def sep_static(path: str, stat_dir: str) -> str:
@ -227,13 +251,14 @@ def sep_static(path: str, stat_dir: str) -> str:
API_STATIC_PATH = API_PREFIX.strip("/") API_STATIC_PATH = API_PREFIX.strip("/")
_DOC_STATIC_PATHS = frozenset(path.strip("/") for path in DOC_PATHS)
class SPAStaticFiles(StaticFiles): class SPAStaticFiles(StaticFiles):
async def get_response(self, path: str, scope): async def get_response(self, path: str, scope):
# Несуществующий маршрут API — честный 404 (в т.ч. для POST), а не index.html SPA и не 405 от StaticFiles: # Несуществующий маршрут API — честный 404 (в т.ч. для POST), а не index.html SPA и не 405 от StaticFiles:
# иначе dev-маршруты, которых при выключенных флагах нет, выглядели бы как «существующие» # иначе dev-маршруты, которых при выключенных флагах нет, выглядели бы как «существующие»
if path == API_STATIC_PATH or path.startswith(API_STATIC_PATH + "/"): if path == API_STATIC_PATH or path.startswith(API_STATIC_PATH + "/") or path in _DOC_STATIC_PATHS:
raise HTTPException(status_code=HTTPStatus.NOT_FOUND) raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
try: try:
return await super().get_response(path, scope) return await super().get_response(path, scope)
@ -247,4 +272,4 @@ class SPAStaticFiles(StaticFiles):
raise raise
app.mount("/", SPAStaticFiles(directory="web", html=True), name="web") app = create_app(settings)

View File

@ -1,7 +1,9 @@
import json
import os import os
from typing import Annotated, Any
from pydantic import NonNegativeInt, PositiveInt from pydantic import NonNegativeInt, PositiveInt, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
class DBSettings(BaseSettings): class DBSettings(BaseSettings):
@ -56,6 +58,9 @@ class Settings(BaseSettings):
server: ServerSettings = ServerSettings() server: ServerSettings = ServerSettings()
ROOT_PATH: str = "/" ROOT_PATH: str = "/"
VERSION: str = "1.0.0" VERSION: str = "1.0.0"
# CORS при DEBUG=false: разрешённые источники (JSON-список или строка через запятую); пусто — кросс-доменные
# запросы запрещены. При DEBUG=true — всегда `*` (OQ-65)
CORS_ORIGINS: Annotated[list[str], NoDecode] = []
# Аутентификация: dev-эмитент (только для разработки; роутер подключается при AUTH_DEV_ISSUER и DEBUG одновременно) # Аутентификация: dev-эмитент (только для разработки; роутер подключается при AUTH_DEV_ISSUER и DEBUG одновременно)
AUTH_DEV_ISSUER: bool = False AUTH_DEV_ISSUER: bool = False
@ -79,6 +84,17 @@ class Settings(BaseSettings):
# Импорт справочников из выгрузки ЕЦП (§13.3): предельный размер файла, больше — 413 # Импорт справочников из выгрузки ЕЦП (§13.3): предельный размер файла, больше — 413
IMPORT_MAX_BYTES: PositiveInt = 20 * 1024 * 1024 IMPORT_MAX_BYTES: PositiveInt = 20 * 1024 * 1024
@field_validator("CORS_ORIGINS", mode="before")
@classmethod
def _split_origins(cls, value: Any) -> Any:
"""`https://a,https://b` или JSON-список `["https://a"]` → список без пробелов и пустых элементов."""
if not isinstance(value, str):
return value
text = value.strip()
if text.startswith("["):
return json.loads(text)
return [item.strip() for item in text.split(",") if item.strip()]
model_config = SettingsConfigDict( model_config = SettingsConfigDict(
extra="ignore", extra="ignore",
env_file=os.environ.get("ENV_FILE", ".env"), env_file=os.environ.get("ENV_FILE", ".env"),

View File

@ -433,3 +433,103 @@ async def test_shutdown_steps_are_independent_and_engine_disposed_last(
assert events == ["nats", "s3", "smb", "db"] assert events == ["nats", "s3", "smb", "db"]
failed_steps = sorted(r["extra"].get("step") for r in records) failed_steps = sorted(r["extra"].get("step") for r in records)
assert failed_steps == ["nats", "s3"] assert failed_steps == ["nats", "s3"]
# --- /docs и CORS по DEBUG (OQ-66, OQ-65, issue #9 п.2) -----------------------------------------------------------
DOC_PATHS = ["/docs", "/redoc", "/openapi.json"]
ORIGIN = "https://arm-ka.example.ru"
def _app_with(tmp_path: Any, **update: Any) -> Any:
"""Приложение `create_app` поверх копии настроек процесса; SPA — пустой временный каталог."""
return app_module.create_app(app_module.settings.model_copy(update=update), web_dir=str(tmp_path))
async def _request(application: Any, method: str, path: str, **kwargs: Any) -> httpx.Response:
transport = httpx.ASGITransport(app=application)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
return await c.request(method, path, **kwargs)
@pytest.mark.parametrize("path", DOC_PATHS)
async def test_docs_disabled_without_debug(tmp_path: Any, path: str) -> None:
(tmp_path / "index.html").write_text("<html>SPA</html>", encoding="utf-8")
application = _app_with(tmp_path, DEBUG=False)
response = await _request(application, "GET", path)
assert response.status_code == HTTPStatus.NOT_FOUND # не схема и не index.html SPA
assert "SPA" not in response.text
@pytest.mark.parametrize("path", DOC_PATHS)
async def test_docs_enabled_with_debug(tmp_path: Any, path: str) -> None:
response = await _request(_app_with(tmp_path, DEBUG=True), "GET", path)
assert response.status_code == HTTPStatus.OK
def _preflight_headers() -> dict[str, str]:
return {"Origin": ORIGIN, "Access-Control-Request-Method": "GET"}
async def test_cors_debug_allows_any_origin(tmp_path: Any) -> None:
response = await _request(_app_with(tmp_path, DEBUG=True), "OPTIONS", "/healthcheck", headers=_preflight_headers())
assert response.status_code == HTTPStatus.OK
assert response.headers["access-control-allow-origin"] in {"*", ORIGIN}
async def test_cors_prod_without_origins_denies_cross_origin(tmp_path: Any) -> None:
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=[])
preflight = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
simple = await _request(application, "GET", "/healthcheck", headers={"Origin": ORIGIN})
assert "access-control-allow-origin" not in preflight.headers
assert simple.status_code == HTTPStatus.OK
assert "access-control-allow-origin" not in simple.headers
async def test_cors_prod_allows_listed_origin_with_credentials(tmp_path: Any) -> None:
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=[ORIGIN])
allowed = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
other = await _request(
application,
"OPTIONS",
"/healthcheck",
headers={"Origin": "https://evil.example.com", "Access-Control-Request-Method": "GET"},
)
assert allowed.status_code == HTTPStatus.OK
assert allowed.headers["access-control-allow-origin"] == ORIGIN
assert allowed.headers["access-control-allow-credentials"] == "true"
assert "access-control-allow-origin" not in other.headers
async def test_cors_prod_wildcard_without_credentials(tmp_path: Any) -> None:
application = _app_with(tmp_path, DEBUG=False, CORS_ORIGINS=["*"])
response = await _request(application, "OPTIONS", "/healthcheck", headers=_preflight_headers())
assert response.headers["access-control-allow-origin"] == "*"
assert "access-control-allow-credentials" not in response.headers
def test_cors_options_rules() -> None:
base = app_module.settings
assert app_module.cors_options(base.model_copy(update={"DEBUG": True, "CORS_ORIGINS": []})) == {
"allow_origins": ["*"],
"allow_credentials": True,
}
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": []})) is None
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": [ORIGIN]})) == {
"allow_origins": [ORIGIN],
"allow_credentials": True,
}
assert app_module.cors_options(base.model_copy(update={"DEBUG": False, "CORS_ORIGINS": ["*", ORIGIN]})) == {
"allow_origins": ["*", ORIGIN],
"allow_credentials": False,
}

View File

@ -123,3 +123,28 @@ def test_env_example_listens_on_loopback(clean_env: pytest.MonkeyPatch) -> None:
"""Dev-запуск по `.env.example` слушает только 127.0.0.1 (финальное ревью M-11).""" """Dev-запуск по `.env.example` слушает только 127.0.0.1 (финальное ревью M-11)."""
s = Settings(_env_file=API_DIR / ".env.example") # type: ignore[call-arg] s = Settings(_env_file=API_DIR / ".env.example") # type: ignore[call-arg]
assert (s.server.HOST, s.server.PORT) == ("127.0.0.1", 8000) assert (s.server.HOST, s.server.PORT) == ("127.0.0.1", 8000)
# --- CORS_ORIGINS (OQ-65, issue #9 п.2) ---------------------------------------------------------------------------
def test_cors_origins_default_is_empty(clean_env: pytest.MonkeyPatch) -> None:
clean_env.delenv("CORS_ORIGINS", raising=False)
s = Settings(_env_file=API_DIR / ".env.test") # type: ignore[call-arg]
assert s.CORS_ORIGINS == []
@pytest.mark.parametrize(
("raw", "expected"),
[
("https://a.example.ru,https://b.example.ru", ["https://a.example.ru", "https://b.example.ru"]),
(" https://a.example.ru , ", ["https://a.example.ru"]),
('["https://a.example.ru", "https://b.example.ru"]', ["https://a.example.ru", "https://b.example.ru"]),
("", []),
],
ids=["comma", "spaces-and-trailing-comma", "json", "empty"],
)
def test_cors_origins_from_env(clean_env: pytest.MonkeyPatch, raw: str, expected: list[str]) -> None:
clean_env.setenv("CORS_ORIGINS", raw)
s = Settings(_env_file=API_DIR / ".env.test") # type: ignore[call-arg]
assert s.CORS_ORIGINS == expected