import configparser import importlib.util import json from pathlib import Path import sys import tempfile import subprocess import os import shutil import unittest from unittest.mock import patch ROOT=Path(__file__).resolve().parents[1] sys.path.insert(0,str(ROOT/'scripts')) import root_custody_finish as finish import root_grant_attended as owner spec=importlib.util.spec_from_file_location('validator',ROOT.parent/'net-kingdom/tools/custody-record/validate.py') validator=importlib.util.module_from_spec(spec);spec.loader.exec_module(validator) class FinishTests(unittest.TestCase): def test_retained_source_copy_contains_latest_valid_history(self): path=ROOT/'docs/evidence/root-custody-history/baseline.json' state,bundle=finish.checked_bundle(path,validator) self.assertEqual(state['document']['entries'],[]) self.assertIn(state,json.loads(bundle)['history']) def test_current_historical_share_is_classified_without_values(self): with tempfile.TemporaryDirectory() as tmp: path=Path(tmp)/'rclone.conf' path.write_text('[nextcloud-backup]\nurl=https://example.invalid/public.php/dav/filesdrop/TEST-SHARE-VALUE\nuser=TEST-SHARE-VALUE\n') result=finish.historical_config_comparison(path,{'NC_WEBDAV_TOKEN':'TEST-SHARE-VALUE'}) self.assertTrue(result['historical_share_matches_current_lane']) self.assertNotIn('TEST-SHARE-VALUE',json.dumps(result)) result=finish.historical_config_comparison(path,{'NC_WEBDAV_TOKEN':'CURRENT-DIFFERENT'}) self.assertFalse(result['historical_share_matches_current_lane']) self.assertEqual(result['historical_credential_status'],'historical-owner-and-revocation-unverified') def test_unverified_upload_destination_refused_before_network(self): lane={'data':{'NC_WEBDAV_URL':'https://unapproved.invalid/private','NC_WEBDAV_TOKEN':'TEST'}} with patch.object(finish,'request') as network: with self.assertRaises(ValueError): finish.roundtrip(b'{}',lane,{'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'TEST'},network=network) network.assert_not_called() def test_actual_age_roundtrip_with_disposable_key_and_memory_provider(self): key=subprocess.run(['age-keygen'],capture_output=True,check=True).stdout.decode() lane={'data':{'NC_WEBDAV_URL':finish.HOST+'/public.php/dav/filesdrop/DISPOSABLE-SHARE', 'NC_WEBDAV_TOKEN':'DISPOSABLE-SHARE','AGE_PRIVATE_KEY':key},'metadata':{'version':9}} stored={} def memory_provider(url,method='GET',body=None,auth=None,headers=None): if method=='PUT': self.assertEqual(headers['If-None-Match'],'*') stored['encrypted']=body return 201,b'' return 200,stored['encrypted'] with tempfile.TemporaryDirectory() as tmp: escrow=Path(tmp)/'escrow.age.key' escrow.write_text(key);escrow.chmod(0o600) with patch.object(finish,'INDEPENDENT_ESCROW',escrow): receipt=finish.roundtrip(b'{"nonsecret":"receipt-fixture"}',lane, {'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'DISPOSABLE-PASSWORD'}, network=memory_provider,get_quota=lambda auth:{'available_bytes':1048576}) self.assertTrue(receipt['decrypted_history_matches']) self.assertNotIn(key,json.dumps(receipt)) self.assertNotIn('DISPOSABLE-SHARE',json.dumps(receipt)) def test_missing_latest_history_cannot_make_backup(self): with tempfile.TemporaryDirectory() as tmp: path=Path(tmp)/'baseline.json' path.write_text((ROOT/'docs/evidence/root-custody-history/baseline.json').read_text()) (path.parent/'history').mkdir() with self.assertRaises(ValueError): finish.checked_bundle(path,validator) class MainIntegrationTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory() self.directory=Path(self.temp.name) shutil.copytree(ROOT/'docs/evidence/root-custody-history',self.directory/'holder') self.state=self.directory/'holder/baseline.json' self.receipt=self.directory/'receipt.json' self.args=['finish','--state',str(self.state),'--net-kingdom',str(ROOT.parent/'net-kingdom'),'--receipt',str(self.receipt)] self.baseline=json.loads(self.state.read_text()) key=subprocess.run(['/usr/bin/age-keygen'],capture_output=True,check=True).stdout.decode() self.lane={'data':{'NC_WEBDAV_URL':finish.HOST+'/public.php/dav/filesdrop/TEST-SHARE', 'NC_WEBDAV_TOKEN':'TEST-SHARE','AGE_PRIVATE_KEY':key},'metadata':{'version':4}} self.operator={'data':{'BACKUP_USERNAME':'Backup','BACKUP_PASSWORD':'TEST-PASSWORD'}} self.escrow=self.directory/'disposable-escrow.key' self.escrow.write_text(key);self.escrow.chmod(0o600) self.stored={} def tearDown(self):self.temp.cleanup() def read(self,path): if path==finish.LANE:return self.lane if path==finish.OPERATOR_PATH:return self.operator return {'data':{'document':self.baseline['document']},'metadata':{'version':1}} def provider(self,url,method='GET',body=None,auth=None,headers=None): if method=='PUT':self.stored['encrypted']=body;return 201,b'' return 200,self.stored['encrypted'] def invoke(self,reader=None,provider=None,quota_failure=None,comparison=None): # The actual attended-envelope predicate runs with private fake home; # only owner I/O boundaries are doubled. System deps and age stay real. actor=self.directory/'.warden-attended-login/operator' with patch.object(sys,'argv',self.args), patch.object(owner.Path,'home',return_value=actor), \ patch.dict(os.environ,{'BAO_ADDR':'http://127.0.0.1:18200','PATH':'/usr/bin:/bin'},clear=True), \ patch.object(finish.Bao,'read',side_effect=reader or self.read), \ patch.object(finish,'request',side_effect=provider or self.provider), \ patch.object(finish,'INDEPENDENT_ESCROW',self.escrow), \ patch.object(finish,'quota',side_effect=quota_failure,return_value={'available_bytes':1048576}), \ patch.object(finish,'historical_config_comparison',return_value=comparison or {'historical_config_present':False,'historical_share_matches_current_lane':None}): return finish.main() def test_full_attended_main_success_real_crypto_fake_owner_and_provider(self): self.assertEqual(self.invoke(),0) receipt=json.loads(self.receipt.read_text()) self.assertTrue(receipt['ok']);self.assertEqual(receipt['stage'],'complete') self.assertTrue(receipt['current_live_baseline_matches']) self.assertTrue(receipt['decrypted_history_matches']) self.assertEqual(receipt['upload_http_status'],201) self.assertNotIn('TEST-PASSWORD',self.receipt.read_text()) def test_permission_or_backend_error_writes_safe_exact_stage(self): def failure(path): if path==finish.OPERATOR_PATH:raise RuntimeError('DO-NOT-LOG-CREDENTIAL-OR-BACKEND-TEXT') return self.read(path) self.assertEqual(self.invoke(reader=failure),1) receipt=json.loads(self.receipt.read_text()) self.assertFalse(receipt['ok']);self.assertEqual(receipt['stage'],'backup_operator_read') self.assertTrue(receipt['current_live_baseline_matches']) self.assertNotIn('DO-NOT-LOG',self.receipt.read_text()) def test_upload_failure_has_status_and_artifact_without_raw_error(self): self.assertEqual(self.invoke(provider=lambda *args,**kwargs:(403,b'NEVER-LOG-RESPONSE')),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'provider_upload');self.assertEqual(receipt['upload_http_status'],403) self.assertIn('artifact',receipt);self.assertNotIn('NEVER-LOG',self.receipt.read_text()) def test_download_failure_preserves_successful_created_artifact(self): def failure(url,method='GET',body=None,auth=None,headers=None): return (201,b'') if method=='PUT' else (401,b'RAW-SECRET') self.assertEqual(self.invoke(provider=failure),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'provider_download');self.assertEqual(receipt['upload_http_status'],201) self.assertEqual(receipt['readback_http_status'],401);self.assertIn('artifact',receipt) def test_quota_failure_has_safe_stage_and_no_secret(self): self.assertEqual(self.invoke(quota_failure=RuntimeError('PRIVATE-QUOTA-URL')),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'provider_quota') self.assertNotIn('PRIVATE-QUOTA-URL',self.receipt.read_text()) def test_missing_crypto_binary_is_identified_without_exception_text(self): with patch.object(finish.subprocess,'run',side_effect=FileNotFoundError('SECRET-ENV-PATH')): self.assertEqual(self.invoke(),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'derive_age_recipient') self.assertEqual(receipt['error_code'],'crypto_binary_unavailable') self.assertNotIn('SECRET-ENV',self.receipt.read_text()) def test_ciphertext_mismatch_never_decrypts_or_claims_backup(self): def bad_download(url,method='GET',body=None,auth=None,headers=None): return (201,b'') if method=='PUT' else (200,b'WRONG-CIPHERTEXT') self.assertEqual(self.invoke(provider=bad_download),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'provider_ciphertext_check') self.assertEqual(receipt['error_code'],'provider_ciphertext_mismatch') self.assertEqual(receipt['upload_http_status'],201) def test_decryption_failure_reports_fixed_code_after_matching_download(self): original=subprocess.run def protected_process(command,*args,**kwargs): if command[0]=='age' and command[1]=='-d': return subprocess.CompletedProcess(command,1,b'',b'RAW-KEY-MUST-NOT-LOG') return original(command,*args,**kwargs) with patch.object(finish.subprocess,'run',side_effect=protected_process): self.assertEqual(self.invoke(),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'decrypt_and_compare_history') self.assertEqual(receipt['error_code'],'crypto_command_failed') self.assertEqual(receipt['upload_http_status'],201) self.assertNotIn('RAW-KEY-MUST-NOT-LOG',self.receipt.read_text()) def test_current_incident_stops_before_any_provider_action(self): comparison={'historical_config_present':True,'historical_share_matches_current_lane':True, 'historical_credential_status':'current-lane-incident'} provider=unittest.mock.Mock() self.assertEqual(self.invoke(provider=provider,comparison=comparison),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'historical_share_comparison') self.assertEqual(receipt['error_code'],'current_backup_lane_incident') provider.assert_not_called() def test_independent_escrow_missing_does_not_upload(self): self.escrow.unlink() provider=unittest.mock.Mock() self.assertEqual(self.invoke(provider=provider),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'independent_key_escrow') self.assertEqual(receipt['error_code'],'independent_key_escrow_missing') provider.assert_not_called() def test_failed_attended_envelope_writes_safe_receipt(self): with patch.object(sys,'argv',self.args),patch.object(finish,'require_attended',side_effect=RuntimeError('TOKEN-DO-NOT-LOG')): self.assertEqual(finish.main(),1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'attended_envelope') self.assertEqual(receipt['error_code'],'attended_envelope_required') self.assertNotIn('TOKEN-DO-NOT-LOG',self.receipt.read_text()) def test_old_success_receipt_is_preserved(self): self.receipt.write_text('{"ok":true}') self.assertEqual(self.invoke(),1) self.assertEqual(self.receipt.read_text(),'{"ok":true}') def test_bootstrap_dependency_import_failure_still_writes_receipt(self): entry=self.directory/'root_custody_finish_entry.py' shutil.copy2(ROOT/'scripts/root_custody_finish_entry.py',entry) (self.directory/'root_custody_finish.py').write_text('raise RuntimeError("IMPORT-MUST-NOT-LOG")') result=subprocess.run(['/usr/bin/python3',str(entry),*self.args[1:]],capture_output=True) self.assertEqual(result.returncode,1) receipt=json.loads(self.receipt.read_text()) self.assertEqual(receipt['stage'],'helper_import');self.assertEqual(receipt['error_code'],'helper_startup_failed') self.assertEqual(result.stdout,b'');self.assertEqual(result.stderr,b'') self.assertNotIn('IMPORT-MUST-NOT-LOG',self.receipt.read_text()) if __name__=='__main__':unittest.main()