Calling HTTP APIs safely

Timeouts, status checks, TLS with your own CA, pagination limits and 429 handling.

Security automation talks to HTTP APIs all the time: a threat-intelligence feed, a ticketing system, a scanner's REST endpoint. This lesson makes those calls so that a stalled server cannot hang your job, a wrong certificate is refused instead of trusted, a token only goes where you meant to send it, and a buggy API cannot keep your loop running forever. You will use urllib.request from the standard library first, then the requests package where it earns its place, against a small HTTPS API on the lab machine with its own certificate authority. Retry policy was the previous lesson's subject and is used here, not taught again.

A local API with its own certificate authority

HTTPS protects a request in two ways: it encrypts the connection, and it lets the client check who it is talking to. The server presents a certificate, a signed statement that says "this public key belongs to the name localhost". The client accepts it only if it was signed by a certificate authority (CA) the client trusts, and only if the name matches the host it asked for. Companies run private CAs for internal services, so you will meet certificates your system does not trust out of the box. Unpack the lesson files (the box at the top of this page) in your home directory and work in ~/ps-http. Besides the scripts shown here, they hold three small servers the lesson starts and describes but does not list: api_server.py, stuck_listener.py and collector.py. First, make a CA:

#!/usr/bin/env bash
# Creates a throwaway lab CA and a server certificate for "localhost", valid for 7 days.
# The two .key files are private keys: openssl writes them with mode 0600. Never print or copy them.
set -euo pipefail
 
# The CA: self-signed, allowed to sign certificates (keyCertSign). Python 3.13 and later
# reject a CA certificate without a keyUsage extension.
openssl req -x509 -new -newkey ec -pkeyopt ec_paramgen_curve:P-256 -noenc \
  -keyout ca.key -out ca.pem -days 7 -subj "/CN=ps-http lab CA" \
  -addext "keyUsage=critical,keyCertSign,cRLSign"
 
# The server: a key and a signing request, then the CA signs it for the name "localhost" only.
openssl req -new -newkey ec -pkeyopt ec_paramgen_curve:P-256 -noenc \
  -keyout server.key -out server.csr -subj "/CN=localhost"
printf '%s\n' "subjectAltName=DNS:localhost" "basicConstraints=critical,CA:FALSE" \
  "keyUsage=critical,digitalSignature" "extendedKeyUsage=serverAuth" > server.ext
openssl x509 -req -in server.csr -CA ca.pem -CAkey ca.key -days 7 \
  -extfile server.ext -out server.pem
rm -f server.csr server.ext
echo "lab CA: ca.pem   server certificate: server.pem (localhost)"

openssl reports each step. ca.pem and server.pem are certificates, public by design; the two .key files are the private keys, created with mode 0600, and nothing in this lesson prints them. The server certificate is valid for DNS:localhost only, issued by the lab CA. The API, api_server.py, is http.server wrapped in TLS on localhost:18110. /v1/status answers at once, /v1/slow after ten seconds, /v1/drip one byte every quarter second, and /v1/moved redirects to another host. With a valid token, /v1/indicators returns three pages (with a 429 the first time page 2 is asked for), /v1/loop repeats its cursor, /v1/busy asks you to wait an hour, and /v1/html, /v1/down, /v1/drop and /v1/garbled answer with HTML, 503, a closed connection and broken JSON. The server reads the valid token from api-token; the lesson also needs an expired one in old-token. Create both readable only by you (the secrets lesson explains why), then start the server in the background and keep its process ID:

Now ask for its status the way most first scripts do:

Python checked the certificate against the system's trusted CAs, found no issuer it knew, and refused: CERTIFICATE_VERIFY_FAILED, "unable to get local issuer certificate". This is the check working. The fix is to trust the lab CA, and only for this client: ssl.create_default_context(cafile="ca.pem") builds a context with verification on that trusts that one CA file.

