fix(qa): isolate legacy database clients with owned transport guard

scripts/qa/qa.py

@@ -17,6 +17,7 @@ import sys
 import time
 import urllib.parse
 import xml.etree.ElementTree as ET
+import qa_isolation
 
 ROOT = Path(__file__).resolve().parents[2]
 DATABASE = 'railwaysystem_fusion_qa_20261013'
@@ -50,6 +51,8 @@ def environment(config):
         raise ValueError('Refusing non-target database URL')
     if any(k.lower() in ('user', 'password') for k in urllib.parse.parse_qs(parsed.query)):
         raise ValueError('URL credentials forbidden')
+    if parsed.port == qa_isolation.LEGACY_PORT:
+        raise ValueError('Legacy database endpoint is forbidden')
     password = config.get('FUSION_QA_APP_PASSWORD', '')
     if not password:
         raise ValueError('Missing FUSION_QA_APP_PASSWORD')
@@ -115,11 +118,14 @@ def launch(name, command, env):
     child = subprocess.Popen(command, cwd=ROOT, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
     log.close()
     started = subprocess.run(['ps', '-p', str(child.pid), '-o', 'lstart='], text=True, capture_output=True).stdout.strip()
-    ledger.write_text(json.dumps({'pid': child.pid, 'command': command, 'cwd': str(ROOT), 'started': started}))
+    ledger.write_text(json.dumps({'pid': child.pid, 'command': command, 'cwd': str(ROOT), 'started': started,
+                                 'isolation_id': env.get('FUSION_QA_ISOLATION_ID')}))
     print(name + ' PID=' + str(child.pid) + ' LOG=' + str(RUNTIME / (name + '.log')))
 
 
 def start(config, env, backend_port, redis_port):
+    lease = qa_isolation.require(env.get('QA_DB_URL'))
+    env['FUSION_QA_ISOLATION_ID'] = lease['isolation_id']
     if inspect(env, ['--require-migrated']):
         raise ValueError('Backend not started: QA migration preflight failed')
     available(backend_port)
@@ -189,13 +195,15 @@ def wait_for_exit(pid):
     raise ValueError('Task-owned process did not stop; ledger retained, no forced kill')
 
 
-def stop():
+def stop(isolation_id=None):
     private_directory()
     for name in ['frontend', 'backend', 'redis']:
         path = RUNTIME / (name + '.json')
         if not path.exists():
             continue
         item = json.loads(path.read_text())
+        if isolation_id is not None and item.get('isolation_id') != isolation_id:
+            raise ValueError('Other isolation lease owns this runtime ledger; refusing to signal it')
         actual = subprocess.run(['ps', '-p', str(item['pid']), '-o', 'command='], text=True, capture_output=True).stdout.strip()
         if actual:
             # macOS Python wrappers exec Python.app; Redis rewrites its process title.
@@ -233,11 +241,20 @@ def main():
         stop()
         return 0
     if args.action == 'frontend':
+        lease = qa_isolation.require()
+        backend = json.loads((RUNTIME / 'backend.json').read_text())
+        if backend.get('isolation_id') != lease['isolation_id'] or '--server.port=' + str(args.backend_port) not in backend.get('command', []):
+            raise ValueError('Frontend backend belongs to another isolation lease')
         available(args.frontend_port)
+        env = os.environ.copy()
+        env['FUSION_QA_ISOLATION_ID'] = lease['isolation_id']
         launch('frontend', [sys.executable, str(ROOT / 'scripts/qa/serve_frontend.py'),
-               '--port', str(args.frontend_port), '--backend-port', str(args.backend_port)], os.environ.copy())
+               '--port', str(args.frontend_port), '--backend-port', str(args.backend_port)], env)
         return 0
     config = load_config(args.config)
+    if args.action == 'start' or (RUNTIME / qa_isolation.LEDGER).exists():
+        lease = qa_isolation.require()
+        config.setdefault('FUSION_QA_DB_URL', qa_isolation.url_for(lease))
     env = environment(config)
     if args.action == 'inspect':
         return inspect(env, [])

scripts/qa/qa_isolation.py

