110 lines
3.3 KiB
Python
Executable File
110 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Best-effort asynchronous delivery of a pre-sanitized CASAN run envelope."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
def canonicalize(value):
|
|
if isinstance(value, dict):
|
|
return {key: canonicalize(value[key]) for key in sorted(value)}
|
|
if isinstance(value, list):
|
|
return [canonicalize(item) for item in value]
|
|
if isinstance(value, float) and value.is_integer():
|
|
return int(value)
|
|
return value
|
|
|
|
|
|
def canonical_bytes(payload):
|
|
return json.dumps(
|
|
canonicalize(payload),
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
|
|
|
|
def deliver(spool_path, url, token):
|
|
try:
|
|
with open(spool_path, "r", encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
except (OSError, ValueError):
|
|
return False
|
|
body = canonical_bytes(payload)
|
|
timestamp = str(int(time.time()))
|
|
signed = timestamp.encode("ascii") + b"." + body
|
|
signature = hmac.new(
|
|
token.encode("utf-8"), signed, hashlib.sha256
|
|
).hexdigest()
|
|
request = Request(
|
|
url,
|
|
data=body,
|
|
method="POST",
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"X-CASAN-Timestamp": timestamp,
|
|
"X-CASAN-Signature": "sha256=%s" % signature,
|
|
"User-Agent": "CASAN-Core-Telemetry/1",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=3) as response:
|
|
if not 200 <= response.status < 300:
|
|
return False
|
|
except Exception:
|
|
# The pending spool remains for a later retry; prompt execution was
|
|
# already finalized and is never coupled to delivery availability.
|
|
return False
|
|
delivered_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(spool_path)), "delivered"
|
|
)
|
|
try:
|
|
os.makedirs(delivered_dir, exist_ok=True)
|
|
os.replace(spool_path, os.path.join(
|
|
delivered_dir, os.path.basename(spool_path)
|
|
))
|
|
except OSError:
|
|
# The server already accepted the trace. A retry is safe because the
|
|
# ingest endpoint is idempotent by trace_id.
|
|
return False
|
|
return True
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--spool", required=True)
|
|
parser.add_argument("--url", required=True)
|
|
parser.add_argument("--token-env", required=True)
|
|
args = parser.parse_args(argv)
|
|
token = os.environ.get(args.token_env)
|
|
if not token:
|
|
return 3
|
|
pending_dir = os.path.dirname(os.path.abspath(args.spool))
|
|
pending = sorted(
|
|
os.path.join(pending_dir, name)
|
|
for name in os.listdir(pending_dir)
|
|
if name.endswith(".json")
|
|
)[:100]
|
|
if os.path.abspath(args.spool) not in pending:
|
|
pending.append(os.path.abspath(args.spool))
|
|
success = True
|
|
for spool_path in pending:
|
|
if not os.path.isfile(spool_path):
|
|
continue
|
|
if not deliver(spool_path, args.url, token):
|
|
success = False
|
|
# Avoid a thundering herd while the Control Plane is unavailable.
|
|
break
|
|
return 0 if success else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|