pftest.py

#!/usr/bin/env python3
"""
pf.conf packet flow simulator — traces packets through the full rule chain,
respecting quick, last-match-wins, anchor ordering, tags, and NAT.
"""

import platform
import re
import shutil
import subprocess
import sys
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import Optional

BASE = Path(__file__).parent
MAIN = BASE / "pf.conf"
PFD = BASE / "pf.d"


# ── data model ───────────────────────────────────────────────────────────────

@dataclass
class Packet:
    """A simulated packet."""
    src: str
    dst: str
    proto: str          # tcp, udp, icmp
    sport: int = 0      # source port
    dport: int = 0      # dest port
    iface: str = ""     # ingress or egress interface
    direction: str = "" # in or out
    tag: str = ""       # assigned tag
    icmp_type: str = "" # echoreq, unreach, timex


@dataclass
class Rule:
    """A parsed pf rule."""
    action: str         # pass, block, match
    direction: str      # in, out, ""
    quick: bool = False
    iface: str = ""
    af: str = "inet"    # inet, inet6, ""
    proto: str = ""     # tcp, udp, icmp, ""
    src: str = ""       # source address/network or "any"
    dst: str = ""       # dest address/network or "any"
    sport: str = ""     # source port or range
    dport: str = ""     # dest port, range, or list
    tag_assign: str = ""  # tag X
    tag_match: str = ""   # tagged X
    nat_to: str = ""
    rdr_to: str = ""
    rdr_to_port: str = ""  # rdr-to port rewrite
    route_to: str = ""
    flags: str = ""
    icmp_type: str = ""
    keep_state: bool = False
    log: bool = False
    raw: str = ""       # original text
    source_file: str = ""
    line_num: int = 0
    anchor: str = ""    # which anchor this belongs to
    max_src_conn: int = 0  # state option: max-src-conn N


@dataclass
class FlowResult:
    action: str             # pass, block
    rule: Optional[Rule]    # matching rule
    tag: str = ""           # assigned tag
    nat_applied: bool = False
    route_to: str = ""
    rdr_applied: bool = False
    rdr_dst: str = ""
    rdr_dport: str = ""
    trace: list = field(default_factory=list)


# ── network helpers ──────────────────────────────────────────────────────────

def ip_to_int(ip):
    parts = ip.split(".")
    return (int(parts[0]) << 24) + (int(parts[1]) << 16) + (int(parts[2]) << 8) + int(parts[3])


def ip_in_net(ip, network):
    """Check if IP is in a CIDR network."""
    if network == "any" or network == "":
        return True
    if "/" not in network:
        return ip == network
    net, bits = network.split("/")
    bits = int(bits)
    mask = (0xFFFFFFFF << (32 - bits)) & 0xFFFFFFFF
    return (ip_to_int(ip) & mask) == (ip_to_int(net) & mask)


def port_in_spec(port, spec):
    """Check if port matches a port specification (number, range, or list)."""
    if not spec or spec == "" or spec == "any":
        return True
    if port == 0:
        return True

    spec = spec.strip()
    if spec.startswith("{") and spec.endswith("}"):
        inner = spec[1:-1].strip()
        parts = re.split(r'[,\s]+', inner)
        return any(port_in_spec(port, p.strip()) for p in parts if p.strip())

    if ":" in spec:
        lo, hi = spec.split(":", 1)
        return int(lo) <= port <= int(hi)

    return port == int(spec)


# ── parser helpers ───────────────────────────────────────────────────────────

def join_continuation_lines(text):
    """Join backslash-continued lines, strip comments."""
    lines = []
    buf = ""
    for raw in text.splitlines():
        stripped = raw.split("#")[0].rstrip()
        if stripped.endswith("\\"):
            buf += stripped[:-1].rstrip() + " "
            continue
        buf += stripped
        if buf.strip():
            lines.append(buf.strip())
        buf = ""
    if buf.strip():
        lines.append(buf.strip())
    return lines


def extract_iface(tokens, idx):
    """Extract interface after 'on' keyword."""
    if idx < len(tokens):
        val = tokens[idx]
        if val == "{":
            end = tokens.index("}", idx)
            return tokens[idx+1:end], end + 1
        return [val], idx + 1
    return [], idx


