Cloudflare Has a Free API: Here's How to Use It for DNS and CDN Automation

#cloudflare #dns #api #devops

Cloudflare's API gives you full control over DNS records, firewall rules, caching, Workers, and analytics — all on the free plan. You can automate everything you'd normally do in the dashboard.

Why Use the Cloudflare API?

Automate DNS record management across domains
Purge cache programmatically after deployments
Manage firewall rules and rate limiting
Deploy Cloudflare Workers via API
Monitor traffic analytics and security events

Getting Started

Get your API token from dash.cloudflare.com > My Profile > API Tokens:

export CF_TOKEN="your-api-token"

# List zones (domains)
curl -s -H "Authorization: Bearer $CF_TOKEN" \
  "https://api.cloudflare.com/client/v4/zones" | jq '.result[] | {name: .name, id: .id, status: .status}'

# List DNS records
curl -s -H "Authorization: Bearer $CF_TOKEN" \
  "https://api.cloudflare.com/client/v4/zones/ZONE_ID/dns_records" | jq '.result[] | {type: .type, name: .name, content: .content, proxied: .proxied}'

Python Client

import requests

class CloudflareClient:
    def __init__(self, token):
        self.url = "https://api.cloudflare.com/client/v4"
        self.headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}

    def list_zones(self):
        resp = requests.get(f"{self.url}/zones", headers=self.headers)
        return resp.json()['result']

    def get_zone_id(self, domain):
        resp = requests.get(f"{self.url}/zones", params={'name': domain}, headers=self.headers)
        results = resp.json()['result']
        return results[0]['id'] if results else None

    def list_dns_records(self, zone_id, record_type=None):
        params = {}
        if record_type:
            params['type'] = record_type
        resp = requests.get(f"{self.url}/zones/{zone_id}/dns_records", params=params, headers=self.headers)
        return resp.json()['result']

    def create_dns_record(self, zone_id, record_type, name, content, proxied=True, ttl=1):
        payload = {'type': record_type, 'name': name, 'content': content, 'proxied': proxied, 'ttl': ttl}
        resp = requests.post(f"{self.url}/zones/{zone_id}/dns_records", json=payload, headers=self.headers)
        return resp.json()

    def update_dns_record(self, zone_id, record_id, record_type, name, content, proxied=True):
        payload = {'type': record_type, 'name': name, 'content': content, 'proxied': proxied}
        resp = requests.put(f"{self.url}/zones/{zone_id}/dns_records/{record_id}", json=payload, headers=self.headers)
        return resp.json()

    def delete_dns_record(self, zone_id, record_id):
        resp = requests.delete(f"{self.url}/zones/{zone_id}/dns_records/{record_id}", headers=self.headers)
        return resp.json()

    def purge_cache(self, zone_id, urls=None):
        payload = {'purge_everything': True} if not urls else {'files': urls}
        resp = requests.post(f"{self.url}/zones/{zone_id}/purge_cache", json=payload, headers=self.headers)
        return resp.json()

# Usage
cf = CloudflareClient('your-api-token')

# List all domains
for zone in cf.list_zones():
    print(f"{zone['name']:30s} Status: {zone['status']}")

Dynamic DNS Updater

def update_dynamic_dns(cf, domain, subdomain):
    # Get current public IP
    my_ip = requests.get('https://api.ipify.org').text

    zone_id = cf.get_zone_id(domain)
    records = cf.list_dns_records(zone_id, record_type='A')

    full_name = f"{subdomain}.{domain}" if subdomain != '@' else domain

    for record in records:
        if record['name'] == full_name:
            if record['content'] != my_ip:
                cf.update_dns_record(zone_id, record['id'], 'A', full_name, my_ip, proxied=False)
                print(f"Updated {full_name}: {record['content']} -> {my_ip}")
            else:
                print(f"{full_name} already points to {my_ip}")
            return

    cf.create_dns_record(zone_id, 'A', full_name, my_ip, proxied=False)
    print(f"Created {full_name} -> {my_ip}")

update_dynamic_dns(cf, 'example.com', 'home')

Post-Deploy Cache Purge

def deploy_and_purge(cf, domain):
    zone_id = cf.get_zone_id(domain)

    # Purge specific URLs after deployment
    critical_urls = [
        f"https://{domain}/",
        f"https://{domain}/api/products",
        f"https://{domain}/static/bundle.js",
        f"https://{domain}/static/styles.css"
    ]

    result = cf.purge_cache(zone_id, urls=critical_urls)
    if result['success']:
        print(f"Cache purged for {len(critical_urls)} URLs")

    # Or purge everything
    # cf.purge_cache(zone_id)  # Nuclear option

deploy_and_purge(cf, 'example.com')

Firewall Rules

def manage_firewall(cf, zone_id):
    # Block bad bots
    resp = requests.post(
        f"{cf.url}/zones/{zone_id}/firewall/rules",
        json=[{
            'filter': {'expression': '(cf.client.bot) or (http.user_agent contains "BadBot")'},
            'action': 'block',
            'description': 'Block bad bots'
        }],
        headers=cf.headers
    )

    # Rate limit API endpoints
    resp = requests.post(
        f"{cf.url}/zones/{zone_id}/rate_limits",
        json={
            'match': {'request': {'url': '*example.com/api/*'}},
            'threshold': 100,
            'period': 60,
            'action': {'mode': 'simulate', 'response': {'content_type': 'application/json', 'body': '{"error": "rate limited"}'}}
        },
        headers=cf.headers
    )
    print("Firewall rules configured")

Analytics

# Get traffic analytics
curl -s -H "Authorization: Bearer $CF_TOKEN" \
  "https://api.cloudflare.com/client/v4/zones/ZONE_ID/analytics/dashboard?since=-1440" \
  | jq '.result.totals | {requests: .requests.all, bandwidth: .bandwidth.all, threats: .threats.all, pageviews: .pageviews.all}'

Real-World Use Case

A SaaS company managed 50+ domains on Cloudflare. They built a DNS automation system that syncs DNS records from a YAML config file in Git. Every merge to main triggers a CI pipeline that updates Cloudflare DNS via API — infrastructure as code for DNS. DNS changes went from "submit a ticket and wait" to "merge a PR and it's live in 30 seconds."