feat(rules): Semgrep-compat YAML rule engine — Phase 1 pattern matching for Python (closes #46 phase-1)

pyproject.toml

@@ -54,12 +54,24 @@ grammars = [
 watch = [
     "watchdog",
 ]
+# Issue #46: Semgrep-compat YAML rule engine — Phase 1.
+# PyYAML is already required transitively but listed for clarity.
+rules = [
+    "PyYAML>=6.0",
+]
+# Issue #48: Native LSP 3.17 server via `codelens lsp`.
+# pygls + lsprotocol provide the JSON-RPC + LSP type bindings.
+lsp = [
+    "pygls>=2.0",
+    "lsprotocol>=2024.0",
+    "tree-sitter-python>=0.23",
+]
 dev = [
     "pytest>=7.0",
     "pytest-cov",
 ]
 all = [
-    "codelens[grammars,watch,dev]",
+    "codelens[grammars,watch,rules,lsp,dev]",
 ]
 
 # Entry point removed — codelens is run directly via python3 codelens.py

scripts/commands/check.py

@@ -24,6 +24,16 @@ def add_args(parser):
     parser.add_argument('--commands', nargs='+',
                         default=['secrets', 'dead-code', 'smell', 'complexity', 'debug-leak', 'circular', 'taint'],
                         help='Commands to run for the quality gate (default: core analysis)')
+    # Issue #46: Semgrep-compatible YAML rule engine — additive flag.
+    # When supplied, the rule engine runs over the workspace and any
+    # findings above --severity are added to the gate's finding list.
+    parser.add_argument('--rule-file', dest='rule_files',
+                        action='append', default=None,
+                        metavar='<path.yaml>',
+                        help='Path to a Semgrep-compatible YAML rule file '
+                             '(issue #46). May be passed multiple times. '
+                             'Additive — rule findings are merged into the '
+                             'quality-gate result.')
 
 
 def execute(args, workspace):
@@ -122,6 +132,48 @@ def execute(args, workspace):
         if severity_order.get(f.get('severity', 'medium'), 2) <= min_sev
     ]
 
+    # Issue #46: merge Semgrep-compat rule-engine findings into the gate.
+    # Rule severities (CRITICAL/HIGH/MEDIUM/LOW/INFO/ERROR/WARNING/HINT)
+    # are normalized to the gate's severity vocabulary before filtering.
+    rule_findings_count = 0
+    rule_files = getattr(args, 'rule_files', None)
+    if rule_files:
+        try:
+            from rule_engine import run_rules_against_file
+            import os
+            sev_map = {
+                'CRITICAL': 'critical', 'HIGH': 'high',
+                'MEDIUM': 'medium', 'LOW': 'low', 'INFO': 'info',
+                'ERROR': 'critical', 'WARNING': 'high',
+                'HINT': 'low',
+            }
+            py_exts = {".py", ".pyw", ".pyi"}
+            for dirpath, _dirs, files in os.walk(workspace):
+                for name in files:
+                    if os.path.splitext(name)[1].lower() not in py_exts:
+                        continue
+                    file_path = os.path.join(dirpath, name)
+                    rr = run_rules_against_file(file_path, rule_files)
+                    if rr.error:
+                        errors.append(f"rule-engine: {rr.error}")
+                        continue
+                    for m in rr.matches:
+                        gate_sev = sev_map.get(m.severity.upper(), 'medium')
+                        if severity_order.get(gate_sev, 2) <= min_sev:
+                            relevant_findings.append({
+                                'severity': gate_sev,
+                                'rule_id': m.rule_id,
+                                'message': m.message,
+                                'file': file_path,
+                                'line': m.range.start_point[0] + 1,
+                                'column': m.range.start_point[1] + 1,
+                                'category': 'rule-engine',
+                                'metavariables': m.metavariables,
+                            })
+                            rule_findings_count += 1
+        except ImportError as exc:
+            errors.append(f"rule-engine unavailable: {exc}")
+
     # Count by severity
     by_severity = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0, 'info': 0}
     for f in relevant_findings:
