Decode traefik acme.json for inspection

certs.py

#!/usr/bin/env python3
"""
traefik_acme_certs.py

List Traefik ACME certificates from acme.json:
- Domains (CN + SANs)
- Serial Number
- Issued (UTC)
- Expires (UTC)
- Days Left

Requires: pip install cryptography
"""

from __future__ import annotations

import argparse
import base64
import json
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, List, Optional, Set, Tuple

from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.x509.oid import NameOID


@dataclass(frozen=True)
class CertInfo:
    domains: Tuple[str, ...]
    serial: str
    issued: datetime
    not_after: datetime

    @property
    def days_left(self) -> int:
        now = datetime.now(timezone.utc)
        return int((self.not_after - now).total_seconds() // 86400)


def _to_utc(dt: datetime) -> datetime:
    return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc)


def _fmt_dt(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")


def _b64_decode_flexible(s: str) -> Optional[bytes]:
    t = "".join(s.strip().split())
    if not t:
        return None
    pad = (-len(t)) % 4
    t_padded = t + ("=" * pad)

    try:
        return base64.b64decode(t_padded, validate=False)
    except Exception:
        try:
            return base64.urlsafe_b64decode(t_padded.encode("ascii"))
        except Exception:
            return None


def _load_cert(decoded: bytes) -> Optional[x509.Certificate]:
    if not decoded:
        return None

    if b"-----BEGIN CERTIFICATE-----" in decoded[:200]:
        try:
            return x509.load_pem_x509_certificate(decoded, default_backend())
        except Exception:
            return None

    try:
        return x509.load_der_x509_certificate(decoded, default_backend())
    except Exception:
        try:
            return x509.load_pem_x509_certificate(decoded, default_backend())
        except Exception:
            return None


def _domains_from_cert(cert: x509.Certificate) -> Tuple[str, ...]:
    domains: Set[str] = set()

    try:
        attrs = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
        if attrs:
            domains.add(str(attrs[0].value).strip())
    except Exception:
        pass

    try:
        ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
        for name in ext.value.get_values_for_type(x509.DNSName):
            domains.add(str(name).strip())
    except Exception:
        pass

    return tuple(sorted(d for d in domains if d))


def _extract_traefik_cert_blobs(acme: Any) -> List[str]:
    blobs: List[str] = []
    if not isinstance(acme, dict):
        return blobs

    for resolver_obj in acme.values():
        if not isinstance(resolver_obj, dict):
            continue
        certs = resolver_obj.get("Certificates")
        if not isinstance(certs, list):
            continue
        for entry in certs:
            if isinstance(entry, dict):
                cert_blob = entry.get("certificate")
                if isinstance(cert_blob, str) and len(cert_blob) > 100:
                    blobs.append(cert_blob)
    return blobs


def _dedupe(certs: List[CertInfo]) -> List[CertInfo]:
    seen = set()
    out: List[CertInfo] = []
    for c in certs:
        key = (c.domains, c.serial)
        if key in seen:
            continue
        seen.add(key)
        out.append(c)
    return out


def _format_serial(cert: x509.Certificate) -> str:
    # Format as colon-separated uppercase hex (OpenSSL style)
    serial_int = cert.serial_number
    hex_str = f"{serial_int:X}"
    if len(hex_str) % 2:
        hex_str = "0" + hex_str
    return ":".join(hex_str[i:i+2] for i in range(0, len(hex_str), 2))


def _print_table(certs: List[CertInfo], max_domain_col: int = 80) -> None:
    certs = sorted(certs, key=lambda c: c.not_after)

    header = ("Domains", "Serial", "Issued (UTC)", "Expires (UTC)", "Days Left")
    rows = []

    for c in certs:
        dom = ", ".join(c.domains) if c.domains else "(no domains)"
        if len(dom) > max_domain_col:
            dom = dom[: max_domain_col - 3] + "..."
        rows.append(
            (
                dom,
                c.serial,
                _fmt_dt(c.issued),
                _fmt_dt(c.not_after),
                str(c.days_left),
            )
        )

    cols = list(zip(*([header] + rows)))
    widths = [max(len(str(x)) for x in col) for col in cols]

    def fmt(parts):
        return "  ".join(str(p).ljust(w) for p, w in zip(parts, widths))

    print(fmt(header))
    print("  ".join("-" * w for w in widths))
    for r in rows:
        print(fmt(r))


def main() -> int:
    ap = argparse.ArgumentParser(description="List Traefik ACME certs from acme.json.")
    ap.add_argument("--acme-file", required=True, help="Host path to Traefik acme.json")
    args = ap.parse_args()

    path = Path(args.acme_file)
    if not path.is_file():
        print(f"ERROR: not a file: {path}", file=sys.stderr)
        return 2

    acme = json.loads(path.read_text(encoding="utf-8"))
    blobs = _extract_traefik_cert_blobs(acme)

    if not blobs:
        print("No certificate entries found in acme.json", file=sys.stderr)
        return 1

    parsed: List[CertInfo] = []

    for b64_blob in blobs:
        decoded = _b64_decode_flexible(b64_blob)
        if not decoded:
            continue

        cert = _load_cert(decoded)
        if not cert:
            continue

        parsed.append(
            CertInfo(
                domains=_domains_from_cert(cert),
                serial=_format_serial(cert),
                issued=_to_utc(cert.not_valid_before),
                not_after=_to_utc(cert.not_valid_after),
            )
        )

    parsed = _dedupe(parsed)

    if not parsed:
        print("Found certificate blobs but could not parse any certificates.", file=sys.stderr)
        return 1

    _print_table(parsed)
    return 0


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