#!/usr/bin/env python3 """Fail-fast validation for the single HAN Chat deployment environment.""" from __future__ import annotations import re import sys from pathlib import Path from urllib.parse import urlparse REQUIRED = { "APP_ENV", "RELEASE_VERSION", "HAN_PG_HOST", "DATABASE_URL", "BITRIX_DATABASE_URL", "BITRIX_SYNC_DATABASE_URL", "MESSAGE_SAFETY_DATABASE_URL", "SMS_DATABASE_URL", "KEYCLOAK_DB_URL", "PUBLIC_HOST", "PUBLIC_WEB_URL", "PUBLIC_API_URL", "PUBLIC_AUTH_URL", "KEYCLOAK_PUBLIC_URL", "KEYCLOAK_INTERNAL_URL", "REDIS_URL", "REDIS_REALTIME_URL", "MESSAGE_SAFETY_REDIS_URL", "MESSAGE_SAFETY_SERVICE_TOKEN", "BITRIX_LOCAL_APP_INTERNAL_TOKEN", "BITRIX_INTERNAL_API_TOKEN", "BITRIX_API_FORWARD_TOKEN", "BITRIX_API_INBOX_TOKEN", "BITRIX_SYNC_SERVICE_TOKEN", "KEYCLOAK_SETTINGS_BRIDGE_TOKEN", "KEYCLOAK_OTP_HMAC_KEY", "KEYCLOAK_SMS_SERVICE_URL", "KEYCLOAK_SMS_SERVICE_TOKEN", "SMS_SERVICE_TOKEN", "IDGTL_SMS_BASE_URL", "IDGTL_SMS_API_KEY", "IDGTL_SMS_CALLBACK_PUBLIC_URL", "IDGTL_SMS_CALLBACK_USERNAME", "IDGTL_SMS_CALLBACK_PASSWORD", "KEYCLOAK_ADMIN", "KEYCLOAK_ADMIN_PASSWORD", "CURSOR_HMAC_SECRET", "BITRIX_TOKEN_ENCRYPTION_KEY", "SELECTEL_S3_ENDPOINT_URL", "SELECTEL_S3_ACCESS_KEY", "SELECTEL_S3_SECRET_KEY", "SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY", "SELECTEL_S3_QUARANTINE_READ_SECRET_KEY", "REDIS_API_PASSWORD", "REDIS_SAFETY_PASSWORD", "REDIS_HEALTH_PASSWORD", } SECRET_KEYS = { key for key in REQUIRED if any(word in key for word in ("TOKEN", "PASSWORD", "SECRET_KEY", "ACCESS_KEY")) } | { "BITRIX_CLIENT_SECRET", "BITRIX_APPLICATION_TOKEN", "IDGTL_SMS_CALLBACK_USERNAME", "IDGTL_SMS_CALLBACK_PASSWORD", } PLACEHOLDER = re.compile(r"(change-me|example\.(com|ru|invalid)|<[^>]+>)", re.I) def load(path: Path) -> tuple[dict[str, str], list[str]]: values: dict[str, str] = {} errors: list[str] = [] for number, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): line = raw.strip() if not line or line.startswith("#"): continue if "=" not in line: errors.append(f"строка {number}: ожидается KEY=VALUE") continue key, value = line.split("=", 1) key = key.strip() if key in values: errors.append(f"строка {number}: повтор ключа {key}") values[key] = value.strip() return values, errors def main() -> int: path = Path(sys.argv[1] if len(sys.argv) > 1 else ".env") if not path.is_file(): print(f"ERROR: файл не найден: {path}", file=sys.stderr) return 2 env, errors = load(path) for key in sorted(REQUIRED): if not env.get(key): errors.append(f"{key}: обязательное значение отсутствует") for key in sorted(SECRET_KEYS): value = env.get(key, "") if value and (len(value) < 16 or PLACEHOLDER.search(value)): errors.append(f"{key}: секрет должен быть непустым, уникальным и длиной >=16") for key in ("SMS_SERVICE_TOKEN", "KEYCLOAK_SMS_SERVICE_TOKEN"): if env.get(key) and len(env[key]) < 32: errors.append(f"{key}: service token должен иметь длину >=32") production = env.get("APP_ENV") in {"production-like", "production"} if production and env.get("FRONTEND_DEV_PROXY_ENABLED", "").lower() != "false": errors.append("FRONTEND_DEV_PROXY_ENABLED: production-like/production требует false") if production and env.get("NGINX_TLS_ENABLED", "").lower() != "true": errors.append("NGINX_TLS_ENABLED: production-like/production требует true") for key in ( "PUBLIC_WEB_URL", "PUBLIC_API_URL", "PUBLIC_AUTH_URL", "KEYCLOAK_PUBLIC_URL", "IDGTL_SMS_BASE_URL", "IDGTL_SMS_CALLBACK_PUBLIC_URL", ): if env.get(key) and urlparse(env[key]).scheme != "https": errors.append(f"{key}: публичный URL должен использовать https") for key in ( "KEYCLOAK_INTERNAL_URL", "KEYCLOAK_SMS_SERVICE_URL", "MESSAGE_SAFETY_URL", "BITRIX_LOCAL_APP_BASE_URL", ): parsed = urlparse(env.get(key, "")) if parsed.scheme != "http" or "." in (parsed.hostname or ""): errors.append(f"{key}: ожидается http URL с Docker DNS service name") if not env.get("KEYCLOAK_INTERNAL_URL", "").rstrip("/").endswith("/auth"): errors.append("KEYCLOAK_INTERNAL_URL: внутренний URL должен заканчиваться на /auth") redis_contract = ( ("REDIS_URL", "api_backend", "REDIS_API_PASSWORD", "/0"), ("REDIS_REALTIME_URL", "api_backend", "REDIS_API_PASSWORD", "/1"), ("MESSAGE_SAFETY_REDIS_URL", "message_safety", "REDIS_SAFETY_PASSWORD", "/2"), ) for key, username, password_key, database in redis_contract: parsed = urlparse(env.get(key, "")) if ( parsed.scheme != "redis" or parsed.hostname != "redis" or parsed.username != username or parsed.password != env.get(password_key) or parsed.path != database ): errors.append(f"{key}: ACL user/password/host/DB не согласованы с {password_key}") for key in ( "DATABASE_URL", "BITRIX_DATABASE_URL", "BITRIX_SYNC_DATABASE_URL", "MESSAGE_SAFETY_DATABASE_URL", "SMS_DATABASE_URL", "KEYCLOAK_DB_URL", ): value = env.get(key, "") if "sslmode=verify-full" not in value or "sslrootcert=" not in value: errors.append(f"{key}: требуется sslmode=verify-full и sslrootcert") if re.search(r"(?:[?&](?:options|currentSchema)=)", value, re.I): errors.append( f"{key}: options/currentSchema запрещены через PgBouncer; " "используйте database-level search_path роли" ) pairs = ( ("BITRIX_LOCAL_APP_INTERNAL_TOKEN", "BITRIX_INTERNAL_API_TOKEN"), ("BITRIX_API_FORWARD_TOKEN", "BITRIX_API_INBOX_TOKEN"), ("KEYCLOAK_SMS_SERVICE_TOKEN", "SMS_SERVICE_TOKEN"), ) for left, right in pairs: if env.get(left) != env.get(right): errors.append(f"{left} должен совпадать с {right}") try: poll = int(env.get("MESSAGE_SAFETY_TASK_POLL_MAX_SEC", "0")) nginx = int(env.get("NGINX_MESSAGE_READ_TIMEOUT_SEC", "0")) ttl = int(env.get("MESSAGE_SAFETY_TASK_TTL_SEC", "0")) if nginx < poll + 30: errors.append("NGINX_MESSAGE_READ_TIMEOUT_SEC должен быть >= poll max + 30") if ttl <= poll + 60: errors.append("MESSAGE_SAFETY_TASK_TTL_SEC должен превышать poll max + recovery margin") except ValueError: errors.append("Safety/NGINX timeout values должны быть целыми числами") if production and env.get("KEYCLOAK_OTP_MOCK_ENABLED", "").lower() == "true": if env.get("KEYCLOAK_OTP_MOCK_RISK_ACCEPTED", "").lower() != "true": errors.append("KEYCLOAK_OTP_MOCK_RISK_ACCEPTED=true обязателен для mock OTP") real_sms = env.get("KEYCLOAK_OTP_MOCK_ENABLED", "").lower() == "false" if production and real_sms and ( env.get("IDGTL_SMS_BASE_URL", "").rstrip("/") != "https://direct.i-dgtl.ru" ): errors.append("IDGTL_SMS_BASE_URL: production contract требует https://direct.i-dgtl.ru") expected_callback = f"https://{env.get('PUBLIC_HOST', '')}/callbacks/idgtl/sms" if production and env.get("IDGTL_SMS_CALLBACK_PUBLIC_URL") != expected_callback: errors.append( "IDGTL_SMS_CALLBACK_PUBLIC_URL должен совпадать с публичным host и callback path" ) if env.get("HAN_PG_HOST") in {"localhost", "127.0.0.1", "postgres", "db"}: errors.append("HAN_PG_HOST: PostgreSQL должен быть внешним managed endpoint") if errors: for error in errors: print(f"ERROR: {error}", file=sys.stderr) return 1 print(f"OK: {path} прошёл проверку ({len(env)} переменных)") return 0 if __name__ == "__main__": raise SystemExit(main())