X.509 Certificate Parser (python, written by Codex)

Codex wrote this python project from the task below. It does not run on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: X.509 Certificate Parser

Write a program that parses X.509 digital certificates in PEM and DER formats, extracts all fields, validates the certificate chain, and checks expiration status.

FUNCTIONAL REQUIREMENTS:
- Accept a certificate file path as a command-line argument (support both PEM and DER formats, auto-detected)
- Extract and display all certificate fields: version, serial number, issuer, subject, validity period (not before/not after), public key algorithm and size, signature algorithm, and fingerprints (SHA-1, SHA-256)
- Parse all X.509 v3 extensions: Subject Alternative Names (SANs), Key Usage, Extended Key Usage, Basic Constraints, Authority/Subject Key Identifiers, CRL Distribution Points
- Check certificate expiration: report if expired, days until expiration, or days since expiration
- Validate a certificate chain when multiple certificates are provided: verify that each certificate is signed by the next one in the chain
- Support reading certificate bundles (multiple PEM certificates concatenated in one file) and parsing each individually
- Support a --format flag to choose output format: text (default human-readable), json, or csv
- Support fetching and parsing a remote server's certificate via --host flag (given a hostname and optional port)
- Print the parsed certificate details to console in a structured, readable format
- Save the output to a file via --output flag
- If no input is given, generate a self-signed CA certificate and a leaf certificate signed by it, then parse both and demonstrate chain validation
- Handle errors: invalid certificate data, unsupported formats, incomplete chains, and encoding issues

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# X.509 Certificate Parser (Python)

Parses X.509 certificates (PEM/DER/bundles/remote host), extracts fields/extensions, checks expiration, and validates chain signatures.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies (Pinned)

- `cryptography==44.0.0`
- `cffi==1.17.1`
- `pycparser==2.22`

## Setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run

```bash
python src/main.py ./cert.pem
python src/main.py ./bundle.pem --format json
python src/main.py ./cert.der --format csv --output ./report.csv
python src/main.py --host example.com:443 --format text
python src/main.py
```

## Features

- PEM/DER auto-detection
- PEM bundle parsing
- Full field extraction
- X.509 extension parsing:
  - SAN
  - Key Usage
  - Extended Key Usage
  - Basic Constraints
  - SKID/AKID
  - CRL Distribution Points
- Expiration status reporting
- Chain signature validation across cert list
- Remote host certificate fetch (`--host`)
- Output formats: text/json/csv
- Output file support (`--output`)
- No-args demo generates CA + signed leaf and validates chain

cryptography==44.0.0
cffi==1.17.1
pycparser==2.22

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
import socket
import ssl
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa, ec
from cryptography.x509.oid import ExtensionOID, NameOID


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="X.509 Certificate Parser")
    p.add_argument("path", nargs="?")
    p.add_argument("--host", help="hostname[:port]")
    p.add_argument("--port", type=int, default=443)
    p.add_argument("--format", default="text", choices=["text", "json", "csv"])
    p.add_argument("--output")
    return p.parse_args()


def load_certs_from_file(path: Path) -> list[x509.Certificate]:
    data = path.read_bytes()
    text = data.decode("utf-8", errors="ignore")
    if "BEGIN CERTIFICATE" in text:
        certs = []
        for block in re.findall(r"-----BEGIN CERTIFICATE-----[\s\S]*?-----END CERTIFICATE-----", text):
            certs.append(x509.load_pem_x509_certificate(block.encode("utf-8")))
        if not certs:
            raise ValueError("No PEM certificates found.")
        return certs
    return [x509.load_der_x509_certificate(data)]


def fetch_host_cert(host: str, port: int) -> list[x509.Certificate]:
    ctx = ssl.create_default_context()
    with socket.create_connection((host, port), timeout=10) as sock:
        with ctx.wrap_socket(sock, server_hostname=host) as ssock:
            leaf_der = ssock.getpeercert(binary_form=True)
    if not leaf_der:
        raise ValueError("Could not fetch certificate from host.")
    return [x509.load_der_x509_certificate(leaf_der)]


