import copy import json from pathlib import Path import tempfile import unittest from samsara.analyzers.assessment import assess from samsara.analyzers.dependencies import CONFIRMED, POTENTIAL, STANDARD, UNKNOWN from samsara.analyzers.policy import load_policy from samsara.models.inventory import validate_inventory from samsara.reporting.export import export_html, export_json from samsara.collectors.importer import load_inventory def fixture(name): return json.loads(Path('tests/fixtures/scoring',name+'.json').read_text()) def scored(result): return [row for row in result['findings'] if row['points']>0] class ScoringRegressionTests(unittest.TestCase): def test_task_and_certificate_failures_reduce_coverage_without_risk(self): data = fixture('clean_windows') baseline = assess(data) for name, status in [('tasks', 'failed'), ('certificates', 'partial')]: data['checks'][name]['status'] = status data['checks'][name]['errors'] = [{'code': 'collection_failed', 'scope': name, 'message': 'Sanitized failure'}] result = assess(data) self.assertEqual(result['score'], baseline['score']) self.assertLess(result['coverage']['percent'], baseline['coverage']['percent']) self.assertEqual(result['confidence']['level'], 'moderate') def test_required_scenarios(self): expected = {'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} for name,score in expected.items(): with self.subTest(name=name): data=fixture(name) validate_inventory(data) before=copy.deepcopy(data) result=assess(data) self.assertEqual(result['score'],score) self.assertEqual(data,before,'Analyzer mutated the raw inventory') self.assertTrue(result['review_required']) self.assertIn('Historical usage validation',result['manual_verification']) self.assertTrue(all(f['evidence'] and f['scoring_reason'] for f in result['findings'])) self.assertEqual(result['score'],max((r['score'] for r in result['scoring']['breakdown']),default=0)) self.assertTrue(all(0<=f['points']<=100 for f in result['findings'])) def test_original_1565_point_incident_is_not_a_workload(self): data=fixture('incomplete_collection') inventory=data['inventory'] self.assertEqual(len(inventory['services']),86) self.assertEqual(len(inventory['roles']),14) self.assertEqual(sum(r['state']=='Established' for r in inventory['network']),19) self.assertEqual(sum(r['state']=='Listen' for r in inventory['network']),23) self.assertEqual(len(inventory['storage']),3) old_points=86*10+19*20+14*10+23*5+3*15+2*10+5 self.assertEqual(old_points,1565) result=assess(data) self.assertEqual(result['score'],0) self.assertFalse(scored(result)) self.assertLess(len(result['findings']),15) self.assertLess(result['coverage']['percent'],100) def test_standard_os_services_features_and_admin_shares_stay_visible(self): data=fixture('clean_windows') result=assess(data) standard=[r for r in result['findings'] if r['classification']==STANDARD] self.assertTrue(standard) self.assertTrue(all(r['points']==0 for r in standard)) for share in data['inventory']['storage']: share.update(observed_open_files=100,activity_source='Get-SmbOpenFile (share-root-correlated)') self.assertEqual(assess(data)['score'],0) # A hidden business share is not a default administrative share. data['inventory']['storage'].append({'type':'SMB Share','name':'Finance$','path':'D:\\Finance','special':True}) self.assertEqual(assess(data)['score'],20) def test_running_service_listener_and_connections_are_one_workload(self): data=fixture('active_sql_server') baseline=assess(data) self.assertEqual(len(scored(baseline)),1) self.assertEqual(scored(baseline)[0]['id'],'sql') self.assertEqual(scored(baseline)[0]['classification'],CONFIRMED) data['inventory']['network'] *= 20 data['inventory']['services'] *= 10 repeated=assess(data) self.assertEqual(repeated['score'],70) self.assertEqual(len(scored(repeated)),1) selected=repeated['scoring']['breakdown'][0] self.assertIn('observed_application_clients',[r['selected_rule'] for r in repeated['scoring']['breakdown']]) self.assertTrue(selected['selected_evidence_refs']) self.assertIn('Observed application TCP activity',scored(repeated)[0]['evidence']) def test_independent_workloads_use_maximum_not_sum(self): data=fixture('active_sql_server') other=fixture('active_file_server')['inventory']['storage'][-1] data['inventory']['storage'].append(other) result=assess(data) self.assertEqual(sorted(r['score'] for r in result['scoring']['breakdown']),[70,80]) self.assertEqual(result['score'],80) self.assertEqual(sum(r['determines_final_score'] for r in result['scoring']['breakdown']),1) def test_coverage_does_not_inflate_or_mask_known_risk(self): clean=assess(fixture('clean_windows')) incomplete=assess(fixture('incomplete_collection')) self.assertEqual(clean['score'],incomplete['score']) self.assertLess(incomplete['coverage']['percent'],clean['coverage']['percent']) self.assertEqual(clean['confidence']['level'],'high') self.assertEqual(incomplete['confidence']['level'],'moderate') data=fixture('active_sql_server') for check in data['checks'].values(): check['status']='failed' result=assess(data) self.assertEqual(result['score'],70,'Missing scope cannot erase actual captured positive evidence') self.assertEqual(result['confidence']['level'],'low') self.assertEqual(result['coverage']['percent'],0) self.assertTrue(all(f['points']==0 for f in result['findings'] if f['category']=='Missing evidence')) def test_browser_steam_loopback_outbound_and_unknown_are_not_inbound(self): self.assertFalse(any(r['id']=='game-server' for r in assess(fixture('inactive_game_server'))['findings'])) data=fixture('active_sql_server') # Remove active external connections, keep the running service/listener. row=data['inventory']['network'][-1] row['remote_address']='127.0.0.1' result=assess(data) self.assertEqual(result['score'],45) self.assertEqual(result['network_observations'][-1]['direction'],'Loopback / same host') self.assertFalse(any(f['classification']==CONFIRMED for f in scored(result))) row['remote_address']='::ffff:127.0.0.1' self.assertEqual(assess(data)['network_observations'][-1]['direction'],'Loopback / same host') row.update(remote_address='192.0.2.50',direction='Outbound') self.assertEqual(assess(data)['score'],45) row.update(pid=99999,direction='Unknown',local_port=1433) self.assertEqual(assess(data)['network_observations'][-1]['direction'],'Unknown') self.assertEqual(assess(data)['score'],45) row.update(local_port=55200,remote_port=443) self.assertEqual(assess(data)['network_observations'][-1]['direction'],'Outbound (inferred)') def test_port_alone_and_iis_http_sys_are_not_proof_of_a_hosted_app(self): data=fixture('clean_windows') data['inventory']['network'].append({'state':'Listen','local_address':'0.0.0.0','local_port':27015,'pid':12345,'process':'unknown','remote_address':'','remote_port':0,'direction':'Unknown'}) self.assertEqual(assess(data)['score'],0) data=fixture('active_iis_server') for row in data['inventory']['network'][-2:]: row.update(pid=4,process='System') result=assess(data) self.assertEqual(result['score'],35) self.assertFalse(any(f['classification']==CONFIRMED for f in scored(result))) def test_role_only_not_domain_controller_and_smb_transport_not_share_use(self): data=fixture('domain_controller');data['server']['product_type']=3 data['inventory']['services']=[r for r in data['inventory']['services'] if r['name']!='NTDS'] self.assertEqual(assess(data)['score'],15) data=fixture('active_file_server') share=data['inventory']['storage'][-1] share['activity_source']='Get-SmbOpenFile (administrative alias ambiguous)' self.assertEqual(assess(data)['score'],40) share.pop('observed_open_files') result=assess(data) self.assertEqual(result['score'],40) self.assertEqual(scored(result)[0]['classification'],POTENTIAL) self.assertIn('accessed share is unknown',scored(result)[0]['evidence']) def test_policy_validation_configuration_and_audit_snapshot(self): policy=load_policy() policy['scores']['observed_application_clients']=55 result=assess(fixture('active_sql_server'),policy=policy) self.assertEqual(result['score'],55) self.assertEqual(result['scoring']['policy']['scores']['observed_application_clients'],55) policy['scores']['observed_application_clients']=2 self.assertEqual(result['scoring']['policy']['scores']['observed_application_clients'],55) for source in [[],{},dict(load_policy(),model_version='1.0')]: with self.subTest(source=type(source)),self.assertRaises(ValueError): load_policy(source) bad=load_policy();bad['scores']['running_application_service']=-1 with self.assertRaises(ValueError): load_policy(bad) bad=load_policy();bad['thresholds']['medium']=90 with self.assertRaises(ValueError): load_policy(bad) def test_actionable_html_and_separate_complete_inventory(self): data=fixture('clean_windows') with tempfile.TemporaryDirectory() as folder: report=Path(folder)/'assessment.html' export_html(report,data,assess(data)) text=report.read_text() self.assertIn('How the score was calculated',text) self.assertIn('Evidence confidence',text) self.assertIn('Historical usage validation',text) self.assertNotIn('medium-risk dependencies',text) self.assertLess(text.count(''),60) inventory=list(Path(folder).glob('*.inventory-*.json')) self.assertEqual(len(inventory),1) raw=json.loads(inventory[0].read_text()) self.assertEqual(raw,data) self.assertEqual(len(raw['inventory']['services']),86) # Re-export reuses an identical artifact rather than losing inventory. export_html(report,data,assess(data)) export_json(Path(folder)/'assessment.json',data,assess(data)) self.assertEqual(load_inventory(inventory[0])['inventory'],load_inventory(Path(folder)/'assessment.json')['inventory']) def test_custom_jobs_not_misclassified_by_substrings_and_zero_points_maintenance(self): data=fixture('clean_windows') data['inventory']['tasks'].append({'name':'capture-report','path':'/etc/cron.d/custom','enabled':True,'triggers':'0 2 * * *'}) self.assertEqual(assess(data)['score'],20) data['inventory']['tasks'][-1]['enabled']=False self.assertEqual(assess(data)['score'],0) data['inventory']['tasks'][-1].update(name='Business export',path='\\Microsoft\\Windows\\PowerShell\\ScheduledJobs\\',enabled=True) self.assertEqual(assess(data)['score'],20) def test_html_escapes_actionable_evidence_and_artifacts_are_collision_safe(self): data=fixture('active_sql_server') data['inventory']['services'][-1]['account']='' result=assess(data) with tempfile.TemporaryDirectory() as folder: report=Path(folder)/'assessment.html' export_html(report,data,result) self.assertNotIn('