With the CA the request works. The second command asked for 127.0.0.1, a name the certificate does not list, and was refused: verification checks the name as well as the signature. The advice you will find for this error is to turn verification off (verify=False in requests, ssl._create_unverified_context() or CERT_NONE). That is unsafe and never the fix: the connection is still encrypted, but to whoever answers, so anyone on the network path can impersonate the API, collect your token and feed your automation invented results. Trust the right CA instead. One trap when you build your own CA: Python 3.13 and later reject a CA certificate without a keyUsage extension ("CA cert does not include key usage extension", which the lab checked), so make_lab_ca.sh adds one.

Timeouts, status codes and a token

api.py wraps one GET request in everything the rest of the lesson relies on:

"""One GET against the lab API with urllib: verified TLS, a token, time limits, checked JSON."""
import http.client
import json
import ssl
import time
import urllib.error
import urllib.request
 
BASE = "https://localhost:18110"
CONTEXT = ssl.create_default_context(cafile="ca.pem")  # verification on, trusting the lab CA
MAX_BODY = 1_000_000  # bytes; a larger answer is a bug or an attack, not data
 
 
class ApiError(Exception):
    """The API refused the request or answered with something unusable: do not retry."""
 
 
class Unavailable(ApiError):
    """Timeouts, refused or dropped connections, 5xx: the same request may work later."""
 
 
class RateLimited(Unavailable):
    def __init__(self, retry_after: float):
        super().__init__(f"HTTP 429, Retry-After {retry_after:.0f}s")
        self.retry_after = retry_after
 
 
def get_json(path: str, token: str, timeout: float = 5.0, budget: float = 10.0) -> dict:
    """timeout limits each step (the connect, every read); budget limits reading the body."""
    request = urllib.request.Request(BASE + path, headers={"Accept": "application/json"})
    request.add_unredirected_header("Authorization", f"Bearer {token}")  # not resent after a redirect
    try:
        with urllib.request.urlopen(request, timeout=timeout, context=CONTEXT) as response:
            content_type = response.headers.get_content_type()
            if content_type != "application/json":
                raise ApiError(f"{path}: expected JSON, got {content_type}")
            body = read_body(response, path, time.monotonic() + budget)
    except urllib.error.HTTPError as err:  # 4xx and 5xx answers
        if err.code == 429:
            raise RateLimited(retry_after_seconds(err.headers.get("Retry-After"))) from err
        kind = Unavailable if err.code >= 500 else ApiError
        raise kind(f"HTTP {err.code} for {path}") from err
    except (OSError, http.client.HTTPException) as err:  # no usable answer at all
        reason = getattr(err, "reason", err)  # URLError keeps the real error in .reason
        kind = ApiError if isinstance(reason, ssl.SSLCertVerificationError) else Unavailable
        raise kind(f"{path}: {reason}") from err
    try:
        return json.loads(body)
    except ValueError as err:  # JSONDecodeError, or bytes that are not UTF-8
        raise ApiError(f"{path}: invalid JSON: {err}") from err
 
 
def read_body(response, path: str, give_up_at: float) -> bytes:
    """Read what arrives, chunk by chunk, until the end, the size cap or the time budget."""
    body = b""
    while True:
        chunk = response.read1(65536)  # returns as soon as some bytes have arrived
        if not chunk:
            return body
        body += chunk
        if len(body) > MAX_BODY:
            raise ApiError(f"{path}: answer larger than {MAX_BODY} bytes")
        if time.monotonic() > give_up_at:
            raise Unavailable(f"{path}: answer still incomplete after the time budget")
 
 
def retry_after_seconds(value: str | None) -> float:
    """Retry-After in seconds; anything else (missing, or an HTTP date) counts as 'too long'."""
    # isascii() too: isdigit() alone accepts characters such as '²' that float() refuses
    return float(value) if value and value.isascii() and value.isdigit() else float("inf")

