"""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 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) 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=load_privacy_policy();policy['report_directory']=folder with patch('samsara.core.storage._verify_windows_acl'), patch('samsara.core.storage.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=load_privacy_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)