mirror of
https://github.com/Skyvern-AI/skyvern.git
synced 2026-07-09 16:09:13 +00:00
fix(SKY-9322): normalize bracketed IPv6 hosts in is_blocked_host (#5759)
This commit is contained in:
parent
4e5d160167
commit
bb345c4ce9
2 changed files with 116 additions and 11 deletions
|
|
@ -46,21 +46,32 @@ def prepend_scheme_and_validate_url(url: str) -> str:
|
|||
|
||||
|
||||
def is_blocked_host(host: str) -> bool:
|
||||
if host.lower() in (h.lower() for h in settings.ALLOWED_HOSTS):
|
||||
return False
|
||||
# RFC 3986 wraps IPv6 literals in [...]; ip_address() only accepts the bare form.
|
||||
bare = host[1:-1] if host.startswith("[") and host.endswith("]") else host
|
||||
ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None
|
||||
try:
|
||||
ip = ipaddress.ip_address(host)
|
||||
# Check if the IP is private, link-local, loopback, or reserved
|
||||
return ip.is_private or ip.is_link_local or ip.is_loopback or ip.is_reserved
|
||||
ip = ipaddress.ip_address(bare)
|
||||
if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
|
||||
ip = ip.ipv4_mapped
|
||||
except ValueError:
|
||||
# If the host is not a valid IP address (e.g., it's a domain name like localhost), handle it here
|
||||
for blocked_host in settings.BLOCKED_HOSTS:
|
||||
if blocked_host == host:
|
||||
return True
|
||||
return False
|
||||
ip = None
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
candidate_forms = {host.lower(), bare.lower()}
|
||||
if ip is not None:
|
||||
candidate_forms.add(str(ip).lower())
|
||||
|
||||
allowed = {h.lower() for h in settings.ALLOWED_HOSTS}
|
||||
if candidate_forms & allowed:
|
||||
return False
|
||||
|
||||
if ip is not None:
|
||||
return ip.is_private or ip.is_link_local or ip.is_loopback or ip.is_reserved
|
||||
|
||||
blocked = {b.lower() for b in settings.BLOCKED_HOSTS}
|
||||
return host.lower() in blocked
|
||||
|
||||
|
||||
def validate_url(url: str) -> str | None:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,4 +1,8 @@
|
|||
from skyvern.utils.url_validators import encode_url
|
||||
import pytest
|
||||
|
||||
from skyvern.config import settings
|
||||
from skyvern.exceptions import BlockedHost
|
||||
from skyvern.utils.url_validators import encode_url, is_blocked_host, validate_url
|
||||
|
||||
|
||||
def test_encode_url_basic():
|
||||
|
|
@ -27,3 +31,93 @@ def test_encode_url_with_pre_encoded_chars():
|
|||
url = "https://example.com/search?q=hello world&type=test%20test"
|
||||
expected = "https://example.com/search?q=hello%20world&type=test%20test"
|
||||
assert encode_url(url) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"[::1]",
|
||||
"[::ffff:127.0.0.1]",
|
||||
"[::ffff:7f00:1]",
|
||||
"[::ffff:169.254.169.254]",
|
||||
"[::ffff:a9fe:a9fe]",
|
||||
"[::ffff:10.0.0.1]",
|
||||
"[::ffff:192.168.1.1]",
|
||||
"[fe80::1]",
|
||||
"[fc00::1]",
|
||||
],
|
||||
)
|
||||
def test_is_blocked_host_bracketed_ipv6_internal(host: str) -> None:
|
||||
assert is_blocked_host(host) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"::1",
|
||||
"::ffff:127.0.0.1",
|
||||
"::ffff:169.254.169.254",
|
||||
"fe80::1",
|
||||
"fc00::1",
|
||||
"10.0.0.1",
|
||||
"127.0.0.1",
|
||||
"169.254.169.254",
|
||||
"192.168.1.1",
|
||||
"localhost",
|
||||
],
|
||||
)
|
||||
def test_is_blocked_host_unbracketed_internal(host: str) -> None:
|
||||
assert is_blocked_host(host) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"[2001:4860:4860::8888]",
|
||||
"2001:4860:4860::8888",
|
||||
"8.8.8.8",
|
||||
"example.com",
|
||||
],
|
||||
)
|
||||
def test_is_blocked_host_public_allowed(host: str) -> None:
|
||||
assert is_blocked_host(host) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"https://[::1]/",
|
||||
"https://[::ffff:127.0.0.1]/",
|
||||
"https://[::ffff:169.254.169.254]/admin",
|
||||
"https://[fc00::1]/internal",
|
||||
],
|
||||
)
|
||||
def test_validate_url_rejects_bracketed_ipv6_internal(url: str) -> None:
|
||||
with pytest.raises(BlockedHost):
|
||||
validate_url(url)
|
||||
|
||||
|
||||
def test_validate_url_allows_public_ipv6() -> None:
|
||||
assert validate_url("https://[2001:4860:4860::8888]/") is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("allowed_entry", "host"),
|
||||
[
|
||||
("::1", "[::1]"),
|
||||
("[::1]", "[::1]"),
|
||||
("127.0.0.1", "[::ffff:127.0.0.1]"),
|
||||
("127.0.0.1", "[::ffff:7f00:1]"),
|
||||
("FC00::1", "[fc00::1]"),
|
||||
],
|
||||
)
|
||||
def test_is_blocked_host_allowed_hosts_normalize_brackets_and_mapped(
|
||||
monkeypatch: pytest.MonkeyPatch, allowed_entry: str, host: str
|
||||
) -> None:
|
||||
monkeypatch.setattr(settings, "ALLOWED_HOSTS", [allowed_entry])
|
||||
assert is_blocked_host(host) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("host", ["LOCALHOST", "LocalHost", "localhost"])
|
||||
def test_is_blocked_host_blocked_hosts_case_insensitive(host: str) -> None:
|
||||
assert is_blocked_host(host) is True
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue