144 lines
4.1 KiB
Python
144 lines
4.1 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
import pytest
|
|
|
|
from devplacepy.services.nodes.policy import DestinationPolicy, PolicyError, is_public_address
|
|
from devplacepy.services.nodes.spec import SPEC
|
|
|
|
POLICY = DestinationPolicy()
|
|
|
|
|
|
def code(url, method="GET", mode="any"):
|
|
with pytest.raises(PolicyError) as info:
|
|
POLICY.check(url, method, mode)
|
|
return info.value.code
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"https://reddit.com/r/python.json",
|
|
"https://www.reddit.com/r/python/top.json?t=week",
|
|
"http://old.reddit.com/",
|
|
"https://REDDIT.COM/",
|
|
"https://reddit.com./r/x",
|
|
"https://redd.it/abc",
|
|
"https://reddit.com:443/x",
|
|
"http://reddit.com:80/x",
|
|
],
|
|
)
|
|
def test_allowlisted_urls_pass(url):
|
|
target = POLICY.check(url)
|
|
assert target.method == "GET" and target.adapter == "reddit"
|
|
assert target.port in (80, 443)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"https://evilreddit.com/",
|
|
"https://reddit.com.evil.net/",
|
|
"https://notreddit.com/",
|
|
"https://example.com/",
|
|
"https://localhost/",
|
|
"https://reddit.evil.com/",
|
|
"https://evil.com/reddit.com",
|
|
],
|
|
)
|
|
def test_other_hosts_are_refused(url):
|
|
assert code(url) == "host_not_allowed"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"https://127.0.0.1/",
|
|
"https://[::1]/",
|
|
"http://10.0.0.5/",
|
|
"https://169.254.169.254/latest/meta-data",
|
|
"https://0x7f.1/",
|
|
"https://2130706433/",
|
|
],
|
|
)
|
|
def test_address_literals_are_refused(url):
|
|
assert code(url) in ("host_not_allowed", "bad_url")
|
|
|
|
|
|
@pytest.mark.parametrize("port", [22, 25, 8080, 8443, 1, 65535])
|
|
def test_only_web_ports(port):
|
|
assert code(f"https://reddit.com:{port}/") == "port_not_allowed"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
["ftp://reddit.com/", "file:///etc/passwd", "javascript:alert(1)", "//reddit.com/x", "reddit.com/x", ""],
|
|
)
|
|
def test_other_schemes_and_shapes_are_bad_urls(url):
|
|
assert code(url) == "bad_url"
|
|
|
|
|
|
def test_credentials_in_the_url_are_refused():
|
|
assert code("https://user:pw@reddit.com/") == "bad_url"
|
|
assert code("https://user@reddit.com/") == "bad_url"
|
|
assert code("https://evil.com@reddit.com.evil.com/") == "bad_url"
|
|
assert code("https://reddit.com@evil.com/") == "bad_url"
|
|
|
|
|
|
def test_overlong_and_non_string_urls_are_refused():
|
|
assert code("https://reddit.com/" + "a" * SPEC.limits.url_max_chars) == "bad_url"
|
|
with pytest.raises(PolicyError):
|
|
POLICY.check(None)
|
|
with pytest.raises(PolicyError):
|
|
POLICY.check(["https://reddit.com/"])
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "PATCH", "CONNECT", ""])
|
|
def test_only_declared_methods(method):
|
|
assert code("https://reddit.com/", method) == "method_not_allowed"
|
|
|
|
|
|
def test_method_is_case_insensitive():
|
|
assert POLICY.check("https://reddit.com/", "get").method == "GET"
|
|
|
|
|
|
def test_unknown_mode_is_refused():
|
|
assert code("https://reddit.com/", mode="raw") == "bad_mode"
|
|
|
|
|
|
def test_a_redirect_off_the_allowlist_is_a_redirect_refusal():
|
|
with pytest.raises(PolicyError) as info:
|
|
POLICY.check_redirect("https://evil.example/")
|
|
assert info.value.code == "redirect_not_allowed"
|
|
assert POLICY.check_redirect("https://www.reddit.com/new").host == "www.reddit.com"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"address,expected",
|
|
[
|
|
("8.8.8.8", True),
|
|
("1.1.1.1", True),
|
|
("2606:4700:4700::1111", True),
|
|
("127.0.0.1", False),
|
|
("10.1.2.3", False),
|
|
("172.16.0.1", False),
|
|
("192.168.1.1", False),
|
|
("169.254.1.1", False),
|
|
("100.64.0.1", False),
|
|
("100.127.255.254", False),
|
|
("224.0.0.1", False),
|
|
("0.0.0.0", False),
|
|
("::1", False),
|
|
("fe80::1", False),
|
|
("fc00::1", False),
|
|
("not an ip", False),
|
|
],
|
|
)
|
|
def test_public_address_classification(address, expected):
|
|
assert is_public_address(address) is expected
|
|
|
|
|
|
def test_every_destination_in_the_spec_passes_its_own_policy():
|
|
for destination in SPEC.destinations:
|
|
target = POLICY.check(f"https://{destination.suffix}/x")
|
|
assert target.destination is destination
|