Interruption and verified resume

#!/usr/bin/env python3
"""Original interrupted SFTP upload and transactional local-consumer fixture. Loopback only, disposable keys."""
import argparse,contextlib,hashlib,json,os,pwd,shlex,signal,socket,sqlite3,subprocess,sys,tempfile,time
from pathlib import Path

def main:
 p=argparse.ArgumentParser(description=__doc__)
 for name in ['ssh','sshd','sshd-session','sshd-auth','keygen','sftp','sftp-server']:p.add_argument('--'+name,required=True)
 p.add_argument('--output',default='sftp-recovery-evidence.json');a=p.parse_args;user=pwd.getpwuid(os.getuid).pw_name
 checks=[];observations={};configurations=[];sessions=[]
 def check(label,ok):assert ok,label;checks.append(label)
 def call(argv):
 x=subprocess.run(list(map(str,argv)),text=True,capture_output=True,timeout=15);assert x.returncode==0,(argv,x.stderr);return x.stdout+x.stderr
 version=call([a.ssh,'-V']).strip;assert 'OpenSSH_10.5p1' in version
 def sha(path):return hashlib.sha256(Path(path).read_bytes).hexdigest
 def quote(path):return '"'+str(path).replace('\\','\\\\').replace('"','\\"')+'"'
 with tempfile.TemporaryDirectory(prefix='dr-sftp-recovery-')as temp:
 root=Path(temp);root.chmod(0o700)
 def key(name):
 f=root/name;call([a.keygen,'-q','-t','ed25519','-N','','-C','dr-disposable-'+name,'-f',f]);return f
 host=key('host');badhost=key('badhost');identity=key('identity');wrong=key('wrong')
 local=root/'local'local.mkdir;source=local/'source.bin'data=(b'DR-original-SFTP-delivery;version=2026-10\n'*4000)+bytes(range(256));source.write_bytes(data)
 @contextlib.contextmanager
 def daemon(label,server_options='',subsystem=True):
 folder=root/label;folder.mkdir;remote=folder/'remote'remote.mkdir;existing=remote/'existing.bin'existing.write_bytes(data)
 auth=folder/'authorized_keys'auth.write_text(Path(str(identity)+'.pub').read_text);auth.chmod(0o600)
 with socket.socketas sock:sock.bind(('127.0.0.1',0));port=sock.getsockname[1]
 command=shlex.quote(a.sftp_server)+' -e -l VERBOSE -d '+shlex.quote(str(remote))+(' '+server_options if server_options else'')
 conf=folder/'sshd.conf'conf.write_text(f'''ListenAddress 127.0.0.1
Port {port}
HostKey {host}
PidFile {folder}/pid
SshdSessionPath {a.sshd_session}
SshdAuthPath {a.sshd_auth}
AuthorizedKeysFile {auth}
AllowUsers {user}
AuthenticationMethods publickey
PubkeyAuthentication yes
PasswordAuthentication no
KbdInteractiveAuthentication no
PermitRootLogin no
PermitUserEnvironment no
PermitUserRC no
PermitTTY no
X11Forwarding no
AllowAgentForwarding no
DisableForwarding yes
StrictModes yes
UseDNS no
LogLevel VERBOSE
ForceCommand {command}
SetEnv ZDOTDIR={folder}
'''+('Subsystem sftp '+command+'\n'if subsystem else''))
 call([a.sshd,'-t','-f',conf]);effective=call([a.sshd,'-T','-f',conf]);configurations.append(dict(label=label,text=conf.read_text,effective=effective))
 log=folder/'server.log'
 with log.open('w')as lf:
 proc=subprocess.Popen([a.sshd,'-D','-e','-f',str(conf)],stdout=lf,stderr=lf,start_new_session=True)
 try:
 for _ in range(200):
 assert proc.pollis None,log.read_text
 if 'Server listening on 'in log.read_text:break
 time.sleep(.02)
 else:raise AssertionError('Server did not listen')
 def transfer(name,lines,offered=identity,bad_host=False,cwd=local,interrupt=False):
 known=folder/(name+'.known_hosts');pub=Path(str(badhost if bad_host else host)+'.pub').read_text.split;known.write_text('[127.0.0.1]:'+str(port)+' '+pub[0]+' '+pub[1]+'\n')
 args=[a.sftp,'-S',a.ssh,'-F','/dev/null','-vv','-b','-','-P',str(port),'-i',str(offered),'-o','BatchMode=yes','-o','IdentitiesOnly=yes','-o','IdentityAgent=none','-o','CertificateFile=none','-o','UserKnownHostsFile='+str(known),'-o','GlobalKnownHostsFile=/dev/null','-o','StrictHostKeyChecking=yes','-o','UpdateHostKeys=no','-o','PreferredAuthentications=publickey','-o','PasswordAuthentication=no','-o','KbdInteractiveAuthentication=no','-o','ControlMaster=no','-o','ControlPath=none','-o','ProxyCommand=none','-o','ProxyJump=none','-o','ConnectTimeout=5',user+'@127.0.0.1']
 if interrupt:
 args=args[:-1]+['-l','128','-B','4096','-R','1',args[-1]]
 out=folder/(name+'.out');err=folder/(name+'.err');batch=folder/(name+'.batch');batch.write_text('\n'.join(lines)+'\n')
 with batch.openas inp,out.open('w')as outf,err.open('w')as errf:
 proc=subprocess.Popen(args,stdin=inp,stdout=outf,stderr=errf,cwd=cwd,start_new_session=True)
 try:
 for _ in range(400):
 assert proc.pollis None,err.read_text
 part=remote/'interrupted.part'
 if part.existsand 4096<=part.stat.st_size<len(data):break
 time.sleep(.025)
 else:raise AssertionError('No partial upload observed')
 finally:
 if proc.pollis None:os.killpg(proc.pid,signal.SIGTERM)
 try:proc.wait(timeout=5)
 except subprocess.TimeoutExpired:os.killpg(proc.pid,signal.SIGKILL);proc.wait(timeout=5)
 time.sleep(.1)
 x=subprocess.CompletedProcess(args,proc.returncode,out.read_text,err.read_text)
 else:x=subprocess.run(args,input='\n'.join(lines)+'\n',text=True,capture_output=True,timeout=15,cwd=cwd)
 result=dict(label=name,exit=x.returncode,stdout=x.stdout,stderr=x.stderr,batch=lines,args=args);sessions.append(result);return result
 yield remote,transfer,log
 finally:
 if proc.pollis None:
 os.killpg(proc.pid,signal.SIGTERM)
 try:proc.wait(timeout=5)
 except subprocess.TimeoutExpired:os.killpg(proc.pid,signal.SIGKILL);proc.wait(timeout=5)
 check(label+': daemon stopped',proc.pollis not None)
 with daemon('recovery')as(remote,run,log):
 interrupted=run('interrupted-upload',[f'put {quote(source)} interrupted.part'],interrupt=True)
 part=remote/'interrupted.part'prefix=part.read_bytes
 check('interruption: client did not complete successfully',interrupted['exit']!=0)
 check('interruption: proper partial file retained',0<len(prefix)<len(data))
 check('interruption: retained bytes match source prefix',prefix==data[:len(prefix)])
 check('interruption: SSH authentication occurred','Authenticated to 'in interrupted['stderr'])
 observations['interruption']=dict(exit=interrupted['exit'],partialBytes=len(prefix),sourceBytes=len(data),prefixMatches=True,mechanism='SIGTERM to local SFTP client process group after observing partial upload; not a network outage or power failure.')
 corrupt=remote/'wrong-prefix.part'corrupt.write_bytes(bytes([prefix[0]^1])+prefix[1:])
 resumed=run('resume-valid',[f'put -a {quote(source)} interrupted.part'])
 check('valid resume: zero exit',resumed['exit']==0);check('valid resume: full content matches',sha(part)==sha(source))
 observations['validResume']=dict(exit=resumed['exit'],bytes=part.stat.st_size,resultHash=sha(part),sourceHash=sha(source))
 wrong=run('resume-wrong-prefix',[f'put -a {quote(source)} wrong-prefix.part'])
 check('wrong prefix: zero exit',wrong['exit']==0);check('wrong prefix: full size',corrupt.stat.st_size==len(data));check('wrong prefix: content mismatch',sha(corrupt)!=sha(source));check('wrong prefix: suffix copied',corrupt.read_bytes[len(prefix):]==data[len(prefix):])
 observations['wrongPrefixResume']=dict(exit=wrong['exit'],bytes=corrupt.stat.st_size,resultHash=sha(corrupt),sourceHash=sha(source),prefixDeliberatelyAltered=True)
 rejected_copy=local/'rejected.bin'rejected_copy.write_bytes(corrupt.read_bytes)
 overwrite=run('overwrite-repair',[f'put {quote(source)} wrong-prefix.part']);check('overwrite: content restored',overwrite['exit']==0 and sha(corrupt)==sha(source))
 published=run('publish-and-download',['rename interrupted.part ready.bin',f'get ready.bin {quote(local/"received.bin")}'])
 check('publication: succeeds',published['exit']==0);check('publication: expected bytes downloaded',sha(local/'received.bin')==sha(source));check('publication: old staging absent',not part.exists)
 observations['publication']=dict(exit=published['exit'],readyHash=sha(remote/'ready.bin'),receivedHash=sha(local/'received.bin'),serverLog=log.read_text)
 dbfile=root/'consumer.sqlite'
 db=sqlite3.connect(dbfile);db.executescript('CREATE TABLE receipts(delivery_id TEXT PRIMARY KEY, digest TEXT NOT NULL, receipt TEXT NOT NULL); CREATE TABLE effects(delivery_id TEXT PRIMARY KEY REFERENCES receipts(delivery_id), byte_count INTEGER NOT NULL);');db.close
 worker_reports=[]
 def consume(label,delivery,filename,expected,mode='normal',wanted=0):
 argv=[sys.executable,str(Path(__file__).resolve),'--consume',str(dbfile),delivery,str(filename),expected,mode]
 x=subprocess.run(argv,capture_output=True,text=True,timeout=10);check(label+': expected worker exit',x.returncode==wanted)
 db=sqlite3.connect(dbfile);receipts=db.execute('SELECT delivery_id,digest,receipt FROM receipts ORDER BY delivery_id').fetchall;effects=db.execute('SELECT delivery_id,byte_count FROM effects ORDER BY delivery_id').fetchall;db.close
 item=dict(label=label,exit=x.returncode,stdout=x.stdout,receipts=receipts,effects=effects);worker_reports.append(item);return item
 rejected=consume('integrity-rejection','delivery-A',rejected_copy,sha(source),wanted=12);check('integrity: no receipt or effect',not rejected['receipts']and not rejected['effects'])
 before=consume('failure-before-commit','delivery-A',local/'received.bin',sha(source),'before-commit',23);check('before commit: both inserts rolled back',not before['receipts']and not before['effects'])
 lost=consume('failure-after-commit','delivery-A',local/'received.bin',sha(source),'after-commit',24);check('after commit: receipt and effect retained',len(lost['receipts'])==len(lost['effects'])==1);check('after commit: no success reply',lost['stdout']=='')
 duplicate=consume('retry-same-identity','delivery-A',local/'received.bin',sha(source));check('same identity: duplicate reply',json.loads(duplicate['stdout'])['status']=='duplicate');check('same identity: no extra effect',duplicate['receipts']==lost['receipts']and duplicate['effects']==lost['effects'])
 changed=local/'changed.bin'changed.write_bytes(b'changed-business-content')
 conflict=consume('identity-content-conflict','delivery-A',changed,sha(changed),wanted=13);check('conflict: original state retained',conflict['receipts']==lost['receipts']and conflict['effects']==lost['effects'])
 new_id=consume('new-identity-same-bytes','delivery-B',local/'received.bin',sha(source));check('new identity: processed again',json.loads(new_id['stdout'])['status']=='accepted'and len(new_id['effects'])==2)
 observations['consumer']=dict(workers=worker_reports,sqliteVersion=sqlite3.sqlite_version,scope='Original local SQLite fixture; receipt and synthetic effect are in one transaction. No financial posting, external effect, concurrent writers, distributed transaction, retention expiry or power-loss test.')
 check('temporary files and keys removed',not root.exists)
 report=dict(sshVersion=version,checks=checks,observations=observations,sessions=sessions,configurations=configurations,runnerSha256=sha(__file__),binarySha256={n:sha(getattr(a,n.replace('-','_')))for n in ['ssh','sshd','sshd-session','sshd-auth','keygen','sftp','sftp-server']},scope='One loopback sshd, five SFTP client sessions including a deliberately interrupted upload, and six original SQLite consumer subprocesses. Current local account, disposable keys, no PAM/chroot, external partner, network fault, financial system, concurrent readers or power-loss durability. Starting directory is not isolation.')
 Path(a.output).write_text(json.dumps(report,indent=2).replace(user,'lab-account')+'\n');print(json.dumps(dict(sessions=len(sessions),checks=len(checks),consumerWorkers=6,output=a.output)))

