For more information regarding the security incident at F5, the actions we are taking to address it, and our ongoing efforts to protect our customers, click here.

BIG-IP security hardening and compliance checks script

 

Hi everyone

After a lot of testing, I'm excited to share a python script I built to help simplify and automate BIG-IP security auditing and compliance checks.

This script connects to a BIG-IP system, validates its security posture, and generates a clean, professional HTML report.

It follows F5’s recommended best-practices (including Article K53108777) and even goes beyond the official guidance to surface additional real-world risks and misconfigurations.

 

What it checks:

  • Access & Identity Security
  • SSH / GUI access restrictions
  • Remote authentication source
  • Root & Admin account status
  • Admin lockout policy
  • Password policy strength
  • and more..

 

What you get:

  • Compliance scoring
  • Visual compliance bar (Green / Yellow / Red)
  • Expandable details

 

Designed to make BIG-IP hardening:

  • Faster
  • Repeatable
  • Audit-friendly
  • Clear for leadership

 

Community-Driven & Free

This tool is available free for the F5 community.

I’d love your feedback, feature requests, or contributions.

If you're upgrading, migrating, securing, or auditing BIG-IP environments — this can save you time and strengthen your posture.

 

My LinkedIn:

https://lnkd.in/ewqsb9Rk

 

And finally.. the script, attached.

import datetime
import customtkinter as ctk
import paramiko
from tkinter import filedialog, messagebox
import threading
import queue

def list_partitions(client):
    try:
        stdin, stdout, stderr = client.exec_command('tmsh list auth partition')
        output = stdout.read().decode().strip()
        # Extract partition names from tmsh list auth partition output
        partitions = re.findall(r'auth partition (\S+)', output)
        if not partitions:
            # Fallback to Common if empty
            return ["Common"]
        return partitions
    except Exception as e:
        # On error fallback to Common partition
        return ["Common"]

def get_port_lockdown_self_ips_all_partitions(client):
    def parse_blocks(output):
        result = {}
        lines = output.splitlines()
        block_name = None
        block_lines = []
        brace_level = 0

        for line in lines:
            # Check for start of a 'net self <name> {' block
            start_match = re.match(r'net self (\S+) \{', line)
            if start_match and brace_level == 0:
                if block_name is not None:
                    # Save previous block before starting new one
                    result[block_name] = "\n".join(block_lines)
                block_name = start_match.group(1)
                block_lines = [line]
                brace_level = 1
                continue

            if brace_level > 0:
                block_lines.append(line)
                brace_level += line.count('{')
                brace_level -= line.count('}')
                if brace_level == 0:
                    # End of current block
                    result[block_name] = "\n".join(block_lines)
                    block_name = None
                    block_lines = []

        # If last block still open, save it
        if block_name is not None:
            result[block_name] = "\n".join(block_lines)

        return result

    try:
        partitions = list_partitions(client)
        all_self_ips = {}
        alert_terms = ['all', 'default', 'tcp:https', 'tcp:ssh', 'tcp:http']
        found_alerts = []

        for partition in partitions:
            cmd = f'tmsh -c "cd /{partition}; list net self allow-service"'
            stdin, stdout, stderr = client.exec_command(cmd)
            output = stdout.read().decode()

            blocks = parse_blocks(output)
            partition_services = {}

            for name, block in blocks.items():
                # Extract allow-service value(s)
                allow_match = re.search(r'allow-service\s+(?:\{([^}]*)\}|(\S+))', block, re.DOTALL)
                if allow_match:
                    allow_val = (allow_match.group(1) or allow_match.group(2)).strip()
                    # Normalize whitespace and join
                    allow_val = ' '.join(allow_val.split())
                    partition_services[name] = allow_val
                    for term in alert_terms:
                        if re.search(rf'\b{re.escape(term)}\b', allow_val, re.IGNORECASE):
                            found_alerts.append(f"{partition}:{name}: {term}")

            all_self_ips[partition] = partition_services

        return all_self_ips, found_alerts

    except Exception as e:
        return {}, [f"Error: {str(e)}"]


