From 281d3a3da4715da389f53408a5bfc13d1b9106e3 Mon Sep 17 00:00:00 2001 From: scarreras Date: Sat, 3 Oct 2026 21:33:55 -0300 Subject: [PATCH 1/3] Make sdkharness large-transfer checks observe concurrency; harden resilience executable - large.concurrent_parts and large.parallel_download now observe overlapping b2_upload_part requests and download streams (ranged GETs) through a loopback observing proxy with a client-side hold, and verify the uploaded bytes. - resilience-upload.py follows LATEST_STABLE_VERSION and asserts it runs from this checkout, accepts contentSha1 'none' only for large files, names the scenario in the result line on configuration failure, and retries only its own authorize setup step. - Proxy variables are scrubbed from every CLI child environment and the printed-URL fetch bypasses ambient proxies. - Unit tests for the loopback and fixed-credential guards. - Changelog fragments, including the migration PRs #1171-#1177. Co-Authored-By: Claude Sonnet 5.5 --- .sdkharness/README.md | 12 +- .../tests/conformance-file-lifecycle.py | 135 +++++- .sdkharness/tests/lib/observing_proxy.py | 210 +++++++++ .sdkharness/tests/lib/simulator_guard.py | 22 +- .sdkharness/tests/resilience-upload.py | 69 ++- .../+sdkharness-migration.infrastructure.md | 1 + ...ness-observe-concurrency.infrastructure.md | 1 + test/unit/test_sdkharness_guards.py | 401 ++++++++++++++++++ 8 files changed, 820 insertions(+), 31 deletions(-) create mode 100644 .sdkharness/tests/lib/observing_proxy.py create mode 100644 changelog.d/+sdkharness-migration.infrastructure.md create mode 100644 changelog.d/+sdkharness-observe-concurrency.infrastructure.md create mode 100644 test/unit/test_sdkharness_guards.py diff --git a/.sdkharness/README.md b/.sdkharness/README.md index 6e72ae43..92c74723 100644 --- a/.sdkharness/README.md +++ b/.sdkharness/README.md @@ -79,7 +79,17 @@ HEALTHCHECK_REALM_URL=$SDKHARNESS_SIMULATOR_URL B2_TEST_APPLICATION_KEY_ID=test- When you are done, stop the simulator with `kill "$SIM_PID"`. Each prints one `SDKHARNESS_RESULT` line and refuses any simulator URL that is not -`http://127.0.0.1:`. Use a fresh simulator per resilience scenario: an +`http://127.0.0.1:`. The CLI child never inherits proxy variables (`HTTP_PROXY`, +`HTTPS_PROXY`, `ALL_PROXY`, `NO_PROXY`, in either case); loopback is exempted explicitly. + +`large.concurrent_parts` and `large.parallel_download` observe concurrency on the wire: they +point the CLI at a small loopback proxy (`tests/lib/observing_proxy.py`) in front of the +simulator, hold each part upload / download stream for half a second so overlap is visible, +and require two or more in flight with `--threads 4` and exactly one with `--threads 1`. +`large.parallel_download` uses a file of just over 200 MiB because b2sdk starts one ranged +stream per 100 MiB of content and the CLI has no option to lower that. + +The guards are covered by `test/unit/test_sdkharness_guards.py` (run with the unit tests). Use a fresh simulator per resilience scenario: an unconsumed injected fault from a failing scenario (for example `upload.retry_408`) carries over to the next one. `api.retry_after_503` and `upload.retry_408` are known SDK findings, not setup problems. diff --git a/.sdkharness/tests/conformance-file-lifecycle.py b/.sdkharness/tests/conformance-file-lifecycle.py index 08c42139..f1344ebd 100755 --- a/.sdkharness/tests/conformance-file-lifecycle.py +++ b/.sdkharness/tests/conformance-file-lifecycle.py @@ -22,11 +22,12 @@ import tempfile import time import uuid -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence +from contextlib import contextmanager from pathlib import Path from urllib.error import HTTPError from urllib.parse import parse_qs, urlsplit -from urllib.request import urlopen +from urllib.request import ProxyHandler, build_opener LEVEL = 'conformance' SCENARIOS = { @@ -60,6 +61,8 @@ 'large.unbound_incremental', 'urls.native_download', } +# The URL the CLI printed is fetched directly: never through an ambient HTTP(S)_PROXY. +NO_PROXY_OPENER = build_opener(ProxyHandler({})) REPOSITORY_ROOT = Path(__file__).resolve().parents[2] # Resolve the CLI implementation from this checkout. The harness installs the @@ -69,12 +72,20 @@ from b2._internal.version_listing import LATEST_STABLE_VERSION # noqa: E402 sys.path.insert(0, str(Path(__file__).resolve().parent / 'lib')) +from observing_proxy import ObservingProxy # noqa: E402 from simulator_guard import ( # noqa: E402 CREDENTIAL_REFUSAL, credential_is_fixed, scrubbed_environment, ) +# How long the observing proxy holds each part upload / ranged download before +# forwarding it. The simulator answers in milliseconds, so without a hold two +# requests could overlap without the check ever catching them at the same time. +OBSERVE_HOLD_SECONDS = 0.5 +# b2sdk's DownloadManager.DEFAULT_MIN_PART_SIZE: the content one ranged download stream retrieves. +DOWNLOAD_PART_SIZE = 100 * 1024 * 1024 + class CheckFailure(Exception): """A failed scenario step with a credential-safe reason.""" @@ -1266,8 +1277,7 @@ def keys_multi_bucket(self) -> None: raise CheckFailure('multi-bucket key', 'unscoped bucket was accessible') @staticmethod - def large_payload() -> bytes: - size = 16 * 1024 * 1024 + def large_payload(size: int = 16 * 1024 * 1024) -> bytes: return (b'large-sdkharness-' * (size // 17 + 1))[:size] def assert_large_version( @@ -1290,7 +1300,13 @@ def assert_large_version( return version def upload_large( - self, bucket_name: str, name: str, payload: bytes, threads: int, step: str + self, + bucket_name: str, + name: str, + payload: bytes, + threads: int, + step: str, + part_size: int = 5_000_000, ) -> None: path = self.scratch / f'large-{uuid.uuid4().hex}.bin' path.write_bytes(payload) @@ -1300,7 +1316,7 @@ def upload_large( 'upload', '--no-progress', '--min-part-size', - '5000000', + str(part_size), '--threads', str(threads), bucket_name, @@ -1348,35 +1364,108 @@ def large_multipart(self) -> None: if self.download_from_bucket('download multipart file', bucket_name, name) != payload: raise CheckFailure('multipart round trip', 'downloaded bytes differ from upload') + @contextmanager + def through_observing_proxy(self, label: str) -> Iterator[ObservingProxy]: + """Run the CLI against a loopback proxy in front of the simulator. + + The CLI authorizes through the proxy (with its own account-info file), so every + API, upload and download URL it is later given leads back through the proxy, + which counts how many part uploads / ranged downloads are open at once. + """ + saved = self.child_environment + try: + with ObservingProxy(self.simulator_url, hold_seconds=OBSERVE_HOLD_SECONDS) as proxy: + self.child_environment = { + **saved, + 'B2_ENVIRONMENT': proxy.origin, + 'B2_ACCOUNT_INFO': str(self.scratch / f'observed-{label}-account-info'), + } + self.invoke(f'authenticate through observer ({label})', 'account', 'authorize') + yield proxy + finally: + self.child_environment = saved + def large_concurrent_parts(self) -> None: bucket_name = self.new_bucket() payload = self.large_payload() + peaks: dict[int, int] = {} for threads, name in ((1, 'st/threads-1.bin'), (4, 'st/threads-4.bin')): - self.upload_large(bucket_name, name, payload, threads, f'upload with {threads} threads') + with self.through_observing_proxy(f'upload-{threads}') as proxy: + self.upload_large( + bucket_name, name, payload, threads, f'upload with {threads} threads' + ) + peaks[threads] = proxy.peak('upload_part') + parts = proxy.total('upload_part') self.assert_large_version(bucket_name, name, payload, f'read {threads}-thread metadata') + if ( + self.download_from_bucket(f'download {threads}-thread upload', bucket_name, name) + != payload + ): + raise CheckFailure( + 'large upload', f'{threads}-thread upload stored different bytes' + ) + if parts < 2: + raise CheckFailure('large upload', f'only {parts} b2_upload_part request(s) seen') + if peaks[1] != 1: + raise CheckFailure('concurrency', f'--threads 1 had {peaks[1]} parts in flight at once') + if peaks[4] < 2: + raise CheckFailure( + 'concurrency', + f'--threads 4 never had two b2_upload_part requests in flight (peak {peaks[4]})', + ) def large_parallel_download(self) -> None: bucket_name = self.new_bucket() - payload = self.large_payload() + # b2sdk starts one ranged stream per DOWNLOAD_PART_SIZE of content and the CLI has no + # option to lower it, so a file must be at least two of them for the CLI to download + # it in parallel at all: a smaller file is always one plain GET, whatever --threads says. + payload = self.large_payload(2 * DOWNLOAD_PART_SIZE + 1024 * 1024) name = 'st/parallel-download.bin' - self.upload_large(bucket_name, name, payload, 4, 'upload parallel-download fixture') + self.upload_large( + bucket_name, + name, + payload, + 4, + 'upload parallel-download fixture', + part_size=DOWNLOAD_PART_SIZE, + ) self.assert_large_version(bucket_name, name, payload, 'read parallel-download metadata') + peaks: dict[int, int] = {} + ranged: dict[int, int] = {} for threads in (1, 4): path = self.scratch / f'download-{threads}.bin' - self.invoke( - f'download with {threads} threads', - 'file', - 'download', - '--no-progress', - '--threads', - str(threads), - '--max-download-streams-per-file', - str(threads), - f'b2://{bucket_name}/{name}', - str(path), - ) + with self.through_observing_proxy(f'download-{threads}') as proxy: + self.invoke( + f'download with {threads} threads', + 'file', + 'download', + '--no-progress', + '--threads', + str(threads), + '--max-download-streams-per-file', + str(threads), + f'b2://{bucket_name}/{name}', + str(path), + ) + peaks[threads] = proxy.peak('download_stream') + ranged[threads] = proxy.total('ranged_get') if path.read_bytes() != payload: raise CheckFailure('parallel download', f'{threads}-thread bytes differ') + path.unlink() + if peaks[1] != 1 or ranged[1] != 0: + raise CheckFailure( + 'concurrency', + f'--threads 1 used {peaks[1]} overlapping streams and {ranged[1]} ranged GETs', + ) + # b2sdk reuses the first, unranged response for the first share of the file and + # fetches the rest with Range requests, so both must be present and overlap. + if ranged[4] < 1: + raise CheckFailure('concurrency', '--threads 4 made no ranged GET') + if peaks[4] < 2: + raise CheckFailure( + 'concurrency', + f'--threads 4 never had two download streams in flight (peak {peaks[4]})', + ) def upload_fifo( self, step: str, bucket_name: str, name: str, payload: bytes, *, incremental: bool @@ -1487,7 +1576,7 @@ def urls_native_download(self) -> None: ).strip() if not prefix_token or '?' in prefix_token or 'x-amz-' in prefix_token.lower(): raise CheckFailure('native URL', 'prefix authorization is not a bare token') - with urlopen(authorized_url, timeout=10) as response: + with NO_PROXY_OPENER.open(authorized_url, timeout=10) as response: if response.status != 200 or response.read() != payload: raise CheckFailure('authorized fetch', 'response did not return fixture bytes') @@ -1502,7 +1591,7 @@ def urls_native_download(self) -> None: ).splitlines()[0] time.sleep(5) try: - urlopen(short_url, timeout=10) + NO_PROXY_OPENER.open(short_url, timeout=10) except HTTPError as error: if error.code != 401: raise CheckFailure( diff --git a/.sdkharness/tests/lib/observing_proxy.py b/.sdkharness/tests/lib/observing_proxy.py new file mode 100644 index 00000000..cdf69255 --- /dev/null +++ b/.sdkharness/tests/lib/observing_proxy.py @@ -0,0 +1,210 @@ +###################################################################### +# +# File: .sdkharness/tests/lib/observing_proxy.py +# +# Copyright 2026 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +"""A loopback forwarding proxy that observes how many requests overlap. + +The B2 CLI is a separate process, so a check cannot see its thread pool. What it +can see is the wire: the CLI is pointed at this proxy as its realm, the proxy +forwards everything to the simulator, and it counts requests that are in flight at +the same moment. Two ``b2_upload_part`` requests (or two ranged downloads) that are +open at once are real concurrency; a client that serialises its transfers can never +produce that, however many threads it was told to use. + +Overlap is only reliably observable if each request lasts long enough, and the +simulator answers in a few milliseconds, so the proxy can hold selected requests +for a fixed time before forwarding them (a client-side throttle). + +The proxy only ever forwards to the one loopback origin it was given, rewrites that +origin to its own in JSON responses (so the URLs the simulator returns keep leading +back through the proxy), and never logs a header or body. +""" + +from __future__ import annotations + +import http.client +import threading +import time +from collections.abc import Callable, Mapping, Sequence +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlsplit + +# Returns the labels a request is counted under (none: not observed). +Classifier = Callable[[str, str, Mapping[str, str]], Sequence[str]] + +_HOP_BY_HOP = { + 'connection', + 'keep-alive', + 'proxy-connection', + 'te', + 'trailer', + 'transfer-encoding', + 'upgrade', + 'host', + 'content-length', +} + + +def classify_transfer(method: str, path: str, headers: Mapping[str, str]) -> Sequence[str]: + """The labels a transfer request is counted under. + + ``upload_part`` for a part upload; ``download_stream`` for a file-body GET (by name or + by id); ``ranged_get`` in addition when that GET carries a Range header. + """ + pathname = urlsplit(path).path + if method == 'POST' and pathname.endswith('/b2_upload_part'): + return ('upload_part',) + if method == 'GET' and ( + pathname.startswith('/file/') or pathname.endswith('/b2_download_file_by_id') + ): + ranged = any(name.lower() == 'range' for name in headers) + return ('download_stream', 'ranged_get') if ranged else ('download_stream',) + return () + + +class ObservingProxy: + def __init__( + self, + upstream_origin: str, + *, + classify: Classifier = classify_transfer, + hold_seconds: float = 0.0, + ) -> None: + parsed = urlsplit(upstream_origin) + if parsed.scheme != 'http' or parsed.hostname != '127.0.0.1' or parsed.port is None: + raise ValueError('the proxy only forwards to a loopback HTTP origin') + self.upstream_host = parsed.hostname + self.upstream_port = parsed.port + self.upstream_origin = f'http://{parsed.hostname}:{parsed.port}' + self.classify = classify + self.hold_seconds = hold_seconds + self._lock = threading.Lock() + self._in_flight: dict[str, int] = {} + self._peak: dict[str, int] = {} + self._total: dict[str, int] = {} + self._server: ThreadingHTTPServer | None = None + self._thread: threading.Thread | None = None + + # -- lifecycle ---------------------------------------------------------- + + @property + def origin(self) -> str: + assert self._server is not None, 'the proxy is not running' + return f'http://127.0.0.1:{self._server.server_address[1]}' + + def __enter__(self) -> ObservingProxy: + proxy = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def log_message(self, *_args) -> None: # never log request lines + pass + + def _forward(self) -> None: + proxy._forward(self) + + do_GET = do_POST = do_PUT = do_DELETE = do_HEAD = _forward + + self._server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self._server.daemon_threads = True + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return self + + def __exit__(self, *_exc) -> None: + if self._server is not None: + self._server.shutdown() + self._server.server_close() + if self._thread is not None: + self._thread.join(timeout=5) + + # -- observation -------------------------------------------------------- + + def peak(self, label: str) -> int: + with self._lock: + return self._peak.get(label, 0) + + def total(self, label: str) -> int: + with self._lock: + return self._total.get(label, 0) + + def reset(self) -> None: + with self._lock: + self._peak.clear() + self._total.clear() + + def _enter(self, label: str) -> None: + with self._lock: + current = self._in_flight.get(label, 0) + 1 + self._in_flight[label] = current + self._peak[label] = max(self._peak.get(label, 0), current) + self._total[label] = self._total.get(label, 0) + 1 + + def _leave(self, label: str) -> None: + with self._lock: + self._in_flight[label] -= 1 + + # -- forwarding --------------------------------------------------------- + + def _forward(self, handler: BaseHTTPRequestHandler) -> None: + headers = {name: value for name, value in handler.headers.items()} + length = int(headers.get('Content-Length') or 0) + body = handler.rfile.read(length) if length else None + labels = tuple(self.classify(handler.command, handler.path, headers)) + for label in labels: + self._enter(label) + try: + if labels and self.hold_seconds: + time.sleep(self.hold_seconds) + try: + status, response_headers, payload = self._upstream( + handler.command, handler.path, headers, body + ) + except OSError: + handler.send_error(502) + return + handler.send_response(status) + for name, value in response_headers: + # send_response() already wrote Date and Server; a second Date is refused + if name.lower() not in _HOP_BY_HOP | {'date', 'server'}: + handler.send_header(name, value) + handler.send_header('Content-Length', str(len(payload))) + handler.send_header('Connection', 'close') + handler.end_headers() + if handler.command != 'HEAD': + # A request stays in flight until its response has been delivered (or the + # client hung up, as a parallel download does once its share is read). + try: + handler.wfile.write(payload) + except OSError: + pass + handler.close_connection = True + finally: + for label in labels: + self._leave(label) + + def _upstream(self, method, path, headers, body): + outgoing = {k: v for k, v in headers.items() if k.lower() not in _HOP_BY_HOP} + connection = http.client.HTTPConnection(self.upstream_host, self.upstream_port, timeout=120) + try: + connection.request(method, path, body=body, headers=outgoing) + response = connection.getresponse() + payload = response.read() if method != 'HEAD' else b'' + response_headers = response.getheaders() + if method == 'HEAD': + # keep the upstream's own Content-Length: there is no body to measure + payload = b'' + elif any( + name.lower() == 'content-type' and 'json' in value.lower() + for name, value in response_headers + ): + payload = payload.replace(self.upstream_origin.encode(), self.origin.encode()) + return response.status, response_headers, payload + finally: + connection.close() diff --git a/.sdkharness/tests/lib/simulator_guard.py b/.sdkharness/tests/lib/simulator_guard.py index 84802306..ebdbfeaf 100644 --- a/.sdkharness/tests/lib/simulator_guard.py +++ b/.sdkharness/tests/lib/simulator_guard.py @@ -27,6 +27,24 @@ def credential_is_fixed(key_id: str, application_key: str) -> bool: return (key_id, application_key) == SIMULATOR_CREDENTIAL +# Variables that route (or exempt) HTTP traffic through a proxy, in either case. +PROXY_VARIABLES = frozenset( + {'http_proxy', 'https_proxy', 'all_proxy', 'no_proxy', 'ftp_proxy', 'grpc_proxy'} +) + + def scrubbed_environment(environment: Mapping[str, str]) -> dict[str, str]: - """A copy of ``environment`` without any ``B2_*`` value (set explicitly afterwards).""" - return {name: value for name, value in environment.items() if not name.startswith('B2_')} + """A copy of ``environment`` without any ``B2_*`` or proxy variable. + + ``B2_*`` values are set explicitly afterwards. Proxy variables are dropped because a + proxy configured on a developer or CI machine would otherwise carry the check's + loopback traffic (and the fixed credential) to a third party, or break it. Loopback + is then exempted explicitly, so nothing a child process inherits can reroute it. + """ + kept = { + name: value + for name, value in environment.items() + if not name.startswith('B2_') and name.lower() not in PROXY_VARIABLES + } + kept['NO_PROXY'] = kept['no_proxy'] = '127.0.0.1' + return kept diff --git a/.sdkharness/tests/resilience-upload.py b/.sdkharness/tests/resilience-upload.py index f395c7d9..fa75815c 100755 --- a/.sdkharness/tests/resilience-upload.py +++ b/.sdkharness/tests/resilience-upload.py @@ -17,6 +17,18 @@ import time import uuid from collections.abc import Mapping +from pathlib import Path + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] + +# Resolve the CLI implementation from this checkout, as the other executables do, and +# follow the latest stable apiver rather than a hard-coded one. +sys.path.insert(0, str(REPOSITORY_ROOT)) +sys.path.insert(0, str(Path(__file__).resolve().parent / 'lib')) +from simulator_guard import scrubbed_environment # noqa: E402 + +import b2 # noqa: E402 +from b2._internal.version_listing import LATEST_STABLE_VERSION # noqa: E402 LEVEL = 'resilience' SCENARIOS = { @@ -87,7 +99,7 @@ SCENARIO = '' OBJECT_NAME = '' FAULT: dict[str, object] = {} -CLI_PREFIX = [sys.executable, '-m', 'b2._internal.b2v5'] +CLI_PREFIX = [sys.executable, '-m', f'b2._internal.{LATEST_STABLE_VERSION}'] ARM_PATH = '/faults' @@ -192,8 +204,12 @@ class Cli: def __init__(self, scratch: str) -> None: self.scratch = scratch - self.env = {k: v for k, v in os.environ.items() if not k.startswith('B2_')} + # No ambient B2_* value or proxy variable reaches the CLI (see simulator_guard). + self.env = scrubbed_environment(os.environ) self.env.update( + PYTHONPATH=os.pathsep.join( + filter(None, (str(REPOSITORY_ROOT), os.environ.get('PYTHONPATH', ''))) + ), B2_APPLICATION_KEY_ID='test-key-id', B2_APPLICATION_KEY='test-key', B2_ENVIRONMENT=f"http://127.0.0.1:{loopback_port(os.environ, 'SDKHARNESS_SIMULATOR_URL')}", @@ -205,6 +221,7 @@ def __init__(self, scratch: str) -> None: def run(self, *args): return subprocess.run( [*CLI_PREFIX, *args], + cwd=REPOSITORY_ROOT, env=self.env, capture_output=True, text=True, @@ -242,6 +259,25 @@ def json_of(text: str): return json.loads(text[min(starts) :]) +def authorize(cli, attempts: int = 3) -> None: + """Authorize the CLI, retrying the setup step only. + + Authorizing is idempotent and is not what any scenario tests, so a transient failure of + the very first CLI start (a cold interpreter and import cache, or a listener that has + printed its URL a moment before it accepts) must not be reported as an SDK failure. + A retry is announced, and a persistent failure still fails with the CLI's own error. + """ + for attempt in range(1, attempts + 1): + try: + cli.must('authenticate', 'account', 'authorize') + return + except Failure: + if attempt == attempts: + raise + note(f'authorize attempt {attempt} failed; retrying (setup step, not under test)') + time.sleep(1.0) + + def setup(): """An authorized CLI and a fresh bucket. There is no unreachable/unauthorized amber: the runner supplies the server and the credential, so either failing @@ -249,7 +285,7 @@ def setup(): scratch = tempfile.mkdtemp(prefix='sdkharness-res-cli.') CLEANUP.append(scratch) cli = Cli(scratch) - cli.must('authenticate', 'account', 'authorize') + authorize(cli) bucket = f'sdkharness-res-{uuid.uuid4().hex[:12]}' cli.must('create bucket', 'bucket', 'create', bucket, 'allPrivate') note( @@ -270,7 +306,7 @@ def put(cli, bucket, name, payload, *flags): raise Failure('upload', type(error).__name__) from error -def uploaded_or_fail(proc, payload, endpoint='b2_upload_file'): +def uploaded_or_fail(proc, payload, endpoint='b2_upload_file', large_file=False): """An upload process that must have recovered and returned the right file record. It checks the exit status, that a JSON file record came back, and that the @@ -279,6 +315,11 @@ def uploaded_or_fail(proc, payload, endpoint='b2_upload_file'): kept because a CLI that stores the right bytes but reports a wrong checksum in its response must still fail. (SHA-1 is B2's wire checksum here, not a security primitive.) + + A large (multipart) file legitimately reports ``contentSha1: none`` -- B2 has no single + checksum for it and carries ``large_file_sha1`` in ``fileInfo`` instead -- so ``none`` is + accepted only when the caller says the upload is a large file, and then the digest in + ``fileInfo`` must match. Every small-file scenario requires the real checksum. """ if proc.returncode != 0: error_note(proc) @@ -292,7 +333,12 @@ def uploaded_or_fail(proc, payload, endpoint='b2_upload_file'): except Exception as error: # noqa: BLE001 raise Failure('upload', 'b2 file upload printed no JSON file record') from error got = (meta.get('contentSha1') or '').split(':')[-1] - if got not in ('none', hashlib.sha1(payload).hexdigest()): + expected = hashlib.sha1(payload).hexdigest() + if large_file and got == 'none': + info = meta.get('fileInfo') or {} + if not isinstance(info, dict) or info.get('large_file_sha1') != expected: + raise Failure('upload', 'the large file record has no matching large_file_sha1') + elif got != expected: raise Failure('upload', 'the returned contentSha1 does not match the bytes sent') return meta @@ -314,6 +360,14 @@ def listed(cli, bucket): CLEANUP = [] +def check_checkout() -> None: + """The CLI under test must be this checkout, not an installed distribution.""" + try: + Path(b2.__file__).resolve().relative_to(REPOSITORY_ROOT) + except ValueError as error: + raise Failure('setup', 'b2 CLI was not imported from this checkout') from error + + def validate_environment(environment: Mapping[str, str]) -> str: if environment.get('SDKHARNESS_TEST_LEVEL') != LEVEL: raise Failure('configuration', 'unexpected test level') @@ -518,6 +572,7 @@ def run_transfer_fault() -> None: ), payload, endpoint, + large_file=True, ) round_trip(cli, f'b2://{bucket}/{OBJECT_NAME}', payload) @@ -568,7 +623,11 @@ def run_transfer_fault() -> None: def main() -> int: global SCENARIO, OBJECT_NAME, FAULT, ARM_PATH signal.signal(signal.SIGTERM, lambda *_: sys.exit(143)) + # Name the scenario in the result line even when the configuration is refused. + requested = os.environ.get('SDKHARNESS_SCENARIO', '') + SCENARIO = requested if requested in ALL_SCENARIOS else 'unknown' try: + check_checkout() SCENARIO = validate_environment(os.environ) OBJECT_NAME = f"res/{SCENARIO.replace('.', '-')}.bin" if SCENARIO in WIRE_SCENARIOS: diff --git a/changelog.d/+sdkharness-migration.infrastructure.md b/changelog.d/+sdkharness-migration.infrastructure.md new file mode 100644 index 00000000..9457b8d2 --- /dev/null +++ b/changelog.d/+sdkharness-migration.infrastructure.md @@ -0,0 +1 @@ +Recorded the repository-owned sdkharness conformance and resilience checks moved into `.sdkharness/` by #1171 (bucket contracts), #1172 (encryption), #1173 (file lock), #1174 (client and key), #1175 (large-transfer and URL), #1176 (upload resilience retries) and #1177 (upload resilience wire faults and the remaining scenarios), which were merged without individual changelog fragments. diff --git a/changelog.d/+sdkharness-observe-concurrency.infrastructure.md b/changelog.d/+sdkharness-observe-concurrency.infrastructure.md new file mode 100644 index 00000000..6202ac27 --- /dev/null +++ b/changelog.d/+sdkharness-observe-concurrency.infrastructure.md @@ -0,0 +1 @@ +The repository-owned sdkharness conformance checks `large.concurrent_parts` and `large.parallel_download` now observe real concurrency (overlapping `b2_upload_part` requests and overlapping download streams including ranged GETs) through a loopback observing proxy instead of only comparing sizes and bytes, and `large.concurrent_parts` also downloads and verifies the uploaded bytes. The resilience executable now follows `LATEST_STABLE_VERSION`, accepts `contentSha1: none` only for large files, names its scenario in the result line when the configuration is refused, and retries only its own authorize setup step. All executables scrub proxy variables from the CLI environment, and unit tests cover the loopback and fixed-credential guards. diff --git a/test/unit/test_sdkharness_guards.py b/test/unit/test_sdkharness_guards.py new file mode 100644 index 00000000..8d4ff2b6 --- /dev/null +++ b/test/unit/test_sdkharness_guards.py @@ -0,0 +1,401 @@ +###################################################################### +# +# File: test/unit/test_sdkharness_guards.py +# +# Copyright 2026 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +"""Tests of the safety guards in the repository-owned ``.sdkharness/`` checks. + +The checks must only ever talk to a loopback simulator with its fixed credential and must +never hand an ambient ``B2_*`` value or proxy setting to the CLI. These tests keep those +guards from regressing silently, and exercise the helpers that decide what a result means. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.request import urlopen + +import pytest + +TESTS_DIR = Path(__file__).resolve().parents[2] / '.sdkharness' / 'tests' +sys.path.insert(0, str(TESTS_DIR / 'lib')) + +import observing_proxy # noqa: E402 +import simulator_guard # noqa: E402 + + +def load(filename: str): + spec = importlib.util.spec_from_file_location( + 'sdkharness_' + filename.replace('-', '_').removesuffix('.py'), TESTS_DIR / filename + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +lifecycle = load('conformance-file-lifecycle.py') +files_upload = load('conformance-files-upload.py') +health = load('health-golden-path.py') +resilience = load('resilience-upload.py') + +REAL_LOOKING_KEY_ID = '0' * 12 + 'REALKEYID' +REAL_LOOKING_KEY = 'K' + '1' * 30 +GOOD_URL = 'http://127.0.0.1:4321' + +# (module, scenario, extra environment) for each executable that validates a simulator URL +# and a credential pair before it runs the CLI. +CLI_CHECKS = [ + pytest.param(lifecycle, 'files.hide', {}, id='conformance-file-lifecycle'), + pytest.param(files_upload, 'files.upload', {}, id='conformance-files-upload'), + pytest.param( + health, 'golden-path', {'HEALTHCHECK_REALM_URL': GOOD_URL}, id='health-golden-path' + ), +] + + +def environment_for(module, scenario, extra, **overrides): + environment = { + 'SDKHARNESS_TEST_LEVEL': module.LEVEL, + 'SDKHARNESS_SCENARIO': scenario, + 'SDKHARNESS_SIMULATOR_URL': GOOD_URL, + 'B2_TEST_APPLICATION_KEY_ID': 'test-key-id', + 'B2_TEST_APPLICATION_KEY': 'test-key', + 'B2_BUCKET_NAME': 'sdkharness-conformance', + **extra, + } + environment.update(overrides) + if 'SDKHARNESS_SIMULATOR_URL' in overrides and 'HEALTHCHECK_REALM_URL' in extra: + environment['HEALTHCHECK_REALM_URL'] = overrides['SDKHARNESS_SIMULATOR_URL'] + return environment + + +# -- simulator_guard --------------------------------------------------------------------- + + +def test_only_the_fixed_credential_is_accepted(): + assert simulator_guard.credential_is_fixed('test-key-id', 'test-key') + assert not simulator_guard.credential_is_fixed(REAL_LOOKING_KEY_ID, REAL_LOOKING_KEY) + assert not simulator_guard.credential_is_fixed('test-key-id', REAL_LOOKING_KEY) + assert not simulator_guard.credential_is_fixed(REAL_LOOKING_KEY_ID, 'test-key') + assert not simulator_guard.credential_is_fixed('', '') + + +def test_scrubbed_environment_drops_b2_and_proxy_variables(): + environment = { + 'PATH': '/usr/bin', + 'B2_APPLICATION_KEY_ID': 'ambient-id', + 'B2_APPLICATION_KEY': 'ambient-key', + 'B2_ENVIRONMENT': 'https://example.invalid', + 'HTTP_PROXY': 'http://proxy.invalid:8080', + 'https_proxy': 'http://proxy.invalid:8080', + 'ALL_PROXY': 'socks5://proxy.invalid', + 'NO_PROXY': 'example.com', + 'no_proxy': 'example.com', + } + original = dict(environment) + + scrubbed = simulator_guard.scrubbed_environment(environment) + + assert environment == original, 'the caller environment must not be modified' + assert scrubbed['PATH'] == '/usr/bin' + assert not [name for name in scrubbed if name.startswith('B2_')] + assert not {name for name in scrubbed if name.lower().endswith('_proxy')} - { + 'NO_PROXY', + 'no_proxy', + } + # loopback is exempted explicitly, whatever the ambient exemptions were + assert scrubbed['NO_PROXY'] == scrubbed['no_proxy'] == '127.0.0.1' + + +# -- the loopback and credential guards of every CLI-driving executable -------------------- + + +@pytest.mark.parametrize('module, scenario, extra', CLI_CHECKS) +def test_validate_environment_accepts_the_simulator_setup(module, scenario, extra): + result = module.validate_environment(environment_for(module, scenario, extra)) + assert GOOD_URL in result + assert 'test-key-id' in result and 'test-key' in result + + +@pytest.mark.parametrize( + 'url', + [ + 'https://127.0.0.1:4321', # TLS: not the plain loopback listener + 'http://localhost:4321', # a name, not the IPv4 literal + 'http://[::1]:4321', + 'http://0.0.0.0:4321', + 'http://127.0.0.2:4321', + 'http://192.168.1.10:4321', + 'https://api.backblazeb2.com', + 'http://127.0.0.1', # no port + 'http://user:pass@127.0.0.1:4321', + 'http://127.0.0.1:4321/b2api', + 'http://127.0.0.1:4321/?next=elsewhere', + 'http://127.0.0.1:4321/#fragment', + 'http://127.0.0.1:notaport', + '', + ], +) +@pytest.mark.parametrize('module, scenario, extra', CLI_CHECKS) +def test_validate_environment_refuses_a_non_loopback_url(module, scenario, extra, url): + environment = environment_for(module, scenario, extra, SDKHARNESS_SIMULATOR_URL=url) + with pytest.raises(module.CheckFailure) as refused: + module.validate_environment(environment) + assert refused.value.step == 'configuration' + + +@pytest.mark.parametrize( + 'key_id, application_key', + [ + (REAL_LOOKING_KEY_ID, REAL_LOOKING_KEY), + ('test-key-id', REAL_LOOKING_KEY), + (REAL_LOOKING_KEY_ID, 'test-key'), + ], +) +@pytest.mark.parametrize('module, scenario, extra', CLI_CHECKS) +def test_validate_environment_refuses_a_real_credential_without_echoing_it( + module, scenario, extra, key_id, application_key +): + environment = environment_for( + module, + scenario, + extra, + B2_TEST_APPLICATION_KEY_ID=key_id, + B2_TEST_APPLICATION_KEY=application_key, + ) + with pytest.raises(module.CheckFailure) as refused: + module.validate_environment(environment) + assert refused.value.detail == simulator_guard.CREDENTIAL_REFUSAL + message = str(refused.value) + assert REAL_LOOKING_KEY not in message and REAL_LOOKING_KEY_ID not in message + + +@pytest.mark.parametrize('module, scenario, extra', CLI_CHECKS) +@pytest.mark.parametrize( + 'missing', ['B2_TEST_APPLICATION_KEY_ID', 'B2_TEST_APPLICATION_KEY', 'B2_BUCKET_NAME'] +) +def test_validate_environment_requires_every_input(module, scenario, extra, missing): + environment = environment_for(module, scenario, extra) + del environment[missing] + with pytest.raises(module.CheckFailure) as refused: + module.validate_environment(environment) + assert refused.value.step == 'configuration' + + +def test_the_health_check_refuses_a_realm_that_is_not_the_simulator_url(): + environment = environment_for( + health, 'golden-path', {'HEALTHCHECK_REALM_URL': 'https://api.backblazeb2.com'} + ) + with pytest.raises(health.CheckFailure, match='mismatch'): + health.validate_environment(environment) + + +def test_the_lifecycle_check_refuses_an_unknown_scenario_or_level(): + with pytest.raises(lifecycle.CheckFailure): + lifecycle.validate_environment( + environment_for(lifecycle, 'files.hide', {}, SDKHARNESS_SCENARIO='bucket.nope') + ) + with pytest.raises(lifecycle.CheckFailure): + lifecycle.validate_environment( + environment_for(lifecycle, 'files.hide', {}, SDKHARNESS_TEST_LEVEL='resilience') + ) + + +# -- the resilience executable --------------------------------------------------------------- + + +def resilience_environment(**overrides): + environment = { + 'SDKHARNESS_TEST_LEVEL': 'resilience', + 'SDKHARNESS_SCENARIO': 'upload.retry_503', + 'SDKHARNESS_SIMULATOR_URL': GOOD_URL, + 'SDKHARNESS_SIMULATOR_CONTROL_URL': 'http://127.0.0.1:4322', + } + environment.update(overrides) + return environment + + +def test_resilience_accepts_loopback_simulator_and_control_urls(): + assert resilience.validate_environment(resilience_environment()) == 'upload.retry_503' + + +@pytest.mark.parametrize('name', ['SDKHARNESS_SIMULATOR_URL', 'SDKHARNESS_SIMULATOR_CONTROL_URL']) +@pytest.mark.parametrize( + 'url', + ['https://127.0.0.1:1', 'http://localhost:1', 'http://[::1]:1', 'http://10.0.0.1:1', '', 'x'], +) +def test_resilience_refuses_a_non_loopback_url(name, url): + with pytest.raises(resilience.Failure) as refused: + resilience.validate_environment(resilience_environment(**{name: url})) + assert refused.value.step == 'configuration' + + +def test_resilience_refuses_an_unknown_scenario(): + with pytest.raises(resilience.Failure): + resilience.validate_environment(resilience_environment(SDKHARNESS_SCENARIO='upload.nope')) + + +def test_resilience_names_the_scenario_in_the_result_line_when_configuration_fails( + monkeypatch, capsys +): + for name, value in resilience_environment(SDKHARNESS_SIMULATOR_URL='https://x.invalid').items(): + monkeypatch.setenv(name, value) + assert resilience.main() == 1 + fields = capsys.readouterr().out.strip().split('\t') + assert fields[:4] == ['SDKHARNESS_RESULT', 'resilience', 'upload.retry_503', 'FAIL'] + assert fields[4].startswith('configuration') + + +def test_resilience_does_not_echo_an_unknown_scenario(monkeypatch, capsys): + for name, value in resilience_environment(SDKHARNESS_SCENARIO='bad\tname').items(): + monkeypatch.setenv(name, value) + assert resilience.main() == 1 + fields = capsys.readouterr().out.strip().split('\t') + assert len(fields) == 5 and fields[2] == 'unknown' + + +def test_resilience_follows_the_latest_stable_apiver(): + from b2._internal.version_listing import LATEST_STABLE_VERSION + + assert resilience.CLI_PREFIX[-1] == f'b2._internal.{LATEST_STABLE_VERSION}' + + +class FakeProcess: + returncode = 0 + stderr = '' + + def __init__(self, record): + self.stdout = json.dumps(record) + + +def test_a_small_file_must_report_its_real_sha1(): + import hashlib + + payload = b'small payload' + real = hashlib.sha1(payload).hexdigest() + resilience.uploaded_or_fail(FakeProcess({'contentSha1': real}), payload) + resilience.uploaded_or_fail(FakeProcess({'contentSha1': f'unverified:{real}'}), payload) + for reported in ('none', '', '0' * 40): + with pytest.raises(resilience.Failure, match='contentSha1'): + resilience.uploaded_or_fail(FakeProcess({'contentSha1': reported}), payload) + + +def test_none_is_accepted_only_for_a_large_file_with_a_matching_digest(): + import hashlib + + payload = b'large payload' + digest = hashlib.sha1(payload).hexdigest() + record = {'contentSha1': 'none', 'fileInfo': {'large_file_sha1': digest}} + resilience.uploaded_or_fail(FakeProcess(record), payload, large_file=True) + with pytest.raises(resilience.Failure): # not a documented large-file case + resilience.uploaded_or_fail(FakeProcess(record), payload) + wrong = {'contentSha1': 'none', 'fileInfo': {'large_file_sha1': '0' * 40}} + with pytest.raises(resilience.Failure, match='large_file_sha1'): + resilience.uploaded_or_fail(FakeProcess(wrong), payload, large_file=True) + with pytest.raises(resilience.Failure, match='large_file_sha1'): + resilience.uploaded_or_fail(FakeProcess({'contentSha1': 'none'}), payload, large_file=True) + + +def test_resilience_child_environment_has_no_ambient_b2_or_proxy_values(monkeypatch, tmp_path): + monkeypatch.setenv('B2_APPLICATION_KEY', REAL_LOOKING_KEY) + monkeypatch.setenv('HTTPS_PROXY', 'http://proxy.invalid:8080') + monkeypatch.setenv('SDKHARNESS_SIMULATOR_URL', GOOD_URL) + cli = resilience.Cli(str(tmp_path)) + assert cli.env['B2_APPLICATION_KEY'] == 'test-key' + assert REAL_LOOKING_KEY not in cli.env.values() + assert 'HTTPS_PROXY' not in cli.env + assert cli.env['NO_PROXY'] == '127.0.0.1' + + +# -- the observing proxy --------------------------------------------------------------------- + + +class Upstream(BaseHTTPRequestHandler): + def log_message(self, *_args): + pass + + def do_POST(self): + self.rfile.read(int(self.headers.get('Content-Length') or 0)) + body = json.dumps({'self': self.server.origin}).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + do_GET = do_POST + + +@pytest.fixture +def upstream(): + server = ThreadingHTTPServer(('127.0.0.1', 0), Upstream) + server.origin = f'http://127.0.0.1:{server.server_address[1]}' + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield server + server.shutdown() + server.server_close() + + +@pytest.mark.parametrize( + 'origin', + ['https://127.0.0.1:1', 'http://localhost:1', 'http://example.com:80', 'http://127.0.0.1'], +) +def test_the_proxy_forwards_only_to_a_loopback_origin(origin): + with pytest.raises(ValueError): + observing_proxy.ObservingProxy(origin) + + +def test_classify_transfer_labels_part_uploads_and_downloads(): + classify = observing_proxy.classify_transfer + assert classify('POST', '/b2api/v4/b2_upload_part', {}) == ('upload_part',) + assert classify('POST', '/b2api/v4/b2_upload_file', {}) == () + assert classify('GET', '/file/bucket/name', {}) == ('download_stream',) + assert classify('GET', '/file/bucket/name', {'range': 'bytes=0-9'}) == ( + 'download_stream', + 'ranged_get', + ) + assert classify('GET', '/b2api/v4/b2_download_file_by_id?fileId=x', {'Range': 'bytes=0-9'}) == ( + 'download_stream', + 'ranged_get', + ) + assert classify('POST', '/b2api/v4/b2_list_buckets', {}) == () + + +def fire(url, count): + threads = [ + threading.Thread(target=lambda: urlopen(url, data=b'x', timeout=10).read()) + for _ in range(count) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + +def test_the_proxy_observes_overlapping_requests_and_rewrites_the_origin(upstream): + with observing_proxy.ObservingProxy(upstream.origin, hold_seconds=0.3) as proxy: + fire(f'{proxy.origin}/b2api/v4/b2_upload_part', 3) + assert proxy.peak('upload_part') == 3 + assert proxy.total('upload_part') == 3 + reply = json.load(urlopen(f'{proxy.origin}/b2api/v4/b2_list_buckets', data=b'', timeout=10)) + assert reply == {'self': proxy.origin}, 'JSON URLs must lead back through the proxy' + + +def test_the_proxy_sees_no_overlap_for_serial_requests(upstream): + with observing_proxy.ObservingProxy(upstream.origin, hold_seconds=0.05) as proxy: + for _ in range(3): + fire(f'{proxy.origin}/b2api/v4/b2_upload_part', 1) + assert proxy.peak('upload_part') == 1 + assert proxy.total('upload_part') == 3 + proxy.reset() + assert proxy.peak('upload_part') == 0 From e7d93f745a7c315c13ce2bd95a43d49ef8946d2a Mon Sep 17 00:00:00 2001 From: scarreras Date: Sat, 3 Oct 2026 21:54:13 -0300 Subject: [PATCH 2/3] Fix existing resilience test for large-file-only sha1 none; harden observing proxy - Update test_sdkharness_resilience for the stricter contentSha1 rule and add a large-file positive case. - observing_proxy: forward only origin-form paths, validate Content-Length, stream non-JSON upstream bodies instead of buffering them. - Merge the two changelog fragments into one; rewrap the README paragraph. Co-Authored-By: Claude Sonnet 5.5 --- .sdkharness/README.md | 12 +-- .sdkharness/tests/lib/observing_proxy.py | 84 +++++++++++-------- .../+sdkharness-migration.infrastructure.md | 1 - ...ness-observe-concurrency.infrastructure.md | 2 +- test/unit/test_sdkharness_resilience.py | 17 +++- 5 files changed, 72 insertions(+), 44 deletions(-) delete mode 100644 changelog.d/+sdkharness-migration.infrastructure.md diff --git a/.sdkharness/README.md b/.sdkharness/README.md index 92c74723..b562d491 100644 --- a/.sdkharness/README.md +++ b/.sdkharness/README.md @@ -82,12 +82,12 @@ Each prints one `SDKHARNESS_RESULT` line and refuses any simulator URL that is n `http://127.0.0.1:`. The CLI child never inherits proxy variables (`HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY`, `NO_PROXY`, in either case); loopback is exempted explicitly. -`large.concurrent_parts` and `large.parallel_download` observe concurrency on the wire: they -point the CLI at a small loopback proxy (`tests/lib/observing_proxy.py`) in front of the -simulator, hold each part upload / download stream for half a second so overlap is visible, -and require two or more in flight with `--threads 4` and exactly one with `--threads 1`. -`large.parallel_download` uses a file of just over 200 MiB because b2sdk starts one ranged -stream per 100 MiB of content and the CLI has no option to lower that. +`large.concurrent_parts` and `large.parallel_download` observe concurrency on the wire. +They point the CLI at a small loopback proxy (`tests/lib/observing_proxy.py`) in front of +the simulator, hold each part upload / download stream for half a second so overlap is +visible, and require two or more in flight with `--threads 4` and exactly one with +`--threads 1`. `large.parallel_download` uses a file of just over 200 MiB because b2sdk +starts one ranged stream per 100 MiB of content and the CLI has no option to lower that. The guards are covered by `test/unit/test_sdkharness_guards.py` (run with the unit tests). Use a fresh simulator per resilience scenario: an unconsumed injected fault from a failing scenario (for example `upload.retry_408`) diff --git a/.sdkharness/tests/lib/observing_proxy.py b/.sdkharness/tests/lib/observing_proxy.py index cdf69255..6f018f30 100644 --- a/.sdkharness/tests/lib/observing_proxy.py +++ b/.sdkharness/tests/lib/observing_proxy.py @@ -153,58 +153,72 @@ def _leave(self, label: str) -> None: # -- forwarding --------------------------------------------------------- def _forward(self, handler: BaseHTTPRequestHandler) -> None: + # Only an origin-form path ("/...") is forwarded, and only to the one fixed loopback + # upstream chosen at construction: a client cannot name another host through the + # request line, a "//host" path, or a Host header (which is never forwarded). + path = handler.path + if not path.startswith('/') or path.startswith('//') or any(c in path for c in '\r\n'): + handler.send_error(400) + return headers = {name: value for name, value in handler.headers.items()} - length = int(headers.get('Content-Length') or 0) + try: + length = int(headers.get('Content-Length') or 0) + except ValueError: + length = -1 + if length < 0: + handler.send_error(400) + return body = handler.rfile.read(length) if length else None - labels = tuple(self.classify(handler.command, handler.path, headers)) + labels = tuple(self.classify(handler.command, path, headers)) for label in labels: self._enter(label) + connection = http.client.HTTPConnection(self.upstream_host, self.upstream_port, timeout=120) try: if labels and self.hold_seconds: time.sleep(self.hold_seconds) + outgoing = {k: v for k, v in headers.items() if k.lower() not in _HOP_BY_HOP} try: - status, response_headers, payload = self._upstream( - handler.command, handler.path, headers, body - ) + connection.request(handler.command, path, body=body, headers=outgoing) + response = connection.getresponse() except OSError: handler.send_error(502) return - handler.send_response(status) - for name, value in response_headers: + is_json = any( + name.lower() == 'content-type' and 'json' in value.lower() + for name, value in response.getheaders() + ) + rewritten = None + if is_json and handler.command != 'HEAD': + # small API documents: buffered so the simulator origin can be replaced + rewritten = response.read().replace( + self.upstream_origin.encode(), self.origin.encode() + ) + handler.send_response(response.status) + for name, value in response.getheaders(): # send_response() already wrote Date and Server; a second Date is refused if name.lower() not in _HOP_BY_HOP | {'date', 'server'}: handler.send_header(name, value) - handler.send_header('Content-Length', str(len(payload))) + if rewritten is not None: + handler.send_header('Content-Length', str(len(rewritten))) + elif response.getheader('Content-Length') is not None: + handler.send_header('Content-Length', response.getheader('Content-Length')) handler.send_header('Connection', 'close') handler.end_headers() - if handler.command != 'HEAD': - # A request stays in flight until its response has been delivered (or the - # client hung up, as a parallel download does once its share is read). - try: - handler.wfile.write(payload) - except OSError: - pass handler.close_connection = True + if handler.command == 'HEAD': + return + # A request stays in flight until its response has been delivered (or the + # client hung up, as a parallel download does once its share is read). File + # bodies are streamed, not held in memory. + try: + if rewritten is not None: + handler.wfile.write(rewritten) + else: + while chunk := response.read(1024 * 1024): + handler.wfile.write(chunk) + except OSError: + pass finally: + connection.close() for label in labels: self._leave(label) - - def _upstream(self, method, path, headers, body): - outgoing = {k: v for k, v in headers.items() if k.lower() not in _HOP_BY_HOP} - connection = http.client.HTTPConnection(self.upstream_host, self.upstream_port, timeout=120) - try: - connection.request(method, path, body=body, headers=outgoing) - response = connection.getresponse() - payload = response.read() if method != 'HEAD' else b'' - response_headers = response.getheaders() - if method == 'HEAD': - # keep the upstream's own Content-Length: there is no body to measure - payload = b'' - elif any( - name.lower() == 'content-type' and 'json' in value.lower() - for name, value in response_headers - ): - payload = payload.replace(self.upstream_origin.encode(), self.origin.encode()) - return response.status, response_headers, payload - finally: - connection.close() diff --git a/changelog.d/+sdkharness-migration.infrastructure.md b/changelog.d/+sdkharness-migration.infrastructure.md deleted file mode 100644 index 9457b8d2..00000000 --- a/changelog.d/+sdkharness-migration.infrastructure.md +++ /dev/null @@ -1 +0,0 @@ -Recorded the repository-owned sdkharness conformance and resilience checks moved into `.sdkharness/` by #1171 (bucket contracts), #1172 (encryption), #1173 (file lock), #1174 (client and key), #1175 (large-transfer and URL), #1176 (upload resilience retries) and #1177 (upload resilience wire faults and the remaining scenarios), which were merged without individual changelog fragments. diff --git a/changelog.d/+sdkharness-observe-concurrency.infrastructure.md b/changelog.d/+sdkharness-observe-concurrency.infrastructure.md index 6202ac27..e01d8e47 100644 --- a/changelog.d/+sdkharness-observe-concurrency.infrastructure.md +++ b/changelog.d/+sdkharness-observe-concurrency.infrastructure.md @@ -1 +1 @@ -The repository-owned sdkharness conformance checks `large.concurrent_parts` and `large.parallel_download` now observe real concurrency (overlapping `b2_upload_part` requests and overlapping download streams including ranged GETs) through a loopback observing proxy instead of only comparing sizes and bytes, and `large.concurrent_parts` also downloads and verifies the uploaded bytes. The resilience executable now follows `LATEST_STABLE_VERSION`, accepts `contentSha1: none` only for large files, names its scenario in the result line when the configuration is refused, and retries only its own authorize setup step. All executables scrub proxy variables from the CLI environment, and unit tests cover the loopback and fixed-credential guards. +The repository-owned sdkharness conformance checks `large.concurrent_parts` and `large.parallel_download` now observe real concurrency (overlapping `b2_upload_part` requests and overlapping download streams including ranged GETs) through a loopback observing proxy instead of only comparing sizes and bytes, and `large.concurrent_parts` also downloads and verifies the uploaded bytes. The resilience executable now follows `LATEST_STABLE_VERSION`, accepts `contentSha1: none` only for large files, names its scenario in the result line when the configuration is refused, and retries only its own authorize setup step. All executables scrub proxy variables from the CLI environment, and unit tests cover the loopback and fixed-credential guards. This also records that the repository-owned conformance and resilience checks moved into `.sdkharness/` by #1171 (bucket), #1172 (encryption), #1173 (file lock), #1174 (client and key), #1175 (large-transfer and URL), #1176 and #1177 (upload resilience) were merged without individual changelog fragments. diff --git a/test/unit/test_sdkharness_resilience.py b/test/unit/test_sdkharness_resilience.py index 7922243f..f7046573 100644 --- a/test/unit/test_sdkharness_resilience.py +++ b/test/unit/test_sdkharness_resilience.py @@ -145,13 +145,28 @@ def test_upload_must_report_the_sha1_of_the_bytes_that_were_sent(): right = hashlib.sha1(PAYLOAD).hexdigest() assert check.uploaded_or_fail(FakeProcess(file_record(right)), PAYLOAD)['fileId'] == '4_zabc' assert check.uploaded_or_fail(FakeProcess(file_record(f'unverified:{right}')), PAYLOAD) - assert check.uploaded_or_fail(FakeProcess(file_record('none')), PAYLOAD) + with pytest.raises(check.Failure, match='contentSha1 does not match'): + # a small file has a real checksum; 'none' is only valid for a large file + check.uploaded_or_fail(FakeProcess(file_record('none')), PAYLOAD) with pytest.raises(check.Failure) as caught: check.uploaded_or_fail(FakeProcess(file_record('0' * 40)), PAYLOAD) assert caught.value.step == 'upload' assert 'contentSha1 does not match' in caught.value.detail +def test_a_large_file_may_report_no_sha1_only_with_a_matching_large_file_sha1(): + check = load_check() + digest = hashlib.sha1(PAYLOAD).hexdigest() + record = json.dumps( + {'fileId': '4_zabc', 'contentSha1': 'none', 'fileInfo': {'large_file_sha1': digest}} + ) + assert ( + check.uploaded_or_fail(FakeProcess(record), PAYLOAD, large_file=True)['fileId'] == '4_zabc' + ) + with pytest.raises(check.Failure, match='large_file_sha1'): + check.uploaded_or_fail(FakeProcess(file_record('none')), PAYLOAD, large_file=True) + + def test_upload_must_exit_zero_and_print_a_json_record(): check = load_check() check.journal = lambda: [] # the control listener is not running in a unit test From 685dc497f8bde0374dcaf5edcae0210d4dc9492b Mon Sep 17 00:00:00 2001 From: scarreras Date: Sat, 3 Oct 2026 22:45:51 -0300 Subject: [PATCH 3/3] Restrict the observing proxy to RFC 3986 request targets Co-Authored-By: Claude Sonnet 5.5 --- .sdkharness/tests/lib/observing_proxy.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/.sdkharness/tests/lib/observing_proxy.py b/.sdkharness/tests/lib/observing_proxy.py index 6f018f30..6aaf7db2 100644 --- a/.sdkharness/tests/lib/observing_proxy.py +++ b/.sdkharness/tests/lib/observing_proxy.py @@ -28,6 +28,7 @@ from __future__ import annotations import http.client +import re import threading import time from collections.abc import Callable, Mapping, Sequence @@ -37,6 +38,9 @@ # Returns the labels a request is counted under (none: not observed). Classifier = Callable[[str, str, Mapping[str, str]], Sequence[str]] +# An origin-form request target: a path and query of RFC 3986 characters, nothing else. +_SAFE_PATH = re.compile(r"/[A-Za-z0-9._~!$&'()*+,;=:@%/?-]*") + _HOP_BY_HOP = { 'connection', 'keep-alive', @@ -157,7 +161,7 @@ def _forward(self, handler: BaseHTTPRequestHandler) -> None: # upstream chosen at construction: a client cannot name another host through the # request line, a "//host" path, or a Host header (which is never forwarded). path = handler.path - if not path.startswith('/') or path.startswith('//') or any(c in path for c in '\r\n'): + if not _SAFE_PATH.fullmatch(path) or path.startswith('//'): handler.send_error(400) return headers = {name: value for name, value in handler.headers.items()}