tegwick
ee4cf14cbc
specs/TrustServiceOnboarding.md defines the mechanism: a Phase Manifest file is committed to the declaring repo (durable, independently foldable forever) and separately registered with the hosted service; once registered, the Ledger's live authoritative copy is the hosted service only, not a second competing file. Licensor token bootstrapping is explicitly out of scope here (a WP-0008-T01 governance action). scripts/trf_onboard.py: a dependency-light CLI (stdlib urllib + target_revenue.validation only, no FastAPI/psycopg needed to onboard a Phase) with validate/register-phase/append-entry/status subcommands. The Licensor token is read only from a named environment variable, never accepted as a literal argument. tests/test_trf_onboard.py (4 tests, no network/Docker) proves invalid-manifest and missing-token-env cases fail before any HTTP attempt, by monkeypatching the request function to raise if called. tests/test_onboarding_hosted.py (1 Docker-gated test) runs an actual uvicorn server on a real socket and drives the full register -> append -> status round trip through the CLI as an external repo would invoke it.
136 lines
5 KiB
Python
136 lines
5 KiB
Python
#!/usr/bin/env python3
|
|
"""Trust Service onboarding CLI (WP-0006-T07).
|
|
|
|
Mechanism only — see `specs/TrustServiceOnboarding.md` for the full
|
|
process this wraps. Deliberately dependency-light: stdlib `urllib` for
|
|
HTTP (no `requests`/`httpx` requirement) plus `target_revenue.validation`
|
|
(already a core, non-service dependency) for offline pre-flight checks —
|
|
a repo onboarding a Phase should not need to install FastAPI/psycopg just
|
|
to register one.
|
|
|
|
Every subcommand that submits data validates it offline first, with the
|
|
same check the hosted service itself runs at registration
|
|
(`validation.py`), so a rejection is caught locally with an identical
|
|
field-by-field diff instead of only being discovered via a failed network
|
|
call (specs/TrustServiceOnboarding.md §3 steps 2 and 4).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "src"))
|
|
|
|
from target_revenue import validation # noqa: E402
|
|
|
|
|
|
def _load(path: str) -> dict:
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _request(url: str, method: str, token: str | None = None, body: dict | None = None) -> dict:
|
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
|
headers = {"Content-Type": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
detail = exc.read().decode("utf-8")
|
|
raise SystemExit(f"{method} {url} -> HTTP {exc.code}: {detail}") from exc
|
|
|
|
|
|
def _token_from_env(env_var: str) -> str:
|
|
token = os.environ.get(env_var)
|
|
if not token:
|
|
raise SystemExit(
|
|
f"environment variable {env_var!r} is not set — the Licensor "
|
|
"API token must never be passed on the command line or "
|
|
"committed to the repo (specs/TrustServiceOnboarding.md §2)"
|
|
)
|
|
return token
|
|
|
|
|
|
def cmd_validate(args: argparse.Namespace) -> None:
|
|
manifest = _load(args.manifest)
|
|
try:
|
|
validation.validate_phase_manifest(manifest)
|
|
except validation.ConformanceError as exc:
|
|
print("REJECTED:", "; ".join(exc.errors), file=sys.stderr)
|
|
raise SystemExit(1)
|
|
print(f"OK: {manifest['phase']['id']} is schema-conformant")
|
|
|
|
|
|
def cmd_register_phase(args: argparse.Namespace) -> None:
|
|
manifest = _load(args.manifest)
|
|
try:
|
|
validation.validate_phase_manifest(manifest)
|
|
except validation.ConformanceError as exc:
|
|
print("REJECTED (offline check, no network call made):", "; ".join(exc.errors), file=sys.stderr)
|
|
raise SystemExit(1)
|
|
token = _token_from_env(args.token_env)
|
|
result = _request(f"{args.url}/phases", "POST", token=token, body=manifest)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
def cmd_append_entry(args: argparse.Namespace) -> None:
|
|
entry = _load(args.entry)
|
|
token = _token_from_env(args.token_env)
|
|
result = _request(
|
|
f"{args.url}/phases/{args.phase_id}/ledger", "POST", token=token, body=entry
|
|
)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
def cmd_status(args: argparse.Namespace) -> None:
|
|
result = _request(f"{args.url}/phases/{args.phase_id}/metrics", "GET")
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="trf_onboard", description=__doc__)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
p_validate = sub.add_parser("validate", help="offline-validate a Phase Manifest file, no network call")
|
|
p_validate.add_argument("manifest")
|
|
p_validate.set_defaults(func=cmd_validate)
|
|
|
|
p_register = sub.add_parser("register-phase", help="validate offline, then POST /phases")
|
|
p_register.add_argument("--url", required=True, help="Trust Service base URL")
|
|
p_register.add_argument("--token-env", required=True, help="env var holding the Licensor API token")
|
|
p_register.add_argument("--manifest", required=True)
|
|
p_register.set_defaults(func=cmd_register_phase)
|
|
|
|
p_append = sub.add_parser("append-entry", help="POST a Ledger entry for an already-registered Phase")
|
|
p_append.add_argument("--url", required=True)
|
|
p_append.add_argument("--token-env", required=True)
|
|
p_append.add_argument("--phase-id", required=True)
|
|
p_append.add_argument("--entry", required=True)
|
|
p_append.set_defaults(func=cmd_append_entry)
|
|
|
|
p_status = sub.add_parser("status", help="GET /phases/{id}/metrics (public, no token)")
|
|
p_status.add_argument("--url", required=True)
|
|
p_status.add_argument("--phase-id", required=True)
|
|
p_status.set_defaults(func=cmd_status)
|
|
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> None:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|