feat(seo-data): label-keyed atomic OAuth token store

2026-07-10 01:03:50 +02:00
parent 10f20438b1
commit fb0484954a
2 changed files with 100 additions and 0 deletions
@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# Deterministic tests for the seo-data engine (no network, no venv).
+set -u
+REPO="$(cd "$(dirname "$0")/../.." && pwd)"
+SD="$REPO/lib/seo-data"
+PASS=0; FAIL=0
+ok()   { echo "  PASS $1"; PASS=$((PASS+1)); }
+no()   { echo "  FAIL $1 — $2"; FAIL=$((FAIL+1)); }
+# assert stdout of a command contains / omits a fixed string
+has()  { if printf '%s' "$2" | grep -qF -- "$3"; then ok "$1"; else no "$1" "missing: $3"; fi; }
+hasnt(){ if printf '%s' "$2" | grep -qF -- "$3"; then no "$1" "forbidden: $3"; else ok "$1"; fi; }
+
+echo "── tokenstore ──"
+TMP="$(mktemp -d)"; STORE="$TMP/tokens.json"
+python3 "$SD/tokenstore.py" set --file "$STORE" --label client-a \
+  --refresh-token RT_AAA --scopes https://www.googleapis.com/auth/webmasters.readonly \
+  --properties sc-domain:a.com,https://www.a.com/ >/dev/null
+python3 "$SD/tokenstore.py" set --file "$STORE" --label client-b \
+  --refresh-token RT_BBB --scopes https://www.googleapis.com/auth/webmasters.readonly \
+  --properties sc-domain:b.com >/dev/null
+LIST="$(python3 "$SD/tokenstore.py" list --file "$STORE")"
+has   "list shows client-a"            "$LIST" '"client-a"'
+has   "list shows client-b"            "$LIST" '"client-b"'
+has   "list shows a property"          "$LIST" 'sc-domain:a.com'
+hasnt "list redacts refresh tokens"    "$LIST" 'RT_AAA'
+PERM="$(stat -c '%a' "$STORE")"
+[ "$PERM" = "600" ] && ok "store file is 0600" || no "store file 0600" "got $PERM"
+rm -rf "$TMP"
+
+echo ""
+echo "seo-data engine: $PASS pass, $FAIL fail"
+[ "$FAIL" -eq 0 ]
@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""Label-keyed OAuth refresh-token store. Atomic writes under an fcntl lock.
+No third-party deps — must run without the venv (used by the offline test path)."""
+import argparse, fcntl, json, os, sys, tempfile
+from datetime import datetime, timezone
+
+def load(path):
+    if not os.path.exists(path):
+        return {"version": 1, "accounts": {}}
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+def list_accounts(path):
+    data = load(path)
+    return [
+        {"label": lbl, "properties": a.get("properties", []),
+         "granted_at": a.get("granted_at")}
+        for lbl, a in data.get("accounts", {}).items()
+    ]  # refresh_token intentionally omitted (redaction)
+
+def get_refresh_token(path, label):
+    return load(path).get("accounts", {}).get(label, {}).get("refresh_token")
+
+def save_account(path, label, refresh_token, scopes, properties):
+    os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True)
+    lock_path = path + ".lock"
+    with open(lock_path, "w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)          # serialize concurrent connects
+        data = load(path)
+        data.setdefault("version", 1)
+        data.setdefault("accounts", {})
+        data["accounts"][label] = {
+            "refresh_token": refresh_token,
+            "scopes": scopes,
+            "granted_at": datetime.now(timezone.utc).isoformat(),
+            "properties": properties,
+        }
+        fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2)
+                f.flush(); os.fsync(f.fileno())
+            os.chmod(tmp, 0o600)
+            os.replace(tmp, path)                 # atomic
+        finally:
+            if os.path.exists(tmp):
+                os.unlink(tmp)
+
+def _cli():
+    p = argparse.ArgumentParser()
+    sub = p.add_subparsers(dest="cmd", required=True)
+    pl = sub.add_parser("list"); pl.add_argument("--file", required=True)
+    ps = sub.add_parser("set")
+    for flag in ("--file", "--label", "--refresh-token"):
+        ps.add_argument(flag, required=True)
+    ps.add_argument("--scopes", default="")
+    ps.add_argument("--properties", default="")
+    args = p.parse_args()
+    if args.cmd == "list":
+        print(json.dumps({"status": "ok", "accounts": list_accounts(args.file)}))
+    else:
+        save_account(args.file, args.label, getattr(args, "refresh_token"),
+                     [s for s in args.scopes.split(",") if s],
+                     [x for x in args.properties.split(",") if x])
+        print(json.dumps({"status": "ok"}))
+
+if __name__ == "__main__":
+    _cli()