arm-core/api/src/core/config.py
Andrey Epifancev f47cbeedda fix: X-Forwarded-For только от доверенных прокси — настройка FORWARDED_ALLOW_IPS (ревью R-2, OQ-65, #9)
`ProxyHeadersMiddleware(trusted_hosts=settings.FORWARDED_ALLOW_IPS)` вместо
`["*"]`: по умолчанию `127.0.0.1`, список через запятую или `*`. Иначе любой
клиент подставлял свой адрес в `audit_event.ip_address` (в том числе в
LOGIN_FAILED). `.env.example`/`.env.test`, README «Конфигурация» и
«Безопасность dev-режима» дополнены; compose `docs/dev` не менялся.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
2026-10-05 13:46:17 +04:00

114 lines
4.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import json
import os
from typing import Annotated, Any
from pydantic import NonNegativeInt, PositiveInt, field_validator
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
class DBSettings(BaseSettings):
DATABASE_URL: str
class MinioSettings(BaseSettings):
MINIO_URL: str
MINIO_ADMIN_LOGIN: str
MINIO_ADMIN_PASSWORD: str
MINIO_BUCKET: str
MINIO_SECURE: bool = True # https (шаблон, платформа); false — http (MinIO из docker compose)
class NATSSettings(BaseSettings):
NATS_URL: str
NATS_SEED: str = "" # пусто — подключение к NATS без nkeys (локальная разработка, тесты)
NATS_JWT: str = ""
class SMBSettings(BaseSettings):
"""SMB (горячие папки ИС Сканирование, шаблон). Необязательны: ядру в Срезе 0 SMB не нужен; пустой `SMB_HOST` —
сессия SMB не регистрируется."""
SMB_HOST: str = ""
SMB_PATH: str = ""
SMB_LOGIN: str = ""
SMB_PASSWORD: str = ""
class OpenBAOSettings(BaseSettings):
DATABASE: DBSettings
MINIO: MinioSettings
NATS: NATSSettings
SMB: SMBSettings = SMBSettings() # все поля необязательны (см. SMBSettings)
class ServerSettings(BaseSettings):
HOST: str = "0.0.0.0"
PORT: int = 8000
class Settings(BaseSettings):
"""Настройки приложения.
Файл окружения выбирается переменной ``ENV_FILE`` (по умолчанию ``.env``; значение читается при импорте модуля,
относительный путь считается от текущего каталога — ``api/``). Тесты используют ``.env.test``.
"""
OPENBAO: OpenBAOSettings
DEBUG: bool = False
server: ServerSettings = ServerSettings()
ROOT_PATH: str = "/"
VERSION: str = "1.0.0"
# CORS при DEBUG=false: разрешённые источники (JSON-список или строка через запятую); пусто — кросс-доменные
# запросы запрещены. При DEBUG=true — всегда `*` (OQ-65)
CORS_ORIGINS: Annotated[list[str], NoDecode] = []
# Доверенные прокси для X-Forwarded-For/-Proto (`ProxyHeadersMiddleware`): адреса через запятую или `*`. Только от
# них берётся адрес клиента (`audit_event.ip_address`); от остальных заголовок игнорируется (OQ-65, ревью R-2)
FORWARDED_ALLOW_IPS: str = "127.0.0.1"
# Аутентификация: dev-эмитент (только для разработки; роутер подключается при AUTH_DEV_ISSUER и DEBUG одновременно)
AUTH_DEV_ISSUER: bool = False
AUTH_DEV_SECRET: str = ""
# Проверка токенов РАИСА: JWKS по URL либо статический публичный ключ (PEM)
AUTH_JWKS_URL: str = ""
AUTH_PUBLIC_KEY: str = ""
AUTH_ISSUER: str = ""
AUTH_AUDIENCE: str = ""
AUTH_EMAIL_CLAIM: str = "email" # claim токена с email пользователя (сопоставляется с app_user.email)
# NATS и outbox/inbox обмена recognition_job
NATS_ENABLED: bool = False
# 0 и отрицательные значения отвергаются при старте: период 0 — горячий цикл, ack-таймаут 0 — повтор каждый проход
OUTBOX_POLL_SECONDS: PositiveInt = 5 # период опроса outbox
ACK_TIMEOUT_SECONDS: PositiveInt = 30 # ожидание ack сервиса распознавания до повторной публикации
OUTBOX_MAX_REPUBLISH: NonNegativeInt = (
3 # число повторных публикаций до перевода в dlq (0 — сразу dlq без повторов)
)
# Импорт справочников из выгрузки ЕЦП (§13.3): предельный размер файла, больше — 413
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(
extra="ignore",
env_file=os.environ.get("ENV_FILE", ".env"),
env_nested_delimiter="__",
nested_model_default_partial_update=False,
)
def get_settings() -> Settings:
return Settings()
settings = get_settings()