216 lines
12 KiB
Python
216 lines
12 KiB
Python
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_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('<tr>'),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']='<script>owned</script>'
|
|
result=assess(data)
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
report=Path(folder)/'assessment.html'
|
|
export_html(report,data,result)
|
|
self.assertNotIn('<script>',report.read_text())
|
|
self.assertIn('<script>',report.read_text())
|
|
artifact=next(Path(folder).glob('*.inventory-*.json'))
|
|
artifact.write_text('unrelated data')
|
|
with self.assertRaises(FileExistsError):
|
|
export_html(report,data,result)
|
|
self.assertEqual(artifact.read_text(),'unrelated data')
|