import copy import asyncio import io import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch, Mock import connection_versions as versions import field_profiles as fp import field_cli as cli import field_ui as ui import field_usb as usb import field_resources as resources import maintenance_process import vscode_safety from tools.secure_bridge_enroll import safe_error, Transport from control_sidecar.protocol import BoundaryError def body(version=1): return dict(connectionConfigVersion=version,advertisedIp='192.0.2.1',controlPort=38443, bboxStreamKey='fixture',redis=dict(host='192.0.2.1',port=16379,db=0,username='reader',password='fixture-only',tls=False)) class RecoveryTests(unittest.TestCase): def test_version_range_and_boolean(self): for value in (0,-1,True,'2',versions.MAX_VERSION+1): with self.assertRaises(ValueError):versions.valid(value) self.assertEqual(versions.valid(100),100) def test_unchanged_reuses_and_changed_advances(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)): key=versions.context('https://192.0.2.42','AI-01');b=body(17) self.assertEqual(versions.choose(key,b,baseline=b),17) newer=copy.deepcopy(b);newer['advertisedIp']='192.0.2.2' self.assertEqual(versions.choose(key,newer,baseline=b),18) versions.remember(key,Path('/fixture'),newer|{'connectionConfigVersion':18},'prepared') self.assertEqual(versions.choose(key,newer,baseline=b),18) self.assertNotIn('successful',versions.read(key)) def test_peer_and_identity_are_independent(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)): a=versions.context('https://192.0.2.42','AI-01');b=body(40) versions.remember(a,Path('/fixture'),b,'successful') for key in (versions.context('https://192.0.2.43','AI-01'),versions.context('https://192.0.2.42','AI-02')): self.assertEqual(versions.choose(key,b),1) self.assertEqual(versions.choose(a,b,100),100) def test_password_and_ca_rotation_not_revision_content(self): a=body();b=copy.deepcopy(a);b.update(password='login-change',aiControlCa='public-rotation') self.assertEqual(versions.fingerprint(a),versions.fingerprint(b)) b['redis']['password']='changed-redis' self.assertNotEqual(versions.fingerprint(a),versions.fingerprint(b)) def test_profile_optional_manual_version(self): p=fp.create('office','fixture','eno1','192.0.2.1/24','192.0.2.42') p['webs'][0]['connection_version']=100 self.assertEqual(fp.validate(p)['version'],1) p['webs'][0]['connection_version']=True with self.assertRaises(ValueError):fp.validate(p) def test_version_change_preserves_pending(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)): p=fp.create('office','fixture','eno1','192.0.2.1/24','192.0.2.42') fp.atomic(fp.STATE/'pending.json',{'stage':'preserve'}) with patch.object(fp,'load',return_value=p),patch.object(fp,'save') as save: with self.assertRaises(ValueError):versions.configure(['2']) save.assert_not_called() self.assertEqual(fp.read_json(fp.STATE/'pending.json')['stage'],'preserve') def test_error_is_allowlisted_and_drops_secrets(self): raw=json.dumps({'error':{'code':'INVALID_PARAMS','message':'private-token', 'details':{'aiConnection':{'code':'CONNECTION_VERSION_CONFLICT','field':'advertisedIp','value':'secret'}}}}).encode() result=safe_error(raw,'LOGIN',400) self.assertEqual(result['remote_code'],'CONNECTION_VERSION_CONFLICT') self.assertNotIn('secret',json.dumps(result));self.assertNotIn('private-token',json.dumps(result)) self.assertEqual(safe_error(b'bad','LOGIN',400),{'phase':'LOGIN','status':400}) self.assertNotIn('remote_code',safe_error(b'{"error":{"code":"leaked-secret"}}','LOGIN',400)) self.assertNotIn('remote_code',safe_error(b'{"error":{"code":{}}}','LOGIN',400)) def test_certificate_reason_is_allowlisted_and_phase_scoped(self): raw=json.dumps({'error':{'code':'INTERNAL_ERROR','details':{ 'reason':'CERTIFICATE_SIGNING_NOT_CONFIGURED','key':'private-value'}}}).encode() result=safe_error(raw,'CERTIFICATE',500) self.assertEqual(result['remote_reason'],'CERTIFICATE_SIGNING_NOT_CONFIGURED') self.assertNotIn('private-value',json.dumps(result)) self.assertNotIn('remote_reason',safe_error(raw,'LOGIN',500)) for value in ('secret-value', 'UNREVIEWED_REASON', {}, ['CERTIFICATE_SIGNING_FAILED']): raw=json.dumps({'error':{'code':'INTERNAL_ERROR','details':{'reason':value}}}).encode() self.assertNotIn('remote_reason',safe_error(raw,'CERTIFICATE',500)) def test_saved_version_output_reads_new_value(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)),patch.dict(os.environ,{'CONTEXTRA_CLI_LANG':'en'}): p=fp.create('office','fixture','eno1','192.0.2.1/24','192.0.2.42') output=io.StringIO() with patch.object(fp,'load',return_value=p) as load,patch.object(fp,'save') as save,patch('sys.stdout',output): versions.configure(['2']) self.assertEqual(p['webs'][0]['connection_version'],2) save.assert_called_once_with(p);load.assert_called_with('office') self.assertIn('connection_version override: 2',output.getvalue()) self.assertNotIn('connection_version override: auto',output.getvalue()) def test_certificate_guidance_preserves_session_and_version(self): import field_diagnostics output=io.StringIO() with patch.dict(os.environ,{'CONTEXTRA_CLI_LANG':'ko'}),patch('sys.stdout',output): field_diagnostics.explain_result({'phase':'CERTIFICATE','http_status':500, 'remote_code':'INTERNAL_ERROR','remote_reason':'CERTIFICATE_SIGNING_NOT_CONFIGURED'}) self.assertIn('Web 인증서 발급 설정이 없습니다',output.getvalue()) self.assertIn('준비본과 세션을 보존',output.getvalue()) self.assertNotIn('ct config cancel;',output.getvalue()) def test_enrollment_summary_preserves_safe_reason(self): import network_prepare with tempfile.TemporaryDirectory() as root: data=Path(root);stage=data/'site'/'fixture' for name in ('tls','bridge-session'):(stage/name).mkdir(parents=True) fp.atomic(stage/'record.json',{'base_active_sha256':'baseline','desired':{'ai_ip':'192.0.2.1'}}) fp.atomic(stage/'host-enrollment.json',{}) detail=safe_error(json.dumps({'error':{'code':'INTERNAL_ERROR','details':{ 'reason':'CERTIFICATE_SIGNING_NOT_CONFIGURED','private':'must-not-log'}}}).encode(),'CERTIFICATE',500) failed=Mock(returncode=2,stdout=json.dumps(detail),stderr='') output=io.StringIO() with patch.object(network_prepare,'DATA',data),patch.object(network_prepare,'sha',return_value='baseline'),patch.object(network_prepare.subprocess,'run',return_value=Mock(stdout='')),patch.object(maintenance_process,'run',return_value=failed),patch('sys.stdout',output): with self.assertRaises(RuntimeError):network_prepare.enroll(stage) result=fp.read_json(stage/'enroll-result.json') self.assertEqual(result['remote_reason'],'CERTIFICATE_SIGNING_NOT_CONFIGURED') self.assertEqual(result['attempt_count'],1) self.assertNotIn('must-not-log',output.getvalue()) def test_retry_cancel_does_not_archive_intent(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)): stage=Path(root)/'stage';(stage/'bridge').mkdir(parents=True) fp.atomic(stage/'enroll-intent.json',{'stage':'INTENT'}) fp.atomic(stage/'bridge/login-body.json',body()) fp.atomic(fp.STATE/'pending.json',{'stage':str(stage)}) p=fp.create('office','fixture','eno1','192.0.2.1/24','192.0.2.42') import field_runtime as runtime with patch.object(runtime,'pending_stage',return_value=stage),patch.object(runtime,'verify_pending',return_value=({},p)),patch.object(runtime,'stage_ready',side_effect=ValueError('unconfirmed')),patch('builtins.input',side_effect=['RETRY','']): cli.config_action(['retry']) self.assertTrue((stage/'enroll-intent.json').exists()) self.assertEqual(list(stage.glob('*.previous.json')),[]) def test_language_ascii_and_persistence(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)),patch.dict(os.environ,{'CONTEXTRA_CLI_LANG':'en'}): self.assertIn('Environment',ui.text('환경:')) self.assertTrue(ui.text('한글폴더/ca.crt').isascii()) output=io.StringIO() with patch('sys.stdout',output):ui.configure(['en']) self.assertEqual(fp.read_json(fp.STATE/'language.json')['language'],'en') self.assertTrue(output.getvalue().isascii()) def test_physical_console_auto_english_ssh_korean(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)),patch.dict(os.environ,{},clear=True): with patch.object(os,'ttyname',return_value='/dev/tty1'):self.assertEqual(ui.language(),'en') with patch.object(os,'ttyname',return_value='/dev/pts/1'):self.assertEqual(ui.language(),'ko') def test_usb_only_safe_filesystems(self): row={'name':'/dev/sdb','type':'disk','tran':'usb','children':[{'name':'/dev/sdb1','type':'part','fstype':'exfat','mountpoints':[],'uuid':'fixture','maj:min':'8:17','size':100}]} self.assertEqual(len(usb.candidates({'blockdevices':[row]})),1) for mount in ('/','/backup','/boot/efi','[SWAP]'): bad=copy.deepcopy(row);bad['children'][0]['mountpoints']=[mount] self.assertEqual(usb.candidates({'blockdevices':[bad]}),[]) row['tran']='sata';self.assertEqual(usb.candidates({'blockdevices':[row]}),[]) def test_usb_changed_identity_never_mounts(self): with patch.object(usb,'inventory',return_value=[]),patch.object(usb.subprocess,'run') as run: with self.assertRaises(ValueError):usb.mount({'name':'/dev/sdb1','uuid':'before'}) run.assert_not_called() def test_process_refuses_protected_or_unconfirmed(self): with self.assertRaises(ValueError):resources.stop({'uid':0,'role':'other'},False) with patch.object(fp,'account',return_value=type('U',(),{'pw_uid':1001})()): with self.assertRaises(ValueError):resources.stop({'uid':0,'role':'other'},True) def test_reused_pid_never_signaled(self): expected=dict(pid=99,ppid=1,uid=1001,start_ticks=5,role='vscode-file-search',exe='/fixture') with patch.object(fp,'account',return_value=type('U',(),{'pw_uid':1001})()),patch.object(os,'pidfd_open',return_value=123),patch.object(os,'close'),patch.object(resources,'process',return_value=expected|{'start_ticks':6}),patch.object(resources.signal,'pidfd_send_signal') as kill: with self.assertRaises(ValueError):resources.stop(expected,True) kill.assert_not_called() def test_owned_subprocess_timeout_reaped(self): with self.assertRaises(subprocess.TimeoutExpired): maintenance_process.run([sys.executable,'-c','import time;time.sleep(20)'],timeout=.05,capture_output=True) self.assertEqual(maintenance_process.run([sys.executable,'-c','print(123)'],capture_output=True,text=True).stdout.strip(),'123') def test_next_step_credentials_before_prepare(self): with tempfile.TemporaryDirectory() as root,patch.object(fp,'PROFILES',Path(root)): p=fp.create('office','fixture','eno1','192.0.2.1/24','192.0.2.42');output=io.StringIO() with patch('sys.stdout',output):cli.next_step(p) self.assertIn('ct profile credentials',output.getvalue()) def test_usb_unmount_protects_data_and_paths(self): for path in ('/','/backup','/training','/workspace/contextra/data/artifacts','/mnt/other'): self.assertFalse(usb.unmountable(path)) with patch.object(usb.subprocess,'run') as run: with self.assertRaises(ValueError):usb.unmount({},path) run.assert_not_called() def test_usb_unmount_rechecks_device_and_nested_mounts(self): selected={'name':'/dev/sdb1','maj:min':'8:17','mountpoints':['/mnt/ct-usb']} doc={'filesystems':[{'source':'/dev/sdb1','target':'/mnt/ct-usb','maj:min':'8:17','children':[{}]}]} with patch.object(usb,'inventory',return_value=[selected]),patch.object(usb.subprocess,'run',return_value=Mock(stdout=json.dumps(doc))) as run: with self.assertRaises(ValueError):usb.unmount(selected,'/mnt/ct-usb') self.assertEqual(run.call_count,1) with patch.object(usb,'inventory',return_value=[]),patch.object(usb.subprocess,'run') as run: with self.assertRaises(ValueError):usb.unmount(selected,'/mnt/ct-usb') run.assert_not_called() def test_usb_unmount_never_forces_busy_volume(self): selected={'name':'/dev/sdb1','maj:min':'8:17','mountpoints':['/mnt/ct-usb']} doc={'filesystems':[{'source':'/dev/sdb1','target':'/mnt/ct-usb','maj:min':'8:17'}]} calls=[] def run(args,**kwargs): calls.append(args) if args[0]=='umount':raise subprocess.CalledProcessError(32,args) return Mock(stdout=json.dumps(doc)) with patch.object(usb,'inventory',return_value=[selected]),patch.object(usb.subprocess,'run',side_effect=run): with self.assertRaises(subprocess.CalledProcessError):usb.unmount(selected,'/mnt/ct-usb') self.assertEqual(calls[-1],['umount','--','/mnt/ct-usb']) def test_vscode_preserves_browsing_and_unrelated_settings(self): original={'editor.fontSize':17,'files.exclude':{'custom':False},'search.exclude':{'custom':True}} new,changes=vscode_safety.plan(original) self.assertEqual(new['files.exclude'],original['files.exclude']) self.assertEqual(new['editor.fontSize'],17);self.assertTrue(new['search.exclude']['custom']) self.assertEqual(new['npm.autoDetect'],'off') self.assertEqual(new['python.analysis.diagnosticMode'],'openFilesOnly') self.assertFalse(new['python.analysis.indexing']) self.assertEqual(vscode_safety.plan(new)[1],[]) def test_gateway_force_warning_after_other_cleanup(self): import ct seen=[] def execute(args,*unused,**kw): seen.append(args) if args[:2]==['docker','ps'] and len([x for x in seen if x[:2]==['docker','ps']])==1: return Mock(stdout='contextra-icerti-ai-gateway-1\n') if args[:2]==['docker','inspect']:return Mock(stdout='{"ExitCode":137,"OOMKilled":false}') return Mock(stdout='') with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)),patch.object(ct,'admin'),patch.object(ct,'command',side_effect=execute): with self.assertRaisesRegex(RuntimeError,'Not graceful PASS'):ct.runtime('stop') self.assertIn(['systemctl','stop',ct.SERVICE],seen) self.assertFalse(fp.read_json(fp.STATE/'last-stop-warning.json')['graceful_verified']) def test_transport_error_summary_without_body_or_retry(self): import tools.secure_bridge_enroll as enroll raw=b'{"error":{"code":"INVALID_PARAMS","details":{"aiConnection":{"code":"STALE_CONNECTION_VERSION"}},"message":"private-value"}}' response=Mock();response.code=400;response.read.return_value=raw response.__enter__=Mock(return_value=response);response.__exit__=Mock(return_value=False) opener=Mock();opener.open.return_value=response;tx=Transport();tx.ca='fixture' output=io.StringIO() with patch.object(enroll.ssl,'create_default_context',return_value=None),patch.object(enroll.urllib.request,'build_opener',return_value=opener),patch('sys.stdout',output): code,_=asyncio.run(tx.request('POST','https://example.invalid/api/v1/auth/login')) self.assertEqual(code,400);opener.open.assert_called_once() self.assertIn('STALE_CONNECTION_VERSION',output.getvalue());self.assertNotIn('private-value',output.getvalue()) if __name__=='__main__':unittest.main()