diff --git a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py index 6366e769..d42d8e9a 100644 --- a/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py +++ b/pipeline/fiftyone_pipeline_engines/src/fiftyone_pipeline_engines/datafile_update_service.py @@ -199,6 +199,20 @@ def update_data_file(self, data_file): self._log_check_error(message) return self._update_failed(data_file, status) + # A response that ended before the whole file arrived is an error + # accessing the service. urllib3 2 raises one, which the request + # above catches, but urllib3 1 gives the bytes that arrived, so the + # length is checked here as well. + short = self._cut_short(response) + if short is not None: + self._log_check_error( + "Error accessing data update service at '" + shown_url + + "' for engine '" + engine + "'", + "The response ended after " + str(short[0]) + " of the " + + str(short[1]) + " bytes it gave as its Content-Length.") + return self._update_failed( + data_file, UpdateStatus.AUTO_UPDATE_HTTPS_ERR) + # Verify the MD5 hash if enabled and the answer gives one. header_md5 = response.headers.get("content-md5") @@ -431,6 +445,23 @@ def _with_cause(text, cause): return (text + ("" if text.endswith(".") else ".") + " Error detail: " + mask_licence_keys_in_text(cause)) + @staticmethod + def _cut_short(response): + # The bytes received and the Content-Length, when fewer arrived than + # it gave, or None. With a Content-Encoding the length counts the + # encoded bytes and requests gives the decoded ones, so the two are + # not compared. + promised = response.headers.get("content-length") + encoding = response.headers.get("content-encoding", "identity") + if promised is None or encoding.strip().lower() != "identity": + return None + try: + promised = int(promised) + except ValueError: + return None + received = len(response.content) + return (received, promised) if received < promised else None + @staticmethod def _engine_type(data_file): return type(getattr(data_file, "flow_element", data_file)).__name__ diff --git a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py index 9d3dba02..be1a1798 100644 --- a/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py +++ b/pipeline/fiftyone_pipeline_engines/tests/test_datafile_update_http.py @@ -22,6 +22,7 @@ import datetime import gzip +import http.server import os import random import shutil @@ -783,5 +784,115 @@ def testFailedMd5CheckOnStartUpIsNotRepeatedAtOnce(self): self.statuses) +# The bytes a short download sends, and how many it promises. +SENT = b"x" * 100 +PROMISED = 1000 + + +class ShortDownloadHandler(http.server.BaseHTTPRequestHandler): + """Answers 200 and closes the connection before the whole file has been + sent, with the length given in Content-Length or in chunks.""" + + protocol_version = "HTTP/1.1" + chunked = False + + def do_GET(self): + self.send_response(200) + if self.chunked: + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + # One chunk, then the connection closes with no last chunk. + self.wfile.write(b"%x\r\n" % len(SENT) + SENT + b"\r\n") + else: + self.send_header("Content-Length", str(PROMISED)) + self.end_headers() + self.wfile.write(SENT) + self.wfile.flush() + self.close_connection = True + + def log_message(self, format, *args): + pass + + +class ShortDownloadTests(unittest.TestCase): + """A response that ends before the whole file arrived is reported as an + error accessing the data update service, and the data in use is kept.""" + + def serve(self, chunked): + handler = type( + "Handler", (ShortDownloadHandler,), {"chunked": chunked}) + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + self.addCleanup(server.server_close) + self.addCleanup(server.shutdown) + return "http://127.0.0.1:" + str(server.server_address[1]) + "/update" + + def check_short_download(self, chunked): + pipeline = RecordingPipeline() + service = DataFileUpdateService(pipeline) + statuses = [] + service.on_complete(lambda status, data_file: statuses.append(status)) + data_file = HttpDataFile() + data_file.verify_if_modified_since = False + data_file.url = self.serve(chunked) + + self.assertFalse(service.update_data_file(data_file)) + + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], statuses) + errors = [message for level, message in pipeline.logs + if level == "error"] + self.assertTrue( + any(message.startswith( + CHECK_ERROR + "Error accessing data update service at '" + + data_file.url + "' for engine 'FakeEngine'") + for message in errors), + errors) + self.assertEqual(b"old", data_file.data) + self.assertEqual([], data_file.refreshed) + + def testAResponseShorterThanItsContentLengthIsAnAccessError(self): + self.check_short_download(chunked=False) + + def testTheLengthIsCheckedWhicheverHttpLibraryIsInstalled(self): + # urllib3 before version 2 gives requests the bytes that arrived + # with no error, so the service compares them with Content-Length + # itself. A reply that holds fewer bytes stands in for that here. + pipeline = RecordingPipeline() + service = DataFileUpdateService(pipeline) + statuses = [] + service.on_complete(lambda status, data_file: statuses.append(status)) + data_file = HttpDataFile() + reply = response(200, SENT) + reply.headers = {"content-length": str(PROMISED)} + with mock.patch.object(requests, "get", lambda url, **kwargs: reply): + self.assertFalse(service.update_data_file(data_file)) + + self.assertEqual([UpdateStatus.AUTO_UPDATE_HTTPS_ERR], statuses) + self.assertIn( + ("error", + CHECK_ERROR + "Error accessing data update service at '" + URL + + "' for engine 'FakeEngine'. Error detail: The response ended " + "after 100 of the 1000 bytes it gave as its Content-Length."), + pipeline.logs) + self.assertEqual(b"old", data_file.data) + + def testAResponseInAContentEncodingIsNotMeasured(self): + # Content-Length counts the encoded bytes, and requests gives the + # decoded ones, so the two are not compared. + pipeline = RecordingPipeline() + service = DataFileUpdateService(pipeline) + data_file = HttpDataFile() + reply = response(200, SENT) + reply.headers = { + "content-length": str(PROMISED), "content-encoding": "gzip"} + with mock.patch.object(requests, "get", lambda url, **kwargs: reply): + self.assertTrue(service.update_data_file(data_file)) + self.assertEqual(SENT, data_file.data) + + def testAChunkedResponseCutShortIsAnAccessError(self): + self.check_short_download(chunked=True) + + if __name__ == "__main__": unittest.main()