parser.py

# SPDX-License-Identifier: GPL-2.0-only
# Copyright (C) 2026 Cyber Defense Institute, Inc.
"""
parser.py  --  SERINUS core logic (1): frame analysis

Role: receives raw Ethernet frames (bytes) and extracts the minimum information
      needed for detection: source MAC, protocol type, and estimated IP.

★ Inside the HAL boundary (platform-independent).
  This is a pure "raw bytes -> event" transformation, translatable almost verbatim
  to ESP32 (C) in the future. Must not depend on OS, scapy, or GUI.
"""

from dataclasses import dataclass
from typing import Optional


# Protocols targeted for detection (early-stage broadcast/multicast traffic)
PROTO_ARP = "ARP"
PROTO_DHCP = "DHCP"
PROTO_IPV6_ND = "IPv6-ND"
PROTO_MDNS = "mDNS"
PROTO_LLMNR = "LLMNR"
PROTO_OTHER = "OTHER"


@dataclass
class FrameEvent:
    """Parsed frame event. The only input format consumed by the core logic (engine)."""
    src_mac: str            # source MAC "aa:bb:cc:dd:ee:ff"
    proto: str              # one of the PROTO_* constants above
    ip_guess: Optional[str] # estimated IP if known, else None


def _mac(b: bytes) -> str:
    return ":".join(f"{x:02x}" for x in b)


def _ipv4(b: bytes) -> str:
    return ".".join(str(x) for x in b)


def parse_frame(raw: bytes) -> Optional[FrameEvent]:
    """
    Parse a raw Ethernet frame. Returns a FrameEvent only for frames worth detecting.
    Returns None for unparseable or non-target frames.
    """
    if raw is None or len(raw) < 14:
        return None

    # --- Ethernet header ---
    dst = raw[0:6]
    src = raw[6:12]
    ethertype = (raw[12] << 8) | raw[13]
    src_mac = _mac(src)

    # ignore broadcast or all-zero source (malformed frame)
    if src_mac in ("ff:ff:ff:ff:ff:ff", "00:00:00:00:00:00"):
        return None

    payload = raw[14:]

    # --- ARP (0x0806) ---
    if ethertype == 0x0806 and len(payload) >= 28:
        # htype ptype hlen plen oper(2) sha(6) spa(4) tha(6) tpa(4)
        # Use Ethernet src (harder to forge) instead of ARP SHA (freely writable)
        spa = payload[14:18]
        return FrameEvent(src_mac, PROTO_ARP, _ipv4(spa))

    # --- IPv4 (0x0800): capture DHCP / mDNS / LLMNR ---
    if ethertype == 0x0800 and len(payload) >= 20:
        ihl = (payload[0] & 0x0F) * 4
        proto = payload[9]
        src_ip = _ipv4(payload[12:16])
        if proto == 17 and len(payload) >= ihl + 8:  # UDP
            udp = payload[ihl:]
            sport = (udp[0] << 8) | udp[1]
            dport = (udp[2] << 8) | udp[3]
            if dport == 67 or sport == 68 or dport == 68:
                return FrameEvent(src_mac, PROTO_DHCP, src_ip)
            if dport == 5353 or sport == 5353:
                return FrameEvent(src_mac, PROTO_MDNS, src_ip)
            if dport == 5355 or sport == 5355:
                return FrameEvent(src_mac, PROTO_LLMNR, src_ip)
        return FrameEvent(src_mac, PROTO_OTHER, src_ip)

    # --- IPv6 (0x86DD): Neighbor Discovery etc. ---
    if ethertype == 0x86DD and len(payload) >= 40:
        nexthdr = payload[6]
        if nexthdr == 58:  # ICMPv6 (includes ND)
            return FrameEvent(src_mac, PROTO_IPV6_ND, None)
        return FrameEvent(src_mac, PROTO_OTHER, None)

    return None


# ----------------------------------------------------------------------
# Test/simulation helper: build a raw ARP frame without scapy.
# Not used on hardware (C). Lets the parser run through the same code path on PC.
# ----------------------------------------------------------------------
def build_arp_request(src_mac: str, src_ip: str,
                      target_ip: str = "0.0.0.0") -> bytes:
    def mac_b(s):
        return bytes(int(p, 16) for p in s.split(":"))

    def ip_b(s):
        return bytes(int(p) for p in s.split("."))

    eth = b"\xff\xff\xff\xff\xff\xff" + mac_b(src_mac) + b"\x08\x06"
    arp = (b"\x00\x01"          # htype = Ethernet
           b"\x08\x00"          # ptype = IPv4
           b"\x06"              # hlen
           b"\x04"              # plen
           b"\x00\x01")         # oper = request
    arp += mac_b(src_mac) + ip_b(src_ip)
    arp += b"\x00\x00\x00\x00\x00\x00" + ip_b(target_ip)
    return eth + arp