"""Observe Paramiko 5.0.0 with prepared SFTP packets, without SSH or a server. The channel below supplies replies, not filesystem persistence. No credential, remote file, host key, directory durability or ERP acceptance is certified. """ from __future__ import annotations import errno import hashlib import http.client import importlib.metadata import io import json import platform import socket import struct import urllib.request from dataclasses import dataclass from unittest.mock import patch import paramiko from paramiko import Message, SFTPAttributes, SFTPClient from paramiko.sftp import ( CMD_ATTRS, CMD_CLOSE, CMD_EXTENDED, CMD_HANDLE, CMD_INIT, CMD_OPEN, CMD_REMOVE, CMD_RENAME, CMD_STAT, CMD_STATUS, CMD_VERSION, CMD_WRITE, SFTP_FAILURE, SFTP_NO_SUCH_FILE, SFTP_OK, SFTP_OP_UNSUPPORTED, SFTP_PERMISSION_DENIED, ) CHECKS: list[str] = [] PACKETS: list[dict] = [] ERRORS: list[dict] = [] DENIED = {"tcp_connections": 0, "ssh_handshakes": 0, "http_calls": 0} SOURCE = b"export;revision;quantity\nEXP-517;r2;12\n" TEMP = "/outbound/EXP-517-r2-A7.csv.part" FINAL = "/outbound/inventory.csv" ERP_STATE = {"last_export_ref": "EXP-516", "new_export_ref": None, "state": "awaiting_document"} def check(condition: bool, name: str) -> None: if not condition: raise AssertionError(name) CHECKS.append(name) def frame(kind: int, payload: Message) -> bytes: raw = payload.asbytes() return struct.pack(">I", len(raw) + 1) + bytes([kind]) + raw @dataclass(frozen=True) class Reply: operation: int status: int = SFTP_OK message: str = "Prepared reply" size: int | None = None class PreparedChannel: """Only an in-memory byte stream consumed by the unmodified native client.""" def __init__(self, name: str, replies: list[Reply], advertised: bool): self.name = name self.replies = list(replies) self.advertised = advertised self.pending = bytearray() self.incoming = bytearray() self.used = 0 self.closed = False def send(self, data: bytes) -> int: self.pending.extend(data) while len(self.pending) >= 4: size = struct.unpack(">I", self.pending[:4])[0] if len(self.pending) < size + 4: break raw = bytes(self.pending[:size + 4]) del self.pending[:size + 4] self._answer(raw) return len(data) def _answer(self, raw: bytes) -> None: kind = raw[4] message = Message(raw[5:]) if kind == CMD_INIT: version = message.get_int() check(version == 3, self.name + ": native INIT version3") reply = Message() reply.add_int(3) if self.advertised: reply.add_string("posix-rename@openssh.com") reply.add_string("1") self.incoming.extend(frame(CMD_VERSION, reply)) PACKETS.append({"case": self.name, "kind": "INIT", "client_version": version, "prepared_extensions": ( {"posix-rename@openssh.com": "1"} if self.advertised else {}), "wire_hex": raw.hex()}) return check(self.used < len(self.replies), self.name + ": request has a prepared response") reply = self.replies[self.used] self.used += 1 check(kind == reply.operation, self.name + ": operation" + str(self.used)) request_id = message.get_int() record = {"case": self.name, "opcode": kind, "request_id": request_id, "wire_hex": raw.hex()} if kind == CMD_OPEN: record.update(path=message.get_text(), flags=message.get_int(), attributes_flags=message.get_int()) elif kind == CMD_WRITE: handle = message.get_binary() offset = message.get_int64() data = message.get_binary() record.update(handle=handle.decode(), offset=offset, size=len(data), sha256=hashlib.sha256(data).hexdigest()) elif kind == CMD_CLOSE: record["handle"] = message.get_text() elif kind == CMD_STAT: record["path"] = message.get_text() elif kind == CMD_RENAME: record.update(oldpath=message.get_text(), newpath=message.get_text()) elif kind == CMD_EXTENDED: record.update(extension=message.get_text(), oldpath=message.get_text(), newpath=message.get_text()) else: raise AssertionError("Unprepared operation, including REMOVE") check(not message.get_remainder(), self.name + ": complete request decoding" + str(self.used)) output = Message() output.add_int(request_id) if kind == CMD_OPEN: output.add_string("handle-" + self.name) response_kind = CMD_HANDLE elif kind == CMD_STAT: attrs = SFTPAttributes() attrs.st_size = reply.size attrs._pack(output) response_kind = CMD_ATTRS record["prepared_stat_size"] = reply.size else: output.add_int(reply.status) output.add_string(reply.message) output.add_string("") response_kind = CMD_STATUS record["prepared_status"] = reply.status PACKETS.append(record) self.incoming.extend(frame(response_kind, output)) def recv(self, count: int) -> bytes: # Deliberately fragmented replies exercise the native packet reader. if not self.incoming: raise EOFError("Prepared channel exhausted; no network fallback") take = min(count, 7, len(self.incoming)) data = bytes(self.incoming[:take]) del self.incoming[:take] return data def recv_ready(self) -> bool: return bool(self.incoming) def get_name(self) -> str: return "prepared-" + self.name def close(self) -> None: self.closed = True def finish(self) -> None: check(self.used == len(self.replies), self.name + ": all responses used") check(not self.pending and not self.incoming, self.name + ": no leftover packet bytes") @dataclass(frozen=True) class Receipt: revision: str temporary_path: str final_path: str source_sha256: str source_size: int def admit(receipt: Receipt, content: bytes, revision: str, temporary_path: str, final_path: str) -> None: """A proposed local guard; it does not inspect any remote filesystem.""" if (receipt.revision != revision or receipt.temporary_path != temporary_path or receipt.final_path != final_path or receipt.source_size != len(content) or receipt.source_sha256 != hashlib.sha256(content).hexdigest()): raise ValueError("Source, revision or publication paths changed") def transfer_replies(size: int) -> list[Reply]: return [Reply(CMD_OPEN), Reply(CMD_WRITE), Reply(CMD_CLOSE), Reply(CMD_STAT, size=size)] def native_error(action, expected_errno: int | None, label: str, message_fragment: str | None = None) -> None: try: action() except OSError as error: check(error.errno == expected_errno, label + ": native errno") ERRORS.append({"case": label, "type": type(error).__name__, "errno": error.errno, "message": str(error)}) if message_fragment is not None: check(message_fragment in str(error), label + ": native message") else: raise AssertionError(label + ": expected native error") def deny(kind: str): def blocked(*args, **kwargs): DENIED[kind] += 1 raise AssertionError("No real connection, HTTP or SSH is permitted") return blocked def run() -> dict: check(paramiko.__version__ == "5.0.0", "pinned Paramiko version") original_state = dict(ERP_STATE) receipt = Receipt("r2", TEMP, FINAL, hashlib.sha256(SOURCE).hexdigest(), len(SOURCE)) admit(receipt, SOURCE, "r2", TEMP, FINAL) normal = PreparedChannel("transfer_then_standard_rename", transfer_replies(len(SOURCE)) + [ Reply(CMD_RENAME, SFTP_FAILURE, "Failure")], advertised=True) client = SFTPClient(normal) attrs = client.putfo(io.BytesIO(SOURCE), TEMP, len(SOURCE), confirm=True) check(attrs.st_size == len(SOURCE), "putfo confirms supplied size") native_error(lambda: client.rename(TEMP, FINAL), None, "generic standard rename failure") client.close() normal.finish() cases = [ ("extension_not_advertised", False, CMD_EXTENDED, SFTP_OP_UNSUPPORTED, None), ("extension_permission_denied", True, CMD_EXTENDED, SFTP_PERMISSION_DENIED, errno.EACCES), ("standard_no_such_file", True, CMD_RENAME, SFTP_NO_SUCH_FILE, errno.ENOENT), ("extension_prepared_success", True, CMD_EXTENDED, SFTP_OK, None), ] for name, advertised, opcode, status, expected_errno in cases: channel = PreparedChannel(name, [Reply(opcode, status)], advertised) client = SFTPClient(channel) action = (lambda: client.rename(TEMP, FINAL)) if opcode == CMD_RENAME \ else (lambda: client.posix_rename(TEMP, FINAL)) if status == SFTP_OK: check(action() is None, name + ": native return") else: native_error(action, expected_errno, name) client.close() channel.finish() mismatch = PreparedChannel("size_mismatch", transfer_replies(len(SOURCE) - 1), True) client = SFTPClient(mismatch) native_error( lambda: client.putfo(io.BytesIO(SOURCE), TEMP, len(SOURCE), confirm=True), None, "confirmation size mismatch", "size mismatch in put!") client.close() mismatch.finish() before = len(PACKETS) for name, content, revision, temporary_path, final_path in [ ("changed same-size source", SOURCE.replace(b"12", b"13"), "r2", TEMP, FINAL), ("changed revision", SOURCE, "r3", TEMP, FINAL), ("changed temporary path", SOURCE, "r2", TEMP + ".other", FINAL), ("changed final path", SOURCE, "r2", TEMP, FINAL + ".other"), ]: try: admit(receipt, content, revision, temporary_path, final_path) except ValueError: check(len(PACKETS) == before, name + ": no further native packet") else: raise AssertionError("Local admission unexpectedly accepted " + name) requests = [p for p in PACKETS if p.get("kind") != "INIT"] writes = [p for p in requests if p["opcode"] == CMD_WRITE] check(len(writes) == 2 and all(p["sha256"] == receipt.source_sha256 and p["offset"] == 0 for p in writes), "actual native write payloads match source") check(all(p["flags"] == 26 for p in requests if p["opcode"] == CMD_OPEN), "putfo wb native flags include create/write/truncate") check(all(p["opcode"] != CMD_REMOVE for p in requests), "no delete-before-rename fallback") unadvertised = next(p for p in requests if p["case"] == "extension_not_advertised") check(unadvertised["opcode"] == CMD_EXTENDED and unadvertised["extension"] == "posix-rename@openssh.com", "native client sends extension despite absent prepared advertisement") check(ERP_STATE == original_state, "prior ERP reference remains unchanged") check(all(value == 0 for value in DENIED.values()), "no real transport used") return { "python": platform.python_version(), "packages": {name: importlib.metadata.version(name) for name in ["paramiko", "cryptography", "bcrypt", "pynacl"]}, "checks_passed": len(CHECKS), "checks": CHECKS, "prepared_sftp_packets": len(PACKETS), "prepared_application_requests": len(requests), "prepared_init_version_exchanges": len(PACKETS) - len(requests), "real_tcp_connections": DENIED["tcp_connections"], "real_ssh_handshakes": DENIED["ssh_handshakes"], "real_http_calls": DENIED["http_calls"], "source_bytes": len(SOURCE), "source_sha256": receipt.source_sha256, "packets": PACKETS, "observed_errors": ERRORS, "erp_state": ERP_STATE, "limits": [ "Prepared SFTP packets do not implement a remote storage service.", "No SSH authentication, host-key checking or filesystem guarantee.", "No real destination existence, overwrite or atomicity is certified.", "No ERP import, durable receipt or connected crash recovery occurs.", ], } if __name__ == "__main__": with patch.object(socket.socket, "connect", deny("tcp_connections")), \ patch("socket.create_connection", deny("tcp_connections")), \ patch.object(paramiko.Transport, "start_client", deny("ssh_handshakes")), \ patch.object(paramiko.SSHClient, "connect", deny("ssh_handshakes")), \ patch("urllib.request.urlopen", deny("http_calls")), \ patch.object(http.client.HTTPConnection, "connect", deny("http_calls")): print(json.dumps(run(), ensure_ascii=False, indent=2))