mirror of
https://github.com/Av0idD4rk/packmate.git
synced 2026-10-23 14:31:01 +00:00
127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
import ipaddress
|
|
from functools import lru_cache
|
|
from typing import Literal
|
|
|
|
import re2
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
class StrictModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
@lru_cache(maxsize=128)
|
|
def compile_regex(value: str):
|
|
options = re2.Options()
|
|
options.log_errors = False
|
|
options.max_mem = 1024 * 1024
|
|
try:
|
|
return re2.compile(value, options=options)
|
|
except re2.error as error:
|
|
raise ValueError(f"INVALID_REGEX: RE2 syntax required: {error}") from None
|
|
|
|
|
|
class AddressFilter(StrictModel):
|
|
type: Literal["exact", "cidr", "regex"] = "exact"
|
|
value: str = Field(min_length=1, max_length=256)
|
|
|
|
@model_validator(mode="after")
|
|
def check(self):
|
|
if self.type == "regex":
|
|
compile_regex(self.value)
|
|
elif self.type == "cidr":
|
|
ipaddress.ip_network(self.value)
|
|
else:
|
|
ipaddress.ip_address(self.value)
|
|
return self
|
|
|
|
|
|
class Match(StrictModel):
|
|
target: Literal["content", "source_ip", "destination_ip", "either_ip"] = "content"
|
|
type: Literal["substring", "regex", "hex"] = "regex"
|
|
value: str = Field(min_length=1, max_length=4096)
|
|
|
|
@model_validator(mode="after")
|
|
def check(self):
|
|
if self.target != "content" and self.type != "regex":
|
|
raise ValueError("IP patterns require regex")
|
|
if self.type == "regex":
|
|
compile_regex(self.value)
|
|
if self.type == "hex":
|
|
# Same representation as Packmate SUBBYTES: no separators, even length.
|
|
if len(self.value) % 2 or any(c not in "0123456789abcdefABCDEF" for c in self.value):
|
|
raise ValueError("hex must contain an even number of hex digits without separators")
|
|
return self
|
|
|
|
|
|
class ScanCursor(StrictModel):
|
|
stream_timestamp: int = Field(ge=0)
|
|
stream_id: int = Field(gt=0)
|
|
packet_id: int = Field(gt=0)
|
|
|
|
|
|
class TrafficQuery(StrictModel):
|
|
"""Time filters refer to the start of a saved stream, not packet arrival time."""
|
|
|
|
last_minutes: int = Field(default=5, ge=1, le=1440)
|
|
from_ms: int | None = Field(default=None, ge=0)
|
|
to_ms: int | None = Field(default=None, ge=0)
|
|
snapshot_ms: int | None = Field(default=None, gt=0)
|
|
cursor: ScanCursor | None = None
|
|
service_ports: list[int] = Field(default_factory=list, max_length=100)
|
|
protocol: Literal["TCP", "UDP"] | None = None
|
|
direction: Literal["incoming", "outgoing"] | None = None
|
|
favorites_only: bool = False
|
|
pattern_id: int | None = Field(default=None, gt=0)
|
|
source_ip: AddressFilter | None = None
|
|
destination_ip: AddressFilter | None = None
|
|
client_ip: AddressFilter | None = None
|
|
match: Match | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def check(self):
|
|
if any(p < 1 or p > 65535 for p in self.service_ports):
|
|
raise ValueError("service_ports must be 1..65535")
|
|
if (self.from_ms is None) != (self.to_ms is None):
|
|
raise ValueError("Specify both from_ms and to_ms, or neither")
|
|
if self.from_ms is not None and not 0 < self.to_ms - self.from_ms <= 86400000:
|
|
raise ValueError("Time window must be positive and at most 24 hours")
|
|
if self.cursor and (self.from_ms is None or self.snapshot_ms is None):
|
|
raise ValueError("Resuming requires the original from_ms, to_ms and snapshot_ms")
|
|
return self
|
|
|
|
|
|
class ServiceSettings(StrictModel):
|
|
name: str = Field(min_length=1, max_length=100)
|
|
http: bool = True
|
|
urldecode_http_requests: bool = True
|
|
merge_adjacent_packets: bool = True
|
|
parse_web_sockets: bool = False
|
|
decrypt_tls: bool = False
|
|
|
|
def backend(self):
|
|
return {"name": self.name, "http": self.http, "urldecodeHttpRequests": self.urldecode_http_requests,
|
|
"mergeAdjacentPackets": self.merge_adjacent_packets, "parseWebSockets": self.parse_web_sockets,
|
|
"decryptTls": self.decrypt_tls}
|
|
|
|
|
|
class PatternSettings(StrictModel):
|
|
name: str = Field(min_length=1, max_length=100)
|
|
match: Match
|
|
action: Literal["FIND", "IGNORE"] = "FIND"
|
|
direction: Literal["INPUT", "OUTPUT", "BOTH"] = "BOTH"
|
|
service_port: int | None = Field(default=None, ge=1, le=65535)
|
|
color: str = Field(default="#ff9800", pattern=r"^#[0-9a-fA-F]{6}$")
|
|
|
|
@model_validator(mode="after")
|
|
def stored_value_limit(self):
|
|
if len(self.match.value) > 255:
|
|
raise ValueError("Stored patterns support up to 255 characters; ad-hoc searches support 4096")
|
|
return self
|
|
|
|
def backend(self):
|
|
return {"name": self.name, "value": self.match.value, "color": self.color,
|
|
"searchType": {"substring": "SUBSTRING", "regex": "REGEX", "hex": "SUBBYTES"}[self.match.type],
|
|
"searchTarget": self.match.target.upper(), "regexEngine": "RE2J",
|
|
"directionType": self.direction, "actionType": self.action, "serviceId": self.service_port}
|