NATS: subject'ы и JWT из OPENBAO__NATS__*, SecretStr, скрытие входа в ошибках (#32)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 12:06:11 +04:00 · 2026-10-13 12:06:11 +04:00 · f1f7d784a5
commit f1f7d784a5
parent 1579295961
7 changed files with 189 additions and 14 deletions
--- a/api/contracts/recognition-request.schema.json
+++ b/api/contracts/recognition-request.schema.json
@ -2,7 +2,7 @@
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "arm-ka/recognition-request/v1",
  "title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0",
-  "description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
+  "description": "Тело сообщения на subject запроса; subject — настройка окружения и должен соответствовать полю pipeline (tabular → NATS_TABULAR_REQUEST_SUBJECT, по умолчанию files.tabular_uploaded; image → NATS_IMAGE_REQUEST_SUBJECT, по умолчанию files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
  "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
  "type": "object",
  "required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"],
--- a/api/contracts/recognition-result-message.schema.json
+++ b/api/contracts/recognition-result-message.schema.json
@ -2,7 +2,7 @@
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "arm-ka/recognition-result-message/v1",
  "title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0",
-  "description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
+  "description": "Тело сообщения на subject ответа (настройка окружения NATS_RESULT_SUBJECT, по умолчанию files.tabular_parsed). Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
  "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.",
  "type": "object",
  "required": ["schema_version", "job_id", "file_id", "status", "service"],
--- a/api/src/contract/subjects.py
+++ b/api/src/contract/subjects.py
@ -22,10 +22,12 @@ REQUEST_SUBJECTS: Mapping[str, str] = {
 }


-def request_subject(pipeline: str) -> str:
-    """Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`."""
+def request_subject(pipeline: str, subjects: Mapping[str, str] = REQUEST_SUBJECTS) -> str:
+    """Subject запроса по pipeline. `subjects` — карта `pipeline → subject` окружения
+    (`NATSSettings.request_subjects()`); по умолчанию — имена контракта: `tabular` → `files.tabular_uploaded`,
+    `image` → `files.img_uploaded`."""
    try:
-        return REQUEST_SUBJECTS[pipeline]
+        return subjects[pipeline]
    except KeyError:
        raise ValueError(f"неизвестный pipeline {pipeline!r}: допустимы {', '.join(get_args(Pipeline))}") from None

