311 lines
13 KiB
Python
311 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from govoplan_core.security.http_fetch import (
|
|
_PolicyRedirectHandler,
|
|
fetch_http,
|
|
is_http_url,
|
|
validate_http_url,
|
|
)
|
|
from govoplan_core.security.outbound_http import (
|
|
DEFAULT_FILE_TRANSFER_BYTES,
|
|
DEFAULT_STRUCTURED_RESPONSE_BYTES,
|
|
OutboundHttpBlocked,
|
|
OutboundResponseTooLarge,
|
|
bounded_chunks_bytes,
|
|
bounded_response_bytes,
|
|
create_outbound_connection,
|
|
outbound_http_policy,
|
|
validate_outbound_http_url,
|
|
validate_unpinned_sdk_host,
|
|
validate_unpinned_sdk_http_url,
|
|
)
|
|
|
|
|
|
class HttpFetchTests(unittest.TestCase):
|
|
def test_fetch_http_forwards_a_bounded_request_body(self) -> None:
|
|
class Response(io.BytesIO):
|
|
status = 200
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
opener = Mock()
|
|
opener.open.return_value = Response(b"{}")
|
|
with patch(
|
|
"govoplan_core.security.http_fetch.validate_outbound_http_url",
|
|
return_value="https://wiki.example.test/api.php",
|
|
), patch(
|
|
"govoplan_core.security.http_fetch.build_outbound_http_opener",
|
|
return_value=opener,
|
|
):
|
|
response = fetch_http(
|
|
"https://wiki.example.test/api.php",
|
|
method="POST",
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
body=b"action=edit",
|
|
max_bytes=1024,
|
|
)
|
|
|
|
request = opener.open.call_args.args[0]
|
|
self.assertEqual("POST", request.get_method())
|
|
self.assertEqual(b"action=edit", request.data)
|
|
self.assertEqual(b"{}", response.body)
|
|
|
|
def test_fetch_http_rejects_an_oversized_request_body_before_transport(self) -> None:
|
|
with patch(
|
|
"govoplan_core.security.http_fetch.validate_outbound_http_url"
|
|
) as validate:
|
|
with self.assertRaisesRegex(ValueError, "request body exceeds"):
|
|
fetch_http(
|
|
"https://wiki.example.test/api.php",
|
|
method="POST",
|
|
body=b"x" * 1_000_001,
|
|
)
|
|
validate.assert_not_called()
|
|
|
|
def test_validate_http_url_accepts_absolute_http_urls_without_credentials(self) -> None:
|
|
self.assertEqual("https://example.test/catalog.json", validate_http_url("https://example.test/catalog.json"))
|
|
self.assertTrue(is_http_url("http://example.test/catalog.json"))
|
|
|
|
def test_validate_http_url_rejects_non_http_urls_and_embedded_credentials(self) -> None:
|
|
for value in (
|
|
"file:///etc/passwd",
|
|
"/relative/catalog.json",
|
|
"https://user@example.test/catalog.json",
|
|
"https://user:secret@example.test/catalog.json",
|
|
):
|
|
with self.subTest(value=value):
|
|
self.assertFalse(is_http_url(value))
|
|
with self.assertRaises(ValueError):
|
|
validate_http_url(value)
|
|
|
|
def test_outbound_policy_has_practical_bounded_defaults(self) -> None:
|
|
policy = outbound_http_policy({"APP_ENV": "production"})
|
|
self.assertFalse(policy.allow_private_networks)
|
|
self.assertEqual(DEFAULT_STRUCTURED_RESPONSE_BYTES, policy.structured_response_bytes)
|
|
self.assertEqual(DEFAULT_FILE_TRANSFER_BYTES, policy.file_transfer_bytes)
|
|
|
|
configured = outbound_http_policy(
|
|
{
|
|
"APP_ENV": "production",
|
|
"GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true",
|
|
"GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "1024",
|
|
"GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES": "2048",
|
|
}
|
|
)
|
|
self.assertTrue(configured.allow_private_networks)
|
|
self.assertEqual(1024, configured.structured_response_bytes)
|
|
self.assertEqual(2048, configured.file_transfer_bytes)
|
|
|
|
def test_private_network_policy_is_deployment_wide(self) -> None:
|
|
denied_policy = outbound_http_policy({"APP_ENV": "production"})
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
|
), self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"):
|
|
validate_outbound_http_url("https://connector.example.test/path", policy=denied_policy)
|
|
|
|
allowed_policy = outbound_http_policy(
|
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
|
|
)
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
|
):
|
|
self.assertEqual(
|
|
"https://127.0.0.1/path",
|
|
validate_outbound_http_url("https://127.0.0.1/path", policy=allowed_policy),
|
|
)
|
|
|
|
def test_private_network_policy_still_blocks_non_peer_and_metadata_addresses(self) -> None:
|
|
policy = outbound_http_policy(
|
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
|
|
)
|
|
for address in (
|
|
"0.0.0.0",
|
|
"169.254.169.254",
|
|
"224.0.0.1",
|
|
"255.255.255.255",
|
|
"::",
|
|
"::ffff:255.255.255.255",
|
|
"64:ff9b::169.254.169.254",
|
|
"64:ff9b::255.255.255.255",
|
|
"fd00:ec2::254",
|
|
"fe80::1",
|
|
"ff02::1",
|
|
):
|
|
with self.subTest(address=address), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", (address, 443))],
|
|
), self.assertRaisesRegex(OutboundHttpBlocked, "forbidden special-purpose network"):
|
|
validate_outbound_http_url("https://connector.example.test/path", policy=policy)
|
|
|
|
def test_public_policy_rejects_reserved_and_ipv4_embedded_ipv6_addresses(self) -> None:
|
|
policy = outbound_http_policy({"APP_ENV": "production"})
|
|
for address in (
|
|
"fec0::1",
|
|
"::127.0.0.1",
|
|
"64:ff9b::8.8.8.8",
|
|
"64:ff9b::127.0.0.1",
|
|
"2002:7f00:1::",
|
|
):
|
|
with self.subTest(address=address), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(10, 1, 6, "", (address, 443, 0, 0))],
|
|
), self.assertRaises(OutboundHttpBlocked):
|
|
validate_outbound_http_url("https://connector.example.test/path", policy=policy)
|
|
|
|
def test_outbound_policy_rejects_mixed_public_and_private_dns_answers(self) -> None:
|
|
policy = outbound_http_policy({"APP_ENV": "production"})
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[
|
|
(2, 1, 6, "", ("93.184.216.34", 443)),
|
|
(2, 1, 6, "", ("10.0.0.4", 443)),
|
|
],
|
|
), self.assertRaises(OutboundHttpBlocked):
|
|
validate_outbound_http_url("https://connector.example.test/path", policy=policy)
|
|
|
|
def test_bounded_response_rejects_declared_and_streamed_oversize_payloads(self) -> None:
|
|
with self.assertRaises(OutboundResponseTooLarge):
|
|
bounded_response_bytes(io.BytesIO(b"small"), headers={"Content-Length": "20"}, max_bytes=10)
|
|
with self.assertRaises(OutboundResponseTooLarge):
|
|
bounded_response_bytes(io.BytesIO(b"eleven-byte"), max_bytes=10)
|
|
self.assertEqual(b"ten-bytes!", bounded_response_bytes(io.BytesIO(b"ten-bytes!"), max_bytes=10))
|
|
with patch.dict("os.environ", {"GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "5"}), self.assertRaises(
|
|
OutboundResponseTooLarge
|
|
):
|
|
bounded_response_bytes(io.BytesIO(b"123456"), max_bytes=10)
|
|
|
|
def test_bounded_chunks_rejects_an_oversized_chunk_before_retaining_it(self) -> None:
|
|
with self.assertRaises(OutboundResponseTooLarge):
|
|
bounded_chunks_bytes((b"one-large-chunk",), max_bytes=8)
|
|
|
|
def test_connection_time_resolution_blocks_dns_rebinding_before_socket_open(self) -> None:
|
|
policy = outbound_http_policy({"APP_ENV": "production"})
|
|
public = [(2, 1, 6, "", ("93.184.216.34", 443))]
|
|
private = [(2, 1, 6, "", ("127.0.0.1", 443))]
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
side_effect=(public, private),
|
|
) as resolver, patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory:
|
|
validate_outbound_http_url("https://connector.example.test/path", policy=policy)
|
|
with self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"):
|
|
create_outbound_connection("connector.example.test", 443, policy=policy)
|
|
|
|
self.assertEqual(2, resolver.call_count)
|
|
socket_factory.assert_not_called()
|
|
|
|
def test_unpinned_sdk_endpoints_fail_closed_in_public_and_private_modes(self) -> None:
|
|
for allow_private, address in ((False, "93.184.216.34"), (True, "10.0.0.5")):
|
|
with self.subTest(allow_private=allow_private):
|
|
policy = outbound_http_policy(
|
|
{
|
|
"APP_ENV": "production",
|
|
"GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": str(allow_private).lower(),
|
|
}
|
|
)
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", (address, 443))],
|
|
), self.assertRaisesRegex(OutboundHttpBlocked, "until that transport supports.*DNS/IP pinning"):
|
|
validate_unpinned_sdk_http_url(
|
|
"https://objects.example.test",
|
|
label="S3 endpoint",
|
|
policy=policy,
|
|
)
|
|
|
|
def test_unpinned_host_sdk_fails_closed_for_dns_and_explicit_ip_targets(self) -> None:
|
|
policy = outbound_http_policy(
|
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"}
|
|
)
|
|
for host in ("files.internal.example", "10.0.0.5"):
|
|
with self.subTest(host=host), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
|
|
), self.assertRaisesRegex(OutboundHttpBlocked, "redirects/referrals.*DNS/IP pinning"):
|
|
validate_unpinned_sdk_host(host, port=445, label="SMB endpoint", policy=policy)
|
|
|
|
def test_core_redirects_strip_credentials_cross_origin_and_reject_https_downgrades(self) -> None:
|
|
import urllib.request
|
|
|
|
request = urllib.request.Request(
|
|
"https://catalog.example.test/releases",
|
|
headers={
|
|
"Authorization": "Bearer secret",
|
|
"Cookie": "session=secret",
|
|
"X-OTRS-Header-Password": "secret",
|
|
"X-Request-ID": "request-1",
|
|
},
|
|
)
|
|
handler = _PolicyRedirectHandler(
|
|
label="Catalog URL",
|
|
sensitive_headers=("X-OTRS-Header-Password",),
|
|
)
|
|
with patch.dict("os.environ", {"APP_ENV": "test"}), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
|
):
|
|
redirected = handler.redirect_request(
|
|
request,
|
|
None,
|
|
302,
|
|
"Found",
|
|
{},
|
|
"https://cdn.example.test/releases",
|
|
)
|
|
downgrade = handler.redirect_request(
|
|
request,
|
|
None,
|
|
302,
|
|
"Found",
|
|
{},
|
|
"http://catalog.example.test/releases",
|
|
)
|
|
|
|
self.assertIsNotNone(redirected)
|
|
self.assertIsNone(redirected.get_header("Authorization"))
|
|
self.assertIsNone(redirected.get_header("Cookie"))
|
|
self.assertIsNone(redirected.get_header("X-otrs-header-password"))
|
|
self.assertEqual("request-1", redirected.get_header("X-request-id"))
|
|
self.assertIsNone(downgrade)
|
|
|
|
def test_core_redirects_preserve_caller_sensitive_headers_on_the_same_origin(self) -> None:
|
|
import urllib.request
|
|
|
|
request = urllib.request.Request(
|
|
"https://desk.example.test/original",
|
|
headers={"X-OTRS-Header-SessionID": "secret"},
|
|
)
|
|
handler = _PolicyRedirectHandler(
|
|
label="Service-desk URL",
|
|
sensitive_headers=("X-OTRS-Header-SessionID",),
|
|
)
|
|
with patch.dict("os.environ", {"APP_ENV": "test"}), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("127.0.0.1", 443))],
|
|
):
|
|
redirected = handler.redirect_request(
|
|
request,
|
|
None,
|
|
302,
|
|
"Found",
|
|
{},
|
|
"https://desk.example.test/final",
|
|
)
|
|
|
|
self.assertIsNotNone(redirected)
|
|
self.assertEqual("secret", redirected.get_header("X-otrs-header-sessionid"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|