346 lines
14 KiB
Python
346 lines
14 KiB
Python
"""Step definitions for security scan hooks configuration feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from behave import given, then, when
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
|
|
|
|
@given("the pre-commit config file exists")
|
|
def step_precommit_config_exists(context: Any) -> None:
|
|
config_path = PROJECT_ROOT / ".pre-commit-config.yaml"
|
|
if not config_path.is_file():
|
|
raise FileNotFoundError(f".pre-commit-config.yaml not found at {config_path}")
|
|
context.precommit_config_path = config_path
|
|
context.precommit_config_content = config_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@when("I parse the pre-commit configuration")
|
|
def step_parse_precommit_config(context: Any) -> None:
|
|
content = context.precommit_config_content
|
|
data = yaml.safe_load(content)
|
|
hooks: list[dict[str, Any]] = []
|
|
for repo in data.get("repos", []):
|
|
for hook in repo.get("hooks", []):
|
|
hooks.append(hook)
|
|
context.precommit_hooks = hooks
|
|
context.precommit_data = data
|
|
|
|
|
|
@then('a hook with id "{hook_id}" should be declared')
|
|
def step_hook_declared(context: Any, hook_id: str) -> None:
|
|
hooks = context.precommit_hooks
|
|
found = any(h.get("id") == hook_id for h in hooks)
|
|
if not found:
|
|
ids = [h.get("id") for h in hooks]
|
|
raise AssertionError(
|
|
f"Hook '{hook_id}' not found in pre-commit config. Found: {ids}"
|
|
)
|
|
|
|
|
|
@then('the "{hook_id}" hook should have a files pattern matching src')
|
|
def step_hook_files_src(context: Any, hook_id: str) -> None:
|
|
hooks = context.precommit_hooks
|
|
for hook in hooks:
|
|
if hook.get("id") == hook_id:
|
|
files_pattern = hook.get("files", "")
|
|
if "src" not in files_pattern:
|
|
raise AssertionError(
|
|
f"Hook '{hook_id}' files pattern '{files_pattern}' "
|
|
f"does not match src"
|
|
)
|
|
return
|
|
raise AssertionError(f"Hook '{hook_id}' not found in pre-commit config")
|
|
|
|
|
|
@then('the "{hook_id}" hook args should reference pyproject.toml')
|
|
def step_hook_args_pyproject(context: Any, hook_id: str) -> None:
|
|
hooks = context.precommit_hooks
|
|
for hook in hooks:
|
|
if hook.get("id") == hook_id:
|
|
args = hook.get("args", [])
|
|
args_str = " ".join(str(a) for a in args)
|
|
if "pyproject.toml" not in args_str:
|
|
raise AssertionError(
|
|
f"Hook '{hook_id}' args do not reference pyproject.toml: {args}"
|
|
)
|
|
return
|
|
raise AssertionError(f"Hook '{hook_id}' not found in pre-commit config")
|
|
|
|
|
|
@then('the "{hook_id}" hook entry should reference .semgrep.yml')
|
|
def step_hook_entry_semgrep_yml(context: Any, hook_id: str) -> None:
|
|
hooks = context.precommit_hooks
|
|
for hook in hooks:
|
|
if hook.get("id") == hook_id:
|
|
entry = hook.get("entry", "")
|
|
if ".semgrep.yml" not in entry:
|
|
raise AssertionError(
|
|
f"Hook '{hook_id}' entry does not reference .semgrep.yml: {entry}"
|
|
)
|
|
return
|
|
raise AssertionError(f"Hook '{hook_id}' not found in pre-commit config")
|
|
|
|
|
|
@when("I parse the dev dependencies from pyproject.toml")
|
|
def step_parse_dev_deps(context: Any) -> None:
|
|
import tomllib
|
|
|
|
content = context.pyproject_content
|
|
data = tomllib.loads(content)
|
|
dev_deps = data.get("project", {}).get("optional-dependencies", {}).get("dev", [])
|
|
context.dev_dependencies = dev_deps
|
|
|
|
|
|
@then('"{package}" should be in the dev dependencies')
|
|
def step_package_in_dev_deps(context: Any, package: str) -> None:
|
|
deps = context.dev_dependencies
|
|
found = any(package in dep for dep in deps)
|
|
if not found:
|
|
raise AssertionError(f"'{package}' not found in dev dependencies: {deps}")
|
|
|
|
|
|
@when("I parse the nox session names from noxfile.py")
|
|
def step_parse_nox_sessions(context: Any) -> None:
|
|
content = context.noxfile_content
|
|
tree = ast.parse(content)
|
|
sessions: list[str] = []
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef):
|
|
for decorator in node.decorator_list:
|
|
if isinstance(decorator, ast.Call):
|
|
func = decorator.func
|
|
if (isinstance(func, ast.Attribute) and func.attr == "session") or (
|
|
isinstance(func, ast.Name) and func.id == "session"
|
|
):
|
|
sessions.append(node.name)
|
|
elif isinstance(decorator, ast.Attribute):
|
|
if decorator.attr == "session":
|
|
sessions.append(node.name)
|
|
context.nox_sessions = sessions
|
|
|
|
|
|
@then('"{session_name}" should be a registered nox session')
|
|
def step_session_registered(context: Any, session_name: str) -> None:
|
|
sessions = context.nox_sessions
|
|
if session_name not in sessions:
|
|
raise AssertionError(
|
|
f"Session '{session_name}' not found in nox sessions: {sessions}"
|
|
)
|
|
|
|
|
|
@when("I read the security_scan session source from noxfile.py")
|
|
def step_read_security_session(context: Any) -> None:
|
|
content = context.noxfile_content
|
|
pattern = r"(def security_scan\(.*?\n(?:(?: .*\n|[ \t]*\n)*))"
|
|
match = re.search(pattern, content)
|
|
if match is None:
|
|
raise ValueError("security_scan function not found in noxfile.py")
|
|
context.security_session_source = match.group(1)
|
|
|
|
|
|
@then("the session source should contain a bandit invocation")
|
|
def step_session_has_bandit(context: Any) -> None:
|
|
source = context.security_session_source
|
|
if "bandit" not in source.lower():
|
|
raise AssertionError(
|
|
"security_scan session does not contain a bandit invocation"
|
|
)
|
|
|
|
|
|
@then("the session source should contain a semgrep invocation")
|
|
def step_session_has_semgrep(context: Any) -> None:
|
|
source = context.security_session_source
|
|
if "semgrep" not in source.lower():
|
|
raise AssertionError(
|
|
"security_scan session does not contain a semgrep invocation"
|
|
)
|
|
|
|
|
|
@then("the session source should contain a vulture invocation")
|
|
def step_session_has_vulture(context: Any) -> None:
|
|
source = context.security_session_source
|
|
if "vulture" not in source.lower():
|
|
raise AssertionError(
|
|
"security_scan session does not contain a vulture invocation"
|
|
)
|
|
|
|
|
|
@given("the semgrep config file exists")
|
|
def step_semgrep_config_exists(context: Any) -> None:
|
|
semgrep_path = PROJECT_ROOT / ".semgrep.yml"
|
|
if not semgrep_path.is_file():
|
|
raise FileNotFoundError(f".semgrep.yml not found at {semgrep_path}")
|
|
context.semgrep_config_path = semgrep_path
|
|
context.semgrep_config_content = semgrep_path.read_text(encoding="utf-8")
|
|
|
|
|
|
@when("I parse the semgrep configuration")
|
|
def step_parse_semgrep_config(context: Any) -> None:
|
|
content = context.semgrep_config_content
|
|
data = yaml.safe_load(content)
|
|
context.semgrep_config = data
|
|
|
|
|
|
@then("the semgrep config should contain at least {count:d} rules")
|
|
def step_semgrep_rule_count(context: Any, count: int) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
actual_count = len(rules)
|
|
if actual_count < count:
|
|
raise AssertionError(
|
|
f"Expected at least {count} semgrep rules, found {actual_count}"
|
|
)
|
|
|
|
|
|
@then('the semgrep config should contain a rule with id "{rule_id}"')
|
|
def step_semgrep_rule_exists(context: Any, rule_id: str) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
found = any(r.get("id") == rule_id for r in rules)
|
|
if not found:
|
|
ids = [r.get("id") for r in rules]
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' not found in semgrep config. Found: {ids}"
|
|
)
|
|
|
|
|
|
@then('the rule "{rule_id}" should include path "{path}"')
|
|
def step_semgrep_rule_includes_path(context: Any, rule_id: str, path: str) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
for rule in rules:
|
|
if rule.get("id") == rule_id:
|
|
paths_config = rule.get("paths", {})
|
|
include_paths = paths_config.get("include", [])
|
|
found = any(path in p for p in include_paths)
|
|
if not found:
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' does not include path '{path}'. "
|
|
f"Include paths: {include_paths}"
|
|
)
|
|
return
|
|
raise AssertionError(f"Rule '{rule_id}' not found in semgrep config")
|
|
|
|
|
|
@then('the rule "{rule_id}" message should mention "{text}"')
|
|
def step_semgrep_rule_message_mentions(context: Any, rule_id: str, text: str) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
for rule in rules:
|
|
if rule.get("id") == rule_id:
|
|
message = rule.get("message", "")
|
|
if text not in message:
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' message does not mention '{text}'. "
|
|
f"Message: {message[:200]}"
|
|
)
|
|
return
|
|
raise AssertionError(f"Rule '{rule_id}' not found in semgrep config")
|
|
|
|
|
|
@then('the rule "{rule_id}" should have a pattern-not for bare re-raise')
|
|
def step_semgrep_rule_has_reraise_pattern_not(context: Any, rule_id: str) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
for rule in rules:
|
|
if rule.get("id") == rule_id:
|
|
# Walk the Semgrep patterns structure to find pattern-not entries with bare raise.
|
|
# In YAML, 'patterns' is a list of alternative match groups (pattern-either).
|
|
patterns = rule.get("patterns", [])
|
|
if not isinstance(patterns, list):
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' has unexpected 'patterns' type: "
|
|
f"{type(patterns).__name__}"
|
|
)
|
|
|
|
found_bare_reraise = False
|
|
for group in patterns:
|
|
either_list = group.get("pattern-either", [])
|
|
if not isinstance(either_list, list):
|
|
continue
|
|
for alt in either_list:
|
|
sub_pats = alt.get("patterns", [])
|
|
if not isinstance(sub_pats, list):
|
|
continue
|
|
for sp in sub_pats:
|
|
pn = sp.get("pattern-not", "")
|
|
if isinstance(pn, str):
|
|
# A bare re-raise pattern-not must contain "raise" on its own line
|
|
# (not raise $EXC or raise ... from ...)
|
|
lines = pn.split("\n")
|
|
for line in lines:
|
|
if line.strip() == "raise":
|
|
found_bare_reraise = True
|
|
break
|
|
|
|
# Also handle nested pattern-either blocks inside alternation groups
|
|
nested_either = (
|
|
alt.get("pattern-either") if isinstance(alt, dict) else None
|
|
)
|
|
if isinstance(nested_either, list):
|
|
for nested_alt in nested_either:
|
|
pn = nested_alt.get("pattern-not", "")
|
|
if isinstance(pn, str):
|
|
lines = pn.split("\n")
|
|
for line in lines:
|
|
if line.strip() == "raise":
|
|
found_bare_reraise = True
|
|
break
|
|
|
|
if not found_bare_reraise:
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' does not have a pattern-not for bare re-raise. "
|
|
f"Expected 'pattern-not' with 'raise' (bare re-raise without arguments)"
|
|
)
|
|
return
|
|
raise AssertionError(f"Rule '{rule_id}' not found in semgrep config")
|
|
|
|
|
|
@then('the rule "{rule_id}" should have a pattern-not for exception chaining')
|
|
def step_semgrep_rule_has_chaining_pattern_not(context: Any, rule_id: str) -> None:
|
|
config = context.semgrep_config
|
|
rules = config.get("rules", [])
|
|
for rule in rules:
|
|
if rule.get("id") == rule_id:
|
|
rule_str = str(rule)
|
|
# Check for exception chaining pattern-not (raise X from Y)
|
|
if "from" not in rule_str or "CAUSE" not in rule_str:
|
|
raise AssertionError(
|
|
f"Rule '{rule_id}' does not appear to have a pattern-not "
|
|
f"for exception chaining (expected 'from $CAUSE' pattern)"
|
|
)
|
|
return
|
|
raise AssertionError(f"Rule '{rule_id}' not found in semgrep config")
|
|
|
|
|
|
@when("I read the lint session source from noxfile.py")
|
|
def step_read_lint_session(context: Any) -> None:
|
|
content = context.noxfile_content
|
|
pattern = r"(def lint\(.*?\n(?:(?: .*\n|[ \t]*\n)*))"
|
|
match = re.search(pattern, content)
|
|
if match is None:
|
|
raise ValueError("lint function not found in noxfile.py")
|
|
context.lint_session_source = match.group(1)
|
|
|
|
|
|
@then("the lint session source should contain a semgrep invocation")
|
|
def step_lint_session_has_semgrep(context: Any) -> None:
|
|
source = context.lint_session_source
|
|
if "semgrep" not in source.lower():
|
|
raise AssertionError("lint session does not contain a semgrep invocation")
|
|
|
|
|
|
@then('the lint session source should reference ".semgrep.yml"')
|
|
def step_lint_session_references_semgrep_yml(context: Any) -> None:
|
|
source = context.lint_session_source
|
|
if ".semgrep.yml" not in source:
|
|
raise AssertionError("lint session does not reference '.semgrep.yml'")
|