"""STRATEGY_ID and external-token presence helpers. Phase1 (product plan): missing STRATEGY_ID → process must fail to start (exit non-zero). Values are never logged or returned in responses. """ from __future__ import annotations import os import sys from typing import Optional # Community alias order (level C); we only check presence, never echo values. _TOKEN_ENV_KEYS = ( "XC_TOKEN_FILE", "XC_TOKEN", "MODELHUB_XC_TOKEN", "EXTERNAL_SERVICE_TOKEN", ) def get_strategy_id() -> Optional[str]: """Return STRATEGY_ID if set and non-empty after strip; else None.""" raw = os.environ.get("STRATEGY_ID") if raw is None: return None value = raw.strip() return value if value else None def require_strategy_id() -> str: """Return STRATEGY_ID or exit the process with a non-zero status. Does not print the value. Safe for startup path. """ sid = get_strategy_id() if sid is None: print( "ERROR: STRATEGY_ID is required and must be non-empty. " "Refusing to start (Phase1 contract).", file=sys.stderr, ) sys.exit(1) return sid def strategy_id_present() -> bool: return get_strategy_id() is not None def _token_from_file(path: str) -> bool: """True if path exists and has non-empty content (value never read into logs).""" try: if not path or not os.path.isfile(path): return False with open(path, "r", encoding="utf-8") as fh: content = fh.read().strip() return bool(content) except OSError: return False def external_service_token_present() -> bool: """Whether any known token env / file alias is present (boolean only).""" for key in _TOKEN_ENV_KEYS: raw = os.environ.get(key) if raw is None: continue value = raw.strip() if not value: continue if key == "XC_TOKEN_FILE": if _token_from_file(value): return True continue # Reject obvious placeholders (community practice, level C) lowered = value.lower() if lowered in ("tmp", "placeholder", "changeme", "xxx"): continue return True return False