← back to Ga Allsites
grant_microsites_gsc.py
268 lines
#!/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.parse
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 _env(*keys):
"""First matching value from process env, else from the secrets .env."""
for k in keys:
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 keys:
if line.startswith(k + "="):
return line.split("=", 1)[1].strip().strip('"').strip("'")
return ""
def cf_token() -> str:
t = _env("CLOUDFLARE_API_TOKEN", "CF_API_TOKEN", "CF_TOKEN")
if not t:
raise SystemExit("ERROR: no Cloudflare API token in env or secrets-manager/.env")
return t
def godaddy_creds():
key = _env("GODADDY_API_KEY", "GODADDY_KEY")
sec = _env("GODADDY_API_SECRET", "GODADDY_SECRET")
return (key, sec) if key and sec else (None, None)
def write_txt(domain: str, txt: str, cftok: str, gd) -> str:
"""Upsert the apex TXT to whichever host is authoritative. Returns
'present' | 'written' | 'no-host'."""
# Cloudflare — only if the zone is ACTIVE (pending zones aren't authoritative)
zones = cf(f"https://api.cloudflare.com/client/v4/zones?name={domain}", cftok)
z = (zones.get("result") or [None])[0]
if z and z.get("status") == "active":
zid = z["id"]
existing = cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records"
f"?type=TXT&name={domain}", cftok)
if any(r["content"].strip('"') == txt for r in existing.get("result", [])):
return "present"
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]}…)")
return "written"
# GoDaddy (domaincontrol.com nameservers) — APPEND via PATCH (never clobbers SPF/etc.)
gkey, gsec = gd
if gkey and gsec:
hdr = {"Authorization": f"sso-key {gkey}:{gsec}", "Content-Type": "application/json"}
base = f"https://api.godaddy.com/v1/domains/{domain}/records"
# read existing @ TXT records (GoDaddy uses sso-key auth, not bearer)
req = urllib.request.Request(f"{base}/TXT/%40", headers=hdr)
try:
cur = json.loads(urllib.request.urlopen(req, timeout=30).read() or "[]")
except Exception:
cur = []
if any(r.get("data", "").strip('"') == txt for r in cur):
return "present"
body = json.dumps([{"type": "TXT", "name": "@", "data": txt, "ttl": 600}]).encode()
preq = urllib.request.Request(base, data=body, headers=hdr, method="PATCH")
urllib.request.urlopen(preq, timeout=30).read()
print(f" [{domain}] TXT appended at GoDaddy")
return "written"
return "no-host"
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, gd) -> 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. write the TXT to whichever host is authoritative (Cloudflare or GoDaddy)
state = write_txt(domain, txt, cftok, gd)
if state == "no-host":
return "SKIP (DNS not on active Cloudflare or GoDaddy)"
if state == "written":
time.sleep(12) # 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
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-fleet" in args:
# every GA4 property in the cache -> its guessed apex domain
rows = json.loads((pathlib.Path(__file__).parent / "cache" / "data.json"
).read_text())["rows"]
def _sl(n):
x = n.lower().replace(" - ga4", "").replace("&", "and")
return "".join(c for c in x if c.isalnum())
targets = sorted({_sl(r["property"]) + ".com" for r in rows if _sl(r["property"])})
elif "--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()
gd = godaddy_creds()
# skip domains the SA can already query
try:
req = urllib.request.Request(
"https://searchconsole.googleapis.com/webmasters/v3/sites",
headers={"Authorization": f"Bearer {gtoken}"})
have = {s["siteUrl"].replace("sc-domain:", "")
for s in json.loads(urllib.request.urlopen(req, timeout=30).read()
).get("siteEntry", [])
if s.get("permissionLevel") != "siteUnverifiedUser"}
except Exception: # noqa: BLE001
have = set()
skipped_have = [d for d in targets if d in have]
targets = [d for d in targets if d not in have]
print(f"SA: {json.loads(KEY.read_text()).get('client_email')}")
print(f"GoDaddy API: {'available' if gd[0] else 'NOT configured'}")
if skipped_have:
print(f"Already granted (skipping {len(skipped_have)}): {', '.join(skipped_have)}")
print(f"Granting {len(targets)} domain(s)\n")
def _do(d):
try:
return grant(d, gtoken, cftok, gd)
except urllib.error.HTTPError as e:
return f"ERROR {e.code}: {e.read().decode()[:120]}"
except Exception as e: # noqa: BLE001
return f"ERROR: {e}"
results = {}
for d in targets:
results[d] = _do(d)
print()
# pass 2: retry domains whose TXT was written but hadn't propagated in time
retry = [d for d, r in results.items() if "VERIFY FAILED" in str(r)]
if retry:
print(f"--- pass 2: retrying {len(retry)} propagation-pending domain(s) ---")
time.sleep(25)
for d in retry:
results[d] = _do(d)
print()
granted = [d for d, r in results.items() if str(r).startswith("GRANTED")]
nohost = [d for d, r in results.items() if "SKIP" in str(r)]
failed = {d: r for d, r in results.items()
if not str(r).startswith("GRANTED") and "SKIP" not in str(r)}
print("=== SUMMARY ===")
print(f" GRANTED : {len(granted)}")
print(f" no DNS host : {len(nohost)} (domain not on our Cloudflare/GoDaddy)")
print(f" failed/other : {len(failed)}")
if granted:
print("\n newly granted:")
for d in sorted(granted):
print(f" ✓ {d}")
if failed:
print("\n failed (verify/token/other — likely wrong domain guess or foreign DNS):")
for d, r in sorted(failed.items()):
print(f" ✗ {d:32} {str(r)[:80]}")
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()