@@ -171,6 +223,7 @@ def execute(args, workspace):
         "health_score": health_score,
         "total_findings": len(all_findings),
         "relevant_findings": len(relevant_findings),
+        "rule_findings": rule_findings_count,
         "findings": relevant_findings,
         "by_severity": by_severity,
         "commands_run": command_results,

scripts/commands/scan.py

@@ -105,6 +105,17 @@
                              "'Scan stats: N files, M nodes, K edges' + "
                              "'Index time: Xs (parse: Ys, write: Zs)'. "
                              "Off by default — does not affect scan output.")
+    # Issue #46: Semgrep-compatible YAML rule engine — additive flag.
+    # When supplied, the rule engine (scripts/rule_engine.py) runs after
+    # the tree-sitter scan completes and prints one finding per match to
+    # stderr. Backward compat: omitting the flag keeps scan output byte-
+    # identical to the pre-#46 behavior.
+    parser.add_argument("--rule-file", dest="rule_files",
+                        action="append", default=None,
+                        metavar="<path.yaml>",
+                        help="Path to a Semgrep-compatible YAML rule file "
+                             "(issue #46). May be passed multiple times. "
+                             "Additive — does not change default scan behavior.")
 
 
 def execute(args, workspace):
@@ -119,14 +130,77 @@
     use_prefilter = getattr(args, 'use_prefilter', True)
     verbose = getattr(args, 'verbose', False)
     scan_stats = getattr(args, 'scan_stats', False)
+    rule_files = getattr(args, 'rule_files', None)
     # Only auto-enable incremental if the user didn't explicitly request a full scan
     # and the registry already exists. We check for explicit --incremental flag.
     # Note: When user runs "scan" without --incremental, they expect a full scan.
     # Auto-incremental was causing confusion where 2nd scan would miss changes.
     # Now: explicit --incremental for incremental, bare "scan" for full scan.
-    return cmd_scan(workspace, incremental, plugins=plugins, max_files=max_files,
-                    use_prefilter=use_prefilter, verbose=verbose,
-                    scan_stats=scan_stats)
+    result = cmd_scan(workspace, incremental, plugins=plugins, max_files=max_files,
+                      use_prefilter=use_prefilter, verbose=verbose,
+                      scan_stats=scan_stats)
+
+    # Issue #46: run the Semgrep-compat rule engine after the scan completes.
+    # Additive — when --rule-file is omitted, _run_rule_files is a no-op and
+    # the scan output stays byte-identical to the pre-#46 behavior.
+    if rule_files:
+        _run_rule_files(workspace, rule_files, verbose=verbose)
+
+    return result
+
+
+def _run_rule_files(workspace: str, rule_files: list, verbose: bool = False) -> None:
+    """Issue #46 — run the Semgrep-compatible rule engine across the workspace.
+
+    Walks the workspace and runs :mod:`rule_engine` against every Python file.
+    Findings are printed to stderr (one per line) so they don't pollute the
+    machine-readable scan result on stdout. Failures (parse errors, missing
+    rule files) are logged but never crash the scan — the rule engine is
+    strictly additive.
+
+    Phase 1: Python-only. Non-Python files are skipped silently.
+    """
+    try:
+        from rule_engine import run_rules_against_file, format_match_for_cli
+    except ImportError as exc:
+        import sys
+        print(f"codelens: rule engine unavailable (--rule-file ignored): {exc}",
+              file=sys.stderr)
+        return
+
+    workspace = os.path.abspath(workspace)
+    py_exts = {".py", ".pyw", ".pyi"}
+    for dirpath, _dirs, files in os.walk(workspace):
+        # Respect .codelensignore if available — skip ignored dirs entirely
+        if _codelensignore_is_ignored is not None:
+            try:
+                if _codelensignore_is_ignored(dirpath + os.sep, workspace):
+                    continue
+            except Exception:
+                pass
+        for name in files:
+            ext = os.path.splitext(name)[1].lower()
+            if ext not in py_exts:
+                continue
+            file_path = os.path.join(dirpath, name)
+            if _codelensignore_is_ignored is not None:
+                try:
+                    if _codelensignore_is_ignored(file_path, workspace):
+                        continue
+                except Exception:
+                    pass
+            result = run_rules_against_file(file_path, rule_files)
+            if result.error:
+                import sys
+                print(f"codelens: {result.error}", file=sys.stderr)
+                continue
+            for m in result.matches:
+                import sys
+                print(format_match_for_cli(m, file_path), file=sys.stderr)
+            if verbose and result.matches:
+                import sys
+                print(f"codelens: {file_path}: {len(result.matches)} rule finding(s)",
+                      file=sys.stderr)
 
 
 def _run_suggest_ignore(workspace: str) -> Dict[str, Any]:

