429cc912ed
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Mint, verify, and revoke a coding-agent AppRole token without printing it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_ADDR = "https://bao.coulomb.social"
|
|
DATA_PATH = "platform/data/workloads/issue-core/issue-core/issue-core-runtime"
|
|
METADATA_PATH = (
|
|
"platform/metadata/workloads/issue-core/issue-core/issue-core-runtime"
|
|
)
|
|
EXPECTED_POLICIES = {
|
|
"agent-high-risk-boundary",
|
|
"workload-kv-read-issue-core-runtime",
|
|
}
|
|
|
|
|
|
def bao_json(args: list[str]) -> dict[str, Any]:
|
|
completed = subprocess.run(
|
|
["bao", *args],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
return json.loads(completed.stdout)
|
|
|
|
|
|
def api_request(
|
|
addr: str,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
token: str | None = None,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
headers = {"Content-Type": "application/json"}
|
|
if token:
|
|
headers["X-Vault-Token"] = token
|
|
body = json.dumps(payload).encode() if payload is not None else None
|
|
request = urllib.request.Request(
|
|
f"{addr.rstrip('/')}/v1/{path}", data=body, headers=headers, method=method
|
|
)
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
content = response.read()
|
|
return json.loads(content) if content else {}
|
|
|
|
|
|
def verify(role: str, addr: str) -> dict[str, Any]:
|
|
role_payload = bao_json(
|
|
["read", "-format=json", f"auth/approle/role/{role}/role-id"]
|
|
)
|
|
role_id = role_payload["data"]["role_id"]
|
|
secret_payload = bao_json(
|
|
["write", "-format=json", "-f", f"auth/approle/role/{role}/secret-id"]
|
|
)
|
|
secret_id = secret_payload["data"]["secret_id"]
|
|
|
|
agent_token: str | None = None
|
|
report: dict[str, Any] = {}
|
|
try:
|
|
login = api_request(
|
|
addr,
|
|
"POST",
|
|
"auth/approle/login",
|
|
payload={"role_id": role_id, "secret_id": secret_id},
|
|
)
|
|
agent_token = login["auth"]["client_token"]
|
|
lookup = api_request(addr, "GET", "auth/token/lookup-self", token=agent_token)
|
|
capabilities = api_request(
|
|
addr,
|
|
"POST",
|
|
"sys/capabilities-self",
|
|
token=agent_token,
|
|
payload={"paths": [DATA_PATH, METADATA_PATH]},
|
|
)
|
|
policies = set(lookup["data"].get("policies", []))
|
|
ttl = int(lookup["data"].get("ttl", 0))
|
|
data_caps = capabilities.get(DATA_PATH, [])
|
|
metadata_caps = capabilities.get(METADATA_PATH, [])
|
|
errors: list[str] = []
|
|
if policies != EXPECTED_POLICIES:
|
|
errors.append(f"unexpected policies: {sorted(policies)}")
|
|
if ttl <= 0 or ttl > 900:
|
|
errors.append(f"TTL outside 1..900 seconds: {ttl}")
|
|
if data_caps != ["deny"]:
|
|
errors.append(f"data capabilities are not deny: {data_caps}")
|
|
if "read" not in metadata_caps or "deny" in metadata_caps:
|
|
errors.append(f"metadata capabilities are not read-only: {metadata_caps}")
|
|
report = {
|
|
"role": role,
|
|
"policies": sorted(policies),
|
|
"ttl": ttl,
|
|
"data_capabilities": data_caps,
|
|
"metadata_capabilities": metadata_caps,
|
|
"deny_wins": data_caps == ["deny"],
|
|
"errors": errors,
|
|
"ok": not errors,
|
|
}
|
|
finally:
|
|
if agent_token:
|
|
try:
|
|
api_request(addr, "POST", "auth/token/revoke-self", token=agent_token)
|
|
except (OSError, urllib.error.HTTPError):
|
|
report.setdefault("errors", []).append("test token self-revocation failed")
|
|
report["ok"] = False
|
|
return report
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--role", default="coding-agent-railiance-platform")
|
|
parser.add_argument("--addr", default=os.environ.get("BAO_ADDR", DEFAULT_ADDR))
|
|
args = parser.parse_args()
|
|
try:
|
|
report = verify(args.role, args.addr)
|
|
except (
|
|
KeyError,
|
|
json.JSONDecodeError,
|
|
OSError,
|
|
subprocess.CalledProcessError,
|
|
subprocess.TimeoutExpired,
|
|
urllib.error.HTTPError,
|
|
) as exc:
|
|
report = {"role": args.role, "ok": False, "errors": [str(exc)]}
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0 if report["ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|