`dns_point.py`

119 lines
#!/usr/bin/env python3
"""Point domains' apex(+www) A records at Kamatera (45.61.58.125), DNS-only.
Handles Cloudflare (API) + GoDaddy (API). Skips CF-Worker-bound apexes.
Idempotent + resumable. Prints a READY list (domains now pointing at Kamatera,
safe to cert) and a SKIP list. Usage: python3 dns_point.py d1.com d2.com ..."""
import json
import os
import pathlib
import subprocess
import sys
import urllib.error
import urllib.request

IP = "45.61.58.125"
ENV = pathlib.Path.home() / "Projects" / "secrets-manager" / ".env"


def ev(*ks):
    for k in ks:
        if os.environ.get(k):
            return os.environ[k]
    for line in ENV.read_text().splitlines():
        line = line.strip()
        for k in ks:
            if line.startswith(k + "="):
                return line.split("=", 1)[1].strip().strip('"').strip("'")
    return ""


CF = ev("CLOUDFLARE_API_TOKEN", "CF_API_TOKEN")
GK, GS = ev("GODADDY_API_KEY"), ev("GODADDY_API_SECRET")


def cf(url, method="GET", payload=None):
    req = urllib.request.Request(url, method=method,
        data=json.dumps(payload).encode() if payload else None,
        headers={"Authorization": f"Bearer {CF}", "Content-Type": "application/json"})
    try:
        return json.loads(urllib.request.urlopen(req, timeout=30).read())
    except urllib.error.HTTPError as e:
        return {"__err__": e.code, "__body__": e.read().decode()[:200]}


def ns_host(d):
    try:
        out = subprocess.run(["dig", "+short", "NS", d], capture_output=True, text=True, timeout=15).stdout.lower()
    except Exception:
        out = ""
    if "cloudflare" in out:
        return "cf"
    if "domaincontrol" in out:
        return "godaddy"
    return "other"


def point_cf(d):
    z = cf(f"https://api.cloudflare.com/client/v4/zones?name={d}").get("result")
    if not z:
        return "SKIP no-cf-zone"
    zid = z[0]["id"]
    for name in (d, f"www.{d}"):
        recs = cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records?name={name}").get("result", [])
        # remove parking placeholders / conflicting CNAME so an A can live here
        for r in recs:
            if (r["type"] == "AAAA" and r["content"] in ("100::", "::")) or r["type"] == "CNAME":
                cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records/{r['id']}", "DELETE")
        a = [r for r in recs if r["type"] == "A"]
        body = {"type": "A", "name": name, "content": IP, "ttl": 300, "proxied": False}
        if a:
            res = cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records/{a[0]['id']}", "PUT", body)
        else:
            res = cf(f"https://api.cloudflare.com/client/v4/zones/{zid}/dns_records", "POST", body)
        if res.get("__err__"):
            body_txt = res.get("__body__", "")
            if "Workers" in body_txt:
                return "SKIP cf-worker"
            if name == d:  # apex failure is fatal for this domain
                return f"ERR cf {res['__err__']}"
    return "READY"


def point_godaddy(d):
    if not (GK and GS):
        return "SKIP no-godaddy-creds"
    hdr = {"Authorization": f"sso-key {GK}:{GS}", "Content-Type": "application/json"}
    body = json.dumps([{"data": IP, "ttl": 600, "type": "A", "name": "@"}]).encode()
    req = urllib.request.Request(f"https://api.godaddy.com/v1/domains/{d}/records/A/%40",
                                 data=body, headers=hdr, method="PUT")
    try:
        urllib.request.urlopen(req, timeout=30).read()
    except urllib.error.HTTPError as e:
        return f"ERR gd {e.code}"
    # www (best-effort; ignore failure)
    try:
        wb = json.dumps([{"data": IP, "ttl": 600, "type": "A", "name": "www"}]).encode()
        urllib.request.urlopen(urllib.request.Request(
            f"https://api.godaddy.com/v1/domains/{d}/records/A/www", data=wb, headers=hdr, method="PUT"), timeout=30).read()
    except Exception:
        pass
    return "READY"


def main():
    doms = sys.argv[1:]
    ready, skip = [], []
    for d in doms:
        host = ns_host(d)
        r = point_cf(d) if host == "cf" else point_godaddy(d) if host == "godaddy" else "SKIP other-registrar"
        (ready if r == "READY" else skip).append(d)
        print(f"  {d:34} [{host}] {r}")
    print(f"\nREADY={len(ready)}  SKIP={len(skip)}")
    print("READY_LIST:" + " ".join(ready))
    if skip:
        print("SKIP_LIST:" + " ".join(skip))


if __name__ == "__main__":
    main()