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