add_unredirected_header() sends the token on this request only, never on a redirect; the last section shows why. urlopen() raises HTTPError for every 4xx and 5xx answer, with the status and the response headers. The classes follow the previous lesson's design: Unavailable means "try later" (timeouts, refused or dropped connections, 5xx, 429), plain ApiError "this request will never work" (401, 404, a certificate that fails verification, a body that is not what was promised). RateLimited has its own __init__, which builds the message through super().__init__() and keeps the wait in self.retry_after. The second except is the previous lesson's rule in one clause: OSError and http.client.HTTPException. getattr(err, "reason", err) reads the attribute reason if the exception has one (URLError keeps the real cause there), else the exception itself. A 200 answer is not proof of good data either: a proxy's maintenance page arrives as 200 text/html, so the Content-Type is checked before reading, and read_body() caps the size. str | None in a type hint means "a string, or None".

timeout= in urlopen() applies to each blocking step on its own: connecting, then every read. Two different failures show it:

/v1/slow accepts the connection and then sends nothing. The read gave up after two seconds with TimeoutError ("The read operation timed out"), which get_json() turned into Unavailable. stuck_listener.py listens but never accepts, and its queue is already full, so the kernel drops new connection attempts, as a firewall that drops packets or a badly overloaded host does. The connect gave up after two seconds, and urllib wrapped the TimeoutError in a URLError. Without timeout=, both calls would wait for the operating system's own limits, which for a stalled read means indefinitely.

Because the limit is per step, a server that keeps sending a little never trips it. /v1/drip sends its 46-byte answer one byte every quarter second, and plain urlopen() with a two-second timeout waits for all of it:

Over eleven seconds with a two-second timeout, because no single read waited two seconds. The previous lesson's deadline would not have helped: retry() checks it only between attempts, and this was one attempt. So get_json() has a second limit, budget, for reading the body: read_body() reads with read1(), which returns as soon as some bytes have arrived, and checks the clock after every chunk. With a three-second budget the drip was cut off after about three seconds. The budget can overrun by one read timeout, and it does not cover the headers, which urlopen() reads first. For a hard limit on a whole program, run it under timeout from the Bash course, as the capstone's schedule does.

Pagination and rate limits

APIs return long lists a page at a time. Each page carries a cursor, an opaque value you send back to get the next page, and the last page has none. A loop that follows cursors trusts the server to end it; a bug that returns the same cursor again turns it into a job that runs all night. pages.py adds two limits and handles 429:

"""Read every page of a paginated endpoint, with limits a buggy or hostile API cannot break."""
import logging
import time
from urllib.parse import quote
 
from api import ApiError, RateLimited, Unavailable, get_json
 
log = logging.getLogger(__name__)
VERDICTS = {"malicious", "suspicious", "benign"}
 
 
def fetch_all(path: str, token: str, *, max_pages: int = 20, max_wait: float = 10.0) -> list[dict]:
    items, seen, cursor = [], set(), None
    for page in range(1, max_pages + 1):
        url = path if cursor is None else f"{path}?cursor={quote(cursor)}"
        data = get_page(url, token, max_wait)
        items.extend(check_items(data, url))
        cursor = data.get("next_cursor")
        if cursor is None:
            return items
        if not isinstance(cursor, str) or cursor in seen:
            raise ApiError(f"page {page} repeats cursor {cursor!r}: the API is looping")
        seen.add(cursor)
    raise ApiError(f"more than {max_pages} pages; stopping")
 
 
def get_page(url: str, token: str, max_wait: float) -> dict:
    """One page; wait out one 429 if the server asks for max_wait seconds or less."""
    try:
        return get_json(url, token)
    except RateLimited as err:
        if err.retry_after > max_wait:
            raise Unavailable(f"{err}, more than the {max_wait:.0f}s we wait") from err
        log.warning("%s: %s; waiting", url, err)
        time.sleep(err.retry_after)
        return get_json(url, token)  # a second 429 propagates
 
 