@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+"""Owned, temporary QA transport. Never revive the legacy database endpoint.
+
+The legacy port is a deny-only socket: no protocol response and no upstream.
+SSH authentication stays interactive; no password is accepted in arguments/files.
+This is runtime isolation, not database credential revocation or a host firewall.
+"""
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import selectors
+import signal
+import socket
+import stat
+import subprocess
+import tempfile
+import time
+import urllib.parse
+
+ROOT = Path(__file__).resolve().parents[2]
+RUNTIME = ROOT / '.qa-runtime'
+LEDGER = 'isolation.json'
+DATABASE = 'railwaysystem_fusion_qa_20261013'
+LEGACY_PORT = 14321
+
+
+def run(args):
+    return subprocess.run(args, text=True, capture_output=True, timeout=8)
+
+
+def identity(pid):
+    born = run(['ps', '-p', str(pid), '-o', 'lstart=']).stdout.strip()
+    command = run(['ps', '-p', str(pid), '-o', 'command=']).stdout.strip()
+    cwd = run(['lsof', '-a', '-p', str(pid), '-d', 'cwd', '-Fn']).stdout.splitlines()
+    return {'pid': pid, 'started': born, 'command': command,
+            'cwd': next((s[1:] for s in cwd if s.startswith('n')), '')}
+
+
+def same_process(item):
+    return bool(item.get('started')) and identity(item['pid']) == item
+
+
+def listener_owned(port, pid):
+    lines = run(['lsof', '-nP', '-iTCP:' + str(port), '-sTCP:LISTEN', '-FpFn']).stdout.splitlines()
+    owners = {int(s[1:]) for s in lines if s.startswith('p')}
+    addresses = {s[1:] for s in lines if s.startswith('n')}
+    return owners == {pid} and addresses == {'127.0.0.1:' + str(port)}
+
+
+def conflicting_clients(port, isolation_id):
+    # Read locally, return only PIDs. Never print/store the captured environments.
+    conflicts = []
+    for line in run(['ps', '-axo', 'pid=,comm=']).stdout.splitlines():
+        fields = line.strip().split(None, 1)
+        if len(fields) != 2 or Path(fields[1]).name != 'java':
+            continue
+        pid = int(fields[0])
+        result = run(['ps', 'eww', '-p', str(pid), '-o', 'command='])
+        if result.returncode or not result.stdout.strip():
+            if run(['ps', '-p', str(pid), '-o', 'pid=']).stdout.strip():
+                raise ValueError('Cannot inspect a running Java client safely')
+            continue
+        text = result.stdout
+        owns_lease = re.search(r'(?:^|\s)FUSION_QA_ISOLATION_ID=' + re.escape(isolation_id) + r'(?:\s|$)', text)
+        if owns_lease and identity(pid)['cwd'] == str(ROOT):
+            continue
+        target = re.search(r'(?:127\.0\.0\.1|localhost|\[::1\]):' + str(port) + r'(?:[/\s?]|$)', text)
+        env_port = re.search(r'(?:^|\s)(?:DB_PORT|QA_DB_PORT)=' + str(port) + r'(?:\s|$)', text)
+        if target or env_port:
+            conflicts.append(pid)
+    return conflicts
+
+
+def private_runtime():
+    RUNTIME.mkdir(mode=0o700, exist_ok=True)
+    info = RUNTIME.stat()
+    if RUNTIME.is_symlink() or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077:
+        raise ValueError('Unsafe isolation directory')
+
+
+def read_lease():
+    private_runtime()
+    path = RUNTIME / LEDGER
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077:
+        raise ValueError('Unsafe isolation ledger')
+    return json.loads(path.read_text())
+
+
+def write_lease(item):
+    fd, name = tempfile.mkstemp(prefix='isolation-', suffix='.tmp', dir=RUNTIME)
+    try:
+        with os.fdopen(fd, 'w') as stream:
+            json.dump(item, stream)
+        os.replace(name, RUNTIME / LEDGER)
+    finally:
+        if Path(name).exists():
+            Path(name).unlink()
+
+
+def validate(item, url=None, scan_clients=True):
+    if item.get('state') != 'READY' or item.get('database') != DATABASE:
+        raise ValueError('No ready QA isolation lease')
+    port = item.get('port')
+    if not isinstance(port, int) or not 1024 <= port <= 65535 or port == LEGACY_PORT:
+        raise ValueError('Legacy or invalid database port refused')
+    if not 0 <= time.time() - item.get('heartbeat', 0) < 10:
+        raise ValueError('Isolation watchdog heartbeat expired')
+    if not same_process(item['owner']) or not same_process(item['ssh']):
+        raise ValueError('Isolation process identity mismatch')
+    if not listener_owned(LEGACY_PORT, item['owner']['pid']) or not listener_owned(port, item['ssh']['pid']):
+        raise ValueError('Isolation listener ownership mismatch')
+    if url:
+        parsed = urllib.parse.urlsplit(url.removeprefix('jdbc:'))
+        if parsed.scheme != 'kingbase8' or parsed.hostname != '127.0.0.1' or parsed.port != port or parsed.path != '/' + DATABASE or parsed.username or parsed.password:
+            raise ValueError('Application must use its owned QA transport')
+    if scan_clients and conflicting_clients(port, item['isolation_id']):
+        raise ValueError('Another Java client depends on the new database endpoint')
+    return item
+
+
+def require(url=None):
+    try:
+        return validate(read_lease(), url)
+    except (OSError, KeyError, TypeError, json.JSONDecodeError) as error:
+        raise ValueError('Managed QA isolation is unavailable') from error
+
+
+def url_for(item):
+    return 'jdbc:kingbase8://127.0.0.1:' + str(item['port']) + '/' + DATABASE
+
+
+def deny_socket():
+    guard = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        # No reuse options: an existing listener/forward must cause refusal.
+        guard.bind(('127.0.0.1', LEGACY_PORT))
+        guard.listen(64)
+        guard.setblocking(False)
+        return guard
+    except BaseException:
+        guard.close()
+        raise
+
+
+def ssh_command(port):
+    # The user's Host * may contain unrelated RemoteForward directives. Read the
+    # jump alias normally but clear its forwards; do not inherit target config.
+    proxy = ('ssh -o ClearAllForwardings=yes -o PermitLocalCommand=no -o ControlMaster=no '
+             '-o ControlPath=none -o ControlPersist=no -o StrictHostKeyChecking=yes '
+             '-W %h:%p wzbmatebook')
+    return ['ssh', '-F', '/dev/null', '-o', 'HostName=192.168.139.218', '-o', 'User=root', '-p', '22',
+            '-o', 'ProxyCommand=' + proxy, '-o', 'StrictHostKeyChecking=yes',
+            '-N', '-o', 'ControlMaster=no', '-o', 'ControlPath=none',
+            '-o', 'ControlPersist=no', '-o', 'ExitOnForwardFailure=yes', '-o', 'ConnectTimeout=8',
+            '-o', 'ServerAliveInterval=10', '-o', 'ServerAliveCountMax=2',
+            '-L127.0.0.1:' + str(port) + ':127.0.0.1:4321', 'sict-218']
+
+
+def serve(port):
+    private_runtime()
+    if (RUNTIME / LEDGER).exists():
+        raise ValueError('Isolation ledger already exists; inspect its owner, never overwrite')
+    if port == LEGACY_PORT or port < 0 or (port and not 1024 <= port <= 65535):
+        raise ValueError('Legacy or invalid database port refused')
+    if not port:
+        with socket.socket() as candidate:
+            candidate.bind(('127.0.0.1', 0))
+            port = candidate.getsockname()[1]
+    lease_id = secrets.token_hex(16)
+    if conflicting_clients(port, lease_id):
+        raise ValueError('New database port is already configured by another Java client')
+    target = run(['ssh', '-G', 'sict-218'])
+    if target.returncode or not all(re.search(pattern, target.stdout, re.M) for pattern in
+            [r'^hostname 192\.168\.139\.218$', r'^user root$', r'^port 22$']):
+        raise ValueError('SSH target alias does not resolve to the approved host')
+    owner = identity(os.getpid())
+    if owner['cwd'] != str(ROOT):
+        raise ValueError('Start the isolation owner from this worktree root')
+    guard = deny_socket()
+    selector = selectors.DefaultSelector()
+    selector.register(guard, selectors.EVENT_READ)
+    child = None
+    item = None
+    def stopped(*_):
+        raise KeyboardInterrupt
+    previous = {s: signal.signal(s, stopped) for s in [signal.SIGTERM, signal.SIGINT]}
+    try:
+        child = subprocess.Popen(ssh_command(port), cwd=ROOT)
+        deadline = time.monotonic() + 90
+        refused = 0
+        while True:
+            if child.poll() is not None:
+                raise ValueError('Owned SSH transport exited; QA isolation refused')
+            for _, _ in selector.select(timeout=0.4):
+                try:
+                    connection, _address = guard.accept()
+                except BlockingIOError:
+                    continue
+                else:
+                    connection.close()  # Deliberately no read, write, proxy or DB authentication.
+                    refused += 1
+            newly_ready = item is None
+            if newly_ready:
+                if not listener_owned(port, child.pid):
+                    if time.monotonic() > deadline:
+                        raise ValueError('Owned SSH transport readiness timed out')
+                    continue
+                item = {'isolation_id': lease_id, 'state': 'READY', 'database': DATABASE, 'port': port,
+                        'target': 'sict-218:127.0.0.1:4321', 'owner': owner, 'ssh': identity(child.pid),
+                        'created': time.time(), 'heartbeat': time.time(), 'denied_legacy_connections': refused}
+            item.update(heartbeat=time.time(), denied_legacy_connections=refused)
+            # Guards and client dependencies are checked on each heartbeat, not just once at launch.
+            validate(item)
+            write_lease(item)
+            if newly_ready:
+                print('QA_ISOLATION_READY port=' + str(port) + ' legacy_port=DENY_ONLY', flush=True)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        # Stop only matching lease-owned applications before removing their transport and guard.
+        cleanup_ok = False
+        try:
+            import qa
+            qa.stop(isolation_id=lease_id)
+            cleanup_ok = True
+        finally:
+            if child is not None and child.poll() is None:
+                child.terminate()
+                try:
+                    child.wait(timeout=12)
+                except subprocess.TimeoutExpired:
+                    # Exact child Popen object, never a name/port-selected process.
+                    child.kill()
+                    child.wait(timeout=5)
+            guard.close()
+            selector.close()
+            for sig, handler in previous.items():
+                signal.signal(sig, handler)
+            if item:
+                item.update(state='STOPPED' if cleanup_ok else 'CLEANUP_FAILED', heartbeat=time.time(),
+                            denied_legacy_connections=refused, cleanup_ok=cleanup_ok)
+                write_lease(item)
+                history = RUNTIME / ('isolation-' + lease_id + '-final.json')
+                os.replace(RUNTIME / LEDGER, history)
+                print('QA_ISOLATION_CLOSED evidence=' + str(history), flush=True)
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--port', type=int, default=0, help='0 selects an unused ephemeral port; legacy port is always refused')
+    args = parser.parse_args()
+    try:
+        serve(args.port)
+    except (ValueError, OSError, subprocess.TimeoutExpired) as error:
+        print('QA_ISOLATION_REFUSED: ' + (str(error) if isinstance(error, ValueError) else type(error).__name__))
+        raise SystemExit(2)

