165 lines
8.0 KiB
Python
165 lines
8.0 KiB
Python
import tempfile
|
||
import unittest
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
from librenet_scanner.models import Host, PortInfo
|
||
from librenet_scanner.storage import HistoryStore
|
||
|
||
|
||
def ports(*values: int) -> list[PortInfo]:
|
||
return [PortInfo(port=value, service="test") for value in values]
|
||
|
||
|
||
class IdentificationFreshness0416Tests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.store = HistoryStore(Path(self.tmp.name) / "history.sqlite3")
|
||
self.scope = "ipv4:192.168.10.0/24"
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _row(self, identity_suffix: str = "mac:00:11:22:33:44:55"):
|
||
with self.store._connect() as conn:
|
||
return conn.execute(
|
||
"SELECT * FROM endpoint_identification WHERE identity LIKE ? ORDER BY identity LIMIT 1",
|
||
(f"%{identity_suffix}",),
|
||
).fetchone()
|
||
|
||
def test_standard_does_not_refresh_os_seen_at(self):
|
||
deep_time = "2025-01-02T03:04:05+00:00"
|
||
standard_time = "2026-08-22T12:00:00+00:00"
|
||
deep = Host(
|
||
ip="192.168.10.20", mac="00:11:22:33:44:55",
|
||
os_name="Debian 13", os_accuracy=97, ports=ports(22, 443),
|
||
)
|
||
self.store.remember_identifications([deep], "Approfondi", observed_at=deep_time, scope=self.scope)
|
||
standard = Host(
|
||
ip="192.168.10.20", mac=deep.mac, os_name="Linux 6.x", ports=ports(22, 443),
|
||
)
|
||
self.store.remember_identifications([standard], "Standard", observed_at=standard_time, scope=self.scope)
|
||
row = self._row()
|
||
self.assertIsNotNone(row)
|
||
self.assertEqual(row["os_name"], "Debian 13")
|
||
self.assertEqual(row["os_seen_at"], deep_time)
|
||
self.assertEqual(row["os_source"], "Approfondi")
|
||
self.assertEqual(row["updated_at"], standard_time)
|
||
|
||
def test_old_os_stays_old_even_when_endpoint_updated_at_is_recent(self):
|
||
old = Host(
|
||
ip="192.168.10.20", mac="00:11:22:33:44:55",
|
||
os_name="Debian 13", os_accuracy=99, ports=ports(22, 443),
|
||
)
|
||
self.store.remember_identifications([old], "Approfondi", scope=self.scope)
|
||
stale = (datetime.now(timezone.utc) - timedelta(days=365)).astimezone().isoformat(timespec="seconds")
|
||
fresh = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
||
with self.store._connect() as conn:
|
||
conn.execute(
|
||
"UPDATE endpoint_identification SET os_seen_at = ?, updated_at = ?",
|
||
(stale, fresh),
|
||
)
|
||
moved = Host(ip="192.168.10.99", mac=old.mac)
|
||
self.store.apply_identifications([moved], scope=self.scope)
|
||
self.assertEqual(moved.remembered_os_name, "")
|
||
|
||
def test_unsafe_mac_move_does_not_overwrite_original_endpoint_before_matching(self):
|
||
old = Host(
|
||
ip="192.168.10.20", mac="00:11:22:33:44:55",
|
||
hostname="old.local", os_name="Debian 13", os_accuracy=99,
|
||
ports=ports(22, 443),
|
||
)
|
||
self.store.remember_identifications([old], "Approfondi", scope=self.scope)
|
||
moved = Host(ip="192.168.10.99", mac=old.mac, hostname="other.local")
|
||
# Même si une future régression écrit avant d'appliquer, l'ancien endpoint
|
||
# doit rester intact : l'observation ambiguë est scindée en macip.
|
||
self.store.remember_identifications([moved], "Standard", scope=self.scope)
|
||
with self.store._connect() as conn:
|
||
base = conn.execute(
|
||
"SELECT * FROM endpoint_identification WHERE identity = ?",
|
||
(f"{self.scope}::mac:{old.mac}",),
|
||
).fetchone()
|
||
split = conn.execute(
|
||
"SELECT * FROM endpoint_identification WHERE identity = ?",
|
||
(f"{self.scope}::macip:{old.mac}@192.168.10.99",),
|
||
).fetchone()
|
||
self.assertIsNotNone(base)
|
||
self.assertEqual(base["ip"], "192.168.10.20")
|
||
self.assertEqual(base["os_name"], "Debian 13")
|
||
self.assertIsNotNone(split)
|
||
|
||
|
||
class DataDeletion0416Tests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.store = HistoryStore(Path(self.tmp.name) / "history.sqlite3")
|
||
self.scope = "ipv4:192.168.10.0/24"
|
||
self.host = Host(
|
||
ip="192.168.10.20", mac="00:11:22:33:44:55", hostname="srv.local",
|
||
os_name="Debian 13", os_accuracy=99, ports=ports(22),
|
||
)
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_clear_scan_history_does_not_delete_identification_or_metadata(self):
|
||
self.store.remember_identifications([self.host], "Approfondi", scope=self.scope)
|
||
self.store.save_host_metadata(self.host, favorite=True, note="Important")
|
||
self.store.save_scan("192.168.10.0/24", "Approfondi", [self.host], "deep-test")
|
||
self.assertEqual(self.store.clear_scan_history(), 1)
|
||
self.assertEqual(self.store.recent_scans(), [])
|
||
current = Host(ip=self.host.ip, mac=self.host.mac, ports=ports(22))
|
||
self.store.apply_identifications([current], scope=self.scope)
|
||
self.assertEqual(current.remembered_os_name, "Debian 13")
|
||
self.assertTrue(self.store.host_metadata(self.host)["favorite"])
|
||
|
||
def test_forget_selected_identification_keeps_metadata(self):
|
||
self.store.remember_identifications([self.host], "Approfondi", scope=self.scope)
|
||
self.store.save_host_metadata(self.host, favorite=True, group_name="Infra", note="Note")
|
||
deleted = self.store.forget_identification_for_host(self.host, scope=self.scope)
|
||
self.assertGreaterEqual(deleted, 1)
|
||
current = Host(ip=self.host.ip, mac=self.host.mac, ports=ports(22))
|
||
self.store.apply_identifications([current], scope=self.scope)
|
||
self.assertEqual(current.remembered_os_name, "")
|
||
metadata = self.store.host_metadata(self.host)
|
||
self.assertTrue(metadata["favorite"])
|
||
self.assertEqual(metadata["group_name"], "Infra")
|
||
|
||
def test_forget_scope_does_not_touch_other_network(self):
|
||
other_scope = "ipv4:192.168.20.0/24"
|
||
other = Host(ip="192.168.20.20", mac="00:11:22:AA:BB:CC", os_name="OpenWrt 24.10", ports=ports(22, 80))
|
||
self.store.remember_identifications([self.host], "Approfondi", scope=self.scope)
|
||
self.store.remember_identifications([other], "Approfondi", scope=other_scope)
|
||
self.assertEqual(self.store.forget_identifications_for_scope(self.scope), 1)
|
||
cur1 = Host(ip=self.host.ip, mac=self.host.mac, ports=ports(22))
|
||
cur2 = Host(ip=other.ip, mac=other.mac, ports=ports(22, 80))
|
||
self.store.apply_identifications([cur1], scope=self.scope)
|
||
self.store.apply_identifications([cur2], scope=other_scope)
|
||
self.assertEqual(cur1.remembered_os_name, "")
|
||
self.assertEqual(cur2.remembered_os_name, "OpenWrt 24.10")
|
||
|
||
def test_forget_all_identifications_leaves_scan_history(self):
|
||
self.store.remember_identifications([self.host], "Approfondi", scope=self.scope)
|
||
self.store.save_scan("192.168.10.0/24", "Approfondi", [self.host], "deep-test")
|
||
self.assertGreaterEqual(self.store.forget_all_identifications(), 1)
|
||
self.assertEqual(len(self.store.recent_scans()), 1)
|
||
current = Host(ip=self.host.ip, mac=self.host.mac, ports=ports(22))
|
||
self.store.apply_identifications([current], scope=self.scope)
|
||
self.assertEqual(current.remembered_os_name, "")
|
||
|
||
|
||
class SourceOrder0416Tests(unittest.TestCase):
|
||
def test_scan_finish_applies_memory_before_updating_endpoint_observation(self):
|
||
source = Path(__file__).parents[1] / "src" / "librenet_scanner" / "ui.py"
|
||
text = source.read_text(encoding="utf-8")
|
||
start = text.index("def _scan_finished")
|
||
end = text.index("# ---------- Constructeurs en ligne", start)
|
||
block = text[start:end]
|
||
self.assertLess(block.index("apply_identifications"), block.index("remember_identifications"))
|
||
self.assertIn("Effacer l’affichage", text)
|
||
self.assertIn("Oublier les identifications", text)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|