mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-28 14:07:55 +00:00
2b14315f3c
* 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>
255 lines
10 KiB
Python
255 lines
10 KiB
Python
"""Tests for SSRF protection module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
from openjarvis.security.ssrf import _check_ssrf_python, check_ssrf, is_private_ip
|
|
|
|
|
|
class TestIsPrivateIp:
|
|
def test_private_10_network(self):
|
|
assert is_private_ip("10.0.0.1") is True
|
|
assert is_private_ip("10.255.255.255") is True
|
|
|
|
def test_private_172_16_network(self):
|
|
assert is_private_ip("172.16.0.1") is True
|
|
assert is_private_ip("172.31.255.255") is True
|
|
|
|
def test_private_192_168_network(self):
|
|
assert is_private_ip("192.168.0.1") is True
|
|
assert is_private_ip("192.168.1.100") is True
|
|
|
|
def test_loopback(self):
|
|
assert is_private_ip("127.0.0.1") is True
|
|
assert is_private_ip("127.255.255.255") is True
|
|
|
|
def test_ipv6_loopback(self):
|
|
assert is_private_ip("::1") is True
|
|
|
|
def test_link_local(self):
|
|
assert is_private_ip("169.254.0.1") is True
|
|
|
|
def test_public_ips(self):
|
|
assert is_private_ip("8.8.8.8") is False
|
|
assert is_private_ip("1.1.1.1") is False
|
|
assert is_private_ip("93.184.216.34") is False
|
|
|
|
def test_invalid_ip(self):
|
|
assert is_private_ip("not-an-ip") is False
|
|
|
|
def test_empty_string(self):
|
|
assert is_private_ip("") is False
|
|
|
|
def test_ipv4_mapped_ipv6_loopback(self):
|
|
# ::ffff:127.0.0.1 — IPv4-mapped form of loopback must be flagged.
|
|
assert is_private_ip("::ffff:127.0.0.1") is True
|
|
|
|
def test_ipv4_mapped_ipv6_rfc1918(self):
|
|
assert is_private_ip("::ffff:10.0.0.1") is True
|
|
assert is_private_ip("::ffff:172.16.0.1") is True
|
|
assert is_private_ip("::ffff:192.168.1.1") is True
|
|
|
|
def test_ipv4_mapped_ipv6_link_local(self):
|
|
assert is_private_ip("::ffff:169.254.0.1") is True
|
|
|
|
def test_ipv4_mapped_ipv6_public_allowed(self):
|
|
# Public IPv4 wrapped as IPv6 must NOT be flagged as private.
|
|
assert is_private_ip("::ffff:8.8.8.8") is False
|
|
|
|
def test_ipv4_compatible_loopback(self):
|
|
# ::127.0.0.1 — deprecated IPv4-compatible form, must still be blocked.
|
|
assert is_private_ip("::127.0.0.1") is True
|
|
|
|
def test_unspecified_addresses(self):
|
|
assert is_private_ip("0.0.0.0") is True
|
|
assert is_private_ip("::") is True
|
|
|
|
def test_zero_subnet_is_private(self):
|
|
# 0.0.0.0/8 routes to localhost on Linux.
|
|
assert is_private_ip("0.1.2.3") is True
|
|
|
|
def test_multicast_and_broadcast_blocked(self):
|
|
assert is_private_ip("239.0.0.1") is True
|
|
assert is_private_ip("255.255.255.255") is True
|
|
assert is_private_ip("ff02::1") is True
|
|
|
|
|
|
class TestCheckSsrf:
|
|
"""Tests for SSRF protection.
|
|
|
|
The Rust backend performs real DNS resolution, so tests that need to
|
|
mock DNS use ``_check_ssrf_python`` (the pure-Python implementation)
|
|
instead of the Rust-backed ``check_ssrf``.
|
|
"""
|
|
|
|
def test_blocks_aws_metadata(self):
|
|
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_blocks_google_metadata(self):
|
|
result = check_ssrf("http://metadata.google.internal/computeMetadata/v1/")
|
|
assert result is not None
|
|
assert "Blocked host" in result
|
|
|
|
def test_blocks_alibaba_metadata(self):
|
|
result = check_ssrf("http://100.100.100.200/latest/meta-data/")
|
|
assert result is not None
|
|
assert "Blocked host" in result
|
|
|
|
def test_allows_normal_urls(self):
|
|
# Use Python impl so we can mock DNS resolution
|
|
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_python("https://example.com")
|
|
assert result is None
|
|
|
|
def test_blocks_localhost_url(self):
|
|
with patch("openjarvis.security.ssrf.socket.getaddrinfo") as mock_dns:
|
|
mock_dns.return_value = [
|
|
(2, 1, 6, "", ("127.0.0.1", 0)),
|
|
]
|
|
result = _check_ssrf_python("http://localhost:8080/admin")
|
|
assert result is not None
|
|
assert "private IP" in result
|
|
|
|
def test_blocks_private_ip_url(self):
|
|
with patch("openjarvis.security.ssrf.socket.getaddrinfo") as mock_dns:
|
|
mock_dns.return_value = [
|
|
(2, 1, 6, "", ("192.168.1.1", 0)),
|
|
]
|
|
result = _check_ssrf_python("http://internal-service.local/api")
|
|
assert result is not None
|
|
assert "private IP" in result
|
|
|
|
def test_no_hostname(self):
|
|
# 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 or "No hostname" in result
|
|
|
|
def test_dns_failure_allowed(self):
|
|
"""DNS resolution failure should not block — request will fail at HTTP time."""
|
|
import socket
|
|
|
|
with patch(
|
|
"openjarvis.security.ssrf.socket.getaddrinfo",
|
|
side_effect=socket.gaierror("Name resolution failed"),
|
|
):
|
|
result = _check_ssrf_python("https://nonexistent.example.com")
|
|
assert result is None
|
|
|
|
def test_blocks_dns_rebinding_to_private(self):
|
|
"""Even if hostname looks normal, block if it resolves to private IP."""
|
|
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_python("https://evil-rebind.example.com")
|
|
assert result is not None
|
|
assert "private IP" in result
|
|
|
|
def test_blocks_ipv4_mapped_loopback_url(self):
|
|
"""IPv4-mapped IPv6 form of 127.0.0.1 must be blocked (Rust path)."""
|
|
result = check_ssrf("http://[::ffff:127.0.0.1]:6666")
|
|
assert result is not None
|
|
assert "private IP" in result
|
|
|
|
def test_blocks_ipv4_mapped_rfc1918_url(self):
|
|
for url in (
|
|
"http://[::ffff:10.0.0.1]/",
|
|
"http://[::ffff:192.168.1.1]/",
|
|
"http://[::ffff:172.16.0.1]/",
|
|
):
|
|
result = check_ssrf(url)
|
|
assert result is not None, f"{url} must be blocked"
|
|
assert "private IP" in result
|
|
|
|
def test_blocks_ipv4_mapped_metadata_url(self):
|
|
# ::ffff:169.254.169.254 — IPv4-mapped form of cloud metadata IP.
|
|
result = check_ssrf("http://[::ffff:169.254.169.254]/latest/meta-data/")
|
|
assert result is not None
|
|
|
|
def test_blocks_ipv6_loopback_literal_url(self):
|
|
result = check_ssrf("http://[::1]:80/")
|
|
assert result is not None
|
|
|
|
def test_allows_ipv4_mapped_public_url(self):
|
|
result = check_ssrf("http://[::ffff:8.8.8.8]/")
|
|
assert result is None
|
|
|
|
def test_python_impl_blocks_ipv4_mapped_loopback(self):
|
|
"""Legacy Python impl must also catch ::ffff:127.0.0.1."""
|
|
result = _check_ssrf_python("http://[::ffff:127.0.0.1]:6666")
|
|
assert result is not None
|
|
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/")
|
|
assert result is not None
|
|
|
|
def test_blocks_ipv4_mapped_alibaba_metadata(self):
|
|
# 100.100.100.200 isn't private — relies on BLOCKED_HOSTS lookup
|
|
# against the embedded v4. Verifies the canonical-host fix.
|
|
result = check_ssrf("http://[::ffff:100.100.100.200]/")
|
|
assert result is not None
|
|
assert "Blocked host" in result
|
|
|
|
def test_blocks_ipv4_compatible_loopback(self):
|
|
result = check_ssrf("http://[::127.0.0.1]/")
|
|
assert result is not None
|
|
|
|
def test_blocks_zero_dot_zero(self):
|
|
result = check_ssrf("http://0.0.0.0/")
|
|
assert result is not None
|
|
|
|
def test_url_userinfo_does_not_bypass(self):
|
|
# Userinfo + bracketed IPv6 literal still gets blocked.
|
|
result = check_ssrf("http://user:pass@[::ffff:127.0.0.1]:8080/admin")
|
|
assert result is not None
|
|
|
|
def test_canonicalized_hex_ipv6_form_blocked(self):
|
|
# url crate canonicalizes [::ffff:127.0.0.1] to [::ffff:7f00:1]
|
|
# internally; ensure this form blocks too.
|
|
result = check_ssrf("http://[::ffff:7f00:1]/")
|
|
assert result is not None
|
|
|
|
|
|
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"]
|