def check_items(data: dict, url: str) -> list[dict]:
    items = data.get("items") if isinstance(data, dict) else None
    if not isinstance(items, list):
        raise ApiError(f"{url}: no 'items' list in the answer")
    for item in items:
        if not isinstance(item, dict) or item.get("verdict") not in VERDICTS \
                or not isinstance(item.get("indicator"), str):
            raise ApiError(f"{url}: unexpected item {item!r}")
    return items

max_pages caps the loop whatever the server does; seen catches the more common failure, a cursor that repeats. A 429 (Too Many Requests) answer often says how long to wait in a Retry-After header. get_page() waits that long once, but only up to max_wait: a job that sleeps an hour on the server's say-so has stopped doing its work. An HTTP-date Retry-After counts as too long (scripting-adv parses it properly), and so does a value such as ², which isdigit() accepts and float() refuses, hence isascii() (the hashing lesson meets the same trap). check_items() validates each page before any of it is used; the backslash continues its long if on the next line. list_indicators.py maps the exceptions to exit statuses 0, 1 and 3, as check_feed.py did.

"""List the malicious indicators from the lab API.
 
Usage: list_indicators.py PATH [TOKEN_FILE]   (TOKEN_FILE defaults to api-token)
Exit status: 0 done, 1 the API refused or answered wrongly, 2 usage, 3 unavailable, try later.
"""
import logging
import sys
from pathlib import Path
 
from api import ApiError, Unavailable
from pages import fetch_all
 
log = logging.getLogger("list_indicators")
 
 
def main(argv: list[str]) -> int:
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
    if len(argv) not in (1, 2):
        log.error("usage: list_indicators.py PATH [TOKEN_FILE]")
        return 2
    token = Path(argv[1] if len(argv) == 2 else "api-token").read_text(encoding="utf-8").strip()
    try:
        items = fetch_all(argv[0], token)
    except Unavailable as err:  # a subclass of ApiError, so it is caught first
        log.error("unavailable: %s", err)
        return 3
    except ApiError as err:
        log.error("%s (not retried)", err)
        return 1
    bad = [item["indicator"] for item in items if item["verdict"] == "malicious"]
    print(f"{len(items)} indicators, {len(bad)} malicious: {', '.join(bad)}")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

Three pages, six indicators. The first request for page 2 got a 429 with Retry-After: 1; the tool waited one second and asked again. The server log shows each request arriving on a new connection (conn 5, 6, 7, 8): urllib opens and closes a connection for every request, and pays for a new TLS handshake each time. Now the failures:

An expired token got a 401 and exit status 1, with no retry: a wrong credential does not fix itself, and retrying it can lock the account. The HTML page was rejected by its Content-Type before the parser saw it. The looping API was stopped on its second page. The Retry-After: 3600 answer ended the run at once with exit status 3, "try later", instead of an hour's sleep. The closed connection is "try later" too (3), and the broken JSON is a wrong answer (1); neither ends in a traceback.

requests: a Session, Retry and the exceptions it raises

urllib did everything so far. The requests package is worth a dependency when a tool makes many calls: a Session keeps connections open and reuses them (connection pooling), holds headers and TLS settings once for all calls, and mounts an adapter that retries with urllib3's Retry policy. It is not in the standard library, so it goes into a venv, pinned exactly like PyYAML in the data lesson; pip install -r FILE installs every requirement listed in the file:

# Exact versions this lesson was tested with (the packaging lesson adds hashes).
requests==2.34.2
urllib3==2.8.0
certifi==2026.7.22
idna==3.20
charset-normalizer==3.5.1

"""A requests Session for the lab API: pooled connections, bounded retries, verified TLS."""
import json
 
import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
 
