diff --git a/backend/app/analysis/checks.py b/backend/app/analysis/checks.py index 4c80b66..a6b1208 100644 --- a/backend/app/analysis/checks.py +++ b/backend/app/analysis/checks.py @@ -608,6 +608,128 @@ def check_wireguard_vpn(device: Device) -> list[RiskData]: ] +# ── Compound / pattern-based checks ────────────────────────────────────────── + + +def check_multiple_admin_panels(device: Device) -> list[RiskData]: + """MEDIUM — 3+ management ports open; unusually large admin surface.""" + admin_ports = _has_port(device, 80, 8080, 8443, 9000, 9443, 9200, 5900, 3389, 22) + if len(admin_ports) < 3: # noqa: PLR2004 — 3 is the meaningful threshold + return [] + port_list = ", ".join(str(p.port_number) for p in admin_ports) + return [ + RiskData( + check_id="multiple_admin_panels", + severity="medium", + title="Multiple admin/management ports open", + description=( + f"Device {device.ip_address} has {len(admin_ports)} management or " + f"admin ports open simultaneously ({port_list}). " + "Each extra management interface is an additional attack surface. " + "Disable any admin panels or remote-access services that are not " + "actively required." + ), + ) + ] + + +def check_database_and_web_exposed(device: Device) -> list[RiskData]: + """HIGH — database port open alongside a public-facing web port.""" + db_ports = _has_port(device, 3306, 5432, 6379, 27017, 1433, 5984) + web_ports = _has_port(device, 80, 8080) + if not db_ports or not web_ports: + return [] + db_list = ", ".join(str(p.port_number) for p in db_ports) + web_list = ", ".join(str(p.port_number) for p in web_ports) + return [ + RiskData( + check_id="database_and_web_exposed", + severity="high", + title="Database and unencrypted web port open together", + description=( + f"Device {device.ip_address} has database port(s) ({db_list}) open " + f"alongside unencrypted web port(s) ({web_list}). " + "An attacker who exploits the web layer may pivot directly to the " + "database. The database port should not be network-accessible, and " + "all web traffic should be served over HTTPS." + ), + ) + ] + + +def check_cleartext_credential_surface(device: Device) -> list[RiskData]: + """HIGH — Telnet, FTP, and HTTP all open; maximum cleartext exposure.""" + telnet = _has_port(device, 23) + ftp = _has_port(device, 21) + http = _has_port(device, 80) + if not (telnet and ftp and http): + return [] + return [ + RiskData( + check_id="cleartext_credential_surface", + severity="high", + title="Cleartext credential surface (Telnet + FTP + HTTP all open)", + description=( + f"Device {device.ip_address} has Telnet (23), FTP (21), and HTTP (80) " + "all open simultaneously. Every one of these protocols transmits " + "credentials and data in plaintext. An attacker on the same network " + "can capture login credentials with a passive packet capture. " + "Disable all three and replace with SSH, SFTP/SCP, and HTTPS." + ), + ) + ] + + +def check_remote_access_no_encryption(device: Device) -> list[RiskData]: + """HIGH — RDP or VNC open with no TLS service detected on the device.""" + rdp = _has_port(device, 3389) + vnc = _has_port(device, 5900) + if not (rdp or vnc): + return [] + # If any TLS-capable port is open, the device at least has *some* encrypted path + tls_ports = _has_port(device, 443, 8443, 22) + if tls_ports: + return [] + remote_list = ", ".join(str(p.port_number) for p in (rdp + vnc)) + return [ + RiskData( + check_id="remote_access_no_encryption", + severity="high", + title="Remote desktop open with no encrypted alternative", + description=( + f"Device {device.ip_address} has remote access port(s) ({remote_list}) " + "open and no TLS-encrypted service (443/8443/22) detected. " + "RDP and VNC can leak screen content and credentials if not tunnelled " + "through an encrypted channel such as SSH or a VPN. " + "Disable direct RDP/VNC exposure and access the device via a VPN or " + "SSH tunnel instead." + ), + ) + ] + + +def check_ssh_and_telnet_both_open(device: Device) -> list[RiskData]: + """MEDIUM — SSH and Telnet both open; Telnet likely a forgotten legacy service.""" + ssh = _has_port(device, 22) + telnet = _has_port(device, 23) + if not (ssh and telnet): + return [] + return [ + RiskData( + check_id="ssh_and_telnet_both_open", + severity="medium", + title="SSH and Telnet both open", + description=( + f"Device {device.ip_address} has both SSH (22) and Telnet (23) open. " + "SSH is the secure replacement for Telnet — having both suggests " + "Telnet was never disabled after SSH was enabled. " + "Disable Telnet immediately; any legitimate remote access should use " + "SSH with key-based authentication only." + ), + ) + ] + + # ── Master list of all checks ───────────────────────────────────────────────── ALL_CHECKS = [ @@ -633,4 +755,10 @@ def check_wireguard_vpn(device: Device) -> list[RiskData]: check_home_assistant_exposed, check_tftp_open, check_wireguard_vpn, + # Compound / pattern checks + check_multiple_admin_panels, + check_database_and_web_exposed, + check_cleartext_credential_surface, + check_remote_access_no_encryption, + check_ssh_and_telnet_both_open, ] diff --git a/backend/app/recommendations/__init__.py b/backend/app/recommendations/__init__.py index 90fb54a..39a8235 100644 --- a/backend/app/recommendations/__init__.py +++ b/backend/app/recommendations/__init__.py @@ -402,6 +402,85 @@ class _Advice: effort="low", impact="low", ), + "multiple_admin_panels": _Advice( + title="Reduce the management attack surface", + description=( + "Having many admin and remote-access ports open simultaneously increases the " + "number of ways an attacker can attempt to compromise the device." + ), + steps=[ + "Audit each open management port and identify which services are actually in use.", + "Disable or stop any admin interfaces, remote desktop, or web UIs not actively needed.", + "Where possible, consolidate management to a single encrypted channel (SSH or HTTPS).", + "Apply firewall rules to restrict remaining management ports to trusted hosts only.", + ], + effort="low", + impact="medium", + ), + "database_and_web_exposed": _Advice( + title="Isolate the database from the network and enforce HTTPS", + description=( + "A database port open alongside an unencrypted web interface creates a " + "two-step path from public HTTP exploit to full database access." + ), + steps=[ + "Bind the database to 127.0.0.1 or a private interface only — " + "it should never be network-accessible from outside the host.", + "If the database must be remotely managed, use an SSH tunnel.", + "Redirect all HTTP (port 80) traffic to HTTPS and obtain a TLS certificate.", + "Apply a firewall rule blocking the database port from all external sources.", + ], + effort="medium", + impact="high", + ), + "cleartext_credential_surface": _Advice( + title="Replace all cleartext protocols immediately", + description=( + "Running Telnet, FTP, and HTTP together means any credential entered on " + "this device can be captured by a passive network observer." + ), + steps=[ + "Disable Telnet and replace with SSH for remote shell access.", + "Disable FTP and replace with SFTP or SCP.", + "Disable plain HTTP and serve all web content over HTTPS only.", + "After disabling, verify the ports are no longer listening with 'ss -tlnp'.", + ], + effort="low", + impact="high", + ), + "remote_access_no_encryption": _Advice( + title="Tunnel remote desktop through an encrypted channel", + description=( + "RDP and VNC without an encrypted wrapper expose session content and " + "credentials to anyone on the same network." + ), + steps=[ + "Do not expose RDP (3389) or VNC (5900) directly on the LAN if avoidable.", + "Set up a VPN (WireGuard, OpenVPN) and access the device only through it.", + "Alternatively, tunnel RDP/VNC over an SSH port forward.", + "If direct access is required, enable NLA (Network Level Authentication) " + "for RDP and set a strong VNC password with TLS where supported.", + "Apply a firewall rule to restrict RDP/VNC to specific trusted host IPs.", + ], + effort="medium", + impact="high", + ), + "ssh_and_telnet_both_open": _Advice( + title="Disable Telnet — SSH is already available", + description=( + "SSH is already running, making Telnet entirely redundant and dangerous. " + "Disable it immediately." + ), + steps=[ + "Identify and stop the Telnet service: " + "'sudo systemctl stop telnet' or 'sudo systemctl disable inetd'.", + "Verify Telnet is no longer listening: 'ss -tlnp | grep 23'.", + "Ensure SSH is configured with key-based authentication and " + "PasswordAuthentication disabled in /etc/ssh/sshd_config.", + ], + effort="low", + impact="medium", + ), } diff --git a/backend/tests/test_analysis.py b/backend/tests/test_analysis.py index 24dc2aa..90037c6 100644 --- a/backend/tests/test_analysis.py +++ b/backend/tests/test_analysis.py @@ -599,6 +599,153 @@ def test_check_wireguard_vpn_no_false_positive_tcp(): assert check_wireguard_vpn(device) == [] +# ── check_multiple_admin_panels ─────────────────────────────────────────────── + + +@pytest.mark.unit +def test_check_multiple_admin_panels_fires_on_three_or_more(): + from app.analysis.checks import check_multiple_admin_panels + + device = _make_device(ports=[_make_port(80), _make_port(8080), _make_port(9000)]) + results = check_multiple_admin_panels(device) + assert len(results) == 1 + assert results[0].check_id == "multiple_admin_panels" + assert results[0].severity == "medium" + + +@pytest.mark.unit +def test_check_multiple_admin_panels_no_false_positive_two_ports(): + from app.analysis.checks import check_multiple_admin_panels + + device = _make_device(ports=[_make_port(80), _make_port(8080)]) + assert check_multiple_admin_panels(device) == [] + + +# ── check_database_and_web_exposed ──────────────────────────────────────────── + + +@pytest.mark.unit +def test_check_database_and_web_exposed_fires_mysql_and_http(): + from app.analysis.checks import check_database_and_web_exposed + + device = _make_device(ports=[_make_port(3306), _make_port(80)]) + results = check_database_and_web_exposed(device) + assert len(results) == 1 + assert results[0].check_id == "database_and_web_exposed" + assert results[0].severity == "high" + + +@pytest.mark.unit +def test_check_database_and_web_exposed_fires_redis_and_http(): + from app.analysis.checks import check_database_and_web_exposed + + device = _make_device(ports=[_make_port(6379), _make_port(8080)]) + results = check_database_and_web_exposed(device) + assert len(results) == 1 + assert results[0].check_id == "database_and_web_exposed" + + +@pytest.mark.unit +def test_check_database_and_web_no_false_positive_db_only(): + from app.analysis.checks import check_database_and_web_exposed + + device = _make_device(ports=[_make_port(3306)]) + assert check_database_and_web_exposed(device) == [] + + +@pytest.mark.unit +def test_check_database_and_web_no_false_positive_https(): + from app.analysis.checks import check_database_and_web_exposed + + # 443 (HTTPS) is not a trigger — only plain HTTP 80/8080 + device = _make_device(ports=[_make_port(3306), _make_port(443)]) + assert check_database_and_web_exposed(device) == [] + + +# ── check_cleartext_credential_surface ─────────────────────────────────────── + + +@pytest.mark.unit +def test_check_cleartext_credential_surface_fires_all_three(): + from app.analysis.checks import check_cleartext_credential_surface + + device = _make_device(ports=[_make_port(23), _make_port(21), _make_port(80)]) + results = check_cleartext_credential_surface(device) + assert len(results) == 1 + assert results[0].check_id == "cleartext_credential_surface" + assert results[0].severity == "high" + + +@pytest.mark.unit +def test_check_cleartext_credential_surface_no_false_positive_missing_one(): + from app.analysis.checks import check_cleartext_credential_surface + + device = _make_device(ports=[_make_port(23), _make_port(21)]) + assert check_cleartext_credential_surface(device) == [] + + +# ── check_remote_access_no_encryption ──────────────────────────────────────── + + +@pytest.mark.unit +def test_check_remote_access_no_encryption_fires_rdp_only(): + from app.analysis.checks import check_remote_access_no_encryption + + device = _make_device(ports=[_make_port(3389)]) + results = check_remote_access_no_encryption(device) + assert len(results) == 1 + assert results[0].check_id == "remote_access_no_encryption" + assert results[0].severity == "high" + + +@pytest.mark.unit +def test_check_remote_access_no_encryption_no_false_positive_with_ssh(): + from app.analysis.checks import check_remote_access_no_encryption + + device = _make_device(ports=[_make_port(3389), _make_port(22)]) + assert check_remote_access_no_encryption(device) == [] + + +@pytest.mark.unit +def test_check_remote_access_no_encryption_fires_vnc_no_tls(): + from app.analysis.checks import check_remote_access_no_encryption + + device = _make_device(ports=[_make_port(5900)]) + results = check_remote_access_no_encryption(device) + assert len(results) == 1 + assert results[0].check_id == "remote_access_no_encryption" + + +# ── check_ssh_and_telnet_both_open ──────────────────────────────────────────── + + +@pytest.mark.unit +def test_check_ssh_and_telnet_both_open_fires(): + from app.analysis.checks import check_ssh_and_telnet_both_open + + device = _make_device(ports=[_make_port(22), _make_port(23)]) + results = check_ssh_and_telnet_both_open(device) + assert len(results) == 1 + assert results[0].check_id == "ssh_and_telnet_both_open" + assert results[0].severity == "medium" + + +@pytest.mark.unit +def test_check_ssh_and_telnet_no_false_positive_ssh_only(): + from app.analysis.checks import check_ssh_and_telnet_both_open + + device = _make_device(ports=[_make_port(22)]) + assert check_ssh_and_telnet_both_open(device) == [] + + +@pytest.mark.unit +def test_check_ssh_and_telnet_no_false_positive_telnet_only(): + from app.analysis.checks import check_ssh_and_telnet_both_open + + device = _make_device(ports=[_make_port(23)]) + assert check_ssh_and_telnet_both_open(device) == [] + + # ── Integration: run_checks with real DB ─────────────────────────────────────