100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Issue or revoke a short-lived API token used by live acceptance scripts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
def call(client: httpx.Client, method: str, path: str, **kwargs) -> httpx.Response:
|
|
response = client.request(method, path, **kwargs)
|
|
if response.is_error:
|
|
raise RuntimeError(f"{method} {path} -> {response.status_code}: {response.text[:1000]}")
|
|
return response
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", required=True)
|
|
password = parser.add_mutually_exclusive_group()
|
|
password.add_argument("--password")
|
|
password.add_argument("--password-stdin", action="store_true")
|
|
password.add_argument("--storage-state", type=Path)
|
|
parser.add_argument("--token-file", type=Path, required=True)
|
|
parser.add_argument("--id-file", type=Path, required=True)
|
|
parser.add_argument("--name", default="ImageFind live acceptance")
|
|
parser.add_argument("--revoke", action="store_true")
|
|
args = parser.parse_args()
|
|
token_revoke_ready = args.revoke and args.token_file.exists() and args.id_file.exists()
|
|
if not token_revoke_ready and not (args.password or args.password_stdin or args.storage_state):
|
|
parser.error("one of --password, --password-stdin or --storage-state is required")
|
|
password_value = sys.stdin.readline().rstrip("\r\n") if args.password_stdin else args.password
|
|
|
|
client = httpx.Client(
|
|
base_url=args.base_url.rstrip("/"),
|
|
timeout=httpx.Timeout(30, connect=10),
|
|
)
|
|
if token_revoke_ready:
|
|
client.headers["Authorization"] = (
|
|
f"Bearer {args.token_file.read_text(encoding='utf-8').strip()}"
|
|
)
|
|
token_id = args.id_file.read_text(encoding="utf-8").strip()
|
|
call(client, "DELETE", f"/api/v1/tokens/{token_id}")
|
|
args.token_file.unlink(missing_ok=True)
|
|
args.id_file.unlink(missing_ok=True)
|
|
print(f"revoked live acceptance token {token_id}")
|
|
return
|
|
|
|
if args.storage_state:
|
|
state = json.loads(args.storage_state.read_text(encoding="utf-8"))
|
|
for cookie in state.get("cookies", []):
|
|
client.cookies.set(
|
|
str(cookie["name"]),
|
|
str(cookie["value"]),
|
|
domain=str(cookie.get("domain") or ""),
|
|
path=str(cookie.get("path") or "/"),
|
|
)
|
|
login = call(client, "GET", "/api/v1/auth/me").json()
|
|
else:
|
|
login = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/auth/login",
|
|
json={"password": password_value, "remember_device": False},
|
|
).json()
|
|
password_value = ""
|
|
csrf = str(login.get("csrf_token") or "")
|
|
if not csrf:
|
|
raise AssertionError("login response omitted CSRF token")
|
|
client.headers["X-CSRF-Token"] = csrf
|
|
|
|
if args.revoke:
|
|
token_id = args.id_file.read_text(encoding="utf-8").strip()
|
|
call(client, "DELETE", f"/api/v1/tokens/{token_id}")
|
|
args.token_file.unlink(missing_ok=True)
|
|
args.id_file.unlink(missing_ok=True)
|
|
print(f"revoked live acceptance token {token_id}")
|
|
return
|
|
|
|
result = call(
|
|
client,
|
|
"POST",
|
|
"/api/v1/tokens",
|
|
json={"name": args.name, "scopes": ["admin"]},
|
|
).json()
|
|
args.token_file.write_text(str(result["token"]), encoding="utf-8")
|
|
args.id_file.write_text(str(result["id"]), encoding="utf-8")
|
|
os.chmod(args.token_file, 0o600)
|
|
os.chmod(args.id_file, 0o600)
|
|
print(f"issued live acceptance token {result['id']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|