scripts/qa/test_qa_isolation.py

@@ -0,0 +1,132 @@
+"""Targeted local tool tests. No server connection, credentials, database or device writes."""
+import json
+from pathlib import Path
+import socket
+import select
+import tempfile
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+import qa
+import qa_isolation as isolation
+
+
+class QaIsolationTest(unittest.TestCase):
+    def item(self):
+        return {'state':'READY','database':isolation.DATABASE,'port':15432,'heartbeat':time.time(),
+                'isolation_id':'synthetic','owner':{'pid':101},'ssh':{'pid':102}}
+
+    def test_legacy_url_is_refused_before_database_access(self):
+        for host in ['127.0.0.1','localhost','192.168.139.218']:
+            with self.assertRaises(ValueError):
+                qa.environment({'FUSION_QA_DB_URL':'jdbc:kingbase8://'+host+':14321/'+qa.DATABASE,
+                                'FUSION_QA_APP_PASSWORD':'synthetic'})
+
+    def test_no_lease_prevents_even_database_preflight(self):
+        with patch.object(isolation,'require',side_effect=ValueError('synthetic refusal')), \
+                patch.object(qa,'inspect') as inspect, patch.object(qa,'launch') as launch:
+            with self.assertRaises(ValueError):qa.start({}, {}, 18096, 16389)
+            inspect.assert_not_called();launch.assert_not_called()
+
+    def test_ready_owned_transport_allowed(self):
+        item=self.item()
+        with patch.object(isolation,'same_process',return_value=True), \
+                patch.object(isolation,'listener_owned',return_value=True), \
+                patch.object(isolation,'conflicting_clients',return_value=[]):
+            self.assertEqual(item,isolation.validate(item,isolation.url_for(item)))
+
+    def test_expired_or_future_heartbeat_refused(self):
+        for delta in [-11,60]:
+            item=self.item();item['heartbeat']=time.time()+delta
+            with self.assertRaises(ValueError):isolation.validate(item)
+
+    def test_legacy_and_invalid_ports_refused(self):
+        for port in [14321,0,-1,65536,'15432']:
+            item=self.item();item['port']=port
+            with self.assertRaises(ValueError):isolation.validate(item)
+
+    def test_wrong_state_or_database_refused(self):
+        for key,value in [('state','STOPPED'),('database','training')]:
+            item=self.item();item[key]=value
+            with self.assertRaises(ValueError):isolation.validate(item)
+
+    def test_reused_pid_refused(self):
+        with patch.object(isolation,'same_process',return_value=False):
+            with self.assertRaises(ValueError):isolation.validate(self.item())
+
+    def test_unknown_listener_refused(self):
+        with patch.object(isolation,'same_process',return_value=True),patch.object(isolation,'listener_owned',return_value=False):
+            with self.assertRaises(ValueError):isolation.validate(self.item())
+
+    def test_wrong_application_endpoint_refused(self):
+        for url in ['jdbc:kingbase8://127.0.0.1:14321/'+qa.DATABASE,
+                    'jdbc:kingbase8://192.168.139.218:4321/'+qa.DATABASE,
+                    'jdbc:kingbase8://127.0.0.1:15432/training']:
+            with patch.object(isolation,'same_process',return_value=True),patch.object(isolation,'listener_owned',return_value=True):
+                with self.assertRaises(ValueError):isolation.validate(self.item(),url)
+
+    def test_another_java_client_is_not_allowed_to_share_endpoint(self):
+        with patch.object(isolation,'same_process',return_value=True), \
+                patch.object(isolation,'listener_owned',return_value=True), \
+                patch.object(isolation,'conflicting_clients',return_value=[999]):
+            with self.assertRaises(ValueError):isolation.validate(self.item())
+
+    def test_deny_socket_has_no_upstream_and_rejects_data(self):
+        # Use an OS-selected temporary port, never touch the actual legacy port in unit tests.
+        with patch.object(isolation,'LEGACY_PORT',0):guard=isolation.deny_socket()
+        try:
+            with socket.create_connection(guard.getsockname(),timeout=1) as client:
+                self.assertTrue(select.select([guard],[],[],2)[0])
+                peer,_=guard.accept();peer.close()
+                self.assertEqual(b'',client.recv(1))
+        finally:guard.close()
+
+    def test_existing_listener_not_replaced(self):
+        with socket.socket() as listener:
+            listener.bind(('127.0.0.1',0));listener.listen(1)
+            with patch.object(isolation,'LEGACY_PORT',listener.getsockname()[1]):
+                with self.assertRaises(OSError):isolation.deny_socket()
+            self.assertNotEqual(-1,listener.fileno())
+
+    def test_tunnel_never_reuses_a_controlmaster_or_exposes_password(self):
+        command=isolation.ssh_command(15432)
+        self.assertIn('ControlMaster=no',command);self.assertIn('ControlPath=none',command)
+        self.assertEqual('/dev/null',command[command.index('-F')+1])
+        self.assertTrue(any(p.startswith('ProxyCommand=') and 'ClearAllForwardings=yes' in p for p in command))
+        self.assertIn('ExitOnForwardFailure=yes',command)
+        self.assertIn('-L127.0.0.1:15432:127.0.0.1:4321',command)
+        self.assertNotIn('-f',command);self.assertNotIn('-g',command)
+        self.assertFalse(any('password=' in p.lower() for p in command))
+
+    def test_lease_file_is_private_and_symlink_refused(self):
+        with tempfile.TemporaryDirectory() as directory,patch.object(isolation,'RUNTIME',Path(directory)):
+            isolation.write_lease({'synthetic':True});path=Path(directory)/isolation.LEDGER
+            self.assertEqual(0o600,path.stat().st_mode & 0o777)
+            self.assertEqual({'synthetic':True},isolation.read_lease())
+            path.unlink();path.symlink_to(Path(directory)/'other')
+            with self.assertRaises((ValueError,OSError)):isolation.read_lease()
+
+    def test_stop_cannot_signal_another_lease(self):
+        with tempfile.TemporaryDirectory() as directory,patch.object(qa,'RUNTIME',Path(directory)),patch.object(qa.os,'kill') as kill:
+            (Path(directory)/'backend.json').write_text(json.dumps({'pid':123,'isolation_id':'other'}))
+            with self.assertRaises(ValueError):qa.stop(isolation_id='synthetic')
+            kill.assert_not_called()
+
+    def test_foreign_client_scan_does_not_report_environment_secrets(self):
+        def fake(args):
+            if args[:2]==['ps','-axo']:return SimpleNamespace(stdout='123 /usr/bin/java\n',returncode=0)
+            return SimpleNamespace(stdout='java DB_PORT=15432 SOME_PASSWORD=synthetic_private_value',returncode=0)
+        with patch.object(isolation,'run',side_effect=fake):
+            self.assertEqual([123],isolation.conflicting_clients(15432,'synthetic'))
+
+    def test_frontend_cannot_proxy_an_old_backend_with_matching_lease_label(self):
+        with tempfile.TemporaryDirectory() as directory,patch.object(qa,'RUNTIME',Path(directory)), \
+                patch.object(isolation,'require',return_value={'isolation_id':'synthetic'}), \
+                patch.object(qa,'launch') as launch,patch('sys.argv',['qa.py','frontend','--backend-port','8096']):
+            (Path(directory)/'backend.json').write_text(json.dumps({'isolation_id':'synthetic','command':['java','--server.port=18096']}))
+            with self.assertRaises(ValueError):qa.main()
+            launch.assert_not_called()
+
+
+if __name__=='__main__':unittest.main()