def name_to_str(name: x509.Name) -> str:
    return ", ".join([f"{attr.oid._name or attr.oid.dotted_string}={attr.value}" for attr in name])


def key_info(pub) -> dict[str, Any]:
    if isinstance(pub, rsa.RSAPublicKey):
        return {"algorithm": "RSA", "size": pub.key_size}
    if isinstance(pub, ec.EllipticCurvePublicKey):
        return {"algorithm": "EC", "size": pub.key_size}
    return {"algorithm": type(pub).__name__, "size": None}


def parse_extensions(cert: x509.Certificate) -> dict[str, Any]:
    out: dict[str, Any] = {
        "subject_alt_names": [],
        "key_usage": {},
        "extended_key_usage": [],
        "basic_constraints": {},
        "subject_key_identifier": None,
        "authority_key_identifier": None,
        "crl_distribution_points": [],
    }
    try:
        san = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value
        out["subject_alt_names"] = [str(x.value) for x in san]
    except x509.ExtensionNotFound:
        pass
    try:
        ku = cert.extensions.get_extension_for_oid(ExtensionOID.KEY_USAGE).value
        out["key_usage"] = {
            "digital_signature": ku.digital_signature,
            "content_commitment": ku.content_commitment,
            "key_encipherment": ku.key_encipherment,
            "data_encipherment": ku.data_encipherment,
            "key_agreement": ku.key_agreement,
            "key_cert_sign": ku.key_cert_sign,
            "crl_sign": ku.crl_sign,
            "encipher_only": ku.encipher_only,
            "decipher_only": ku.decipher_only,
        }
    except x509.ExtensionNotFound:
        pass
    try:
        eku = cert.extensions.get_extension_for_oid(ExtensionOID.EXTENDED_KEY_USAGE).value
        out["extended_key_usage"] = [oid.dotted_string for oid in eku]
    except x509.ExtensionNotFound:
        pass
    try:
        bc = cert.extensions.get_extension_for_oid(ExtensionOID.BASIC_CONSTRAINTS).value
        out["basic_constraints"] = {"ca": bc.ca, "path_length": bc.path_length}
    except x509.ExtensionNotFound:
        pass
    try:
        skid = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_KEY_IDENTIFIER).value
        out["subject_key_identifier"] = skid.digest.hex()
    except x509.ExtensionNotFound:
        pass
    try:
        akid = cert.extensions.get_extension_for_oid(ExtensionOID.AUTHORITY_KEY_IDENTIFIER).value
        out["authority_key_identifier"] = akid.key_identifier.hex() if akid.key_identifier else None
    except x509.ExtensionNotFound:
        pass
    try:
        crl = cert.extensions.get_extension_for_oid(ExtensionOID.CRL_DISTRIBUTION_POINTS).value
        uris: list[str] = []
        for dp in crl:
            if dp.full_name:
                for n in dp.full_name:
                    uris.append(str(n.value))
        out["crl_distribution_points"] = uris
    except x509.ExtensionNotFound:
        pass
    return out


def expiration_info(cert: x509.Certificate) -> dict[str, Any]:
    now = datetime.now(timezone.utc)
    na = cert.not_valid_after.replace(tzinfo=timezone.utc)
    days = (na - now).days
    if days >= 0:
        return {"expired": False, "days_until_expiration": days, "days_since_expiration": 0}
    return {"expired": True, "days_until_expiration": 0, "days_since_expiration": abs(days)}


