Spaces:
Running on Zero
Running on Zero
| 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 | |