opencode
321 lines
8.5 KiB
Python
321 lines
8.5 KiB
Python
#!/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()
|