Day 13/90: Secure REST API Consumption in Python -- Security Automation

#python #cybersecurity #api #tutorial

Day 13/90: Secure REST API Consumption in Python

90 Day Python Security Scripting Challenge

Security automation depends on reliable API integration. Threat intel feeds, SIEM endpoints, cloud provider APIs: they all need to be consumed correctly or your tooling falls apart under pressure. Here are the patterns I use in production.

Bearer Token Authentication with Sessions

import os
import requests

def build_api_session(token_env, base_headers=None):
    """Create a requests session with token from environment."""
    token = os.environ.get(token_env)
    if not token:
        raise RuntimeError(f"Missing env var: {token_env}")
    sess = requests.Session()
    sess.headers["Authorization"] = f"Bearer {token}"
    sess.headers["User-Agent"] = "SecOpsClient/1.0"
    if base_headers:
        sess.headers.update(base_headers)
    return sess

session = build_api_session("THREAT_INTEL_TOKEN")
resp = session.get("https://feed.example.com/api/v1/iocs", timeout=10)
print(f"Status: {resp.status_code}, IOCs: {len(resp.json())}")

The session approach keeps TCP connections alive and applies auth headers consistently. Pull tokens from environment variables so they never touch version control.

Paginated Data Collection

import os
import requests

def collect_pages(url, headers, key="results", limit=5000):
    """Walk paginated endpoint collecting all results."""
    all_items = []
    page = 1
    while len(all_items) < limit:
        resp = requests.get(
            url, headers=headers,
            params={"page": page, "per_page": 100},
            timeout=15
        )
        resp.raise_for_status()
        items = resp.json().get(key, [])
        if not items:
            break
        all_items.extend(items)
        page += 1
    return all_items[:limit]

hdrs = {"X-Api-Key": os.environ["SIEM_KEY"]}
events = collect_pages("https://siem.local/api/events", hdrs, limit=10000)
print(f"Collected {len(events)} security events")

The hard limit prevents runaway collection. In incident response scenarios you want controlled data volumes, not unbounded fetches.

Rate Limit Compliance

import time
import requests

def rate_safe_get(url, headers, max_attempts=3):
    """GET request with rate limit backoff."""
    for i in range(max_attempts):
        r = requests.get(url, headers=headers, timeout=15)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", 15 * (i + 1)))
            print(f"Rate limited, waiting {wait}s")
            time.sleep(wait)
            continue
        r.raise_for_status()
        return r.json()
    raise Exception("Rate limit retries exhausted")

intel = rate_safe_get(
    "https://otx.alienvault.com/api/v1/pulses/subscribed",
    headers={"X-OTX-API-KEY": os.environ["OTX_KEY"]}
)
print(f"Pulses retrieved: {len(intel.get('results', []))}")

Reading Retry-After is critical. Most security APIs tell you exactly how long to wait. Ignoring it and retrying immediately is how you get your key revoked.

Error Handling with Fallback

import logging
import requests

log = logging.getLogger(__name__)

def fetch_with_fallback(url, headers, cache=None):
    """Fetch live data or fall back to cache on failure."""
    try:
        r = requests.get(url, headers=headers, timeout=10)
        r.raise_for_status()
        return r.json()
    except requests.exceptions.RequestException as exc:
        log.warning(f"API call failed ({exc}), using cached data")
        return cache if cache is not None else []

stale_iocs = [{"type": "domain", "value": "malware.example.com"}]
live_data = fetch_with_fallback(
    "https://feed.local/api/iocs/today",
    headers={"Auth": "Bearer demo"},
    cache=stale_iocs
)
print(f"Working with {len(live_data)} indicators")

During incidents, stale threat data beats no data. The fallback pattern keeps your pipeline producing output even when upstream APIs are down.

Key Takeaways

Pull credentials from environment, never source code
Cap pagination to prevent quota and memory exhaustion
Respect rate limits or lose API access at the worst possible time
Fall back to cached data rather than crashing on API failure

Day 13/90 of the 90 Day Python Security Scripting Challenge

Keep all security scripts in a dedicated venv with pinned dependencies. Run bandit against your own code regularly -- security tools should be held to the same standard as the systems they protect.

DEV Community