#!/usr/bin/env python3 """ DNSai API Client — Sample Script ================================= A working example demonstrating how to connect to the DNSai REST API, authenticate, and perform common operations. Requirements: pip install requests Usage: 1. Replace YOUR_API_KEY_HERE with your actual API key 2. Run: python dnsai_api_client.py Get your API key at: https://app.dnsai.com/settings/?tab=apikeys API documentation: https://app.dnsai.com/api-setup/ """ import os import sys import time import requests # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- # Replace with your API key, or set the DNSAI_API_KEY environment variable API_KEY = os.environ.get("DNSAI_API_KEY", "YOUR_API_KEY_HERE") # Base URL for the DNSai API BASE_URL = "https://app.dnsai.com/api/v1" # Maximum retries for rate-limited requests MAX_RETRIES = 5 # --------------------------------------------------------------------------- # API Client # --------------------------------------------------------------------------- class DNSaiClient: """Simple client for the DNSai REST API.""" def __init__(self, api_key, base_url=BASE_URL): self.base_url = base_url.rstrip("/") self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }) def _request(self, method, path, **kwargs): """Make an API request with automatic retry on rate limiting. Handles 429 (Too Many Requests) responses with exponential backoff using the Retry-After header. """ url = f"{self.base_url}{path}" for attempt in range(MAX_RETRIES): try: response = self.session.request(method, url, **kwargs) except requests.ConnectionError: print(f" Connection error. Retrying in {2 ** attempt}s...") time.sleep(2 ** attempt) continue except requests.Timeout: print(f" Request timed out. Retrying in {2 ** attempt}s...") time.sleep(2 ** attempt) continue # Success if response.ok: return response.json() # Rate limited — wait and retry if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 60)) print(f" Rate limited. Waiting {retry_after}s " f"(attempt {attempt + 1}/{MAX_RETRIES})...") time.sleep(retry_after) continue # Authentication error if response.status_code == 401: print("Error: Invalid or expired API key.") print("Get a new key at: https://app.dnsai.com/settings/?tab=apikeys") sys.exit(1) # Plan access error if response.status_code == 403: print("Error: Your plan does not include API access.") print("Upgrade at: https://app.dnsai.com/settings/?tab=billing") sys.exit(1) # Other errors try: error_data = response.json() errors = error_data.get("errors", []) detail = errors[0]["detail"] if errors else response.text except (ValueError, KeyError, IndexError): detail = response.text print(f"Error {response.status_code}: {detail}") return None print("Error: Max retries exceeded.") return None def get(self, path, params=None): """GET request.""" return self._request("GET", path, params=params) def post(self, path, data=None): """POST request.""" return self._request("POST", path, json=data) def delete(self, path): """DELETE request.""" return self._request("DELETE", path) def fetch_all(self, path, params=None): """Fetch all results from a paginated endpoint. Uses limit/offset pagination to iterate through the full result set. Default limit is 5000 (the maximum) for efficient bulk retrieval. """ if params is None: params = {} params.setdefault("limit", 5000) all_results = [] result = self.get(path, params=params) if not result: return all_results all_results.extend(result.get("data", [])) # Follow pagination via meta.next next_url = (result.get("meta") or {}).get("next") while next_url: response = self.session.get(next_url) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 60)) print(f" Rate limited. Waiting {retry_after}s...") time.sleep(retry_after) continue if not response.ok: break page_data = response.json() all_results.extend(page_data.get("data", [])) next_url = (page_data.get("meta") or {}).get("next") return all_results # --------------------------------------------------------------------------- # Example Operations # --------------------------------------------------------------------------- def list_domains(client, limit=250): """List domains in your account with pagination.""" print("\n--- Listing Domains ---") result = client.get("/domains/", params={"limit": limit}) if not result: return [] domains = result.get("data", []) meta = result.get("meta", {}) total = meta.get("total_count", len(domains)) print(f"Page: {len(domains)} domains (of {total} total)") for d in domains[:5]: # Show first 5 gateway = d.get("email_gateway", "unknown") print(f" {d['domain']} — gateway: {gateway}") if len(domains) > 5: print(f" ... and {len(domains) - 5} more on this page") return domains def fetch_all_domains(client): """Fetch ALL domains using automatic pagination.""" print("\n--- Fetching All Domains ---") all_domains = client.fetch_all("/domains/") print(f"Total domains fetched: {len(all_domains)}") # Summarize gateway distribution gateways = {} for d in all_domains: gw = d.get("email_gateway") or "Not Found" gateways[gw] = gateways.get(gw, 0) + 1 print(" Gateway distribution:") for gw, count in sorted(gateways.items(), key=lambda x: -x[1])[:5]: print(f" {gw}: {count}") return all_domains def get_domain_detail(client, domain_id): """Get full details for a single domain.""" print(f"\n--- Domain Detail (ID: {domain_id}) ---") result = client.get(f"/domains/{domain_id}/") if not result: return None data = result.get("data", {}) print(f" Domain: {data.get('domain')}") scan = data.get("scan_result", {}) if scan: print(f" Email Gateway: {scan.get('email_gateway', 'N/A')}") print(f" MX: {scan.get('mx', 'N/A')}") print(f" SPF: {scan.get('spf', 'N/A')}") print(f" DMARC Policy: {scan.get('dmarc_policy_p', 'N/A')}") print(f" DMARC RUA: {scan.get('dmarc_rua', 'N/A')}") print(f" Last Scanned: {scan.get('scanned_at', 'N/A')}") print(f" Tech in DNS: {scan.get('tech_in_dns', 'N/A')}") print(f" MX Server Country: {scan.get('mx_server_country', 'N/A')}") print(f" SPF Total Lookups: {scan.get('spf_total_lookups', 'N/A')}") whois = data.get("whois", {}) if whois: print(f" Registrar: {whois.get('registrar', 'N/A')}") print(f" Expires: {whois.get('expiry_date', 'N/A')}") print(f" Privacy: {whois.get('privacy_enabled', 'N/A')}") changes = data.get("change_history", []) if changes: print(f" Recent Changes: {len(changes)}") for ch in changes[:3]: print(f" {ch['detected_at']}: {ch['field_name']} " f"'{ch.get('old_value', '')}' → '{ch.get('new_value', '')}'") return data def add_domains(client, domain_list): """Add new domains for monitoring.""" print(f"\n--- Adding {len(domain_list)} Domains ---") result = client.post("/domains/", data={"domains": domain_list}) if not result: return None data = result.get("data", {}) print(f" Submitted: {data.get('domains_submitted', 0)}") print(f" Added: {data.get('domains_added', 0)}") print(f" Already existed: {data.get('domains_existing', 0)}") print(f" Scans queued: {data.get('scans_queued', 0)}") return data def get_dashboard_summary(client): """Get aggregated dashboard statistics.""" print("\n--- Dashboard Summary ---") result = client.get("/dashboard/summary/") if not result: return None data = result.get("data", {}) total = data.get("total_domains", 0) print(f" Total Domains: {total}") gateway = data.get("email_gateway_distribution", []) if gateway: print(" Email Gateways:") for item in gateway[:5]: name = item.get("gateway_name", "Unknown") count = item.get("count", 0) pct = item.get("percentage", 0) print(f" {name}: {count} ({pct}%)") dmarc = data.get("dmarc_distribution", {}) if dmarc: print(" DMARC Status:") for policy in ("reject", "quarantine", "none", "missing"): info = dmarc.get(policy, {}) print(f" {policy}: {info.get('count', 0)} ({info.get('percentage', 0)}%)") spf = data.get("spf_status", {}) if spf: print(f" SPF: valid={spf.get('valid', 0)}, " f"over_limit={spf.get('over_limit', 0)}, " f"missing={spf.get('missing', 0)}") attention = data.get("needs_attention_count", 0) print(f" Needs Attention: {attention}") return data def get_scan_results(client, domain_id): """Get the latest scan results for a domain.""" print(f"\n--- Scan Results (Domain ID: {domain_id}) ---") result = client.get(f"/domains/{domain_id}/scan-results/") if not result: return None data = result.get("data", {}) print(f" Email Gateway: {data.get('email_gateway', 'N/A')}") print(f" MX: {data.get('mx', 'N/A')}") print(f" A Records: {data.get('a_records', 'N/A')}") print(f" NS: {data.get('ns', 'N/A')}") print(f" SPF: {data.get('spf', 'N/A')}") print(f" DMARC: {data.get('dmarc', 'N/A')}") print(f" DMARC Policy: {data.get('dmarc_policy_p', 'N/A')}") print(f" DKIM Selectors: {data.get('dkim_selectors', 'N/A')}") print(f" Tech in DNS: {data.get('tech_in_dns', 'N/A')}") print(f" MX Country: {data.get('mx_server_country', 'N/A')}") print(f" A Country: {data.get('a_server_country', 'N/A')}") print(f" SPF Lookups: {data.get('spf_total_lookups', 'N/A')}/10") print(f" Last Scanned: {data.get('scanned_at', 'N/A')}") return data def get_change_history(client, domain_id): """Get DNS change history for a domain.""" print(f"\n--- Change History (Domain ID: {domain_id}) ---") all_changes = client.fetch_all(f"/domains/{domain_id}/changes/") print(f" Total changes: {len(all_changes)}") for ch in all_changes[:10]: print(f" {ch['detected_at']}: {ch['field_name']} " f"'{ch.get('old_value', '')}' → '{ch.get('new_value', '')}'") return all_changes def check_rate_limits(client): """Make a request and display the rate limit headers.""" print("\n--- Rate Limit Status ---") response = client.session.get(f"{client.base_url}/") if response.ok: print(f" Rate Limit: {response.headers.get('X-RateLimit-Limit', 'N/A')} req/min") print(f" Remaining: {response.headers.get('X-RateLimit-Remaining', 'N/A')}") print(f" Reset: {response.headers.get('X-RateLimit-Reset', 'N/A')}") print(f" Quota Limit: {response.headers.get('X-API-Quota-Limit', 'N/A')} calls/day") print(f" Quota Used: {response.headers.get('X-API-Quota-Used', 'N/A')}") # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): """Run all example operations.""" # Validate API key if API_KEY == "YOUR_API_KEY_HERE": print("=" * 60) print("DNSai API Client — Sample Script") print("=" * 60) print() print("Please set your API key before running this script:") print() print(" Option 1: Edit this file and replace YOUR_API_KEY_HERE") print(" Option 2: Set the environment variable:") print(" export DNSAI_API_KEY=dnsai_live_your_key_here") print() print("Get your API key at:") print(" https://app.dnsai.com/settings/?tab=apikeys") print() sys.exit(0) print("=" * 60) print("DNSai API Client — Running Examples") print("=" * 60) client = DNSaiClient(API_KEY) # 1. Check rate limits check_rate_limits(client) # 2. Get dashboard summary get_dashboard_summary(client) # 3. List domains (first page) domains = list_domains(client, limit=10) # 4. Get detail for the first domain (if any) if domains: first_id = domains[0].get("id") if first_id: get_domain_detail(client, first_id) get_scan_results(client, first_id) get_change_history(client, first_id) # 5. Fetch ALL domains (uncomment to run — may be slow for large accounts) # all_domains = fetch_all_domains(client) # 6. Add sample domains (commented out to avoid modifying your account) # add_domains(client, ["example.com", "example.org"]) print("\n" + "=" * 60) print("All examples completed successfully!") print("=" * 60) if __name__ == "__main__": main()