#!/usr/bin/env python3 """Ask the Sellable control plane whether one customer-program job is due.""" from __future__ import annotations import json import os import re import stat from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import Request, urlopen JOBS = { "campaign_daily_review", "campaign_daily_results", "campaign_weekly_review", } SAFE_REASON = re.compile(r"^[a-z0-9_:-]{1,120}$") SAFE_ID = re.compile(r"^[A-Za-z0-9._:/-]{1,1000}$") TOKEN = re.compile(r"^sat_[^_\s]+_\S{1,4070}$") def _emit(value: dict) -> int: print(json.dumps(value, separators=(",", ":"), sort_keys=True)) return 0 def _blocked(reason: str) -> int: safe = reason if SAFE_REASON.fullmatch(reason) else "invalid_runtime_state" return _emit({"wakeAgent": False, "reason": f"blocked:{safe}"}) def _read_json_file(path: Path, *, maximum: int) -> dict: descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) try: metadata = os.fstat(descriptor) if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > maximum: raise ValueError("file_rejected") with os.fdopen(descriptor, "r", encoding="utf-8") as stream: descriptor = -1 value = json.load(stream) finally: if descriptor >= 0: os.close(descriptor) if not isinstance(value, dict): raise ValueError("json_rejected") return value def _config_path() -> Path: explicit = os.environ.get("SELLABLE_CUSTOMER_PROGRAM_CONFIG", "").strip() if explicit: return Path(explicit) profile_root = Path(__file__).absolute().parent.parent return profile_root / ".sellable-agent" / "customer-program" / "review-schedule.json" def _job_key(config: dict) -> str: filename_job = ( Path(__file__).stem .removeprefix("customer-program-due-") .replace("-", "_") ) return ( os.environ.get("SELLABLE_CUSTOMER_PROGRAM_JOB_KEY", "") or (filename_job if filename_job in JOBS else "") or str(config.get("jobKey") or "") ).strip() def _validate_origin(value: object, allow_test_http: bool) -> str: if not isinstance(value, str): raise ValueError("origin_rejected") parsed = urlsplit(value) test_http = ( allow_test_http and parsed.scheme == "http" and parsed.hostname in {"127.0.0.1", "localhost", "host.docker.internal"} ) if ( (parsed.scheme != "https" and not test_http) or not parsed.hostname or parsed.username or parsed.password or parsed.path not in {"", "/"} or parsed.query or parsed.fragment ): raise ValueError("origin_rejected") return value.rstrip("/") def _safe_response(value: object, expected_job: str) -> dict: if not isinstance(value, dict) or type(value.get("wakeAgent")) is not bool: raise ValueError("response_rejected") if value["wakeAgent"] is False: reason = value.get("reason") if not isinstance(reason, str) or not SAFE_REASON.fullmatch(reason): raise ValueError("response_rejected") return {"wakeAgent": False, "reason": reason} job_key = value.get("jobKey") period_key = value.get("periodKey") lease_owner = value.get("leaseOwner") mode = value.get("mode") effect_id = value.get("effectId") fence = value.get("fence") attempt = value.get("attempt") if ( job_key != expected_job or not isinstance(period_key, str) or not SAFE_ID.fullmatch(period_key) or not isinstance(lease_owner, str) or not SAFE_ID.fullmatch(lease_owner) or mode not in {"execute", "resume", "reconcile"} or not isinstance(effect_id, str) or not SAFE_ID.fullmatch(effect_id) or type(fence) is not int or fence < 1 or type(attempt) is not int or attempt < 0 or attempt > 3 ): raise ValueError("response_rejected") return { "wakeAgent": True, "jobKey": job_key, "periodKey": period_key, "leaseOwner": lease_owner, "mode": mode, "effectId": effect_id, "fence": fence, "attempt": attempt, } def main() -> int: try: allow_test_http = ( os.environ.get("SELLABLE_CUSTOMER_PROGRAM_ALLOW_HTTP_LOOPBACK") == "1" ) config = _read_json_file(_config_path(), maximum=16_384) if set(config) != { "schemaVersion", "workspaceId", "apiUrl", "authorityPath", "jobKeys", } or config.get("schemaVersion") != "sellable-customer-program-schedule/v2": return _blocked("config_rejected") workspace_id = config.get("workspaceId") authority_path = config.get("authorityPath") job_keys = config.get("jobKeys") if ( not isinstance(workspace_id, str) or not SAFE_ID.fullmatch(workspace_id) or not isinstance(authority_path, str) or not Path(authority_path).is_absolute() or not isinstance(job_keys, list) or set(job_keys) != JOBS ): return _blocked("config_rejected") if not allow_test_http and not authority_path.startswith( "/run/sellable-managed-auth/" ): return _blocked("authority_path_rejected") job_key = _job_key(config) if job_key not in JOBS or job_key not in job_keys: return _emit({"wakeAgent": False, "reason": "unsupported_job"}) origin = _validate_origin(config.get("apiUrl"), allow_test_http) authority = _read_json_file(Path(authority_path), maximum=16_384) agent = authority.get("agent") token = authority.get("token") if ( authority.get("schemaVersion") != "sellable-managed-auth/v1" or authority.get("apiUrl") != origin or not isinstance(agent, dict) or agent.get("workspaceId") != workspace_id or not isinstance(token, str) or not TOKEN.fullmatch(token) ): return _blocked("authority_rejected") body = json.dumps( {"action": "schedule_due", "jobKey": job_key}, separators=(",", ":"), ).encode("utf-8") request = Request( f"{origin}/api/v3/mcp/customer-program", data=body, method="POST", headers={ "authorization": f"Bearer {token}", "content-type": "application/json", "x-workspaceid": workspace_id, }, ) with urlopen(request, timeout=10) as response: if response.status != 200: return _blocked(f"http_{response.status}") payload = response.read(65_537) if len(payload) > 65_536: return _blocked("response_too_large") return _emit(_safe_response(json.loads(payload), job_key)) except HTTPError as error: return _blocked(f"http_{error.code}") except URLError: return _blocked("control_plane_unavailable") except (OSError, ValueError, TypeError, json.JSONDecodeError): return _blocked("runtime_state_rejected") if __name__ == "__main__": raise SystemExit(main())