def get_httpd_allow(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys httpd allow")
        output = stdout.read().decode().strip()
        if not output or "allow {" not in output:
            return "UNRESTRICTED", None, output
        elif re.search(r'\b(any|all)\b', output, flags=re.IGNORECASE):
            return "UNRESTRICTED", None, output
        else:
            ips = re.findall(r'\b\d{1,3}(?:\.\d{1,3}){3}\b', output)
            if not ips:
                return "UNRESTRICTED", None, output
            return "RESTRICTED", ips, output
    except Exception as e:
        return "ERROR", None, f"(Error: {str(e)})"

def get_httpd_idle_timeout(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys httpd auth-pam-idle-timeout")
        output = stdout.read().decode().strip()
        match = re.search(r'auth-pam-idle-timeout\s+(\d+)', output)
        if match:
            return int(match.group(1)), output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_sshd_inactivity_timeout(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys sshd inactivity-timeout")
        output = stdout.read().decode().strip()
        match = re.search(r'inactivity-timeout\s+(\d+)', output)
        if match:
            return int(match.group(1)), output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_ssh_access(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys sshd allow")
        output = stdout.read().decode().strip()
        if not output or "allow {" not in output:
            return "UNRESTRICTED", None, output
        elif re.search(r'\b(any|all)\b', output, flags=re.IGNORECASE):
            return "UNRESTRICTED", None, output
        else:
            ips = re.findall(r'\b\d{1,3}(?:\.\d{1,3}){3}\b', output)
            if not ips:
                return "UNRESTRICTED", None, output
            return "RESTRICTED", ips, output
    except Exception as e:
        return "ERROR", None, f"(Error: {str(e)})"

def get_bigip_hostname(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys global-settings hostname")
        output = stdout.read().decode().strip()
        match = re.search(r'hostname\s+(\S+)', output)
        if match:
            return match.group(1)
        else:
            return "(Unable to parse BIG-IP hostname)"
    except Exception as e:
        return f"(Error: {str(e)})"

def get_password_policy(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list auth password-policy all-properties")
        output = stdout.read().decode().strip()
        return output
    except Exception as e:
        return f"(Error retrieving password policy: {str(e)})"

import re

def get_syslog_remote_servers(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys syslog remote-servers")
        output = stdout.read().decode()

        # Check if no remote servers configured
        if re.search(r'remote-servers\s+none', output, re.IGNORECASE):
            return None, output

        # Regex to find lines starting with 'host' and capture IP address
        pattern = re.compile(r'^\s*host\s+(\d+\.\d+\.\d+\.\d+)\s*$', re.MULTILINE)
        ips = pattern.findall(output)

        unique_ips = sorted(set(ips))

        if unique_ips:
            msg = f"Syslog Remote Servers configured: {', '.join(unique_ips)}"
        else:
            msg = "No Syslog Remote Servers configured."

        return msg, output

    except Exception as e:
        return None, f"(Error: {str(e)})"


def get_root_login_disabled(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list sys db systemauth.disablerootlogin")
        output = stdout.read().decode()
        match = re.search(r'value\s+"([^"]+)"', output)
        if match:
            val = match.group(1).strip().lower()
            return val, output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_primary_admin_user(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list sys db systemauth.primaryadminuser")
        output = stdout.read().decode().strip()
        match = re.search(r'value\s+"([^"]+)"', output)
        if match:
            return match.group(1).strip(), output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error retrieving primary admin user: {str(e)})"

def get_local_admin_lockout_disabled(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list sys db systemauth.disablelocaladminlockout")
        output = stdout.read().decode()
        match = re.search(r'value\s+"([^"]+)"', output)
        if match:
            val = match.group(1).strip().lower()
            return val, output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_auth_source(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list auth source")
        output = stdout.read().decode()
        if not output.strip() or re.search(r'auth source\s*\{\s*\}', output, re.IGNORECASE):
            return None, output
        match = re.search(r'type\s+(\w+)', output, re.IGNORECASE)
        method = match.group(1).lower() if match else "Unknown"
        return method, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_httpd_ssl_protocol(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list /sys httpd ssl-protocol")
        output = stdout.read().decode().strip()
        match = re.search(r'ssl-protocol\s+(\S+)', output)
        if match:
            return match.group(1), output
        else:
            return None, output
    except Exception as e:
        return None, f"(Error: {str(e)})"

def get_clientside_profiles_in_use(client):
    profiles = []
    try:
        stdin, stdout, stderr = client.exec_command('tmsh list ltm virtual all | grep -B 1 "context clientside"')
        output = stdout.read().decode()
        lines = output.splitlines()

        for line in lines:
            line = line.strip()
            if line == '--' or not line:
                continue
            if line.endswith('{'):
                profile_name = line[:-1].strip()
                # Filter out http2 or other non-clientssl profiles by name pattern
                if profile_name.lower() != 'http2' and 'clientssl' in profile_name.lower():
                    profiles.append(profile_name)

    except Exception as e:
        pass
    return profiles

def check_tls_options_per_profile(client):
    profiles_in_use = get_clientside_profiles_in_use(client)
    reports = {}
    for prof in profiles_in_use:
        try:
            cmd = f"tmsh list ltm profile client-ssl {prof} options"
            stdin, stdout, stderr = client.exec_command(cmd)
            output = stdout.read().decode().lower()
            warnings = []
            if "no-ssl" not in output:
                warnings.append("SSL is allowed")
            if "no-tlsv1.1" not in output:
                warnings.append("TLSv1.1 is allowed")
            if "single-dh-use" not in output:
                warnings.append("Strong Diffie-Hellman ciphers are not in use")
            if "no-tlsv1" not in output:
                warnings.append("TLSv1.0 is allowed")
            reports[prof] = {'warnings': warnings, 'raw': output}
        except Exception as e:
            reports[prof] = {'warnings': [f"Error checking profile {prof}: {str(e)}"], 'raw': ''}
    return reports


def validate_password_policy(policy_output):
    required_policy = {
        "max-duration": "90",
        "max-login-failures": "3",
        "minimum-length": "10",
        "password-memory": "3",
        "required-lowercase": "1",
        "required-numeric": "1",
        "required-special": "1",
        "required-uppercase": "1"
    }
    unmet = []
    for key, val in required_policy.items():
        regex = rf'{re.escape(key)}\s+{re.escape(val)}'
        if not re.search(regex, policy_output):
            unmet.append(f"Set {key} to {val}")
    return (len(unmet) == 0, unmet)

def get_ntp_servers(client):
    try:
        stdin, stdout, stderr = client.exec_command("tmsh list sys ntp")
        output = stdout.read().decode().strip()

        # F5 Output example:
        # sys ntp {
        #    servers { 192.168.1.10 192.168.1.11 }
        # }
        servers = re.findall(r'servers\s+\{([^}]*)\}', output, re.MULTILINE | re.DOTALL)
        if servers:
            server_list = servers[0].split()
            return server_list, output

        # If "servers none" or empty
        return None, output
    except Exception as e:
        return None, f"(Error retrieving NTP configuration: {str(e)})"
def check_slowread_timeout(client):
    """
    Detect SLOWRead-vulnerable TCP profiles only if:
      - zero-window-timeout >= 10000 (10 seconds)
      - profile is attached to a virtual server
    Uses 'tmsh list ltm virtual profiles | grep -e virtual -e tcp' to map profiles.
    """
    import re
    reports = {}

    try:
        cmd = "tmsh list ltm virtual profiles | grep -e virtual -e tcp"
        stdin, stdout, stderr = client.exec_command(cmd)
        vs_lines = stdout.read().decode(errors="ignore").splitlines()

        vs_profile_map = {}
        current_vs = None

        for line in vs_lines:
            line = line.strip()
            m_vs = re.match(r'ltm virtual\s+(\S+)', line)
            if m_vs:
                current_vs = m_vs.group(1)
                vs_profile_map.setdefault(current_vs, [])
                continue

            m_prof = re.match(r'([^ \{]+)\s*\{', line)
            if m_prof and current_vs:
                prof = m_prof.group(1)
                vs_profile_map[current_vs].append(prof)

        stdin, stdout, stderr = client.exec_command("tmsh list ltm profile tcp zero-window-timeout")
        tcp_raw = stdout.read().decode(errors="ignore")
        profile_names = re.findall(r'ltm profile tcp\s+([^\s\{]+)', tcp_raw)

        profile_to_vs = {}
        for vs, profiles in vs_profile_map.items():
            for pf in profiles:
                profile_to_vs.setdefault(pf, []).append(vs)

        for prof in set(profile_names + list(profile_to_vs.keys())):
            stdin, stdout, stderr = client.exec_command(f"tmsh list ltm profile tcp {prof} zero-window-timeout")
            prof_output = stdout.read().decode(errors="ignore")

            m = re.search(r'zero-window-timeout\s+(\d+)', prof_output)
            timeout = int(m.group(1)) if m else None

            # Get the list of virtual servers attached to this profile
            attached_vs = sorted(profile_to_vs.get(prof, []))

            # Only report if profile is attached to at least one virtual server AND timeout >= 10000
            if attached_vs and timeout is not None and timeout > 10000:
                reports[prof] = {
                    "timeout": timeout,
                    "vs_list": attached_vs,
                    "raw": prof_output,
                    "warning": True
                }
            else:
                # If it is not attached to any virtual server, don't include
                if not attached_vs:
                    continue
                # If attached but timeout is below threshold, add as safe or ignore based on your needs
                reports[prof] = {
                    "timeout": timeout,
                    "vs_list": attached_vs,
                    "raw": prof_output,
                    "warning": False
                }

        return reports

    except Exception as e:
        return {
            "Error": {
                "timeout": None,
                "vs_list": [],
                "raw": f"SLOWRead error: {str(e)}",
                "warning": True
            }
        }

def check_cert_expiration(client):
    try:
        stdin, stdout, stderr = client.exec_command(
            'tmsh -c "cd /; run /sys crypto check-cert ignore-large-cert-bundles enabled verbose enabled" | grep -v OK'
        )
        output = stdout.read().decode()
        expired = []
        expiring_soon = []
        now = datetime.datetime.now()
        warning_days = 30

        for line in output.splitlines():
            cn = re.search(r'CN=([^,]+)', line)
            dt = re.search(r'(expired on|will expire on) (\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}\s+\d{4})', line)
            if cn and dt:
                cn = cn.group(1)
                exp = datetime.datetime.strptime(dt.group(2), "%b %d %H:%M:%S %Y")
                days = (exp - now).days
                if days < 0:
                    expired.append(line)
                elif days <= warning_days:
                    expiring_soon.append(f"{line} (expires in {days} days)")

        return {'raw': output, 'expired': expired, 'expiring_soon': expiring_soon}
    except Exception as e:
        return {'raw': f"(Error: {str(e)})", 'expired': [], 'expiring_soon': []}

def extract_cn_and_date(text):
    cn_match = re.search(r'CN=([^,]+)', text)
    date_match = re.search(r'(expired on|will expire on) (\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}\s+\d{4})', text)
    cn_value = cn_match.group(1) if cn_match else "Unknown"
    date_value = date_match.group(2) if date_match else "Unknown"
    return cn_value, date_value



def highlight_date(text, color):
    return f"<span style='color:{color};'>{text}</span>"

def run_all_checks(host, username, password):
    results = {}
    try:

        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(host, username=username, password=password, timeout=5)
        results['hostname'] = get_bigip_hostname(client)
        results['ssh_status'], results['ssh_ips'], results['ssh_raw'] = get_ssh_access(client)
        results['http_status'], results['http_ips'], results['http_raw'] = get_httpd_allow(client)
        results['idle_timeout'], results['idle_raw'] = get_httpd_idle_timeout(client)
        results['sshd_timeout'], results['sshd_timeout_raw'] = get_sshd_inactivity_timeout(client)
        results['password_policy_raw'] = get_password_policy(client)
        results['syslog_servers'], results['syslog_raw'] = get_syslog_remote_servers(client)
        results['root_disabled'], results['root_disabled_raw'] = get_root_login_disabled(client)
        results['local_admin_disabled'], results['local_admin_disabled_raw'] = get_local_admin_lockout_disabled(client)
        results['primary_admin_user'], results['primary_admin_user_raw'] = get_primary_admin_user(client)
        results['auth_source'], results['auth_source_raw'] = get_auth_source(client)
        results['ssl_protocol'], results['ssl_protocol_raw'] = get_httpd_ssl_protocol(client)
        results['port_lockdown_output'], results['port_lockdown_alerts'] = get_port_lockdown_self_ips_all_partitions(client)
        results['tls_options'] = check_tls_options_per_profile(client)
        results['ntp_servers'], results['ntp_raw'] = get_ntp_servers(client)
        results['slowread_profiles'] = check_slowread_timeout(client)
        results['cert_expiration'] = check_cert_expiration(client)

        client.close()
    except Exception as e:
        results["error"] = str(e)
    return results

def calculate_cis_score(results):
    """
    Returns CIS compliance score based on existing audit results.
    Each check = 1 point if compliant.
    """

    checks = {
        "SSH restricted": results.get('ssh_status') == "RESTRICTED",
        "HTTP restricted": results.get('http_status') == "RESTRICTED",
        "GUI idle timeout": results.get('idle_timeout') and results['idle_timeout'] > 0,
        "SSH idle timeout": results.get('sshd_timeout') and 0 < results['sshd_timeout'] <= 900,
        "Password policy": validate_password_policy(results.get("password_policy_raw", ""))[0],
        "Syslog enabled": results.get('syslog_servers') is not None,
        "Root login disabled": results.get('root_disabled') == "true",
        "Primary admin changed": results.get('primary_admin_user') and results.get('primary_admin_user').lower() != "admin",
        "Admin lockout enabled": results.get('local_admin_disabled') == "false",
        "Remote auth configured": results.get('auth_source') is not None,
        "TLS 1.2+ enforced": results.get('ssl_protocol') in ("TLSv1.2", "TLSv1.3"),
        "Self-IP lockdown enforced": len(results.get('port_lockdown_alerts', [])) == 0,
        "NTP configured": results.get('ntp_servers') is not None,
        "Certificates valid": len(results.get('cert_expiration', {}).get('expired', [])) == 0,
        "TLS profiles hardened": all(
            len(details.get('warnings', [])) == 0 for details in results.get('tls_options', {}).values()),
        "SLOWRead TCP hardening": all(
            not details.get('warning', False) for details in results.get('slowread_profiles', {}).values()),

    }

    passed = sum(1 for v in checks.values() if v)
    total = len(checks)

    score_percent = round((passed / total) * 100, 2)

    return score_percent, passed, total

def compose_sections(results):
    sections = []
    def green(msg): return f'<span class="ok">{msg}</span>'
    def warn(msg): return f'<span class="warning">{msg}</span>'
    def black(msg): return f'<span style="color: black; font-weight: normal;">{msg}</span>'

    # SSH Access Restriction
    if results.get('ssh_status') == "RESTRICTED":
        msg = (
                '<span style="color:green;">SSH access is restricted to these IP addresses:</span><br>'
                + '<span style="color:black;">' + "<br>".join(results['ssh_ips']) + '</span>'
        )

    elif results.get('ssh_status') == "UNRESTRICTED":
        msg = warn("WARNING: SSH access is NOT restricted to a specific network!")
    else:
        msg = f"SSH: {results.get('ssh_status')}"
    sections.append({'summary': "SSH access restriction", 'content': f'{msg}<br><pre class="cmd-output">{results.get("ssh_raw", "")}</pre>'})

    # GUI Access Restriction
    if results.get('http_status') == "RESTRICTED":
        msg = (
                '<span style="color:green;">GUI access is restricted to these IP addresses:</span><br>'
                + '<span style="color:black;">' + "<br>".join(results['http_ips']) + '</span>'
        )

    elif results.get('http_status') == "UNRESTRICTED":
        msg = warn("WARNING: GUI/HTTP access is NOT restricted to a specific network!")
    else:
        msg = f"GUI/HTTP: {results.get('http_status')}"
    sections.append({'summary': "GUI access restriction", 'content': f'{msg}<br><pre class="cmd-output">{results.get("http_raw", "")}</pre>'})

    # GUI Inactivity Timeout
    idle_timeout = results.get('idle_timeout')
    idle_raw = results.get('idle_raw', '')
    idle_msg = f"Inactivity timeout is set to {idle_timeout} seconds." if idle_timeout else "Could not determine inactivity timeout value."
    sections.append({'summary': "GUI inactivity timeout", 'content': f'{green(idle_msg)}<br><pre class="cmd-output">{idle_raw}</pre>'})

    # SSH Inactivity Timeout
    sshd_timeout = results.get('sshd_timeout')
    sshd_raw = results.get('sshd_timeout_raw', '')
    if sshd_timeout is None:
        sshd_msg = warn("Could not determine SSH inactivity timeout value.")
    elif sshd_timeout == 0:
        sshd_msg = warn('WARNING: SSH inactivity timeout is unlimited (0)!')
    elif sshd_timeout > 900:
        sshd_msg = warn(f'WARNING: SSH inactivity timeout is too high ({sshd_timeout} seconds)!')
    else:
        sshd_msg = green(f"SSH inactivity timeout is set to {sshd_timeout} seconds.")
    sections.append({'summary': "SSH inactivity timeout", 'content': f'{sshd_msg}<br><pre class="cmd-output">{sshd_raw}</pre>'})

    # Password Policy
    pw_policy_raw = results.get("password_policy_raw", "")
    passes, unmet_conditions = validate_password_policy(pw_policy_raw)
    pw_output = f'<pre class="cmd-output">{pw_policy_raw}</pre>'
    if passes:
        pw_msg = green("Password policy meets all requirements.<br>") + pw_output
    else:
        explanation = "<br>".join(unmet_conditions)
        pw_msg = warn("WARNING: Password policy does NOT meet all requirements!<br>") + black("Required changes:<br>" + explanation + "<br>") + pw_output
    sections.append({'summary': "Password policy", 'content': pw_msg})

    # Syslog Servers
    syslog_msg, syslog_raw = results.get('syslog_servers'), results.get('syslog_raw', '')
    if syslog_msg is None:
        syslog_msg = warn("WARNING: No syslog server was configured!")
    else:
        # Separate prefix and IPs, add a line break after prefix
        prefix = "Syslog Remote Servers configured:<br>"
        ips_part = syslog_msg.replace(prefix.replace("<br>", ""), "").strip()
        ips_lines = ips_part.replace(", ", "<br>")  # put each IP on new line via <br>
        # Combine prefix (green) with IPs (black default)
        syslog_msg = green(prefix) + ips_lines

    sections.append({
        'summary': "Syslog server configuration",
        'content': f'{syslog_msg}<br><pre class="cmd-output">{syslog_raw}</pre>'
    })

    # Root User Disabled
    root_disabled = results.get('root_disabled')
    root_raw = results.get('root_disabled_raw', '')
    if root_disabled == "false":
        root_msg = warn("WARNING: Root user is not disabled!")
    else:
        root_msg = green("Root user is disabled.")
    sections.append({'summary': "Root user status", 'content': f'{root_msg}<br><pre class="cmd-output">{root_raw}</pre>'})

    # NTP / Time Sync Configuration
    ntp_servers = results.get('ntp_servers')
    ntp_raw = results.get('ntp_raw', '')

    if ntp_servers:
        ntp_msg = (
                '<span style="color:green;">NTP servers configured:</span><br>'  # Green text
                + '<span style="color:black;">' + "<br>".join(ntp_servers) + '</span>'  # Black list
        )

    else:
        ntp_msg = warn("WARNING: No NTP servers configured! Time drift may break logs, HA sync, & security.")

    sections.append({
        'summary': "NTP servers configuration check",
        'content': f'{ntp_msg}<br><pre class="cmd-output">{ntp_raw}</pre>'
    })

    # Primary Admin User Check
    primary_admin = results.get('primary_admin_user')
    primary_admin_raw = results.get('primary_admin_user_raw', '')

    if primary_admin is None:
        primary_admin_msg = warn("WARNING: Unable to determine primary admin user.")
    elif primary_admin.lower() == "admin":
        primary_admin_msg = warn('WARNING: Primary admin user is still "admin". Change it!')
    else:
        primary_admin_msg = green(f"Primary admin user is set to: {primary_admin}")


    sections.append({
        'summary': "Admin user status",
        'content': f'{primary_admin_msg}<br><pre class="cmd-output">{primary_admin_raw}</pre>'
    })

    # Admin Lockout Disabled
    admin_disabled = results.get('local_admin_disabled')
    admin_raw = results.get('local_admin_disabled_raw', '')
    if admin_disabled == "false":
        admin_msg = warn("WARNING: Admin lockout is disabled!")
    else:
        admin_msg = green("Admin lockout is enabled.")
    sections.append({'summary': "Admin User Lockout function Status", 'content': f'{admin_msg}<br><pre class="cmd-output">{admin_raw}</pre>'})


    # Remote Auth Source
    auth_source = results.get('auth_source')
    auth_raw = results.get('auth_source_raw', '')
    if auth_source is None:
        auth_msg = warn("WARNING: No remote authentication method was configured.")
    else:
        auth_msg = green(f"<b>{auth_source}</b> Remote authentication method is configured.")
    sections.append({'summary': "Remote authentication source", 'content': f'{auth_msg}<br><pre class="cmd-output">{auth_raw}</pre>'})

    # TLS 1.2 GUI Access Enforcement
    ssl_protocol = results.get('ssl_protocol')
    ssl_raw = results.get('ssl_protocol_raw', '')
    if ssl_protocol is None or ssl_protocol != "TLSv1.2":
        ssl_msg = warn("WARNING: GUI access is not restricted to TLSv1.2")
    else:
        ssl_msg = green("GUI management access is properly restricted to TLSv1.2.")
    sections.append({'summary': "TLS 1.2 GUI access enforcement", 'content': f'{ssl_msg}<br><pre class="cmd-output">{ssl_raw}</pre>'})

    # TLS Profile Options per SSL Profile
    tls_options = results.get('tls_options', {})
    tls_content = ""
    if not tls_options:
        tls_content = "No client SSL profiles found."
    else:
        for profile, detail in tls_options.items():
            warnings_list = detail.get('warnings', [])
            raw_output = detail.get('raw', '')
            if warnings_list:
                warnings_formatted = "<ul>" + "".join(f"<li>{w}</li>" for w in warnings_list) + "</ul>"

                warn_msg = (
                    f"<span style='color:#d9534f; font-weight:bold;'>"
                    f"Warnings for profile '{profile}':"
                    f"</span><br>{warnings_formatted}"
                )



            else:
                warn_msg = green(f"Profile '{profile}' options are properly configured.")
            tls_content += f"<h4>{profile}</h4>{warn_msg}<br><pre class='cmd-output'>{raw_output}</pre><br>"
    sections.append({'summary': "Client-SSL profiles hardening", 'content': tls_content})

    # SLOWRead TCP Timeout Check
    slowread_profiles = results.get('slowread_profiles', {})
    slowread_content = ""
    for prof, detail in slowread_profiles.items():
        timeout_sec = detail.get('timeout', 0) / 1000 if detail.get('timeout') else 0
        raw_output = detail.get('raw', '')

        if detail.get('warning'):  # Warning condition
            for vs in detail.get('vs_list', []):
                slowread_content += (
                    f"<b><span style='color:red;'>SLOWRead risk detected in TCP profile '{prof}' in virtual server {vs}:</span></b><br>"
                    f"<span style='color:black;'>zero-window-timeout={detail['timeout']}</span><br>"
                    f"<pre class='cmd-output'>{raw_output}</pre><br>"
                )

        else:  # Safe condition

            vs_list = detail.get('vs_list', [])

            vs_names = ", ".join(vs_list) if vs_list else "No virtual servers assigned"

            slowread_content += (

                f"<b><span style='color:green;'>TCP profile '{prof}' zero-window-timeout value is safe </span></b>"
                f"<span style='color:black;'>({timeout_sec:.2f} seconds)</span><br> "
                f"It is assigned to the following virtual servers: {vs_names}\n<br>"
                f"<pre class='cmd-output'>{raw_output}</pre><br>"

            )

    sections.append({'summary': "SLOWRead TCP profile check", 'content': slowread_content})

    # Port Lockdown State on Self IPs
    port_lockdown = results.get('port_lockdown_output', '')
    port_alerts = results.get('port_lockdown_alerts', [])

    if port_alerts:
        formatted_alerts = []
        for alert in port_alerts:
            parts = alert.split(':', 2)
            if len(parts) == 3:
                partition, self_ip, ports = parts
                formatted_alerts.append(
                    f"<b>Partition:</b> {partition.strip():<15}\n"
                    f"<b>Self-IP:</b> {self_ip.strip():<20}\n"
                    f"<b>Allowed-Ports:</b> {ports.strip()}"
                )
            else:
                # fallback if unexpected format
                formatted_alerts.append(f"<b>Partition info:</b> {alert.strip()}")

        # Wrap the formatted alerts as HTML bullet points preserving multiline with <br>
        bullet_html = "<ul style='margin-top:0; padding-left: 20px;'>\n"
        for alert in formatted_alerts:
            bullet_html += f"<li style='margin-bottom:10px; white-space: pre-wrap;'>{alert.replace(chr(10), '<br>')}</li>\n"
        bullet_html += "</ul>"

        port_msg = (
                '<span style="color:red; font-weight:bold;">WARNING: Management access should NOT be allowed from Self IP!</span><br>'
                + '<span style="color:black; font-weight:bold;">Found:</span><br>'
                + bullet_html
        )

    else:
        port_msg = green("No management access risks detected on Self IPs.")

    sections.append({
        'summary': "Port Lockdown state on Self IPs",
        'content': f'{port_msg}<br><pre class="cmd-output">{port_lockdown}</pre>'
    })

    cert_check = results.get('cert_expiration', {})
    cert_raw = cert_check.get('raw', '')
    expired = cert_check.get('expired', [])
    expiring_soon = cert_check.get('expiring_soon', [])

    cert_content = ""

    if expired:
        expired_msgs = ""
        for c in expired:
            cn, date = extract_cn_and_date(c)
            expired_msgs += f"<li><b>CN:</b> {cn}, <b>Expired at:</b> {date}</li>"
        cert_content += f"<b><span style='color:red;'>Expired certificates:</span></b><ul style='color:black;'>{expired_msgs}</ul>"

    if expiring_soon:
        soon_msgs = ""
        for c in expiring_soon:
            cn, date = extract_cn_and_date(c)
            soon_msgs += f"<li><b>CN:</b> {cn}, <b>Will expire at:</b> {date}</li>"
        cert_content += f"<b><span style='color:orange;'>Certificates expiring soon (within 30 days):</span></b><ul style='color:black;'>{soon_msgs}</ul>"

    if not expired and not expiring_soon:
        cert_content = '<span style="color:green;">All certificates are valid within the next 30 days.</span>'

    cert_content += f"<br><pre class='cmd-output'>{cert_raw}</pre>"

    sections.append({
        'summary': "Certificate Expiration Check",
        'content': cert_content
    })

    return sections

def generate_html_report(device, timestamp, audit_results, html_path, cis_score, cis_passed, cis_total):

    html = f"""<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>BIG-IP Security Audit Report - {device}</title>

    <style>
        body {{ font-family: Arial, Helvetica, sans-serif; margin: 24px; }}
        .header {{ background: #dc0024; color: #fff; padding: 12px; border-radius: 8px; font-size: 1.5em; margin-bottom: 18px; }}
        .audit-metadata {{ margin-bottom: 18px; }}
        .audit-metadata b {{ margin-right: 4px; }}
        details {{ margin-bottom: 22px; border-radius:8px; box-shadow: 0 0 6px #efefef; }}
        summary {{ background: #3399ff; color: #fff; border-radius:8px; font-weight: bold; padding: 8px; cursor: pointer; }}
        details[open] summary {{ background: #2577d2; }}
        .section-content {{ padding: 12px 22px; background:#fafbfc; border-radius:0 0 8px 8px; }}
        .warning {{ color: #e03b3b; font-weight: 700; }}
        .ok {{ color: #28a745; font-weight: 700; }}
        pre.cmd-output {{ background: #eeeeee; padding: 8px 12px; border-radius: 5px; white-space: pre-wrap; word-wrap: break-word; }}
        button.toggle-all {{
            margin: 6px 8px 12px 0;
            padding: 8px 14px;
            border-radius: 8px;
            background: linear-gradient(135deg, #4f8df7, #2a6ee8);
            color: #fff;
            border: none;
            cursor: pointer;
            font-size: 14px;
            font-weight: 500;
            box-shadow: 0 2px 5px rgba(0,0,0,0.18);
            transition: all 0.2s ease;
        }}

        button.toggle-all:hover {{
           background: linear-gradient(135deg, #3a7bf0, #1f5ecc);
           transform: translateY(-2px);
           box-shadow: 0 4px 8px rgba(0,0,0,0.22);
        }}

        button.toggle-all:active {{
          transform: translateY(0px);
          box-shadow: 0 1px 3px rgba(0,0,0,0.16);
        }}

    </style>
</head>
<body>
    <div class="header">BIG-IP Security Audit Report - {device}</div>
    <div class="audit-metadata">
    <b>Timestamp:</b> {timestamp}<br><br>
    <b>Compliance Score:</b> {cis_score}% ({cis_passed}/{cis_total})<br>

    <div style="width:15%;background:#ddd;border-radius:8px;margin-top:6px;">
        <div style="
            width:{cis_score}%;
            background:{'#d9534f' if cis_score<70 else '#f0ad4e' if cis_score<90 else '#5cb85c'};
            height:14px;
            border-radius:8px;">
        </div>
    </div>
</div>
        <button class="toggle-all" onclick="expandAll()">⤢ Expand All</button>
        <button class="toggle-all" onclick="collapseAll()">⤡ Collapse All</button>
<br><br>

</div>

"""
    for idx, section in enumerate(audit_results, start=1):
        html += f"""
    <details>
        <summary>{idx}. {section['summary']}</summary>
        <div class="section-content">
            {section['content']}
        </div>
    </details>
"""
    html += """
<script>
function expandAll() {
  document.querySelectorAll("details").forEach(d => d.setAttribute("open", true));
}
function collapseAll() {
  document.querySelectorAll("details").forEach(d => d.removeAttribute("open"));
}
</script>

</body>
</html>"""
    with open(html_path, 'w', encoding='utf-8') as f:
        f.write(html)
# This new function will run in the background thread
def run_audit_in_thread(host, username, password, results_queue):
    """
    Runs the full audit in a separate thread and puts the
    results dictionary into the provided queue.
    """
    try:
        # This is your existing blocking function
        results = run_all_checks(host, username, password)
        # Put the results in the queue for the main thread to find
        results_queue.put(results)
    except Exception as e:
        # If anything unexpected goes wrong, put the error in the queue
        results_queue.put({"error": f"A critical thread error occurred: {str(e)}"})


# This new function runs in the main GUI thread
def check_audit_queue(results_queue):
    """
    Checks the results_queue for data from the background thread.
    If results are found, it processes them (saves report, shows popups).
    If not, it schedules itself to check again.
    """
    try:
        # Check the queue for results WITHOUT blocking
        results = results_queue.get_nowait()

        # --- RESULTS ARE BACK! PROCESS THEM IN THE GUI ---

        # Check for an error from the worker thread
        if "error" in results:
            messagebox.showerror("Connection or Command Error", results["error"])
            status_label.configure(text="Audit failed. Check connection or input.", text_color="#e54646")
        else:
            # No error, proceed with report generation
            timestamp = datetime.datetime.now().strftime("%d-%m-%Y %H:%M")
            # Get host/user from the entry boxes (since we're in the GUI thread)
            device = results.get('hostname', host_entry.get().strip()).strip()
            username = user_entry.get().strip()

            sections = compose_sections(results)
            cis_score, cis_passed, cis_total = calculate_cis_score(results)
            results['cis_score'] = cis_score
            results['cis_passed'] = cis_passed
            results['cis_total'] = cis_total

            path = filedialog.asksaveasfilename(defaultextension=".html",
                                                filetypes=[("HTML files", "*.html"), ("All files", "*.*")],
                                                initialfile=f"{device.replace(' ', '_')}_audit_{timestamp.replace(':', '_')}.html")
            if path:
                generate_html_report(
                    device,timestamp, sections, path,
                    cis_score=results['cis_score'],
                    cis_passed=results['cis_passed'],
                    cis_total=results['cis_total']
                )
                status_label.configure(text=f"HTML report saved successfully", text_color="#28a745")
                messagebox.showinfo("Report Saved", f"Security audit HTML report was saved to:\n\n{path}")
            else:
                status_label.configure(text="No save location selected.", text_color="#e54646")

        # --- Re-enable the button, whether it passed or failed ---
        run_button.configure(state='normal', text="Run Audit & Export to HTML")

    except queue.Empty:
        # No results yet, schedule this function to run again in 100ms
        root.after(100, check_audit_queue, results_queue)
def run_audit():
    host = host_entry.get().strip()
    username = user_entry.get().strip()
    password = pass_entry.get()
    status_label.configure(text="Running audit checks...", text_color="#285fb4")
    root.update()
    results = run_all_checks(host, username, password)
    if "error" in results:
        messagebox.showerror("Connection or Command Error", results["error"])
        status_label.configure(text="Audit failed. Check connection or input.", text_color="#e54646")
        return
    timestamp = datetime.datetime.now().strftime("%d-%m-%Y %H:%M")
    sections = compose_sections(results)
    cis_score, cis_passed, cis_total = calculate_cis_score(results)
    results['cis_score'] = cis_score
    results['cis_passed'] = cis_passed
    results['cis_total'] = cis_total

    path = filedialog.asksaveasfilename(defaultextension=".html", filetypes=[("HTML files", "*.html"), ("All files", "*.*")],
                                        initialfile=f"BIG-IP_security_audit_.html")
    if path:
        generate_html_report(
            timestamp, sections, path,
            cis_score=results['cis_score'],
            cis_passed=results['cis_passed'],
            cis_total=results['cis_total']
        )

        status_label.configure(text=f"HTML report saved successfully", text_color="#28a745")
        messagebox.showinfo("Report Saved", f"Security audit HTML report was saved to:\n\n{path}")
    else:
        status_label.configure(text="No save location selected.", text_color="#e54646")

# GUI initialization
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
root = ctk.CTk()
root.title("BIG‑IP Security Audit Tool v1.2")
root.geometry("440x340")
root.resizable(False, False)
frame = ctk.CTkFrame(root, corner_radius=16)
frame.pack(expand=True, fill="both", padx=18, pady=18)
ctk.CTkLabel(frame, text="BIG‑IP Security Audit Tool v1.2", font=("Segoe UI", 18)).pack(pady=(0,16))
host_entry = ctk.CTkEntry(frame, placeholder_text="BIG‑IP Host (IP or FQDN)", width=315)
host_entry.pack(pady=5)
user_entry = ctk.CTkEntry(frame, placeholder_text="Username", width=315)
user_entry.pack(pady=5)
pass_entry = ctk.CTkEntry(frame, placeholder_text="Password", show="*", width=315)
pass_entry.pack(pady=5)
status_label = ctk.CTkLabel(frame, text="", font=("Segoe UI", 12), text_color="#285fb4")
status_label.pack(pady=10)
footer_label = ctk.CTkLabel(frame, text="BIG-IP Security Audit Tool by Amit Zakay", font=("Segoe UI", 11), text_color="#808080")
footer_label.pack(side="bottom", pady=(14,3))


def on_run_click():
    # 1. Disable button and update status (this is fast)
    run_button.configure(state='disabled', text="Running...")
    status_label.configure(text="Running audit checks...", text_color="#285fb4")

    # We don't need root.update_idletasks() anymore because we aren't blocking

    # 2. Get credentials from the GUI
    host = host_entry.get().strip()
    username = user_entry.get().strip()
    password = pass_entry.get()

    # 3. Create the queue to pass results
    results_queue = queue.Queue()

    # 4. Create and start the background thread
    audit_thread = threading.Thread(
        target=run_audit_in_thread,  # Function to run in thread
        args=(host, username, password, results_queue),  # Args to pass to it
        daemon=True  # A daemon thread exits when the main app exits
    )
    audit_thread.start()

    # 5. Start polling for the results from the main GUI thread
    root.after(100, check_audit_queue, results_queue)

run_button = ctk.CTkButton(frame, text="Run Audit & Export to HTML", command=on_run_click, width=220, height=38)
run_button.pack(pady=8)
root.mainloop()

 

Updated Nov 05, 2025
Version 6.0

2 Comments

    • amit-zakay's avatar
      amit-zakay
      Icon for Nimbostratus rankNimbostratus

      You're more than welcomed. 

      Basically , I have created it for my own use in order to get a quick overview of my customer's systems.

      However, this community helped me a lot as a passive reader, so I felt that this could be my small contribution 🙂