diff --git a/lib/seo-data/seo-data.test.sh b/lib/seo-data/seo-data.test.sh new file mode 100644 index 0000000..1a22e08 --- /dev/null +++ b/lib/seo-data/seo-data.test.sh @@ -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 ] diff --git a/lib/seo-data/tokenstore.py b/lib/seo-data/tokenstore.py new file mode 100644 index 0000000..b1788f4 --- /dev/null +++ b/lib/seo-data/tokenstore.py @@ -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()