--- a/api/src/core/config.py
+++ b/api/src/core/config.py
@ -11,16 +11,32 @@
 """

 import os
+import re
 from functools import lru_cache
 from typing import Any, Literal

-from pydantic import NonNegativeFloat, PositiveFloat, PositiveInt, model_validator
+from pydantic import (
+    NonNegativeFloat,
+    PositiveFloat,
+    PositiveInt,
+    SecretStr,
+    ValidationInfo,
+    field_validator,
+    model_validator,
+)
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict

+from src.contract.subjects import IMAGE_REQUEST_SUBJECT, RESULT_SUBJECT, TABULAR_REQUEST_SUBJECT
 from src.core.settings_sources import OpenBaoAppSettingsSource

+# Subject NATS без шаблонов: непустые токены через точку, без пробельных символов, `*` и `>` (docs nats-env-config)
+NATS_SUBJECT_RE = re.compile(r"[^\s.*>]+(?:\.[^\s.*>]+)*")
+NATS_SUBJECT_FIELDS = ("NATS_TABULAR_REQUEST_SUBJECT", "NATS_IMAGE_REQUEST_SUBJECT", "NATS_RESULT_SUBJECT")
+

 class MinioSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
    MINIO_URL: str
    MINIO_ADMIN_LOGIN: str
    MINIO_ADMIN_PASSWORD: str
@ -29,15 +45,57 @@ class MinioSettings(BaseSettings):


 class NATSSettings(BaseSettings):
+    """Подключение и subject'ы NATS — секреты окружения `OPENBAO__NATS__*` (прод и тест — свои наборы).
+
+    Пустые `NATS_SEED` и `NATS_JWT` — подключение без nkeys (локальная разработка, тесты). Subject'ы по умолчанию —
+    имена контракта v1.0; у ядра и сервисов одного окружения они обязаны совпадать (docs, change nats-env-config).
+    """
+
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
    NATS_URL: str
-    NATS_SEED: str = ""  # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты)
-    NATS_JWT: str = ""
+    NATS_SEED: SecretStr = SecretStr("")  # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты)
+    NATS_JWT: SecretStr = SecretStr("")
+    NATS_TABULAR_REQUEST_SUBJECT: str = TABULAR_REQUEST_SUBJECT
+    NATS_IMAGE_REQUEST_SUBJECT: str = IMAGE_REQUEST_SUBJECT
+    NATS_RESULT_SUBJECT: str = RESULT_SUBJECT
+
+    @field_validator(*NATS_SUBJECT_FIELDS)
+    @classmethod
+    def _check_subject(cls, value: str, info: ValidationInfo) -> str:
+        """Конкретный subject без шаблонов: иначе подписка молча слушала бы чужой трафик или ничего."""
+        if not NATS_SUBJECT_RE.fullmatch(value):
+            raise ValueError(
+                f"{info.field_name}={value!r}: нужен subject NATS из непустых токенов через точку, "
+                "без пробелов, '*' и '>'"
+            )
+        return value
+
+    @model_validator(mode="after")
+    def _check_distinct_subjects(self) -> "NATSSettings":
+        """Три subject'а попарно различны: совпадение запроса и ответа — сервис получает свои ответы, двух запросов —
+        каждый запрос получают оба сервиса."""
+        seen: dict[str, str] = {}
+        for field in NATS_SUBJECT_FIELDS:
+            subject = getattr(self, field)
+            if subject in seen:
+                raise ValueError(
+                    f"{seen[subject]} и {field} совпадают ({subject!r}): subject'ы NATS должны различаться"
+                )
+            seen[subject] = field
+        return self
+
+    def request_subjects(self) -> dict[str, str]:
+        """Subject запроса по `pipeline` (`tabular`, `image`) — аргумент `request_subject` контракта."""
+        return {"tabular": self.NATS_TABULAR_REQUEST_SUBJECT, "image": self.NATS_IMAGE_REQUEST_SUBJECT}


 class SMBSettings(BaseSettings):
    """SMB-шара горячих папок ИС Сканирование. Нужна только при `SCAN_HOTFOLDER_MODE=smb` (там все поля обязательны,
    проверяет `Settings`); в режиме local весь блок можно не задавать."""

+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
    SMB_HOST: str = ""
    SMB_PATH: str = ""
    SMB_LOGIN: str = ""
@ -45,6 +103,8 @@ class SMBSettings(BaseSettings):


 class OpenBAOSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
    MINIO: MinioSettings
    NATS: NATSSettings
    SMB: SMBSettings = SMBSettings()  # весь блок необязателен: в режиме local не требуется
@ -61,6 +121,8 @@ class OpenBAOSettings(BaseSettings):


 class ServerSettings(BaseSettings):
+    model_config = SettingsConfigDict(hide_input_in_errors=True)
+
    HOST: str = "0.0.0.0"
    PORT: int = 8000

@ -99,6 +161,7 @@ class Settings(BaseSettings):
        env_file_encoding="utf-8",
        env_nested_delimiter="__",
        nested_model_default_partial_update=False,
+        hide_input_in_errors=True,
    )

    @classmethod
--- a/api/src/core/nats.py
+++ b/api/src/core/nats.py
@ -92,8 +92,8 @@ class PublishFailed(NatsError):

 def build_nats_auth(settings: NATSSettings) -> dict[str, Any]:
    """Параметры аутентификации для `Client.connect`: `{}` без учётных данных или колбэки nkeys/JWT шаблона."""
-    seed = settings.NATS_SEED.strip()
-    jwt = settings.NATS_JWT.strip()
+    seed = settings.NATS_SEED.get_secret_value().strip()
+    jwt = settings.NATS_JWT.get_secret_value().strip()
    if not seed and not jwt:
        return {}
    if not seed or not jwt:
--- a/api/tests/unit/test_config.py
+++ b/api/tests/unit/test_config.py
@ -1,3 +1,6 @@
+import os
+from collections.abc import Callable
+
 import pytest
 from pydantic import ValidationError

@ -35,8 +38,8 @@ def test_defaults_local_mode_without_smb_and_nats_credentials(monkeypatch, minim
    assert s.SCAN_TIMEOUT_SECONDS == 1200
    assert s.SCAN_MAX_INFLIGHT == 4
    assert s.OPENBAO.SMB.SMB_HOST == ""
-    assert s.OPENBAO.NATS.NATS_SEED == ""
-    assert s.OPENBAO.NATS.NATS_JWT == ""
+    assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""
+    assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
    assert not hasattr(s.OPENBAO, "DATABASE")


@ -171,7 +174,7 @@ def test_env_test_file_loads_for_tests():
    assert s.OPENBAO.MINIO.MINIO_URL == "127.0.0.1:9000"
    assert s.OPENBAO.MINIO.MINIO_SECURE is False
    assert s.OPENBAO.NATS.NATS_URL == "nats://127.0.0.1:4222"
-    assert s.OPENBAO.NATS.NATS_SEED == ""
+    assert s.OPENBAO.NATS.NATS_SEED.get_secret_value() == ""


 def test_env_example_loads(monkeypatch):
@ -185,4 +188,102 @@ def test_env_example_loads(monkeypatch):
    assert not hasattr(s, "SCAN_TMP_DIR")
    assert s.SCAN_SETTLE_SECONDS == 2
    assert s.OPENBAO.MINIO.MINIO_BUCKET == "arm-ka"
-    assert s.OPENBAO.NATS.NATS_JWT == ""
+    assert s.OPENBAO.NATS.NATS_JWT.get_secret_value() == ""
+
+
+# --- настройки NATS: subject'ы и секреты окружения OPENBAO__NATS__* (change nats-env-config) ----------------------
+
+REQUIRED_OPENBAO_ENV = {
+    "OPENBAO__MINIO__MINIO_URL": "localhost:9000",
+    "OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "minioadmin",
+    "OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "minioadmin",
+    "OPENBAO__MINIO__MINIO_BUCKET": "arm-ka",
+    "OPENBAO__NATS__NATS_URL": "nats://localhost:4222",
+}
+
+
+@pytest.fixture
+def settings_from_env(monkeypatch, tmp_path) -> Callable[..., Settings]:
+    """Фабрика `Settings(_env_file=None)`: минимальный набор обязательных `OPENBAO__*` плюс `env` поверх; значение
+    `None` удаляет переменную. Файл окружения не читается: иначе удалённая переменная осталась бы в нём."""
+
+    def build(**env: str | None) -> Settings:
+        _clean_env(monkeypatch)
+        for key in [key for key in os.environ if key.startswith("OPENBAO__")]:
+            monkeypatch.delenv(key)
+        values = {**REQUIRED_OPENBAO_ENV, "SCAN_HOTFOLDER_PATH": str(tmp_path), **env}
+        for key, value in values.items():
+            if value is None:
+                monkeypatch.delenv(key, raising=False)
+            else:
+                monkeypatch.setenv(key, value)
+        return Settings(_env_file=None)  # type: ignore[call-arg]
+
+    return build
+
+
+JWT = "eyJ0eXAiOiJKV1QiLCJhbGciOiJlZDI1NTE5LW5rZXkifQ.secret-payload.sig"
+SEED = "SUACSSL3UAHUDXKFSNVUZRF5UHPMWZ6BFDTJ7M6USDXIEDNPPQYYYCU3VY"
+
+
+def test_nats_subjects_default_to_contract_names(settings_from_env) -> None:
+    nats = settings_from_env().OPENBAO.NATS
+    assert nats.NATS_TABULAR_REQUEST_SUBJECT == "files.tabular_uploaded"
+    assert nats.NATS_IMAGE_REQUEST_SUBJECT == "files.img_uploaded"
+    assert nats.NATS_RESULT_SUBJECT == "files.tabular_parsed"
+    assert nats.request_subjects() == {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}
+
+
+def test_nats_subjects_from_openbao(settings_from_env) -> None:
+    nats = settings_from_env(
+        OPENBAO__NATS__NATS_TABULAR_REQUEST_SUBJECT="test.files.tabular_uploaded",
+        OPENBAO__NATS__NATS_IMAGE_REQUEST_SUBJECT="test.files.img_uploaded",
+        OPENBAO__NATS__NATS_RESULT_SUBJECT="test.files.tabular_parsed",
+    ).OPENBAO.NATS
+    assert nats.request_subjects() == {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
+    assert nats.NATS_RESULT_SUBJECT == "test.files.tabular_parsed"
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["", "files.*", "files.>", ">", "files..tabular", ".files", "files.", "files tabular", "files.tabular_uploaded "],
+)
+def test_invalid_subject_rejected(settings_from_env, value: str) -> None:
+    with pytest.raises(ValidationError, match="NATS_RESULT_SUBJECT"):
+        settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT=value)
+
+
+def test_equal_subjects_rejected(settings_from_env) -> None:
+    with pytest.raises(ValidationError, match="NATS_TABULAR_REQUEST_SUBJECT и NATS_RESULT_SUBJECT совпадают"):
+        settings_from_env(OPENBAO__NATS__NATS_RESULT_SUBJECT="files.tabular_uploaded")
+
+
+def test_nats_secrets_hidden_in_repr(settings_from_env) -> None:
+    settings = settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED)
+    assert settings.OPENBAO.NATS.NATS_JWT.get_secret_value() == JWT
+    assert JWT not in repr(settings) and SEED not in repr(settings)
+    assert JWT not in str(settings) and SEED not in str(settings)
+
+
+MINIO_KEYS = ("MINIO_URL", "MINIO_ADMIN_LOGIN", "MINIO_ADMIN_PASSWORD", "MINIO_BUCKET", "MINIO_SECURE")
+
+
+@pytest.mark.parametrize(
+    "missing",
+    [
+        ("OPENBAO__NATS__NATS_URL",),  # нет поля в группе NATS: в ошибке — вход группы NATS (с JWT)
+        tuple(f"OPENBAO__MINIO__{key}" for key in MINIO_KEYS),  # нет группы MINIO: в ошибке — вход всего OPENBAO
+    ],
+)
+def test_validation_error_hides_nats_secrets(settings_from_env, missing: tuple[str, ...]) -> None:
+    """Ошибка «нет поля» печатает вход всей группы — JWT и seed туда попадать не должны (Review Focus 1).
+
+    pydantic урезает длинные значения в тексте ошибки: видны начало и конец секрета, а не он целиком. Поэтому проверяем
+    и голову, и хвост каждого секрета (без скрытия входа в тексте видны оба куска)."""
+    with pytest.raises(ValidationError) as exc_info:
+        settings_from_env(OPENBAO__NATS__NATS_JWT=JWT, OPENBAO__NATS__NATS_SEED=SEED, **dict.fromkeys(missing))
+    text = str(exc_info.value)
+    for secret in (JWT, SEED):
+        assert secret not in text
+        assert secret[:10] not in text
+        assert secret[-10:] not in text
--- a/api/tests/unit/test_contract.py
+++ b/api/tests/unit/test_contract.py
@ -640,6 +640,15 @@ def test_request_subject_rejects_unknown_pipeline() -> None:
        request_subject("ocr")


+def test_request_subject_uses_environment_map() -> None:
+    """Карта `pipeline → subject` окружения (NATSSettings.request_subjects()) подменяет имена контракта."""
+    subjects = {"tabular": "test.files.tabular_uploaded", "image": "test.files.img_uploaded"}
+    assert request_subject("tabular", subjects) == "test.files.tabular_uploaded"
+    assert request_subject("image", subjects) == "test.files.img_uploaded"
+    with pytest.raises(ValueError, match="pipeline"):
+        request_subject("ocr", subjects)
+
+
 def test_request_subject_matches_example_pipeline() -> None:
    for name in ("request-tabular.json", "request-image.json"):
        request = RecognitionRequest.model_validate(_example(name))