BASE = "https://localhost:18110"
TIMEOUT = (3.05, 5)  # (connect, read) in seconds; pass it on every call
MAX_BODY = 1_000_000  # bytes, as in api.py
 
 
def make_session(token: str) -> requests.Session:
    retry = Retry(
        total=3,                     # at most 3 retries: 4 attempts in all
        backoff_factor=0.5,          # no wait before the first retry, then 1 s, then 2 s
        backoff_jitter=0.5,          # plus a random 0-0.5 s
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods={"GET"},     # never repeat a write on its own
        retry_after_max=10,          # a longer Retry-After is cut down to 10 s
        other=0,                     # a failed certificate check is not retried
    )
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=retry))
    session.verify = "ca.pem"        # verification on, trusting the lab CA
    session.headers["Authorization"] = f"Bearer {token}"
    return session
 
 
def read_json(response: requests.Response) -> dict:
    """The body of a stream=True response as JSON, after checking its type and size."""
    content_type = response.headers.get("Content-Type", "").split(";")[0].strip()
    if content_type != "application/json":
        raise ValueError(f"expected JSON, got {content_type or 'no Content-Type'}")
    body = b""
    for chunk in response.iter_content(65536):
        body += chunk
        if len(body) > MAX_BODY:
            raise ValueError(f"answer larger than {MAX_BODY} bytes")
    return json.loads(body)

Retry counts attempts, waits between them and, for 429 and 503, uses the server's Retry-After instead of its own backoff; the urllib3 documentation gives the wait as backoff_factor * 2 ** (retries so far), with none before the first retry. retry_after_max matters: its default is 21600 seconds, so one Retry-After could park your job for six hours. With a longer value it waits 10 seconds and still retries, where pages.py refused. allowed_methods={"GET"} keeps it from repeating writes. other=0 matters for security: urllib3 would retry a failed certificate check like a network hiccup, although it may be someone impersonating the API. requests has no default timeout, so TIMEOUT goes on every call: 3.05 seconds to connect, 5 for each read. read_json() brings back the checks api.py makes, the Content-Type and a size cap while reading, which is why req_get.py asks for stream=True (the body is read only when iter_content() asks).

"""GET one lab API path with requests and report what happened.
 
Usage: req_get.py PATH [BASE_URL]
Exit status: 0 answer printed, 1 the API refused or answered wrongly, 3 unavailable, try later.
"""
import sys
from pathlib import Path
 
import requests
from requests.exceptions import RetryError  # not exported as requests.RetryError
 
from session import BASE, TIMEOUT, make_session, read_json
 
 
def main(path: str, base: str = BASE) -> int:
    token = Path("api-token").read_text(encoding="utf-8").strip()
    with make_session(token) as session:
        try:
            response = session.get(base + path, timeout=TIMEOUT, allow_redirects=False, stream=True)
            response.raise_for_status()  # a 4xx, or a 5xx that is not retried
            if response.is_redirect:
                raise requests.HTTPError(f"unexpected redirect to {response.headers['Location']}")
            data = read_json(response)
        except requests.HTTPError as err:
            print(f"refused: {err}", file=sys.stderr)
            return 1
        except RetryError as err:  # retries used up on 429/5xx answers
            print(f"gave up: {err}", file=sys.stderr)
            return 3
        except requests.exceptions.SSLError as err:  # a subclass of ConnectionError: first
            print(f"TLS verification failed (not retried): {err}", file=sys.stderr)
            return 1
        except requests.ConnectionError as err:  # refused, dropped, read timeouts after retries
            print(f"no answer: {err}", file=sys.stderr)
            return 3
        except ValueError as err:  # from read_json(): wrong type, too large, not JSON
            print(f"not usable: {err}", file=sys.stderr)
            return 1
    print(f"HTTP {response.status_code}: {data}")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main(*sys.argv[1:]))

Three requests in one session arrived on one connection (conn 17), where urllib needed a new one each time. Here is what req_get.py sees when things go wrong:

