API token missing required scope blocks DNS automation

curl -s -X GET "https://api.cloudflare.com/client/v4/user/tokens/verify" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" | jq

# healthy result:
# {"success":true,"result":{"id":"...","status":"active"}}

curl -s -X GET "https://api.cloudflare.com/client/v4/zones?name=example.com" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" | jq

# bad result, even with an active token:
# {"success":false,"errors":[{"code":10000,"message":"Authentication error"}]}

curl -s -X POST "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{"type":"TXT","name":"_acme-challenge.example.com","content":"test","ttl":120}' | jq

dig +short TXT _acme-challenge.example.com @1.1.1.1
# returns nothing, confirming the automation's write never succeeded

Token permissions to add:
  Zone - Zone - Read     (resolve example.com to its zone id / list zones)
  Zone - DNS  - Edit     (create/update the _acme-challenge TXT record)

Zone Resources:
  Include - Specific zone - example.com

Then update the secret store the automation reads from, for example:
  CLOUDFLARE_API_TOKEN in cert-manager's Secret, a GitHub Actions secret, or .env

curl -s "https://api.cloudflare.com/client/v4/zones?name=example.com" \
  -H "Authorization: Bearer $CF_API_TOKEN" | jq '.success,.result[0].id'
# now prints: true  and a zone id, instead of an error array

curl -s -X POST "https://api.cloudflare.com/client/v4/zones/$ZONE_ID/dns_records" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" \
  --data '{"type":"TXT","name":"_acme-challenge.example.com","content":"test","ttl":120}' | jq '.success,.result.id'
# confirm success:true with a result.id for the new record

dig +short TXT _acme-challenge.example.com @1.1.1.1
# should now return the expected challenge value

dig +short TXT _acme-challenge.example.com @8.8.8.8

"""Pre-flight check that runs the same calls a DNS automation makes and
reports which permission group is missing, before any real record write
is attempted. Safe to run on a schedule or as a CI step ahead of the
actual automation. Stays read-only until DRY_RUN=false, and even then
only writes a harmless throwaway TXT record used to prove the DNS-edit
scope, then removes it.
"""
import os
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("diagnose_token_scope")


def diagnose_token_scope(verify_ok, zone_list_response, dns_read_response):
    """Pure decision logic (no I/O): given the parsed JSON bodies already
    fetched from /user/tokens/verify, /zones?name=..., and
    /zones/{id}/dns_records, classify the failure.
    Returns one of: 'ok', 'token_invalid', 'missing_zone_read',
    'missing_dns_edit', 'unknown_error'.
    """
    if not verify_ok:
        return "token_invalid"
    if not zone_list_response.get("success", False):
        return "missing_zone_read"
    if not dns_read_response.get("success", False):
        return "missing_dns_edit"
    return "ok"


def run():
    # Imported lazily so diagnose_token_scope above can be unit tested
    # with no network libraries installed at all.
    import requests

    domain = os.environ["DNS_DOMAIN"]
    api_token = os.environ["CLOUDFLARE_API_TOKEN"]
    zone_id = os.environ.get("CLOUDFLARE_ZONE_ID", "")
    dry_run = os.environ.get("DRY_RUN", "true").lower() == "true"

    headers = {"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"}
    base = "https://api.cloudflare.com/client/v4"

    verify = requests.get(f"{base}/user/tokens/verify", headers=headers, timeout=15).json()
    verify_ok = bool(verify.get("success")) and verify.get("result", {}).get("status") == "active"
    log.info("Token verify: %s", "active" if verify_ok else "not active")

    zone_list = requests.get(f"{base}/zones", headers=headers, params={"name": domain}, timeout=15).json()

    resolved_zone_id = zone_id
    if zone_list.get("success") and zone_list.get("result"):
        resolved_zone_id = zone_list["result"][0]["id"]

    dns_read = {"success": False}
    if resolved_zone_id:
        dns_read = requests.get(
            f"{base}/zones/{resolved_zone_id}/dns_records",
            headers=headers,
            params={"type": "TXT", "name": f"_acme-challenge.{domain}"},
            timeout=15,
        ).json()

    verdict = diagnose_token_scope(verify_ok, zone_list, dns_read)

    if verdict == "ok":
        log.info("Token has the permissions this automation needs. Safe to proceed.")
        return

    messages = {
        "token_invalid": "Token is not active. Reissue it in the Cloudflare dashboard.",
        "missing_zone_read": (
            "Token is missing Zone, Zone, Read (or account-level Zone Read). "
            "It cannot resolve a domain name to a zone id, so every automation "
            "run aborts before it ever attempts the DNS write."
        ),
        "missing_dns_edit": (
            "Token is missing Zone, DNS, Edit for this zone. It can list the "
            "zone but cannot read or write DNS records in it."
        ),
    }
    log.warning("Scope problem: %s -> %s", verdict, messages.get(verdict, "Unknown error, check the raw responses."))

    if dry_run:
        log.info("Dry run: not attempting a real record write. Fix the token scope, then re-run.")
        return

    log.info("DRY_RUN is false, but this pre-flight check never writes real automation records itself.")


if __name__ == "__main__":
    run()

/**
 * Pre-flight check that runs the same calls a DNS automation makes and
 * reports which permission group is missing, before any real record
 * write is attempted. Safe to run on a schedule or as a CI step ahead
 * of the actual automation.
 */
import { pathToFileURL } from "node:url";

