diff --git a/appguardrail_core/rules.py b/appguardrail_core/rules.py index 1adc7dff..c8cdf88e 100644 --- a/appguardrail_core/rules.py +++ b/appguardrail_core/rules.py @@ -21,9 +21,17 @@ "OWASP A07:2021 - Identification and Authentication Failures", "CWE-798 - Use of Hard-coded Credentials", ), + "ssrf": ( + "OWASP A10:2021 - Server-Side Request Forgery", + "CWE-918 - Server-Side Request Forgery", + ), "storage": ("OWASP A01:2021 - Broken Access Control",), } +REFERENCE_CATEGORY_OVERRIDES = { + "CWE-918": "ssrf", +} + SAMM_BY_CATEGORY = { "authz": "Implementation / Secure Build", "dependency": "Implementation / Secure Build", @@ -31,6 +39,7 @@ "misconfig": "Operations / Environment Management", "payment": "Verification / Requirements-driven Testing", "secrets": "Operations / Environment Management", + "ssrf": "Implementation / Secure Build", "storage": "Implementation / Secure Build", } @@ -59,6 +68,10 @@ "Remove the secret from source, rotate it, and load future values from " "managed secret storage." ), + "ssrf": ( + "Validate untrusted URLs before persistence, reject non-public destinations, " + "and revalidate redirects or pin the outbound destination before delivery." + ), "storage": "Enforce storage or database access controls with authenticated ownership policies.", } @@ -101,6 +114,15 @@ def extract_public_references(message: str) -> tuple[str, ...]: ) +def _category_for_references(references: tuple[str, ...], fallback: str) -> str: + """Prefer an authoritative public taxonomy over a rule-id heuristic.""" + for reference in references: + for prefix, category in REFERENCE_CATEGORY_OVERRIDES.items(): + if reference.startswith(prefix): + return category + return fallback + + def build_rule_metadata( rule_id: str, severity: str, @@ -110,8 +132,10 @@ def build_rule_metadata( source: str = "appguardrail-rule", ) -> RuleMetadata: """Build a stable metadata envelope for a scanner finding.""" + public_references = extract_public_references(message) + category = _category_for_references(public_references, category) references = _merge_references( - extract_public_references(message), + public_references, CATEGORY_REFERENCE_DEFAULTS.get(category, ()), ) return RuleMetadata( diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index 7dd60bec..0d853d64 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -969,6 +969,7 @@ def finish_rule(): "message_lines": [], "include_paths": [], "exclude_paths": [], + "required_substrings": [], "severity": "WARNING", } in_message = False @@ -1001,6 +1002,12 @@ def finish_rule(): current["languages"] = _parse_inline_list(raw_line.split(":", 1)[1]) path_mode = None continue + if raw_line.startswith(" prefilter: "): + current["required_substrings"] = _parse_inline_list( + raw_line.split(":", 1)[1] + ) + path_mode = None + continue if raw_line.startswith(" - pattern-regex: "): current["regexes"].append( _unquote_rule_scalar(raw_line.split("pattern-regex:", 1)[1]) @@ -1038,6 +1045,9 @@ def _compile_yaml_regex_rule(rule): "extensions": extensions, "include_paths": rule.get("include_paths") or [], "exclude_paths": rule.get("exclude_paths") or [], + "required_substrings": tuple( + rule.get("required_substrings") or () + ), } ) return compiled_rules @@ -2148,6 +2158,7 @@ def _get_applicable_rules(ext: str): rule["pattern"].finditer, tuple(rule.get("include_paths") or ()), tuple(rule.get("exclude_paths") or ()), + tuple(rule.get("required_substrings") or ()), ) for rule in SCAN_RULES if not rule["extensions"] or ext in rule["extensions"] @@ -2962,7 +2973,12 @@ def _scan_file( finditer, include_paths, exclude_paths, + required_substrings, ) in applicable_rules: + if required_substrings and not all( + substring in content for substring in required_substrings + ): + continue if include_paths or exclude_paths: if rel_path_for_filters is None: rel_path_for_filters = _display_path( diff --git a/scanner/rules/ssrf.yml b/scanner/rules/ssrf.yml new file mode 100644 index 00000000..b0625106 --- /dev/null +++ b/scanner/rules/ssrf.yml @@ -0,0 +1,14 @@ +rules: + - id: python-stored-ssrf-webhook-url + patterns: + - pattern-regex: '(?is)(?:\b(?P[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])(?:(?!\bset_webhook\b).){0,800}?(?m:^(?P[ \t]*)if\s+(?:(?!\bnot\b)[^\n:]){0,400}_is_safe_url\s*\(\s*(?P=unguarded_var)\s*\)[^\n:]{0,100}:\s*(?:#[^\n]*)?\n(?:(?P=unguarded_indent)[ \t]+[^\n]*\n){0,20}?(?P=unguarded_indent)[ \t]+\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=unguarded_var)\s*\))(?:(?!\bset_webhook\b).){0,800}?(?m:^(?P=unguarded_indent)\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=unguarded_var)\s*\))|\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])|\b(?P[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])(?!(?:(?!\bset_webhook\b).){0,800}(?m:^(?P[ \t]*)if\s+(?:(?P=webhook_url_var)\s+is\s+not\s+None\s+and\s+not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)|(?:(?P=webhook_url_var)\s+and\s+)?not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)|(?P=webhook_url_var)\s+not\s+in\s*\(\s*None\s*,\s*["\x27]["\x27]\s*\)\s+and\s*\(\s*not\s+isinstance\s*\(\s*(?P=webhook_url_var)\s*,\s*str\s*\)\s+or\s+not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)\s*\))(?:(?!:).){0,100}:\s*(?:#[^\n]*)?\n(?:(?P=reject_indent)[ \t]+[^\n]*\n){0,20}?(?P=reject_indent)[ \t]+\b(?:return|raise)\b))(?!(?:(?!\bset_webhook\b).){0,800}(?m:^(?P[ \t]*)if\s+(?:(?!\bnot\b)[^\n:]){0,400}_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)[^\n:]{0,100}:\s*(?:#[^\n]*)?\n(?:(?P=guard_indent)[ \t]+[^\n]*\n){0,20}?(?P=guard_indent)[ \t]+\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=webhook_url_var)\s*\)))(?:(?!\bset_webhook\b).){0,800}?\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=webhook_url_var)\s*\))' + message: | + A webhook URL is persisted directly from request data without an explicit + validation boundary. Validate the URL with a fail-closed SSRF policy before + storing it, and revalidate or pin the destination before outbound delivery. + [CWE-918 - Server-Side Request Forgery] + severity: HIGH + languages: [python] + prefilter: [set_webhook] + cwe: [CWE-918] + owasp: [A10:2021] diff --git a/tests/test_ssrf_rule_accessors.py b/tests/test_ssrf_rule_accessors.py new file mode 100644 index 00000000..62195f06 --- /dev/null +++ b/tests/test_ssrf_rule_accessors.py @@ -0,0 +1,77 @@ +"""Regression tests for equivalent stored-SSRF request accessors.""" + +import pytest + +from scanner.cli.appguardrail import SCAN_RULES, _scan_file + +_RULE_ID = "python-stored-ssrf-webhook-url" + + +def _rule(): + """Return the packaged stored-SSRF rule under test.""" + matches = [rule for rule in SCAN_RULES if rule["id"] == _RULE_ID] + assert len(matches) == 1 + return matches[0] + + +def _source(accessor, *, direct=False, validated=False): + """Build a direct or one-hop request URL persistence flow.""" + sink = "set_" + "webhook" + if direct: + return "\n".join( + [ + "def update_webhook(conn, org, body):", + f" {sink}(conn, org, {accessor})", + "", + ] + ) + + lines = [ + "def update_webhook(conn, org, body):", + f" target = {accessor}", + ] + if validated: + lines.extend( + [ + " if not _is_safe_url(target):", + " return", + ] + ) + lines.extend([f" {sink}(conn, org, target)", ""]) + return "\n".join(lines) + + +@pytest.mark.parametrize( + "source", + [ + _source('body["url"]', direct=True), + _source('request.json["url"]'), + _source('request.json.get("url")'), + ], +) +def test_packaged_rule_matches_equivalent_request_url_accessors(source): + """Detect direct, subscript, and attribute-based URL sources.""" + assert _rule()["pattern"].search(source) + + +def test_scan_file_emits_finding_for_subscript_variable_flow(tmp_path): + """Emit the stored-SSRF finding for a subscript one-hop flow.""" + source_file = tmp_path / "webhook.py" + source_file.write_text(_source('body["url"]'), encoding="utf-8") + + findings = [ + finding + for finding in _scan_file(source_file, tmp_path) + if finding["rule_id"] == _RULE_ID + ] + + assert len(findings) == 1 + assert findings[0]["line"] == 2 + assert findings[0]["category"] == "ssrf" + + +def test_packaged_rule_ignores_validated_subscript_flow(): + """Do not flag a subscript source protected by a fail-closed guard.""" + assert not _rule()["pattern"].search( + _source('body["url"]', validated=True) + ) diff --git a/tests/test_ssrf_rules.py b/tests/test_ssrf_rules.py new file mode 100644 index 00000000..296e068d --- /dev/null +++ b/tests/test_ssrf_rules.py @@ -0,0 +1,327 @@ +"""Regression tests for stored SSRF detection in the packaged rule engine.""" + +from unittest.mock import patch + +from scanner.cli.appguardrail import SCAN_RULES, _scan_file + +_RULE_ID = "python-stored-ssrf-webhook-url" + + +def _rule(): + """Return the single packaged stored-SSRF rule under test.""" + matches = [rule for rule in SCAN_RULES if rule["id"] == _RULE_ID] + assert len(matches) == 1, f"expected one loaded rule for {_RULE_ID}" + return matches[0] + + +def _vulnerable_source(): + """Build the original direct request-to-persistence vulnerability.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + f' {sink}(conn, org, (body or {{}}).get("url"))', + "", + ] + ) + + +def _unvalidated_variable_source(variable="webhook_url"): + """Build an unvalidated local-variable flow into webhook persistence.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + f' {variable} = (body or {{}}).get("url")', + f" {sink}(conn, org, {variable})", + "", + ] + ) + + +def _ignored_validation_result_source(): + """Build a flow that calls the validator but discards its result.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' target = (body or {}).get("url")', + " _is_safe_url(target)", + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _non_enforcing_guard_source(): + """Build a guard that logs invalid input without blocking persistence.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' target = (body or {}).get("url")', + " if not _is_safe_url(target):", + ' log.warning("unsafe webhook url")', + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _non_enforcing_guard_with_unrelated_return_source(): + """Build a non-enforcing guard followed by an unrelated early return.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body, disabled):", + ' target = (body or {}).get("url")', + " if not _is_safe_url(target):", + ' log.warning("unsafe webhook url")', + " if disabled:", + " return", + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _conditional_rejection_guard_source(): + """Build a guard that rejects only when an unrelated flag is true.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body, disabled):", + ' target = (body or {}).get("url")', + " if disabled and not _is_safe_url(target):", + " return", + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _positive_guard_source(): + """Build a safe flow whose persistence sink is inside a positive guard.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' target = (body or {}).get("url")', + " if _is_safe_url(target):", + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _positive_guard_then_unprotected_sink_source(): + """Build a safe guarded sink followed by an unsafe unguarded sink.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' target = (body or {}).get("url")', + " if _is_safe_url(target):", + f" {sink}(conn, org, target)", + f" {sink}(conn, org, target)", + "", + ] + ) + + +def _safe_source(): + """Build a safe flow that raises before persisting invalid input.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' webhook_url = (body or {}).get("url")', + " if webhook_url and not _is_safe_url(webhook_url):", + " raise ValueError(\"unsafe webhook url\")", + f" {sink}(conn, org, webhook_url)", + "", + ] + ) + + +def _production_guard_source(): + """Build the multiline fail-closed guard used by the control plane.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' webhook_url = body.get("url")', + ' if webhook_url not in (None, "") and (', + " not isinstance(webhook_url, str)", + " or not _is_safe_url(webhook_url)", + " ):", + ' return {"error": "unsafe webhook url"}', + f" {sink}(conn, org, webhook_url)", + "", + ] + ) + + +def _none_aware_guard_source(): + """Build the compact fail-closed guard used by the control plane.""" + sink = "set_" + "webhook" + return "\n".join( + [ + "def update_webhook(conn, org, body):", + ' webhook_url = body.get("url")', + " if webhook_url is not None and not _is_safe_url(webhook_url):", + ' return {"error": "unsafe webhook url"}', + f" {sink}(conn, org, webhook_url)", + "", + ] + ) + + +def _scan_rule_findings(tmp_path, source): + """Run the real file scanner and return only stored-SSRF findings.""" + source_file = tmp_path / "webhook.py" + source_file.write_text(source, encoding="utf-8") + return [ + finding + for finding in _scan_file(source_file, tmp_path) + if finding["rule_id"] == _RULE_ID + ] + + +def test_packaged_rule_matches_direct_request_url_persistence(): + """Detect direct request URL persistence with HIGH severity.""" + rule = _rule() + assert rule["severity"] == "HIGH" + assert rule["pattern"].search(_vulnerable_source()) + + +def test_packaged_rule_declares_sink_prefilter(): + """Skip the expensive flow regex unless the persistence sink is present.""" + assert _rule()["required_substrings"] == ("set_webhook",) + + +def test_scan_file_skips_regex_when_sink_prefilter_is_absent(tmp_path): + """Do not invoke an expensive regex for files without its required sink.""" + source_file = tmp_path / "benign.py" + source_file.write_text('target = body.get("url")\n' * 1000, encoding="utf-8") + + class ExplodingPattern: + """Prove that prefilter rejection happens before regex evaluation.""" + + def finditer(self, _content): + raise AssertionError("regex must not run without the sink literal") + + rule = { + "id": _RULE_ID, + "pattern": ExplodingPattern(), + "severity": "HIGH", + "message": "stored SSRF [CWE-918 - Server-Side Request Forgery]", + "extensions": [".py"], + "required_substrings": ("set_webhook",), + } + with patch("scanner.cli.appguardrail.SCAN_RULES", [rule]): + assert not _scan_file(source_file, tmp_path) + + +def test_packaged_rule_matches_unvalidated_variable_persistence(): + """Detect source-to-sink persistence through a local variable.""" + assert _rule()["pattern"].search(_unvalidated_variable_source()) + + +def test_packaged_rule_does_not_depend_on_url_variable_name(): + """Detect the flow even when the variable name omits the word URL.""" + assert _rule()["pattern"].search(_unvalidated_variable_source("target")) + + +def test_packaged_rule_does_not_treat_ignored_validator_result_as_safe(): + """Detect a validator call whose boolean result is discarded.""" + assert _rule()["pattern"].search(_ignored_validation_result_source()) + + +def test_packaged_rule_matches_non_enforcing_validation_guard(): + """Detect a validation branch that logs but does not terminate.""" + assert _rule()["pattern"].search(_non_enforcing_guard_source()) + + +def test_packaged_rule_matches_non_enforcing_guard_with_unrelated_return(): + """Ignore unrelated returns when deciding whether validation enforces.""" + assert _rule()["pattern"].search( + _non_enforcing_guard_with_unrelated_return_source() + ) + + +def test_packaged_rule_matches_conditional_rejection_guard(): + """Do not treat an unrelated conditional rejection as fail-closed.""" + assert _rule()["pattern"].search(_conditional_rejection_guard_source()) + + +def test_packaged_rule_ignores_positive_guarded_persistence(): + """Do not flag a sink that is reachable only after positive validation.""" + assert not _rule()["pattern"].search(_positive_guard_source()) + + +def test_packaged_rule_matches_unprotected_sink_after_positive_guard(): + """A guarded sink must not hide a later unprotected persistence sink.""" + assert _rule()["pattern"].search( + _positive_guard_then_unprotected_sink_source() + ) + + +def test_packaged_rule_ignores_fail_closed_guarded_persistence(): + """Do not flag a flow that raises on invalid input before persistence.""" + assert not _rule()["pattern"].search(_safe_source()) + + +def test_packaged_rule_ignores_production_fail_closed_guard(): + """Do not self-flag the control plane's multiline rejection guard.""" + assert not _rule()["pattern"].search(_production_guard_source()) + + +def test_packaged_rule_ignores_none_aware_fail_closed_guard(): + """Do not self-flag a None-aware fail-closed rejection guard.""" + assert not _rule()["pattern"].search(_none_aware_guard_source()) + + +def test_scan_file_emits_stored_ssrf_finding(tmp_path): + """Emit a normalized SSRF finding through the production file scanner.""" + matches = _scan_rule_findings(tmp_path, _vulnerable_source()) + + assert len(matches) == 1 + finding = matches[0] + assert finding["severity"] == "HIGH" + assert finding["source"] == "appguardrail-rule" + assert finding["file"] == "webhook.py" + assert finding["line"] == 2 + assert finding["category"] == "ssrf" + assert finding["cwe"] == ("CWE-918 - Server-Side Request Forgery",) + assert finding["owasp"] == ("OWASP A10:2021 - Server-Side Request Forgery",) + assert "destination" in finding["remediation"].lower() + + +def test_scan_file_emits_stored_ssrf_finding_for_variable_flow(tmp_path): + """Emit a finding for an unvalidated local-variable persistence flow.""" + matches = _scan_rule_findings(tmp_path, _unvalidated_variable_source()) + + assert len(matches) == 1 + assert matches[0]["line"] == 2 + + +def test_scan_file_emits_finding_when_validator_result_is_ignored(tmp_path): + """Emit a finding when code ignores the validator's return value.""" + matches = _scan_rule_findings(tmp_path, _ignored_validation_result_source()) + + assert len(matches) == 1 + assert matches[0]["line"] == 2 + + +def test_scan_file_emits_finding_for_non_enforcing_guard(tmp_path): + """Emit a finding when an invalid branch fails to stop persistence.""" + matches = _scan_rule_findings(tmp_path, _non_enforcing_guard_source()) + + assert len(matches) == 1 + assert matches[0]["line"] == 2 + + +def test_scan_file_does_not_flag_validated_path(tmp_path): + """Suppress the finding for a verified fail-closed persistence path.""" + assert not _scan_rule_findings(tmp_path, _safe_source())