from pathlib import Path
import subprocess as sp, socket, threading, http.server, urllib.request, time, json, hashlib, os, shlex
import sys, platform, getpass
P=Path(sys.argv[1]).resolve(); log=(P/'runtime-transcript.txt').open('w'); results=[]
def say(s): log.write(s+'\n'); log.flush(); print(s,flush=True)
def port():
 with socket.socket() as s: s.bind(('127.0.0.1',0)); return s.getsockname()[1]
def run(args, data=None):
 r=sp.run(args,input=data,stdout=sp.PIPE,stderr=sp.PIPE,timeout=12)
 say('$ '+shlex.join(args)); say('exit='+str(r.returncode)+' stdout='+repr(r.stdout.decode(errors='replace'))+' stderr='+repr(r.stderr.decode(errors='replace')))
 return r
sshport=port(); config=P/'ssh_config'; cfg=''
for alias,opts in [('remote',' RemoteCommand printf USER_REMOTE_COMMAND\\n\n'),('tty',' RequestTTY force\n'),('both',' RemoteCommand printf USER_REMOTE_COMMAND\\n\n RequestTTY force\n'),('plain','')]:
 cfg+=f'Host {alias}\n HostName 127.0.0.1\n Port {sshport}\n User {getpass.getuser()}\n IdentityFile {P}/client\n IdentitiesOnly yes\n StrictHostKeyChecking no\n UserKnownHostsFile /dev/null\n GlobalKnownHostsFile /dev/null\n LogLevel ERROR\n'+opts
config.write_text(cfg); original=config.read_bytes()
(P/'sshd_config').write_text(f'Port {sshport}\nListenAddress 127.0.0.1\nHostKey {P}/host\nPidFile {P}/sshd.pid\nAuthorizedKeysFile {P}/client.pub\nStrictModes no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nUsePAM yes\nAllowUsers {getpass.getuser()}\nAllowTcpForwarding yes\nPermitTTY yes\nLogLevel VERBOSE\n')
say('Runtime sequence transcript, not UI stills/recording. OpenSSH local loopback fixture; no real hosts or user SSH config.'); say('ENVIRONMENT '+platform.platform()+'; '+sp.run(['ssh','-V'],capture_output=True,text=True).stderr.strip()+'; Node '+sp.check_output(['node','--version'],text=True).strip())
sshd=sp.Popen(['sudo','-n','/usr/sbin/sshd','-D','-e','-f',str(P/'sshd_config')],stderr=(P/'sshd.log').open('w'))
class Handler(http.server.BaseHTTPRequestHandler):
 def do_POST(self):
  data=self.rfile.read(int(self.headers['Content-Length'])); self.send_response(200); self.end_headers(); self.wfile.write(data)
 def log_message(self,*a): pass
