78 lines
2.4 KiB
Python
Executable File
78 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Call the loopback identity-overlay API without exposing the token in argv."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
def _read_secret(path: Path) -> str:
|
|
if path.is_symlink() or not path.is_file():
|
|
raise RuntimeError("SECRET_REF_INVALID")
|
|
if path.stat().st_size <= 0 or path.stat().st_size > 16_384:
|
|
raise RuntimeError("SECRET_REF_INVALID")
|
|
value = path.read_text(encoding="utf-8").strip()
|
|
if not value:
|
|
raise RuntimeError("SECRET_REF_EMPTY")
|
|
return value
|
|
|
|
|
|
def _safe_write(path: Path, value: dict) -> None:
|
|
data = json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(data, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--secret-ref", required=True, type=Path)
|
|
parser.add_argument("--request-json", required=True, type=Path)
|
|
parser.add_argument(
|
|
"--url",
|
|
default="http://127.0.0.1:19864/api/governance/owner-authorizations",
|
|
)
|
|
parser.add_argument("--result-json", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
token = _read_secret(args.secret_ref)
|
|
body = args.request_json.read_bytes()
|
|
request = urllib.request.Request(
|
|
args.url,
|
|
data=body,
|
|
method="POST",
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
payload = json.loads(response.read())
|
|
status = response.status
|
|
except urllib.error.HTTPError as exc:
|
|
status = exc.code
|
|
payload = json.loads(exc.read())
|
|
if not isinstance(payload, dict):
|
|
raise RuntimeError("RESPONSE_OBJECT_REQUIRED")
|
|
payload["http_status"] = status
|
|
_safe_write(args.result_json, payload)
|
|
return 0 if status == 200 else 2
|
|
except Exception:
|
|
_safe_write(
|
|
args.result_json,
|
|
{"status": "FAILED", "error": "OVERLAY_CLIENT_FAILED"},
|
|
)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|