feat: add EtherType and IP Protocol enums for enhanced packet parsing and logging

backend/src/network_sniffer.py

@@ -4,7 +4,6 @@ import threading
 import os
 import time
 from typing import List, Dict, Optional
-
 import asyncpg
 from scapy.all import (
     Ether,
@@ -18,13 +17,15 @@ from scapy.all import (
     Dot1Q,
     Raw,
 )
-
-# ---- New imports for AF_PACKET optimized reader ----------------------
 import socket
 import selectors
 import errno
 import struct
 
+from src.Models.etherType import EtherTypeEnum, ethertype_from_int
+from src.Models.ip_protocol import IPProtocolEnum, protocol_from_number
+
+
 # ---- Logging ----------------------------------------------------------
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("af_packet_sniffer")
@@ -170,7 +171,7 @@ async def db_insert_packet(pkt_info: dict, bridge: str) -> None:
             pkt_info.get("vlan_id"),
             pkt_info.get("src_ip"),
             pkt_info.get("dst_ip"),
-            pkt_info.get("protocol_name"),
+            pkt_info.get("protocol"),
             pkt_info.get("src_port"),
             pkt_info.get("dst_port"),
             pkt_info["length"],
@@ -192,13 +193,15 @@ def parse_packet(pkt, bridge: str) -> None:
     Parse a scapy packet object and collect a normalized dict of metadata
     which is then scheduled to be written to the database asynchronously.
 
-    The function expects that 'pkt' is a Scapy Packet and that we set
-    'pkt.sniffed_on' before calling this function.
+    Enhancements:
+      - Uses EtherType and IP Protocol enums (human-readable) for API documentation.
+      - Records both raw numeric values and enum descriptions.
+      - Proper VLAN handling (Dot1Q inner ethertype).
+      - Keeps `protocol_name` (string) for backward compatibility.
     """
     pkt_iface = getattr(pkt, "sniffed_on", None)
     if not pkt_iface:
-        # If sniffed_on is missing we cannot determine the interface context;
-        # skip this packet.
+        # If sniffed_on is missing we cannot determine the interface context; skip this packet.
         return
 
     logger.debug("Packet captured on %s, bridge %s", pkt_iface, bridge)
@@ -210,78 +213,163 @@ def parse_packet(pkt, bridge: str) -> None:
         "raw": bytes(pkt),
         "src_mac": None,
         "dst_mac": None,
-        "eth_type": None,
+        # eth types: both raw numeric and enum/description
+        "eth_type_raw": None,
+        "eth_type": EtherTypeEnum.UNKNOWN,  # enum / human description
         "vlan_id": None,
+        # IP protocol: raw numeric and enum/description
+        "protocol_raw": None,
+        "protocol": IPProtocolEnum.UNKNOWN,
+        # backward-compatible string label (older code)
+        "protocol_name": None,
+        # L3/L4 fields
         "src_ip": None,
         "dst_ip": None,
-        "protocol_name": None,
         "src_port": None,
         "dst_port": None,
     }
 
     # --- Layer extraction ---
-    # Ethernet layer
+    # Ethernet layer + EtherType
     if Ether in pkt:
-        pkt_info["src_mac"] = pkt[Ether].src
-        pkt_info["dst_mac"] = pkt[Ether].dst
-        pkt_info["eth_type"] = hex(pkt[Ether].type)
+        try:
+            pkt_info["src_mac"] = pkt[Ether].src
+        except Exception:
+            pkt_info["src_mac"] = None
+        try:
+            pkt_info["dst_mac"] = pkt[Ether].dst
+        except Exception:
+            pkt_info["dst_mac"] = None
 
-    # VLAN (802.1Q)
-    if Dot1Q in pkt:
-        pkt_info["vlan_id"] = pkt[Dot1Q].vlan
+        # Base ethertype (may be 0x8100 for VLAN)
+        eth_type_raw: Optional[int] = None
+        try:
+            eth_type_raw = int(pkt[Ether].type)
+        except Exception:
+            eth_type_raw = None
 
-    # ARP
+        # VLAN (Dot1Q) may contain the inner ethertype
+        if Dot1Q in pkt:
+            try:
+                # Dot1Q.type is the encapsulated ethertype
+                inner = int(pkt[Dot1Q].type)
+                if inner:
+                    eth_type_raw = inner
+            except Exception:
+                # ignore and keep whatever eth_type_raw was
+                pass
+            # capture vlan id if present
+            try:
+                pkt_info["vlan_id"] = int(pkt[Dot1Q].vlan)
+            except Exception:
+                pkt_info["vlan_id"] = None
+
+        # Fill eth_type fields (raw + enum)
+        if eth_type_raw is not None:
+            pkt_info["eth_type_raw"] = eth_type_raw
+            try:
+                pkt_info["eth_type"] = ethertype_from_int(eth_type_raw)
+            except Exception:
+                pkt_info["eth_type"] = EtherTypeEnum.UNKNOWN
+        else:
+            # Unknown / missing ethertype
+            pkt_info["eth_type_raw"] = None
+            pkt_info["eth_type"] = EtherTypeEnum.UNKNOWN
+
+    # ARP (layer 2/3)
     if ARP in pkt:
         pkt_info["protocol_name"] = "ARP"
-        pkt_info["src_ip"] = pkt[ARP].psrc
-        pkt_info["dst_ip"] = pkt[ARP].pdst
+        pkt_info["src_ip"] = getattr(pkt[ARP], "psrc", None)
+        pkt_info["dst_ip"] = getattr(pkt[ARP], "pdst", None)
         pkt_info["src_port"] = None
         pkt_info["dst_port"] = None
+        # ARP is a L2 protocol — leave protocol_raw/protocol as UNKNOWN (or set to a sentinel if desired)
 
     # IPv4
     if IP in pkt:
-        pkt_info["src_ip"] = pkt[IP].src
-        pkt_info["dst_ip"] = pkt[IP].dst
-        proto = pkt[IP].proto
-        if proto == 6 and TCP in pkt:
+        try:
+            pkt_info["src_ip"] = pkt[IP].src
+        except Exception:
+            pkt_info["src_ip"] = None
+        try:
+            pkt_info["dst_ip"] = pkt[IP].dst
+        except Exception:
+            pkt_info["dst_ip"] = None
+
+        # protocol number (IPv4 'protocol' field)
+        try:
+            proto_num = int(pkt[IP].proto)
+        except Exception:
+            proto_num = None
+
+        if proto_num is not None:
+            pkt_info["protocol_raw"] = proto_num
+            try:
+                pkt_info["protocol"] = protocol_from_number(proto_num)
+            except Exception:
+                pkt_info["protocol"] = IPProtocolEnum.UNKNOWN
+
+        # Common transports
+        if proto_num == 6 and TCP in pkt:
             pkt_info["protocol_name"] = "TCP"
-            pkt_info["src_port"] = pkt[TCP].sport
-            pkt_info["dst_port"] = pkt[TCP].dport
-        elif proto == 17 and UDP in pkt:
+            pkt_info["src_port"] = getattr(pkt[TCP], "sport", None)
+            pkt_info["dst_port"] = getattr(pkt[TCP], "dport", None)
+        elif proto_num == 17 and UDP in pkt:
             pkt_info["protocol_name"] = "UDP"
-            pkt_info["src_port"] = pkt[UDP].sport
-            pkt_info["dst_port"] = pkt[UDP].dport
-        elif proto == 1 and ICMP in pkt:
+            pkt_info["src_port"] = getattr(pkt[UDP], "sport", None)
+            pkt_info["dst_port"] = getattr(pkt[UDP], "dport", None)
+        elif proto_num == 1 and ICMP in pkt:
             pkt_info["protocol_name"] = "ICMP"
         else:
-            pkt_info["protocol_name"] = f"IP_PROTO_{proto}"
+            # leave protocol_name as numeric fallback if not matched
+            if pkt_info["protocol_name"] is None:
+                pkt_info["protocol_name"] = f"IP_PROTO_{proto_num}" if proto_num is not None else None
 
     # IPv6
     if IPv6 in pkt:
-        pkt_info["src_ip"] = pkt[IPv6].src
-        pkt_info["dst_ip"] = pkt[IPv6].dst
-        nh = pkt[IPv6].nh
+        try:
+            pkt_info["src_ip"] = pkt[IPv6].src
+        except Exception:
+            pkt_info["src_ip"] = None
+        try:
+            pkt_info["dst_ip"] = pkt[IPv6].dst
+        except Exception:
+            pkt_info["dst_ip"] = None
+
+        # next header / nh value
+        try:
+            nh = int(pkt[IPv6].nh)
+        except Exception:
+            nh = None
+
+        if nh is not None:
+            pkt_info["protocol_raw"] = nh
+            try:
+                pkt_info["protocol"] = protocol_from_number(nh)
+            except Exception:
+                pkt_info["protocol"] = IPProtocolEnum.UNKNOWN
+
         if nh == 6 and TCP in pkt:
             pkt_info["protocol_name"] = "TCP"
-            pkt_info["src_port"] = pkt[TCP].sport
-            pkt_info["dst_port"] = pkt[TCP].dport
+            pkt_info["src_port"] = getattr(pkt[TCP], "sport", None)
+            pkt_info["dst_port"] = getattr(pkt[TCP], "dport", None)
         elif nh == 17 and UDP in pkt:
             pkt_info["protocol_name"] = "UDP"
-            pkt_info["src_port"] = pkt[UDP].sport
-            pkt_info["dst_port"] = pkt[UDP].dport
+            pkt_info["src_port"] = getattr(pkt[UDP], "sport", None)
+            pkt_info["dst_port"] = getattr(pkt[UDP], "dport", None)
         elif ICMPv6Unknown in pkt:
             pkt_info["protocol_name"] = "ICMPv6"
         else:
-            pkt_info["protocol_name"] = f"IPV6_PROTO_{nh}"
+            if pkt_info["protocol_name"] is None:
+                pkt_info["protocol_name"] = f"IPV6_PROTO_{nh}" if nh is not None else None
 
     # Raw payload / fallback protocol label
     if Raw in pkt and not pkt_info["protocol_name"]:
         pkt_info["protocol_name"] = "RAW"
 
-    # Submit DB insert to background asyncio loop from this thread.
+    # Submit DB insert to background asyncio loop from this thread
     asyncio.run_coroutine_threadsafe(db_insert_packet(pkt_info, bridge), async_loop)
 
-
 # -------------------------
 # AF_PACKET optimized reader
 # -------------------------