"""Synthetic disclosure regressions. No real inventory or patient data is used.""" import copy import json import os from pathlib import Path import stat import sys import tempfile import unittest from unittest.mock import patch from samsara.core.privacy import minimize_inventory, load_privacy_policy, canonical_share_root from samsara.core.storage import write_private, require_local_path from samsara.core.process import capture, ProcessFailure, CollectionCancelled from samsara.models.inventory import validate_inventory from samsara.reporting.export import export_html, export_json from samsara.analyzers.assessment import assess CANARY='SYNTHETIC_PERSON_CANARY_9f4' def strict_policy(): policy=load_privacy_policy() policy['retain']={key:False for key in policy['retain']} return policy def fixture(name='active_sql_server'): return json.loads(Path('tests/fixtures/scoring',name+'.json').read_text()) class PrivacyTests(unittest.TestCase): def test_all_free_text_and_stale_findings_are_removed(self): raw=fixture();original=copy.deepcopy(raw) raw['server'].update(hostname=CANARY,domain=CANARY,network_adapters=[CANARY]) raw['inventory']['services'][-1].update(name='MSSQL$'+CANARY,account=CANARY,display_name=CANARY) raw['inventory']['tasks']=[{'name':CANARY,'path':'\\'+CANARY+'\\','enabled':True,'account':CANARY,'triggers':CANARY,'arguments':CANARY}] raw['inventory']['software']=[{'name':'Microsoft SQL Server '+CANARY,'publisher':CANARY,'version':'16.0.1 '+CANARY}] raw['inventory']['storage']=[{'type':'SMB Share','name':CANARY,'path':'C:\\'+CANARY}] raw['inventory']['certificates']=[{'subject':CANARY,'issuer':CANARY,'thumbprint':CANARY,'expires':'2027-01-01T00:00:00Z'}] raw['checks']['services'].update(message=CANARY,errors=[{'code':CANARY,'scope':CANARY,'message':CANARY}]) raw['checks'][CANARY]={'status':'failed','message':CANARY} raw['privacy']={'profile':'minimized-metadata-v1'} # A marker cannot bypass projection. snapshot=copy.deepcopy(raw) clean=minimize_inventory(raw,strict_policy()) self.assertEqual(raw,snapshot) self.assertNotIn(CANARY,json.dumps(clean)) validate_inventory(clean) self.assertEqual(assess(clean)['score'],70) with tempfile.TemporaryDirectory() as folder: policy=strict_policy();policy['report_directory']=folder with patch('samsara.core.storage._verify_windows_acl'), patch('samsara.core.storage.load_privacy_policy',return_value=policy), patch('samsara.core.privacy.load_privacy_policy',return_value=policy): assessment=assess(original) assessment['findings'][0]['evidence']=CANARY export_html(Path(folder)/'test.html',raw,assessment) export_json(Path(folder)/'test.json',raw,assessment) for path in Path(folder).iterdir(): self.assertNotIn(CANARY,path.read_text()) self.assertEqual(len(list(Path(folder).iterdir())),2) def test_exact_roots_only_no_descendant_or_network_approval(self): policy=strict_policy();policy['approved_share_roots']=['D:\\Approved'] raw=fixture();raw['inventory']['storage']=[{'type':'SMB Share','name':'private','path':p} for p in ['D:\\Approved','D:\\Approved\\child','D:\\Other']] clean=minimize_inventory(raw,policy) self.assertEqual(clean['inventory']['storage'][0]['path'],'D:\\Approved') self.assertNotIn('child',json.dumps(clean)) for bad in ['\\\\server\\share','//server/share','D:\\Approved\\..\\Other']: with self.assertRaises(ValueError): canonical_share_root(bad) def test_workload_scores_survive_minimization(self): for name,score in [('clean_windows',0),('inactive_game_server',0),('active_sql_server',70),('active_iis_server',70),('domain_controller',95),('active_file_server',80),('incomplete_collection',0)]: with self.subTest(name=name): clean=minimize_inventory(fixture(name)) self.assertEqual(assess(clean)['score'],score) self.assertTrue(assess(clean)['review_required']) self.assertEqual(minimize_inventory(clean),clean) def test_dates_closed_schema_and_oversized_input_fail(self): raw=fixture();raw['collected_at']='not a date' with self.assertRaises(ValueError):validate_inventory(raw) raw=fixture();raw['inventory']['services'][0]['unexpected']={'sensitive':CANARY} with self.assertRaises(ValueError):validate_inventory(raw) raw=fixture();raw['inventory']['tasks']=[{}]*20001 with self.assertRaises(ValueError):minimize_inventory(raw) def test_default_collectors_do_not_request_high_risk_scopes(self): for name,forbidden in [('Get-ServerStorage.ps1','Get-SmbOpenFile'),('Get-ServerCertificates.ps1','Get-ChildItem'),('Get-ServerTasks.ps1','Get-ScheduledTask'),('Get-ServerServices.ps1','StartName')]: self.assertNotIn(forbidden,Path('powershell',name).read_text()) self.assertNotIn('TemporaryFile',Path('samsara/collectors/windows.py').read_text()) self.assertNotIn('TemporaryFile',Path('samsara/collectors/linux.py').read_text()) class StorageTests(unittest.TestCase): @unittest.skipIf(os.name=='nt','POSIX permission test; native Windows ACL acceptance requires a provisioned lab root.') def test_private_root_exclusive_files_and_destination_boundary(self): with tempfile.TemporaryDirectory() as base: root=Path(base)/'reports';policy=load_privacy_policy();policy['report_directory']=str(root) with patch('samsara.core.storage.load_privacy_policy',return_value=policy): output=root/'report.json';write_private(output,'{}') if os.name!='nt': self.assertEqual(stat.S_IMODE(root.stat().st_mode),0o700) self.assertEqual(stat.S_IMODE(output.stat().st_mode),0o600) with self.assertRaises(FileExistsError):write_private(output,'overwrite') with self.assertRaises(ValueError):write_private(Path(base)/'outside.json','{}') self.assertEqual(output.read_text(),'{}') if os.name!='nt': root.chmod(0o755) with self.assertRaises(ValueError):write_private(root/'unsafe.json','{}') def test_network_uri_symlink_paths_rejected(self): for value in ['\\\\server\\share\\data.json','//server/share/data.json','https://host/path']: with self.assertRaises(ValueError):require_local_path(value) if os.name!='nt': with tempfile.TemporaryDirectory() as base: link=Path(base)/'link';link.symlink_to(Path(base),target_is_directory=True) with self.assertRaises(ValueError):require_local_path(link/'data.json') @unittest.skipIf(os.name=='nt','POSIX FIFO regression') def test_import_rejects_fifo_without_waiting_for_a_writer(self): from samsara.collectors.importer import load_inventory with tempfile.TemporaryDirectory() as base: pipe=Path(base)/'inventory.json' os.mkfifo(pipe) with self.assertRaises(ValueError): load_inventory(pipe) class CaptureTests(unittest.TestCase): def test_bounded_memory_capture_timeout_cancel_and_sanitized_errors(self): env={'PATH':os.defpath, **{key:os.environ[key] for key in ('SystemRoot','WINDIR') if key in os.environ}} self.assertEqual(capture([sys.executable,'-c','print("synthetic")'],env=env,timeout=3),b'synthetic\n') for code,script,options in [('output_limit','print("x"*200000)',{'limit':100}),('timeout','import time; time.sleep(10)',{'timeout':.05}),('collection_failed','raise RuntimeError("'+CANARY+'")',{})]: with self.subTest(code=code), self.assertRaises(ProcessFailure) as raised: capture([sys.executable,'-c',script],env=env,**dict({'timeout':3},**options)) self.assertEqual(raised.exception.code,code) self.assertNotIn(CANARY,str(raised.exception)) with self.assertRaises(CollectionCancelled): capture([sys.executable,'-c','print(1)'],env=env,timeout=3,cancelled=lambda:True) class RetainedNameTests(unittest.TestCase): def test_useful_names_and_roots_are_kept_but_identifier_like_values_withheld(self): raw=fixture('active_file_server') raw['server']['hostname']='fileserver01' raw['inventory']['software']=[{'name':'Epic Hyperspace Client 2024','version':'1.2.3'},{'name':'Scanner 123456789','version':'1'}] raw['inventory']['services'][-1]['name']='CustomImagingService' raw['inventory']['storage']=[ {'type':'SMB Share','name':'RadiologyArchive','path':'D:\\Shares\\Radiology'}, {'type':'SMB Share','name':'Dept','path':'D:\\Shares\\a\\b\\c\\d.pdf'}, {'type':'Remote Mount','name':'nas01:/export/imaging','path':'/mnt/imaging'}, {'type':'SMB Share','name':'jane.doe@hospital.org','path':'D:\\Shares\\123-45-6789'}] clean=minimize_inventory(raw) text=json.dumps(clean) self.assertEqual(clean['server']['hostname'],'fileserver01') names=[r['name'] for r in clean['inventory']['software']] self.assertIn('Epic Hyperspace Client 2024',names); self.assertNotIn('Scanner 123456789',names) self.assertIn('CustomImagingService',[r['name'] for r in clean['inventory']['services']]) storage=clean['inventory']['storage'] self.assertEqual(storage[0]['name'],'RadiologyArchive'); self.assertEqual(storage[0]['path'],'D:\\Shares\\Radiology') self.assertEqual(storage[2]['name'],'nas01:/export/imaging'); self.assertEqual(storage[2]['path'],'/mnt/imaging') for leaked in ('d.pdf','jane.doe','123-45-6789','123456789'): self.assertNotIn(leaked,text) self.assertEqual(minimize_inventory(clean),clean) validate_inventory(clean) def test_every_category_can_be_switched_off(self): raw=fixture('active_file_server'); raw['server']['hostname']='fileserver01' raw['inventory']['software']=[{'name':'Epic Hyperspace Client','version':'1'}] raw['inventory']['storage']=[{'type':'SMB Share','name':'RadiologyArchive','path':'D:\\Shares\\Radiology'}] text=json.dumps(minimize_inventory(raw,strict_policy())) for value in ('fileserver01','Epic','RadiologyArchive','Radiology'): self.assertNotIn(value,text) def test_policy_retain_block_is_validated(self): with tempfile.TemporaryDirectory() as folder: bad=load_privacy_policy(); bad['retain']['hostname']='yes' path=Path(folder)/'policy.json'; path.write_text(json.dumps(bad)) with self.assertRaises(ValueError): load_privacy_policy(path)