🛡️ Sentinel: [HIGH] Fix SSRF bypass via HTTP redirects

.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()`.

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

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 <url> for authorized DAST")
+            _console_print(
+                "   ZAP baseline: provide --zap-baseline <url> 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,15 +3250,19 @@ 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)
         conn.close()
         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,15 +3272,19 @@ 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")
         try:
             _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]

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,