def analyze_cert(cert: x509.Certificate) -> dict[str, Any]:
    der = cert.public_bytes(serialization.Encoding.DER)
    sig_hash = cert.signature_hash_algorithm.name if cert.signature_hash_algorithm else "unknown"
    return {
        "version": cert.version.name,
        "serial_number": format(cert.serial_number, "x"),
        "issuer": name_to_str(cert.issuer),
        "subject": name_to_str(cert.subject),
        "validity": {
            "not_before": cert.not_valid_before.replace(tzinfo=timezone.utc).isoformat(),
            "not_after": cert.not_valid_after.replace(tzinfo=timezone.utc).isoformat(),
        },
        "expiration": expiration_info(cert),
        "public_key": key_info(cert.public_key()),
        "signature_algorithm": sig_hash,
        "fingerprints": {
            "sha1": hashlib.sha1(der).hexdigest(),
            "sha256": hashlib.sha256(der).hexdigest(),
        },
        "extensions": parse_extensions(cert),
    }


def verify_child_signed_by_issuer(child: x509.Certificate, issuer: x509.Certificate) -> bool:
    pub = issuer.public_key()
    try:
        if isinstance(pub, rsa.RSAPublicKey):
            pub.verify(child.signature, child.tbs_certificate_bytes, padding.PKCS1v15(), child.signature_hash_algorithm)
        elif isinstance(pub, ec.EllipticCurvePublicKey):
            pub.verify(child.signature, child.tbs_certificate_bytes, ec.ECDSA(child.signature_hash_algorithm))
        else:
            return False
        return True
    except InvalidSignature:
        return False


def validate_chain(certs: list[x509.Certificate]) -> dict[str, Any]:
    steps = []
    ok = True
    for i in range(len(certs) - 1):
        valid = verify_child_signed_by_issuer(certs[i], certs[i + 1])
        steps.append({
            "index": i,
            "child_subject": name_to_str(certs[i].subject),
            "issuer_subject": name_to_str(certs[i + 1].subject),
            "valid_signature": valid,
        })
        if not valid:
            ok = False
    return {"valid": ok, "steps": steps}


def generate_demo_chain() -> list[x509.Certificate]:
    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    now = datetime.now(timezone.utc)

    ca_subject = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, "Demo Root CA"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "TMLR Demo"),
    ])
    ca_cert = (
        x509.CertificateBuilder()
        .subject_name(ca_subject)
        .issuer_name(ca_subject)
        .public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(days=365))
        .add_extension(x509.BasicConstraints(ca=True, path_length=1), critical=True)
        .add_extension(x509.KeyUsage(
            digital_signature=False, content_commitment=False, key_encipherment=False, data_encipherment=False,
            key_agreement=False, key_cert_sign=True, crl_sign=True, encipher_only=False, decipher_only=False
        ), critical=True)
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
        .sign(private_key=ca_key, algorithm=hashes.SHA256())
    )

    leaf_subject = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, "demo.local"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "TMLR Demo"),
    ])
    leaf_cert = (
        x509.CertificateBuilder()
        .subject_name(leaf_subject)
        .issuer_name(ca_subject)
        .public_key(leaf_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(days=180))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .add_extension(x509.KeyUsage(
            digital_signature=True, content_commitment=False, key_encipherment=True, data_encipherment=False,
            key_agreement=False, key_cert_sign=False, crl_sign=False, encipher_only=False, decipher_only=False
        ), critical=True)
        .add_extension(x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
        .add_extension(x509.SubjectAlternativeName([x509.DNSName("demo.local"), x509.DNSName("www.demo.local")]), critical=False)
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False)
        .add_extension(x509.CRLDistributionPoints([
            x509.DistributionPoint(
                full_name=[x509.UniformResourceIdentifier("http://example.com/demo.crl")],
                relative_name=None,
                reasons=None,
                crl_issuer=None,
            )
        ]), critical=False)
        .sign(private_key=ca_key, algorithm=hashes.SHA256())
    )
    return [leaf_cert, ca_cert]


