F5 Certificate Expiry Audit

#!/usr/bin/env python3
"""
F5 Certificate Expiry Audit

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/f5-cert-audit

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "requests>=2.28" "PyYAML>=6.0"
"""

import argparse
import datetime
import os
import sys

import requests
import urllib3
import yaml

sys.stdout.reconfigure(encoding="utf-8", errors="replace")


# --- iControl REST client -------------------------------------------------
#
# Inlined verbatim rather than imported: these scripts download standalone
# from the site, so there is no shared module to import. The same block
# appears in every F5 script here; change it in all of them or in none.

def credentials():
    user = os.environ.get("F5_USER")
    password = os.environ.get("F5_PASS")
    if not user or not password:
        raise SystemExit(
            "ERROR: F5_USER and F5_PASS must be set in the environment.\n"
            "\n"
            "  Linux / macOS:   export F5_USER=svc_automation\n"
            "                   export F5_PASS='...'\n"
            "  PowerShell:      $env:F5_USER='svc_automation'\n"
            "                   $env:F5_PASS='...'\n"
            "\n"
            "They are read from the environment on purpose: an argument is\n"
            "visible in `ps auxww` while the process runs and lands in shell\n"
            "history."
        )
    return user, password


class BigIP:
    """Minimal iControl REST client.

    Token auth rather than Basic on every call: a token is scoped, expires,
    and does not put the password on the wire repeatedly.
    """

    def __init__(self, host, user, password, verify=True, timeout=30):
        self.base = "https://%s/mgmt" % host
        self.host = host
        self.timeout = timeout
        self.s = requests.Session()
        self.s.verify = verify
        if not verify:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        self._login(user, password)

    def _login(self, user, password):
        r = self.s.post(
            self.base + "/shared/authn/login",
            json={"username": user, "password": password, "loginProviderName": "tmos"},
            timeout=self.timeout,
        )
        if r.status_code == 401:
            raise SystemExit("ERROR: %s rejected the credentials (401)." % self.host)
        r.raise_for_status()
        token = r.json()["token"]["token"]
        self.s.headers.update({"X-F5-Auth-Token": token})

    def get(self, path, **kw):
        r = self.s.get(self.base + path, timeout=self.timeout, **kw)
        r.raise_for_status()
        return r.json()

    def patch(self, path, body):
        r = self.s.patch(self.base + path, json=body, timeout=self.timeout)
        r.raise_for_status()
        return r.json()

    def post(self, path, body):
        r = self.s.post(self.base + path, json=body, timeout=self.timeout)
        r.raise_for_status()
        return r.json()


def load_inventory(path):
    """Read the device inventory. Credentials never live in it."""
    try:
        with open(path, "r", encoding="utf-8") as fh:
            doc = yaml.safe_load(fh) or {}
    except FileNotFoundError:
        raise SystemExit("ERROR: inventory file not found: %s" % path)
    except yaml.YAMLError as exc:
        raise SystemExit("ERROR: %s is not valid YAML: %s" % (path, exc))

    defaults = doc.get("defaults") or {}
    devices = doc.get("devices")
    if not devices:
        raise SystemExit(
            "ERROR: %s has no `devices:` list.\n"
            "\n"
            "  defaults:\n"
            "    verify_ssl: true\n"
            "  devices:\n"
            "    - host: ltm01.example.com\n"
            "    - host: 10.0.0.10\n"
            "      verify_ssl: false" % path
        )
    out = []
    for i, d in enumerate(devices, 1):
        merged = dict(defaults)
        merged.update(d or {})
        if not merged.get("host"):
            raise SystemExit("ERROR: device #%d in %s has no `host`." % (i, path))
        out.append(merged)
    return out


def encode(name):
    """tmsh paths use ~ for / in the REST URI: /Common/pool -> ~Common~pool"""
    return name.replace("/", "~")

# --- certificate audit ----------------------------------------------------


def parse_expiry(epoch):
    try:
        return datetime.datetime.fromtimestamp(int(epoch), datetime.timezone.utc)
    except (TypeError, ValueError):
        return None