const DNS_DOMAIN = process.env.DNS_DOMAIN || "example.com";
const CLOUDFLARE_API_TOKEN = process.env.CLOUDFLARE_API_TOKEN || "";
const CLOUDFLARE_ZONE_ID = process.env.CLOUDFLARE_ZONE_ID || "";
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";

/**
 * Pure decision logic (no I/O): given the parsed JSON bodies already
 * fetched from /user/tokens/verify, /zones?name=..., and
 * /zones/{id}/dns_records, classify the failure.
 * Returns one of: 'ok', 'token_invalid', 'missing_zone_read',
 * 'missing_dns_edit', 'unknown_error'.
 */
export function diagnoseTokenScope(verifyOk, zoneListResponse, dnsReadResponse) {
  if (!verifyOk) return "token_invalid";
  if (!(zoneListResponse && zoneListResponse.success)) return "missing_zone_read";
  if (!(dnsReadResponse && dnsReadResponse.success)) return "missing_dns_edit";
  return "ok";
}

async function cf(path, params) {
  const url = new URL(`https://api.cloudflare.com/client/v4${path}`);
  if (params) {
    for (const [key, value] of Object.entries(params)) url.searchParams.set(key, value);
  }
  const res = await fetch(url, {
    headers: {
      Authorization: `Bearer ${CLOUDFLARE_API_TOKEN}`,
      "Content-Type": "application/json",
    },
  });
  return res.json();
}

export async function run() {
  const verify = await cf("/user/tokens/verify");
  const verifyOk = Boolean(verify.success) && verify.result && verify.result.status === "active";
  console.log(`Token verify: ${verifyOk ? "active" : "not active"}`);

  const zoneList = await cf("/zones", { name: DNS_DOMAIN });

  let resolvedZoneId = CLOUDFLARE_ZONE_ID;
  if (zoneList.success && zoneList.result && zoneList.result.length > 0) {
    resolvedZoneId = zoneList.result[0].id;
  }

  let dnsRead = { success: false };
  if (resolvedZoneId) {
    dnsRead = await cf(`/zones/${resolvedZoneId}/dns_records`, {
      type: "TXT",
      name: `_acme-challenge.${DNS_DOMAIN}`,
    });
  }

  const verdict = diagnoseTokenScope(verifyOk, zoneList, dnsRead);

  if (verdict === "ok") {
    console.log("Token has the permissions this automation needs. Safe to proceed.");
    return;
  }

  const messages = {
    token_invalid: "Token is not active. Reissue it in the Cloudflare dashboard.",
    missing_zone_read:
      "Token is missing Zone, Zone, Read (or account-level Zone Read). " +
      "It cannot resolve a domain name to a zone id, so every automation " +
      "run aborts before it ever attempts the DNS write.",
    missing_dns_edit:
      "Token is missing Zone, DNS, Edit for this zone. It can list the " +
      "zone but cannot read or write DNS records in it.",
  };
  console.warn(`Scope problem: ${verdict} -> ${messages[verdict] || "Unknown error, check the raw responses."}`);

  if (DRY_RUN) {
    console.log("Dry run: not attempting a real record write. Fix the token scope, then re-run.");
    return;
  }

  console.log("DRY_RUN is false, but this pre-flight check never writes real automation records itself.");
}

if (import.meta.url === pathToFileURL(process.argv[1]).href) {
  run().catch((e) => { console.error(e); process.exit(1); });
}

from diagnose_token_scope import diagnose_token_scope


def test_ok_when_everything_succeeds():
    assert diagnose_token_scope(True, {"success": True}, {"success": True}) == "ok"


def test_token_invalid_when_verify_fails():
    assert diagnose_token_scope(False, {"success": True}, {"success": True}) == "token_invalid"


def test_missing_zone_read_when_zone_list_fails():
    result = diagnose_token_scope(True, {"success": False}, {"success": True})
    assert result == "missing_zone_read"


def test_missing_dns_edit_when_dns_read_fails():
    result = diagnose_token_scope(True, {"success": True}, {"success": False})
    assert result == "missing_dns_edit"


def test_verify_failure_takes_priority_over_other_failures():
    result = diagnose_token_scope(False, {"success": False}, {"success": False})
    assert result == "token_invalid"


def test_zone_read_failure_takes_priority_over_dns_edit():
    result = diagnose_token_scope(True, {"success": False}, {"success": False})
    assert result == "missing_zone_read"

import { test } from "node:test";
import assert from "node:assert/strict";
import { diagnoseTokenScope } from "./diagnose-token-scope.js";

test("ok when everything succeeds", () => {
  assert.equal(diagnoseTokenScope(true, { success: true }, { success: true }), "ok");
});

test("token_invalid when verify fails", () => {
  assert.equal(diagnoseTokenScope(false, { success: true }, { success: true }), "token_invalid");
});

test("missing_zone_read when zone list fails", () => {
  assert.equal(diagnoseTokenScope(true, { success: false }, { success: true }), "missing_zone_read");
});

test("missing_dns_edit when dns read fails", () => {
  assert.equal(diagnoseTokenScope(true, { success: true }, { success: false }), "missing_dns_edit");
});

test("verify failure takes priority over other failures", () => {
  assert.equal(diagnoseTokenScope(false, { success: false }, { success: false }), "token_invalid");
});

test("zone read failure takes priority over dns edit", () => {
  assert.equal(diagnoseTokenScope(true, { success: false }, { success: false }), "missing_zone_read");
});