mi
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from jsonschema import validate
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuleResult:
|
|
deny_rule: str | None
|
|
monitor_rules: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CompiledRule:
|
|
rule_id: str
|
|
action: str
|
|
pattern: re.Pattern[str]
|
|
|
|
|
|
class RuleBundle:
|
|
def __init__(self, version: str, rules: tuple[CompiledRule, ...]) -> None:
|
|
self.version = version
|
|
self.rules = rules
|
|
|
|
@classmethod
|
|
def load(cls, bundle_path: Path, schema_path: Path) -> RuleBundle:
|
|
bundle = yaml.safe_load(bundle_path.read_text(encoding="utf-8"))
|
|
schema = yaml.safe_load(schema_path.read_text(encoding="utf-8"))
|
|
validate(bundle, schema)
|
|
compiled: list[CompiledRule] = []
|
|
ids: set[str] = set()
|
|
for rule in bundle["rules"]:
|
|
if rule["rule_id"] in ids:
|
|
raise ValueError("duplicate rule_id")
|
|
ids.add(rule["rule_id"])
|
|
pattern = re.compile(rule["pattern"], re.IGNORECASE)
|
|
compiled_rule = CompiledRule(rule["rule_id"], rule["action"], pattern)
|
|
for sample in rule["positive"]:
|
|
if not pattern.search(sample):
|
|
raise ValueError(f"positive vector failed: {rule['rule_id']}")
|
|
for sample in rule["negative"]:
|
|
if pattern.search(sample):
|
|
raise ValueError(f"negative vector failed: {rule['rule_id']}")
|
|
compiled.append(compiled_rule)
|
|
return cls(bundle["rules_version"], tuple(compiled))
|
|
|
|
def evaluate(self, text: str) -> RuleResult:
|
|
deny: list[str] = []
|
|
monitor: list[str] = []
|
|
for rule in self.rules:
|
|
if rule.pattern.search(text):
|
|
(deny if rule.action == "deny" else monitor).append(rule.rule_id)
|
|
return RuleResult(min(deny) if deny else None, tuple(sorted(monitor)))
|