#!/usr/bin/env python3 """Original offline Azure Blob SDK witness; no Azure account or ERP is used. Install azure-storage-blob==12.31.0 azure-core==1.41.0 requests==2.34.2. Prepared replies exercise the actual SDK pipeline, serializers and decoders. The manifest admission rule is our proposed local rule, not an Azure guarantee. """ from __future__ import annotations import base64 import copy import hashlib import importlib.metadata import json import platform import socket import xml.etree.ElementTree as ET from dataclasses import dataclass from urllib.parse import parse_qs, urlsplit CONNECTIONS = [] REAL_HTTP = [] CHECKS = [] TRACE = [] def deny_connection(*args, **kwargs): CONNECTIONS.append("blocked") raise AssertionError("Connection forbidden in this offline witness") socket.socket.connect = deny_connection socket.socket.connect_ex = deny_connection socket.create_connection = deny_connection import requests from azure.core import MatchConditions from azure.core.credentials import AzureSasCredential from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceModifiedError from azure.core.pipeline.transport import HttpResponse, HttpTransport, RequestsTransport from azure.storage.blob import BlobBlock, BlobClient, BlockState, ContentSettings def deny_http(*args, **kwargs): REAL_HTTP.append("blocked") raise AssertionError("Real HTTP forbidden") requests.sessions.Session.request = deny_http RequestsTransport.send = deny_http def check(label, condition): if not condition: raise AssertionError(label) CHECKS.append(label) def sha(raw): return hashlib.sha256(raw).hexdigest() def encode_id(raw): return base64.b64encode(raw.encode()).decode() @dataclass(frozen=True) class Scope: account_url: str = "https://stockage-fictif.invalid" container: str = "echanges" blob: str = "exports/EXP-208.csv" source_revision: str = "revision-fictive-7" SCOPE = Scope() SOURCE = b"reference;quantite\nSKU-41;2\nSKU-42;1\n" # Opaque IDs deliberately have a lexical order different from source order. RAW_IDS = ["EXP-208:000002", "EXP-208:000001"] PARTS = [SOURCE[:20], SOURCE[20:]] OLD_SAS = "sv=2026-06-06&sr=b&sp=rw&se=2026-10-16T10:00:00Z&sig=fictif-ancien" NEW_SAS = "sv=2026-06-06&sr=b&sp=rw&se=2026-10-16T12:00:00Z&sig=fictif-renouvele" class PreparedResponse(HttpResponse): def __init__(self, request, status, body=b"", headers=None): super().__init__(request, None) self.status_code = status self.headers = requests.structures.CaseInsensitiveDict( {"content-type": "application/xml", "x-ms-request-id": "requete-fictive-208", **(headers or {})}) self.content_type = self.headers["content-type"] self.reason = "Prepared offline reply" self._body = body def body(self): return self._body def text(self, encoding=None): return self._body.decode(encoding or "utf-8") class PreparedTransport(HttpTransport): def __init__(self, case, replies): self.case = case self.replies = list(replies) def open(self): pass def close(self): pass def __enter__(self): return self def __exit__(self, *args): self.close() def send(self, request, **kwargs): if not self.replies: raise AssertionError("Unexpected prepared SDK transport call") method, comp, status, body, headers = self.replies.pop(0) parsed = urlsplit(request.url) query = parse_qs(parsed.query) assert parsed.hostname == "stockage-fictif.invalid" assert request.method == method and query.get("comp", [None])[0] == comp payload = request.body or b"" if isinstance(payload, str): payload = payload.encode() assert isinstance(payload, bytes), "Witness accepts only explicit byte bodies" signature = query.get("sig", [""])[0] marker = {"fictif-ancien": "old", "fictif-renouvele": "renewed"}.get(signature, "unknown") TRACE.append({"case": self.case, "method": request.method, "path": parsed.path, "operation": comp or "properties", "block_id_wire": query.get("blockid", [None])[0], "block_list_type": query.get("blocklisttype", [None])[0], "api_version": request.headers.get("x-ms-version"), "credential_marker": marker, "if_none_match": request.headers.get("If-None-Match"), "body_length": len(payload), "body_sha256": sha(payload), "body_xml": payload.decode() if comp == "blocklist" and method == "PUT" else None, "prepared_status": status}) return PreparedResponse(request, status, body, headers) def client(transport, credential): return BlobClient(account_url=SCOPE.account_url, container_name=SCOPE.container, blob_name=SCOPE.blob, credential=credential, transport=transport, api_version="2026-06-06", retry_total=0, retry_connect=0, retry_read=0, retry_status=0) def reply(method, comp, status, body=b"", headers=None): return method, comp, status, body, headers or {} def error_reply(code): root = ET.Element("Error") ET.SubElement(root, "Code").text = code ET.SubElement(root, "Message").text = "Erreur fournie au SDK dans un laboratoire hors reseau" if code == "AuthenticationFailed": ET.SubElement(root, "AuthenticationErrorDetail").text = "Signature not valid in the supplied fictional time frame" return ET.tostring(root) def block_list_reply(uncommitted, committed=()): root = ET.Element("BlockList") for tag, items in [("CommittedBlocks", committed), ("UncommittedBlocks", uncommitted)]: group = ET.SubElement(root, tag) for identifier, size in items: block = ET.SubElement(group, "Block") ET.SubElement(block, "Name").text = encode_id(identifier) ET.SubElement(block, "Size").text = str(size) return ET.tostring(root) def make_manifest(): return {"scope": SCOPE.__dict__, "source_sha256": sha(SOURCE), "blocks": [{"id": identifier, "size": len(part), "source_sha256": sha(part)} for identifier, part in zip(RAW_IDS, PARTS)]} def proposed_plan(manifest, current_scope, current_source, committed, uncommitted): """Narrow admission for a new blob; neither durable state nor cloud integrity proof.""" if manifest["scope"] != current_scope.__dict__ or manifest["source_sha256"] != sha(current_source): raise ValueError("Source or scope changed") if committed: raise ValueError("Existing committed blob needs separate investigation") actual = {block.id: block.size for block in uncommitted} expected = {block["id"]: block["size"] for block in manifest["blocks"]} if len(actual) != len(uncommitted) or actual != expected: raise ValueError("Incomplete or foreign staged block inventory") # Use own source order. This narrow workflow also sends If-None-Match: *. # The pinned SDK state-selector behavior is examined separately below. return [BlobBlock(block["id"], state=BlockState.LATEST) for block in manifest["blocks"]] def commit(blob_client, plan): return blob_client.commit_block_list(plan, match_condition=MatchConditions.IfMissing, content_settings=ContentSettings(content_type="text/csv")) def run(): for package, version in [("azure-storage-blob", "12.31.0"), ("azure-core", "1.41.0"), ("requests", "2.34.2")]: check(package + " pinned", importlib.metadata.version(package) == version) manifest = make_manifest() manifest_hash = sha(json.dumps(manifest, sort_keys=True).encode()) erp = {"export": "EXP-208", "state": "awaiting_commit", "document_reference": None} erp_initial = copy.deepcopy(erp) # Azure documents alphabetical order for uncommitted IDs. This differs from our source order. inventory = sorted(zip(RAW_IDS, map(len, PARTS))) success_headers = {"etag": '"etag-fictif-208"', "last-modified": "Fri, 16 Oct 2026 10:30:00 GMT"} transport = PreparedTransport("renewed_upload", [ reply("PUT", "block", 201), reply("PUT", "block", 201), reply("PUT", "blocklist", 403, error_reply("AuthenticationFailed"), {"x-ms-error-code": "AuthenticationFailed"}), reply("GET", "blocklist", 200, block_list_reply(inventory)), reply("PUT", "blocklist", 201, headers=success_headers), reply("HEAD", None, 200, headers={**success_headers, "content-length": str(len(SOURCE)), "x-ms-blob-type": "BlockBlob"}), ]) credential = AzureSasCredential(OLD_SAS) blob_client = client(transport, credential) for identifier, part in zip(RAW_IDS, PARTS): blob_client.stage_block(identifier, part, length=len(part)) check("two real SDK stage serializations", len(TRACE) == 2) check("SDK encodes raw IDs exactly once", [x["block_id_wire"] for x in TRACE] == list(map(encode_id, RAW_IDS))) check("staged request bytes match both source slices", [x["body_sha256"] for x in TRACE] == list(map(sha, PARTS))) attempted = [BlobBlock(x, state=BlockState.LATEST) for x in RAW_IDS] try: commit(blob_client, attempted) raise AssertionError("Expected supplied authentication failure") except ClientAuthenticationError as exc: check("SDK surfaces supplied 403", exc.status_code == 403) check("native error code retained", str(exc.error_code) == "StorageErrorCode.AUTHENTICATION_FAILED" or exc.error_code == "AuthenticationFailed") check("expiry detail comes from supplied XML", "fictional time frame" in str(exc)) check("failed commit does not release ERP reference", erp == erp_initial) check("zero implicit retry after refused commit", len(TRACE) == 3) credential.update(NEW_SAS) # Supplied fictional replacement, not creation of a valid SAS. committed, uncommitted = blob_client.get_block_list(block_list_type="all") check("actual SDK decodes no committed blocks", committed == []) check("actual SDK decodes raw IDs from XML", [b.id for b in uncommitted] == [x[0] for x in inventory]) check("enumeration order differs from source order", [b.id for b in uncommitted] != RAW_IDS) check("listing asks for both block lists", TRACE[-1]["block_list_type"] == "all") check("same client uses updated SAS object", TRACE[-1]["credential_marker"] == "renewed") plan = proposed_plan(manifest, SCOPE, SOURCE, committed, uncommitted) check("proposed plan keeps source order", [b.id for b in plan] == RAW_IDS) check("proposed plan deliberately uses Latest", all(b.state == BlockState.LATEST for b in plan)) result = commit(blob_client, plan) commit_request = TRACE[-1] xml = ET.fromstring(commit_request["body_xml"]) check("SDK serializes two Latest elements", [x.tag for x in xml] == ["Latest", "Latest"]) check("wire XML retains manifest order", [x.text for x in xml] == list(map(encode_id, RAW_IDS))) check("create-only condition reaches wire", commit_request["if_none_match"] == "*") check("supplied commit ETag decoded", result["etag"] == '"etag-fictif-208"') properties = blob_client.get_blob_properties() check("supplied properties expose expected byte length", properties.size == len(SOURCE)) check("supplied properties expose same ETag", properties.etag == result["etag"]) check("metadata success is not ERP acceptance", erp == erp_initial) check("credential renewal keeps manifest unchanged", sha(json.dumps(manifest, sort_keys=True).encode()) == manifest_hash) check("main case consumes six replies", not transport.replies and len(TRACE) == 6) for case, status, code, cls in [ ("permission_refusal", 403, "AuthorizationPermissionMismatch", HttpResponseError), ("missing_block_refusal", 400, "InvalidBlockList", HttpResponseError), ("conditional_conflict", 412, "ConditionNotMet", ResourceModifiedError), ]: failing = PreparedTransport(case, [reply("PUT", "blocklist", status, error_reply(code), {"x-ms-error-code": code})]) try: commit(client(failing, AzureSasCredential(NEW_SAS)), plan) raise AssertionError("Expected prepared refusal") except cls as exc: check(case + " status retained", exc.status_code == status) check(case + " code retained", code.lower().replace("_", "") in str(exc.error_code).lower().replace("_", "")) check(case + " one request without retry", len([x for x in TRACE if x["case"] == case]) == 1 and not failing.replies) check(case + " ERP state unchanged", erp == erp_initial) if case == "permission_refusal": check("permission refusal is not relabelled expired SAS", not isinstance(exc, ClientAuthenticationError)) for case, supplied in [("incomplete_inventory", inventory[:1]), ("foreign_inventory", inventory + [("EXP-999:000001", 6)])]: t = PreparedTransport(case, [reply("GET", "blocklist", 200, block_list_reply(supplied))]) old_count = len(TRACE) got_committed, got_uncommitted = client(t, AzureSasCredential(NEW_SAS)).get_block_list("all") try: proposed_plan(manifest, SCOPE, SOURCE, got_committed, got_uncommitted) raise AssertionError("Expected inventory rejection") except ValueError: check(case + " no commit sent", len(TRACE) == old_count + 1) check(case + " retains ERP waiting state", erp == erp_initial) before = len(TRACE) for case, scope, source in [("changed_source", SCOPE, SOURCE.replace(b"SKU-41", b"SKU-99")), ("changed_blob", Scope(blob="exports/AUTRE.csv"), SOURCE)]: try: proposed_plan(manifest, scope, source, committed, uncommitted) raise AssertionError("Expected scope or source rejection") except ValueError: check(case + " blocked before any SDK request", len(TRACE) == before) # Same metadata cannot attest block contents: a byte substitution leaves sizes unchanged. altered = PARTS[0].replace(b"reference", b"REFERENCE") check("same length counterexample", len(altered) == len(PARTS[0]) and sha(altered) != sha(PARTS[0])) check("proposed admission does not inspect native bytes", [b.id for b in proposed_plan(manifest, SCOPE, SOURCE, committed, uncommitted)] == RAW_IDS) selected = PreparedTransport("requested_uncommitted", [reply("PUT", "blocklist", 201, headers=success_headers)]) requested = [BlobBlock(x, state=BlockState.UNCOMMITTED) for x in RAW_IDS] check("requested enum value is capitalized Uncommitted", all(b.state.value == "Uncommitted" for b in requested)) commit(client(selected, AzureSasCredential(NEW_SAS)), requested) selected_xml = ET.fromstring(TRACE[-1]["body_xml"]) check("pinned SDK emits Latest despite requested Uncommitted", [x.tag for x in selected_xml] == ["Latest", "Latest"]) check("selector case still serializes create-only condition", TRACE[-1]["if_none_match"] == "*") check("selector case performs exactly one prepared call", not selected.replies) wrong = PreparedTransport("double_encoded_input", [reply("PUT", "block", 201)]) client(wrong, AzureSasCredential(NEW_SAS)).stage_block(encode_id(RAW_IDS[0]), PARTS[0]) check("preencoding ID causes second SDK encoding", TRACE[-1]["block_id_wire"] == encode_id(encode_id(RAW_IDS[0]))) check("double encoding differs from original wire ID", TRACE[-1]["block_id_wire"] != encode_id(RAW_IDS[0])) check("all prepared requests use pinned API version", all(x["api_version"] == "2026-06-06" for x in TRACE)) check("no real HTTP transport invocation", REAL_HTTP == []) check("no instrumented connection attempt", CONNECTIONS == []) check("all ERP references remain pending", erp == erp_initial) return {"checks_passed": len(CHECKS), "checks": CHECKS, "prepared_transport_invocations": len(TRACE), "trace": TRACE, "real_http_requests": len(REAL_HTTP), "connection_attempts": len(CONNECTIONS), "python": platform.python_version(), "packages": {p: importlib.metadata.version(p) for p in ["azure-storage-blob", "azure-core", "requests"]}, "source_size": len(SOURCE), "source_sha256": sha(SOURCE), "manifest": manifest, "supplied_commit_etag": result["etag"], "erp_state": erp, "scope": "Prepared SDK serialization/decoding and proposed local manifest admission only. No real SAS time/signature/permissions, Azure storage, persistence, concurrency, CSV business acceptance or ERP write certified."} if __name__ == "__main__": print(json.dumps(run(), ensure_ascii=False, indent=2))