From df0ceffad96b6a726da02a2f2dd2aede2aaaa5d2 Mon Sep 17 00:00:00 2001 From: opencode Date: Sat, 13 Jun 2026 10:17:51 +0000 Subject: [PATCH] Add DRL ingestion script and IngestReport/IngestedRule models --- scripts/ingest.py | 320 ++++++++++++++++++++++++++++ src/iriusrisk_drl_toolkit/models.py | 34 +++ 2 files changed, 354 insertions(+) create mode 100644 scripts/ingest.py diff --git a/scripts/ingest.py b/scripts/ingest.py new file mode 100644 index 0000000..7a754d2 --- /dev/null +++ b/scripts/ingest.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +""" +Ingest one or more IriusRisk Drools (.drl) files and produce a structured +catalog of every rule, classified by how well it maps to our supported subset. + +Usage: + python scripts/ingest.py path/to/rules.drl + python scripts/ingest.py path/to/drl/directory/ + python scripts/ingest.py path/to/rules.drl --json > catalog.json +""" + +import argparse +import json +import re +import sys +from dataclasses import asdict +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from iriusrisk_drl_toolkit.models import IngestedRule, IngestReport + +PACKAGE_RE = re.compile(r"^\s*package\s+([A-Za-z0-9_.]+)\s*$", re.MULTILINE) +IMPORT_RE = re.compile(r"^\s*import\s+([A-Za-z0-9_.*]+)\s*$", re.MULTILINE) + +RULE_RE = re.compile( + r""" + rule\s+"(?P[^"]+)"\s* + (?P.*?) + when\s + (?P.*?) + then\s + (?P.*?) + end + """, + re.DOTALL | re.VERBOSE, +) + +SALIENCE_RE = re.compile(r"\bsalience\s+(-?\d+)\b") +NO_LOOP_RE = re.compile(r"\bno-loop\s+true\b") + +PATTERN_RE = re.compile( + r""" + \$\w+\s*:\s* + (\w+)\s*\( + (.*?) + \) + """, + re.DOTALL | re.VERBOSE, +) + +RELATION_TYPE_RE = re.compile(r'relationType\s+==\s*"([^"]+)"') +COMPONENT_TYPE_RE = re.compile(r'\btype\s+==\s*"([^"]+)"') + +ACTION_RE = re.compile( + r""" + (\w+(?:\.\w+)*)\s* + \.\s* + (\w+)\s*\( + (.*?) + \)\s*; + """, + re.DOTALL | re.VERBOSE, +) + +ACTIVITY_KINDS = frozenset({ + "CREATE_THREAT", + "CREATE_COUNTERMEASURE", + "AUTO_RESOLVE", + "CREATE_RISK_PATTERN", +}) + + +def split_drl(file_text: str) -> tuple[str | None, list[str]]: + package = PACKAGE_RE.search(file_text) + package_name = package.group(1) if package else None + imports = IMPORT_RE.findall(file_text) + return package_name, imports + + +def parse_rules(file_text: str) -> list[dict]: + return [m.groupdict() for m in RULE_RE.finditer(file_text)] + + +def parse_patterns(when_block: str) -> list[dict]: + patterns = [] + for m in PATTERN_RE.finditer(when_block): + patterns.append({ + "type": m.group(1), + "body": m.group(2).strip(), + }) + return patterns + + +def parse_actions(then_block: str) -> list[dict]: + actions = [] + for m in ACTION_RE.finditer(then_block): + actions.append({ + "object": m.group(1), + "method": m.group(2), + "args": m.group(3).strip(), + }) + return actions + + +def classify_rule( + name: str, + when_block: str, + then_block: str, + patterns: list[dict], + actions: list[dict], +) -> tuple[str, str]: + pattern_types = [p["type"] for p in patterns] + has_relation = "Relation" in pattern_types + has_component = "Component" in pattern_types + + if has_relation and has_component: + rel_match = RELATION_TYPE_RE.search(when_block) + activity_kind: str | None = None + for a in actions: + if a["method"] == "createActivity": + args = a["args"] + m = re.search(r'"([^"]+)"', args) + if m and m.group(1) in ACTIVITY_KINDS: + activity_kind = m.group(1) + + if rel_match and activity_kind: + return ( + "relation_rule", + f"relation_type={rel_match.group(1)!r}, activity={activity_kind!r}", + ) + return ("relation_rule", "relation+component pattern (unrecognised action)") + + if "eval(" in when_block: + return ( + "unsupported", + "contains eval() — unsupported dynamic construct", + ) + + if has_component: + return ("component_rule", "component-only pattern (no relation)") + + return ( + "unsupported", + f"patterns: {pattern_types} — no recognised Relation+Component combo", + ) + + +def ingest_drl( + file_path: Path, + *, + parse_body: bool = True, +) -> IngestReport: + text = file_path.read_text(encoding="utf-8", errors="replace") + + package_name, imports = split_drl(text) + raw_rules = parse_rules(text) + + supported: list[IngestedRule] = [] + unsupported: list[IngestedRule] = [] + errors: list[tuple[str, str]] = [] + + for r in raw_rules: + name = r["name"] + attrs = r["attrs"] + when_block = r["when"] + then_block = r["then"] + + salience_m = SALIENCE_RE.search(attrs) + no_loop = bool(NO_LOOP_RE.search(attrs)) + + patterns = parse_patterns(when_block) if parse_body else [] + actions = parse_actions(then_block) if parse_body else [] + + kind, reason = classify_rule(name, when_block, then_block, patterns, actions) + + rule = IngestedRule( + name=name, + kind=kind, + package=package_name, + imports=imports, + salience=int(salience_m.group(1)) if salience_m else None, + no_loop=no_loop, + when_block=when_block.strip(), + then_block=then_block.strip(), + raw_drl=text, + source_file=str(file_path), + patterns=patterns, + actions=actions, + support_reason=reason, + ) + + if kind == "unsupported": + unsupported.append(rule) + else: + supported.append(rule) + + return IngestReport( + source_file=str(file_path), + package=package_name, + total_rules=len(raw_rules), + supported=supported, + unsupported=unsupported, + errors=errors, + ) + + +def ingest_directory( + path: Path, + *, + glob_pattern: str = "*.drl", + recursive: bool = True, + parse_body: bool = True, +) -> list[IngestReport]: + if path.is_file(): + return [ingest_drl(path, parse_body=parse_body)] + + pattern = f"**/{glob_pattern}" if recursive else glob_pattern + reports = [] + for f in sorted(path.glob(pattern)): + if f.is_file(): + try: + reports.append(ingest_drl(f, parse_body=parse_body)) + except Exception as e: + print(f"[WARN] {f}: {e}", file=sys.stderr) + return reports + + +def format_summary(reports: list[IngestReport]) -> str: + total_files = len(reports) + total_rules = sum(r.total_rules for r in reports) + total_supported = sum(len(r.supported) for r in reports) + total_unsupported = sum(len(r.unsupported) for r in reports) + + by_kind: dict[str, int] = {} + for r in reports: + for s in r.supported: + by_kind[s.kind] = by_kind.get(s.kind, 0) + 1 + + lines = [ + f"Files: {total_files}", + f"Total rules: {total_rules}", + f"Supported: {total_supported}", + f"Unsupported: {total_unsupported}", + "", + "Breakdown by kind:", + ] + for kind, count in sorted(by_kind.items()): + lines.append(f" {kind}: {count}") + if total_unsupported: + lines.append("") + lines.append("Unsupported rules:") + for r in reports: + for u in r.unsupported: + lines.append(f" {u.source_file}:{u.name} — {u.support_reason}") + return "\n".join(lines) + + +def main() -> None: + ap = argparse.ArgumentParser( + description="Ingest IriusRisk DRL files into a structured catalog." + ) + ap.add_argument( + "path", + type=str, + help="DRL file or directory containing .drl files", + ) + ap.add_argument( + "--json", + action="store_true", + help="Output full catalog as JSON (default: summary text)", + ) + ap.add_argument( + "--no-parse-body", + action="store_true", + help="Skip detailed pattern/action parsing (faster, header-only)", + ) + ap.add_argument( + "--flat", + action="store_true", + help="Flatten all rules into a single array (used with --json)", + ) + ap.add_argument( + "--glob", + type=str, + default="*.drl", + help="Glob pattern when scanning a directory (default: *.drl)", + ) + args = ap.parse_args() + + target = Path(args.path) + if not target.exists(): + print(f"Error: {target} does not exist", file=sys.stderr) + sys.exit(1) + + reports = ingest_directory( + target, + glob_pattern=args.glob, + parse_body=not args.no_parse_body, + ) + + if not reports: + print("No DRL files found.", file=sys.stderr) + sys.exit(1) + + if args.json: + if args.flat: + all_rules = [] + for r in reports: + all_rules.extend(r.supported) + all_rules.extend(r.unsupported) + print(json.dumps([asdict(r) for r in all_rules], indent=2)) + else: + print(json.dumps([asdict(r) for r in reports], indent=2)) + else: + print(format_summary(reports)) + + +if __name__ == "__main__": + main() diff --git a/src/iriusrisk_drl_toolkit/models.py b/src/iriusrisk_drl_toolkit/models.py index 69e2d1c..b23d06b 100644 --- a/src/iriusrisk_drl_toolkit/models.py +++ b/src/iriusrisk_drl_toolkit/models.py @@ -65,3 +65,37 @@ class RuleEvaluation: applies: bool reasons: list[str] failed: list[str] + + +RuleKind = Literal[ + "relation_rule", + "component_rule", + "unsupported", +] + + +@dataclass +class IngestedRule: + name: str + kind: str + package: str | None + imports: list[str] + salience: int | None + no_loop: bool + when_block: str + then_block: str + raw_drl: str + source_file: str + patterns: list[dict] # parsed condition patterns + actions: list[dict] # parsed action statements + support_reason: str # why supported or not + + +@dataclass +class IngestReport: + source_file: str + package: str | None + total_rules: int + supported: list[IngestedRule] + unsupported: list[IngestedRule] + errors: list[tuple[str, str]] # rule_name, error_msg