133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from threading import Thread
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import httpcore
|
|
|
|
from govoplan_core.security.outbound_http import outbound_http_policy, validate_outbound_http_url
|
|
from govoplan_files.backend.storage.http_client import (
|
|
ConnectorHttpError,
|
|
_OutboundPolicyNetworkBackend,
|
|
_SocketNetworkStream,
|
|
request_connector_bytes,
|
|
)
|
|
|
|
|
|
class _TestHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802 - stdlib handler contract
|
|
body = b"connector-ok"
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
|
|
class ConnectorHttpClientTests(unittest.TestCase):
|
|
def test_public_transport_adapter_performs_a_real_http_request(self) -> None:
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), _TestHandler)
|
|
thread = Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"APP_ENV": "test", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
|
|
clear=False,
|
|
):
|
|
result = request_connector_bytes(
|
|
"GET",
|
|
f"http://127.0.0.1:{server.server_port}/object",
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
self.assertEqual(200, result.status_code)
|
|
self.assertEqual(b"connector-ok", result.content)
|
|
|
|
def test_connector_responses_are_streamed_without_redirects(self) -> None:
|
|
response = MagicMock()
|
|
response.status_code = 200
|
|
response.headers = {"content-length": "6"}
|
|
response.iter_bytes.return_value = iter((b"abc", b"def"))
|
|
context = MagicMock()
|
|
context.__enter__.return_value = response
|
|
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("93.184.216.34", 443))],
|
|
), patch("govoplan_files.backend.storage.http_client._stream_connector_request", return_value=context):
|
|
result = request_connector_bytes("GET", "https://example.test/object", max_bytes=10)
|
|
|
|
self.assertEqual(b"abcdef", result.content)
|
|
|
|
def test_connector_response_limit_is_enforced_while_streaming(self) -> None:
|
|
response = MagicMock()
|
|
response.status_code = 200
|
|
response.headers = {}
|
|
response.iter_bytes.return_value = iter((b"12345", b"67890", b"!"))
|
|
context = MagicMock()
|
|
context.__enter__.return_value = response
|
|
|
|
with patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("93.184.216.34", 443))],
|
|
), patch(
|
|
"govoplan_files.backend.storage.http_client._stream_connector_request",
|
|
return_value=context,
|
|
), self.assertRaisesRegex(
|
|
ConnectorHttpError,
|
|
"configured limit",
|
|
):
|
|
request_connector_bytes("GET", "https://example.test/object", max_bytes=10)
|
|
|
|
def test_private_destination_is_blocked_before_http_connection(self) -> None:
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
|
|
), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
return_value=[(2, 1, 6, "", ("10.0.0.5", 443))],
|
|
), patch("govoplan_files.backend.storage.http_client._stream_connector_request") as stream, self.assertRaisesRegex(
|
|
ConnectorHttpError,
|
|
"non-public network",
|
|
):
|
|
request_connector_bytes("GET", "https://connector.example.test/object")
|
|
stream.assert_not_called()
|
|
|
|
def test_connection_backend_rejects_private_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.dict(
|
|
"os.environ",
|
|
{"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
|
|
), patch(
|
|
"govoplan_core.security.outbound_http.socket.getaddrinfo",
|
|
side_effect=(public, private),
|
|
), patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory:
|
|
validate_outbound_http_url("https://connector.example.test/object", policy=policy)
|
|
with self.assertRaises(httpcore.ConnectError):
|
|
_OutboundPolicyNetworkBackend().connect_tcp("connector.example.test", 443)
|
|
socket_factory.assert_not_called()
|
|
|
|
def test_network_backend_returns_public_network_stream_adapter(self) -> None:
|
|
sock = MagicMock()
|
|
with patch(
|
|
"govoplan_files.backend.storage.http_client.create_outbound_connection",
|
|
return_value=sock,
|
|
):
|
|
stream = _OutboundPolicyNetworkBackend().connect_tcp("connector.example.test", 443)
|
|
|
|
self.assertIsInstance(stream, _SocketNetworkStream)
|
|
self.assertIs(stream.get_extra_info("socket"), sock)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|