diff --git a/src/scm/git/relay.py b/src/scm/git/relay.py index aa8d785..b65848a 100644 --- a/src/scm/git/relay.py +++ b/src/scm/git/relay.py @@ -112,8 +112,9 @@ class GitRelay: one, the relay uses a plain ``HTTPAdapter``, which checks nothing. The relay replaces a requests or urllib3 exception from the upstream request with a coded error, so that it - cannot carry the token. ``map_error`` gets any other exception, such as the adapter's own for a refused host, - and returns either a coded error, which the relay raises in its place without the original, or None to let the + cannot carry the token, and so is one from reading the response body, whose frames hold the bytes read so far. + ``map_error`` gets any other exception from the request, such as the adapter's own for a refused host, and + returns either a coded error, which the relay raises in its place without the original, or None to let the original reach the caller unchanged. An error that reaches the caller this way does carry the token: an adapter refusing the host as it connects raises through urllib3's frames, which hold the Authorization header, so report such an error without its frames' local variables. GitServer answers a coded error like its other refusals. For @@ -279,7 +280,7 @@ class _UpstreamStream: def __init__(self, session: requests.Session, upstream: requests.Response, chunk_size: int) -> None: self.session = session self.upstream = upstream - self.chunks = upstream.raw.stream(chunk_size, decode_content=False) + self.chunks = _read_body(upstream, chunk_size) def __iter__(self) -> "_UpstreamStream": return self @@ -301,6 +302,20 @@ def close(self) -> None: self.session.close() +def _read_body(upstream: requests.Response, chunk_size: int) -> Iterator[bytes]: + """Yield the provider's response body, raising the coded error ``forward`` would for the same failure.""" + try: + yield from upstream.raw.stream(chunk_size, decode_content=False) + # "from None" drops urllib3's exception and the http.client one it chains, whose frames hold the partial chunk. + except urllib3.exceptions.ReadTimeoutError: + raise ResourceGatewayTimeout() from None + # A broken connection or a body cut off early. An SSLError here is the TLS connection breaking, not a certificate. + except (urllib3.exceptions.ProtocolError, urllib3.exceptions.IncompleteRead, urllib3.exceptions.SSLError): + raise ResourceServiceUnavailable() from None + except urllib3.exceptions.HTTPError: + raise ResourceBadGateway() from None + + def _read_until_eof(body: Readable, chunk_size: int) -> Iterator[bytes]: while chunk := _read(body, chunk_size): yield chunk diff --git a/src/scm/git/server.py b/src/scm/git/server.py index 101a6bf..e49bedf 100644 --- a/src/scm/git/server.py +++ b/src/scm/git/server.py @@ -80,6 +80,10 @@ class GitServer: ``client_closed_request`` when the client goes away first. A response the provider sends with a status other than 2xx is passed on to the client but counted as failed too, as ``provider_error``. The bytes sent each way are counted then too, as ``request_size`` and ``response_size``. + + A coded error from the provider's stream breaking, such as a ``ResourceGatewayTimeout``, comes too late for a + refusal, so it goes to ``emit_error`` with the request's tags and is then raised, making the WSGI server drop + the connection rather than end the response as if it were whole. The caller need not report it again. """ def __init__( @@ -193,10 +197,13 @@ def finish(failure: str | None, bytes_sent: int) -> None: if bytes_sent: self.record_count("sentry.scm.git.response_size", bytes_sent, tags) + def report(error: SCMCodedError) -> None: + self.emit_error(error, context) + return StreamResponse( status_code=response.status_code, headers=response.headers, - content=_MeteredStream(response.content, finish), + content=_MeteredStream(response.content, finish, report), ) @@ -228,9 +235,15 @@ def read(self, size: int = -1, /) -> bytes: class _MeteredStream: """The relay's response stream, reporting once how it ended and how many bytes it passed on.""" - def __init__(self, content: ResponseBody, finish: Callable[[str | None, int], None]) -> None: + def __init__( + self, + content: ResponseBody, + finish: Callable[[str | None, int], None], + report: Callable[[SCMCodedError], None], + ) -> None: self.content = content self.finish = finish + self.report = report self.bytes_sent = 0 self.finished = False @@ -243,6 +256,11 @@ def __next__(self) -> bytes: except StopIteration: self._end(None) raise + except SCMCodedError as e: + # Counted like any other break; the report says which it was. + self._end("stream_interrupted") + self.report(e) + raise except Exception: self._end("stream_interrupted") raise diff --git a/tests/unit/error_reports.py b/tests/unit/error_reports.py index af4a846..1cd60f6 100644 --- a/tests/unit/error_reports.py +++ b/tests/unit/error_reports.py @@ -11,6 +11,16 @@ def assert_not_in_frames(error: BaseException, value: str) -> None: tb = tb.tb_next +def assert_not_reported(error: BaseException, value: bytes) -> None: + """Fail if a report of ``error`` or what it chains, followed as Sentry's SDK does, would show ``value``.""" + shown = repr(value)[2:-1] + reported: BaseException | None = error + while reported is not None: + assert shown not in str(reported) and shown not in repr(reported), f"{type(reported).__name__} holds it" + assert_not_in_frames(reported, shown) + reported = reported.__cause__ if reported.__suppress_context__ else reported.__context__ + + def reported_local(error: BaseException, function: str, name: str) -> str: """How an error report shows the local variable ``name`` of ``function``'s frame in the error's traceback.""" tb = error.__traceback__ diff --git a/tests/unit/test_git_relay.py b/tests/unit/test_git_relay.py index db2ec1a..fe304ed 100644 --- a/tests/unit/test_git_relay.py +++ b/tests/unit/test_git_relay.py @@ -1,6 +1,8 @@ +import contextlib import gzip import io import socket +import ssl import threading import time from collections.abc import Callable, Iterator @@ -26,7 +28,7 @@ from scm.git.relay import GitRelay, _UpstreamStream, authorize_request from scm.secret import Secret from scm.types import GitRemote -from tests.unit.error_reports import assert_not_in_frames, reported_local +from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local from tests.unit.git_builders import make_grant from tests.unit.git_upstream import ( ClosingAdapter, @@ -538,6 +540,48 @@ def handler(h: UpstreamHandler) -> None: finally: release.set() + def test_provider_cutting_off_the_body(self, relay, upstream): + def handler(h: UpstreamHandler) -> None: + h.start_chunked(200) + h.write_chunk(b"first") + # A chunk that promises more than it sends, so http.client raises IncompleteRead. + h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_CHUNK)) + PARTIAL_CHUNK) + h.wfile.flush() + # A clean TLS close, so the relay sees the chunk end early rather than TLS breaking. It then fails + # waiting for the relay's close_notify. + with contextlib.suppress(ssl.SSLError): + h.connection.unwrap() # type: ignore[attr-defined] + h.close_connection = True + + upstream.handler = handler + content = fetch_refs(relay, upstream).content + assert next(content) == b"first" + with pytest.raises(ResourceServiceUnavailable) as exc_info: + next(content) + assert isinstance(exc_info.value.__context__, urllib3.exceptions.ProtocolError) + assert_body_not_in(exc_info.value) + + def test_provider_going_silent_mid_body(self, upstream): + release = threading.Event() + + def handler(h: UpstreamHandler) -> None: + h.start_chunked(200) + h.write_chunk(b"first") + h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_CHUNK)) + PARTIAL_CHUNK) + h.wfile.flush() + release.wait(WAIT) + + upstream.handler = handler + try: + content = fetch_refs(GitRelay(ca_bundle=ca_bundle(), timeout=(1, 0.2)), upstream).content + assert next(content) == b"first" + with pytest.raises(ResourceGatewayTimeout) as exc_info: + next(content) + assert isinstance(exc_info.value.__context__, urllib3.exceptions.ReadTimeoutError) + assert_body_not_in(exc_info.value) + finally: + release.set() + @pytest.mark.parametrize( "make_error", [ @@ -590,6 +634,16 @@ def assert_token_not_in(error: Exception) -> None: assert basic_auth("x-access-token", TOKEN.reveal()) not in repr(error) +# Part of a pack, which must not reach an error report. +PARTIAL_CHUNK = b"PACK-partial-repository-contents" * 64 + + +def assert_body_not_in(error: Exception) -> None: + assert error.__cause__ is None + assert error.__suppress_context__ + assert_not_reported(error, PARTIAL_CHUNK[:32]) + + def upstream_stream(*chunks: bytes | Exception) -> tuple[_UpstreamStream, MagicMock, MagicMock]: """An _UpstreamStream over a response whose body yields ``chunks``, raising any that is an exception.""" @@ -620,11 +674,43 @@ def test_closes_the_response_and_session_when_the_provider_stream_breaks(self): stream, session, response = upstream_stream(b"a", urllib3.exceptions.ProtocolError("connection broken")) assert next(stream) == b"a" response.close.assert_not_called() - with pytest.raises(urllib3.exceptions.ProtocolError): + with pytest.raises(ResourceServiceUnavailable): next(stream) response.close.assert_called_once() session.close.assert_called_once() + @pytest.mark.parametrize( + ("error", "expected"), + [ + ( + urllib3.exceptions.ReadTimeoutError(urllib3.HTTPSConnectionPool("127.0.0.1"), "", "Read timed out."), + ResourceGatewayTimeout, + ), + (urllib3.exceptions.ProtocolError("Connection broken"), ResourceServiceUnavailable), + (urllib3.exceptions.IncompleteRead(10, 5), ResourceServiceUnavailable), + # Not a certificate the relay does not trust: the connection is already up. + (urllib3.exceptions.SSLError("bad record mac"), ResourceServiceUnavailable), + (urllib3.exceptions.HTTPError("something else"), ResourceBadGateway), + ], + ids=lambda value: value.__name__ if isinstance(value, type) else type(value).__name__, + ) + def test_provider_stream_breaking_is_a_coded_error(self, error, expected): + stream, _, _ = upstream_stream(b"a", error) + next(stream) + with pytest.raises(expected) as exc_info: + next(stream) + assert exc_info.value.__cause__ is None + assert exc_info.value.__suppress_context__ + + def test_other_errors_reach_the_caller_unchanged(self): + error = RuntimeError("not a urllib3 error") + stream, session, _ = upstream_stream(b"a", error) + next(stream) + with pytest.raises(RuntimeError) as exc_info: + next(stream) + assert exc_info.value is error + session.close.assert_called_once() + def test_close_before_reading(self): # A response GitServer replaces, such as a 401, is closed without being read. stream, session, response = upstream_stream(b"a") diff --git a/tests/unit/test_git_server.py b/tests/unit/test_git_server.py index b6a5e79..b6dc082 100644 --- a/tests/unit/test_git_server.py +++ b/tests/unit/test_git_server.py @@ -1,4 +1,6 @@ +import contextlib import io +import ssl import time from collections.abc import Generator from types import SimpleNamespace @@ -12,6 +14,7 @@ GitGrantKeyError, GitRemoteRefused, RateLimitExceeded, + ResourceGatewayTimeout, ResourceServiceUnavailable, ) from scm.git.grant import sign_grant @@ -21,13 +24,22 @@ from scm.rpc.types import StreamResponse from scm.secret import Secret from scm.types import GitRemote, Provider, Repository, RepositoryId -from tests.unit.error_reports import assert_not_in_frames, reported_local +from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local from tests.unit.git_builders import make_grant, make_key_pair, make_provider, make_repository -from tests.unit.git_upstream import ClosingAdapter, HostRefused, RefusingAdapter, basic_auth, running_upstream +from tests.unit.git_upstream import ( + ClosingAdapter, + HostRefused, + RefusingAdapter, + UpstreamHandler, + basic_auth, + running_upstream, +) KEY = make_key_pair() TOKEN = Secret("ghs_secrettoken") REFERRER = make_grant().referrer +# A constant, not a test's local, since the test's own frame is in the traceback. +PARTIAL_PACK = b"PACK-partial-repository-contents" * 64 def signed_grant(**overrides) -> str: @@ -348,6 +360,53 @@ def broken_stream(): ) ] assert recorder.bytes == {("sentry.scm.git.response_size", "fetch"): 11} + # Not coded, so the caller reports it. + assert recorder.errors == [] + + def test_provider_stream_breaking_with_a_coded_error_is_reported_and_raised(self): + error = ResourceGatewayTimeout() + + def broken_stream(): + yield b"half a pack" + raise error + + recorder = Recorder() + content = request(make_server(recorder, relay_returning(broken_stream())), signed_grant()).content + assert next(content) == b"half a pack" + assert recorder.errors == [] + # Raised, so the WSGI server drops the connection. + with pytest.raises(ResourceGatewayTimeout) as exc_info: + next(content) + assert exc_info.value is error + assert recorder.errors == [ + ( + error, + { + "repository_id": "5", + "organization_id": "1", + "referrer": REFERRER, + "operation": "fetch", + "provider": "github", + "remote_host": "github.com", + }, + ) + ] + assert recorder.counts == [ + ( + "sentry.scm.git.request.failed", + { + "referrer": REFERRER, + "operation": "fetch", + "provider": "github", + "status": "200", + "code": "stream_interrupted", + }, + ) + ] + # Reported once, however the caller goes on to close the stream. + content.close() # type: ignore[attr-defined] + assert len(recorder.errors) == 1 + assert len(recorder.counts) == 1 def test_client_leaving_early_is_a_failure_and_closes_the_provider_stream(self): closed = [] @@ -586,6 +645,34 @@ def test_relays_end_to_end(self): assert received.headers["Authorization"] == basic_auth("x-access-token", TOKEN.reveal()) assert GRANT_HEADER not in received.headers + def test_provider_cutting_off_the_body_end_to_end(self): + def handler(h: UpstreamHandler) -> None: + h.start_chunked(200) + h.write_chunk(b"first") + h.wfile.write(b"%x\r\n" % (2 * len(PARTIAL_PACK)) + PARTIAL_PACK) + h.wfile.flush() + # See test_git_relay's test_provider_cutting_off_the_body. + with contextlib.suppress(ssl.SSLError): + h.connection.unwrap() # type: ignore[attr-defined] + h.close_connection = True + + with running_upstream() as upstream: + upstream.handler = handler + recorder = Recorder( + repository=make_repository(provider_name="github_enterprise", web_base_url=upstream.url) + ) + content = request(make_server(recorder, GitRelay(ca_bundle=upstream.ca_bundle)), signed_grant()).content + assert next(content) == b"first" + with pytest.raises(ResourceServiceUnavailable) as exc_info: + next(content) + assert exc_info.value.__cause__ is None + assert exc_info.value.__suppress_context__ + assert_not_reported(exc_info.value, PARTIAL_PACK[:32]) + [(error, context)] = recorder.errors + assert error is exc_info.value + assert set(context) == {"repository_id", "organization_id", "referrer", "operation", "provider", "remote_host"} + assert [tags["code"] for _, tags in recorder.counts] == ["stream_interrupted"] + class TestKeptOutOfErrorReports: """What GitServer passes to emit_error or raises is reported to Sentry, with every frame's local variables."""