Fetch the (trust anchor) root CA certificate for a given HTTPS hostname and print it in two formats

get_root_cert.py

#!/usr/bin/env python3
"""
Fetch the (trust anchor) root CA certificate for a given HTTPS hostname and
print it in two formats:

1) As standard PEM
2) As a Python string literal suitable for CircuitPython's
   ssl_context.load_verify_locations(cadata=ssl_cert)

Why this exists:
- CircuitPython often requires embedding a specific CA/root cert as a string
  so https requests can validate a given site.
- Servers frequently do not send the root certificate in the TLS handshake.
  This tool attempts to retrieve the certificate chain and, if needed, follow
  AIA (Authority Information Access) pointers to download intermediates until
  the self-signed root CA is found.

Requirements:
- Python 3.9+
- OpenSSL available on PATH (openssl command line).
  We use it to obtain the presented chain and to parse AIA info if needed.
- Network access to the host and potential AIA URLs.

Usage:
  python tools/get_root_cert.py <hostname> [--port 443] [--out FILE]

The output includes a ready-to-copy Python variable snippet you can paste into
`core/network.py` similar to `ssl_cert = '-----BEGIN CERTIFICATE-----\n...`.
"""
import argparse
import base64
import re
import subprocess
import sys
import textwrap
import urllib.request
from typing import List, Optional, Tuple

PEM_BEGIN = "-----BEGIN CERTIFICATE-----"
PEM_END = "-----END CERTIFICATE-----"


