"""Independent synthetic scanner controls; no real private-data collection.""" import ast import hashlib import io import json import shlex import tempfile import unittest import zipfile from pathlib import Path HERE=Path(__file__).resolve().parent ROOT=HERE.parents[2] SCANNER=ROOT/'release/comparison-v20/check-public-privacy.py' TREE=ast.parse(SCANNER.read_text()) FUNCTIONS=ast.Module(body=[n for n in TREE.body if isinstance(n,ast.FunctionDef) and n.name in {'add_text','text_parts','add_private_file','process_keys','check'}],type_ignores=[]) INGEST=ast.Module(body=[n for n in TREE.body if isinstance(n,ast.For) and isinstance(n.target,ast.Name) and n.target.id=='p' and 'private-unplanned-input' in ast.unparse(n.iter)],type_ignores=[]) SYNTHETIC_SHORT='Q7' assert hashlib.sha256(SYNTHETIC_SHORT.encode()).hexdigest()!='f056b8cae37dc3ed60ee52da9a2951f8a6c531fe3222a3ad6835dbdd905c7d2a' def namespace(): ns={'Path':Path,'json':json,'io':io,'zipfile':zipfile,'shlex':shlex, 'sha':lambda b:hashlib.sha256(b).hexdigest(),'load':lambda p:json.loads(Path(p).read_text()), 'needles':set(),'private_hashes':set(),'draft_values':set(),'checked':[],'failures':[]} exec(compile(FUNCTIONS,'independent-synthetic-scanner-functions','exec'),ns) return ns def ingest(ns,root,value): folder=root/'private-unplanned-input';folder.mkdir(exist_ok=True) path=folder/'synthetic.conversation.json' path.write_text(json.dumps({'clientInput':value,'messages':[]})) ns['OLD']=root exec(compile(INGEST,'independent-synthetic-short-ingestion','exec'),ns) def blocked(ns,label,raw): ns['failures'].clear();ns['check'](label,raw) return bool(ns['failures']) def archive(name,raw): stream=io.BytesIO() with zipfile.ZipFile(stream,'w') as z:z.writestr(name,raw) return stream.getvalue() class IndependentScannerControls(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name);self.ns=namespace() ingest(self.ns,self.root,SYNTHETIC_SHORT) def tearDown(self):self.temp.cleanup() def test_short_ordinary_substring_is_not_a_private_evidence_match(self): self.assertFalse(blocked(self.ns,'ordinary.txt',('Synthetic ordinary before '+SYNTHETIC_SHORT+' after').encode())) def test_exact_short_client_input_in_json_is_blocked(self): self.assertTrue(blocked(self.ns,'case.json',json.dumps({'clientInput':SYNTHETIC_SHORT,'synthetic':True}).encode())) def test_nested_exact_client_input_is_blocked(self): self.assertTrue(blocked(self.ns,'case.json',json.dumps({'nested':[{'clientInput':SYNTHETIC_SHORT}]}).encode())) def test_unicode_escaped_short_field_is_blocked_after_json_decoding(self): raw=('{'+'"clientInput":"'+''.join('\\u%04x'%ord(c) for c in SYNTHETIC_SHORT)+'"}').encode() self.assertTrue(blocked(self.ns,'escaped.json',raw)) def test_exact_short_client_input_jsonl_record_is_blocked(self): raw=(json.dumps({'synthetic':True})+'\n'+json.dumps({'clientInput':SYNTHETIC_SHORT})+'\n').encode() self.assertTrue(blocked(self.ns,'records.jsonl',raw)) def test_short_jsonl_record_inside_nested_archives_is_blocked(self): raw=(json.dumps({'nested':{'clientInput':SYNTHETIC_SHORT}})+'\n').encode() self.assertTrue(blocked(self.ns,'outer.zip',archive('inner.zip',archive('records.jsonl',raw)))) def test_malformed_jsonl_line_does_not_hide_later_exact_private_field(self): raw=b'not-json\n'+json.dumps({'clientInput':SYNTHETIC_SHORT}).encode()+b'\n' self.assertTrue(blocked(self.ns,'records.jsonl',raw)) def test_case_variant_jsonl_extension_keeps_exact_field_protection(self): raw=json.dumps({'clientInput':SYNTHETIC_SHORT}).encode()+b'\n' self.assertTrue(blocked(self.ns,'records.JSONL',raw)) def test_short_json_inside_archive_is_blocked(self): self.assertTrue(blocked(self.ns,'outer.zip',archive('case.json',json.dumps({'clientInput':SYNTHETIC_SHORT}).encode()))) def test_four_character_draft_keeps_arbitrary_substring_protection(self): ingest(self.ns,self.root,'M4z8') self.assertTrue(blocked(self.ns,'ordinary.txt',b'Synthetic prefix M4z8 suffix')) def test_whitespace_short_draft_keeps_exact_untrimmed_field_protection(self): value=' '+SYNTHETIC_SHORT+' ';ingest(self.ns,self.root,value) self.assertTrue(blocked(self.ns,'case.json',json.dumps({'clientInput':value}).encode())) def test_whole_private_original_hash_remains_blocked(self): raw=(self.root/'private-unplanned-input/synthetic.conversation.json').read_bytes() self.assertTrue(blocked(self.ns,'renamed.txt',raw)) def test_private_filename_is_blocked_without_content_needle(self): self.assertTrue(blocked(self.ns,'archive.zip!private-ui/synthetic.json',b'{}')) def test_process_api_credential_remains_substring_protected(self): token='SYNTHETIC_AUTH_TOKEN_'+('x'*32) self.ns['process_keys']({'argv':['owned-synthetic-server','--api-key',token]}) self.assertTrue(blocked(self.ns,'case.txt',('synthetic '+token).encode())) def test_equals_style_api_key_inside_nested_archive_is_blocked(self): token='SYNTHETIC_EQ_TOKEN_'+('z'*32) self.ns['process_keys']({'command':'owned-synthetic-server --api-key='+token}) self.assertTrue(blocked(self.ns,'outer.zip',archive('inner.zip',archive('case.txt',token.encode())))) def test_private_raw_ui_json_wrapping_remains_blocked(self): value='Synthetic private UI only: "draft"\n'+('long unrelated observation '*8) path=self.root/'synthetic.private.txt';path.write_text(json.dumps({'snapshot':value})) self.ns['add_private_file'](path) self.assertTrue(blocked(self.ns,'wrapped.json',json.dumps({'differentField':value}).encode())) def test_private_whole_binary_file_inside_archive_remains_blocked(self): raw=b'\x00\x01\x02SYNTHETIC-PRIVATE-BINARY\xff';path=self.root/'synthetic.bin';path.write_bytes(raw) self.ns['add_private_file'](path) self.assertTrue(blocked(self.ns,'outer.zip',archive('renamed.bin',raw))) def test_unsafe_nested_archive_path_remains_blocked(self): self.assertTrue(blocked(self.ns,'outer.zip',archive('../escape.txt',b'synthetic'))) if __name__=='__main__':unittest.main(verbosity=2)