Recipe: Google Workspace on a subdomain#

curl https://api.maildns.io/api3.json \
  --data-urlencode key=KEY \
  --data-urlencode command=set_dns2 \
  --data-urlencode domain=advisor.example.com \
  --data-urlencode add_dns_to_current_setting=1 \
  --data-urlencode main_record_type0=txt \
  --data-urlencode main_record0=google-site-verification=3J9tq0iV8pTt1yW5zKc2mE7aR4bN6hD0sF1gL8xQ

curl https://api.maildns.io/api3.json \
  --data-urlencode key=KEY \
  --data-urlencode command=dns_status \
  --data-urlencode domain=advisor.example.com \
  --data-urlencode subdomain=@ \
  --data-urlencode record_type=txt \
  --data-urlencode value=google-site-verification=3J9tq0iV8pTt1yW5zKc2mE7aR4bN6hD0sF1gL8xQ

curl https://api.maildns.io/api3.json \
  --data-urlencode key=KEY \
  --data-urlencode command=set_dns2 \
  --data-urlencode domain=advisor.example.com \
  --data-urlencode add_dns_to_current_setting=1 \
  --data-urlencode main_record_type0=mx \
  --data-urlencode main_record0=smtp.google.com \
  --data-urlencode main_recordx0=1 \
  --data-urlencode main_record_type1=txt \
  --data-urlencode 'main_record1=v=spf1 include:_spf.google.com ~all' \
  --data-urlencode subdomain0=_dmarc \
  --data-urlencode sub_record_type0=txt \
  --data-urlencode 'sub_record0=v=DMARC1; p=quarantine'

curl https://api.maildns.io/api3.json \
  --data-urlencode key=KEY \
  --data-urlencode command=set_dns2 \
  --data-urlencode domain=advisor.example.com \
  --data-urlencode add_dns_to_current_setting=1 \
  --data-urlencode subdomain0=google._domainkey \
  --data-urlencode sub_record_type0=txt \
  --data-urlencode 'sub_record0=v=DKIM1; k=rsa; p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...IDAQAB'

pip install requests
export MAILDNS_KEY='KEY'

python3 workspace_dns.py verify tokens.csv   # host,token        -> TXT @ google-site-verification=...
# ... press Verify for each host in Google ...
python3 workspace_dns.py mail hosts.txt      # one host per line -> MX, SPF, DMARC
python3 workspace_dns.py dkim dkim.csv       # host,dkim_value   -> TXT google._domainkey
# ... press Start authentication for each host ...

advisor.example.com,google-site-verification=3J9tq0iV8pTt1yW5zKc2mE7aR4bN6hD0sF1gL8xQ
billing.example.com,google-site-verification=Qm2pX7rT0vW4yA9cE1gH3jK5lN8oS6uB2dF4hJ6kL

#!/usr/bin/env python3
"""Google Workspace DNS for many hosts, through MailDNS (Dynadot-style API).

    export MAILDNS_KEY=KEY
    python3 workspace_dns.py verify tokens.csv   # host,token       -> TXT @ google-site-verification=...
    python3 workspace_dns.py mail hosts.txt      # one host a line  -> MX @, SPF @, DMARC _dmarc
    python3 workspace_dns.py dkim dkim.csv       # host,dkim_value  -> TXT google._domainkey

Every step only adds records (records already there are skipped), so re-running is safe.
Every step waits until each new record is live on every authoritative nameserver.
"""
import csv
import os
import sys
import time

import requests

API = "https://api.maildns.io/api3.json"
KEY = os.environ["MAILDNS_KEY"]
MX = "smtp.google.com"
SPF = "v=spf1 include:_spf.google.com ~all"
DMARC = "v=DMARC1; p=quarantine"      # add "; rua=mailto:..." if you collect reports
POLL_EVERY, POLL_FOR = 15, 900        # seconds

session = requests.Session()


def call(command, **params):
    """One Dynadot-style call. Returns the inner reply, retries rate limits and busy roots, raises on errors."""
    params.update(key=KEY, command=command)
    for attempt in range(8):
        r = session.post(API, data=params, timeout=90)
        r.raise_for_status()
        reply = next(iter(r.json().values()))
        if str(reply.get("ResponseCode")) == "0":
            return reply
        error = reply.get("Error", "unknown error")
        if error.startswith(("too many requests", "another change", "this domain is temporarily")):
            time.sleep(float(r.headers.get("Retry-After", 0)) + 2 * (attempt + 1))
            continue
        raise RuntimeError(f"{command} {params.get('domain', '')}: {error}")
    raise RuntimeError(f"{command} {params.get('domain', '')}: still refused after 8 tries")


def add(host, records):
    """Add records at or under one host. records: [(name, type, value, mx_priority)], name "@" = the host."""
    params = {"domain": host, "add_dns_to_current_setting": "1"}
    m = s = 0
    for name, rtype, value, prio in records:
        if name == "@":
            params[f"main_record_type{m}"], params[f"main_record{m}"] = rtype, value
            if prio is not None:
                params[f"main_recordx{m}"] = str(prio)
            m += 1
        else:
            params[f"subdomain{s}"], params[f"sub_record_type{s}"], params[f"sub_record{s}"] = name, rtype, value
            if prio is not None:
                params[f"sub_recordx{s}"] = str(prio)
            s += 1
    call("set_dns2", **params)


