fix(security): detect Rust at import time + SSRF Python fallback (#467)

* fix(security): detect Rust at import time + SSRF Python fallback (#225)

RUST_AVAILABLE was hardcoded to True, so the exported flag never
reflected reality and the pure-Python fallbacks it was meant to gate
were unreachable.

- _rust_bridge: compute RUST_AVAILABLE dynamically by probing the
  compiled extension once at import time.
- security.ssrf: check_ssrf now falls back to the existing
  _check_ssrf_python implementation when the Rust extension is not
  built, instead of raising ImportError. The SSRF guard is
  security-critical and must never be silently skipped or crash just
  because Rust was not compiled.
- tools.browser: drop the `except ImportError: pass` around the SSRF
  check, which previously disabled SSRF protection entirely on installs
  without the compiled backend (internal/metadata endpoints reachable).
- tests: cover the Python fallback path (metadata IP, private IP, and
  public URL) with RUST_AVAILABLE patched False.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* test(security): make test_no_hostname backend-agnostic

The cherry-picked #451 fix adds a pure-Python SSRF fallback, but
test_no_hostname asserted the Rust-specific message "Invalid URL". The
Python fallback returns "No hostname in URL" for the same input, so the
SSRF suite failed on exactly the uncompiled-install path #451 targets
(in CI the Rust extension is built, masking it).

Assert the security behavior (URL blocked, non-None reason) and accept
either backend's wording, so the suite passes on both the Rust and
Python paths.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(tools): update browser SSRF test for non-bypassable check

The #225 fix removes the `except ImportError: pass` that silently
disabled the SSRF check in browser navigation. The existing
test_execute_ssrf_module_missing asserted that very anti-pattern ("skip
check and proceed"), so it failed once the swallow was removed.

Replace it with tests that assert the SECURE behavior: the SSRF check
runs unconditionally and is honored (a private-IP URL is blocked even
when navigation would otherwise succeed), and a public URL still
navigates. check_ssrf's pure-Python fallback means the import never
fails anymore, so the old skip path no longer exists.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Rahul <therahulll56@gmail.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Jon Saad-Falcon
2026-06-01 11:57:44 -07:00
committed by GitHub
co-authored by Claude Opus 4.8 Rahul
parent 09b19193fe
commit 2b14315f3c
5 changed files with 114 additions and 35 deletions
+15 -1
View File
@@ -32,7 +32,21 @@ def get_rust_module() -> _types.ModuleType:
return openjarvis_rust
RUST_AVAILABLE: bool = True
def _detect_rust() -> bool:
"""Return ``True`` if the compiled ``openjarvis_rust`` extension is importable.
Computed once at import time. Modules with a Python fallback (e.g.
``security.ssrf``) consult this flag instead of hardcoding availability,
so the fallback is actually reachable when the extension was not built.
"""
try:
get_rust_module()
except ImportError:
return False
return True
RUST_AVAILABLE: bool = _detect_rust()
# ---------------------------------------------------------------------------
+13 -5
View File
@@ -70,15 +70,23 @@ def is_private_ip(ip_str: str) -> bool:
def check_ssrf(url: str) -> Optional[str]:
"""Check a URL for SSRF vulnerabilities — always via Rust backend."""
from openjarvis._rust_bridge import get_rust_module
"""Check a URL for SSRF vulnerabilities.
_rust = get_rust_module()
return _rust.check_ssrf(url)
Prefers the Rust backend, but falls back to the pure-Python
implementation when the compiled extension is unavailable. The SSRF
guard is security-critical, so it must never be silently skipped — or
crash with ``ImportError`` — merely because Rust was not built.
"""
from openjarvis._rust_bridge import RUST_AVAILABLE, get_rust_module
if RUST_AVAILABLE:
return get_rust_module().check_ssrf(url)
return _check_ssrf_python(url)
def _check_ssrf_python(url: str) -> Optional[str]:
"""Legacy Python SSRF check — kept for reference only."""
"""Pure-Python SSRF check — fallback used when the Rust extension is
unavailable (e.g. an install without the compiled backend)."""
from urllib.parse import urlparse
parsed = urlparse(url)
+11 -12
View File
@@ -101,19 +101,18 @@ class BrowserNavigateTool(BaseTool):
if wait_for not in ("load", "domcontentloaded", "networkidle"):
wait_for = "load"
# SSRF check
try:
from openjarvis.security.ssrf import check_ssrf
# SSRF check — never skipped. check_ssrf falls back to a pure-Python
# implementation when the Rust backend is unavailable, so an
# uncompiled extension must not silently disable SSRF protection.
from openjarvis.security.ssrf import check_ssrf
ssrf_error = check_ssrf(url)
if ssrf_error:
return ToolResult(
tool_name="browser_navigate",
content=f"SSRF blocked: {ssrf_error}",
success=False,
)
except ImportError:
pass # ssrf module not available, skip check
ssrf_error = check_ssrf(url)
if ssrf_error:
return ToolResult(
tool_name="browser_navigate",
content=f"SSRF blocked: {ssrf_error}",
success=False,
)
try:
page = _session.page
+37 -6
View File
@@ -126,10 +126,15 @@ class TestCheckSsrf:
assert "private IP" in result
def test_no_hostname(self):
# Rust returns "Invalid URL" for malformed URLs (no scheme => parse error)
# A URL with no usable hostname must be blocked (non-None reason).
# The exact wording is backend-specific — Rust's URL parser errors
# with "Invalid URL", while the Python fallback's urlparse yields no
# hostname and returns "No hostname in URL". Assert the security
# behavior (blocked), not the backend-specific message, so the test
# passes on both paths.
result = check_ssrf("not-a-url")
assert result is not None
assert "Invalid URL" in result
assert "Invalid URL" in result or "No hostname" in result
def test_dns_failure_allowed(self):
"""DNS resolution failure should not block — request will fail at HTTP time."""
@@ -188,9 +193,7 @@ class TestCheckSsrf:
assert "private IP" in result
def test_python_impl_blocks_ipv4_mapped_metadata(self):
result = _check_ssrf_python(
"http://[::ffff:169.254.169.254]/latest/meta-data/"
)
result = _check_ssrf_python("http://[::ffff:169.254.169.254]/latest/meta-data/")
assert result is not None
def test_blocks_ipv4_mapped_alibaba_metadata(self):
@@ -220,4 +223,32 @@ class TestCheckSsrf:
assert result is not None
__all__ = ["TestCheckSsrf", "TestIsPrivateIp"]
class TestCheckSsrfPythonFallback:
"""When the Rust extension is not compiled, ``check_ssrf`` must fall back
to the pure-Python implementation rather than raising ``ImportError`` or
being silently skipped — the SSRF guard is security-critical.
"""
def test_falls_back_to_python_when_rust_unavailable(self):
with patch("openjarvis._rust_bridge.RUST_AVAILABLE", False):
result = check_ssrf("http://169.254.169.254/latest/meta-data/")
assert result is not None
assert "cloud metadata" in result.lower() or "Blocked host" in result
def test_fallback_blocks_private_ip(self):
with patch("openjarvis._rust_bridge.RUST_AVAILABLE", False):
with patch("openjarvis.security.ssrf.socket.getaddrinfo") as mock_dns:
mock_dns.return_value = [(2, 1, 6, "", ("10.0.0.5", 0))]
result = check_ssrf("http://internal-service.local/api")
assert result is not None
assert "private IP" in result
def test_fallback_allows_public_url_without_rust(self):
with patch("openjarvis._rust_bridge.RUST_AVAILABLE", False):
with patch("openjarvis.security.ssrf.socket.getaddrinfo") as mock_dns:
mock_dns.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
result = check_ssrf("https://example.com")
assert result is None
__all__ = ["TestCheckSsrf", "TestCheckSsrfPythonFallback", "TestIsPrivateIp"]
+38 -11
View File
@@ -135,28 +135,55 @@ class TestBrowserNavigateTool:
assert result.success is False
assert "SSRF blocked" in result.content
def test_execute_ssrf_module_missing(self):
"""When ssrf module is not available, skip check and proceed."""
def test_execute_ssrf_check_is_not_bypassable(self):
"""The SSRF check must never be silently skipped (#225/#467).
Previously browser navigation wrapped the SSRF check in
``except ImportError: pass``, so an environment where the check
couldn't be imported silently disabled SSRF protection. ``check_ssrf``
now has a pure-Python fallback and is always reachable, so the import
no longer fails — and the guard must not be bypassed. Here a real
blocked URL is rejected even though navigation would otherwise
succeed, proving the check runs unconditionally.
"""
from openjarvis.tools.browser import BrowserNavigateTool
mock_ssrf_module = MagicMock()
mock_ssrf_module.check_ssrf.return_value = (
"URL resolves to private IP: 127.0.0.1"
)
page = _make_mock_page()
session = _make_mock_session(page)
# Make the ssrf import fail inside execute
import builtins
with patch("openjarvis.tools.browser._session", session):
with patch.dict(
"sys.modules",
{"openjarvis.security.ssrf": mock_ssrf_module},
):
tool = BrowserNavigateTool()
result = tool.execute(url="http://127.0.0.1:8080/admin")
original_import = builtins.__import__
# Blocked — the check ran and was honored, not skipped.
assert result.success is False
assert "SSRF blocked" in result.content
def _mock_import(name, *args, **kwargs):
if name == "openjarvis.security.ssrf":
raise ImportError("No module named 'openjarvis.security.ssrf'")
return original_import(name, *args, **kwargs)
def test_execute_allows_public_url(self):
"""The always-on SSRF check must not break legitimate navigation."""
from openjarvis.tools.browser import BrowserNavigateTool
mock_ssrf_module = MagicMock()
mock_ssrf_module.check_ssrf.return_value = None # public URL, allowed
page = _make_mock_page()
session = _make_mock_session(page)
with patch("openjarvis.tools.browser._session", session):
with patch.object(builtins, "__import__", side_effect=_mock_import):
with patch.dict(
"sys.modules",
{"openjarvis.security.ssrf": mock_ssrf_module},
):
tool = BrowserNavigateTool()
result = tool.execute(url="https://example.com")
# Should succeed since SSRF check is skipped
assert result.success is True
def test_execute_success(self):