def run(cmd: List[str], input_bytes: bytes | None = None, timeout: int = 30) -> Tuple[int, bytes, bytes]:
    proc = subprocess.Popen(cmd, stdin=subprocess.PIPE if input_bytes else None, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = proc.communicate(input=input_bytes, timeout=timeout)
    return proc.returncode, out, err


def parse_pems_from_text(text: str) -> List[str]:
    certs = []
    pattern = re.compile(r"{}[\s\S]*?{}".format(re.escape(PEM_BEGIN), re.escape(PEM_END)))
    for m in pattern.finditer(text):
        p = m.group(0)
        # Normalize newlines
        p = "\n".join(line.strip() for line in p.strip().splitlines()) + "\n"
        certs.append(p)
    return certs


def is_self_signed(pem: str) -> bool:
    # Quick heuristic: subject == issuer
    code, out, _ = run(["openssl", "x509", "-noout", "-subject", "-issuer"], input_bytes=pem.encode("utf-8"))
    if code != 0:
        return False
    text = out.decode("utf-8", errors="ignore")
    subj = None
    iss = None
    for line in text.splitlines():
        if line.startswith("subject="):
            subj = line.strip()
        elif line.startswith("issuer="):
            iss = line.strip()
    return subj is not None and iss is not None and subj == iss


def is_ca_certificate(pem: str) -> bool:
    # Look for Basic Constraints: CA:TRUE
    code, out, _ = run(["openssl", "x509", "-noout", "-text"], input_bytes=pem.encode("utf-8"))
    if code != 0:
        return False
    text = out.decode("utf-8", errors="ignore").lower()
    return "basic constraints:" in text and "ca:true" in text


def get_aia_issuer_uris(pem: str) -> List[str]:
    # Parse AIA (Authority Information Access) URIs for CA Issuers
    code, out, _ = run(["openssl", "x509", "-noout", "-text"], input_bytes=pem.encode("utf-8"))
    if code != 0:
        return []
    text = out.decode("utf-8", errors="ignore")
    uris = []
    for line in text.splitlines():
        line = line.strip()
        if "CA Issuers - URI:" in line:
            uri = line.split("CA Issuers - URI:", 1)[1].strip()
            uris.append(uri)
    return uris


def der_bytes_to_pem(der: bytes) -> str:
    b64 = base64.encodebytes(der).decode("ascii")
    # Ensure proper 64-char wrapping
    b64_wrapped = "".join(b64)
    lines = [b64_wrapped[i : i + 64] for i in range(0, len(b64_wrapped), 64)]
    return PEM_BEGIN + "\n" + "\n".join(l for l in lines if l.strip()) + "\n" + PEM_END + "\n"


def ensure_pem(cert_bytes: bytes) -> str:
    text = cert_bytes.decode("utf-8", errors="ignore")
    if PEM_BEGIN in text and PEM_END in text:
        # Already PEM
        pems = parse_pems_from_text(text)
        return pems[0] if pems else text
    # Treat as DER
    return der_bytes_to_pem(cert_bytes)


def fetch_via_aia(pem: str) -> Optional[str]:
    uris = get_aia_issuer_uris(pem)
    for uri in uris:
        try:
            with urllib.request.urlopen(uri, timeout=20) as resp:
                data = resp.read()
            return ensure_pem(data)
        except Exception:
            continue
    return None


def get_presented_chain(host: str, port: int) -> List[str]:
    # Use s_client to get everything the server sent
    # We pass an empty line to quit interactive mode
    cmd = [
        "openssl",
        "s_client",
        "-servername",
        host,
        "-showcerts",
        "-connect",
        f"{host}:{port}",
    ]
    code, out, err = run(cmd, input_bytes=b"\n")
    text = (out + err).decode("utf-8", errors="ignore")
    return parse_pems_from_text(text)


def find_root_cert(host: str, port: int) -> Tuple[str, List[str]]:
    """Return (root_pem, chain_collected) where chain_collected are all PEMs followed."""
    collected: List[str] = []
    presented = get_presented_chain(host, port)
    collected.extend(presented)

    # If any presented is already a root CA, return it
    for pem in collected:
        if is_ca_certificate(pem) and is_self_signed(pem):
            return pem, collected

    # Otherwise, try to chase AIA starting from the last collected cert
    safety_hops = 6  # avoid infinite loops
    current = collected[-1] if collected else None
    while current and safety_hops > 0:
        safety_hops -= 1
        parent = fetch_via_aia(current)
        if not parent:
            break
        # If already have it, stop to avoid cycles
        if any(parent == c for c in collected):
            break
        collected.append(parent)
        if is_ca_certificate(parent) and is_self_signed(parent):
            return parent, collected
        current = parent

    # Fallback: choose the last CA in the collected list if no self-signed found
    for pem in reversed(collected):
        if is_ca_certificate(pem):
            return pem, collected

    # As a last resort, return the last certificate we saw (may be intermediate)
    if collected:
        return collected[-1], collected
    raise RuntimeError("No certificates could be obtained for the host")


def to_python_string_literal(pem: str, var_name: str = "ssl_cert") -> str:
    # Escape backslashes and single quotes, and ensure \n line endings are explicit
    body = pem.replace("\\", "\\\\").replace("'", "\\'")
    body = body.replace("\r\n", "\n").replace("\r", "\n")
    # Keep a trailing newline to match expected format
    body = body if body.endswith("\n") else (body + "\n")
    # Convert actual newlines to \n sequences for single-quoted Python string
    body = body.replace("\n", "\\n\n").rstrip("\n")  # double-newline for readability
    wrapped = (
        f"{var_name} = '\n" + body + "\n'\n"
    )
    return wrapped


def main(argv: List[str]) -> int:
    parser = argparse.ArgumentParser(description="Fetch root CA certificate for a host and print as PEM and Python string literal.")
    parser.add_argument("host", help="Hostname (e.g., example.com)")
    parser.add_argument("--port", type=int, default=443, help="Port (default: 443)")
    parser.add_argument("--out", help="Optional path to write the Python string literal (e.g., paste into CircuitPython code)")
    parser.add_argument("--var", default="ssl_cert", help="Variable name to use in the Python output (default: ssl_cert)")
    args = parser.parse_args(argv)

    try:
        root_pem, chain = find_root_cert(args.host, args.port)
    except Exception as e:
        sys.stderr.write(f"Error: {e}\n")
        return 2

    # Print summaries
    print("Collected certificate chain ({} certs):".format(len(chain)))
    for i, pem in enumerate(chain, start=1):
        code, out, _ = run(["openssl", "x509", "-noout", "-subject", "-issuer"], input_bytes=pem.encode("utf-8"))
        subj_iss = out.decode("utf-8", errors="ignore").strip()
        print(f"  [{i}]\n{textwrap.indent(subj_iss, '     ')}")

    print("\nSelected root (or best-available CA):")
    code, out, _ = run(["openssl", "x509", "-noout", "-subject", "-issuer"], input_bytes=root_pem.encode("utf-8"))
    print(out.decode("utf-8", errors="ignore").strip())

    print("\n----- Root Certificate (PEM) -----")
    print(root_pem, end="")

    py_snippet = to_python_string_literal(root_pem, var_name=args.var)

    print("\n----- Python string literal (for CircuitPython) -----")
    print(py_snippet)

    if args.out:
        with open(args.out, "w", encoding="utf-8") as f:
            f.write(py_snippet)
        print(f"Saved Python string literal to: {args.out}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))