forked from luopingyi/xc_agent_platform_demo
Phase1 skeleton: replace demo main.py with src.server contract
- Overlay ModelHub XC Phase1 (health/status, STRATEGY_ID gate, dry-run) - Dockerfile CMD -> python -m src.server; FROM stays platform TCR base - Remove demo main.py and requirements.txt (stdlib only) - Add tests/, scripts/, docs/, Makefile, config/examples (no secrets)
This commit is contained in:
81
src/strategy.py
Normal file
81
src/strategy.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user