#!/bin/bash

show_help() {
cat <<'EOF'

  Script:        scfanalyzerv2.sh
  Description:   Safe analyzer and fixer for the CSF firewall deny list.

  Purpose:
    This script scans /etc/csf/csf.deny and performs two main functions:

    1. ANALYZE Mode (default):
       - Finds IPv4 /24 and IPv6 /64 networks with THRESHOLD or more individual IPs.
       - Reports duplicate IP and subnet entries.
       - Reports individual IPs already covered by an existing subnet block.
       - Reports entries older than the configured age.
       - Reports entries without a date stamp.
       - Preserves blank lines, comment-only lines, tcp|/udp| advanced rules,
         and entries containing "do not delete".

    2. FIX Mode (-fix):
       - Creates IPv4 /24 and IPv6 /64 aggregate subnet blocks.
       - Auto-generated aggregate subnet blocks receive the current date and expire
         normally after the configured age; they are NOT marked "do not delete".
       - Removes redundant individual IPs covered by existing or newly-created subnets.
       - Removes duplicate non-protected IP/subnet entries.
       - Adds a current date stamp to entries that are missing one.
       - Removes non-protected entries older than the configured age.
       - Creates a verified backup before any replacement.
       - Keeps only the 10 most recent backups in /etc/csf/backups.
       - Builds a temporary file, fsyncs it, preserves owner/mode, and atomically
         replaces csf.deny only after all processing succeeds.
       - Aborts if csf.deny changes externally while FIX is being prepared.
       - Reloads CSF quietly; normal CSF reload chatter is hidden.
       - If the CSF reload fails, the error and captured CSF output are shown.

  Requirements:
       - Bash 4.0+
       - Python 3.6+ (standard-library ipaddress module)
       - flock

  Usage:
       /root/scfanalyzerv2.sh             -> threshold=3, age=60 days
       /root/scfanalyzerv2.sh 2           -> threshold=2, age=60 days
       /root/scfanalyzerv2.sh 2 30        -> threshold=2, age=30 days
       /root/scfanalyzerv2.sh -fix        -> FIX with defaults
       /root/scfanalyzerv2.sh -fix 2      -> FIX with threshold=2
       /root/scfanalyzerv2.sh -fix 2 30   -> FIX with threshold=2, age=30
       /root/scfanalyzerv2.sh -r          -> CSF optimization notes
       /root/scfanalyzerv2.sh -h          -> this help

  Cron examples:
       # Analyze every 6 hours
       0 */6 * * * /root/scfanalyzerv2.sh >> /var/log/scfanalyzer.log 2>&1

       # Fix daily at 3:30 AM
       30 3 * * * /root/scfanalyzerv2.sh -fix >> /var/log/scfanalyzer.log 2>&1

  Safety notes:
       - Run ANALYZE mode first when changing threshold/age values.
       - FIX mode never intentionally modifies entries you manually mark "do not delete".
       - Auto-generated aggregate subnet blocks are intentionally temporary and age out.
       - The script preserves unrecognized/non-IP lines rather than guessing.
       - If another process changes csf.deny during FIX preparation, FIX aborts.

  Author:        Emilius
  Version:       3.04-py36
  Date Updated:  2026-09-02
  Tested for:    CWP + CSF (CentOS / AlmaLinux)

EOF
exit 0
}

DENY_FILE="${DENY_FILE:-/etc/csf/csf.deny}"
BACKUP_DIR="${BACKUP_DIR:-/etc/csf/backups}"
THRESHOLD=3
DAYS_OLD=60
MODE="ANALYZE"
SHOW_RECOMMENDATIONS=0

# Parse arguments. Numeric arguments are threshold first, age second.
numeric_seen=0
while (($#)); do
    case "$1" in
        -h|--help)
            show_help
            ;;
        -v|--version)
            echo "scfanalyzerv2.sh 3.04-py36"
            exit 0
            ;;
        -fix|--fix)
            MODE="FIX"
            ;;
        -r|--recommendations)
            SHOW_RECOMMENDATIONS=1
            ;;
        ''|*[!0-9]*)
            echo "Unknown argument: $1" >&2
            echo "Use -h for help." >&2
            exit 2
            ;;
        *)
            if (( numeric_seen == 0 )); then
                THRESHOLD="$1"
            elif (( numeric_seen == 1 )); then
                DAYS_OLD="$1"
            else
                echo "Too many numeric arguments. Use: [threshold] [days-old]" >&2
                exit 2
            fi
            ((numeric_seen++))
            ;;
    esac
    shift