def consumer:
 dbfile,delivery,filename,expected,mode=sys.argv[2:]
 content=Path(filename).read_bytes;digest=hashlib.sha256(content).hexdigest
 if digest!=expected:print(json.dumps({'status':'integrity-rejected'}));return 12
 db=sqlite3.connect(dbfile,isolation_level=None);db.execute('PRAGMA foreign_keys=ON')
 try:
 db.execute('BEGIN IMMEDIATE');old=db.execute('SELECT digest,receipt FROM receipts WHERE delivery_id=?',(delivery,)).fetchone
 if old:
 db.execute('ROLLBACK')
 if old[0]!=digest:print(json.dumps({'status':'identity-conflict'}));return 13
 print(json.dumps({'status':'duplicate','receipt':old[1]}));return 0
 receipt='receipt-'+delivery
 db.execute('INSERT INTO receipts VALUES(?,?,?)',(delivery,digest,receipt));db.execute('INSERT INTO effects VALUES(?,?)',(delivery,len(content)))
 if mode=='before-commit':os._exit(23)
 db.execute('COMMIT')
 if mode=='after-commit':os._exit(24)
 print(json.dumps({'status':'accepted','receipt':receipt}));return 0
 finally:db.close
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1]=='--consume':sys.exit(consumer)
 main

Related topics: Staging and publication · Identity and reconciliation

Reference: OpenSSH SFTP resume and client options · BigSavant SFTP 2026-09; selected OpenSSH client, server and extension behavior