scripts/qa/test_qa_tools.py

@@ -63,13 +63,15 @@ class QaToolsTest(unittest.TestCase):
             self.assertEqual('$(touch should-not-exist)', qa.load_config(path)['KEY'])
 
     def test_migration_preflight_prevents_all_processes(self):
-        with patch.object(qa, 'inspect', return_value=2), patch.object(qa, 'launch') as launch:
+        with patch.object(qa.qa_isolation, 'require', return_value={'isolation_id':'synthetic'}), \
+                patch.object(qa, 'inspect', return_value=2), patch.object(qa, 'launch') as launch:
             with self.assertRaises(ValueError):
                 qa.start({}, {}, 18096, 16389)
             launch.assert_not_called()
 
     def test_start_flags_and_secret_handling(self):
         with tempfile.TemporaryDirectory() as directory, patch.object(qa, 'RUNTIME', Path(directory)), \
+                patch.object(qa.qa_isolation, 'require', return_value={'isolation_id':'synthetic'}), \
                 patch.object(qa, 'inspect', return_value=0), patch.object(qa, 'available'), patch.object(qa, 'launch') as launch:
             config = {'FUSION_QA_APP_PASSWORD':'synthetic', 'TRAIN_STUDENT_IDENTITY_ENCRYPTION_KEY':'synthetic', 'TRAIN_STUDENT_IDENTITY_HASH_KEY':'synthetic'}
             qa.start(config, qa.environment(config), 18096, 16389)