fix: harden custom SAST rule selection + filtering behavior

Signed-off-by: lelia <2418071+lelia@users.noreply.github.com>

Author: lelia

socket_basics/core/connector/opengrep/__init__.py

@@ -40,6 +40,12 @@ def scan(self) -> Dict[str, Any]:
 			rule_files = self.config.build_opengrep_rules() or []
 		except Exception:
 			rule_files = []
+		logger.info(
+			"OpenGrep config summary: all_languages_enabled=%s all_rules_enabled=%s requested_rule_files=%s",
+			bool(self.config.get('all_languages_enabled', False)),
+			bool(self.config.get('all_rules_enabled', False)),
+			rule_files,
+		)
 
 		# If no languages selected and not explicitly allowing all, skip
 		if not rule_files and not self.config.get('all_languages_enabled', False):
@@ -51,6 +57,12 @@ def scan(self) -> Dict[str, Any]:
 		# Check if custom rules mode is enabled
 		custom_rules_path = self.config.get_custom_rules_path()
 		custom_rule_files: Dict[str, Path] = {}
+		logger.info(
+			"Custom SAST requested=%s custom_path=%s resolved_path=%s",
+			bool(self.config.get('use_custom_sast_rules', False)),
+			self.config.get('custom_sast_rule_path', ''),
+			str(custom_rules_path) if custom_rules_path else '(none)',
+		)
 
 		if custom_rules_path:
 			logger.info(f"Custom SAST rules enabled, loading from: {custom_rules_path}")
@@ -74,6 +86,11 @@ def scan(self) -> Dict[str, Any]:
 			filtered = self.config.build_filtered_opengrep_rules() or {}
 		except Exception:
 			filtered = {}
+		if filtered:
+			filtered_counts = {k: len(v or []) for k, v in filtered.items()}
+			logger.info("Per-language enabled-rule filters detected: %s", filtered_counts)
+		else:
+			logger.info("Per-language enabled-rule filters disabled for this run")
 
 		# Debugging: log computed rule files and filtered rules for diagnosis
 		try:
@@ -91,25 +108,42 @@ def scan(self) -> Dict[str, Any]:
 		# Process all enabled languages - use filtered rules if specified, otherwise use all rules
 		for rf in rule_files:
 			# Check if we have a custom rule file for this language
+			using_custom_rules = bool(custom_rule_files and rf in custom_rule_files)
 			if custom_rule_files and rf in custom_rule_files:
 				p = custom_rule_files[rf]
-				logger.info(f"Using custom rules for {rf}")
+				logger.info("Using custom rules for %s from %s", rf, p)
 			else:
 				# Fall back to bundled rules
 				p = Path(rules_dir) / rf
 				if not p.exists():
 					logger.debug('Rule file missing: %s', p)
 					continue
+				logger.info("Using bundled rules for %s from %s", rf, p)
 
 			# Check if this language has specific rules enabled (filtered mode)
 			if filtered and rf in filtered:
 				enabled_ids = filtered[rf]
-				logger.debug(f"Using filtered rules for {rf}: {len(enabled_ids)} rules enabled")
+				logger.info("Filtering rules for %s: %d enabled IDs configured", rf, len(enabled_ids))
 				try:
 					with open(p, 'r') as fh:
 						data = yaml.safe_load(fh) or {}
 					all_ids = [r.get('id') for r in (data.get('rules') or []) if r.get('id')]
-					to_exclude = [rid for rid in all_ids if rid not in (enabled_ids or [])]
+					# Custom-rule mode can coexist with legacy bundled allowlists.
+					# If none of the configured enabled IDs match custom IDs, keep all
+					# custom IDs active to avoid silently disabling user-authored rules.
+					if using_custom_rules:
+						matched_enabled_ids = [rid for rid in all_ids if rid in (enabled_ids or [])]
+						if enabled_ids and not matched_enabled_ids:
+							logger.warning(
+								"No configured enabled-rule IDs matched custom rules for %s; using all custom rules from %s",
+								rf,
+								p,
+							)
+							config_args.extend(['--config', str(p)])
+							continue
+						to_exclude = [rid for rid in all_ids if rid not in matched_enabled_ids]
+					else:
+						to_exclude = [rid for rid in all_ids if rid not in (enabled_ids or [])]
 					config_args.extend(['--config', str(p)])
 					for ex in to_exclude:
 						config_args.extend(['--exclude-rule', ex])

