from __future__ import annotations import argparse import json import os from pathlib import Path import sys import httpx ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from services.shared.security import issue_app_token def _get_json(url: str) -> dict: with httpx.Client(timeout=10, trust_env=False) as client: response = client.get(url, headers=_ops_headers()) response.raise_for_status() return response.json() def _parse_queue_map(raw: str) -> dict[str, str]: payload = json.loads(raw or "{}") if not isinstance(payload, dict): raise ValueError("ASTERISK_QUEUE_MAP_JSON must be a JSON object") result: dict[str, str] = {} for key, value in payload.items(): key_s = str(key).strip() val_s = str(value).strip() if key_s and val_s: result[key_s] = val_s return result def _parse_subjects(raw: str) -> set[str]: return {item.strip() for item in (raw or "").split(",") if item.strip()} def _load_env_file(path: str) -> None: env_path = Path(path).expanduser().resolve() if not env_path.exists() or not env_path.is_file(): return for line in env_path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, value = stripped.split("=", 1) key = key.strip() if key and key not in os.environ: os.environ[key] = value.strip() def _bool_env(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _ops_headers() -> dict[str, str]: if _bool_env("ALLOW_LEGACY_HEADER_AUTH", True): return {"X-User": "admin", "X-Role": "admin"} token = issue_app_token( subject="ops:track9-preflight", username="track9-preflight", role="admin", auth_source="service", provider="track9-script", ttl_seconds=300, ) return {"Authorization": f"Bearer {token}"} def _bridge_auth_mode() -> str: raw = os.getenv("ASTERISK_BRIDGE_AUTH_MODE", "legacy_headers").strip().lower() if raw in {"legacy_headers", "bearer", "bearer_first"}: return raw raise ValueError( "ASTERISK_BRIDGE_AUTH_MODE must be one of: legacy_headers, bearer_first, bearer" ) def _check_sftp() -> tuple[bool, str]: host = os.getenv("ASTERISK_SFTP_HOST", "").strip() username = os.getenv("ASTERISK_SFTP_USERNAME", "").strip() password = os.getenv("ASTERISK_SFTP_PASSWORD", "").strip() base_path = os.getenv("ASTERISK_SFTP_BASE_PATH", "/var/spool/asterisk/monitor").strip() try: port = int(os.getenv("ASTERISK_SFTP_PORT", "22").strip()) except ValueError: return False, "ASTERISK_SFTP_PORT is not a valid integer" if not (host and username and password): return False, "SFTP env is incomplete (host/username/password)" try: import paramiko # type: ignore except ImportError: return False, "paramiko is not installed" transport = paramiko.Transport((host, port)) try: transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) try: sftp.listdir(base_path) finally: sftp.close() except Exception as exc: # noqa: BLE001 return False, f"SFTP check failed: {exc}" finally: try: transport.close() except Exception: # noqa: BLE001 pass return True, "ok" def main() -> int: parser = argparse.ArgumentParser(description="Track 9 preflight checks") parser.add_argument("--base-url", default="http://127.0.0.1:8080") parser.add_argument("--env-file", default=".env.production") parser.add_argument("--check-sftp", action="store_true") parser.add_argument("--require-strict-service-auth", action="store_true") args = parser.parse_args() _load_env_file(args.env_file) failures: list[str] = [] base_url = args.base_url.rstrip("/") health = _get_json(f"{base_url}/proxy/asterisk-bridge/health") status = _get_json(f"{base_url}/proxy/asterisk-bridge/asterisk/status") queues = _get_json(f"{base_url}/proxy/routing/queues") if health.get("status") != "ok": failures.append("asterisk-bridge health is not ok") if status.get("status") not in {"ok", "disabled"}: failures.append("asterisk-bridge status is invalid") try: queue_map = _parse_queue_map(os.getenv("ASTERISK_QUEUE_MAP_JSON", "{}")) except ValueError as exc: failures.append(str(exc)) queue_map = {} try: auth_mode = _bridge_auth_mode() except ValueError as exc: failures.append(str(exc)) auth_mode = "legacy_headers" if auth_mode in {"bearer", "bearer_first"}: app_secret = os.getenv("APP_TOKEN_SECRET", "").strip() if not app_secret: failures.append("APP_TOKEN_SECRET is required for bearer bridge auth modes") elif app_secret == "dev-secret-change-me": failures.append("APP_TOKEN_SECRET must not use default value in bearer bridge auth modes") bridge_subject = os.getenv("ASTERISK_BRIDGE_AUTH_SUBJECT", "svc:asterisk-bridge").strip() voice_subjects = _parse_subjects(os.getenv("VOICE_ADAPTER_TRUSTED_SERVICE_SUBJECTS", "")) recording_subjects = _parse_subjects(os.getenv("RECORDING_IMPORT_TRUSTED_SERVICE_SUBJECTS", "")) if bridge_subject and voice_subjects and bridge_subject not in voice_subjects: failures.append("VOICE_ADAPTER_TRUSTED_SERVICE_SUBJECTS does not include ASTERISK_BRIDGE_AUTH_SUBJECT") if bridge_subject and recording_subjects and bridge_subject not in recording_subjects: failures.append("RECORDING_IMPORT_TRUSTED_SERVICE_SUBJECTS does not include ASTERISK_BRIDGE_AUTH_SUBJECT") if args.require_strict_service_auth: if auth_mode != "bearer": failures.append("Strict mode requires ASTERISK_BRIDGE_AUTH_MODE=bearer") if _bool_env("ALLOW_LEGACY_HEADER_AUTH", True): failures.append("Strict mode requires ALLOW_LEGACY_HEADER_AUTH=0") if _bool_env("ASTERISK_BRIDGE_AUTH_FALLBACK_LEGACY", False): failures.append("Strict mode requires ASTERISK_BRIDGE_AUTH_FALLBACK_LEGACY=0") if _bool_env("RECORDING_IMPORT_ALLOW_ADMIN", True): failures.append("Strict mode requires RECORDING_IMPORT_ALLOW_ADMIN=0") if not voice_subjects: failures.append("Strict mode requires VOICE_ADAPTER_TRUSTED_SERVICE_SUBJECTS") if not recording_subjects: failures.append("Strict mode requires RECORDING_IMPORT_TRUSTED_SERVICE_SUBJECTS") queue_ids = {str(item.get("queue_id")) for item in queues if isinstance(item, dict)} missing_queue_ids = sorted({value for value in queue_map.values() if value not in queue_ids}) if not queue_map: failures.append("ASTERISK_QUEUE_MAP_JSON is empty") if missing_queue_ids: failures.append( f"Queue map contains unknown queue_id values: {', '.join(missing_queue_ids)}" ) if args.check_sftp: sftp_ok, sftp_message = _check_sftp() if not sftp_ok: failures.append(sftp_message) if failures: print("[FAIL] Track 9 preflight failed") for item in failures: print(f"- {item}") return 1 print("[PASS] Track 9 preflight passed") print(f"- bridge_status: {status.get('status')}") print(f"- bridge_auth_mode: {auth_mode}") print(f"- queue_codes_loaded: {len(queue_map)}") print(f"- mapped_queue_ids: {len(set(queue_map.values()))}") if args.check_sftp: print("- sftp: ok") return 0 if __name__ == "__main__": raise SystemExit(main())