Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 128 additions & 0 deletions backend/app/analysis/checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand All @@ -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,
]
79 changes: 79 additions & 0 deletions backend/app/recommendations/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
),
}


Expand Down
147 changes: 147 additions & 0 deletions backend/tests/test_analysis.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ─────────────────────────────────────


Expand Down
Loading