diff --git a/.jules/sentinel.md b/.jules/sentinel.md index f88bf46..e295aee 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -112,3 +112,7 @@ **Vulnerability:** Cross-Site Scripting (XSS) via `javascript:` URIs in `href` attributes in `scanner/dashboard/index.html`. **Learning:** `f.references` values (URLs) were injected into `href` attributes with only HTML entity escaping via `esc()`. HTML escaping `javascript:alert(1)` does not neutralize the `javascript:` URI protocol, meaning that clicking the link executes the injected script in the context of the dashboard UI. This allowed an attacker controlling finding references to achieve XSS. **Prevention:** Always validate the URL scheme (allow-listing `http:` and `https:`) using `new URL()` parser to ensure user-provided URLs cannot leverage dangerous schemes like `javascript:`, `file:`, or `data:` when injected into `href` attributes, before applying standard HTML escaping. +## 2026-07-25 - Prevent SSRF Bypasses via HTTP Redirects +**Vulnerability:** HTTP clients using `urllib.request.urlopen` followed HTTP redirects automatically, bypassing initial SSRF and LFI checks (like `_is_safe_url`) on the origin URL and allowing redirection to internal network locations or sensitive IP addresses (e.g., `http://127.0.0.1/` or `http://169.254.169.254/`). +**Learning:** `_is_safe_url` checking only the initial URL does not protect against SSRF bypasses via `301`/`302` HTTP redirects since standard library clients follow redirects blindly to any valid URL scheme/host. +**Prevention:** Always implement and enforce a custom redirect handler (`urllib.request.HTTPRedirectHandler`) attached to a custom opener (`urllib.request.build_opener`) and re-apply `_is_safe_url` checks inside `redirect_request` for every redirect hop. diff --git a/appguardrail_core/controlplane.py b/appguardrail_core/controlplane.py index 07300b0..3896715 100644 --- a/appguardrail_core/controlplane.py +++ b/appguardrail_core/controlplane.py @@ -18,9 +18,11 @@ import secrets import sqlite3 from datetime import datetime, timezone -from importlib import resources # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2 +import importlib.resources as resources # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2 from typing import Any, Iterable from urllib.parse import parse_qs, urlparse +import urllib.request +import urllib.error from .findings import is_deploy_blocking, normalize_findings, severity_counts @@ -220,7 +222,9 @@ def _is_safe_url(url: str) -> bool: import socket 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 @@ -270,6 +274,17 @@ def is_bad_ip(ip) -> bool: return True +import urllib.request +import urllib.error + + +class SafeRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + if not _is_safe_url(newurl): + raise urllib.error.URLError(f"Unsafe redirect URL: {newurl}") + return super().redirect_request(req, fp, code, msg, headers, newurl) + + def _send_alert( url: str, payload: dict[str, Any], @@ -301,7 +316,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(SafeRedirectHandler()) + 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 43a7678..059a4d5 100644 --- a/scanner/cli/appguardrail.py +++ b/scanner/cli/appguardrail.py @@ -60,20 +60,28 @@ 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" @@ -1661,6 +1669,17 @@ def is_bad_ip(ip) -> bool: return True +import urllib.request +import urllib.error + + +class SafeRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + if not _is_safe_url(newurl): + raise urllib.error.URLError(f"Unsafe redirect URL: {newurl}") + return super().redirect_request(req, fp, code, msg, headers, newurl) + + def _push_findings(url, findings): """POST normalized findings to a control-plane /api/v1/scans endpoint.""" import urllib.error @@ -1695,7 +1714,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(SafeRedirectHandler()) + 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"{}") diff --git a/tests/test_controlplane.py b/tests/test_controlplane.py index eb6b40d..beea829 100644 --- a/tests/test_controlplane.py +++ b/tests/test_controlplane.py @@ -8,13 +8,23 @@ import pytest -from appguardrail_core.controlplane import (_is_slack_webhook, _send_alert, - _slack_blocks, add_scan, connect, - create_key, create_org, get_scan, - has_role, list_scans, - make_control_plane_server, - org_for_key, role_for_key, - scan_trend, set_webhook) +from appguardrail_core.controlplane import ( + _is_slack_webhook, + _send_alert, + _slack_blocks, + add_scan, + connect, + create_key, + create_org, + get_scan, + has_role, + list_scans, + make_control_plane_server, + org_for_key, + role_for_key, + scan_trend, + set_webhook, +) FINDINGS = [ {"severity": "CRITICAL", "rule_id": "x", "context": "app-code"}, @@ -361,7 +371,14 @@ class _R: # minimal stand-in, urlopen result is ignored return _R() - monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen) + def _fake_build_opener(*handlers): + class _Opener: + def open(self, req, timeout=None): + return _fake_urlopen(req, timeout=timeout) + + return _Opener() + + monkeypatch.setattr(urllib.request, "build_opener", _fake_build_opener) generic = { "event": "drift.new_blocking", "org_id": 3, diff --git a/tests/test_ssrf_protection.py b/tests/test_ssrf_protection.py index 8cb0884..3bcfdfa 100644 --- a/tests/test_ssrf_protection.py +++ b/tests/test_ssrf_protection.py @@ -58,3 +58,40 @@ def test_is_safe_url_mapped_ips(): def test_is_safe_url_reserved_and_not_global_ips(): assert not _is_safe_url("http://255.255.255.255/") assert not _is_safe_url("http://0.0.0.0/") + + +def test_safe_redirect_blocks_unsafe_url(): + from appguardrail_core.controlplane import SafeRedirectHandler + import urllib.error + import pytest + + handler = SafeRedirectHandler() + + with pytest.raises(urllib.error.URLError) as exc: + handler.redirect_request(None, None, 302, "Found", None, "http://127.0.0.1/") + + assert "Unsafe redirect URL" in str(exc.value) + + with pytest.raises(urllib.error.URLError) as exc: + handler.redirect_request( + None, None, 302, "Found", None, "http://169.254.169.254/" + ) + + assert "Unsafe redirect URL" in str(exc.value) + + # Safe url should fall through and not raise URLError on our end, might raise AttributeError on None but we can mock it + class DummyReq: + pass + + class DummyFP: + pass + + class DummyHeaders: + pass + + try: + handler.redirect_request( + DummyReq(), DummyFP(), 302, "Found", DummyHeaders(), "https://github.com/" + ) + except AttributeError: + pass # this means super().redirect_request was called