Add DRL ingestion script and IngestReport/IngestedRule models
This commit is contained in:
opencode
@@ -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<name>[^"]+)"\s*
|
||||
(?P<attrs>.*?)
|
||||
when\s
|
||||
(?P<when>.*?)
|
||||
then\s
|
||||
(?P<then>.*?)
|
||||
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()
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user