def ssl_profiles_using(bip):
    """Map certificate name -> the client-SSL profiles that reference it."""
    used = {}
    data = bip.get("/tm/ltm/profile/client-ssl")
    for prof in data.get("items", []):
        names = []
        if prof.get("cert"):
            names.append(prof["cert"])
        for entry in prof.get("certKeyChain", []) or []:
            if entry.get("cert"):
                names.append(entry["cert"])
        for n in names:
            used.setdefault(n, []).append(prof["fullPath"])
    return used


def virtuals_using(bip, profiles):
    """Map client-SSL profile -> the virtual servers that attach it."""
    hits = {}
    data = bip.get("/tm/ltm/virtual")
    for vs in data.get("items", []):
        ref = vs.get("profilesReference", {}).get("link")
        if not ref:
            continue
        path = ref.split("/mgmt")[-1].split("?")[0]
        try:
            attached = bip.get(path)
        except requests.exceptions.RequestException:
            continue
        for p in attached.get("items", []):
            if p.get("fullPath") in profiles:
                hits.setdefault(p["fullPath"], []).append(vs["fullPath"])
    return hits


def main():
    ap = argparse.ArgumentParser(
        description="Audit certificates on BIG-IPs: what expires when, and "
                    "which client-SSL profiles and virtual servers would "
                    "break. Read-only.")
    ap.add_argument("--inventory", required=True, help="YAML inventory of BIG-IPs")
    ap.add_argument("--days", type=int, default=30,
                    help="Warn below this many days remaining (default 30)")
    ap.add_argument("--critical-days", type=int, default=7,
                    help="Treat as critical below this many days (default 7)")
    ap.add_argument("--unused", action="store_true",
                    help="Also list certificates no client-SSL profile references")
    args = ap.parse_args()

    user, password = credentials()
    devices = load_inventory(args.inventory)
    now = datetime.datetime.now(datetime.timezone.utc)
    worst = 0

    for dev in devices:
        host = dev["host"]
        print("=" * 72)
        print(host)
        print("=" * 72)
        try:
            bip = BigIP(host, user, password,
                        verify=dev.get("verify_ssl", True),
                        timeout=dev.get("timeout", 60))
            certs = bip.get("/tm/sys/crypto/cert").get("items", [])
            used = ssl_profiles_using(bip)
            vs_map = virtuals_using(bip, set(p for ps in used.values() for p in ps))

            rows, unused = [], []
            for c in certs:
                name = c.get("fullPath", c.get("name", "?"))
                exp = parse_expiry(c.get("expirationDate"))
                profiles = used.get(name, [])
                if not profiles:
                    unused.append(name)
                if exp is None:
                    print("  COULD NOT VERIFY  %-44s no expiry in the API response" % name)
                    continue
                days = (exp - now).days
                if days <= args.days:
                    rows.append((days, name, exp, profiles))

            rows.sort()
            if not rows:
                print("  No certificate expires within %d days." % args.days)
            for days, name, exp, profiles in rows:
                sev = "CRITICAL" if days <= args.critical_days else "WARNING"
                if days < 0:
                    sev = "EXPIRED"
                print("  [%-8s] %-44s %s (%d days)"
                      % (sev, name, exp.strftime("%Y-%m-%d"), days))
                if profiles:
                    for p in profiles:
                        vss = vs_map.get(p, [])
                        print("             profile %s" % p)
                        for v in vss:
                            print("               -> virtual server %s" % v)
                else:
                    print("             not referenced by any client-SSL profile")
                worst = max(worst, 2 if sev in ("CRITICAL", "EXPIRED") else 1)

            if args.unused:
                print()
                print("  Certificates no client-SSL profile references (%d):" % len(unused))
                for n in sorted(unused):
                    print("    %s" % n)
                print("  NOTE: this only checks client-SSL profiles. A certificate")
                print("  can still be used by server-SSL, LDAP, SAML or device trust.")
                print("  Treat the list as candidates to investigate, not to delete.")
        except requests.exceptions.SSLError as exc:
            print("  ERROR: TLS failed: %s" % exc)
            worst = 2
        except requests.exceptions.RequestException as exc:
            print("  ERROR: %s" % exc)
            worst = 2
        print()

    return worst


if __name__ == "__main__":
    sys.exit(main())