server=http.server.ThreadingHTTPServer(('127.0.0.1',0),Handler); threading.Thread(target=server.serve_forever,daemon=True).start()
payload=bytes(range(256))*257+b'T3-SSH-PROOF\x00\xff'; digest=hashlib.sha256(payload).hexdigest()
try:
 for i in range(100):
  if sshd.poll() is not None: raise RuntimeError('sshd exited: '+(P/'sshd.log').read_text())
  try:
   with socket.create_connection(('127.0.0.1',sshport),timeout=.1): break
  except OSError: time.sleep(.02)
 for rev in ['base','head']:
  for alias in ['remote','tty','both','plain']:
   # Import the complete unchanged production module snapshot, never imitate its builder.
   args=json.loads(sp.check_output(['node',str(P/'args.ts'),str(P/rev/'command.ts'),str(sshport),'yes'],text=True))
   common=['ssh',*args,'-F',str(config)]
   g=run([*common,'-G','--',alias]); fields=dict(line.split(' ',1) for line in g.stdout.decode().splitlines() if ' ' in line)
   say(f'CONFIG {rev}/{alias}: '+json.dumps({k:fields.get(k) for k in ['hostname','user','port','remotecommand','requesttty']}))
   assert g.returncode==0
   assert fields['hostname']=='127.0.0.1' and fields['user']==getpass.getuser() and fields['port']==str(sshport)
   if rev=='head': assert fields.get('remotecommand') in [None,'none'] and fields['requesttty']=='false'
   script=b'if test -t 0 || test -t 1; then printf "TTY_PRESENT\\n"; exit 42; fi\nprintf "SCRIPT_OK_NO_TTY\\n"\n'
   say('SCRIPT_STDIN '+repr(script))
   r=run([*common,'--',alias,'sh','-s'],script)
   results.append(dict(rev=rev,alias=alias,path='script',ok=r.returncode==0 and r.stdout==b'SCRIPT_OK_NO_TTY\n',exit=r.returncode,stdout=r.stdout.decode(errors='replace'),stderr=r.stderr.decode(errors='replace')))
   for kind in ['desktop','device']:
    local=port()
    opts=['-o','ExitOnForwardFailure=yes','-o','ControlMaster=no','-o','ControlPath=none','-o','ControlPersist=no','-o',f'ServerAliveInterval={15 if kind=="desktop" else 10}','-o','ServerAliveCountMax=3']
    if kind=='desktop':
     generated=json.loads(sp.check_output(['node',str(P/'args.ts'),str(P/rev/'command.ts'),str(sshport),'no'],text=True))
     cmd=['ssh',*generated,'-F',str(config),*opts,'-n','-N','-L',f'{local}:127.0.0.1:{server.server_port}','--',alias]
    else: cmd=[*common,*opts,'-T','-L',f'127.0.0.1:{local}:127.0.0.1:{server.server_port}',alias,'sh','-c',"'exec cat >/dev/null'"]
    say('$ '+shlex.join(cmd)); proc=sp.Popen(cmd,stdin=sp.PIPE,stdout=sp.PIPE,stderr=sp.PIPE)
    ok=False; err=''
    try:
     for attempt in range(80):
      if proc.poll() is not None: break
      try:
       response=urllib.request.urlopen(urllib.request.Request(f'http://127.0.0.1:{local}/proof',data=payload,method='POST'),timeout=.3).read()
       ok=response==payload
       say(f'HTTP POST bytes={len(payload)} response_bytes={len(response)} sha256={hashlib.sha256(response).hexdigest()} expected={digest} equal={ok}')
       break
      except Exception as e: err=str(e); time.sleep(.025)
    finally:
     before=proc.poll()
     if proc.poll() is None: proc.terminate()
     out,stderr=proc.communicate(timeout=5)
    say(f'{rev}/{alias}/{kind}: byte_match={ok} exited_before_cleanup={before} stderr={stderr.decode()!r} last_http_error={err!r}')
    results.append(dict(rev=rev,alias=alias,path=kind,ok=ok,exitBeforeCleanup=before,stderr=stderr.decode()))
 for alias in ['remote','tty','both','plain']:
  g=run(['ssh','-F',str(config),'-G',alias]); f=dict(line.split(' ',1) for line in g.stdout.decode().splitlines() if ' ' in line)
  say('INDEPENDENT_CONFIG '+alias+' '+json.dumps({k:f.get(k) for k in ['remotecommand','requesttty']}))
  assert f.get('remotecommand') == ('printf USER_REMOTE_COMMAND\\n' if alias in ['remote','both'] else None)
  assert f['requesttty']==('force' if alias in ['tty','both'] else 'auto')
 assert config.read_bytes()==original
 say('CONFIG_PRESERVED sha256='+hashlib.sha256(original).hexdigest())
 for row in results:
  expected=not (row['rev']=='base' and ((row['path']=='script' and row['alias']!='plain') or (row['path']=='device' and row['alias'] in ['remote','both'])))
  row['classification']='affected' if ((row['path']=='script' and row['alias']!='plain') or (row['path']=='device' and row['alias'] in ['remote','both'])) else 'compatibility control'
  row['expectedOk']=expected
  assert row['ok']==expected, row
 assert len(results)==24
 (P/'results.json').write_text(json.dumps(results,indent=2))
 say('ASSERTIONS PASSED: 24 executions, all expected outcomes; configuration resolution and preservation verified.')
finally:
 server.shutdown(); sshd.terminate()
 try: sshd.wait(timeout=5)
 except sp.TimeoutExpired:
  sp.run(['sudo','-n','kill','-TERM',str(sshd.pid)],check=True); sshd.wait(timeout=5)
 # root-owned daemon is the explicitly captured PID only.
 pidfile=P/'sshd.pid'
 if pidfile.exists():
  pid=int(pidfile.read_text()); sp.run(['sudo','-n','kill','-TERM',str(pid)],check=False)
 for f in ['client','client.pub','host','host.pub']: (P/f).unlink(missing_ok=True)
 server.server_close()
 assert sshd.poll() is not None
 say('CLEANUP: owned forwarding children stopped, sshd stopped, throwaway private/public keys removed.')
 log.close()