feat(#441): lot 6 netcross_core, exploit_signatures et banners

Résumés DEBUG, détections et bannières en TRACE ; charge utile malformée en TRACE au lieu de logger.exception.

Author: deusyv

src/netcross_core/application/banners.py

@@ -395,9 +395,18 @@ def extract_banners(proto: str, sport: int | None, dport: int | None, payload: b
                 banners = _greeting_banners(payload, sport)
         elif sport == 53:
             banners = _dns_banners(payload, proto)
+        if banners:
+            logger.trace(
+                "extract_banners: {} banniere(s) {} (port source {}){}",
+                len(banners),
+                "/".join(sorted({b.protocol for b in banners})),
+                sport,
+                ", tronque a " + str(MAX_BANNERS_PER_PACKET) if len(banners) > MAX_BANNERS_PER_PACKET else "",
+            )
         return tuple(banners[:MAX_BANNERS_PER_PACKET])
-    except (ValueError, IndexError, struct.error, UnicodeError):
-        logger.exception("échec dans extract_banners")
+    except (ValueError, IndexError, struct.error, UnicodeError) as exc:
+        # charge utile malformee : resultat attendu, a chaque paquet -> TRACE sans traceback
+        logger.trace("extract_banners: charge utile {} malformee ({} octets) : {!r}", proto, len(payload), exc)
         return ()
 
 
@@ -439,6 +448,11 @@ def build_service_fingerprints(packets: Iterable[Pkt]) -> list[dict]:
                 }
             elif len(b.raw) > len(entry["banner"]):
                 entry["banner"] = b.raw
+    logger.debug(
+        "build_service_fingerprints: {} logiciel(s) identifie(s) sur {} hote(s)",
+        len(found),
+        len({(e["point"], e["host"]) for e in found.values()}),
+    )
     return sorted(
         found.values(),
         key=lambda e: (e["point"], e["host"], e["port"] or 0, e["service"].lower(), e["version"] or "", e["role"]),

src/netcross_core/exploit_signatures.py

@@ -729,12 +729,19 @@ class ExploitScanner:
     def __init__(self, signatures: Sequence[Signature] | None = None) -> None:
         self.signatures: list[Signature] = list(signatures) if signatures is not None else load_signatures()
         self._flows: dict[tuple[str, int | None, str, int | None, str], _FlowState] = {}
+        logger.debug(
+            "ExploitScanner: {} signature(s) ({}), dont {} a detecteur structurel",
+            len(self.signatures),
+            "fournies" if signatures is not None else "base par defaut",
+            sum(1 for sig in self.signatures if sig.detector is not None),
+        )
 
     def _state(self, pkt: PacketLike) -> _FlowState:
         key = (pkt.src, pkt.sport, pkt.dst, pkt.dport, pkt.proto)
         state = self._flows.get(key)
         if state is None:
             if len(self._flows) >= MAX_TRACKED_FLOWS:
+                logger.trace("ExploitScanner._state: {} flux suivis, eviction du plus ancien", len(self._flows))
                 del self._flows[next(iter(self._flows))]  # evince le flux le plus ancien
             state = self._flows[key] = _FlowState()
         return state
@@ -781,6 +788,16 @@ def scan_packet(self, pkt: PacketLike, point: str = "") -> list[Detection]:
             else:
                 hit = self._match_pattern(sig, view)
             if hit is not None:
+                logger.trace(
+                    "scan_packet: signature {} sur la trame {} ({}:{} -> {}:{}), cible {}",
+                    sig.id,
+                    pkt.frame_number,
+                    pkt.src,
+                    pkt.sport,
+                    pkt.dst,
+                    pkt.dport,
+                    hit[0],
+                )
                 detections.append(
                     Detection(
                         signature_id=sig.id,
@@ -807,8 +824,18 @@ def scan_packet(self, pkt: PacketLike, point: str = "") -> list[Detection]:
     def scan_packets(self, packets: Iterable[PacketLike], point: str = "") -> list[Detection]:
         """Detections pour des paquets fournis dans l'ordre de capture."""
         detections: list[Detection] = []
+        scanned = 0
         for pkt in packets:
+            scanned += 1
             detections.extend(self.scan_packet(pkt, point))
+        logger.debug(
+            "scan_packets: point {!r}, {} paquet(s), {} detection(s) sur {} signature(s) distincte(s), {} flux suivis",
+            point,
+            scanned,
+            len(detections),
+            len({d.signature_id for d in detections}),
+            len(self._flows),
+        )
         return detections
 
 
@@ -825,7 +852,9 @@ def detect_exploits(
 
 def detections_to_dicts(detections: Iterable[Detection]) -> list[dict[str, Any]]:
     """Serialisation JSON stable (sans charge utile)."""
-    return [asdict(d) | {"cves": list(d.cves)} for d in detections]
+    out = [asdict(d) | {"cves": list(d.cves)} for d in detections]
+    logger.debug("detections_to_dicts: {} detection(s)", len(out))
+    return out
 
 
 __all__ = [

tests/test_journalisation_signatures.py

@@ -0,0 +1,51 @@
+"""#441, lot 6 : journalisation de exploit_signatures et application/banners."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from loguru import logger
+
+from netcross_core.application.banners import build_service_fingerprints, extract_banners
+from netcross_core.exploit_signatures import ExploitScanner, detections_to_dicts
+
+
+@pytest.fixture
+def journal():
+    logger.enable("netcross_core")
+    lignes: list[tuple[str, str]] = []
+    sink = logger.add(lambda m: lignes.append((m.record["level"].name, m.record["message"])), level="TRACE")
+    yield lignes
+    logger.remove(sink)
+
+
+def _vu(journal, niveau, fragment):
+    return any(n == niveau and fragment in m for n, m in journal)
+
+
+def test_scanner_resume_en_debug(journal):
+    scanner = ExploitScanner()
+    pkt = SimpleNamespace(
+        payload=b"GET / HTTP/1.1\\r\\n\\r\\n", proto="TCP", src="10.0.0.1", sport=50000,
+        dst="10.0.0.2", dport=80, ts=1.0, frame_number=1, payload_hash=None,
+    )  # fmt: skip
+    detections = scanner.scan_packets([pkt], point="A")
+    detections_to_dicts(detections)
+    assert _vu(journal, "DEBUG", "base par defaut")
+    assert _vu(journal, "DEBUG", "scan_packets: point 'A', 1 paquet(s)")
+    assert _vu(journal, "DEBUG", "detections_to_dicts:")
+
+
+def test_bannieres_trace_et_resume(journal):
+    banners = extract_banners("TCP", 22, 50000, b"SSH-2.0-OpenSSH_9.6\\r\\n")
+    assert banners
+    assert _vu(journal, "TRACE", "extract_banners: 1 banniere(s)")
+    build_service_fingerprints([])
+    assert _vu(journal, "DEBUG", "build_service_fingerprints: 0 logiciel(s)")
+
+
+def test_charge_utile_malformee_sans_traceback(journal):
+    # reponse DNS tronquee sur le port 53 : parsee puis rejetee
+    assert extract_banners("UDP", 53, 40000, b"\\x00\\x01\\x85\\x00\\x00\\x01\\x00\\x01") == ()
+    assert not any(n in ("ERROR", "CRITICAL") for n, _m in journal)