links/routermon/parser.py

"""
Parse syslog datagrams from AsusWRT-Merlin's dnsmasq.

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 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
"""
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)?)\[\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*)')

# 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


# ── 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 ParseResult:
    query: Optional[QueryEvent] = None
    reply: Optional[ReplyEvent] = None
    dhcp: Optional[DhcpEvent] = 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.
        body = raw
    else:
        body = m.group(2).strip()

    # 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