"""Phase 4 admin P4-04..P4-06 API verification. Runs against the running API on 127.0.0.1:8000 using the admin account whose credentials live in /etc/kaotings/p4-ops.secrets (root-only). It must never print the admin password or any secret. Requires the venv to have httpx. """ from __future__ import annotations import json import os import sys from datetime import datetime, timedelta, timezone from uuid import uuid4 import httpx API_ROOT = "https://192.168.199.22/api/v1" def load_secret(name: str) -> str: value = os.getenv(name) if not value: raise SystemExit(f"missing env {name}") return value def client() -> httpx.Client: session = httpx.Client(base_url=API_ROOT, timeout=15, follow_redirects=False, verify=False) csrf = session.get("/auth/csrf").json()["csrf_token"] resp = session.post("/auth/login", json={"email": load_secret("P4_OPS_EMAIL"), "password": load_secret("P4_OPS_PASSWORD")}, headers={"X-CSRF-Token": csrf}) assert resp.status_code == 200, resp.text session.headers.update({"X-CSRF-Token": csrf}) return session def mk_user(session: httpx.Client, prefix: str) -> tuple[str, str]: email = f"p4-{prefix}-{uuid4().hex[:10]}@example.com" password = f"P4T-{uuid4().hex}-Aa1!" body = {"email": email, "password": password} csrf = session.get("/auth/csrf").json()["csrf_token"] assert session.post("/auth/register", json=body, headers={"X-CSRF-Token": csrf}).status_code == 201, "register failed" return email, password def find_user_by_email(session: httpx.Client, email: str) -> dict: r = session.get("/admin/users", params={"email": email}) assert r.status_code == 200, r.text items = r.json()["items"] assert items, f"user {email} not found" return next(i for i in items if i["email"] == email) def main() -> None: results: dict = {} session = client() # P4-04: pagination + filters + detail r = session.get("/admin/users", params={"limit": 5, "page": 1}) assert r.status_code == 200, r.text body = r.json() results["users_page"] = {"total": body["total"], "pages": body["pages"], "items": len(body["items"])} email, user_password = mk_user(session, "detail") target = find_user_by_email(session, email) results["users_filter_email"] = {"found": True, "status": target["status"]} detail_resp = session.get(f"/admin/users/{target['id']}") assert detail_resp.status_code == 200, f"detail {detail_resp.status_code}: {detail_resp.text[:300]}" detail = detail_resp.json() results["user_detail"] = {"has_quota": "quota" in detail, "effective_plan": detail["effective_plan"]} # disable -> sessions revoked; verify last-admin protection separately csrf = session.headers["X-CSRF-Token"] dis = session.patch(f"/admin/users/{target['id']}/status", json={"status": "disabled", "reason": "P4-04 disable test"}, headers={"X-CSRF-Token": csrf}) assert dis.status_code == 200, dis.text results["disable"] = dis.json() # disabled user cannot log in with its real password unauth = httpx.Client(base_url=API_ROOT, timeout=10, follow_redirects=False, verify=False) dcs = unauth.get("/auth/csrf").json()["csrf_token"] dl = unauth.post("/auth/login", json={"email": email, "password": user_password}, headers={"X-CSRF-Token": dcs}) results["disabled_login_blocked"] = dl.status_code in {401, 403} unauth.close() rev = session.patch(f"/admin/users/{target['id']}/status", json={"status": "active", "reason": "P4-04 re-enable test"}, headers={"X-CSRF-Token": csrf}) assert rev.status_code == 200, rev.text results["restore"] = rev.json() # last-admin protection: try disabling the acting admin me = session.get("/auth/me").json() admin_target = session.get("/admin/users", params={"email": me["email"]}).json()["items"][0] # there is more than one admin, so disabling one is allowed; check non-admin guard instead below # P4-05: membership view + grant + revoke mem = session.get(f"/admin/users/{target['id']}/membership") results["membership_view"] = mem.json() grant = session.put(f"/admin/users/{target['id']}/membership", json={"plan": "vip", "expires_at": (datetime.now(timezone.utc) + timedelta(days=30)).isoformat(), "reason": "P4-05 vip grant"}, headers={"X-CSRF-Token": csrf}) assert grant.status_code == 200, grant.text results["vip_grant"] = grant.json() mem2 = session.get(f"/admin/users/{target['id']}/membership").json() results["vip_effective"] = mem2["effective_plan"] revoke = session.post(f"/admin/users/{target['id']}/membership/revoke", json={"reason": "P4-05 revoke"}, headers={"X-CSRF-Token": csrf}) assert revoke.status_code == 200, revoke.text results["vip_revoke"] = revoke.json() mem3 = session.get(f"/admin/users/{target['id']}/membership").json() results["vip_effective_after_revoke"] = mem3["effective_plan"] # quota view + adjustment with reason + idempotency + balance constraint quota = session.get(f"/admin/users/{target['id']}/quota").json() results["quota_view"] = quota["quota"] idem = f"p4-adjust-{uuid4().hex}" adj = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 500, "reason": "P4-05 grant bonus", "idempotency_key": idem}, headers={"X-CSRF-Token": csrf}) results["quota_adjust"] = adj.status_code dup = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 500, "reason": "P4-05 dup", "idempotency_key": idem}, headers={"X-CSRF-Token": csrf}) results["quota_adjust_idempotent"] = dup.status_code == 409 # balance constraint: cannot push below used hard = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": -100000, "reason": "P4-05 too low", "idempotency_key": f"p4-adjust-{uuid4().hex}"}, headers={"X-CSRF-Token": csrf}) hard_body = hard.json().get("error", hard.json()) if hard.status_code >= 400 else {} results["quota_balance_constraint"] = {"status": hard.status_code, "code": hard_body.get("code")} # missing reason rejected noreason = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 10, "idempotency_key": f"p4-adjust-{uuid4().hex}"}, headers={"X-CSRF-Token": csrf}) results["quota_reason_required"] = noreason.status_code == 422 # P4-06: tts tasks view + audit logs + usage summary tasks = session.get("/admin/tts/tasks", params={"page": 1, "limit": 10}) assert tasks.status_code == 200, tasks.text results["tts_tasks"] = {"total": tasks.json()["total"], "items": len(tasks.json()["items"])} summary = session.get("/admin/usage/summary") assert summary.status_code == 200, summary.text results["usage_summary"] = summary.json() audit = session.get("/admin/audit-logs", params={"limit": 30}) assert audit.status_code == 200, audit.text audit_items = audit.json()["items"] # never show secrets / never include user text fields in the payload shape for item in audit_items: keys = set(item.keys()) assert "text" not in keys and "password" not in keys results["audit"] = {"total": audit.json()["total"], "sample_keys": sorted(set().union(*(set(i.keys()) for i in audit_items[0:1])) if audit_items else [])} # permission guard: a normal (non-admin) user account cannot reach admin endpoints known_email = f"p4-known-{uuid4().hex}@example.com" known_pw = f"K-{uuid4().hex}-Aa1!" reg_session = httpx.Client(base_url=API_ROOT, timeout=10, follow_redirects=True, verify=False) rcs_k = reg_session.get("/auth/csrf").json()["csrf_token"] assert reg_session.post("/auth/register", json={"email": known_email, "password": known_pw}, headers={"X-CSRF-Token": rcs_k}).status_code == 201 assert reg_session.post("/auth/login", json={"email": known_email, "password": known_pw}, headers={"X-CSRF-Token": rcs_k}).status_code == 200 guard = reg_session.get("/admin/users") guard_detail = reg_session.get(f"/admin/users/{target['id']}") results["regular_forbidden_users"] = guard.status_code == 403 results["regular_forbidden_detail"] = guard_detail.status_code == 403 reg_session.close() print(json.dumps(results, ensure_ascii=False, indent=2, default=str)) if __name__ == "__main__": main()