SFTP sobre SSH e fases de falha

#!/usr/bin/env python3
"""Original SFTP over loopback SSH lab. Disposable keys, synthetic files, current local account, no PAM or chroot."""
import argparse,contextlib,hashlib,json,os,pwd,shlex,signal,socket,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-transport-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-transport-')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):
                        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']
                        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('roundtrip')as(remote,run,log):
            final=remote/'final.bin';final.write_bytes(b'old-version');old=sha(final)
            staged=run('stage',[f'put {quote(source)} stage.part']);check('upload succeeds over SSH',staged['exit']==0);check('staged bytes match source',sha(remote/'stage.part')==sha(source));check('old final unchanged before publication',sha(final)==old)
            published=run('publish',['rename stage.part final.bin',f'get final.bin {quote(local/"roundtrip.bin")}']);check('publication and download succeed',published['exit']==0);check('roundtrip bytes match source',sha(local/'roundtrip.bin')==sha(source)==sha(final));check('staging name removed',not(remote/'stage.part').exists());check('actual SSH authentication observed','Authenticated to 'in staged['stderr']);check('SFTP version negotiation observed','Remote version: 3'in staged['stderr'])
            observations['roundtrip']=dict(stageExit=staged['exit'],publishExit=published['exit'],sourceHash=sha(source),finalHash=sha(final),roundtripHash=sha(local/'roundtrip.bin'),oldFinalHash=old,bytes=len(data),serverLog=log.read_text())
        for label,bad_host,offered in [('wrongHost',True,identity),('wrongKey',False,wrong)]:
            with daemon(label)as(remote,run,log):
                x=run(label,[f'put {quote(source)} rejected.bin'],offered,bad_host);check(label+': client fails',x['exit']!=0);check(label+': no upload',not(remote/'rejected.bin').exists());check(label+': no accepted authentication','Authenticated to 'not in x['stderr']);term='REMOTE HOST IDENTIFICATION HAS CHANGED'if bad_host else'Permission denied';check(label+': reason recorded',term in x['stderr']);observations[label]=dict(exit=x['exit'],fileCreated=False,serverLog=log.read_text())
        with daemon('missingSubsystem',subsystem=False)as(remote,run,log):
            x=run('missingSubsystem',['pwd']);check('missing subsystem: authentication succeeds','Authenticated to 'in x['stderr']);check('missing subsystem: request fails',x['exit']!=0 and 'subsystem request failed'in x['stderr']);check('missing subsystem: no protocol version','Remote version: 3'not in x['stderr']);observations['missingSubsystem']=dict(exit=x['exit'],authenticated=True,protocolStarted=False,serverLog=log.read_text())
        with daemon('readOnly','-R')as(remote,run,log):
            read=run('readOnlyGet',[f'get existing.bin {quote(local/"readonly.bin")}']);write=run('readOnlyPut',[f'put {quote(source)} denied.bin']);check('read-only: read succeeds',read['exit']==0 and sha(local/'readonly.bin')==sha(source));check('read-only: write rejected',write['exit']!=0 and not(remote/'denied.bin').exists());check('read-only: both authenticate',all('Authenticated to 'in x['stderr']for x in [read,write]));check('read-only: filesystem writable',os.access(remote,os.W_OK));observations['readOnly']=dict(readExit=read['exit'],writeExit=write['exit'],readHash=sha(local/'readonly.bin'),directoryWritable=True,serverLog=log.read_text())
        with daemon('renameDenied','-P rename,posix-rename')as(remote,run,log):
            x=run('renameDenied',[f'put {quote(source)} complete.part','rename complete.part ready.bin']);check('rename denied: batch fails',x['exit']!=0);check('rename denied: staging retained',sha(remote/'complete.part')==sha(source));check('rename denied: final absent',not(remote/'ready.bin').exists());observations['renameDenied']=dict(exit=x['exit'],stagingHash=sha(remote/'complete.part'),finalExists=False,serverLog=log.read_text())
        with daemon('relativeContext')as(remote,run,log):
            expected=local/'expected';stale=local/'stale';expected.mkdir();stale.mkdir();(expected/'delivery.csv').write_bytes(b'version,current\n');(stale/'delivery.csv').write_bytes(b'version,previous\n')
            x=run('relativeWrongSource',['put delivery.csv uploaded.csv'],cwd=stale);y=run('explicitSource',[f'put {quote(expected/"delivery.csv")} correct.csv'],cwd=stale)
            check('relative paths: both transfers succeed',x['exit']==y['exit']==0);check('relative paths: wrong local version uploaded',sha(remote/'uploaded.csv')==sha(stale/'delivery.csv')!=sha(expected/'delivery.csv'));check('relative paths: explicit path selects expected version',sha(remote/'correct.csv')==sha(expected/'delivery.csv'));observations['relativeContext']=dict(relativeExit=x['exit'],explicitExit=y['exit'],expectedHash=sha(expected/'delivery.csv'),uploadedHash=sha(remote/'uploaded.csv'),explicitHash=sha(remote/'correct.csv'),serverLog=log.read_text())
        with daemon('flushRequest')as(remote,run,log):
            x=run('flushRequest',[f'put -f {quote(source)} flushed.bin']);check('flush: transfer succeeds',x['exit']==0);check('flush: content matches',sha(remote/'flushed.bin')==sha(source));check('flush: fsync extension advertised','fsync@openssh.com'in x['stderr']);check('flush: request observed','Sending SSH2_FXP_EXTENDED(fsync@openssh.com)'in x['stderr']);observations['flushRequest']=dict(exit=x['exit'],contentHash=sha(remote/'flushed.bin'),serverLog=log.read_text(),powerLossTested=False)
    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='Eight loopback sshd processes and eleven SFTP client sessions with synthetic files and disposable keys. Current local account, no PAM, chroot, account isolation, external partner, power-loss durability, network interruption or business consumer. Starting directory is not a security boundary. Existing filesystem was only accessed inside the temporary experiment tree.')
    Path(a.output).write_text(json.dumps(report,indent=2).replace(user,'lab-account')+'\n');print(json.dumps(dict(experiments=len(observations),sessions=len(sessions),checks=len(checks),output=a.output)))
if __name__=='__main__':main()

Tópicos relacionados: SSH e identidade · Permissões e políticas SFTP

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