def records_of(host):
    """[(name, TYPE, value)] for everything at ("@") and under the host."""
    ns = call("get_dns", domain=host)["GetDns"]["NameServerSettings"]
    return ([("@", r["RecordType"].upper(), r["Value"]) for r in ns.get("MainDomains", [])]
            + [(r["Subhost"], r["RecordType"].upper(), r["Value"]) for r in ns.get("SubDomains", [])])


def is_live(host, name, rtype, value):
    reply = call("dns_status", domain=host, subdomain=name, record_type=rtype, value=value)
    return bool(reply["Records"]) and all(rec["Live"] == "yes" for rec in reply["Records"])


def wait_live(checks):
    """checks: [(host, name, type, value)]. Polls until all are live; returns the ones that are not."""
    pending, deadline = list(checks), time.time() + POLL_FOR
    while pending:
        pending = [c for c in pending if not is_live(*c)]
        if not pending or time.time() > deadline:
            break
        print(f"  {len(pending)} record(s) not live yet, checking again in {POLL_EVERY}s")
        time.sleep(POLL_EVERY)
    return pending


def finish(pending, next_step):
    if pending:
        print("Not live yet (run the same command again later; nothing will be added twice):")
        for host, name, rtype, _ in pending:
            print(f"  {host}  {name}  {rtype}")
        sys.exit(1)
    print(f"All live. {next_step}")


def rows(path):
    """CSV rows (host, value), skipping blank lines and a header row."""
    with open(path, newline="") as f:
        for row in csv.reader(f):
            if len(row) >= 2 and row[0].strip() and row[0].strip().lower() != "host":
                yield row[0].strip().lower().rstrip("."), row[1].strip()


def verify(path):
    checks = []
    for host, token in rows(path):
        value = token if token.startswith("google-site-verification=") else f"google-site-verification={token}"
        add(host, [("@", "txt", value, None)])
        checks.append((host, "@", "txt", value))
        print(f"added  {host}  TXT @  {value}")
    finish(wait_live(checks), "Press Verify for each host in Google, then run: mail")


def mail(path):
    checks = []
    with open(path) as f:
        hosts = [h.strip().lower().rstrip(".") for h in f if h.strip()]
    for host in hosts:
        have = records_of(host)
        spf = [v for n, t, v in have if n == "@" and t == "TXT" and v.startswith("v=spf1")]
        dmarc = [v for n, t, v in have if n == "_dmarc" and t == "TXT" and v.startswith("v=DMARC1")]
        wanted = [("@", "mx", MX, 1)]
        if spf and spf != [SPF]:
            print(f"skip SPF on {host}: it already has {spf[0]!r} (merge the two by hand)")
        else:
            wanted.append(("@", "txt", SPF, None))
        if dmarc and dmarc != [DMARC]:
            print(f"skip DMARC on {host}: it already has {dmarc[0]!r}")
        else:
            wanted.append(("_dmarc", "txt", DMARC, None))
        add(host, wanted)
        checks += [(host, name, rtype, value) for name, rtype, value, _ in wanted]
        print(f"added  {host}  " + ", ".join(f"{t.upper()} {n}" for n, t, _, _ in wanted))
    finish(wait_live(checks), "Mail records are live. Generate DKIM keys in Google, then run: dkim")


def dkim(path):
    checks = []
    for host, value in rows(path):
        add(host, [("google._domainkey", "txt", value, None)])
        checks.append((host, "google._domainkey", "txt", value))
        print(f"added  {host}  TXT google._domainkey")
    finish(wait_live(checks), "Press Start authentication for each host in Google.")


if __name__ == "__main__":
    steps = {"verify": verify, "mail": mail, "dkim": dkim}
    if len(sys.argv) != 3 or sys.argv[1] not in steps:
        sys.exit(__doc__)
    steps[sys.argv[1]](sys.argv[2])

# 1. Verification TXT
curl -X PUT https://api.maildns.io/api/v1/dns/records/advisor.example.com \
  -H 'X-API-Key: KEY' -H 'X-API-Secret: SECRET' -H 'Content-Type: application/json' \
  -d '{"force": false, "items": [{"type": "TXT", "name": "@", "ttl": 3600,
       "value": "google-site-verification=3J9tq0iV8pTt1yW5zKc2mE7aR4bN6hD0sF1gL8xQ"}]}'

# 2, 3, 5. MX, SPF, DMARC
curl -X PUT https://api.maildns.io/api/v1/dns/records/advisor.example.com \
  -H 'X-API-Key: KEY' -H 'X-API-Secret: SECRET' -H 'Content-Type: application/json' \
  -d '{"force": false, "items": [
        {"type": "MX", "name": "@", "exchange": "smtp.google.com", "preference": 1},
        {"type": "TXT", "name": "@", "value": "v=spf1 include:_spf.google.com ~all"},
        {"type": "TXT", "name": "_dmarc", "value": "v=DMARC1; p=quarantine"}]}'

# 4. DKIM
curl -X PUT https://api.maildns.io/api/v1/dns/records/advisor.example.com \
  -H 'X-API-Key: KEY' -H 'X-API-Secret: SECRET' -H 'Content-Type: application/json' \
  -d '{"force": false, "items": [{"type": "TXT", "name": "google._domainkey",
       "value": "v=DKIM1; k=rsa; p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...IDAQAB"}]}'

# Is a record live?
curl -G https://api.maildns.io/api/v1/dns/status/advisor.example.com \
  -H 'X-API-Key: KEY' -H 'X-API-Secret: SECRET' \
  --data-urlencode name=google._domainkey --data-urlencode type=TXT \
  --data-urlencode 'value=v=DKIM1; k=rsa; p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...IDAQAB'