def render_text(result: dict[str, Any]) -> str:
    lines = []
    for i, cert in enumerate(result["certificates"], start=1):
        lines.append(f"Certificate #{i}")
        lines.append(f"  Subject: {cert['subject']}")
        lines.append(f"  Issuer : {cert['issuer']}")
        lines.append(f"  Version: {cert['version']}")
        lines.append(f"  Serial : {cert['serial_number']}")
        lines.append(f"  Validity: {cert['validity']['not_before']} -> {cert['validity']['not_after']}")
        lines.append(f"  Signature Algorithm: {cert['signature_algorithm']}")
        lines.append(f"  Public Key: {cert['public_key']['algorithm']} {cert['public_key']['size']}")
        lines.append(f"  SHA-1 : {cert['fingerprints']['sha1']}")
        lines.append(f"  SHA-256: {cert['fingerprints']['sha256']}")
        lines.append(
            f"  Expired: {cert['expiration']['expired']} | Days until: {cert['expiration']['days_until_expiration']} | Days since: {cert['expiration']['days_since_expiration']}"
        )
        lines.append(f"  SANs: {'; '.join(cert['extensions']['subject_alt_names'])}")
        lines.append(f"  Key Usage: {', '.join([k for k, v in cert['extensions']['key_usage'].items() if v])}")
        lines.append(f"  Extended Key Usage: {', '.join(cert['extensions']['extended_key_usage'])}")
        lines.append(f"  Basic Constraints: {cert['extensions']['basic_constraints']}")
        lines.append(f"  Subject Key ID: {cert['extensions']['subject_key_identifier']}")
        lines.append(f"  Authority Key ID: {cert['extensions']['authority_key_identifier']}")
        lines.append(f"  CRL DP: {'; '.join(cert['extensions']['crl_distribution_points'])}")
        lines.append("")
    lines.append(f"Chain validation: {result['chain_validation']['valid']}")
    for s in result["chain_validation"]["steps"]:
        lines.append(f"  [{s['index']}] {'OK' if s['valid_signature'] else 'FAIL'} :: {s['child_subject']} <- {s['issuer_subject']}")
    return "\n".join(lines) + "\n"


def render_csv(result: dict[str, Any]) -> str:
    from io import StringIO
    buf = StringIO()
    writer = csv.writer(buf)
    writer.writerow([
        "index", "subject", "issuer", "serial_number", "not_before", "not_after", "expired",
        "days_until_expiration", "days_since_expiration", "public_key_algorithm", "public_key_size",
        "signature_algorithm", "fingerprint_sha1", "fingerprint_sha256",
    ])
    for i, cert in enumerate(result["certificates"], start=1):
        writer.writerow([
            i, cert["subject"], cert["issuer"], cert["serial_number"], cert["validity"]["not_before"], cert["validity"]["not_after"],
            cert["expiration"]["expired"], cert["expiration"]["days_until_expiration"], cert["expiration"]["days_since_expiration"],
            cert["public_key"]["algorithm"], cert["public_key"]["size"], cert["signature_algorithm"],
            cert["fingerprints"]["sha1"], cert["fingerprints"]["sha256"],
        ])
    return buf.getvalue()


def main() -> int:
    args = parse_args()
    if args.host:
        host, *port = args.host.split(":")
        p = int(port[0]) if port else args.port
        certs = fetch_host_cert(host, p)
        source = f"host:{host}:{p}"
    elif args.path:
        certs = load_certs_from_file(Path(args.path).resolve())
        source = str(Path(args.path).resolve())
    else:
        certs = generate_demo_chain()
        source = "demo-generated"

    result = {
        "source": source,
        "certificate_count": len(certs),
        "certificates": [analyze_cert(c) for c in certs],
        "chain_validation": validate_chain(certs),
    }
    if args.format == "json":
        out = json.dumps(result, indent=2)
    elif args.format == "csv":
        out = render_csv(result)
    else:
        out = render_text(result)

    if args.output:
        out_path = Path(args.output).resolve()
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(out + ("" if out.endswith("\n") else "\n"), encoding="utf-8")
    print(out, end="" if out.endswith("\n") else "\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())