/v1/down was tried four times on one connection in about 3.9 seconds (waits of 0, 1 and 2 seconds plus jitter), then requests raised RetryError. There was no response for raise_for_status() to look at, so code that only catches HTTPError misses exactly the case it was written for. RetryError lives in requests.exceptions, not requests; a typo there only surfaces when a 503 reaches the except line, as an AttributeError. /v1/slow: the read timeout was retried too, four attempts of five seconds, and what came out was ConnectionError with "Read timed out" inside, not requests.Timeout. /v1/busy: three waits of 10 seconds instead of three hours, then RetryError. The 404, the HTML body and the redirect ended with 1. The last command asked for 127.0.0.1: SSLError, exit status 1, in a fraction of a second; without other=0 the lab measured about three seconds of retries before the same exception.

Redirects and credentials

A 3xx answer tells the client to repeat the request somewhere else, and both libraries follow it by default. /v1/moved redirects to the collector on 127.0.0.1:18111, a different host and port over plain HTTP, which logs any credentials it receives. redirect_check.py sends an Authorization header and a custom X-Api-Key header four ways (lambda: urllib_get(False) is a function with no name and no parameters, so the loop can call each variant the same way):

"""Which credentials follow a redirect to another host? /v1/moved sends the client to the
collector on 127.0.0.1:18111, which logs the Authorization and X-Api-Key headers it receives."""
import ssl
import urllib.request
 
import requests
 
URL = "https://localhost:18110/v1/moved"
HEADERS = {"Authorization": "Bearer lab-token-not-secret", "X-Api-Key": "lab-apikey-not-secret"}
CONTEXT = ssl.create_default_context(cafile="ca.pem")
 
 
def urllib_get(unredirected: bool) -> str:
    request = urllib.request.Request(URL)
    for name, value in HEADERS.items():
        if unredirected:
            request.add_unredirected_header(name, value)
        else:
            request.add_header(name, value)
    with urllib.request.urlopen(request, timeout=5, context=CONTEXT) as response:
        return response.url
 
 
def requests_get(follow: bool) -> str:
    with requests.Session() as session:
        session.headers.update(HEADERS)
        response = session.get(URL, timeout=5, verify="ca.pem", allow_redirects=follow)
        return response.url if follow else f"{response.status_code}, not followed"
 
 
for name, call in [("urllib add_header", lambda: urllib_get(False)),
                   ("urllib add_unredirected_header", lambda: urllib_get(True)),
                   ("requests", lambda: requests_get(True)),
                   ("requests allow_redirects=False", lambda: requests_get(False))]:
    print(f"{name}: ended at {call()}")

Read the collector's lines in order. urllib with add_header() forwarded both headers to the other host, in clear text; with add_unredirected_header() it forwarded neither, which is why api.py uses it. requests removed Authorization because the host changed, but forwarded X-Api-Key: it only knows that one header name is a credential, and many APIs (VirusTotal, AbuseIPDB and others) take their key in a custom header. With allow_redirects=False nothing reached the collector. An API that answers a token-bearing request with a redirect is misconfigured or compromised; report it, as req_get.py does. For fleet-scale clients, see the scripting-adv lesson on fleet automation.

Try this

Write req_pages.py PATH in ~/ps-http that reads every page through the requests session from session.py: params={"cursor": cursor} for the next page, allow_redirects=False, at most 20 pages, and a set of seen cursors. It exits 0 and prints 6 indicators from 3 pages for /v1/indicators; exits 1 with error: cursor 'p2' repeated for /v1/loop; and exits 3 for /v1/down after the retries (catch RetryError from requests.exceptions). Then break it: disable the seen-cursor check, run it against /v1/loop, and confirm that only the page cap stops it, after 20 requests, with error: more than 20 pages. When you are done, stop the three servers the lesson started:

Takeaway

Every HTTP call gets a timeout, verified TLS against the CA that really issued the certificate, a check of the status and content type before parsing, and hard limits on pages, waits and the time spent reading one answer. Send credentials only to the host you meant: do not follow redirects on authenticated API calls, and catch RetryError, SSLError and ConnectionError as well as HTTPError when a retry adapter is mounted.

Related