Interrupção e retoma verificada

#!/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.socket()as 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.poll()is 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.open()as 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.poll()is None,err.read_text()
                                        part=remote/'interrupted.part'
                                        if part.exists()and 4096<=part.stat().st_size<len(data):break
                                        time.sleep(.025)
                                    else:raise AssertionError('No partial upload observed')
                                finally:
                                    if proc.poll()is 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.poll()is 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.poll()is 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()

Tópicos relacionados: Staging e publicação · Identidade e reconciliação

Referência: OpenSSH SFTP resume and client options · DR SFTP 2026-09; selected OpenSSH client, server and extension behavior