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)))