done

if (( SHOW_RECOMMENDATIONS )); then
    cat <<'EOF'

🛡️ CSF Optimization Notes:
 - DENY_IP_LIMIT = 20000                  # Increase only if the server has enough memory and your policy needs it
 - LF_IPSET = 1                           # Prefer ipset for large deny lists when supported by your CSF installation
 - LF_IPSET_HASHSIZE = 4096               # Reasonable starting point for thousands of entries; size for your workload
 - LF_PERMBLOCK = 1                       # Permanently block repeat offenders when this matches your policy
 - LF_PERMBLOCK_INTERVAL = 21600          # Example: 6-hour repeat-offender observation window
 - LF_PERMBLOCK_COUNT = 2                 # Example: permanent block after two qualifying temporary blocks
 - PORTFLOOD = 22;tcp;5;60,25;tcp;10;60   # Example only; tune to your actual SSH/SMTP traffic

Important:
 - LF_SELECT controls selective/port-specific blocking behavior; do not enable it solely to reduce logging.
 - LF_DSHIELD and LF_SPAMHAUS behavior/values vary by CSF version. Read the comments in your installed
   /etc/csf/csf.conf before changing them; do not assume that "1" is the correct enable value.
 - Keep a copy of csf.conf before changing firewall behavior and reload with: csf -r

EOF
    exit 0
fi

if [[ "${BASH_VERSINFO:-0}" -lt 4 ]]; then
    echo "This script requires Bash 4.0 or higher." >&2
    exit 1
fi

if ! command -v python3 >/dev/null 2>&1; then
    echo "python3 is required for safe IPv4/IPv6 parsing." >&2
    exit 1
fi

# This build intentionally supports the Python 3.6 shipped on many older CWP systems.
# Keep the embedded helper compatible with Python 3.6; avoid newer-only syntax/APIs.
if ! python3 -c 'import sys; sys.exit(0 if sys.version_info >= (3,6) else 1)'; then
    echo "Python 3.6 or newer is required." >&2
    exit 1
fi

if ! command -v flock >/dev/null 2>&1; then
    echo "flock is required." >&2
    exit 1
fi

if [[ ! -f "$DENY_FILE" ]]; then
    echo "Deny file not found: $DENY_FILE" >&2
    exit 1
fi

if (( THRESHOLD < 1 )); then
    echo "Threshold must be at least 1." >&2
    exit 2
fi

# Acquire our script lock before FIX analysis starts. This prevents two copies of
# scfanalyzerv2 from racing each other. The Python side additionally checks that
# csf.deny itself did not change externally before the atomic replacement.
if [[ "$MODE" == "FIX" ]]; then
    exec 200>"${DENY_FILE}.scfanalyzerv2.lock"
    if ! flock -n 200; then
        echo "Could not acquire scfanalyzerv2 lock for $DENY_FILE; another FIX may be running." >&2
        exit 1
    fi
fi

START_TIME=$(date +%s.%N)
export DENY_FILE BACKUP_DIR THRESHOLD DAYS_OLD MODE START_TIME

python3 <<'PY'
import datetime as dt
import hashlib
import ipaddress
import os
import random
import re
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from pathlib import Path

DENY_FILE = Path(os.environ["DENY_FILE"])
BACKUP_DIR = Path(os.environ["BACKUP_DIR"])
THRESHOLD = int(os.environ["THRESHOLD"])
DAYS_OLD = int(os.environ["DAYS_OLD"])
MODE = os.environ["MODE"]
START_TIME = float(os.environ["START_TIME"])

DATE_RE = re.compile(r"[A-Z][a-z]{2} [A-Z][a-z]{2} [ 0-9][0-9] [0-9]{2}:[0-9]{2}:[0-9]{2} [0-9]{4}")
ADVANCED_RE = re.compile(r"^(?:tcp|udp)\|", re.IGNORECASE)
DO_NOT_DELETE_RE = re.compile(r"\bdo\s+not\s+delete\b", re.IGNORECASE)


class Entry(object):
    def __init__(self, idx, original, newline, token=None, obj=None, key=None,
                 protected=False, metadata=False, comment=""):
        self.idx = idx
        self.original = original
        self.newline = newline
        self.token = token
        self.obj = obj
        self.key = key
        self.protected = protected
        self.metadata = metadata
        self.old = False
        self.duplicate = False
        self.redundant = False
        self.missing_date = False
        self.comment = comment


