← back to Ga Allsites
add grant_microsites_gsc.py: no-click GSC grant via DNS-TXT owner-verification (Steve-run, gated)
69d7b8242dff843d3120dc282a888bb3ab484a03 · 2026-08-17 09:44:55 -0700 · Steve Abrams
Files touched
A grant_microsites_gsc.py
Diff
commit 69d7b8242dff843d3120dc282a888bb3ab484a03
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Mon Aug 17 09:44:55 2026 -0700
add grant_microsites_gsc.py: no-click GSC grant via DNS-TXT owner-verification (Steve-run, gated)
---
grant_microsites_gsc.py | 172 ++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 172 insertions(+)
diff --git a/grant_microsites_gsc.py b/grant_microsites_gsc.py
new file mode 100644
index 0000000..5d02e85
--- /dev/null
+++ b/grant_microsites_gsc.py
@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""
+Grant the ga-allsites service account read access to microsite Search Console
+properties WITHOUT any console clicks — by verifying the SA as a domain OWNER
+via a DNS TXT record (Google Site Verification API) and adding the sc-domain
+property. Steve-run only (identity + DNS change; gated to Steve by design).
+
+What it does per domain (idempotent, reversible):
+ 1. SA mints a DNS-TXT verification token (Site Verification API).
+ 2. Writes that TXT record to the domain's Cloudflare zone.
+ 3. Verifies -> the SA becomes a VERIFIED OWNER of the domain.
+ 4. Adds the `sc-domain:<domain>` Search Console property.
+ -> keyword queries flow on the next ETL run (every 30 min).
+
+REVERSIBLE: remove the SA in Search Console (Users & permissions) and delete the
+google-site-verification TXT record in Cloudflare any time.
+
+NOTE: this grants OWNER-level (more than the "Restricted user" a manual add would
+give). The ga-allsites ETL only ever issues read-only searchAnalytics.query calls.
+
+Usage:
+ python3 grant_microsites_gsc.py # canary: novasuede.com only
+ python3 grant_microsites_gsc.py --site X.com # one specific domain
+ python3 grant_microsites_gsc.py --all # all recommended microsites
+
+Creds:
+ SA key : ~/.config/ga-analytics-agent/service-account.json
+ CF token: env CLOUDFLARE_API_TOKEN / CF_API_TOKEN, else ~/Projects/secrets-manager/.env
+"""
+import json
+import os
+import pathlib
+import sys
+import time
+import urllib.error
+import urllib.request
+
+import google.auth.transport.requests as gareq
+from google.oauth2 import service_account as sa
+
+KEY = pathlib.Path.home() / ".config" / "ga-analytics-agent" / "service-account.json"
+SECRETS_ENV = pathlib.Path.home() / "Projects" / "secrets-manager" / ".env"
+SCOPES = [
+ "https://www.googleapis.com/auth/siteverification",
+ "https://www.googleapis.com/auth/webmasters",
+]
+# The recommended queue (organic-floored, verifiability-filtered) — novasuede first.
+RECOMMENDED = [
+ "novasuede.com", "wallpapercanada.com", "67calc.com",
+ "nationalpaperhangers.com", "hospitalitywallcoverings.com",
+ "glassbeadedwallpaper.com", "flockedwallpaper.com",
+ "1890swallpaper.com", "retrowalls.com",
+]
+
+
+def cf_token() -> str:
+ for k in ("CLOUDFLARE_API_TOKEN", "CF_API_TOKEN", "CF_TOKEN"):
+ if os.environ.get(k):
+ return os.environ[k]
+ if SECRETS_ENV.exists():
+ for line in SECRETS_ENV.read_text().splitlines():
+ line = line.strip()
+ for k in ("CLOUDFLARE_API_TOKEN", "CF_API_TOKEN", "CF_TOKEN"):
+ if line.startswith(k + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit("ERROR: no Cloudflare API token in env or secrets-manager/.env")
+
+
+def google_creds():
+ creds = sa.Credentials.from_service_account_file(str(KEY), scopes=SCOPES)
+ creds.refresh(gareq.Request())
+ return creds
+
+
+def gget(url, token, payload=None, method=None):
+ m = method or ("POST" if payload else "GET")
+ req = urllib.request.Request(
+ url, method=m,
+ data=json.dumps(payload).encode() if payload else None,
+ headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"})
+ with urllib.request.urlopen(req, timeout=30) as r:
+ body = r.read().decode()
+ return json.loads(body) if body else {}
+
+
+def cf(url, tok, payload=None, method=None):
+ return gget(url, tok, payload, method)
+
+
+def grant(domain: str, gtoken: str, cftok: str) -> str:
+ # 1. mint DNS TXT token
+ tok = gget("https://www.googleapis.com/siteVerification/v1/token", gtoken, {
+ "verificationMethod": "DNS_TXT",
+ "site": {"type": "INET_DOMAIN", "identifier": domain},
+ })
+ txt = tok["token"]
+ print(f" [{domain}] token: {txt}")
+
+ # 2. find CF zone + upsert the TXT record on the apex
+ zones = cf(f"https://api.cloudflare.com/client/v4/zones?name={domain}", cftok)
+ if not zones.get("result"):
+ return "SKIP (no Cloudflare zone)"
+ zid = zones["result"][0]["id"]
+ existing = cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records"
+ f"?type=TXT&name={domain}", cftok)
+ already = any(r["content"].strip('"') == txt for r in existing.get("result", []))
+ if not already:
+ cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records", cftok,
+ {"type": "TXT", "name": domain, "content": txt, "ttl": 120})
+ print(f" [{domain}] TXT written to Cloudflare (zone {zid[:8]}…)")
+ time.sleep(8) # let DNS propagate before asking Google to check
+ else:
+ print(f" [{domain}] TXT already present")
+
+ # 3. verify -> SA becomes a verified owner
+ try:
+ gget("https://www.googleapis.com/siteVerification/v1/webResource"
+ "?verificationMethod=DNS_TXT", gtoken, {
+ "site": {"type": "INET_DOMAIN", "identifier": domain}})
+ print(f" [{domain}] verified as owner ✓")
+ except urllib.error.HTTPError as e:
+ return f"VERIFY FAILED ({e.code}: {e.read().decode()[:160]})"
+
+ # 4. add the sc-domain Search Console property
+ import urllib.parse
+ site = urllib.parse.quote(f"sc-domain:{domain}", safe="")
+ try:
+ gget(f"https://searchconsole.googleapis.com/webmasters/v3/sites/{site}",
+ gtoken, method="PUT")
+ print(f" [{domain}] sc-domain property added ✓")
+ except urllib.error.HTTPError as e:
+ if e.code not in (409,): # 409 = already present
+ return f"ADDED-OWNER but sites.add {e.code}"
+ return "GRANTED ✓"
+
+
+def main():
+ args = sys.argv[1:]
+ if "--all" in args:
+ targets = RECOMMENDED
+ elif "--site" in args:
+ targets = [args[args.index("--site") + 1]]
+ else:
+ targets = ["novasuede.com"] # canary default
+
+ if not KEY.exists():
+ raise SystemExit(f"ERROR: SA key not found at {KEY}")
+ creds = google_creds()
+ gtoken = creds.token
+ cftok = cf_token()
+ print(f"SA: {json.loads(KEY.read_text()).get('client_email')}")
+ print(f"Granting {len(targets)} domain(s): {', '.join(targets)}\n")
+
+ results = {}
+ for d in targets:
+ try:
+ results[d] = grant(d, gtoken, cftok)
+ except urllib.error.HTTPError as e:
+ results[d] = f"ERROR {e.code}: {e.read().decode()[:160]}"
+ except Exception as e: # noqa: BLE001
+ results[d] = f"ERROR: {e}"
+ print()
+
+ print("=== SUMMARY ===")
+ for d, r in results.items():
+ print(f" {d:32} {r}")
+ print("\nNext: cd ~/Projects/ga-allsites && /opt/homebrew/bin/python3 etl.py")
+ print("then check keywords.json gsc_site_count climbs (data lags ~2 days).")
+
+
+if __name__ == "__main__":
+ main()
← b1a8d9e dashboard: align grankStatus text order with grankOrder sort
·
back to Ga Allsites
·
grant script: hoist urllib.parse to module scope (fix Unboun 0217519 →