93 lines
2.8 KiB
Python
Executable File
93 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Root-governed CLI for issuing and revoking task identity overlays.
|
|
|
|
The CLI never prints the bearer token. It returns only the root-owned secret
|
|
reference and non-secret metadata.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from task_identity_overlay import (
|
|
IdentityOverlayConfig,
|
|
IdentityOverlayError,
|
|
TaskIdentityOverlayStore,
|
|
)
|
|
|
|
|
|
def _write_result(path: str, value: dict) -> None:
|
|
data = json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n"
|
|
if path == "-":
|
|
sys.stdout.write(data)
|
|
return
|
|
target = Path(path)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(data, encoding="utf-8")
|
|
os.chmod(target, 0o600)
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
value = argparse.ArgumentParser(
|
|
description="Issue/status/revoke a task-scoped identity overlay"
|
|
)
|
|
sub = value.add_subparsers(dest="command", required=True)
|
|
|
|
issue = sub.add_parser("issue")
|
|
issue.add_argument("--ttl-seconds", type=int, default=900)
|
|
issue.add_argument("--authorization-assertion-ref", required=True)
|
|
issue.add_argument(
|
|
"--operator-identity",
|
|
default=os.getenv("SUDO_USER") or os.getenv("USER") or "root",
|
|
)
|
|
issue.add_argument("--result-json", default="-")
|
|
|
|
status = sub.add_parser("status")
|
|
status.add_argument("--overlay-id", required=True)
|
|
status.add_argument("--result-json", default="-")
|
|
|
|
revoke = sub.add_parser("revoke")
|
|
revoke.add_argument("--overlay-id", required=True)
|
|
revoke.add_argument("--reason", required=True)
|
|
revoke.add_argument("--result-json", default="-")
|
|
return value
|
|
|
|
|
|
def main() -> int:
|
|
args = parser().parse_args()
|
|
try:
|
|
store = TaskIdentityOverlayStore(
|
|
IdentityOverlayConfig.from_environment()
|
|
)
|
|
if args.command == "issue":
|
|
result = store.issue(
|
|
ttl_seconds=args.ttl_seconds,
|
|
authorization_assertion_ref=args.authorization_assertion_ref,
|
|
operator_identity=args.operator_identity,
|
|
)
|
|
elif args.command == "status":
|
|
result = store.status(args.overlay_id)
|
|
else:
|
|
result = store.revoke(args.overlay_id, reason=args.reason)
|
|
_write_result(args.result_json, result)
|
|
return 0
|
|
except IdentityOverlayError as exc:
|
|
body = {"status": "BLOCKED", "error": exc.code}
|
|
if exc.details:
|
|
body["details"] = exc.details
|
|
_write_result(getattr(args, "result_json", "-"), body)
|
|
return 2
|
|
except Exception:
|
|
_write_result(
|
|
getattr(args, "result_json", "-"),
|
|
{"status": "FAILED", "error": "IDENTITY_OVERLAY_CLI_INTERNAL_ERROR"},
|
|
)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|