links/routermon/parser.py

"""
Parse syslog datagrams from AsusWRT-Merlin's dnsmasq and kernel/iptables.

Handles RFC 3164 syslog envelope:
  Mmm DD HH:MM:SS hostname dnsmasq[PID]: message
  or with PRI prefix: <N>Mmm DD HH:MM:SS hostname dnsmasq[PID]: message

Recognised dnsmasq message formats:
  query[TYPE] domain from IP          — DNS query event
  reply domain is IP                  — resolved to an IP
  reply domain is NXDOMAIN            — domain does not exist
  reply domain is NODATA              — exists but no records of requested type
  DHCPACK(iface) IP MAC [hostname]    — DHCP lease acknowledged

Recognised kernel/iptables message formats:
  WAN_IN: IN=eth0 SRC=x.x.x.x DST=y.y.y.y ... PROTO=TCP SPT=N DPT=N ...
"""
import re
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

logger = logging.getLogger(__name__)

# ── Compiled regexes ──────────────────────────────────────────────────────────

# Strip optional RFC 3164 PRI prefix: <N>
_RE_PRI = re.compile(r'^<\d+>')

# Full syslog line: optional PRI + "Mmm DD HH:MM:SS hostname process[pid]: msg"
# We capture the process token and the message body.
_RE_SYSLOG = re.compile(
    r'(?:<\d+>)?'                        # optional PRI
    r'\w{3}\s+\d+\s+\d{2}:\d{2}:\d{2}'  # timestamp (Mmm DD HH:MM:SS)
    r'\s+\S+'                             # hostname
    r'\s+(dnsmasq(?:-dhcp)?|kernel)(?:\[\d+\])?:'  # process (group 1)
    r'\s*(.*)',                           # message body (group 2)
    re.DOTALL,
)

# dnsmasq DNS query: query[TYPE] domain from IP
_RE_QUERY = re.compile(r'^query\[(\w+)\]\s+([\w.\-]+)\s+from\s+([\d.a-fA-F:]+)')

# dnsmasq reply: reply domain is ANSWER
# ANSWER may be: an IP address, NXDOMAIN, NODATA, NODATA-IPv4, NODATA-IPv6, etc.
_RE_REPLY = re.compile(r'^reply\s+([\w.\-]+)\s+is\s+(\S+)')

# dnsmasq DHCP ack: DHCPACK(iface) IP MAC [hostname]
_RE_DHCP = re.compile(r'^DHCPACK\(\S+\)\s+([\d.]+)\s+([\da-fA-F:]+)\s*(\S*)')

# kernel: iptables log prefix marker
_RE_WAN_PREFIX = re.compile(r'WAN_IN:\s+')

# Simple IPv4/IPv6 check (enough to distinguish from NXDOMAIN/NODATA strings)
_RE_IP = re.compile(r'^[\d.a-fA-F:]+$')

# Private IP ranges to skip for geo (same check as nginxmon)
import ipaddress
_PRIVATE_NETS = (
    ipaddress.ip_network('10.0.0.0/8'),
    ipaddress.ip_network('172.16.0.0/12'),
    ipaddress.ip_network('192.168.0.0/16'),
    ipaddress.ip_network('127.0.0.0/8'),
    ipaddress.ip_network('::1/128'),
    ipaddress.ip_network('fc00::/7'),
)


def _is_public_ip(addr: str) -> bool:
    try:
        ip = ipaddress.ip_address(addr)
        return not any(ip in net for net in _PRIVATE_NETS)
    except ValueError:
        return False


def _parse_kv(text: str) -> dict:
    """Parse KEY=value pairs from an iptables log line."""
    return dict(re.findall(r'(\w+)=([^\s]+)', text))


# ── Result dataclasses ────────────────────────────────────────────────────────

@dataclass
class QueryEvent:
    client_ip: str
    domain: str
    query_type: str


@dataclass
class ReplyEvent:
    domain: str
    answer: str          # raw answer (IP, NXDOMAIN, NODATA, …)
    is_nxdomain: bool
    resolved_ip: Optional[str]  # public IP only, or None


@dataclass
class DhcpEvent:
    ip: str
    mac: str
    hostname: str


@dataclass
class WanEventParsed:
    src_ip: str
    protocol: str        # TCP / UDP / ICMP
    dst_port: Optional[int]
    src_port: Optional[int]


@dataclass
class ParseResult:
    query: Optional[QueryEvent] = None
    reply: Optional[ReplyEvent] = None
    dhcp: Optional[DhcpEvent] = None
    wan_event: Optional[WanEventParsed] = None


# ── Public API ────────────────────────────────────────────────────────────────

def parse_line(raw: str) -> Optional[ParseResult]:
    """
    Parse a single syslog datagram string.
    Returns a ParseResult if a recognised event was found, else None.
    """
    raw = raw.strip()
    if not raw:
        return None

    m = _RE_SYSLOG.match(raw)
    if not m:
        # Some Merlin builds omit the standard syslog prefix; try matching
        # dnsmasq message body directly.
        process = None
        body = raw
    else:
        process = m.group(1)
        body = m.group(2).strip()

    # ── Kernel / iptables WAN events ─────────────────────────────────────────
    if process == 'kernel' or (process is None and _RE_WAN_PREFIX.search(raw)):
        # Strip optional kernel timestamp "[12345.678] "
        body_stripped = re.sub(r'^\[\d+\.\d+\]\s*', '', body)
        if _RE_WAN_PREFIX.search(body_stripped):
            return _parse_wan_event(body_stripped)
        return None

    # ── dnsmasq events ───────────────────────────────────────────────────────

    # DNS query
    mq = _RE_QUERY.match(body)
    if mq:
        return ParseResult(query=QueryEvent(
            client_ip=mq.group(3),
            domain=mq.group(2).rstrip('.'),
            query_type=mq.group(1),
        ))

    # DNS reply
    mr = _RE_REPLY.match(body)
    if mr:
        domain = mr.group(1).rstrip('.')
        answer = mr.group(2)
        is_nxdomain = answer.upper() in ('NXDOMAIN', 'NODATA', 'NODATA-IPV4', 'NODATA-IPV6')
        resolved_ip = None
        if not is_nxdomain and _RE_IP.match(answer) and _is_public_ip(answer):
            resolved_ip = answer
        return ParseResult(reply=ReplyEvent(
            domain=domain,
            answer=answer,
            is_nxdomain=is_nxdomain,
            resolved_ip=resolved_ip,
        ))

    # DHCP ack
    md = _RE_DHCP.match(body)
    if md:
        return ParseResult(dhcp=DhcpEvent(
            ip=md.group(1),
            mac=md.group(2).lower(),
            hostname=md.group(3),
        ))

    return None


def _parse_wan_event(body: str) -> Optional[ParseResult]:
    """Parse an iptables WAN_IN log line body into a WanEventParsed."""
    kv = _parse_kv(body)
    src_ip = kv.get('SRC', '')
    if not src_ip or not _is_public_ip(src_ip):
        return None
    protocol = kv.get('PROTO', 'UNKNOWN').upper()
    try:
        dst_port = int(kv['DPT']) if 'DPT' in kv else None
    except (ValueError, KeyError):
        dst_port = None
    try:
        src_port = int(kv['SPT']) if 'SPT' in kv else None
    except (ValueError, KeyError):
        src_port = None
    return ParseResult(wan_event=WanEventParsed(
        src_ip=src_ip,
        protocol=protocol,
        dst_port=dst_port,
        src_port=src_port,
    ))