APRENDER. PRATICAR. EVOLUIR.Desenha os extremos antes de abrir o túnel
Um encaminhamento SSH tem pelo menos três elementos: o listener consumido pela aplicação, a ligação SSH e o serviço final. Em -L, o cliente cria o listener e o servidor SSH estabelece a ligação ao destino pedido. Em -R, o listener está do lado do servidor e o cliente SSH liga ao backend. O mesmo texto 127.0.0.1 pode assim referir máquinas diferentes num desenho real. No laboratório, tudo corre em loopback na mesma máquina; a separação é feita por processos e portas. Desenha a direção dos pedidos e identifica quem resolve o nome final antes de investigar DNS, firewall ou disponibilidade. Uma sessão autenticada não demonstra que o consumidor obtém resposta.
Compara criação, política e disponibilidade
O runner cria uma fixture TCP original que devolve DR-FORWARDING-OK. localAllowed e remoteAllowed recebem esse marcador pelo caminho esperado. bindCollision ocupa deliberadamente a porta de entrada antes de iniciar SSH; ExitOnForwardFailure=yes termina o cliente quando não consegue criar o listener. backendDown é diferente: o listener existe, mas o serviço final não escuta e o canal regista Connection refused. O cliente continua ativo. Esta opção não transforma um túnel num monitor de disponibilidade aplicacional. localDenied também mantém o listener, mas regista administratively prohibited porque a política recusa o destino. Compara fase, mensagem, configuração efetiva e resultado antes de escolher uma correção. O mesmo sintoma de aplicação sem resposta pode ter causas e responsáveis distintos.
Mantém a autorização e a exposição explícitas
PermitOpen limita destinos pedidos ao servidor por canais de encaminhamento. Na experiência literalMismatch, a política permite o endereço 127.0.0.1, mas o pedido usa localhost. A rejeição não prova falha de resolução: a regra não resolve os nomes para estabelecer equivalência. PermitListen trata os endereços e portas de escuta pedidos por encaminhamento remoto. remoteDenied demonstra a recusa da criação quando essa lista é none. Os listeners deste laboratório usam explicitamente loopback e GatewayPorts no. Não foi ensaiada exposição em interfaces externas. Num projeto, abrir 0.0.0.0 altera quem pode alcançar a entrada; precisa de análise do consumidor, controlos de acesso e âmbito autorizado. Evita substituir uma permissão exata por any para ocultar um desajuste de configuração.
Executa, interpreta e entrega evidência útil
Copia o código completo para run.py num diretório de trabalho descartável. Fornece os cinco executáveis correspondentes de OpenSSH 10.5p1, conforme o guião da aula seguinte. O build utilizado foi compilado sem PAM e a conta local existente precisa de poder autenticar; não se cria qualquer conta. O programa inicia onze experiências, quinze processos sshd temporários e uma fixture TCP, terminando-os no fim. Usa chaves descartáveis, ficheiros de confiança próprios e um comando fixo no destino. Passaram 58 verificações. Repete com novas portas e chaves, comparando resultados sem exigir logs literalmente iguais. Para RUN, acrescenta uma operação sintética autorizada e distingue saúde do processo, sucesso do canal e resultado funcional. O marcador não valida TLS nem autenticação de uma aplicação real. Guarda também a relação entre a porta de entrada e a configuração do consumidor, para que a recuperação não deixe o job a usar o destino anterior.
#!/usr/bin/env python3
"""Original SSH forwarding and jump-host lab: disposable keys, loopback only, fixed target command."""
import argparse,contextlib,hashlib,json,os,pwd,re,shlex,signal,socket,socketserver,subprocess,sys,tempfile,threading,time
from pathlib import Path
def main():
ap=argparse.ArgumentParser(description=__doc__)
for x in ['ssh','sshd','sshd-session','sshd-auth','keygen']:ap.add_argument('--'+x,required=True)
ap.add_argument('--output',default='ssh-forwarding-evidence.json');a=ap.parse_args()
user=pwd.getpwuid(os.getuid()).pw_name;checks=[];observations={};configs=[];count=0
def call(argv):
x=subprocess.run(list(map(str,argv)),capture_output=True,text=True,timeout=15)
assert x.returncode==0,(argv,x.returncode,x.stderr);return x.stdout+x.stderr
version=call([a.ssh,'-V']).strip();assert 'OpenSSH_10.5p1' in version
def check(name,ok):assert ok,name;checks.append(name)
def port():
with socket.socket()as s:s.bind(('127.0.0.1',0));return s.getsockname()[1]
def await_log(proc,log,term):
for _ in range(200):
if term in log.read_text():return
assert proc.poll()is None,(term,log.read_text())
time.sleep(.02)
raise AssertionError((term,log.read_text()))
@contextlib.contextmanager
def running(argv,log,label):
with log.open('w')as out:
proc=subprocess.Popen(list(map(str,argv)),stdout=out,stderr=out,start_new_session=True)
try:yield proc
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+': process stopped',proc.poll()is not None)
with tempfile.TemporaryDirectory(prefix='dr-ssh-forwarding-')as tmp:
root=Path(tmp);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');jumpkey=key('jump');targetkey=key('target');wrong=key('wrong')
forced=root/'forced.py';forced.write_text('import json\nprint(json.dumps({"fixedTarget":True}))\n')
class Backend(socketserver.BaseRequestHandler):
def handle(self):self.request.sendall(b'DR-FORWARDING-OK\n')
backend=socketserver.ThreadingTCPServer(('127.0.0.1',0),Backend);backend.daemon_threads=True
thread=threading.Thread(target=backend.serve_forever,daemon=True);thread.start();backend_port=backend.server_address[1]
@contextlib.contextmanager
def daemon(label,identity,allow='local',permitopen='none',permitlisten='none'):
nonlocal count
count+=1;folder=root/label;folder.mkdir();p=port();authorized=folder/'authorized_keys';authorized.write_text(Path(str(identity)+'.pub').read_text());authorized.chmod(0o600)
conf=folder/'sshd.conf';log=folder/'daemon.log';conf.write_text(f'''ListenAddress 127.0.0.1
Port {p}
HostKey {host}
PidFile {folder}/pid
SshdSessionPath {a.sshd_session}
SshdAuthPath {a.sshd_auth}
AuthorizedKeysFile {authorized}
AllowUsers {user}
AuthenticationMethods publickey
PubkeyAuthentication yes
PasswordAuthentication no
KbdInteractiveAuthentication no
PermitRootLogin no
PermitUserEnvironment no
PermitUserRC no
PermitTTY no
X11Forwarding no
AllowAgentForwarding no
AllowTcpForwarding {allow}
AllowStreamLocalForwarding no
PermitTunnel no
GatewayPorts no
PermitOpen {permitopen}
PermitListen {permitlisten}
StrictModes yes
UseDNS no
LogLevel VERBOSE
ForceCommand {shlex.quote(sys.executable)} {shlex.quote(str(forced))}
SetEnv ZDOTDIR={folder}
''')
call([a.sshd,'-t','-f',conf]);effective=call([a.sshd,'-T','-f',conf]);configs.append(dict(label=label,configuration=conf.read_text(),effective=effective))
with running([a.sshd,'-D','-e','-f',conf],log,label+' daemon')as proc:
await_log(proc,log,'Server listening on ');yield p,log
known=root/'known_hosts';pub=Path(str(host)+'.pub').read_text().split();known.write_text('lab-host '+pub[0]+' '+pub[1]+'\n')
common=f'''Host *
User {user}
HostName 127.0.0.1
HostKeyAlias lab-host
UserKnownHostsFile {known}
GlobalKnownHostsFile /dev/null
StrictHostKeyChecking yes
UpdateHostKeys no
IdentityAgent none
IdentitiesOnly yes
CertificateFile none
BatchMode yes
PasswordAuthentication no
KbdInteractiveAuthentication no
PreferredAuthentications publickey
ConnectTimeout 5
ConnectionAttempts 1
ControlMaster no
ControlPath none
ForwardAgent no
RequestTTY no
ExitOnForwardFailure yes
'''
def clientconfig(label,entries):
f=root/(label+'.conf');f.write_text(''.join(f'Host {name}\n Port {p}\n IdentityFile {identity}\n'for name,p,identity in entries)+common);return f
def read_marker(p):
try:
with socket.create_connection(('127.0.0.1',p),timeout=3)as s:return s.recv(100).decode()
except (ConnectionResetError,OSError):return ''
try:
for label in ['localAllowed','localDenied','backendDown','bindCollision','literalMismatch','remoteAllowed','remoteDenied']:
listen=port();dest=port()if label=='backendDown'else backend_port
remote=label.startswith('remote');allowed=label not in ['localDenied','remoteDenied']
allow='remote'if remote else 'local';po='127.0.0.1:'+str(dest)if allowed and not remote else 'none';pl='127.0.0.1:'+str(listen)if allowed and remote else 'none'
occupied=None
if label=='bindCollision':occupied=socket.socket();occupied.bind(('127.0.0.1',listen));occupied.listen()
try:
with daemon(label,jumpkey,allow,po,pl)as(p,serverlog):
cfg=clientconfig(label,[('endpoint',p,jumpkey)]);log=root/(label+'.client.log');targethost='localhost'if label=='literalMismatch'else '127.0.0.1'
spec='127.0.0.1:'+str(listen)+':'+targethost+':'+str(dest)
argv=[a.ssh,'-F',cfg,'-vv','-N','-R'if remote else '-L',spec,'endpoint']
with running(argv,log,label+' client')as proc:
if label in ['remoteDenied','bindCollision']:
code=proc.wait(timeout=10);check(label+': setup fails',code==255)
term='remote port forwarding failed'if remote else 'Address already in use';check(label+': setup reason recorded',term in log.read_text())
marker='';alive=False
else:
await_log(proc,log,'remote forward success'if remote else 'Local forwarding listening on 127.0.0.1')
marker=read_marker(listen);time.sleep(.05);alive=proc.poll()is None
expected=label in ['localAllowed','remoteAllowed'];check(label+': payload matches expectation',(marker=='DR-FORWARDING-OK\n')==expected)
check(label+': client remains running',alive)
if label in ['localDenied','literalMismatch']:check(label+': policy denial recorded','administratively prohibited' in log.read_text())
if label=='backendDown':check(label+': backend refusal recorded','Connection refused' in log.read_text())
observations[label]=dict(marker=marker,clientAliveBeforeCleanup=alive,clientLog=log.read_text(),serverLog=serverlog.read_text(),requestedDestination=targethost+':'+str(dest))
finally:
if occupied:occupied.close()
for label in ['jumpAllowed','jumpWrongKey','targetWrongKey','jumpForwardDenied']:
with daemon(label+'-target',targetkey,'no')as(tp,tlog):
permitted='none'if label=='jumpForwardDenied'else '127.0.0.1:'+str(tp)
with daemon(label+'-jump',jumpkey,'local',permitted)as(jp,jlog):
cfg=clientconfig(label,[('jump',jp,wrong if label=='jumpWrongKey'else jumpkey),('target',tp,wrong if label=='targetWrongKey'else targetkey)])
x=subprocess.run([a.ssh,'-F',str(cfg),'-vv','-J','jump','target','requested-check'],capture_output=True,text=True,timeout=15)
good=label=='jumpAllowed';check(label+': expected exit',x.returncode==(0 if good else 255))
check(label+': target output',(x.stdout.strip()=='{"fixedTarget": true}')==good)
jumpaccepted='Accepted publickey' in jlog.read_text();targetaccepted='Accepted publickey' in tlog.read_text()
check(label+': jump authentication',jumpaccepted==(label!='jumpWrongKey'))
check(label+': target authentication',targetaccepted==good)
if label=='jumpForwardDenied':check(label+': channel denial','administratively prohibited' in x.stderr)
observations[label]=dict(exit=x.returncode,stdout=x.stdout,stderr=x.stderr,jumpAuthenticated=jumpaccepted,targetAuthenticated=targetaccepted,jumpLog=jlog.read_text(),targetLog=tlog.read_text())
finally:
backend.shutdown();backend.server_close();thread.join(timeout=5);check('backend stopped',not thread.is_alive())
check('temporary keys removed',not root.exists())
report=dict(sshVersion=version,checks=checks,experiments=len(observations),serverProcesses=count,observations=observations,configurations=configs,runnerSha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),scope='Eleven loopback forwarding and ProxyJump experiments. Current local account, disposable keys, fixed target command, no PAM, no personal configuration or agents. No external bastion, production network segmentation, TLS application, dynamic SOCKS, IPv6 or CA rotation tested.')
Path(a.output).write_text(json.dumps(report,indent=2).replace(user,'lab-account')+'\n');print(json.dumps(dict(experiments=len(observations),serverProcesses=count,checks=len(checks),output=a.output)))
if __name__=='__main__':main()
Ria mantém o túnel ativo, mas a reconciliação falha porque o backend não escuta. A equipa restaura o serviço e confirma uma consulta autorizada antes de repetir o processamento.
Armadilhas comuns
Confundir -L com -R, tratar PID como saúde da aplicação, atribuir uma recusa administrativa ao DNS ou ampliar PermitOpen sem autorização para resolver um nome diferente.
Tópicos relacionados: Monitorização e observabilidade · TCP/IP e controlo de acesso
Prova separadamente criação do listener, autorização do canal, disponibilidade do destino e resultado do consumidor. Regista o âmbito que cada experiência realmente cobre.
Referência: OpenSSH client forwarding configuration · OpenSSH concepts and OpenBSD-current manuals consulted 2026-09-29; distribution defaults vary