approval-engine/approval_engine/cli.py

from __future__ import annotations

import argparse
import json
import threading
from pathlib import Path
from wsgiref.simple_server import make_server

from .api import App
from .audit import AuditCoreSink, OutboxWorker
from .auth import Identity, JWTAuthenticator, StaticTokenAuthenticator
from .store import Engine

ALL_SCOPES = frozenset(
    {
        "approval:observe",
        "approval:emit",
        "approval:create",
        "approval:read",
        "approval:approve",
        "approval:revoke",
        "approval:supersede",
        "approval:consume",
    }
)


def _common_db(sub: argparse.ArgumentParser) -> None:
    sub.add_argument("--db", default="approvals.sqlite")


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="approval-engine")
    sub = parser.add_subparsers(dest="cmd", required=True)
    serve = sub.add_parser("serve", help="serve the authenticated HTTP API")
    _common_db(serve)
    serve.add_argument("--host", default="127.0.0.1")
    serve.add_argument("--port", type=int, default=8787)
    serve.add_argument("--production", action="store_true")
    serve.add_argument("--tenant", default="platform")
    serve.add_argument("--jwt-issuer")
    serve.add_argument("--jwt-audience")
    serve.add_argument("--jwks-url")
    serve.add_argument("--dev-token-file")
    serve.add_argument("--audit-url")
    serve.add_argument("--audit-token-file")
    serve.add_argument("--outbox-poll-seconds", type=float, default=5)
    serve.add_argument("--heartbeat-seconds", type=int, default=86400)
    migrate = sub.add_parser("migrate", help="apply repeatable schema migrations")
    _common_db(migrate)
    verify = sub.add_parser("verify", help="verify schema and SQLite integrity")
    _common_db(verify)
    backup = sub.add_parser("backup", help="create and verify an online backup")
    _common_db(backup)
    backup.add_argument("--output", required=True)
    return parser


def _authenticator(args: argparse.Namespace, parser: argparse.ArgumentParser):
    jwt_values = (args.jwt_issuer, args.jwt_audience, args.jwks_url)
    if all(jwt_values):
        return JWTAuthenticator(
            issuer=args.jwt_issuer,
            audience=args.jwt_audience,
            jwks_url=args.jwks_url,
        )
    if any(jwt_values):
        parser.error("--jwt-issuer, --jwt-audience, and --jwks-url are a set")
    if args.production:
        parser.error("production requires KeyCape JWT verifier configuration")
    if not args.dev_token_file:
        parser.error("configure JWT verification or explicit --dev-token-file")
    token = Path(args.dev_token_file).read_text(encoding="utf-8").strip()
    if not token:
        parser.error("development token file is empty")
    identity = Identity(
        subject="development-agent",
        issuer="local-development",
        audiences=("approval-engine",),
        principal_type="agent",
        tenant="development",
        roles=frozenset({"developer"}),
        scopes=ALL_SCOPES,
        assurance={"method": "explicit-development-token"},
        evidence_ref="local-development-token",
    )
    return StaticTokenAuthenticator({token: identity})


def _serve(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int:
    if args.production and args.db == ":memory:":
        parser.error("production requires a persistent database")
    if bool(args.audit_url) != bool(args.audit_token_file):
        parser.error("--audit-url and --audit-token-file are a set")
    if args.production and not args.audit_url:
        parser.error("production requires authenticated audit delivery")
    authenticator = _authenticator(args, parser)
    engine = Engine(args.db, auto_migrate=not args.production, tenant=args.tenant)
    storage = engine.storage_status(integrity=True)
    if args.production and (not storage["ok"] or not storage["persistent"]):
        parser.error("production store is not ready; run migrate and verify")
    app = App(engine, authenticator, require_persistent=args.production)
    stop = threading.Event()
    if args.audit_url:
        sink = AuditCoreSink(args.audit_url, args.audit_token_file)
        worker = OutboxWorker(
            engine,
            sink,
            heartbeat_interval_seconds=args.heartbeat_seconds,
        )
        threading.Thread(
            target=worker.run_forever,
            args=(stop, args.outbox_poll_seconds),
            daemon=True,
            name="approval-outbox",
        ).start()
    print(f"approval-engine listening on {args.host}:{args.port} db={args.db}")
    try:
        if args.production:
            from waitress import serve

            serve(app, host=args.host, port=args.port, threads=4)
        else:
            make_server(args.host, args.port, app).serve_forever()
    finally:
        stop.set()
        engine.close()
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = _parser()
    args = parser.parse_args(argv)
    if args.cmd == "migrate":
        engine = Engine(args.db, auto_migrate=True)
        print(json.dumps(engine.storage_status(integrity=True), sort_keys=True))
        engine.close()
        return 0
    if args.cmd == "verify":
        engine = Engine(args.db, auto_migrate=False)
        status = engine.storage_status(integrity=True)
        print(json.dumps(status, sort_keys=True))
        engine.close()
        return 0 if status["ok"] else 1
    if args.cmd == "backup":
        engine = Engine(args.db, auto_migrate=False)
        print(json.dumps(engine.backup(args.output), sort_keys=True))
        engine.close()
        return 0
    if args.cmd == "serve":
        return _serve(args, parser)
    return 2


if __name__ == "__main__":
    raise SystemExit(main())