Spaces:
Running on Zero
Running on Zero
File size: 2,936 Bytes
261a83d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | import logging
logger = logging.getLogger(__name__)
def validate_rfc_compliance(packet_data: list, has_forensics: bool = False) -> list:
"""
Protocol validation engine. Checks for impossible TCP flags,
malformed headers, and other protocol RFC violations.
"""
if not packet_data:
return []
alerts = []
reported_anomalies = set()
anomaly_id_counter = 3000000
for pkt in packet_data:
src_ip = pkt.get("src_ip")
if not src_ip: continue
proto = pkt.get("protocol", "")
flags = pkt.get("flags", "")
if proto == "TCP" and flags:
anomaly_type = None
note = ""
# Rule 1: SYN and FIN together (Impossible under RFC 793)
if "S" in flags and "F" in flags:
anomaly_type = "SYN-FIN"
note = "TCP SYN and FIN flags set simultaneously (Invalid RFC 793)."
# Rule 2: FIN but no ACK (Rare, often a stealth scan like Nmap FIN scan)
elif "F" in flags and "A" not in flags:
anomaly_type = "FIN-No-ACK"
note = "TCP FIN flag set without ACK (Stealth Scan pattern)."
# Rule 3: NULL scan (No flags set)
elif flags == "None" or flags == "":
anomaly_type = "NULL-Scan"
note = "TCP packet with no flags set (NULL scan pattern)."
# Rule 4: SYN, FIN, PSH, URG together (XMAS scan)
elif "F" in flags and "P" in flags and "U" in flags:
anomaly_type = "XMAS-Scan"
note = "TCP FIN, PSH, URG flags set simultaneously (XMAS Scan pattern)."
if anomaly_type:
sig_key = f"{anomaly_type}_{src_ip}"
if sig_key not in reported_anomalies:
anomaly_id_counter += 1
alerts.append({
"sid": anomaly_id_counter,
"msg": f"NetNerve Protocol Validation: {anomaly_type} Detected",
"signature": f"NetNerve Protocol Validation: {anomaly_type} Detected",
"category": "Protocol Anomaly",
"severity": "HIGH",
"priority": 1,
"src_ip": src_ip,
"dst_ip": pkt.get("dst_ip", "Multiple"),
"src_port": pkt.get("src_port", "*"),
"dst_port": pkt.get("dst_port", "*"),
"protocol": "TCP",
"timestamp": pkt.get("timestamp_iso", ""),
"count": 1,
"note": note
})
reported_anomalies.add(sig_key)
logger.info(f"Protocol Validator generated {len(alerts)} alerts.")
return alerts
|