diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 64c03f5c..b0431edb 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -107,3 +107,8 @@ **Vulnerability:** Server-Side Request Forgery (SSRF) bypass due to `_is_safe_url` only checking `is_loopback` and `is_private`. This fails to correctly evaluate mapped IPv4 addresses disguised as IPv6 (e.g. `[::ffff:127.0.0.1]`) and misses restricted IP designations like `is_reserved` or non `is_global` IPs, allowing SSRF to `0.0.0.0` or `255.255.255.255`. **Learning:** Python's `ipaddress` objects for mapped IPv6 don't inherit properties of their IPv4 wrapped content directly. Using `is_loopback` without checking `.ipv4_mapped` leaves blind spots. **Prevention:** Always extract `getattr(ip, 'ipv4_mapped', None)` before evaluation, and combine checks spanning `is_reserved`, `not is_global`, `is_multicast`, `is_unspecified`, `is_private`, and `is_loopback` to fully protect endpoints. + +## 2026-07-20 - Block HTTP Redirects in URL fetchers to prevent SSRF bypass +**Vulnerability:** Server-Side Request Forgery (SSRF) bypass through HTTP redirects. When a user provides an external URL that passes `_is_safe_url` validation, `urllib.request.urlopen` automatically follows redirects. If the external server redirects to an internal or restricted IP (e.g., `http://127.0.0.1`), the request succeeds, bypassing the initial safety checks. +**Learning:** `urllib.request.urlopen` follows HTTP redirects by default. Validating the initial URL is insufficient if the client subsequently fetches an unvalidated redirect target. +**Prevention:** Explicitly disable HTTP redirects when fetching user-provided URLs by passing a custom `urllib.request.HTTPRedirectHandler` (that prevents redirects) to `urllib.request.build_opener()`. diff --git a/appguardrail_core/controlplane.py b/appguardrail_core/controlplane.py index 07300b0f..642e5494 100644 --- a/appguardrail_core/controlplane.py +++ b/appguardrail_core/controlplane.py @@ -18,7 +18,11 @@ import secrets import sqlite3 from datetime import datetime, timezone -from importlib import resources # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2 + +try: + from importlib import resources # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2 +except ImportError: + import importlib_resources as resources from typing import Any, Iterable from urllib.parse import parse_qs, urlparse @@ -216,11 +220,13 @@ def _slack_blocks( def _is_safe_url(url: str) -> bool: import ipaddress - import urllib.parse import socket + import urllib.parse try: - parsed = urllib.parse.urlparse(url) # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + parsed = urllib.parse.urlparse( + url + ) # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected except ValueError: return False @@ -286,6 +292,10 @@ def _send_alert( import urllib.error import urllib.request + class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + if not _is_safe_url(url): return False @@ -301,7 +311,8 @@ def _send_alert( method="POST", headers={"Content-Type": "application/json"}, ) - urllib.request.urlopen( # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + opener = urllib.request.build_opener(_NoRedirectHandler) + opener.open( # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected req, timeout=10 ) # noqa: S310 - Safe URL scheme validated return True diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py index 94e75ef6..06fb5284 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -60,28 +60,20 @@ from appguardrail_core.config import load_config from appguardrail_core.external import build_external_scan_plan from appguardrail_core.findings import NON_BLOCKING_CONTEXTS -from appguardrail_core.findings import is_deploy_blocking as core_is_deploy_blocking +from appguardrail_core.findings import \ + is_deploy_blocking as core_is_deploy_blocking from appguardrail_core.findings import normalize_findings -from appguardrail_core.language import ( - LANGUAGE_EXTENSIONS, - detect_language_axes, - detect_stack_profile, -) -from appguardrail_core.org_bundle import ( - OrgBundleError, - annotate_missing_pr_repositories, - gh_error_message, - gh_pr_list, - gh_repo_list, -) +from appguardrail_core.language import (LANGUAGE_EXTENSIONS, + detect_language_axes, + detect_stack_profile) +from appguardrail_core.org_bundle import (OrgBundleError, + annotate_missing_pr_repositories, + gh_error_message, gh_pr_list, + gh_repo_list) from appguardrail_core.org_bundle import load_json as load_org_json from appguardrail_core.org_bundle import render_org_evidence, write_bundle -from appguardrail_core.reports import ( - REPORT_TYPE_LABELS, - ReportContext, - render_report, - supported_report_types, -) +from appguardrail_core.reports import (REPORT_TYPE_LABELS, ReportContext, + render_report, supported_report_types) from appguardrail_core.rules import build_rule_metadata __version__ = "0.1.1" @@ -100,10 +92,7 @@ def _format_msg(msg: str) -> str: def _console_print(*values, **kwargs) -> None: """Print CLI values after applying accessibility formatting to strings.""" _ORIGINAL_PRINT( - *( - _format_msg(value) if isinstance(value, str) else value - for value in values - ), + *(_format_msg(value) if isinstance(value, str) else value for value in values), **kwargs, ) @@ -1397,7 +1386,9 @@ def cmd_scan(args): _console_print(f"\nšŸ” AppGuardrail scanning: {scan_path}\n") if run_codegraph: - _console_print("šŸ§­ CodeGraph enabled: initializing or syncing structural index\n") + _console_print( + "šŸ§­ CodeGraph enabled: initializing or syncing structural index\n" + ) try: status = _run_codegraph_index(scan_path) except RuntimeError as exc: @@ -1435,9 +1426,13 @@ def cmd_scan(args): if profile.frameworks: _console_print(f" Framework signals: {', '.join(profile.frameworks)}") if profile.external_tools: - _console_print(f" Optional external engines: {', '.join(profile.external_tools)}") + _console_print( + f" Optional external engines: {', '.join(profile.external_tools)}" + ) if profile.zap_recommended: - _console_print(" ZAP baseline: provide --zap-baseline for authorized DAST") + _console_print( + " ZAP baseline: provide --zap-baseline for authorized DAST" + ) _console_print() external_plan = build_external_scan_plan( @@ -1610,11 +1605,13 @@ def _write_findings_json(findings, output_path: Path): def _is_safe_url(url: str) -> bool: import ipaddress - import urllib.parse import socket + import urllib.parse try: - parsed = urllib.parse.urlparse(url) # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + parsed = urllib.parse.urlparse( + url + ) # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected except ValueError: return False @@ -1688,6 +1685,11 @@ def _push_findings(url, findings): "commit": os.environ.get("GITHUB_SHA"), } endpoint = url.rstrip("/") + "/api/v1/scans" + + class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + req = urllib.request.Request( # noqa: S310 - Safe URL scheme validated endpoint, data=json.dumps(payload).encode("utf-8"), @@ -1698,7 +1700,8 @@ def _push_findings(url, findings): }, ) try: - with urllib.request.urlopen( # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + opener = urllib.request.build_opener(_NoRedirectHandler) + with opener.open( # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected req, timeout=15 ) as resp: # noqa: S310 - Safe URL scheme validated body = json.loads(resp.read() or b"{}") @@ -1799,7 +1802,9 @@ def cmd_fix(args): f"\nšŸ”§ {total_fixes} safe fix{fix_s} available in {changed_files} file{file_s}. " "Re-run with --apply to write them." ) - _console_print(" Other findings need review ā€” see 'appguardrail report fix-pack'.") + _console_print( + " Other findings need review ā€” see 'appguardrail report fix-pack'." + ) return 0 @@ -1837,7 +1842,9 @@ def cmd_report(args): """Generate markdown reports from normalized AppGuardrail findings JSON.""" report_type = getattr(args, "report_type", None) if report_type not in supported_report_types(): - _console_print(f"āŒ Error: Unsupported report type: {report_type}", file=sys.stderr) + _console_print( + f"āŒ Error: Unsupported report type: {report_type}", file=sys.stderr + ) _console_print( "šŸ’” Hint: Supported report types are: " + ", ".join(supported_report_types()), @@ -2096,7 +2103,9 @@ def _path_matches_glob(path: str, pattern: str) -> bool: @functools.lru_cache(maxsize=2048) -def _path_allowed_by_rule_cached(path: str, include_paths: tuple, exclude_paths: tuple) -> bool: +def _path_allowed_by_rule_cached( + path: str, include_paths: tuple, exclude_paths: tuple +) -> bool: """Return whether a path passes optional YAML include/exclude filters (cached).""" if include_paths and not any( _path_matches_glob(path, glob) for glob in include_paths @@ -2106,9 +2115,14 @@ def _path_allowed_by_rule_cached(path: str, include_paths: tuple, exclude_paths: return False return True + def _path_allowed_by_rule(path: str, include_paths, exclude_paths) -> bool: """Return whether a path passes optional YAML include/exclude filters.""" - return _path_allowed_by_rule_cached(path, tuple(include_paths) if include_paths else (), tuple(exclude_paths) if exclude_paths else ()) + return _path_allowed_by_rule_cached( + path, + tuple(include_paths) if include_paths else (), + tuple(exclude_paths) if exclude_paths else (), + ) def _collect_files(base_path: Path): @@ -3010,18 +3024,30 @@ def _print_scan_results(findings, files_scanned): _console_print("\nāš ļø No files were scanned. Are you in the right directory?") elif counts["CRITICAL"] > 0: issue_word = "issue" if counts["CRITICAL"] == 1 else "issues" - _console_print(_format_msg(f"\nāŒ Critical {issue_word} found. Fix before deploying.")) + _console_print( + _format_msg(f"\nāŒ Critical {issue_word} found. Fix before deploying.") + ) elif counts["HIGH"] > 0: issue_word = "issue" if counts["HIGH"] == 1 else "issues" - _console_print(_format_msg(f"\nāš ļø High-severity {issue_word} found. Review before deploying.")) + _console_print( + _format_msg( + f"\nāš ļø High-severity {issue_word} found. Review before deploying." + ) + ) elif not findings: _console_print(_format_msg("\nāœ… No issues found in this scan.")) else: - _console_print(_format_msg("\nāœ… No deploy-blocking critical or high issues found.")) + _console_print( + _format_msg("\nāœ… No deploy-blocking critical or high issues found.") + ) if findings: these_word = "this issue" if len(findings) == 1 else "these issues" - _console_print(_format_msg(f"\nšŸ’” Run 'appguardrail review' to get an AI prompt for fixing {these_word}.")) + _console_print( + _format_msg( + f"\nšŸ’” Run 'appguardrail review' to get an AI prompt for fixing {these_word}." + ) + ) _console_print() @@ -3051,8 +3077,12 @@ def cmd_review(args): _console_print("ā•" * 60 + "\n") _console_print("šŸ’” Tips:") _console_print(" - Paste this into Claude Code, Cursor, or any AI assistant") - _console_print(" - Include relevant files as context (API routes, DB schema, etc.)") - _console_print(" - Run 'appguardrail scan .' first to identify specific files to review") + _console_print( + " - Include relevant files as context (API routes, DB schema, etc.)" + ) + _console_print( + " - Run 'appguardrail scan .' first to identify specific files to review" + ) _console_print() @@ -3220,7 +3250,9 @@ def cmd_serve(args): key_path = _api_key_output_path(args, db) if key_path.exists(): conn.close() - _console_print(f"āŒ API key file already exists: {key_path}", file=sys.stderr) + _console_print( + f"āŒ API key file already exists: {key_path}", file=sys.stderr + ) _console_print("šŸ’” Pass --api-key-file with a new path.", file=sys.stderr) return 1 oid, key = cp.create_org(conn, create) @@ -3228,7 +3260,9 @@ def cmd_serve(args): try: _persist_api_key(key_path, key) except FileExistsError: - _console_print(f"āŒ API key file already exists: {key_path}", file=sys.stderr) + _console_print( + f"āŒ API key file already exists: {key_path}", file=sys.stderr + ) _console_print("šŸ’” Pass --api-key-file with a new path.", file=sys.stderr) return 1 _console_print(f"āœ… Created org '{create}' (id {oid}).") @@ -3238,7 +3272,9 @@ def cmd_serve(args): key_path = _api_key_output_path(args, db) if key_path.exists(): conn.close() - _console_print(f"āŒ API key file already exists: {key_path}", file=sys.stderr) + _console_print( + f"āŒ API key file already exists: {key_path}", file=sys.stderr + ) _console_print("šŸ’” Pass --api-key-file with a new path.", file=sys.stderr) return 1 _oid, key = cp.create_org(conn, "default") @@ -3246,7 +3282,9 @@ def cmd_serve(args): _persist_api_key(key_path, key) except FileExistsError: conn.close() - _console_print(f"āŒ API key file already exists: {key_path}", file=sys.stderr) + _console_print( + f"āŒ API key file already exists: {key_path}", file=sys.stderr + ) _console_print("šŸ’” Pass --api-key-file with a new path.", file=sys.stderr) return 1 _console_print("ā„¹ļø No orgs yet ā€” created 'default'.") @@ -3323,7 +3361,9 @@ def cmd_dashboard(args): index = dashboard_index_path() if not index.is_file(): - _console_print(f"āŒ Error: Dashboard assets not found at {index}", file=sys.stderr) + _console_print( + f"āŒ Error: Dashboard assets not found at {index}", file=sys.stderr + ) _console_print( "šŸ’” Hint: Check if the path is correct or if you are in the right directory.", file=sys.stderr, @@ -3342,7 +3382,9 @@ def cmd_dashboard(args): " Generate one with: " "appguardrail scan --findings-json reports/findings.json ." ) - _console_print(" The dashboard opens with instructions ā€” reload after generating.\n") + _console_print( + " The dashboard opens with instructions ā€” reload after generating.\n" + ) tokens_css = b"" tokens_file = dashboard_tokens_path() @@ -3364,8 +3406,12 @@ def cmd_dashboard(args): host, port, index.read_bytes(), findings_path, tokens_css ) except OSError as exc: - _console_print(f"āŒ Cannot start dashboard on {host}:{port} ({exc}).", file=sys.stderr) - _console_print("šŸ’” Pass a free port with --port, e.g. --port 8899.", file=sys.stderr) + _console_print( + f"āŒ Cannot start dashboard on {host}:{port} ({exc}).", file=sys.stderr + ) + _console_print( + "šŸ’” Pass a free port with --port, e.g. --port 8899.", file=sys.stderr + ) return 1 actual_port = server.server_address[1] diff --git a/tests/test_controlplane.py b/tests/test_controlplane.py index eb6b40dd..482de893 100644 --- a/tests/test_controlplane.py +++ b/tests/test_controlplane.py @@ -352,16 +352,19 @@ def test_slack_blocks_caps_and_escapes(): def test_send_alert_slack_vs_generic(monkeypatch): posted = {} - def _fake_urlopen(req, timeout=None): - posted["url"] = req.full_url - posted["body"] = json.loads(req.data.decode()) + def _fake_open(self, req, data=None, timeout=None): + posted["url"] = req.full_url if hasattr(req, "full_url") else req + posted["body"] = json.loads(req.data.decode()) if hasattr(req, "data") else None - class _R: # minimal stand-in, urlopen result is ignored - pass + class _R: + def close(self): pass + @property + def status(self): return 200 + def read(self): return b"{}" return _R() - monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen) + monkeypatch.setattr(urllib.request.OpenerDirector, "open", _fake_open) generic = { "event": "drift.new_blocking", "org_id": 3,