scripts/rule_engine.py

@@ -0,0 +1,117 @@
+"""
+CodeLens — rule-engine entry point.
+
+Provides a single :func:`run_rules_against_file` function that the
+``scan`` and ``check`` commands can call when the user passes
+``--rule-file <path.yaml>``.
+
+The integration is purely additive — when no ``--rule-file`` is supplied,
+nothing in this module is called and CodeLens behaves exactly as before.
+
+File header — CodeLens rule engine (Phase 1).
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Iterable
+
+from rule_matcher import Match, match_source
+from rule_pattern_parser import Rule, RuleParseError, parse_rule_file
+
+
+@dataclass
+class RuleFileResult:
+    """Aggregate result of running one or more rule files against one file."""
+
+    file_path: str
+    matches: list[Match]
+    rules_loaded: int
+    error: str | None = None
+
+
+def load_rules(rule_files: Iterable[str]) -> tuple[list[Rule], list[str]]:
+    """
+    Load rules from one or more YAML files.
+
+    Returns ``(rules, errors)`` where ``errors`` is a list of human-readable
+    error strings (one per file that failed to parse). Files that fail
+    do not abort the whole batch — partial results are still returned.
+    """
+    rules: list[Rule] = []
+    errors: list[str] = []
+    for path in rule_files:
+        try:
+            rules.extend(parse_rule_file(path))
+        except RuleParseError as exc:
+            errors.append(f"{path}: {exc}")
+    return rules, errors
+
+
+def run_rules_against_file(
+    file_path: str,
+    rule_files: Iterable[str],
+) -> RuleFileResult:
+    """
+    Load rules from ``rule_files`` and run them against ``file_path``.
+
+    Returns a :class:`RuleFileResult`. Never raises for parse errors or
+    file-not-found — those are surfaced via ``result.error``.
+    """
+    rule_files = list(rule_files)
+    rules, errors = load_rules(rule_files)
+    if errors:
+        return RuleFileResult(
+            file_path=file_path,
+            matches=[],
+            rules_loaded=len(rules),
+            error="; ".join(errors),
+        )
+
+    if not rules:
+        return RuleFileResult(
+            file_path=file_path,
+            matches=[],
+            rules_loaded=0,
+            error=None,
+        )
+
+    try:
+        with open(file_path, "rb") as fh:
+            source_bytes = fh.read()
+    except OSError as exc:
+        return RuleFileResult(
+            file_path=file_path,
+            matches=[],
+            rules_loaded=len(rules),
+            error=f"cannot read {file_path}: {exc}",
+        )
+
+    # Use file extension to decide language. Phase 1: only Python.
+    ext = os.path.splitext(file_path)[1].lower()
+    if ext not in {".py", ".pyw", ".pyi"}:
+        return RuleFileResult(
+            file_path=file_path,
+            matches=[],
+            rules_loaded=len(rules),
+            error=None,
+        )
+
+    matches = match_source(rules, source_bytes)
+    return RuleFileResult(
+        file_path=file_path,
+        matches=matches,
+        rules_loaded=len(rules),
+        error=None,
+    )
+
+
+def format_match_for_cli(m: Match, file_path: str) -> str:
+    """One-line human-readable summary, suitable for stderr / CLI output."""
+    row = m.range.start_point[0] + 1
+    col = m.range.start_point[1] + 1
+    mv = ""
+    if m.metavariables:
+        mv = " " + ", ".join(f"{k}={v!r}" for k, v in m.metavariables.items())
+    return f"{file_path}:{row}:{col}: [{m.severity}] {m.rule_id}: {m.message}{mv}"