diff --git a/deploy-steps.md b/deploy-steps.md index 274aa48..745f72d 100644 --- a/deploy-steps.md +++ b/deploy-steps.md @@ -1,14 +1,153 @@ # usefull commands - Убрать переносы строк: sed -i 's/\r$//' name-file.sh #sed -i 's/\r$//' /tmp/setup-vm.sh - docker compose stop # мягко останавливает контейнеры (не удаляет) - sudo shutdown -h now # выглючить ВМ + - Убрать переносы строк: sed -i 's/\r$//' name-file.sh + - docker compose stop # мягко останавливает контейнеры (не удаляет) + - sudo shutdown -h now # выглючить ВМ + - Генерация паролей + - (со спец.символами) openssl rand -base64 32 + - (без спец.символов) openssl rand -hex 16 + - openssl rand -base64 16 | xclip -selection clipboard # Linux -Туннель до БД: ssh -i C:\Users\MI\.ssh\hansel -L 5432:10.105.90.166:5432 root@111.88.219.213 -N +Туннель до БД: ssh -i C:\Users\MI\.ssh\hansel -L 5433:192.168.0.211:5433 root@135.106.164.58 -N -++ ssh -i C:\Users\MI\.ssh\hansel root@111.88.219.213 +# Разворачиваем инфраструктуру в Селектел ч1 -++ scp -i C:\Users\MI\.ssh\hansel -r "C:\Users\MI\Documents\Assistent\HAN_chat_specification\codebase\backend\deployment\scripts\setup-vm.sh" root@111.88.219.213:/tmp/setup-vm.sh +## Создание сети + +## Создание групп безопасности для сети и для БД + Для сети открываем порты 80, 22, 443 + Для БД открываем порты 5432, 5433 но только из CIDR (192.168.0.211\24) + +## Создаем ВМ, подключаем к новой сети (через порт) и перезагружаем (либо руками вводить новые сетевые настройки - надо выбрать) + +# Подключение к ВМ и первичный скрипт +++ ssh -i C:\Users\MI\.ssh\hansel root@135.106.164.58 +++ scp -i C:\Users\MI\.ssh\hansel -r "C:\Users\MI\Documents\Assistent\HAN_chat_specification\codebase\backend\deployment\scripts\setup-vm.sh" root@135.106.164.58:/tmp/setup-vm.sh chmod +x /tmp/setup-vm.sh sudo /tmp/setup-vm.sh +# Скрипт создал деплой пользователя. Нужно Перенести туда публичный ключ и дать права на файл и папку +## Вставляем публичный ключ новой строкой Далее даем права +sudo nano /home/deploy/.ssh/authorized_keys +sudo chmod 700 /home/deploy/.ssh +sudo chmod 600 /home/deploy/.ssh/authorized_keys +sudo chown -R deploy:deploy /home/deploy/.ssh + +## Пароль деплою для использования sudo +sudo passwd deploy + +## После этого можно подключаться под деплой пользователем +ssh -i C:\Users\MI\.ssh\han_chat_deploy deploy@135.106.164.58 + +## GIT не работал. Проект пришлось копировать. +tar -C HAN_chat_specification/codebase/backend -czf han-chat-backend.tar.gz . +scp -i C:\Users\MI\.ssh\han_chat_deploy C:\Users\MI\Documents\Assistent\han-chat-backend.tar.gz deploy@135.106.164.58:/tmp/ + +## Дальше распаковка +cd /opt/han-chat/backend +tar -xzf /tmp/han-chat-backend.tar.gz + +## Правим переносы строк и даем права на исполнение +find . -type f \( -name '*.sh' -o -name 'validate-env' \) -exec dos2unix {} + +chmod +x scripts/validate-env deployment/scripts/*.sh redis/scripts/*.sh nginx/scripts/*.sh + +# Разворачиваем инфраструктуру в Селектел ч2 + +## Создаем БД +В базе подключаем pgcrypto + +### Копируем СА сертификат + +Вариант Деплоя: +mkdir -p /opt/han-chat/backend/secrets/pg +scp -i C:\Users\MI\.ssh\han_chat_deploy -r C:\Users\MI\Documents\job\HAN_new_life\HANapp\Production\sertificates\CA.pem deploy@135.106.164.58:/opt/han-chat/backend/secrets/pg/ca.pem +chmod 644 /opt/han-chat/backend/secrets/pg/ca.pem + +Вариант Селектела: +mkdir -p ~/.postgresql/ +wget https://storage.dbaas.selcloud.ru/CA.pem -O ~/.postgresql/root.crt +chmod 0600 ~/.postgresql/root.crt + +### Проверка подключения к Postgers в ВМ +nc -zv 192.168.0.211 5433 + +sudo apt update +sudo apt install postgresql-client + +psql "host=master.ef54e3e4-ad3d-4b80-a6af-d63269e0895a.c.dbaas.selcloud.ru \ + port=5432 \ + dbname=han_chat \ + user=dbAdmin \ + sslmode=verify-ca" + +### Заводим пользователей и схемы + +Пользователей создаем через интерфейс селектела. +Под админом даем права на создание в БД: + GRANT CREATE ON DATABASE han_chat TO han_app; + GRANT CREATE ON DATABASE han_chat TO bitrix_local_app; + GRANT CREATE ON DATABASE han_chat TO bitrix_sync_user; + GRANT CREATE ON DATABASE han_chat TO message_safety_app; + GRANT CREATE ON DATABASE han_chat TO keycloak_user; + +Схемы создаем от лица пользователей, заходя каждым из них в БД. ++ Запрещаем всем посторонним входить в схему han_app и др. +CREATE SCHEMA IF NOT EXISTS han_app; +REVOKE ALL ON SCHEMA han_app FROM PUBLIC; + +CREATE SCHEMA IF NOT EXISTS bitrix_local; +REVOKE ALL ON SCHEMA bitrix_local FROM PUBLIC; + +CREATE SCHEMA IF NOT EXISTS bitrix_sync; +REVOKE ALL ON SCHEMA bitrix_sync FROM PUBLIC; + +CREATE SCHEMA IF NOT EXISTS message_safety; +REVOKE ALL ON SCHEMA message_safety FROM PUBLIC; + +CREATE SCHEMA IF NOT EXISTS keycloak; +REVOKE ALL ON SCHEMA keycloak FROM PUBLIC; + + +После реализации bitrix_sync (проверить, вероятно не на все таблицы права нужны): +-- Даем право на чтение (SELECT) всех СУЩЕСТВУЮЩИХ таблиц в схеме +GRANT SELECT, INSERT ON ALL TABLES IN SCHEMA han_app TO bitrix_sync_user; +-- Настраиваем права по умолчанию для новых таблиц +ALTER DEFAULT PRIVILEGES IN SCHEMA han_app +GRANT SELECT ON TABLES TO bitrix_sync_user; +-- Если нужно дать право на чтение и для новых последовательностей (sequences): +ALTER DEFAULT PRIVILEGES IN SCHEMA han_app +GRANT USAGE, SELECT ON SEQUENCES TO bitrix_sync_user; + +## Заводим S3 хранилища: + +Создаем 2 сервисных пользователя и заводим им ключи: + - API backend SELECTEL_S3_ACCESS_KEY / SELECTEL_S3_SECRET_KEY + - Message Safety SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY / SELECTEL_S3_QUARANTINE_READ_SECRET_KEY + +Создаем 3 приватных бакета + - han-chat-quarantine (политика доступа API backend RW, Message Safety R) + - han-chat-attachments (политика доступа API backend RW) + - han-chat-documents (политика доступа API backend RW) + +RW: + ListBucket + GetBucketLocation + PutObject + GetObject + DeleteObject + ListBucketMultipartUploads + ListMultipartUploadParts + AbortMultipartUpload + +R: + ListBucket + GetBucketLocation + GetObject + ListBucketMultipartUploads + ListMultipartUploadParts + +Политика CORS: +Allowed origin: https://chat.example.ru +Methods: POST, PUT, GET, HEAD +Headers: Content-Type, x-amz-* +Expose headers: ETag \ No newline at end of file diff --git a/infratest/.env.example b/infratest/.env.example new file mode 100644 index 0000000..892aad2 --- /dev/null +++ b/infratest/.env.example @@ -0,0 +1,53 @@ +# Скопируйте в .env, заполните секреты и выполните: +# chmod 600 .env +# python3 infratest.py --mode full + +# Managed PostgreSQL. Указывайте адрес из ВМ и CA-файл на самой ВМ. +HAN_PG_HOST=managed-pg.private.example +HAN_PG_PORT=6432 +HAN_PG_DATABASE=han_chat +HAN_PG_SSLMODE=verify-full +HAN_PG_SSLROOTCERT=/opt/han-chat/secrets/pg/ca.pem + +# Имена ролей и схем совпадают с контрактом HAN Chat. +HAN_PG_USER_HAN_APP=han_app +HAN_PG_SCHEMA_HAN_APP=han_app +HAN_PG_PASSWORD_HAN_APP= + +HAN_PG_USER_BITRIX=bitrix_local_app +HAN_PG_SCHEMA_BITRIX=bitrix_local +HAN_PG_PASSWORD_BITRIX= + +HAN_PG_USER_BITRIX_SYNC=bitrix_sync_user +HAN_PG_SCHEMA_BITRIX_SYNC=bitrix_sync +HAN_PG_PASSWORD_BITRIX_SYNC= + +HAN_PG_USER_MESSAGE_SAFETY=message_safety_app +HAN_PG_SCHEMA_MESSAGE_SAFETY=message_safety +HAN_PG_PASSWORD_MESSAGE_SAFETY= + +HAN_PG_USER_KEYCLOAK=keycloak_user +HAN_PG_SCHEMA_KEYCLOAK=keycloak +HAN_PG_PASSWORD_KEYCLOAK= + +# Selectel S3: основной RW-ключ API backend. +SELECTEL_S3_ENDPOINT_URL=https://s3.storage.selcloud.ru +# Не задавайте регион, если приложение также использует значение SDK по умолчанию. +SELECTEL_S3_REGION= +SELECTEL_S3_BUCKET_QUARANTINE=han-chat-quarantine +SELECTEL_S3_BUCKET_ATTACHMENTS=han-chat-attachments +SELECTEL_S3_BUCKET_DOCUMENTS=han-chat-documents +SELECTEL_S3_ACCESS_KEY= +SELECTEL_S3_SECRET_KEY= + +# Отдельный read-only ключ message-safety: R только для quarantine. +SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY= +SELECTEL_S3_QUARANTINE_READ_SECRET_KEY= + +# Точный origin веб-приложения для CORS-проверки. +PUBLIC_WEB_URL=https://chat.example.ru + +# Проверки после миграций Alembic включайте только после deployment/scripts/migrate.sh. +INFRATEST_CHECK_MIGRATIONS=false +INFRATEST_CHECK_CORS=true +INFRATEST_TIMEOUT_SECONDS=15 diff --git a/infratest/.gitignore b/infratest/.gitignore new file mode 100644 index 0000000..2070686 --- /dev/null +++ b/infratest/.gitignore @@ -0,0 +1,6 @@ +.env +.venv/ +.ruff_cache/ +__pycache__/ +*.pyc +report*.json diff --git a/infratest/README.md b/infratest/README.md new file mode 100644 index 0000000..c9692a3 --- /dev/null +++ b/infratest/README.md @@ -0,0 +1,180 @@ +# Инфраструктурный тест HAN Chat + +Автономный тест запускается непосредственно на ВМ и проверяет managed PostgreSQL и +три приватных Selectel S3-бакета в соответствии с контрактами HAN Chat. Для запуска +приложение и Docker Compose не требуются. + +Тест не создаёт постоянных данных: + +- PostgreSQL DDL/DML выполняются в транзакции и завершаются `ROLLBACK`; +- S3-объекты создаются только под уникальным префиксом `infratest//`; +- все S3-объекты удаляются в блоке очистки даже при ошибке одной из проверок. + +## Что проверяется + +### PostgreSQL + +- подключение пяти сервисных ролей с TLS `verify-full`; +- `current_user`, `search_path`, наличие своей схемы и отсутствие `USAGE` на четыре + чужие схемы; +- `CREATE TABLE`, `INSERT`, `SELECT` и фактическое отсутствие таблицы после + `ROLLBACK`; +- расширение `pgcrypto`; +- опционально — ключевые таблицы и Alembic revision после миграций. + +Проверяются роли: + +- `han_app` → `han_app`; +- `bitrix_local_app` → `bitrix_local`; +- `bitrix_sync_user` → `bitrix_sync`; +- `message_safety_app` → `message_safety`; +- `keycloak_user` → `keycloak`. + +### S3 + +- `HeadBucket` для quarantine, attachments и documents; +- прямые `PutObject`, `HeadObject`, `GetObject`, `ListObjectsV2` с ВМ; +- presigned PUT и GET с побайтовой проверкой содержимого; +- проектный переход `CopyObject` из quarantine в attachments с удалением + исходного объекта; +- запрет анонимного GET; +- CORS preflight для точного `PUBLIC_WEB_URL`; +- отдельный ключ message-safety: чтение/list только quarantine, запрет записи, + удаления и доступа к attachments/documents. + +Multipart API не проверяется: текущая реализация HAN Chat использует single PUT. +Прикладные сценарии авторизации, создания диалога и отправки сообщения также не +входят в этот инструмент — это E2E приложения, а не проверка инфраструктуры. + +## Подготовка на ВМ + +Рекомендуется скопировать всю папку в отдельный каталог: + +```bash +mkdir -p /opt/han-chat/infratest +cd /opt/han-chat/infratest +``` + +Установите Python 3.10+ и создайте изолированное окружение: + +```bash +python3 -m venv .venv +. .venv/bin/activate +python3 -m pip install --upgrade pip +python3 -m pip install -r requirements.txt +``` + +Создайте настройки: + +```bash +cp .env.example .env +chmod 600 .env +nano .env +``` + +В `.env` нужны сервисные пароли PostgreSQL, основной S3 RW-ключ и отдельный +quarantine read-only ключ. Административный пароль PostgreSQL не используется. +`HAN_PG_SSLROOTCERT` должен указывать на CA-файл по пути, доступному из процесса +на ВМ, а не на контейнерный путь `/run/secrets/pg-ca.pem`. + +Не передавайте секреты аргументами командной строки и не прикладывайте `.env` к +отчётам. Файл исключён через `.gitignore`. + +## Запуск + +Полная самочищающаяся проверка: + +```bash +cd /opt/han-chat/infratest +. .venv/bin/activate +python3 infratest.py --mode full +``` + +Проверка без DDL и записи в S3: + +```bash +python3 infratest.py --mode readonly +``` + +Другой файл настроек и JSON-отчёт: + +```bash +python3 infratest.py \ + --env-file /secure/path/infratest.env \ + --mode full \ + --json-report report.json +``` + +Скрипт возвращает: + +- `0` — нет проваленных проверок; +- `1` — одна или несколько инфраструктурных проверок завершились ошибкой; +- `2` — неверная конфигурация или отсутствуют Python-зависимости. + +В консоль и JSON не выводятся пароли, access keys и query-параметры presigned URL. + +## Запуск до и после миграций + +До запуска Alembic оставьте: + +```dotenv +INFRATEST_CHECK_MIGRATIONS=false +``` + +После успешного `deployment/scripts/migrate.sh` измените значение на: + +```dotenv +INFRATEST_CHECK_MIGRATIONS=true +``` + +В этом режиме дополнительно проверяются ключевые таблицы схем `han_app`, +`bitrix_local`, `bitrix_sync` и ожидаемая revision `0001_initial` для `han_app`. + +## Ожидаемая IAM-модель + +Основной `SELECTEL_S3_ACCESS_KEY` должен иметь: + +- RW на quarantine, attachments и documents; +- `ListBucket`, `GetBucketLocation`, `PutObject`, `GetObject`, `DeleteObject`; +- `CopyObject` обеспечивается правами чтения источника и записи назначения. + +Ключ `SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY` должен иметь только: + +- `ListBucket`, `GetBucketLocation`, `GetObject` для quarantine; +- никаких прав на Put/Delete; +- никаких прав на attachments и documents. + +Если IAM-политика ограничивает основной ключ другими object prefixes, разрешите +служебный префикс `infratest/*` либо запускайте только `--mode readonly`. + +## CORS + +При `INFRATEST_CHECK_CORS=true` quarantine должен отвечать на preflight: + +- origin — точное значение `PUBLIC_WEB_URL`, без wildcard; +- method — `PUT`; +- headers — `Content-Type`, `x-amz-*`. + +Чтобы временно исключить CORS из диагностики: + +```dotenv +INFRATEST_CHECK_CORS=false +``` + +## Интерпретация результата + +- `PASS` — контракт подтверждён реальной операцией; +- `FAIL` — контракт нарушен или ресурс недоступен; +- `SKIP` — проверка отключена режимом или настройкой. + +При `FAIL` сначала проверьте сетевые ACL ВМ, CA/hostname PostgreSQL, GRANT и +`search_path`, затем IAM/CORS бакетов. После аварийного прерывания процесса можно +найти остатки только под префиксом `infratest/`; удалять другие ключи не требуется. + +## Локальные unit-тесты + +Они не обращаются к PostgreSQL или S3: + +```bash +python3 -m unittest -v test_infratest.py +``` diff --git a/infratest/infratest.py b/infratest/infratest.py new file mode 100644 index 0000000..fe63c03 --- /dev/null +++ b/infratest/infratest.py @@ -0,0 +1,844 @@ +#!/usr/bin/env python3 +"""Самоочищающийся инфраструктурный тест PostgreSQL и S3 для HAN Chat.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable +from urllib.parse import urlsplit, urlunsplit + +try: + import boto3 + import psycopg + import requests + from botocore.config import Config as BotoConfig + from botocore.exceptions import ClientError + from dotenv import dotenv_values + from psycopg import sql +except ImportError as exc: # pragma: no cover - понятная ошибка до запуска тестов + print( + f"Не установлена зависимость {exc.name!r}. " + "Выполните: python3 -m pip install -r requirements.txt", + file=sys.stderr, + ) + raise SystemExit(2) from exc + + +URL_PATTERN = re.compile(r"https?://[^\s]+") +EXPECTED_ROLES = ( + ("han_app", "han_app", "HAN_PG_PASSWORD_HAN_APP"), + ("bitrix_local_app", "bitrix_local", "HAN_PG_PASSWORD_BITRIX"), + ("bitrix_sync_user", "bitrix_sync", "HAN_PG_PASSWORD_BITRIX_SYNC"), + ("message_safety_app", "message_safety", "HAN_PG_PASSWORD_MESSAGE_SAFETY"), + ("keycloak_user", "keycloak", "HAN_PG_PASSWORD_KEYCLOAK"), +) +EXPECTED_TABLES = { + "han_app": ( + "alembic_version", + "app_settings", + "dialogs", + "messages", + "message_attachments", + "documents", + "sync_queue", + ), + "bitrix_local": ( + "alembic_version", + "portal_installations", + "dialog_sessions", + "inbox_events", + "outbound_messages", + ), + "bitrix_sync": ("alembic_version",), +} + + +class ConfigError(ValueError): + """Ошибка пользовательской конфигурации.""" + + +def as_bool(value: str | None, default: bool = False) -> bool: + if value is None or not value.strip(): + return default + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise ConfigError(f"Ожидалось логическое значение, получено: {value!r}") + + +def sanitize(value: Any) -> str: + """Удаляет секреты и query-параметры URL из диагностического текста.""" + text = str(value) + text = URL_PATTERN.sub( + lambda match: urlunsplit( + (*urlsplit(match.group(0))[:3], "", "") + ), + text, + ) + text = re.sub( + r"(?i)\b(password|secret|access[_-]?key|authorization|token|dsn)" + r"\s*[:=]\s*[^\s,;]+", + lambda match: f"{match.group(1)}=", + text, + ) + return text[:1000] + + +def require(values: dict[str, str], name: str) -> str: + value = values.get(name, "").strip() + if not value: + raise ConfigError(f"Не задана обязательная переменная {name}") + return value + + +@dataclass(frozen=True) +class PgRole: + user: str + schema: str + password: str = field(repr=False) + + +@dataclass(frozen=True) +class Settings: + pg_host: str + pg_port: int + pg_database: str + pg_sslmode: str + pg_sslrootcert: str + pg_roles: tuple[PgRole, ...] + s3_endpoint: str + s3_region: str | None + buckets: dict[str, str] + s3_access_key: str = field(repr=False) + s3_secret_key: str = field(repr=False) + s3_read_access_key: str = field(repr=False) + s3_read_secret_key: str = field(repr=False) + public_web_url: str | None + check_cors: bool + check_migrations: bool + timeout_seconds: int + + @classmethod + def from_env_file(cls, path: Path) -> "Settings": + if not path.is_file(): + raise ConfigError(f"Файл настроек не найден: {path}") + raw = dotenv_values(path) + values = { + key: str(value) + for key, value in raw.items() + if value is not None + } + # Явно экспортированные переменные имеют приоритет над файлом. + values.update({key: value for key, value in os.environ.items() if key in values}) + + sslmode = values.get("HAN_PG_SSLMODE", "verify-full").strip() + if sslmode != "verify-full": + raise ConfigError("HAN_PG_SSLMODE должен быть verify-full") + ca_path = require(values, "HAN_PG_SSLROOTCERT") + if not Path(ca_path).is_file(): + raise ConfigError(f"CA-файл PostgreSQL не найден: {ca_path}") + + endpoint = require(values, "SELECTEL_S3_ENDPOINT_URL") + parsed_endpoint = urlsplit(endpoint) + if parsed_endpoint.scheme != "https" or not parsed_endpoint.netloc: + raise ConfigError("SELECTEL_S3_ENDPOINT_URL должен быть корректным HTTPS URL") + + roles = tuple( + PgRole( + values.get(f"HAN_PG_USER_{env_suffix}", default_user).strip(), + values.get(f"HAN_PG_SCHEMA_{env_suffix}", default_schema).strip(), + require(values, password_env), + ) + for default_user, default_schema, password_env in EXPECTED_ROLES + for env_suffix in ( + { + "han_app": "HAN_APP", + "bitrix_local": "BITRIX", + "bitrix_sync": "BITRIX_SYNC", + "message_safety": "MESSAGE_SAFETY", + "keycloak": "KEYCLOAK", + }[default_schema], + ) + ) + timeout = int(values.get("INFRATEST_TIMEOUT_SECONDS", "15")) + if timeout < 1 or timeout > 300: + raise ConfigError("INFRATEST_TIMEOUT_SECONDS должен быть от 1 до 300") + + public_web_url = values.get("PUBLIC_WEB_URL", "").strip() or None + check_cors = as_bool(values.get("INFRATEST_CHECK_CORS"), True) + if check_cors and not public_web_url: + raise ConfigError("Для CORS-проверки задайте PUBLIC_WEB_URL") + + return cls( + pg_host=require(values, "HAN_PG_HOST"), + pg_port=int(values.get("HAN_PG_PORT", "5432")), + pg_database=require(values, "HAN_PG_DATABASE"), + pg_sslmode=sslmode, + pg_sslrootcert=ca_path, + pg_roles=roles, + s3_endpoint=endpoint.rstrip("/"), + s3_region=values.get("SELECTEL_S3_REGION", "").strip() or None, + buckets={ + "quarantine": require(values, "SELECTEL_S3_BUCKET_QUARANTINE"), + "attachments": require(values, "SELECTEL_S3_BUCKET_ATTACHMENTS"), + "documents": require(values, "SELECTEL_S3_BUCKET_DOCUMENTS"), + }, + s3_access_key=require(values, "SELECTEL_S3_ACCESS_KEY"), + s3_secret_key=require(values, "SELECTEL_S3_SECRET_KEY"), + s3_read_access_key=require( + values, "SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY" + ), + s3_read_secret_key=require( + values, "SELECTEL_S3_QUARANTINE_READ_SECRET_KEY" + ), + public_web_url=public_web_url, + check_cors=check_cors, + check_migrations=as_bool( + values.get("INFRATEST_CHECK_MIGRATIONS"), False + ), + timeout_seconds=timeout, + ) + + +@dataclass(frozen=True) +class Result: + check: str + status: str + detail: str + + +class Reporter: + def __init__(self) -> None: + self.results: list[Result] = [] + + def add(self, check: str, status: str, detail: Any) -> None: + result = Result(check, status, sanitize(detail)) + self.results.append(result) + print(f"[{status:4}] {check}: {result.detail}") + + def run( + self, + check: str, + operation: Callable[[], Any], + success_detail: str = "OK", + ) -> bool: + try: + detail = operation() + except Exception as exc: # noqa: BLE001 - каждая проверка должна продолжить отчёт + self.add(check, "FAIL", f"{type(exc).__name__}: {exc}") + return False + self.add(check, "PASS", success_detail if detail is None else detail) + return True + + def skip(self, check: str, detail: str) -> None: + self.add(check, "SKIP", detail) + + @property + def exit_code(self) -> int: + return 1 if any(item.status == "FAIL" for item in self.results) else 0 + + def write_json(self, path: Path) -> None: + path.write_text( + json.dumps( + [item.__dict__ for item in self.results], + ensure_ascii=False, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + + +class InfraTest: + def __init__(self, settings: Settings, reporter: Reporter, mode: str) -> None: + self.settings = settings + self.reporter = reporter + self.mode = mode + self.run_id = uuid.uuid4().hex + self.prefix = f"infratest/{self.run_id}" + self.payload = ( + b"HAN Chat infrastructure test\n" + + self.run_id.encode("ascii") + + b"\n" + ) + self.api_s3 = self._s3_client( + settings.s3_access_key, settings.s3_secret_key + ) + self.read_s3 = self._s3_client( + settings.s3_read_access_key, settings.s3_read_secret_key + ) + self.created_objects: set[tuple[str, str]] = set() + + def _s3_client(self, access_key: str, secret_key: str): + kwargs: dict[str, Any] = { + "service_name": "s3", + "endpoint_url": self.settings.s3_endpoint, + "aws_access_key_id": access_key, + "aws_secret_access_key": secret_key, + "config": BotoConfig( + connect_timeout=self.settings.timeout_seconds, + read_timeout=self.settings.timeout_seconds, + retries={"max_attempts": 2}, + ), + } + if self.settings.s3_region: + kwargs["region_name"] = self.settings.s3_region + return boto3.client(**kwargs) + + def _pg_connect(self, role: PgRole): + return psycopg.connect( + host=self.settings.pg_host, + port=self.settings.pg_port, + dbname=self.settings.pg_database, + user=role.user, + password=role.password, + sslmode=self.settings.pg_sslmode, + sslrootcert=self.settings.pg_sslrootcert, + connect_timeout=self.settings.timeout_seconds, + application_name=f"han-infratest-{self.run_id[:8]}", + ) + + def run(self) -> None: + self.run_postgres() + self.run_s3() + + def run_postgres(self) -> None: + print("\nPostgreSQL") + schemas = tuple(role.schema for role in self.settings.pg_roles) + for role in self.settings.pg_roles: + connection_ok = self.reporter.run( + f"postgres.{role.user}.connect_tls", + lambda role=role: self._check_pg_connection(role), + ) + if not connection_ok: + continue + self.reporter.run( + f"postgres.{role.user}.schema_isolation", + lambda role=role: self._check_schema_isolation(role, schemas), + ) + if self.mode == "full": + self.reporter.run( + f"postgres.{role.user}.ddl_dml_rollback", + lambda role=role: self._check_pg_write(role), + ) + else: + self.reporter.skip( + f"postgres.{role.user}.ddl_dml_rollback", + "режим readonly", + ) + + self.reporter.run("postgres.pgcrypto", self._check_pgcrypto) + if self.settings.check_migrations: + self.reporter.run("postgres.migrations", self._check_migrations) + else: + self.reporter.skip( + "postgres.migrations", + "INFRATEST_CHECK_MIGRATIONS=false", + ) + + def _check_pg_connection(self, role: PgRole) -> str: + with self._pg_connect(role) as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT current_user, current_setting('search_path'), + EXISTS ( + SELECT 1 FROM pg_stat_ssl + WHERE pid = pg_backend_pid() AND ssl + ) + """ + ) + current_user, search_path, tls_active = cur.fetchone() + if current_user != role.user: + raise AssertionError( + f"current_user={current_user!r}, ожидался {role.user!r}" + ) + if role.schema not in search_path.split(","): + raise AssertionError( + f"search_path={search_path!r}, ожидалась {role.schema!r}" + ) + if not tls_active: + raise AssertionError("соединение установлено без TLS") + return f"TLS active, search_path={search_path}" + + def _check_schema_isolation( + self, role: PgRole, all_schemas: tuple[str, ...] + ) -> str: + with self._pg_connect(role) as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT EXISTS ( + SELECT 1 FROM information_schema.schemata + WHERE schema_name = %s + ), has_schema_privilege(current_user, %s, 'USAGE') + """, + (role.schema, role.schema), + ) + schema_exists, own_usage = cur.fetchone() + if not schema_exists or not own_usage: + raise AssertionError( + f"нет схемы или USAGE для {role.schema}" + ) + unexpected = [] + for other in all_schemas: + if other == role.schema: + continue + cur.execute( + "SELECT has_schema_privilege(current_user, %s, 'USAGE')", + (other,), + ) + if cur.fetchone()[0]: + unexpected.append(other) + if unexpected: + raise AssertionError( + "обнаружен USAGE на чужие схемы: " + ", ".join(unexpected) + ) + return f"доступ только к схеме {role.schema}" + + def _check_pg_write(self, role: PgRole) -> str: + table_name = f"_infratest_{self.run_id}" + with self._pg_connect(role) as conn: + try: + with conn.cursor() as cur: + cur.execute( + sql.SQL("CREATE TABLE {}.{} (id integer PRIMARY KEY, value text)") + .format(sql.Identifier(role.schema), sql.Identifier(table_name)) + ) + cur.execute( + sql.SQL("INSERT INTO {}.{} VALUES (%s, %s)").format( + sql.Identifier(role.schema), sql.Identifier(table_name) + ), + (1, "ok"), + ) + cur.execute( + sql.SQL("SELECT value FROM {}.{} WHERE id = %s").format( + sql.Identifier(role.schema), sql.Identifier(table_name) + ), + (1,), + ) + if cur.fetchone()[0] != "ok": + raise AssertionError("прочитано неожиданное значение") + finally: + conn.rollback() + with conn.cursor() as cur: + cur.execute("SELECT to_regclass(%s)", (f"{role.schema}.{table_name}",)) + if cur.fetchone()[0] is not None: + raise AssertionError("таблица сохранилась после ROLLBACK") + return "CREATE/INSERT/SELECT успешны, DDL откачен" + + def _check_pgcrypto(self) -> str: + role = next(item for item in self.settings.pg_roles if item.schema == "han_app") + with self._pg_connect(role) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname='pgcrypto')" + ) + if not cur.fetchone()[0]: + raise AssertionError("расширение pgcrypto не установлено") + return "расширение pgcrypto установлено" + + def _check_migrations(self) -> str: + role_by_schema = {role.schema: role for role in self.settings.pg_roles} + missing: list[str] = [] + revisions: dict[str, str] = {} + for schema, tables in EXPECTED_TABLES.items(): + with self._pg_connect(role_by_schema[schema]) as conn: + with conn.cursor() as cur: + for table in tables: + cur.execute("SELECT to_regclass(%s)", (f"{schema}.{table}",)) + if cur.fetchone()[0] is None: + missing.append(f"{schema}.{table}") + cur.execute( + sql.SQL("SELECT version_num FROM {}.alembic_version LIMIT 1") + .format(sql.Identifier(schema)) + ) + row = cur.fetchone() + revisions[schema] = row[0] if row else "" + if missing: + raise AssertionError("нет таблиц: " + ", ".join(missing)) + if revisions.get("han_app") != "0001_initial": + raise AssertionError( + f"han_app revision={revisions.get('han_app')!r}, ожидался '0001_initial'" + ) + return "таблицы и Alembic revisions присутствуют" + + def run_s3(self) -> None: + print("\nSelectel S3") + reachable: dict[str, bool] = {} + for logical_name, bucket in self.settings.buckets.items(): + reachable[logical_name] = self.reporter.run( + f"s3.{logical_name}.head_bucket", + lambda bucket=bucket: self.api_s3.head_bucket(Bucket=bucket), + f"бакет {bucket} доступен", + ) + + self.reporter.run( + "s3.quarantine_read_key.boundaries", + self._check_read_key_bucket_boundaries, + ) + if self.mode != "full": + for name in ( + "direct_vm_operations", + "presigned_put_get", + "copy_promote", + "cors", + "anonymous_access", + "quarantine_read_key.object_permissions", + ): + self.reporter.skip(f"s3.{name}", "режим readonly") + return + if not all(reachable.values()): + self.reporter.add( + "s3.full_operations", + "FAIL", + "полные проверки невозможны: не все бакеты доступны", + ) + return + + try: + direct_keys: dict[str, str] = {} + for logical_name, bucket in self.settings.buckets.items(): + key = f"{self.prefix}/direct-{logical_name}.bin" + direct_keys[logical_name] = key + self.reporter.run( + f"s3.{logical_name}.direct_vm_operations", + lambda bucket=bucket, key=key: self._check_direct_object( + bucket, key + ), + ) + + presigned_key = f"{self.prefix}/presigned.bin" + presigned_ok = self.reporter.run( + "s3.quarantine.presigned_put_get", + lambda: self._check_presigned(presigned_key), + ) + if presigned_ok: + promoted_key = f"{self.prefix}/promoted.bin" + self.reporter.run( + "s3.quarantine_to_attachments.copy_promote", + lambda: self._check_promote(presigned_key, promoted_key), + ) + + privacy_key = direct_keys["documents"] + self.reporter.run( + "s3.documents.anonymous_access_denied", + lambda: self._check_anonymous_denied( + self.settings.buckets["documents"], privacy_key + ), + ) + self.reporter.run( + "s3.quarantine_read_key.object_permissions", + lambda: self._check_read_key_object_permissions( + direct_keys["quarantine"], + direct_keys["attachments"], + direct_keys["documents"], + ), + ) + if self.settings.check_cors: + cors_key = f"{self.prefix}/cors.bin" + self.reporter.run( + "s3.quarantine.cors_preflight", + lambda: self._check_cors(cors_key), + ) + else: + self.reporter.skip( + "s3.quarantine.cors_preflight", + "INFRATEST_CHECK_CORS=false", + ) + finally: + self._cleanup() + + def _check_direct_object(self, bucket: str, key: str) -> str: + self.api_s3.put_object( + Bucket=bucket, + Key=key, + Body=self.payload, + ContentType="application/octet-stream", + Metadata={"infratest-run": self.run_id}, + ) + self.created_objects.add((bucket, key)) + head = self.api_s3.head_object(Bucket=bucket, Key=key) + if head["ContentLength"] != len(self.payload): + raise AssertionError("HeadObject вернул неверный размер") + response = self.api_s3.get_object(Bucket=bucket, Key=key) + if response["Body"].read() != self.payload: + raise AssertionError("GetObject вернул другие байты") + listed = self.api_s3.list_objects_v2( + Bucket=bucket, Prefix=key, MaxKeys=2 + ).get("Contents", []) + if not any(item["Key"] == key for item in listed): + raise AssertionError("ListObjectsV2 не вернул тестовый объект") + return "Put/Head/Get/List с ВМ успешны" + + def _check_presigned(self, key: str) -> str: + bucket = self.settings.buckets["quarantine"] + put_url = self.api_s3.generate_presigned_url( + "put_object", + Params={ + "Bucket": bucket, + "Key": key, + "ContentType": "application/octet-stream", + }, + ExpiresIn=300, + ) + response = requests.put( + put_url, + data=self.payload, + headers={"Content-Type": "application/octet-stream"}, + timeout=self.settings.timeout_seconds, + ) + response.raise_for_status() + self.created_objects.add((bucket, key)) + self.api_s3.head_object(Bucket=bucket, Key=key) + get_url = self.api_s3.generate_presigned_url( + "get_object", + Params={"Bucket": bucket, "Key": key}, + ExpiresIn=300, + ) + downloaded = requests.get( + get_url, timeout=self.settings.timeout_seconds + ) + downloaded.raise_for_status() + if downloaded.content != self.payload: + raise AssertionError("presigned GET вернул другие байты") + return "presigned PUT/GET успешны, содержимое совпадает" + + def _check_promote(self, source_key: str, destination_key: str) -> str: + quarantine = self.settings.buckets["quarantine"] + attachments = self.settings.buckets["attachments"] + self.api_s3.copy_object( + Bucket=attachments, + Key=destination_key, + CopySource={"Bucket": quarantine, "Key": source_key}, + ) + self.created_objects.add((attachments, destination_key)) + copied = self.api_s3.get_object( + Bucket=attachments, Key=destination_key + )["Body"].read() + if copied != self.payload: + raise AssertionError("скопированный объект повреждён") + self.api_s3.delete_object(Bucket=quarantine, Key=source_key) + self.created_objects.discard((quarantine, source_key)) + try: + self.api_s3.head_object(Bucket=quarantine, Key=source_key) + except ClientError as exc: + if exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode") != 404: + raise + else: + raise AssertionError("исходный quarantine-объект не удалён") + return "CopyObject успешен, исходный объект удалён" + + def _check_anonymous_denied(self, bucket: str, key: str) -> str: + signed_url = self.api_s3.generate_presigned_url( + "get_object", + Params={"Bucket": bucket, "Key": key}, + ExpiresIn=300, + ) + parsed = urlsplit(signed_url) + unsigned_url = urlunsplit( + (parsed.scheme, parsed.netloc, parsed.path, "", "") + ) + response = requests.get( + unsigned_url, + allow_redirects=False, + timeout=self.settings.timeout_seconds, + ) + if response.status_code not in {401, 403}: + raise AssertionError( + f"анонимный GET вернул HTTP {response.status_code}, ожидался 401/403" + ) + return f"анонимный GET запрещён (HTTP {response.status_code})" + + @staticmethod + def _assert_denied(operation: Callable[[], Any], description: str) -> None: + try: + operation() + except ClientError as exc: + status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode") + code = exc.response.get("Error", {}).get("Code", "") + if status in {401, 403} or code in { + "AccessDenied", + "AllAccessDisabled", + "InvalidAccessKeyId", + }: + return + raise + raise AssertionError(f"операция неожиданно разрешена: {description}") + + def _check_read_key_bucket_boundaries(self) -> str: + quarantine = self.settings.buckets["quarantine"] + self.read_s3.head_bucket(Bucket=quarantine) + self.read_s3.list_objects_v2(Bucket=quarantine, MaxKeys=1) + for logical_name in ("attachments", "documents"): + bucket = self.settings.buckets[logical_name] + self._assert_denied( + lambda bucket=bucket: self.read_s3.list_objects_v2( + Bucket=bucket, MaxKeys=1 + ), + f"read-key ListBucket {logical_name}", + ) + return "read-key видит только quarantine" + + def _check_read_key_object_permissions( + self, + quarantine_key: str, + attachments_key: str, + documents_key: str, + ) -> str: + quarantine = self.settings.buckets["quarantine"] + attachments = self.settings.buckets["attachments"] + documents = self.settings.buckets["documents"] + response = self.read_s3.get_object( + Bucket=quarantine, Key=quarantine_key + ) + if response["Body"].read() != self.payload: + raise AssertionError("read-key получил повреждённые данные") + + denied_put_key = f"{self.prefix}/read-key-must-not-put.bin" + try: + self._assert_denied( + lambda: self.read_s3.put_object( + Bucket=quarantine, Key=denied_put_key, Body=self.payload + ), + "read-key PutObject quarantine", + ) + finally: + # Если политика ошибочно разрешила PUT, удалить объект API-ключом. + try: + self.api_s3.delete_object( + Bucket=quarantine, Key=denied_put_key + ) + except Exception: # noqa: BLE001 - основная очистка будет продолжена + pass + + self._assert_denied( + lambda: self.read_s3.delete_object( + Bucket=quarantine, Key=quarantine_key + ), + "read-key DeleteObject quarantine", + ) + self._assert_denied( + lambda: self.read_s3.get_object( + Bucket=attachments, Key=attachments_key + ), + "read-key GetObject attachments", + ) + self._assert_denied( + lambda: self.read_s3.get_object( + Bucket=documents, Key=documents_key + ), + "read-key GetObject documents", + ) + return "GET quarantine разрешён; запись, удаление и другие бакеты запрещены" + + def _check_cors(self, key: str) -> str: + url = self.api_s3.generate_presigned_url( + "put_object", + Params={ + "Bucket": self.settings.buckets["quarantine"], + "Key": key, + "ContentType": "application/octet-stream", + }, + ExpiresIn=300, + ) + response = requests.options( + url, + headers={ + "Origin": self.settings.public_web_url or "", + "Access-Control-Request-Method": "PUT", + "Access-Control-Request-Headers": "content-type,x-amz-meta-infratest", + }, + timeout=self.settings.timeout_seconds, + ) + if response.status_code not in {200, 204}: + raise AssertionError(f"preflight вернул HTTP {response.status_code}") + allow_origin = response.headers.get("Access-Control-Allow-Origin") + if allow_origin != self.settings.public_web_url: + raise AssertionError( + f"Access-Control-Allow-Origin={allow_origin!r}, " + f"ожидался точный PUBLIC_WEB_URL" + ) + allow_methods = response.headers.get("Access-Control-Allow-Methods", "") + if "PUT" not in allow_methods.upper(): + raise AssertionError("CORS не разрешает PUT") + return "CORS разрешает presigned PUT только с PUBLIC_WEB_URL" + + def _cleanup(self) -> None: + failures: list[str] = [] + for bucket, key in sorted(self.created_objects): + try: + self.api_s3.delete_object(Bucket=bucket, Key=key) + except Exception as exc: # noqa: BLE001 - удалить остальные объекты + failures.append(f"{bucket}/{key}: {sanitize(exc)}") + if failures: + self.reporter.add( + "s3.cleanup", + "FAIL", + "не удалены тестовые объекты: " + "; ".join(failures), + ) + else: + self.reporter.add( + "s3.cleanup", + "PASS", + f"удалены все объекты префикса {self.prefix}", + ) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Проверка PostgreSQL и Selectel S3 для HAN Chat" + ) + parser.add_argument( + "--env-file", + type=Path, + default=Path(__file__).with_name(".env"), + help="путь к .env (по умолчанию .env рядом со скриптом)", + ) + parser.add_argument( + "--mode", + choices=("readonly", "full"), + default="full", + help="full выполняет самочищающиеся DDL/S3 операции", + ) + parser.add_argument( + "--json-report", + type=Path, + help="сохранить обезличенный JSON-отчёт", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + try: + settings = Settings.from_env_file(args.env_file.resolve()) + except (ConfigError, ValueError) as exc: + print(f"Ошибка конфигурации: {sanitize(exc)}", file=sys.stderr) + return 2 + + reporter = Reporter() + test = InfraTest(settings, reporter, args.mode) + print(f"HAN Chat infrastructure test: mode={args.mode}, run={test.run_id[:8]}") + test.run() + if args.json_report: + reporter.write_json(args.json_report) + + passed = sum(item.status == "PASS" for item in reporter.results) + failed = sum(item.status == "FAIL" for item in reporter.results) + skipped = sum(item.status == "SKIP" for item in reporter.results) + print(f"\nИтог: PASS={passed}, FAIL={failed}, SKIP={skipped}") + return reporter.exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/infratest/requirements.txt b/infratest/requirements.txt new file mode 100644 index 0000000..d83e6a0 --- /dev/null +++ b/infratest/requirements.txt @@ -0,0 +1,4 @@ +boto3 +psycopg[binary] +python-dotenv +requests diff --git a/infratest/test_infratest.py b/infratest/test_infratest.py new file mode 100644 index 0000000..662a5d1 --- /dev/null +++ b/infratest/test_infratest.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import patch + +from infratest import ConfigError, Reporter, Settings, as_bool, sanitize + + +def valid_env(ca_path: Path) -> str: + return f""" +HAN_PG_HOST=db.example.test +HAN_PG_PORT=6432 +HAN_PG_DATABASE=han_chat +HAN_PG_SSLMODE=verify-full +HAN_PG_SSLROOTCERT={ca_path} +HAN_PG_PASSWORD_HAN_APP=han-password +HAN_PG_PASSWORD_BITRIX=bitrix-password +HAN_PG_PASSWORD_BITRIX_SYNC=sync-password +HAN_PG_PASSWORD_MESSAGE_SAFETY=safety-password +HAN_PG_PASSWORD_KEYCLOAK=keycloak-password +SELECTEL_S3_ENDPOINT_URL=https://s3.storage.selcloud.ru +SELECTEL_S3_BUCKET_QUARANTINE=quarantine +SELECTEL_S3_BUCKET_ATTACHMENTS=attachments +SELECTEL_S3_BUCKET_DOCUMENTS=documents +SELECTEL_S3_ACCESS_KEY=api-access +SELECTEL_S3_SECRET_KEY=api-secret +SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY=read-access +SELECTEL_S3_QUARANTINE_READ_SECRET_KEY=read-secret +PUBLIC_WEB_URL=https://chat.example.test +INFRATEST_CHECK_CORS=true +INFRATEST_CHECK_MIGRATIONS=false +INFRATEST_TIMEOUT_SECONDS=10 +""".strip() + + +class SettingsTests(unittest.TestCase): + def test_loads_complete_configuration_and_default_roles(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + ca = root / "ca.pem" + ca.write_text("test ca", encoding="utf-8") + env_file = root / ".env" + env_file.write_text(valid_env(ca), encoding="utf-8") + + with patch.dict(os.environ, {}, clear=True): + settings = Settings.from_env_file(env_file) + + self.assertEqual(settings.pg_host, "db.example.test") + self.assertEqual(settings.pg_port, 6432) + self.assertEqual( + [(role.user, role.schema) for role in settings.pg_roles], + [ + ("han_app", "han_app"), + ("bitrix_local_app", "bitrix_local"), + ("bitrix_sync_user", "bitrix_sync"), + ("message_safety_app", "message_safety"), + ("keycloak_user", "keycloak"), + ], + ) + self.assertEqual(settings.buckets["documents"], "documents") + self.assertTrue(settings.check_cors) + self.assertFalse(settings.check_migrations) + + def test_rejects_non_verifying_postgres_tls(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + ca = root / "ca.pem" + ca.write_text("test ca", encoding="utf-8") + env_file = root / ".env" + env_file.write_text( + valid_env(ca).replace( + "HAN_PG_SSLMODE=verify-full", + "HAN_PG_SSLMODE=require", + ), + encoding="utf-8", + ) + with patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(ConfigError, "verify-full"): + Settings.from_env_file(env_file) + + def test_requires_public_url_when_cors_enabled(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + ca = root / "ca.pem" + ca.write_text("test ca", encoding="utf-8") + env_file = root / ".env" + env_file.write_text( + valid_env(ca).replace( + "PUBLIC_WEB_URL=https://chat.example.test", + "PUBLIC_WEB_URL=", + ), + encoding="utf-8", + ) + with patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(ConfigError, "PUBLIC_WEB_URL"): + Settings.from_env_file(env_file) + + +class SafetyTests(unittest.TestCase): + def test_sanitize_redacts_secret_and_url_query(self) -> None: + value = ( + "password=do-not-print " + "url=https://bucket.example/object?X-Amz-Credential=secret&X-Amz-Signature=x" + ) + sanitized = sanitize(value) + self.assertNotIn("do-not-print", sanitized) + self.assertNotIn("X-Amz", sanitized) + self.assertNotIn("Signature", sanitized) + self.assertIn("password=", sanitized) + self.assertIn("https://bucket.example/object", sanitized) + + def test_reporter_exit_code_and_json_are_secret_free(self) -> None: + reporter = Reporter() + with redirect_stdout(io.StringIO()): + reporter.add("ok", "PASS", "url=https://example.test/a?token=secret") + reporter.add("bad", "FAIL", "password=hidden") + + with tempfile.TemporaryDirectory() as directory: + report_path = Path(directory) / "report.json" + reporter.write_json(report_path) + data = json.loads(report_path.read_text(encoding="utf-8")) + + self.assertEqual(reporter.exit_code, 1) + serialized = json.dumps(data) + self.assertNotIn("hidden", serialized) + self.assertNotIn("?token", serialized) + + def test_boolean_parser(self) -> None: + self.assertTrue(as_bool("yes")) + self.assertFalse(as_bool("OFF")) + with self.assertRaises(ConfigError): + as_bool("sometimes") + + +if __name__ == "__main__": + unittest.main()