LEARN. PRACTICE. GROW.Map the endpoints before opening the tunnel
An SSH forward has at least three elements: the listener consumed by the application, the SSH connection, and the final service. With -L, the client creates the listener and the SSH server connects to the requested destination. With -R, the listener is on the server side and the SSH client connects to the backend. The same text 127.0.0.1 can therefore refer to different machines in a real design. In this lab, everything runs on loopback on one machine; processes and ports provide separation. Draw request direction and identify who resolves the final name before investigating DNS, firewalls, or availability. An authenticated session does not demonstrate that the consumer receives a response.
Compare setup, policy, and availability
The runner creates an original TCP fixture returning DR-FORWARDING-OK. localAllowed and remoteAllowed receive that marker through the expected path. bindCollision deliberately occupies the entry port before starting SSH; ExitOnForwardFailure=yes terminates the client when it cannot create the listener. backendDown differs: the listener exists, but the final service is not listening and the channel records Connection refused. The client stays alive. This option does not turn a tunnel into an application availability monitor. localDenied also retains the listener but records administratively prohibited because policy refuses the destination. Compare stage, message, effective configuration, and result before choosing a correction. The same application symptom of no response can have different causes and owners.
Keep authorization and exposure explicit
PermitOpen limits destinations requested from the server by forwarding channels. In literalMismatch, policy allows the address 127.0.0.1 but the request uses localhost. Rejection does not prove a resolution failure: the rule does not resolve names to establish equivalence. PermitListen handles listening addresses and ports requested by remote forwarding. remoteDenied demonstrates setup refusal when that list is none. This lab’s listeners explicitly use loopback and GatewayPorts no. Exposure on external interfaces was not exercised. In a project, binding 0.0.0.0 changes who can reach the entry point; it requires analysis of consumers, access controls, and authorized scope. Avoid replacing an exact permission with any to conceal a configuration mismatch.
Run, interpret, and hand over useful evidence
Copy the complete code into run.py in a disposable working directory. Supply the five matching OpenSSH 10.5p1 executables as described in the next lesson’s guide. The build used was compiled without PAM, and the existing local account must be able to authenticate; no account is created. The program starts eleven experiments, fifteen temporary sshd processes, and a TCP fixture, stopping them afterward. It uses disposable keys, dedicated trust files, and a fixed target command. All 58 checks passed. Repeat with new ports and keys, comparing results without requiring literally identical logs. For RUN, add an authorized synthetic operation and distinguish process health, channel success, and functional outcome. The marker does not validate TLS or real application authentication. Also record the relationship between the entry port and consumer configuration, so recovery does not leave the job using its previous destination.
#!/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.socketas 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.pollis 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.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+': process stopped',proc.pollis 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=portif 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.pollis 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 retains a live tunnel, but reconciliation fails because the backend is not listening. The team restores the service and confirms an authorized query before retrying processing.
Common pitfalls
Confusing -L with -R, treating a PID as application health, attributing administrative rejection to DNS, or broadening PermitOpen without authorization to resolve a different name.
Related topics: Monitoring and observability · TCP/IP and access control
Prove listener creation, channel authorization, destination availability, and consumer result separately. Record the scope each experiment actually covers.
Reference: OpenSSH client forwarding configuration · OpenSSH concepts and OpenBSD-current manuals consulted 2026-09-29; distribution defaults vary