def extract_addr_port(tokens, idx):
    """Extract address and optional port from token stream."""
    addr = "any"
    port = ""

    if idx >= len(tokens):
        return addr, port, idx

    if tokens[idx] == "!":
        # negation is baked into the address string for addr_matches()
        if idx + 1 < len(tokens):
            addr = "!" + tokens[idx + 1]
            idx += 2
        else:
            idx += 1
            return addr, port, idx
    elif tokens[idx] == "{":
        end_idx = idx + 1
        depth = 1
        while end_idx < len(tokens) and depth > 0:
            if tokens[end_idx] == "{":
                depth += 1
            elif tokens[end_idx] == "}":
                depth -= 1
            end_idx += 1
        addr = " ".join(tokens[idx:end_idx])
        idx = end_idx
    elif tokens[idx] not in ("port", "flags", "keep", "tag", "tagged", "rdr-to",
                              "nat-to", "route-to", "icmp-type", "scrub", "log",
                              "quick", "proto", "on", "inet", "inet6"):
        addr = tokens[idx]
        idx += 1

    # check for port
    if idx < len(tokens) and tokens[idx] == "port":
        idx += 1
        if idx < len(tokens):
            if tokens[idx] == "{":
                end_idx = idx + 1
                depth = 1
                while end_idx < len(tokens) and depth > 0:
                    if tokens[end_idx] == "{":
                        depth += 1
                    elif tokens[end_idx] == "}":
                        depth -= 1
                    end_idx += 1
                port = " ".join(tokens[idx:end_idx])
                idx = end_idx
            else:
                port = tokens[idx]
                idx += 1

    return addr, port, idx


# ── PFSimulator ──────────────────────────────────────────────────────────────