def plural(n, one, many=None):
    return one if n == 1 else (many if many is not None else one + "s")


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def parse_last_date(line):
    matches = DATE_RE.findall(line)
    if not matches:
        return None
    try:
        return dt.datetime.strptime(matches[-1], "%a %b %d %H:%M:%S %Y")
    except ValueError:
        return None


def strip_date_from_comment(comment):
    comment = DATE_RE.sub("", comment)
    comment = re.sub(r"\s*[-–—]+\s*$", "", comment).strip()
    return comment


def address_token(line):
    stripped = line.strip()
    if not stripped or stripped.startswith("#") or ADVANCED_RE.match(stripped):
        return None
    return stripped.split(None, 1)[0]


def parse_ip_token(token):
    try:
        if "/" in token:
            return ipaddress.ip_network(token, strict=False)
        return ipaddress.ip_address(token)
    except ValueError:
        return None


def covered_address(ip, network_sets_by_prefix):
    """Return an existing network containing an address, preferring the most specific."""
    fam = ip.version
    prefixes = sorted(network_sets_by_prefix[fam].keys(), reverse=True)
    for prefix in prefixes:
        candidate = ipaddress.ip_network("{0}/{1}".format(ip, prefix), strict=False)
        if candidate in network_sets_by_prefix[fam][prefix]:
            return candidate
    return None


def covering_network(net, network_sets_by_prefix):
    """Return an existing network that fully covers net, preferring the most specific."""
    fam = net.version
    prefixes = sorted(network_sets_by_prefix[fam].keys(), reverse=True)
    for prefix in prefixes:
        if prefix > net.prefixlen:
            continue
        candidate = ipaddress.ip_network("{0}/{1}".format(net.network_address, prefix), strict=False)
        if candidate in network_sets_by_prefix[fam][prefix]:
            return candidate
    return None


def append_date(line, stamp):
    # Preserve existing content; avoid producing "# - DATE" when a line ends with '#'.
    line = line.rstrip()
    if line.endswith("#"):
        return line[:-1].rstrip() + " - " + stamp
    return line + " - " + stamp


def safe_unlink(path):
    try:
        path.unlink()
    except OSError:
        pass


def print_timing(total):
    elapsed = time.time() - START_TIME
    if elapsed >= 60:
        print("Total entries scanned: {0} in {1:.2f} minutes".format(total, elapsed / 60.0))
    else:
        print("Total entries scanned: {0} in {1:.2f} seconds".format(total, elapsed))


try:
    original_bytes = DENY_FILE.read_bytes()
except OSError as exc:
    print("Unable to read {0}: {1}".format(DENY_FILE, exc), file=sys.stderr)
    sys.exit(1)

initial_hash = sha256_bytes(original_bytes)
raw_lines = original_bytes.decode("utf-8", errors="surrogateescape").splitlines(keepends=True)
entries = []
now = dt.datetime.now()

