F5 UCS Backup

#!/usr/bin/env python3
"""
F5 UCS Backup

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

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("/", "~")

# --- UCS backup -----------------------------------------------------------

CHUNK = 512 * 1024


def create_ucs(bip, name, dry_run):
    if dry_run:
        print("  [dry-run] would create UCS %s" % name)
        return False
    bip.post("/tm/sys/ucs", {"command": "save", "name": name})
    return True


def download_ucs(bip, name, dest):
    """Pull the archive OFF the device.

    This is the whole point. A UCS sitting on the BIG-IP is not a backup: if
    the reason you need it is that the device is gone, so is the file.
    """
    uri = bip.base + "/shared/file-transfer/ucs-downloads/" + name
    start, total = 0, None
    with open(dest, "wb") as fh:
        while True:
            end = start + CHUNK - 1
            headers = {"Content-Range": "%d-%d/%s" % (start, end, total or "*")}
            r = bip.s.get(uri, headers=headers, timeout=bip.timeout, stream=True)
            if r.status_code == 400 and total is not None and start >= total:
                break
            r.raise_for_status()
            crange = r.headers.get("Content-Range", "")
            if "/" in crange:
                total = int(crange.split("/")[-1])
            written = 0
            for block in r.iter_content(64 * 1024):
                fh.write(block)
                written += len(block)
            start += written
            if written == 0 or (total is not None and start >= total):
                break
    return os.path.getsize(dest)


def prune(outdir, host, keep):
    """Keep the newest N archives per device, delete the rest."""
    mine = sorted(
        (f for f in os.listdir(outdir)
         if f.startswith(host + "_") and f.endswith(".ucs")),
        reverse=True,
    )
    removed = 0
    for old in mine[keep:]:
        try:
            os.remove(os.path.join(outdir, old))
            removed += 1
        except OSError as exc:
            print("  WARNING: could not remove %s: %s" % (old, exc))
    return removed


def main():
    ap = argparse.ArgumentParser(
        description="Create a UCS archive on each BIG-IP and download it off "
                    "the device. A UCS that only exists on the box it came "
                    "from is not a backup.")
    ap.add_argument("--inventory", required=True, help="YAML inventory of BIG-IPs")
    ap.add_argument("--out", default="./f5-backups", help="Local directory (default ./f5-backups)")
    ap.add_argument("--keep", type=int, default=7, help="Archives to keep per device (default 7)")
    ap.add_argument("--dry-run", action="store_true", help="Change nothing")
    args = ap.parse_args()

    user, password = credentials()
    devices = load_inventory(args.inventory)
    if not args.dry_run:
        os.makedirs(args.out, exist_ok=True)

    stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    worst, ok = 0, 0

    for dev in devices:
        host = dev["host"]
        print("=" * 72)
        print(host)
        print("=" * 72)
        name = "%s_%s.ucs" % (host, stamp)
        dest = os.path.join(args.out, name)
        try:
            bip = BigIP(host, user, password,
                        verify=dev.get("verify_ssl", True),
                        timeout=dev.get("timeout", 300))
            ver = bip.get("/tm/sys/version")
            print("  version query ok")
            del ver
            if not create_ucs(bip, name, args.dry_run):
                print()
                continue
            print("  created on device: %s" % name)
            size = download_ucs(bip, name, dest)
            print("  downloaded: %s (%.1f MB)" % (dest, size / 1048576.0))
            if size < 1024:
                print("  ERROR: archive is suspiciously small. Treat as FAILED.")
                worst = 2
                continue
            n = prune(args.out, host, args.keep)
            if n:
                print("  pruned %d old archive(s), keeping %d" % (n, args.keep))
            ok += 1
        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
        except OSError as exc:
            print("  ERROR: could not write %s: %s" % (dest, exc))
            worst = 2
        print()

    print("%d of %d device(s) backed up." % (ok, len(devices)))
    if ok == 0 and devices:
        return 2
    return worst


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