feat: improve IoT device detection via two-phase scan and mDNS discovery
- Two-phase nmap: ARP sweep first to find all alive hosts (incl. IoT with no open TCP ports), then port scan only alive hosts - mDNS/Bonjour discovery via zeroconf for Shelly, ESPHome, HomeKit devices - Add CoAP ports (5683, 5684, 4915) to port scan and IoT type hints - Expand MAC OUI table with Shelly, Espressif, Sonoff, Tapo, Hue, IKEA, Tuya - IoT vendor MAC takes precedence over generic HTTP port type hints - Reorder suggest_node_type priority: iot now beats server
This commit is contained in:
Pouzor
@@ -0,0 +1,306 @@
|
||||
"""Tests for scanner: two-phase nmap, mDNS discovery, run_scan integration."""
|
||||
import uuid
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from app.db.database import Base
|
||||
from app.db.models import PendingDevice, ScanRun
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_run_id() -> str:
|
||||
return str(uuid.uuid4())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def mem_db():
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
yield factory
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
def _make_scan_run(run_id: str) -> ScanRun:
|
||||
return ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _nmap_arp_sweep
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_nmap_arp_sweep_returns_alive_hosts():
|
||||
from app.services.scanner import _nmap_arp_sweep
|
||||
|
||||
mock_nm = MagicMock()
|
||||
mock_nm.all_hosts.return_value = ["192.168.1.1", "192.168.1.2"]
|
||||
mock_nm.__getitem__ = lambda self, host: MagicMock(
|
||||
state=lambda: "up",
|
||||
get=lambda key, default=None: {"mac": "aa:bb:cc:dd:ee:ff"} if key == "addresses" else default,
|
||||
)
|
||||
|
||||
with patch("app.services.scanner.nmap.PortScanner", return_value=mock_nm), \
|
||||
patch("app.services.scanner._resolve_hostname", return_value=None):
|
||||
result = _nmap_arp_sweep("192.168.1.0/24")
|
||||
|
||||
assert set(result.keys()) == {"192.168.1.1", "192.168.1.2"}
|
||||
for host in result.values():
|
||||
assert host["open_ports"] == [] # empty until phase 2
|
||||
|
||||
|
||||
def test_nmap_arp_sweep_skips_down_hosts():
|
||||
from app.services.scanner import _nmap_arp_sweep
|
||||
|
||||
states = {"192.168.1.1": "up", "192.168.1.2": "down"}
|
||||
|
||||
mock_nm = MagicMock()
|
||||
mock_nm.all_hosts.return_value = list(states.keys())
|
||||
|
||||
def getitem(host):
|
||||
m = MagicMock()
|
||||
m.state.return_value = states[host]
|
||||
m.get.return_value = {}
|
||||
return m
|
||||
|
||||
mock_nm.__getitem__ = lambda self, host: getitem(host)
|
||||
|
||||
with patch("app.services.scanner.nmap.PortScanner", return_value=mock_nm), \
|
||||
patch("app.services.scanner._resolve_hostname", return_value=None):
|
||||
result = _nmap_arp_sweep("192.168.1.0/24")
|
||||
|
||||
assert "192.168.1.1" in result
|
||||
assert "192.168.1.2" not in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _nmap_port_scan
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_nmap_port_scan_merges_ports():
|
||||
from app.services.scanner import _nmap_port_scan
|
||||
|
||||
alive = {
|
||||
"192.168.1.10": {"ip": "192.168.1.10", "hostname": None, "mac": None, "os": None, "open_ports": []},
|
||||
}
|
||||
|
||||
mock_nm = MagicMock()
|
||||
mock_nm.all_hosts.return_value = ["192.168.1.10"]
|
||||
mock_nm.__getitem__ = lambda self, host: MagicMock(
|
||||
all_protocols=lambda: ["tcp"],
|
||||
**{"__getitem__": lambda self2, proto: {
|
||||
80: {"state": "open", "product": "nginx", "version": "1.24"},
|
||||
}},
|
||||
get=lambda key, default=None: default,
|
||||
)
|
||||
|
||||
with patch("app.services.scanner.nmap.PortScanner", return_value=mock_nm), \
|
||||
patch("app.services.scanner._extract_os", return_value=None):
|
||||
result = _nmap_port_scan(alive)
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["open_ports"][0]["port"] == 80
|
||||
|
||||
|
||||
def test_nmap_port_scan_returns_arp_only_on_failure():
|
||||
from app.services.scanner import _nmap_port_scan
|
||||
|
||||
alive = {
|
||||
"192.168.1.20": {"ip": "192.168.1.20", "hostname": None, "mac": None, "os": None, "open_ports": []},
|
||||
}
|
||||
|
||||
mock_nm = MagicMock()
|
||||
mock_nm.scan.side_effect = Exception("nmap error")
|
||||
|
||||
with patch("app.services.scanner.nmap.PortScanner", return_value=mock_nm):
|
||||
result = _nmap_port_scan(alive)
|
||||
|
||||
# Should return the ARP-found host even though port scan failed
|
||||
assert len(result) == 1
|
||||
assert result[0]["ip"] == "192.168.1.20"
|
||||
assert result[0]["open_ports"] == []
|
||||
|
||||
|
||||
def test_nmap_port_scan_includes_hosts_with_no_open_ports():
|
||||
"""IoT devices found by ARP but with no open TCP ports must still be returned."""
|
||||
from app.services.scanner import _nmap_port_scan
|
||||
|
||||
alive = {
|
||||
"192.168.1.30": {"ip": "192.168.1.30", "hostname": "shelly1.lan", "mac": "34:94:54:aa:bb:cc", "os": None, "open_ports": []},
|
||||
"192.168.1.31": {"ip": "192.168.1.31", "hostname": None, "mac": None, "os": None, "open_ports": []},
|
||||
}
|
||||
|
||||
# Port scan returns only 192.168.1.31 (e.g., .30 filtered all ports)
|
||||
mock_nm = MagicMock()
|
||||
mock_nm.all_hosts.return_value = ["192.168.1.31"]
|
||||
mock_nm.__getitem__ = lambda self, host: MagicMock(
|
||||
all_protocols=lambda: [],
|
||||
get=lambda key, default=None: default,
|
||||
)
|
||||
|
||||
with patch("app.services.scanner.nmap.PortScanner", return_value=mock_nm), \
|
||||
patch("app.services.scanner._extract_os", return_value=None):
|
||||
result = _nmap_port_scan(alive)
|
||||
|
||||
ips = {h["ip"] for h in result}
|
||||
assert "192.168.1.30" in ips, "ARP-found device with no open ports must still be returned"
|
||||
assert "192.168.1.31" in ips
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _nmap_scan (integration of both phases)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_nmap_scan_uses_mock_when_nmap_unavailable():
|
||||
from app.services.scanner import _nmap_scan
|
||||
|
||||
with patch("app.services.scanner._NMAP_AVAILABLE", False):
|
||||
result = _nmap_scan("192.168.1.0/24")
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["ip"] == "192.168.1.99"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _mdns_discover
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mdns_discover_returns_empty_when_zeroconf_unavailable():
|
||||
from app.services.scanner import _mdns_discover
|
||||
|
||||
with patch("app.services.scanner._ZEROCONF_AVAILABLE", False):
|
||||
result = await _mdns_discover()
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mdns_discover_returns_devices():
|
||||
from app.services.scanner import _mdns_discover
|
||||
|
||||
mock_info = MagicMock()
|
||||
mock_info.addresses = [b"\xc0\xa8\x01\x50"] # 192.168.1.80
|
||||
mock_info.server = "shelly1.local."
|
||||
mock_info.port = 80
|
||||
mock_info.async_request = AsyncMock(return_value=True)
|
||||
|
||||
mock_browser = AsyncMock()
|
||||
mock_browser.async_cancel = AsyncMock()
|
||||
|
||||
# Simulate a service being found during the sleep
|
||||
captured_handler: list = []
|
||||
|
||||
def fake_browser(zc, types, handlers):
|
||||
captured_handler.extend(handlers)
|
||||
return mock_browser
|
||||
|
||||
from zeroconf import ServiceStateChange
|
||||
|
||||
async def fake_sleep(t):
|
||||
# Fire the handler as if a device was discovered
|
||||
for h in captured_handler:
|
||||
h(None, "_shelly._tcp.local.", "Shelly1._shelly._tcp.local.", ServiceStateChange.Added)
|
||||
|
||||
mock_azc = AsyncMock()
|
||||
mock_azc.__aenter__ = AsyncMock(return_value=mock_azc)
|
||||
mock_azc.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_azc.zeroconf = MagicMock()
|
||||
|
||||
with patch("app.services.scanner._ZEROCONF_AVAILABLE", True), \
|
||||
patch("app.services.scanner.AsyncZeroconf", return_value=mock_azc), \
|
||||
patch("app.services.scanner.AsyncServiceBrowser", side_effect=fake_browser), \
|
||||
patch("app.services.scanner.AsyncServiceInfo", return_value=mock_info), \
|
||||
patch("asyncio.sleep", side_effect=fake_sleep):
|
||||
result = await _mdns_discover(timeout=0.01)
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["ip"] == "192.168.1.80"
|
||||
assert result[0]["hostname"] == "shelly1.local."
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# run_scan integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_adds_nmap_devices_as_pending(mem_db):
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
await session.commit()
|
||||
|
||||
nmap_hosts = [{"ip": "192.168.1.5", "hostname": "device.lan", "mac": None, "os": None, "open_ports": []}]
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=nmap_hosts), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=[]), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
result = await session.execute(
|
||||
__import__("sqlalchemy", fromlist=["select"]).select(PendingDevice)
|
||||
)
|
||||
devices = result.scalars().all()
|
||||
|
||||
assert any(d.ip == "192.168.1.5" for d in devices)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_mdns_only_device_added(mem_db):
|
||||
"""Devices found only by mDNS (not nmap) should appear in pending_devices."""
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
await session.commit()
|
||||
|
||||
mdns_hosts = [{"ip": "192.168.1.80", "hostname": "shelly1.local.", "mac": None, "os": None, "open_ports": [{"port": 80, "protocol": "tcp", "banner": ""}]}]
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=[]), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=mdns_hosts), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
from sqlalchemy import select as sa_select
|
||||
result = await session.execute(sa_select(PendingDevice).where(PendingDevice.ip == "192.168.1.80"))
|
||||
device = result.scalar_one_or_none()
|
||||
|
||||
assert device is not None
|
||||
assert device.status == "pending"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_mdns_skipped_if_already_in_nmap(mem_db):
|
||||
"""If nmap and mDNS both find the same IP, it should not be double-counted."""
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
await session.commit()
|
||||
|
||||
shared_host = {"ip": "192.168.1.10", "hostname": "device.lan", "mac": None, "os": None, "open_ports": []}
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=[shared_host]), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=[shared_host]), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
from sqlalchemy import select as sa_select
|
||||
result = await session.execute(sa_select(PendingDevice).where(PendingDevice.ip == "192.168.1.10"))
|
||||
devices = result.scalars().all()
|
||||
|
||||
assert len(devices) == 1 # not duplicated
|
||||
Reference in New Issue
Block a user