for idx, raw in enumerate(raw_lines):
    if raw.endswith("\r\n"):
        line, newline = raw[:-2], "\r\n"
    elif raw.endswith("\n"):
        line, newline = raw[:-1], "\n"
    else:
        line, newline = raw, ""

    stripped = line.strip()
    token = address_token(line)
    obj = parse_ip_token(token) if token else None
    metadata = (not stripped) or stripped.startswith("#") or bool(ADVANCED_RE.match(stripped)) or obj is None
    protected = bool(DO_NOT_DELETE_RE.search(line)) or metadata
    key = None
    if obj is not None:
        key = "{0}:{1}".format(obj.version, obj.compressed)
    comment = ""
    if "#" in line:
        comment = line.split("#", 1)[1].strip()

    e = Entry(idx, line, newline, token=token, obj=obj, key=key,
              protected=protected, metadata=metadata, comment=comment)

    # Protected/metadata lines are never aged out.
    if obj is not None and not protected:
        last_date = parse_last_date(line)
        if last_date is not None:
            age_days = int((now - last_date).total_seconds() // 86400)
            if age_days > DAYS_OLD:
                e.old = True
    entries.append(e)

# Duplicate handling: protected keys win. Otherwise the first surviving occurrence wins.
protected_keys = set(e.key for e in entries if e.key and e.protected and not e.old)
seen = set()
duplicated_ips = []
duplicated_subnets = []
for e in entries:
    if e.obj is None or e.old or e.protected:
        continue
    if e.key in protected_keys or e.key in seen:
        e.duplicate = True
        if isinstance(e.obj, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
            duplicated_subnets.append(e)
        else:
            duplicated_ips.append(e)
    else:
        seen.add(e.key)

# Build efficient lookup for all existing surviving subnet blocks, including protected ones.
network_sets_by_prefix = {4: {}, 6: {}}
for e in entries:
    if e.obj is None or e.old or e.duplicate:
        continue
    if isinstance(e.obj, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
        network_sets_by_prefix[e.obj.version].setdefault(e.obj.prefixlen, set()).add(e.obj)

# Existing CIDR blocks make individual IP entries redundant immediately.
covered_existing = {}
for e in entries:
    if e.obj is None or e.old or e.duplicate or e.protected:
        continue
    if isinstance(e.obj, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
        covering = covered_address(e.obj, network_sets_by_prefix)
        if covering is not None:
            e.redundant = True
            covered_existing[e.idx] = covering

# Aggregate remaining modifiable IPs by IPv4 /24 and IPv6 /64.
groups = {}
for e in entries:
    if e.obj is None or e.old or e.duplicate or e.redundant or e.protected:
        continue
    if isinstance(e.obj, ipaddress.IPv4Address):
        net = ipaddress.ip_network("{0}/24".format(e.obj), strict=False)
    elif isinstance(e.obj, ipaddress.IPv6Address):
        net = ipaddress.ip_network("{0}/64".format(e.obj), strict=False)
    else:
        continue
    groups.setdefault(net, []).append(e)

new_networks = []
for net, members in sorted(groups.items(), key=lambda x: (x[0].version, int(x[0].network_address), x[0].prefixlen)):
    if len(members) < THRESHOLD:
        continue
    existing_cover = covering_network(net, network_sets_by_prefix)
    if existing_cover is not None:
        for e in members:
            e.redundant = True
            covered_existing[e.idx] = existing_cover
        continue
    new_networks.append((net, members))
    for e in members:
        e.redundant = True

stamp = time.strftime("%a %b %d %H:%M:%S %Y")
for e in entries:
    if e.obj is None or e.old or e.duplicate or e.redundant or e.protected:
        continue
    if parse_last_date(e.original) is None:
        e.missing_date = True

old_entries = [e for e in entries if e.old]
redundant_entries = [e for e in entries if e.redundant]
missing_date_entries = [e for e in entries if e.missing_date]

fix_count = (
    len(old_entries)
    + len(duplicated_ips)
    + len(duplicated_subnets)
    + len(new_networks)
    + len(redundant_entries)
    + len(missing_date_entries)
)

print("=== {0} mode enabled: scanning {1} ===".format(MODE, DENY_FILE))

phrase_remove = "Removing" if MODE == "FIX" else "Found"
phrase_add = "Adding" if MODE == "FIX" else "Found"
phrase_fix = "Fixing" if MODE == "FIX" else "Found"

if old_entries:
    print("{0} {1} {2} older than {3} {4}".format(
        phrase_remove, len(old_entries), plural(len(old_entries), "entry", "entries"),
        DAYS_OLD, plural(DAYS_OLD, "day", "days")))
    for e in old_entries:
        print("       {0}".format(e.token))
elif MODE != "FIX":
    print("Found 0 entries older than {0} days".format(DAYS_OLD))

if duplicated_ips:
    print("{0} {1} {2}".format(phrase_remove, len(duplicated_ips),
                               plural(len(duplicated_ips), "duplicated IP", "duplicated IPs")))
    for e in duplicated_ips:
        print("       {0}".format(e.token))
elif MODE != "FIX":
    print("Found 0 duplicated IPs")

if duplicated_subnets:
    print("{0} {1} {2}".format(phrase_remove, len(duplicated_subnets),
                               plural(len(duplicated_subnets), "duplicated subnet block", "duplicated subnet blocks")))
    for e in duplicated_subnets:
        print("       {0}".format(e.token))
elif MODE != "FIX":
    print("Found 0 duplicated subnet blocks")

if new_networks:
    print("{0} {1} {2} with {3} or more {4}".format(
        phrase_add, len(new_networks), plural(len(new_networks), "subnet", "subnets"),
        THRESHOLD, plural(THRESHOLD, "entry", "entries")))
    for net, members in new_networks:
        print("       {0:<30} ({1} IPs)".format(str(net), len(members)))
elif MODE != "FIX":
    print("Found 0 subnets with {0} or more entries".format(THRESHOLD))

if redundant_entries:
    print("{0} {1} redundant {2}".format(
        phrase_remove, len(redundant_entries), plural(len(redundant_entries), "IP", "IPs")))
    for e in redundant_entries:
        if e.idx in covered_existing:
            print("       {0}  (covered by {1})".format(e.token, covered_existing[e.idx]))
        else:
            print("       {0}".format(e.token))
elif MODE != "FIX":
    print("Found 0 redundant IPs")

if missing_date_entries:
    print("{0} {1} {2} without date stamp".format(
        phrase_fix, len(missing_date_entries), plural(len(missing_date_entries), "entry", "entries")))
    for e in missing_date_entries:
        print("       {0}".format(e.token))
elif MODE != "FIX":
    print("Found 0 entries without date stamp")

if MODE == "FIX" and fix_count:
    try:
        if not BACKUP_DIR.exists():
            BACKUP_DIR.mkdir(parents=True)
    except OSError as exc:
        print("ERROR: cannot create backup directory {0}: {1}".format(BACKUP_DIR, exc), file=sys.stderr)
        sys.exit(1)

    backup = BACKUP_DIR / "{0}.bak.{1}".format(DENY_FILE.name, time.strftime("%Y-%m-%d_%H-%M-%S"))
    try:
        shutil.copy2(str(DENY_FILE), str(backup))
        if not backup.is_file() or backup.stat().st_size != len(original_bytes):
            raise OSError("backup verification failed (size mismatch)")
        with backup.open("rb") as bf:
            if sha256_bytes(bf.read()) != initial_hash:
                raise OSError("backup verification failed (checksum mismatch)")
    except OSError as exc:
        print("ERROR: backup failed; no changes made: {0}".format(exc), file=sys.stderr)
        safe_unlink(backup)
        sys.exit(1)

    print("Backup saved to {0}".format(backup))

    # Retain only ten newest matching backups. Failure to prune is non-fatal.
    try:
        backups = sorted(BACKUP_DIR.glob("{0}.bak.*".format(DENY_FILE.name)),
                         key=lambda p: p.stat().st_mtime, reverse=True)
        for old in backups[10:]:
            try:
                old.unlink()
            except OSError as exc:
                print("WARNING: could not remove old backup {0}: {1}".format(old, exc), file=sys.stderr)
    except OSError as exc:
        print("WARNING: could not prune backups: {0}".format(exc), file=sys.stderr)

    # Construct the new file while preserving original order for surviving lines.
    output_lines = []
    for e in entries:
        if e.old or e.duplicate or e.redundant:
            continue
        line = e.original
        if e.missing_date:
            line = append_date(line, stamp)
        output_lines.append(line + e.newline)

    # Use the file's dominant newline style for generated entries.
    newline = "\n"
    crlf = sum(1 for e in entries if e.newline == "\r\n")
    lf = sum(1 for e in entries if e.newline == "\n")
    if crlf > lf:
        newline = "\r\n"

    # Ensure a separator newline before appending generated subnet blocks.
    if new_networks:
        if output_lines and not output_lines[-1].endswith(("\n", "\r\n")):
            output_lines[-1] += newline
        for net, members in new_networks:
            first_comment = ""
            for m in members:
                c = strip_date_from_comment(m.comment)
                if c:
                    first_comment = c
                    break
            extra = " - " + first_comment if first_comment else ""
            output_lines.append(
                "{0} # aggregated by scfanalyzerv2 from {1} IPs{2} - {3}{4}".format(
                    net, len(members), extra, stamp, newline)
            )

    new_text = "".join(output_lines)
    new_bytes = new_text.encode("utf-8", errors="surrogateescape")

    # Re-check the live file immediately before replacement. If LFD/CSF/another
    # process changed it during our analysis, abort rather than overwrite new data.
    try:
        current_bytes = DENY_FILE.read_bytes()
    except OSError as exc:
        print("ERROR: could not re-read {0}; no replacement made: {1}".format(DENY_FILE, exc), file=sys.stderr)
        sys.exit(1)
    if sha256_bytes(current_bytes) != initial_hash:
        print("ERROR: csf.deny changed externally while FIX was being prepared; aborting to avoid losing new entries.", file=sys.stderr)
        print("Backup remains at: {0}".format(backup), file=sys.stderr)
        sys.exit(75)

    st = DENY_FILE.stat()
    temp_path = None
    try:
        fd, temp_name = tempfile.mkstemp(prefix=".{0}.scfanalyzerv2.".format(DENY_FILE.name),
                                         dir=str(DENY_FILE.parent))
        temp_path = Path(temp_name)
        with os.fdopen(fd, "wb") as tf:
            tf.write(new_bytes)
            tf.flush()
            os.fsync(tf.fileno())
        os.chmod(str(temp_path), stat.S_IMODE(st.st_mode))
        try:
            os.chown(str(temp_path), st.st_uid, st.st_gid)
        except PermissionError:
            pass
        os.replace(str(temp_path), str(DENY_FILE))
        temp_path = None
        try:
            dir_fd = os.open(str(DENY_FILE.parent), os.O_DIRECTORY)
            try:
                os.fsync(dir_fd)
            finally:
                os.close(dir_fd)
        except OSError:
            pass
    except OSError as exc:
        print("ERROR: atomic replacement failed: {0}".format(exc), file=sys.stderr)
        print("Original backup remains at: {0}".format(backup), file=sys.stderr)
        if temp_path is not None:
            safe_unlink(temp_path)
        sys.exit(1)

    print_timing(len(entries))
    print("Reloading CSF firewall...")
    sys.stdout.flush()
    csf_bin = shutil.which("csf")
    if csf_bin is None:
        for candidate in ("/usr/sbin/csf", "/usr/local/sbin/csf", "/sbin/csf"):
            if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
                csf_bin = candidate
                break
    if csf_bin is None:
        print("ERROR: CSF executable not found; firewall was not reloaded.", file=sys.stderr)
        print("The updated deny file is in place; verified backup: {0}".format(backup), file=sys.stderr)
        sys.exit(127)
    # Keep successful reloads quiet so FIX output stays compact. If CSF fails,
    # show the captured reload output so the failure can still be diagnosed.
    with tempfile.TemporaryFile(mode="w+b") as csf_output:
        result_code = subprocess.call(
            [csf_bin, "-r"],
            stdout=csf_output,
            stderr=subprocess.STDOUT,
        )
        if result_code != 0:
            csf_output.seek(0)
            raw_output = csf_output.read()
            try:
                reload_output = raw_output.decode("utf-8", "replace")
            except AttributeError:
                reload_output = str(raw_output)
            print("ERROR: csf -r failed with exit status {0}.".format(result_code), file=sys.stderr)
            print("The updated deny file is in place; verified backup: {0}".format(backup), file=sys.stderr)
            if reload_output.strip():
                print("--- CSF reload output ---", file=sys.stderr)
                print(reload_output.rstrip(), file=sys.stderr)
                print("--- end CSF reload output ---", file=sys.stderr)
            sys.exit(result_code or 1)
    print("[CSF reloaded]")

else:
    print_timing(len(entries))

if fix_count == 0:
    funny = [
        "Everything is locked down tighter than a submarine's mailbox. Nothing to fix!",
        "Your deny list is cleaner than a freshly formatted drive. All clear!",
        "No issues found. The firewall sleeps… but with one eye open.",
        "All IPs accounted for. The blacklist is as tidy as a sysadmin’s desk — on a good day.",
        "System secure. Even the bots are bored.",
        "No redundant entries. It’s like digital feng shui in here.",
        "Well optimized. The hackers called — they said it’s too hard now.",
        "Subnets are in check. Redundancies eliminated. CSF approved.",
        "Congratulations, your deny list has reached nirvana.",
        "Nothing to fix. Your deny list is living its best life.",
        "Clean as a whistle. Bots denied. Peace restored.",
        "No clutter, no chaos. Just pure CSF zen.",
        "Firewall status: chef’s kiss.",
        "Clean report. Go ahead and brag to the other sysadmins.",
        "The force is strong with this deny list.",
        "Peace has returned to the firewall galaxy.",
        "All entries verified. You are the One.",
        "99 problems but a bot ain’t one.",
    ]
    print(random.choice(funny))

sys.exit(0)
PY
PY_STATUS=$?

# Preserve Python's meaningful failure code (including 75 for concurrent modification).
exit "$PY_STATUS"