tests/test_opengrep_custom_rules.py

@@ -0,0 +1,183 @@
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+from socket_basics.core.config import Config
+from socket_basics.core.connector.opengrep import OpenGrepScanner
+
+
+def _write_rule_file(path: Path, rule_ids: list[str]) -> None:
+    rules = [{"id": rid, "languages": ["javascript"], "pattern": "eval(...)"} for rid in rule_ids]
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"rules": rules}), encoding="utf-8")
+
+
+def _write_custom_rules_file(path: Path, rule_ids: list[str]) -> None:
+    lines = ["rules:"]
+    for rid in rule_ids:
+        lines.extend(
+            [
+                f"  - id: {rid}",
+                "    pattern: eval(...)",
+                "    languages: [javascript, typescript]",
+                f'    message: Rule {rid}',
+                "    severity: ERROR",
+            ]
+        )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def _mock_subprocess_run(monkeypatch, captured_cmd: list[str]):
+    def _runner(cmd, capture_output, text):
+        captured_cmd.extend(cmd)
+        out_file = cmd[cmd.index("--output") + 1]
+        Path(out_file).write_text(json.dumps({"results": []}), encoding="utf-8")
+        return SimpleNamespace(stdout="", stderr="", returncode=0)
+
+    monkeypatch.setattr("socket_basics.core.connector.opengrep.subprocess.run", _runner)
+
+
+def test_scan_uses_custom_rule_file_when_available(tmp_path, monkeypatch):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir(parents=True, exist_ok=True)
+
+    custom_rules_dir = workspace / ".socket" / "rules"
+    # Custom file can be any yaml name; builder groups by languages.
+    custom_rules_file = custom_rules_dir / "org-rules.yml"
+    _write_custom_rules_file(custom_rules_file, ["org.no-eval"])
+
+    bundled_rules_dir = tmp_path / "bundled-rules"
+    _write_rule_file(bundled_rules_dir / "javascript_typescript.yml", ["js-default-rule"])
+
+    config = Config(
+        {
+            "workspace": str(workspace),
+            "output_dir": str(workspace),
+            "javascript_sast_enabled": True,
+            "use_custom_sast_rules": True,
+            "custom_sast_rule_path": ".socket/rules",
+            "opengrep_rules_dir": str(bundled_rules_dir),
+            "all_languages_enabled": False,
+            "all_rules_enabled": False,
+            "verbose": False,
+        }
+    )
+    scanner = OpenGrepScanner(config)
+    scanner._convert_to_socket_facts = lambda _: {"components": []}
+    scanner.generate_notifications = lambda _: {}
+
+    captured_cmd: list[str] = []
+    _mock_subprocess_run(monkeypatch, captured_cmd)
+    scanner.scan()
+
+    cmd_str = " ".join(captured_cmd)
+    assert "socket_custom_rules_" in cmd_str
+    assert str(bundled_rules_dir / "javascript_typescript.yml") not in cmd_str
+
+
+def test_scan_falls_back_to_bundled_file_when_custom_missing(tmp_path, monkeypatch):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir(parents=True, exist_ok=True)
+
+    bundled_rules_dir = tmp_path / "bundled-rules"
+    bundled_file = bundled_rules_dir / "javascript_typescript.yml"
+    _write_rule_file(bundled_file, ["js-default-rule"])
+
+    config = Config(
+        {
+            "workspace": str(workspace),
+            "output_dir": str(workspace),
+            "javascript_sast_enabled": True,
+            "use_custom_sast_rules": True,
+            "custom_sast_rule_path": ".socket/missing-rules",
+            "opengrep_rules_dir": str(bundled_rules_dir),
+            "all_languages_enabled": False,
+            "all_rules_enabled": False,
+            "verbose": False,
+        }
+    )
+    scanner = OpenGrepScanner(config)
+    scanner._convert_to_socket_facts = lambda _: {"components": []}
+    scanner.generate_notifications = lambda _: {}
+
+    captured_cmd: list[str] = []
+    _mock_subprocess_run(monkeypatch, captured_cmd)
+    scanner.scan()
+
+    assert str(bundled_file) in " ".join(captured_cmd)
+
+
+def test_custom_rules_ignore_nonmatching_bundled_allowlist_ids(tmp_path, monkeypatch):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir(parents=True, exist_ok=True)
+
+    custom_rules_file = workspace / ".socket" / "rules" / "org-rules.yml"
+    _write_custom_rules_file(custom_rules_file, ["org.no-eval", "org.no-innerhtml"])
+
+    bundled_rules_dir = tmp_path / "bundled-rules"
+    _write_rule_file(bundled_rules_dir / "javascript_typescript.yml", ["js-default-rule"])
+
+    config = Config(
+        {
+            "workspace": str(workspace),
+            "output_dir": str(workspace),
+            "javascript_sast_enabled": True,
+            "javascript_enabled_rules": "js-default-rule",
+            "use_custom_sast_rules": True,
+            "custom_sast_rule_path": ".socket/rules",
+            "opengrep_rules_dir": str(bundled_rules_dir),
+            "all_languages_enabled": False,
+            "all_rules_enabled": False,
+            "verbose": False,
+        }
+    )
+    scanner = OpenGrepScanner(config)
+    scanner._convert_to_socket_facts = lambda _: {"components": []}
+    scanner.generate_notifications = lambda _: {}
+
+    captured_cmd: list[str] = []
+    _mock_subprocess_run(monkeypatch, captured_cmd)
+    scanner.scan()
+
+    cmd_str = " ".join(captured_cmd)
+    assert "socket_custom_rules_" in cmd_str
+    assert "--exclude-rule org.no-eval" not in cmd_str
+    assert "--exclude-rule org.no-innerhtml" not in cmd_str
+
+
+def test_custom_rules_apply_allowlist_when_custom_ids_match(tmp_path, monkeypatch):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir(parents=True, exist_ok=True)
+
+    custom_rules_file = workspace / ".socket" / "rules" / "org-rules.yml"
+    _write_custom_rules_file(custom_rules_file, ["org.no-eval", "org.no-innerhtml"])
+
+    bundled_rules_dir = tmp_path / "bundled-rules"
+    _write_rule_file(bundled_rules_dir / "javascript_typescript.yml", ["js-default-rule"])
+
+    config = Config(
+        {
+            "workspace": str(workspace),
+            "output_dir": str(workspace),
+            "javascript_sast_enabled": True,
+            "javascript_enabled_rules": "org.no-eval",
+            "use_custom_sast_rules": True,
+            "custom_sast_rule_path": ".socket/rules",
+            "opengrep_rules_dir": str(bundled_rules_dir),
+            "all_languages_enabled": False,
+            "all_rules_enabled": False,
+            "verbose": False,
+        }
+    )
+    scanner = OpenGrepScanner(config)
+    scanner._convert_to_socket_facts = lambda _: {"components": []}
+    scanner.generate_notifications = lambda _: {}
+
+    captured_cmd: list[str] = []
+    _mock_subprocess_run(monkeypatch, captured_cmd)
+    scanner.scan()
+
+    cmd_str = " ".join(captured_cmd)
+    assert "--exclude-rule org.no-innerhtml" in cmd_str
+    assert "--exclude-rule org.no-eval" not in cmd_str