from __future__ import annotations import ast import re import subprocess import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] class InfrastructureConfigTests(unittest.TestCase): def test_migrations_pass_one_top_level_statement_per_execute(self) -> None: migration_roots = ( ROOT / "api-backend/alembic/versions", ROOT / "bitrix-local-app/alembic/versions", ROOT / "sms-service/migrations/versions", ) dollar_quoted = re.compile( r"\$\$.*?\$\$|\$(?P[A-Za-z_][A-Za-z0-9_]*)\$.*?\$(?P=tag)\$", re.DOTALL, ) single_quoted = re.compile(r"'(?:''|[^'])*'", re.DOTALL) for migration_root in migration_roots: for migration in migration_root.glob("*.py"): tree = ast.parse( migration.read_text(encoding="utf-8"), filename=str(migration), ) for call in ast.walk(tree): if not ( isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) and call.func.attr == "execute" and call.args and isinstance(call.args[0], ast.Constant) and isinstance(call.args[0].value, str) ): continue sql = dollar_quoted.sub("DOLLAR_QUOTED_BODY", call.args[0].value) sql = single_quoted.sub("STRING_LITERAL", sql) sql = re.sub(r"--[^\n]*|/\*.*?\*/", "", sql, flags=re.DOTALL) statements = [part for part in sql.split(";") if part.strip()] self.assertLessEqual( len(statements), 1, f"{migration}:{call.lineno} passes multiple SQL commands to execute", ) def test_structlog_event_is_not_passed_twice(self) -> None: log_methods = {"debug", "info", "warning", "error", "exception", "critical"} for source_path in (ROOT / "api-backend/app").rglob("*.py"): tree = ast.parse(source_path.read_text(encoding="utf-8"), filename=str(source_path)) for node in ast.walk(tree): if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr in log_methods and node.args and any(keyword.arg == "event" for keyword in node.keywords) ): self.fail(f"duplicate structlog event in {source_path}:{node.lineno}") def test_root_compose_uses_only_infra_fragments(self) -> None: compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") for fragment in ( "infra/compose/application.yml", "nginx/docker-compose.yml", "redis/docker-compose.yml", "observability/docker-compose.yml", "deployment/docker-compose.jobs.yml", ): self.assertIn(fragment, compose) self.assertNotIn("postgres:", compose.lower()) def test_external_dependencies_use_dedicated_egress_network(self) -> None: root = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") self.assertIn("egress:\n name: han-chat-egress", root) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("networks: [backend, observability, egress]", application) self.assertIn("networks: [public, backend, observability, egress]", application) self.assertIn( "KEYCLOAK_YANDEX_CAPTCHA_SERVER_KEY_FILE: " "/run/secrets/keycloak_yandex_captcha_server_key", application, ) self.assertEqual( application.count("IDGTL_SMS_API_KEY_FILE: /run/secrets/idgtl_sms_api_key"), 1, ) jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") self.assertEqual(jobs.count("networks: [backend, egress]"), 4) observability = (ROOT / "observability/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("networks: [observability, egress]", observability) redis = (ROOT / "redis/docker-compose.yml").read_text(encoding="utf-8") self.assertNotIn("egress", redis) def test_only_nginx_fragment_publishes_ports(self) -> None: forbidden = ( ROOT / "infra/compose/application.yml", ROOT / "api-backend/docker-compose.yml", ROOT / "keycloak/docker-compose.yml", ROOT / "redis/docker-compose.yml", ROOT / "observability/docker-compose.yml", ROOT / "deployment/docker-compose.jobs.yml", ) for path in forbidden: self.assertNotIn("\n ports:", path.read_text(encoding="utf-8"), path) nginx = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") self.assertEqual(nginx.count("\n ports:"), 1) self.assertIn('NGINX_HTTP_PORT:-80}:80', nginx) self.assertIn('NGINX_HTTPS_PORT:-443}:443', nginx) def test_production_compose_uses_only_required_digest_images(self) -> None: image_variables = ( "API_BACKEND_IMAGE", "BITRIX_LOCAL_APP_IMAGE", "FRONTEND_STATIC_IMAGE", "KEYCLOAK_IMAGE", "NGINX_IMAGE", "OTEL_COLLECTOR_IMAGE", "OTEL_QUEUE_INIT_IMAGE", "REDIS_IMAGE", "SMS_SERVICE_IMAGE", "TOOLBOX_IMAGE", ) compose_paths = ( ROOT / "infra/compose/application.yml", ROOT / "nginx/docker-compose.yml", ROOT / "redis/docker-compose.yml", ROOT / "observability/docker-compose.yml", ROOT / "deployment/docker-compose.jobs.yml", ) combined = "\n".join(path.read_text(encoding="utf-8") for path in compose_paths) self.assertNotIn("\n build:", combined) for variable in image_variables: self.assertIn(f"${{{variable}:?", combined, variable) example = (ROOT / ".env.example").read_text(encoding="utf-8") for variable in image_variables: match = re.search(rf"^{variable}=(.+)$", example, re.MULTILINE) self.assertIsNotNone(match, variable) self.assertRegex(match.group(1), r"@sha256:[0-9a-f]{64}$") preflight = (ROOT / "deployment/preflight.sh").read_text(encoding="utf-8") self.assertIn("config --images", preflight) self.assertIn("config --services", preflight) self.assertIn("only nginx may publish production host ports", preflight) self.assertIn("still contains the example image/digest", preflight) self.assertIn("socket.getaddrinfo", preflight) self.assertIn('ipaddress.ip_network("10.0.0.0/8")', preflight) self.assertIn("hostname must resolve only to private VPC addresses", preflight) self.assertIn('--runtime-manifest "$MANIFEST"', preflight) for variable in image_variables: self.assertIn(variable, preflight) def test_legacy_compose_tls_is_removed(self) -> None: root = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") nginx = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") combined = f"{root}\n{nginx}" self.assertNotIn("nginx-certs", combined) self.assertNotIn("nginx-acme", combined) self.assertNotIn("\n certbot:", nginx) self.assertIn("/var/lib/han-chat/public-tls:/run/tls:ro", nginx) self.assertIn("/var/lib/han-chat/acme:/var/www/certbot:ro", nginx) example = (ROOT / ".env.example").read_text(encoding="utf-8") self.assertIn("NGINX_TLS_CERTIFICATE=/run/tls/fullchain.pem", example) self.assertIn("NGINX_TLS_CERTIFICATE_KEY=/run/tls/privkey.pem", example) def test_runtime_services_have_compose_hardening(self) -> None: application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") for anchor in ("x-api-runtime: &api-runtime", "x-sms-runtime: &sms-runtime"): start = application.index(anchor) end = application.index("\n\n", start) runtime = application[start:end] for setting in ( "read_only: true", 'cap_drop: ["ALL"]', 'security_opt: ["no-new-privileges:true"]', "/tmp:size=", "pids_limit:", "mem_limit:", "cpus:", ): self.assertIn(setting, runtime, f"{anchor}: {setting}") bitrix = application[ application.index("\n bitrix-local-app:") : application.index("\nsecrets:") ] for setting in ( "read_only: true", 'cap_drop: ["ALL"]', 'security_opt: ["no-new-privileges:true"]', "/tmp:size=", "pids_limit:", "mem_limit:", "cpus:", ): self.assertIn(setting, bitrix) keycloak = application[ application.index("\n keycloak:") : application.index("\n sms-service:") ] self.assertNotIn("\n read_only:", keycloak) self.assertIn('cap_drop: ["ALL"]', keycloak) self.assertIn("pids_limit:", keycloak) self.assertIn("read_only is intentionally omitted", keycloak) jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") self.assertEqual(jobs.count("<<: *python-job-runtime"), 4) for setting in ("read_only: true", 'cap_drop: ["ALL"]', "/tmp:size="): self.assertIn(setting, jobs) def test_vm_and_nginx_security_defaults(self) -> None: setup = (ROOT / "deployment/scripts/setup-vm.sh").read_text(encoding="utf-8") env_example = (ROOT / ".env.example").read_text(encoding="utf-8") compose = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") ssl_renew = ( ROOT / "deployment/scripts/ssl-renew.sh" ).read_text(encoding="utf-8") self.assertIn('LOCK_ACCOUNT_PASSWORDS="${LOCK_ACCOUNT_PASSWORDS:-true}"', setup) self.assertIn('passwd --lock root', setup) self.assertIn('passwd --lock "$DEPLOY_USER"', setup) self.assertIn("X11Forwarding no", setup) self.assertIn("PasswordAuthentication no", setup) self.assertIn("NGINX_HSTS_MAX_AGE=31536000", env_example) self.assertIn("NGINX_HSTS_MAX_AGE:-31536000", compose) self.assertIn("compose kill --signal HUP nginx", ssl_renew) executable_ssl_renew = "\n".join( line for line in ssl_renew.splitlines() if not line.lstrip().startswith("#") ) self.assertNotIn("nginx -s reload", executable_ssl_renew) def test_nginx_internal_denies_precede_spa(self) -> None: site = (ROOT / "nginx/templates/site-tls.conf.template").read_text(encoding="utf-8") compose = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") config = (ROOT / "nginx/nginx.conf.template").read_text(encoding="utf-8") entrypoint = (ROOT / "nginx/scripts/entrypoint.sh").read_text(encoding="utf-8") proxy_common = (ROOT / "nginx/snippets/proxy-common.conf").read_text(encoding="utf-8") proxy_keycloak = (ROOT / "nginx/snippets/proxy-keycloak.conf").read_text( encoding="utf-8" ) internal = site.index("location ^~ /internal/") api = site.index("location ^~ /api/") frontend = site.index("include /etc/nginx/generated/frontend-location.conf") self.assertLess(internal, api) self.assertLess(api, frontend) self.assertIn("location = /api/v1/realtime", site) self.assertNotIn("message-safety:", site) self.assertIn("uid=0,gid=0", compose) self.assertIn('cap_add: ["CHOWN", "NET_BIND_SERVICE", "SETUID", "SETGID"]', compose) self.assertTrue(config.startswith("user nginx;\n")) self.assertIn("chown root:nginx \"$cache_root\"", entrypoint) self.assertIn("chmod 2770 \"$cache_public\"", entrypoint) self.assertLess(entrypoint.index("cache_root="), entrypoint.index("nginx -t")) self.assertIn("proxy_read_timeout 30s;", config) self.assertNotIn("proxy_read_timeout", proxy_common) self.assertNotIn("proxy_send_timeout", proxy_common) self.assertIn("include /etc/nginx/snippets/proxy-keycloak.conf;", site) self.assertIn("location = /auth/callback", site) self.assertIn("location ^~ /auth/resources/", site) self.assertIn("location ^~ /auth/realms/", site) self.assertIn( "location = /auth/realms/han-chat/protocol/openid-connect/auth", site ) self.assertIn( "location = /auth/realms/han-chat/login-actions/authenticate", site ) captcha_csp = ( ROOT / "nginx/snippets/proxy-keycloak-captcha-csp.conf" ).read_text(encoding="utf-8") self.assertIn("proxy_hide_header Content-Security-Policy", captcha_csp) self.assertIn("smartcaptcha.cloud.yandex.ru", captcha_csp) self.assertIn("yastatic.net", captcha_csp) for directive in ("default-src 'self'", "base-uri 'self'", "form-action 'self'"): self.assertIn(directive, captcha_csp) self.assertNotIn("browserSecurityHeaders", ( ROOT / "keycloak/realm/han-chat-realm.json" ).read_text(encoding="utf-8")) self.assertIn("location = /callbacks/idgtl/sms", site) self.assertIn("allow 185.203.96.7;", site) self.assertIn("proxy_pass http://sms_service_upstream;", site) self.assertIn("upstream sms_service_upstream", config) self.assertNotIn("security-headers.conf", proxy_keycloak) self.assertNotIn("X-Frame-Options", proxy_keycloak) def test_redis_persistence_acl_and_no_host_port(self) -> None: config = (ROOT / "redis/redis.conf").read_text(encoding="utf-8") acl = (ROOT / "redis/users.acl.template").read_text(encoding="utf-8") self.assertIn("appendonly yes", config) self.assertIn("appendfsync everysec", config) self.assertIn("save 900 1", config) self.assertIn("user default off", acl) self.assertIn("~han:api:*", acl) self.assertNotIn("message_safety", acl) self.assertNotIn("REDIS_SAFETY_PASSWORD", acl) def test_otel_has_redaction_and_persistent_queue(self) -> None: config = (ROOT / "observability/otel-collector.yaml").read_text(encoding="utf-8") for forbidden_attribute in ( "http.request.header.authorization", "http.request.header.cookie", "url.query", "url.full", "db.statement", "db.query.text", "messaging.message.body", "aws.s3.key", ): self.assertIn(forbidden_attribute, config) self.assertIn("storage: file_storage", config) self.assertIn("retry_on_failure:", config) self.assertIn('insecure: "${env:OTEL_REMOTE_TLS_INSECURE}"', config) self.assertIn( "authorization: ${file:/run/secrets/otel_remote_auth_header}", config ) self.assertNotIn("OTEL_REMOTE_AUTH_HEADER", config) self.assertIn("tail_sampling:", config) self.assertNotIn("probabilistic_sampler:", config) self.assertIn('targets: ["sms-service:8080"]', config) self.assertIn('targets: ["sms-worker:9464"]', config) self.assertIn('targets: ["keycloak:9000"]', config) self.assertIn("metric_relabel_configs:", config) compose = (ROOT / "observability/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("OTEL_REMOTE_TLS_INSECURE: ${OTEL_REMOTE_TLS_INSECURE:-false}", compose) self.assertIn("networks: [observability, egress]", compose) self.assertNotIn("networks: [observability, backend, egress]", compose) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME_API:-api-backend}", application) self.assertIn("OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME_SMS_API:-sms-service}", application) self.assertIn("OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME_SMS_WORKER:-sms-worker}", application) self.assertIn('expose: ["9464"]', application) for service_main in ( ROOT / "api-backend/app/main.py", ROOT / "sms-service/app/main.py", ): source = service_main.read_text(encoding="utf-8") self.assertNotIn("route=request.url.path", source) self.assertIn('getattr(request.scope.get("route"), "path"', source) for telemetry_path in ( ROOT / "api-backend/app/telemetry.py", ROOT / "sms-service/app/telemetry.py", ROOT / "bitrix-local-app/app/telemetry.py", ): telemetry = telemetry_path.read_text(encoding="utf-8") self.assertIn("OTLPLogExporter", telemetry) self.assertIn("BatchLogRecordProcessor", telemetry) self.assertIn("LoggingHandler", telemetry) worker = (ROOT / "sms-service/app/worker.py").read_text(encoding="utf-8") for span_name in ("sms.claim", "sms.process", "sms.provider", "sms.save_result"): self.assertIn(f'"{span_name}"', worker) bitrix = (ROOT / "bitrix-local-app/app/main.py").read_text(encoding="utf-8") self.assertIn('EXPECTED_BITRIX_DB_REVISION = "0002_inbox_trace_context"', bitrix) self.assertIn("links=origin_links(row.traceparent)", bitrix) bitrix_migration = ( ROOT / "bitrix-local-app/alembic/versions/0002_inbox_trace_context.py" ).read_text(encoding="utf-8") self.assertIn('down_revision: str | None = "0001_bitrix_local"', bitrix_migration) def test_host_otel_collector_is_hardened_and_allowlisted(self) -> None: config = ( ROOT / "deployment/observability/otel-host-collector.yaml" ).read_text(encoding="utf-8") unit = (ROOT / "deployment/han-host-otel-collector@.service").read_text( encoding="utf-8" ) setup = (ROOT / "deployment/scripts/setup-vm.sh").read_text(encoding="utf-8") for required in ( "filelog/docker", "journald/host", "filter/allowlist", "storage: file_storage", 'layout: "2006-01-02T15:04:05.000000000Z07:00"', 'set(body, "nginx.access")', 'set(body, "keycloak.event")', '{key: uri, action: delete}', 'resource.attributes["service.name"] != "nginx"', 'resource.attributes["service.name"] != "keycloak"', 'resource.attributes["service.name"] != "redis"', ): self.assertIn(required, config) self.assertNotIn("/var/run/docker.sock", config + unit) self.assertNotIn('from: attributes["service.name"]', config) self.assertIn("CapabilityBoundingSet=", unit) self.assertIn("NoNewPrivileges=yes", unit) self.assertIn("ReadOnlyPaths=/var/lib/docker/containers", unit) self.assertIn("ReadWritePaths=/var/lib/han-otel/host-collector", unit) self.assertIn("OTEL_HOST_COLLECTOR_SHA256", setup) self.assertIn("sha256sum -c -", setup) self.assertIn("journalctl --flush", setup) self.assertNotIn("systemctl start han-host-otel-collector", setup) verify = ( ROOT / "deployment/scripts/verify-observability.sh" ).read_text(encoding="utf-8") self.assertIn("TELEMETRYGEN_IMAGE", verify) self.assertNotIn("telemetrygen:latest", verify) def test_settings_cli_and_workers_are_deployable(self) -> None: cli = ROOT / "api-backend/app/cli" self.assertTrue((cli / "__init__.py").is_file()) self.assertTrue((cli / "seed_settings.py").is_file()) self.assertTrue((cli / "validate_settings.py").is_file()) jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") self.assertIn("python -m app.cli.seed_settings", jobs) self.assertIn("python -m app.cli.validate_settings", jobs) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("\n frontend-static:", application) self.assertIn("${FRONTEND_STATIC_IMAGE:?", application) self.assertIn("frontend-static:/output", application) for service, command in ( ("sms-worker:", "han-sms-worker"), ("delivery-worker:", "han-delivery-worker"), ("safety-recovery-worker:", "han-safety-worker"), ("cleanup-worker:", "han-cleanup-worker"), ): self.assertIn(f"\n {service}", application) self.assertIn(f'command: ["{command}"]', application) self.assertNotIn("\n ports:", application) nginx = (ROOT / "nginx/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("frontend-static: {condition: service_completed_successfully}", nginx) def test_smoke_script_does_not_source_env_as_shell(self) -> None: smoke = (ROOT / "deployment/scripts/smoke.sh").read_text(encoding="utf-8") self.assertNotIn('. "./$CONFIG_FILE"', smoke) for variable in ("PUBLIC_HOST", "PUBLIC_WEB_URL", "KEYCLOAK_REALM"): self.assertIn(f"{variable}=$(env_value {variable})", smoke) def test_expo_public_environment_uses_static_property_access(self) -> None: config = (ROOT / "frontend-test-site/src/config.ts").read_text(encoding="utf-8") auth = (ROOT / "frontend-test-site/src/auth.ts").read_text(encoding="utf-8") dockerfile = (ROOT / "frontend-test-site/Dockerfile").read_text(encoding="utf-8") realm = (ROOT / "keycloak/realm/han-chat-realm.json").read_text(encoding="utf-8") application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") callback = (ROOT / "frontend-test-site/app/auth/callback.tsx").read_text( encoding="utf-8" ) self.assertNotIn("process.env[name]", config) self.assertIn("process.env.EXPO_PUBLIC_API_BASE_URL", config) self.assertIn("process.env.EXPO_PUBLIC_AUTH_BASE_URL", config) self.assertIn('path: "auth/callback"', auth) self.assertIn('window.location.assign(url)', auth) self.assertIn('test -n "$EXPO_PUBLIC_AUTH_BASE_URL"', dockerfile) self.assertIn("${PUBLIC_WEB_URL}/auth/callback", realm) self.assertNotIn("chat.han0107.ru", realm) self.assertIn("PUBLIC_WEB_URL: ${PUBLIC_WEB_URL:?", application) self.assertIn("completionStarted.current", callback) self.assertTrue((ROOT / "frontend-test-site/app/auth/callback.tsx").is_file()) def test_alembic_escapes_percent_encoded_dsn_options(self) -> None: for relative_path in ( "api-backend/alembic/env.py", "bitrix-local-app/alembic/env.py", "sms-service/migrations/env.py", ): env_script = (ROOT / relative_path).read_text(encoding="utf-8") self.assertIn('.replace("%", "%%")', env_script, relative_path) self.assertIn("create_postgres_engine", env_script, relative_path) sms_db = (ROOT / "sms-service/app/db.py").read_text(encoding="utf-8") self.assertNotIn("server_settings", sms_db) def test_contact_sync_qualifies_digest_and_deduplicates_initial_map(self) -> None: initial = ( ROOT / "api-backend/alembic/versions/0001_initial_han_app.py" ).read_text(encoding="utf-8") fix = ( ROOT / "api-backend/alembic/versions/0002_qualify_pgcrypto_digest.py" ).read_text(encoding="utf-8") dedup_fix = ( ROOT / "api-backend/alembic/versions/0010_contact_map_dedup.py" ).read_text(encoding="utf-8") main = (ROOT / "api-backend/app/main.py").read_text(encoding="utf-8") self.assertIn("public.digest(", initial) self.assertIn("public.digest(", fix) self.assertIn('down_revision: str | None = "0001_initial"', fix) self.assertIn("IF v_task_type = 'contact.map_or_create' THEN", dedup_fix) self.assertIn("v_dedup := v_task_type || ':' || v_entity_id::text;", dedup_fix) self.assertIn( "NEW.phone_number IS NOT DISTINCT FROM OLD.phone_number", dedup_fix, ) self.assertIn( "NEW.record_status IS NOT DISTINCT FROM OLD.record_status", dedup_fix, ) self.assertIn('down_revision: str | None = "0009_chat_message_max"', dedup_fix) self.assertIn('EXPECTED_API_DB_REVISION = "0013_delivery_trace_context"', main) def test_consent_audit_migration_supports_existing_and_fresh_databases(self) -> None: migration = ( ROOT / "api-backend/alembic/versions/0003_consent_device_audit_context.py" ).read_text(encoding="utf-8") self.assertIn("ADD COLUMN IF NOT EXISTS device_json", migration) self.assertIn("DROP COLUMN IF EXISTS ip", migration) self.assertIn('down_revision: str | None = "0002_pgcrypto_digest"', migration) def test_keycloak_management_health_and_bridge_environment(self) -> None: standalone = (ROOT / "keycloak/docker-compose.yml").read_text(encoding="utf-8") self.assertIn("GET /auth/health/ready", standalone) dockerfile = (ROOT / "keycloak/Dockerfile").read_text(encoding="utf-8") self.assertLess( dockerfile.index("COPY realm ./realm"), dockerfile.index("mvn -B -ntp clean verify"), ) application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") self.assertIn("GET /auth/health/ready", application) for variable in ( "KC_DB_SCHEMA", "KEYCLOAK_OTP_MOCK_ENABLED", "KEYCLOAK_YANDEX_CAPTCHA_ENABLED", "KEYCLOAK_OTP_SETTINGS_MAX_STALE_SEC", "KEYCLOAK_SETTINGS_BRIDGE_URL", "KEYCLOAK_SMS_SERVICE_URL", ): self.assertIn(f" {variable}:", application) for variable in ( "KEYCLOAK_OTP_MOCK_CODE", "KEYCLOAK_YANDEX_CAPTCHA_SERVER_KEY", "KEYCLOAK_OTP_HMAC_KEY", "KEYCLOAK_SETTINGS_BRIDGE_TOKEN", "KEYCLOAK_SMS_SERVICE_TOKEN", ): self.assertIn(f" {variable}_FILE:", application) self.assertIn( " KEYCLOAK_YANDEX_CAPTCHA_CLIENT_KEY: " "${KEYCLOAK_YANDEX_CAPTCHA_CLIENT_KEY:-}", application, ) def test_compose_secrets_do_not_enter_config_environment(self) -> None: compose_paths = ( ROOT / "infra/compose/application.yml", ROOT / "redis/docker-compose.yml", ROOT / "observability/docker-compose.yml", ROOT / "deployment/docker-compose.jobs.yml", ) combined = "\n".join(path.read_text(encoding="utf-8") for path in compose_paths) self.assertNotIn("env_file:", combined) for path in compose_paths: self.assertIn( "core: {soft: 0, hard: 0}", path.read_text(encoding="utf-8"), path, ) for variable in ( "DATABASE_URL", "REDIS_URL", "SMS_SERVICE_TOKEN", "KEYCLOAK_DB_PASSWORD", "OTEL_REMOTE_AUTH_HEADER", ): self.assertNotIn(f"${{{variable}", combined) self.assertIn( "${HAN_SECRETS_DIR:-/run/han-chat/secrets}/DATABASE_URL", combined ) self.assertNotIn("entrypoint: []", combined) for service in ( "api-backend", "bitrix-local-app", "keycloak", "nginx", "redis", "sms-service", ): dockerfile = (ROOT / service / "Dockerfile").read_text(encoding="utf-8") self.assertIn("sed -i 's/\\r$//'", dockerfile, service) self.assertIn("/bin/sh -n", dockerfile, service) frontend = (ROOT / "frontend-test-site/Dockerfile").read_text(encoding="utf-8") self.assertIn('ENTRYPOINT ["/bin/sh", "-ec"]', frontend) self.assertNotIn("entrypoint.sh", frontend) def test_legacy_stubs_are_absent_and_safety_is_remote_tls(self) -> None: for stub in ("message-safety", "bitrix-sync"): self.assertFalse((ROOT / stub / "pyproject.toml").exists()) self.assertFalse((ROOT / stub / "Dockerfile").exists()) self.assertFalse((ROOT / stub / "app/main.py").exists()) example = (ROOT / ".env.example").read_text(encoding="utf-8") application = (ROOT / "infra/compose/application.yml").read_text(encoding="utf-8") jobs = (ROOT / "deployment/docker-compose.jobs.yml").read_text(encoding="utf-8") nginx = (ROOT / "nginx/nginx.conf.template").read_text(encoding="utf-8") for config in (example, application, jobs): self.assertIn("https://processing.internal:8443", config) self.assertIn("/internal/safety/v2", config) self.assertIn( "MESSAGE_SAFETY_EXTRA_HOST=processing.internal=192.168.0.4", example ) self.assertIn("${MESSAGE_SAFETY_EXTRA_HOST:?", application) self.assertIn("MESSAGE_SAFETY_CA_HOST_PATH", example) self.assertNotIn("message-safety:", application) self.assertNotIn("bitrix-sync:", application) self.assertNotIn("bitrix_sync_upstream", nginx) def test_env_validator_accepts_materialized_example(self) -> None: example = (ROOT / ".env.example").read_text(encoding="utf-8") self.assertNotIn("options=-csearch_path", example) self.assertNotIn("currentSchema=", example) self.assertIn("KEYCLOAK_DB_SCHEMA=keycloak", example) self.assertIn("HAN_PG_PORT=5433", example) self.assertIn("SECRETS_SOURCE=file", example) self.assertNotIn("CURSOR_HMAC_SECRET=", example) self.assertNotIn("BITRIX_TOKEN_ENCRYPTION_KEY=", example) materialized = example materialized = materialized.replace( "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false", "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true", ) with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / ".env" env_file.write_text(materialized, encoding="utf-8") result = subprocess.run( [sys.executable, str(ROOT / "scripts/validate-env"), str(env_file)], text=True, capture_output=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) def test_env_validator_keeps_captcha_server_key_out_of_config(self) -> None: example = (ROOT / ".env.example").read_text(encoding="utf-8") materialized = example.replace( "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=false", "KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true", ) enabled_without_keys = materialized.replace( "KEYCLOAK_YANDEX_CAPTCHA_ENABLED=false", "KEYCLOAK_YANDEX_CAPTCHA_ENABLED=true", ) with tempfile.TemporaryDirectory() as directory: env_file = Path(directory) / ".env" env_file.write_text(enabled_without_keys, encoding="utf-8") result = subprocess.run( [sys.executable, str(ROOT / "scripts/validate-env"), str(env_file)], text=True, capture_output=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertNotIn("KEYCLOAK_YANDEX_CAPTCHA_SERVER_KEY=", example) if __name__ == "__main__": unittest.main()