class PFSimulator:
    """Stateful PF ruleset simulator.

    Encapsulates all parsed state (macros, tables, topology, rules)
    so multiple configs can be loaded in the same process without
    cross-contamination.
    """

    def __init__(self, pf_conf=MAIN, pf_d=PFD):
        self.pf_conf = Path(pf_conf)
        self.pf_d = Path(pf_d)
        self.macros = {}
        self.tables = {}
        self.iface_networks = {}
        self.iface_groups = {}
        self.rules = []
        self._state_table = None

    # ── macro/table parsing ──────────────────────────────────────

    def _resolve_macro(self, val):
        """Resolve $macro references."""
        while "$" in val:
            changed = False
            for k, v in self.macros.items():
                if f"${k}" in val:
                    val = val.replace(f"${k}", v)
                    changed = True
            if not changed:
                break
        return val

    def _parse_macros(self, lines):
        self.macros.clear()
        for line in lines:
            m = re.match(r'^(\w+)\s*=\s*"([^"]*)"', line)
            if m:
                self.macros[m.group(1)] = m.group(2)

    def _parse_tables(self, lines):
        self.tables.clear()
        for line in lines:
            m = re.match(r'table\s+<(\w+)>\s+(?:const|persist)\s*(?:\{([^}]*)\})?', line)
            if m:
                name = m.group(1)
                members = []
                if m.group(2):
                    for part in re.split(r'[,\s]+', m.group(2).strip()):
                        part = part.strip()
                        if part:
                            members.append(self._resolve_macro(part))
                self.tables[name] = members

    # ── topology detection ───────────────────────────────────────

    def _auto_detect_topology(self, main_lines):
        """Auto-detect interface-to-network mapping from pf.conf.

        Three-phase detection:
          1. Pair *_if macros with *_net macros by shared prefix
          2. Detect WAN interface from 'wan' macro or NAT rules
          3. Scan match-tag rules for remaining interface->network bindings
        """
        iface_networks = {}
        wan_iface = None

        # Phase 1: pair *_if with *_net macros by prefix
        if_macros = {}
        net_macros = {}
        for name, val in self.macros.items():
            resolved = self._resolve_macro(val)
            if name.endswith("_if"):
                if_macros[name[:-3]] = resolved
            elif name.endswith("_net"):
                net_macros[name[:-4]] = resolved

        for prefix in if_macros:
            if prefix in net_macros:
                iface_networks[if_macros[prefix]] = net_macros[prefix]

        # Phase 2: detect WAN
        if "wan" in self.macros:
            wan_iface = self._resolve_macro(self.macros["wan"])
        else:
            for line in main_lines:
                if "nat-to" in line and line.startswith("match"):
                    m = re.search(r'on\s+(\S+)', line)
                    if m:
                        candidate = self._resolve_macro(m.group(1))
                        if candidate not in iface_networks:
                            wan_iface = candidate
                            break

        if wan_iface:
            iface_networks[wan_iface] = None

        # Phase 3: scan match-tag rules for remaining pairs
        for line in main_lines:
            m = re.match(
                r'match\s+in\s+(?:log\s+)?on\s+(\S+)\s+inet\s+from\s+(\S+)\s+',
                line,
            )
            if m:
                iface_raw = self._resolve_macro(m.group(1))
                net_raw = self._resolve_macro(m.group(2))
                if "/" in net_raw and iface_raw not in iface_networks:
                    iface_networks[iface_raw] = net_raw

        self.iface_networks = iface_networks
        self.iface_groups = {"egress": wan_iface} if wan_iface else {}

    # ── antispoof ────────────────────────────────────────────────

    def _parse_antispoof(self, line):
        """Parse antispoof line and return (interfaces, quick)."""
        # handle: antispoof [log] [quick] for { iface1 iface2 }
        m = re.match(r'antispoof\s+(?:log\s+)?(?:(quick)\s+)?for\s+\{([^}]+)\}', line)
        if m:
            quick = m.group(1) is not None
            ifaces = re.split(r'[,\s]+', m.group(2).strip())
            return [self._resolve_macro(i.strip()) for i in ifaces if i.strip()], quick
        m = re.match(r'antispoof\s+(?:log\s+)?(?:(quick)\s+)?for\s+(\S+)', line)
        if m:
            quick = m.group(1) is not None
            return [self._resolve_macro(m.group(2))], quick
        return [], False

    def _expand_antispoof(self, ifaces, quick, anchor="main"):
        """Expand antispoof into the block rules PF generates.

        For each interface X with network N, PF creates:
          block drop in [quick] on X from !N to any
          block drop in [quick] on !X from N to any
        """
        rules = []
        for iface in ifaces:
            net = self.iface_networks.get(iface)
            if not net:
                continue

            r1 = Rule(
                action="block", direction="in", quick=quick,
                iface=iface, af="inet",
                src=f"!{net}", dst="any",
                raw=f"[antispoof] block drop in{' quick' if quick else ''} on {iface} from !{net} to any",
                source_file="pf.conf", anchor=anchor,
            )
            rules.append(r1)

            for other_iface in self.iface_networks:
                if other_iface == iface:
                    continue
                r2 = Rule(
                    action="block", direction="in", quick=quick,
                    iface=other_iface, af="inet",
                    src=net, dst="any",
                    raw=f"[antispoof] block drop in{' quick' if quick else ''} on {other_iface} from {net} to any",
                    source_file="pf.conf", anchor=anchor,
                )
                rules.append(r2)

        return rules

    # ── rule parsing ─────────────────────────────────────────────

    def _parse_rule(self, line, source_file="", line_num=0, anchor=""):
        """Parse a single pf rule line into a Rule object."""
        r = Rule(action="", direction="", raw=line, source_file=source_file,
                 line_num=line_num, anchor=anchor)

        tokens = re.findall(r'\{|\}|[^\s{}]+', line)
        if not tokens:
            return None

        idx = 0
        if tokens[idx] in ("pass", "block", "match"):
            r.action = tokens[idx]
            idx += 1
        else:
            return None

        # resolve macros in tokens
        tokens = [self._resolve_macro(t) if t.startswith("$") or t.startswith("!$")
                  else t for t in tokens]

        while idx < len(tokens):
            tok = tokens[idx]

            if tok in ("in", "out"):
                r.direction = tok
                idx += 1
            elif tok == "log":
                r.log = True
                idx += 1
                if idx < len(tokens) and tokens[idx] == "(all)":
                    idx += 1
            elif tok == "quick":
                r.quick = True
                idx += 1
            elif tok == "on":
                idx += 1
                ifaces, idx = extract_iface(tokens, idx)
                r.iface = ",".join(ifaces) if ifaces else ""
            elif tok in ("inet", "inet6"):
                r.af = tok
                idx += 1
            elif tok == "proto":
                idx += 1
                if idx < len(tokens):
                    if tokens[idx] == "{":
                        end = tokens.index("}", idx)
                        protos = [t.strip(",") for t in tokens[idx+1:end]]
                        r.proto = ",".join(protos)
                        idx = end + 1
                    else:
                        r.proto = tokens[idx]
                        idx += 1
            elif tok == "from":
                idx += 1
                r.src, r.sport, idx = extract_addr_port(tokens, idx)
                if idx < len(tokens) and tokens[idx] == "to":
                    idx += 1
                    r.dst, r.dport, idx = extract_addr_port(tokens, idx)
            elif tok == "to":
                idx += 1
                r.dst, r.dport, idx = extract_addr_port(tokens, idx)
            elif tok == "tag":
                idx += 1
                if idx < len(tokens):
                    r.tag_assign = tokens[idx]
                    idx += 1
            elif tok == "tagged":
                idx += 1
                if idx < len(tokens):
                    r.tag_match = tokens[idx]
                    idx += 1
            elif tok == "nat-to":
                idx += 1
                if idx < len(tokens):
                    r.nat_to = tokens[idx]
                    idx += 1
            elif tok == "rdr-to":
                idx += 1
                if idx < len(tokens):
                    r.rdr_to = tokens[idx]
                    idx += 1
                    if idx < len(tokens) and tokens[idx] == "port":
                        idx += 1
                        if idx < len(tokens):
                            r.rdr_to_port = tokens[idx]
                            idx += 1
            elif tok == "route-to":
                idx += 1
                if idx < len(tokens):
                    r.route_to = tokens[idx]
                    idx += 1
            elif tok == "flags":
                idx += 1
                if idx < len(tokens):
                    r.flags = tokens[idx]
                    idx += 1
            elif tok == "icmp-type":
                idx += 1
                if idx < len(tokens):
                    if tokens[idx] == "{":
                        end = tokens.index("}", idx)
                        r.icmp_type = ",".join(tokens[idx+1:end])
                        idx = end + 1
                    else:
                        r.icmp_type = tokens[idx]
                        idx += 1
            elif tok == "keep":
                r.keep_state = True
                idx += 1
                if idx < len(tokens) and tokens[idx] == "state":
                    idx += 1
            elif tok == "drop":
                idx += 1
            elif tok == "scrub":
                idx += 1
                if idx < len(tokens) and tokens[idx] == "(":
                    while idx < len(tokens) and tokens[idx] != ")":
                        idx += 1
                    idx += 1
            elif tok.startswith("("):
                # state options — parse max-src-conn
                state_tokens = [tok]
                while idx < len(tokens) and not tokens[idx].endswith(")"):
                    idx += 1
                    if idx < len(tokens):
                        state_tokens.append(tokens[idx])
                idx += 1
                state_str = " ".join(state_tokens)
                m_conn = re.search(r'max-src-conn\s+(\d+)', state_str)
                if m_conn:
                    r.max_src_conn = int(m_conn.group(1))
            else:
                idx += 1

        return r

    # ── loading ────────────────────────────────────────────���─────

    def syntax_check(self):
        """Run pfctl -nf to syntax-check the config.

        Works on any OS with pfctl (OpenBSD, macOS, FreeBSD).

        Returns (ok, output):
          - (True, "")           on success
          - (False, error_msg)   on pfctl syntax error
          - (None, reason)       if pfctl is unavailable or can't run
        """
        pfctl = shutil.which("pfctl")
        if not pfctl:
            return None, "pfctl not found in PATH"

        try:
            result = subprocess.run(
                [pfctl, "-nf", str(self.pf_conf)],
                capture_output=True, text=True, timeout=10,
            )
            if result.returncode == 0:
                return True, ""
            return False, result.stderr.strip()
        except PermissionError:
            return None, "pfctl requires root — run with doas/sudo to syntax-check"
        except subprocess.TimeoutExpired:
            return None, "pfctl timed out"
        except OSError as e:
            return None, f"pfctl failed: {e}"

    def load(self, syntax_check=True):
        """Load main config and all anchors in evaluation order.

        Sequence:
          0. (OpenBSD only) Run pfctl -nf to syntax-check
          1. Parse macros and tables
          2. Auto-detect network topology
          3. Parse rules, expanding antispoof
          4. Parse anchor files in declared order
        """
        if syntax_check:
            ok, msg = self.syntax_check()
            if ok is False:
                print(f"pfctl syntax error:\n{msg}", file=sys.stderr)
                raise SystemExit(1)
            elif ok is None and msg:
                pass  # silently skip — not OpenBSD or not root
            # ok is True — config is valid, continue
        main_text = self.pf_conf.read_text()
        main_lines = join_continuation_lines(main_text)

        self._parse_macros(main_lines)
        self._parse_tables(main_lines)
        self._auto_detect_topology(main_lines)

        self.rules = []
        anchor_order = []

        for i, line in enumerate(main_lines):
            if line.startswith("anchor") or line.startswith("load anchor"):
                m = re.search(r'"(\w+)"', line)
                if m and m.group(1) not in anchor_order:
                    anchor_order.append(m.group(1))
                continue

            if line.startswith("antispoof"):
                ifaces, quick = self._parse_antispoof(line)
                if ifaces:
                    self.rules.extend(self._expand_antispoof(ifaces, quick))
                continue

            r = self._parse_rule(line, source_file="pf.conf", line_num=i, anchor="main")
            if r:
                self.rules.append(r)

        for anchor_name in anchor_order:
            conf = self.pf_d / f"{anchor_name}.conf"
            if conf.exists():
                anchor_lines = join_continuation_lines(conf.read_text())
                for i, line in enumerate(anchor_lines):
                    r = self._parse_rule(line, source_file=f"{anchor_name}.conf",
                                         line_num=i, anchor=anchor_name)
                    if r:
                        self.rules.append(r)

        return self.rules

    # ── matching ─────────────────────────────────────────────────

    def _addr_matches(self, pkt_addr, rule_addr):
        if not rule_addr or rule_addr == "any":
            return True

        rule_addr = self._resolve_macro(rule_addr)

        if rule_addr.startswith("!"):
            return not self._addr_matches(pkt_addr, rule_addr[1:])

        # (iface) self-address — approximation: matches router's own traffic.
        # Real PF matches only the IP assigned to that interface.
        if rule_addr.startswith("(") and rule_addr.endswith(")"):
            return True

        if rule_addr.startswith("{") and rule_addr.endswith("}"):
            inner = rule_addr[1:-1].strip()
            parts = re.split(r'[,\s]+', inner)
            return any(self._addr_matches(pkt_addr, self._resolve_macro(p.strip()))
                       for p in parts if p.strip())

        m = re.match(r'<(\w+)>', rule_addr)
        if m:
            table_name = m.group(1)
            if table_name in self.tables:
                return any(ip_in_net(pkt_addr, net) for net in self.tables[table_name])
            return False

        return ip_in_net(pkt_addr, rule_addr)

    def _resolve_addr_to_ip(self, addr):
        """Resolve a table/macro reference to a concrete IP (first member)."""
        addr = self._resolve_macro(addr)
        m = re.match(r'<(\w+)>', addr)
        if m:
            table_name = m.group(1)
            if table_name in self.tables and self.tables[table_name]:
                return self.tables[table_name][0]
        return addr

    def _iface_matches(self, pkt_iface, rule_iface):
        if not rule_iface:
            return True
        ifaces = [i.strip() for i in rule_iface.split(",")]
        resolved = [self.iface_groups.get(i, i) for i in ifaces]
        return pkt_iface in resolved

    def _rule_matches(self, pkt, rule):
        if rule.direction and rule.direction != pkt.direction:
            return False
        if not self._iface_matches(pkt.iface, rule.iface):
            return False
        if rule.af == "inet6":
            return False
        if rule.proto and pkt.proto not in rule.proto.split(","):
            return False
        if not self._addr_matches(pkt.src, rule.src):
            return False
        if not self._addr_matches(pkt.dst, rule.dst):
            return False
        if rule.sport and not port_in_spec(pkt.sport, rule.sport):
            return False
        if rule.dport and not port_in_spec(pkt.dport, rule.dport):
            return False
        if rule.tag_match and rule.tag_match != pkt.tag:
            return False
        if rule.icmp_type and pkt.proto == "icmp":
            types = [t.strip() for t in rule.icmp_type.split(",")]
            if pkt.icmp_type not in types and pkt.icmp_type:
                return False
        return True

    # ── state table ──────────────────────────────────────────────

    def enable_state_tracking(self):
        """Enable stateful evaluation with max-src-conn enforcement."""
        self._state_table = {}
        return self._state_table

    def disable_state_tracking(self):
        self._state_table = None

    def _check_state_limit(self, src_ip, rule):
        """Returns True if connection allowed, False if rate-limited."""
        if self._state_table is None or rule.max_src_conn <= 0:
            return True
        key = (src_ip, rule.source_file, rule.line_num, rule.anchor)
        current = self._state_table.get(key, 0)
        if current >= rule.max_src_conn:
            return False
        self._state_table[key] = current + 1
        return True

    # ── evaluation ───────────────────────────────────────────────

    def evaluate(self, pkt, rules=None):
        """Evaluate packet against ruleset. Returns FlowResult."""
        if rules is None:
            rules = self.rules

        result = FlowResult(action="block", rule=None)
        pkt = replace(pkt)  # shallow copy

        for rule in rules:
            if not self._rule_matches(pkt, rule):
                continue

            # match rules: transformations only
            if rule.action == "match":
                if rule.tag_assign:
                    pkt = replace(pkt, tag=rule.tag_assign)
                    result.tag = rule.tag_assign
                    result.trace.append((rule, f"tag={rule.tag_assign}"))
                if rule.nat_to:
                    result.nat_applied = True
                    result.trace.append((rule, f"nat-to={rule.nat_to}"))
                if rule.rdr_to:
                    resolved = self._resolve_addr_to_ip(rule.rdr_to)
                    result.rdr_applied = True
                    result.rdr_dst = resolved
                    result.rdr_dport = rule.rdr_to_port
                    pkt = replace(pkt, dst=resolved)
                    if rule.rdr_to_port:
                        pkt = replace(pkt, dport=int(rule.rdr_to_port))
                    result.trace.append((rule, f"rdr-to={resolved}" +
                        (f" port {rule.rdr_to_port}" if rule.rdr_to_port else "")))
                continue

            # pass/block rules with rdr-to
            if rule.rdr_to:
                resolved = self._resolve_addr_to_ip(rule.rdr_to)
                result.rdr_applied = True
                result.rdr_dst = resolved
                result.rdr_dport = rule.rdr_to_port
                pkt = replace(pkt, dst=resolved)
                if rule.rdr_to_port:
                    pkt = replace(pkt, dport=int(rule.rdr_to_port))
                result.trace.append((rule, f"rdr-to={resolved}" +
                    (f" port {rule.rdr_to_port}" if rule.rdr_to_port else "")))

            if rule.quick:
                if rule.action == "pass" and not self._check_state_limit(pkt.src, rule):
                    result.action = "block"
                    result.rule = rule
                    result.trace.append((rule, f"QUICK pass -> RATE LIMITED (max-src-conn {rule.max_src_conn})"))
                    return result
                result.action = rule.action
                result.rule = rule
                if rule.route_to:
                    result.route_to = rule.route_to
                result.trace.append((rule, f"QUICK {rule.action}"))
                return result

            # last-match-wins
            result.action = rule.action
            result.rule = rule
            if rule.route_to:
                result.route_to = rule.route_to
            result.trace.append((rule, f"match {rule.action} (last-match)"))

        # state check for final last-match-wins rule
        if result.action == "pass" and result.rule:
            if not self._check_state_limit(pkt.src, result.rule):
                result.action = "block"
                result.trace.append((result.rule,
                    f"RATE LIMITED (max-src-conn {result.rule.max_src_conn})"))

        return result


# ── module-level API (backwards compatibility) ───────────────────────────────
#
# All existing consumers use:
#   from pftest import Packet, load_all_rules, evaluate, MACROS, TABLES
#
# These delegate to a default PFSimulator instance.

_default = PFSimulator()

# mutable dicts — updated in place by load_all_rules()
MACROS = _default.macros
TABLES = _default.tables
IFACE_NETWORKS = _default.iface_networks
IFACE_GROUPS = _default.iface_groups


def load_all_rules():
    """Load rules from pf.conf + pf.d/ using the default simulator."""
    rules = _default.load()
    # update module-level references (for code that imported them at load time)
    global MACROS, TABLES, IFACE_NETWORKS, IFACE_GROUPS
    MACROS = _default.macros
    TABLES = _default.tables
    IFACE_NETWORKS = _default.iface_networks
    IFACE_GROUPS = _default.iface_groups
    return rules


def evaluate(pkt, rules):
    """Evaluate a packet using the default simulator."""
    return _default.evaluate(pkt, rules)


def enable_state_tracking():
    return _default.enable_state_tracking()


def disable_state_tracking():
    _default.disable_state_tracking()