scrapy/tests/test_http2_client_protocol.py

799 lines
28 KiB
Python

from __future__ import annotations
import json
import random
import re
import string
from ipaddress import IPv4Address
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from unittest import mock
from urllib.parse import urlencode
import pytest
from pytest_twisted import async_yield_fixture
from twisted.internet.defer import Deferred, DeferredList, inlineCallbacks
from twisted.internet.endpoints import SSL4ClientEndpoint, SSL4ServerEndpoint
from twisted.internet.ssl import Certificate, PrivateCertificate, optionsForClientTLS
from twisted.internet.task import deferLater
from twisted.web.client import URI, ResponseFailed
from twisted.web.http import H2_ENABLED
from twisted.web.http import Request as TxRequest
from twisted.web.server import NOT_DONE_YET, Site
from twisted.web.static import File
from scrapy.exceptions import DownloadCancelledError, DownloadTimeoutError
from scrapy.http import JsonRequest, Request, Response
from scrapy.spiders import Spider
from scrapy.utils.defer import (
deferred_f_from_coro_f,
deferred_from_coro,
maybe_deferred_to_future,
)
from scrapy.utils.test import get_crawler
from tests.mockserver.http_resources import LeafResource, Status, put_child
from tests.mockserver.utils import ssl_context_factory
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Callable, Coroutine, Generator
from scrapy.core._http2.protocol import H2ClientProtocol
from scrapy.crawler import Crawler
pytestmark = [
pytest.mark.requires_reactor, # H2ClientProtocol requires a reactor
pytest.mark.skipif(
not H2_ENABLED, reason="HTTP/2 support in Twisted is not enabled"
),
]
def generate_random_string(size: int) -> str:
return "".join(random.choices(string.ascii_uppercase + string.digits, k=size))
def make_html_body(val: str) -> bytes:
response = f"""<html>
<h1>Hello from HTTP2<h1>
<p>{val}</p>
</html>"""
return bytes(response, "utf-8")
class DummySpider(Spider):
name = "dummy"
start_urls = []
def parse(self, response):
print(response)
class Data:
SMALL_SIZE = 1024 # 1 KB
LARGE_SIZE = 1024**2 # 1 MB
STR_SMALL = generate_random_string(SMALL_SIZE)
STR_LARGE = generate_random_string(LARGE_SIZE)
EXTRA_SMALL = generate_random_string(1024 * 15)
EXTRA_LARGE = generate_random_string(LARGE_SIZE)
HTML_SMALL = make_html_body(STR_SMALL)
HTML_LARGE = make_html_body(STR_LARGE)
JSON_SMALL = {"data": STR_SMALL}
JSON_LARGE = {"data": STR_LARGE}
DATALOSS = b"Dataloss Content"
NO_CONTENT_LENGTH = b"This response do not have any content-length header"
class GetDataHtmlSmall(LeafResource):
def render_GET(self, request: TxRequest):
request.setHeader("Content-Type", "text/html; charset=UTF-8")
return Data.HTML_SMALL
class GetDataHtmlLarge(LeafResource):
def render_GET(self, request: TxRequest):
request.setHeader("Content-Type", "text/html; charset=UTF-8")
return Data.HTML_LARGE
class PostDataJsonMixin:
@staticmethod
def make_response(request: TxRequest, extra_data: str) -> bytes:
assert request.content is not None
response = {
"request-headers": {},
"request-body": json.loads(request.content.read()),
"extra-data": extra_data,
}
for k, v in request.requestHeaders.getAllRawHeaders():
response["request-headers"][str(k, "utf-8")] = str(v[0], "utf-8")
response_bytes = bytes(json.dumps(response), "utf-8")
request.setHeader("Content-Type", "application/json; charset=UTF-8")
request.setHeader("Content-Encoding", "UTF-8")
return response_bytes
class PostDataJsonSmall(LeafResource, PostDataJsonMixin):
def render_POST(self, request: TxRequest):
return self.make_response(request, Data.EXTRA_SMALL)
class PostDataJsonLarge(LeafResource, PostDataJsonMixin):
def render_POST(self, request: TxRequest):
return self.make_response(request, Data.EXTRA_LARGE)
class Dataloss(LeafResource):
def render_GET(self, request: TxRequest):
request.setHeader(b"Content-Length", b"1024")
self.deferRequest(request, 0, self._delayed_render, request)
return NOT_DONE_YET
@staticmethod
def _delayed_render(request: TxRequest):
request.write(Data.DATALOSS)
request.finish()
class NoContentLengthHeader(LeafResource):
def render_GET(self, request: TxRequest):
request.responseHeaders.removeHeader("Content-Length")
self.deferRequest(request, 0, self._delayed_render, request)
return NOT_DONE_YET
@staticmethod
def _delayed_render(request: TxRequest):
request.write(Data.NO_CONTENT_LENGTH)
request.finish()
class TimeoutResponse(LeafResource):
def render_GET(self, request: TxRequest):
return NOT_DONE_YET
class QueryParams(LeafResource):
def render_GET(self, request: TxRequest):
request.setHeader("Content-Type", "application/json; charset=UTF-8")
request.setHeader("Content-Encoding", "UTF-8")
query_params: dict[str, str] = {}
assert request.args is not None
for k, v in request.args.items():
query_params[str(k, "utf-8")] = str(v[0], "utf-8")
return bytes(json.dumps(query_params), "utf-8")
class RequestHeaders(LeafResource):
"""Sends all the headers received as a response"""
def render_GET(self, request: TxRequest):
request.setHeader("Content-Type", "application/json; charset=UTF-8")
request.setHeader("Content-Encoding", "UTF-8")
headers = {}
for k, v in request.requestHeaders.getAllRawHeaders():
headers[str(k, "utf-8")] = str(v[0], "utf-8")
return bytes(json.dumps(headers), "utf-8")
def make_request_dfd(client: H2ClientProtocol, request: Request) -> Deferred[Response]:
return client.request(request, DummySpider())
async def make_request(client: H2ClientProtocol, request: Request) -> Response:
return await maybe_deferred_to_future(make_request_dfd(client, request))
class TestHttps2ClientProtocol:
scheme = "https"
host = "localhost"
key_file = Path(__file__).parent / "keys" / "localhost.key"
certificate_file = Path(__file__).parent / "keys" / "localhost.crt"
@pytest.fixture
def site(self, tmp_path):
r = File(str(tmp_path))
put_child(r, b"get-data-html-small", GetDataHtmlSmall())
put_child(r, b"get-data-html-large", GetDataHtmlLarge())
put_child(r, b"post-data-json-small", PostDataJsonSmall())
put_child(r, b"post-data-json-large", PostDataJsonLarge())
put_child(r, b"dataloss", Dataloss())
put_child(r, b"no-content-length-header", NoContentLengthHeader())
put_child(r, b"status", Status())
put_child(r, b"query-params", QueryParams())
put_child(r, b"timeout", TimeoutResponse())
put_child(r, b"request-headers", RequestHeaders())
return Site(r, timeout=None)
@async_yield_fixture # type: ignore[untyped-decorator]
async def server_port(self, site: Site) -> AsyncGenerator[int]:
from twisted.internet import reactor
context_factory = ssl_context_factory(
str(self.key_file), str(self.certificate_file)
)
server_endpoint = SSL4ServerEndpoint(
reactor, 0, context_factory, interface=self.host
)
server = await server_endpoint.listen(site)
yield server.getHost().port
await server.stopListening()
@pytest.fixture
def client_certificate(self) -> PrivateCertificate:
pem = self.key_file.read_text(
encoding="utf-8"
) + self.certificate_file.read_text(encoding="utf-8")
return PrivateCertificate.loadPEM(pem) # type: ignore[no-any-return]
@pytest.fixture
def crawler(self, request: pytest.FixtureRequest) -> Crawler:
return get_crawler(settings_dict=getattr(request, "param", None))
@async_yield_fixture # type: ignore[untyped-decorator]
async def client(
self,
server_port: int,
client_certificate: PrivateCertificate,
crawler: Crawler,
) -> AsyncGenerator[H2ClientProtocol]:
from twisted.internet import reactor
from scrapy.core._http2.protocol import H2ClientFactory # noqa: PLC0415
client_options = optionsForClientTLS(
hostname=self.host,
trustRoot=client_certificate,
acceptableProtocols=[b"h2"],
)
uri = URI.fromBytes(bytes(self.get_url(server_port, "/"), "utf-8"))
h2_client_factory = H2ClientFactory(uri, crawler, Deferred())
client_endpoint = SSL4ClientEndpoint(
reactor, self.host, server_port, client_options
)
client = await client_endpoint.connect(h2_client_factory)
yield client
if not client.transport.disconnecting:
client.transport.loseConnection()
client.transport.abortConnection()
def get_url(self, portno: int, path: str) -> str:
"""
:param path: Should have / at the starting compulsorily if not empty
:return: Complete url
"""
assert len(path) > 0
assert path[0] == "/" or path[0] == "&"
return f"{self.scheme}://{self.host}:{portno}{path}"
@staticmethod
async def _check_repeat(
get_coro: Callable[[], Coroutine[Any, Any, None]], count: int
) -> None:
d_list = []
for _ in range(count):
d = deferred_from_coro(get_coro())
d_list.append(d)
await maybe_deferred_to_future(DeferredList(d_list, fireOnOneErrback=True))
async def _check_GET(
self,
client: H2ClientProtocol,
request: Request,
expected_body: bytes,
expected_status: int,
) -> None:
response = await make_request(client, request)
assert response.status == expected_status
assert response.body == expected_body
content_length_header = response.headers.get("Content-Length")
assert content_length_header is not None
content_length = int(content_length_header)
assert len(response.body) == content_length
@deferred_f_from_coro_f
async def test_GET_small_body(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = Request(self.get_url(server_port, "/get-data-html-small"))
await self._check_GET(client, request, Data.HTML_SMALL, 200)
@deferred_f_from_coro_f
async def test_GET_large_body(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = Request(self.get_url(server_port, "/get-data-html-large"))
await self._check_GET(client, request, Data.HTML_LARGE, 200)
@pytest.mark.parametrize(
"crawler", [{"HTTP2_MAX_FRAME_SIZE": 1024**2}], indirect=True
)
@deferred_f_from_coro_f
async def test_GET_large_frames(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = Request(self.get_url(server_port, "/get-data-html-large"))
await self._check_GET(client, request, Data.HTML_LARGE, 200)
assert client.conn.local_settings.max_frame_size == 1024**2
async def _check_GET_x10(
self,
client: H2ClientProtocol,
request: Request,
expected_body: bytes,
expected_status: int,
) -> None:
async def get_coro() -> None:
await self._check_GET(client, request, expected_body, expected_status)
await self._check_repeat(get_coro, 10)
@deferred_f_from_coro_f
async def test_GET_small_body_x10(
self, server_port: int, client: H2ClientProtocol
) -> None:
await self._check_GET_x10(
client,
Request(self.get_url(server_port, "/get-data-html-small")),
Data.HTML_SMALL,
200,
)
@deferred_f_from_coro_f
async def test_GET_large_body_x10(
self, server_port: int, client: H2ClientProtocol
) -> None:
await self._check_GET_x10(
client,
Request(self.get_url(server_port, "/get-data-html-large")),
Data.HTML_LARGE,
200,
)
@staticmethod
async def _check_POST_json(
client: H2ClientProtocol,
request: Request,
expected_request_body: dict[str, str],
expected_extra_data: str,
expected_status: int,
) -> None:
response = await make_request(client, request)
assert response.status == expected_status
content_length_header = response.headers.get("Content-Length")
assert content_length_header is not None
content_length = int(content_length_header)
assert len(response.body) == content_length
# Parse the body
content_encoding_header = response.headers[b"Content-Encoding"]
assert content_encoding_header is not None
content_encoding = str(content_encoding_header, "utf-8")
body = json.loads(str(response.body, content_encoding))
assert "request-body" in body
assert "extra-data" in body
assert "request-headers" in body
request_body = body["request-body"]
assert request_body == expected_request_body
extra_data = body["extra-data"]
assert extra_data == expected_extra_data
# Check if headers were sent successfully
request_headers = body["request-headers"]
for k, v in request.headers.items():
k_str = str(k, "utf-8")
assert k_str in request_headers
assert request_headers[k_str] == str(v[0], "utf-8")
@deferred_f_from_coro_f
async def test_POST_small_json(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = JsonRequest(
url=self.get_url(server_port, "/post-data-json-small"),
method="POST",
data=Data.JSON_SMALL,
)
await self._check_POST_json(
client, request, Data.JSON_SMALL, Data.EXTRA_SMALL, 200
)
@deferred_f_from_coro_f
async def test_POST_large_json(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = JsonRequest(
url=self.get_url(server_port, "/post-data-json-large"),
method="POST",
data=Data.JSON_LARGE,
)
await self._check_POST_json(
client, request, Data.JSON_LARGE, Data.EXTRA_LARGE, 200
)
async def _check_POST_json_x10(
self,
client: H2ClientProtocol,
request: Request,
expected_request_body: dict[str, str],
expected_extra_data: str,
expected_status: int,
) -> None:
async def get_coro() -> None:
await self._check_POST_json(
client,
request,
expected_request_body,
expected_extra_data,
expected_status,
)
await self._check_repeat(get_coro, 10)
@deferred_f_from_coro_f
async def test_POST_small_json_x10(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = JsonRequest(
url=self.get_url(server_port, "/post-data-json-small"),
method="POST",
data=Data.JSON_SMALL,
)
await self._check_POST_json_x10(
client, request, Data.JSON_SMALL, Data.EXTRA_SMALL, 200
)
@deferred_f_from_coro_f
async def test_POST_large_json_x10(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = JsonRequest(
url=self.get_url(server_port, "/post-data-json-large"),
method="POST",
data=Data.JSON_LARGE,
)
await self._check_POST_json_x10(
client, request, Data.JSON_LARGE, Data.EXTRA_LARGE, 200
)
@inlineCallbacks
def test_invalid_negotiated_protocol(
self, server_port: int, client: H2ClientProtocol
) -> Generator[Deferred[Any], Any, None]:
with mock.patch("scrapy.core._http2.protocol.PROTOCOL_NAME", new=b"not-h2"):
request = Request(url=self.get_url(server_port, "/status?n=200"))
with pytest.raises(ResponseFailed):
yield make_request_dfd(client, request)
@inlineCallbacks
def test_cancel_request(
self, server_port: int, client: H2ClientProtocol
) -> Generator[Deferred[Any], Any, None]:
request = Request(url=self.get_url(server_port, "/get-data-html-large"))
d = make_request_dfd(client, request)
d.cancel()
response = cast("Response", (yield d))
assert response.status == 499
@deferred_f_from_coro_f
async def test_download_maxsize_exceeded(
self,
caplog: pytest.LogCaptureFixture,
server_port: int,
client: H2ClientProtocol,
) -> None:
request = Request(
url=self.get_url(server_port, "/get-data-html-large"),
meta={"download_maxsize": 1000},
)
with pytest.raises(
DownloadCancelledError,
match=r"Expected to receive \d+ bytes which is larger than download max size \(1000\)",
):
await make_request(client, request)
@inlineCallbacks
def test_received_dataloss_response(
self, server_port: int, client: H2ClientProtocol
) -> Generator[Deferred[Any], Any, None]:
"""In case when value of Header Content-Length != len(Received Data)
ProtocolError is raised"""
from h2.exceptions import InvalidBodyLengthError # noqa: PLC0415
request = Request(url=self.get_url(server_port, "/dataloss"))
with pytest.raises(ResponseFailed) as exc_info:
yield make_request_dfd(client, request)
assert len(exc_info.value.reasons) > 0
assert any(
isinstance(error, InvalidBodyLengthError)
for error in exc_info.value.reasons
)
@deferred_f_from_coro_f
async def test_missing_content_length_header(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = Request(url=self.get_url(server_port, "/no-content-length-header"))
response = await make_request(client, request)
assert response.status == 200
assert response.body == Data.NO_CONTENT_LENGTH
assert "Content-Length" not in response.headers
async def _check_log_warnsize(
self,
client: H2ClientProtocol,
request: Request,
warn_pattern: re.Pattern[str],
expected_body: bytes,
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level("WARNING", "scrapy.core._http2.stream"):
response = await make_request(client, request)
assert response.status == 200
assert response.body == expected_body
# Check the warning is raised only once for this request
assert len(re.findall(warn_pattern, caplog.text)) == 1
@deferred_f_from_coro_f
async def test_log_expected_warnsize(
self,
server_port: int,
client: H2ClientProtocol,
caplog: pytest.LogCaptureFixture,
) -> None:
request = Request(
url=self.get_url(server_port, "/get-data-html-large"),
meta={"download_warnsize": 1000},
)
warn_pattern = re.compile(
rf"Expected to receive \d+ bytes which is larger than "
rf"download warn size \(1000\) in request {request}"
)
await self._check_log_warnsize(
client, request, warn_pattern, Data.HTML_LARGE, caplog
)
@deferred_f_from_coro_f
async def test_log_received_warnsize(
self,
server_port: int,
client: H2ClientProtocol,
caplog: pytest.LogCaptureFixture,
) -> None:
request = Request(
url=self.get_url(server_port, "/no-content-length-header"),
meta={"download_warnsize": 10},
)
warn_pattern = re.compile(
rf"Received \d+ bytes which is larger than "
rf"download warn size \(10\) in request {request}"
)
await self._check_log_warnsize(
client, request, warn_pattern, Data.NO_CONTENT_LENGTH, caplog
)
@deferred_f_from_coro_f
async def test_max_concurrent_streams(
self, server_port: int, client: H2ClientProtocol
) -> None:
"""Send 500 requests at one to check if we can handle
very large number of request.
"""
async def get_coro() -> None:
await self._check_GET(
client,
Request(self.get_url(server_port, "/get-data-html-small")),
Data.HTML_SMALL,
200,
)
await self._check_repeat(get_coro, 500)
@inlineCallbacks
def test_inactive_stream(
self, server_port: int, client: H2ClientProtocol
) -> Generator[Deferred[Any], Any, None]:
"""Here we send 110 requests considering the MAX_CONCURRENT_STREAMS
by default is 100. After sending the first 100 requests we close the
connection."""
d_list = []
def assert_inactive_stream(failure):
assert failure.check(ResponseFailed) is not None
from scrapy.core._http2.stream import InactiveStreamClosed # noqa: PLC0415
assert any(
isinstance(e, InactiveStreamClosed) for e in failure.value.reasons
)
# Send 100 request (we do not check the result)
for _ in range(100):
d = make_request_dfd(
client, Request(self.get_url(server_port, "/get-data-html-small"))
)
d.addBoth(lambda _: None)
d_list.append(d)
# Now send 10 extra request and save the response deferred in a list
for _ in range(10):
d = make_request_dfd(
client, Request(self.get_url(server_port, "/get-data-html-small"))
)
d.addCallback(lambda _: pytest.fail("This request should have failed"))
d.addErrback(assert_inactive_stream)
d_list.append(d)
# Close the connection now to fire all the extra 10 requests errback
# with InactiveStreamClosed
assert client.transport
client.transport.loseConnection()
yield DeferredList(d_list, consumeErrors=True, fireOnOneErrback=True)
@deferred_f_from_coro_f
async def test_invalid_request_type(self, client: H2ClientProtocol):
with pytest.raises(TypeError):
await make_request(client, "https://InvalidDataTypePassed.com") # type: ignore[arg-type]
@deferred_f_from_coro_f
async def test_query_parameters(
self, server_port: int, client: H2ClientProtocol
) -> None:
params = {
"a": generate_random_string(20),
"b": generate_random_string(20),
"c": generate_random_string(20),
"d": generate_random_string(20),
}
request = Request(
self.get_url(server_port, f"/query-params?{urlencode(params)}")
)
response = await make_request(client, request)
content_encoding_header = response.headers[b"Content-Encoding"]
assert content_encoding_header is not None
content_encoding = str(content_encoding_header, "utf-8")
data = json.loads(str(response.body, content_encoding))
assert data == params
@deferred_f_from_coro_f
async def test_status_codes(
self, server_port: int, client: H2ClientProtocol
) -> None:
for status in [200, 404]:
request = Request(self.get_url(server_port, f"/status?n={status}"))
response = await make_request(client, request)
assert response.status == status
@pytest.mark.filterwarnings(
r"ignore:.*You should use cryptography's X\.509 APIs:DeprecationWarning"
)
@deferred_f_from_coro_f
async def test_response_has_correct_certificate_ip_address(
self,
server_port: int,
client: H2ClientProtocol,
client_certificate: PrivateCertificate,
) -> None:
request = Request(self.get_url(server_port, "/status?n=200"))
response = await make_request(client, request)
assert isinstance(response.certificate, Certificate)
assert response.certificate.original is not None
assert response.certificate.getIssuer() == client_certificate.getIssuer()
assert response.certificate.getPublicKey().matches(
client_certificate.getPublicKey()
)
assert isinstance(response.ip_address, IPv4Address)
assert str(response.ip_address) == "127.0.0.1"
@staticmethod
async def _check_invalid_netloc(client: H2ClientProtocol, url: str) -> None:
from scrapy.core._http2.stream import InvalidHostname # noqa: PLC0415
request = Request(url)
with pytest.raises(InvalidHostname) as exc_info:
await make_request(client, request)
error_msg = str(exc_info.value)
assert "localhost" in error_msg
assert "127.0.0.1" in error_msg
assert str(request) in error_msg
@deferred_f_from_coro_f
async def test_invalid_hostname(self, client: H2ClientProtocol) -> None:
await self._check_invalid_netloc(
client, "https://notlocalhost.notlocalhostdomain"
)
@deferred_f_from_coro_f
async def test_invalid_host_port(
self, server_port: int, client: H2ClientProtocol
) -> None:
port = server_port + 1
await self._check_invalid_netloc(client, f"https://127.0.0.1:{port}")
@deferred_f_from_coro_f
async def test_connection_stays_with_invalid_requests(
self, server_port: int, client: H2ClientProtocol
):
await maybe_deferred_to_future(self.test_invalid_hostname(client))
await maybe_deferred_to_future(self.test_invalid_host_port(server_port, client))
await maybe_deferred_to_future(self.test_GET_small_body(server_port, client))
await maybe_deferred_to_future(self.test_POST_small_json(server_port, client))
@inlineCallbacks
def test_connection_timeout(
self, server_port: int, client: H2ClientProtocol
) -> Generator[Deferred[Any], Any, None]:
request = Request(self.get_url(server_port, "/timeout"))
# Update the timer to 1s to test connection timeout
client.setTimeout(1) # type: ignore[no-untyped-call]
with pytest.raises(ResponseFailed) as exc_info:
yield make_request_dfd(client, request)
for err in exc_info.value.reasons:
from scrapy.core._http2.protocol import H2ClientProtocol # noqa: PLC0415
if isinstance(err, DownloadTimeoutError):
assert (
f"Connection was IDLE for more than {H2ClientProtocol.IDLE_TIMEOUT}s"
in str(err)
)
break
else:
pytest.fail("No TimeoutError raised.")
@deferred_f_from_coro_f
async def test_connection_timeout_no_streams(
self, client: H2ClientProtocol
) -> None:
"""A connection with no streams is closed when it times out."""
from twisted.internet import reactor
client.setTimeout(0.1) # type: ignore[no-untyped-call]
await maybe_deferred_to_future(deferLater(reactor, 0.5))
assert client.transport.disconnecting
@deferred_f_from_coro_f
async def test_request_headers_received(
self, server_port: int, client: H2ClientProtocol
) -> None:
request = Request(
self.get_url(server_port, "/request-headers"),
headers={"header-1": "header value 1", "header-2": "header value 2"},
)
response = await make_request(client, request)
assert response.status == 200
response_headers = json.loads(str(response.body, "utf-8"))
assert isinstance(response_headers, dict)
for k, v in request.headers.items():
k_decoded, v_decoded = str(k, "utf-8"), str(v[0], "utf-8")
assert k_decoded in response_headers
assert v_decoded == response_headers[k_decoded]