← back to Cf Dns Migration
stage.py
223 lines
#!/usr/bin/env python3
"""
Phase-2 CF-DNS staging engine.
For each domain: read its LIVE records from GoDaddy, ensure a Cloudflare zone
exists, and replicate the records into that zone idempotently. NOTHING cuts over
here — the domain keeps its GoDaddy nameservers until a human does the NS-swap,
so a staged CF zone is dormant/safe. The point is that when the swap happens the
zone is already a faithful copy, so there's zero downtime.
python3 stage.py --dry-run homesonspec.com # show what would happen
python3 stage.py homesonspec.com foo.com # stage specific domains
python3 stage.py --from worklist.csv --limit 5 # stage first 5 from worklist
python3 stage.py --from worklist.csv # stage the whole worklist
Records are replicated DNS-only (grey cloud) to mirror GoDaddy exactly; enable
the CF proxy per-site later if wanted. NS/SOA and GoDaddy-proprietary
`_domainconnect` records are never copied.
"""
import json, os, sys, time, urllib.request, urllib.error
ENV = os.path.expanduser("~/Projects/secrets-manager/.env")
def load_env(path):
e = {}
for line in open(path, encoding="utf-8", errors="ignore"):
line = line.strip()
if line and not line.startswith("#") and "=" in line:
k, v = line.split("=", 1)
e[k.strip()] = v.strip().strip('"').strip("'")
return e
E = load_env(ENV)
GD_KEY = E.get("GODADDY_API_KEY"); GD_SEC = E.get("GODADDY_API_SECRET")
GD_PAT = E.get("GODADDY_PAT")
# Prefer a Bearer PAT when present (TK-10 key rotation); else legacy sso-key.
GD_AUTH = f"Bearer {GD_PAT}" if GD_PAT else f"sso-key {GD_KEY}:{GD_SEC}"
CF_TOKEN = E.get("CLOUDFLARE_API_TOKEN") or E.get("CF_API_TOKEN") or E.get("CLOUDFLARE_DNS_TOKEN")
COPY_TYPES = {"A", "AAAA", "CNAME", "MX", "TXT", "SRV", "CAA", "NS_SUB"} # zone-apex NS handled by CF
SKIP_NAMES = {"_domainconnect"}
def req(url, method="GET", headers=None, body=None):
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(url, data=data, method=method, headers=headers or {})
try:
with urllib.request.urlopen(r, timeout=30) as resp:
return resp.status, json.loads(resp.read().decode())
except urllib.error.HTTPError as ex:
try: return ex.code, json.loads(ex.read().decode())
except Exception: return ex.code, {"error": str(ex)}
except Exception as ex:
return 0, {"error": str(ex)}
# ---- GoDaddy ----
def gd_records(domain, tries=5):
"""Read with backoff — GoDaddy throttles ~60 req/min and returns 429; the
earlier run's 135 'no records' were false empties from being throttled."""
for i in range(tries):
st, d = req(f"https://api.godaddy.com/v1/domains/{domain}/records",
headers={"Authorization": GD_AUTH})
if st == 200 and isinstance(d, list):
return d
if st == 429 or st == 0 or (isinstance(d, dict) and d.get("code") == "TOO_MANY_REQUESTS"):
wait = 0
if isinstance(d, dict):
try: wait = int(d.get("retryAfterSec", 0))
except Exception: wait = 0
time.sleep(max(wait, 2 * (i + 1)))
continue
if isinstance(d, list):
return d
time.sleep(1.5)
return []
def is_ipv4(s):
p = str(s).split(".")
return len(p) == 4 and all(x.isdigit() and 0 <= int(x) <= 255 for x in p)
# ---- Cloudflare ----
CF_H = {"Authorization": f"Bearer {CF_TOKEN}", "Content-Type": "application/json"}
def cf_zone(domain):
_, d = req(f"https://api.cloudflare.com/client/v4/zones?name={domain}", headers=CF_H)
res = d.get("result") or []
return res[0] if res else None
def cf_create_zone(domain):
_, d = req("https://api.cloudflare.com/client/v4/zones", "POST", CF_H,
{"name": domain, "type": "full"})
return d.get("result") if d.get("success") else None
def cf_records(zid):
out, page = [], 1
while True:
_, d = req(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records?per_page=100&page={page}", headers=CF_H)
res = d.get("result") or []
out += res
if len(res) < 100: break
page += 1
return out
def cf_add(zid, rec):
return req(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records", "POST", CF_H, rec)
def gd_to_cf(domain, r):
"""Map a GoDaddy record to a Cloudflare dns_records payload, or None to skip."""
t = r.get("type")
name = r.get("name", "@")
# SRV skipped — GoDaddy's are almost always autodiscover/client defaults and
# need weight/port fields CF requires; not worth carrying.
if t in ("NS", "SOA", "SRV") or name in SKIP_NAMES or t not in ("A","AAAA","CNAME","MX","TXT","CAA"):
return None
fqdn = domain if name == "@" else f"{name}.{domain}"
data = r.get("data", "")
if t == "CNAME" and data in ("@", ""):
data = domain
# Guard against GoDaddy parked/forwarding A records whose data isn't an IP.
if t == "A" and not is_ipv4(data): return None
if t == "AAAA" and ":" not in str(data): return None
rec = {"type": t, "name": fqdn, "content": data, "ttl": max(int(r.get("ttl", 600) or 600), 60), "proxied": False}
if t == "MX":
rec["priority"] = int(r.get("priority", 10) or 10)
rec.pop("proxied", None)
if t in ("TXT", "SRV", "CAA"):
rec.pop("proxied", None)
return rec
def rec_key(t, name, content):
return (t, name.rstrip("."), str(content).strip().strip('"'))
def stage(domain, dry=False):
gd = gd_records(domain)
if not gd:
return {"domain": domain, "status": "no-godaddy-records", "added": 0, "existing": 0, "ns": ""}
zone = cf_zone(domain)
created = False
if not zone:
if dry:
zone = {"id": "(would-create)", "name_servers": ["(assigned-on-create)"]}
else:
zone = cf_create_zone(domain)
created = True
if not zone:
return {"domain": domain, "status": "cf-zone-create-failed", "added": 0, "existing": 0, "ns": ""}
zid = zone["id"]
existing = [] if zid == "(would-create)" else cf_records(zid)
have = {rec_key(r["type"], r["name"], r.get("content", "")) for r in existing}
# Structural state so we never create an invalid zone when CF is already curated:
types_at = {} # name -> {types present}
dmarc_at, spf_at = set(), set() # names that already have a DMARC / SPF TXT
for r in existing:
nm = r["name"].rstrip(".")
types_at.setdefault(nm, set()).add(r["type"])
c = str(r.get("content", "")).strip('"').lower()
if r["type"] == "TXT" and c.startswith("v=dmarc1"): dmarc_at.add(nm)
if r["type"] == "TXT" and c.startswith("v=spf1"): spf_at.add(nm)
added = kept = 0
for r in gd:
payload = gd_to_cf(domain, r)
if not payload: continue
nm = payload["name"].rstrip("."); c = str(payload["content"]).strip('"').lower()
here = types_at.get(nm, set())
# CNAME must be the ONLY record at a name; and DMARC/SPF are singletons.
if payload["type"] == "CNAME" and here: kept += 1; continue
if payload["type"] != "CNAME" and "CNAME" in here: kept += 1; continue
if payload["type"] == "TXT" and c.startswith("v=dmarc1") and nm in dmarc_at: kept += 1; continue
if payload["type"] == "TXT" and c.startswith("v=spf1") and nm in spf_at: kept += 1; continue
k = rec_key(payload["type"], payload["name"], payload["content"])
if k in have:
kept += 1; continue
# commit to adding — update running structural state so later records see it
types_at.setdefault(nm, set()).add(payload["type"]); have.add(k)
if payload["type"] == "TXT" and c.startswith("v=dmarc1"): dmarc_at.add(nm)
if payload["type"] == "TXT" and c.startswith("v=spf1"): spf_at.add(nm)
if dry:
print(f" + {payload['type']:6} {payload['name']:34} -> {str(payload['content'])[:46]}")
added += 1
else:
code, resp = cf_add(zid, payload)
if resp.get("success"): added += 1
else:
# duplicate/exists errors are fine (idempotent); log others
errs = resp.get("errors", [])
if any("already exists" in str(e).lower() for e in errs): kept += 1
else: print(f" ! {payload['type']} {payload['name']}: {errs}")
time.sleep(0.05)
ns = ",".join(zone.get("name_servers", []) or [])
return {"domain": domain, "status": ("staged" + ("+created" if created else "")) if not dry else "dry",
"added": added, "existing": kept, "ns": ns, "zone_id": zid}
def main():
args = sys.argv[1:]
dry = "--dry-run" in args; args = [a for a in args if a != "--dry-run"]
limit = None; domains = []
i = 0
while i < len(args):
if args[i] == "--limit": limit = int(args[i+1]); i += 2
elif args[i] == "--from":
import csv
for row in csv.DictReader(open(args[i+1])): domains.append(row["domain"])
i += 2
else: domains.append(args[i]); i += 1
if limit: domains = domains[:limit]
if not domains:
print("usage: stage.py [--dry-run] [--from worklist.csv] [--limit N] [domain ...]"); sys.exit(1)
if not (GD_KEY and GD_SEC and CF_TOKEN):
print("missing creds (GODADDY_API_KEY/SECRET, CLOUDFLARE_API_TOKEN)"); sys.exit(1)
print(f"{'DRY-RUN' if dry else 'STAGING'} {len(domains)} domain(s)\n")
results = []
for d in domains:
print(f"── {d}", flush=True)
res = stage(d, dry)
results.append(res)
print(f" {res['status']}: +{res['added']} new, {res['existing']} already present"
+ (f" | NS: {res['ns']}" if res['ns'] else ""), flush=True)
time.sleep(0.7) # pace GoDaddy reads (~60/min cap)
# emit the NS-swap worklist
out = "staged-nameservers.csv"
if not dry:
import csv
with open(out, "w", newline="") as f:
w = csv.writer(f); w.writerow(["domain", "status", "added", "cf_nameservers"])
for r in results: w.writerow([r["domain"], r["status"], r["added"], r["ns"]])
print(f"\nNS-swap worklist → {out} (set each domain's GoDaddy nameservers to its cf_nameservers)")
if __name__ == "__main__":
main()