oceans2alaska
610a3bb447
- Add privacy projection (core/privacy.py), private report storage, bounded
process capture; omit identities, free text and unapproved paths
- Mask remote peer IPs: only loopback/host/wildcard stay numeric, others are
reduced to remote_scope private/public (schema + scoring updated)
- Map ss TCP states to canonical names; ASCII-only numeric fields; fix
"Microsoft Visual C++" product matching (Python and PowerShell)
- Merge 7fed3b9: -Diagnostic switch, serialization exit code 3, PS7 module
path, UI error detail; keep hardened task/certificate collectors
- Add audit docs, privacy docs and regression tests
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from samsara.core.validation import validate_hostname
|
|
from samsara.collectors.demo import collect_demo, DEMO_HOST
|
|
from samsara.analyzers.assessment import assess
|
|
from report_helpers import export_html, export_json
|
|
|
|
class CoreTests(unittest.TestCase):
|
|
def test_hostname(self):
|
|
self.assertEqual(validate_hostname(' DEMO-WEB-01.example.test. '), DEMO_HOST)
|
|
for value in ['', '-host', 'a..b', 'https://host', 'host;whoami', 'host$(id)', 'a'*64]:
|
|
with self.subTest(value=value), self.assertRaises(ValueError):
|
|
validate_hostname(value)
|
|
|
|
def test_demo_rejects_real_hosts(self):
|
|
with self.assertRaises(ValueError):
|
|
collect_demo('production.example.test')
|
|
data = collect_demo(DEMO_HOST)
|
|
data['server']['hostname'] = 'changed'
|
|
self.assertEqual(collect_demo(DEMO_HOST)['server']['hostname'], DEMO_HOST)
|
|
|
|
def test_evidence_and_missing_activity(self):
|
|
data = collect_demo(DEMO_HOST)
|
|
result = assess(data)
|
|
self.assertTrue(all(f['evidence'] for f in result['findings']))
|
|
self.assertTrue(any(f['classification'] == 'Potential workload dependency' for f in result['findings']))
|
|
self.assertFalse(any(f['classification'] == 'Confirmed workload dependency' for f in result['findings']))
|
|
data['inventory']['network'] = []
|
|
self.assertTrue(any('does not prove' in f['evidence'] for f in assess(data)['findings']))
|
|
self.assertTrue(any('ad:' in f['evidence'] for f in result['findings']))
|
|
|
|
def test_export_and_escape(self):
|
|
data = collect_demo(DEMO_HOST)
|
|
data['server']['hostname'] = '<script>alert(1)</script>'
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
report = Path(directory) / 'report.html'
|
|
export_html(report, data, assess(data))
|
|
text = report.read_text()
|
|
self.assertNotIn('<script>', text)
|
|
self.assertIn('SYNTHETIC', text)
|
|
self.assertIn('<title>Samsara</title>', text)
|
|
self.assertIn('Samsara assessment', text)
|
|
self.assertIn('missing evidence', text)
|
|
self.assertIn('manual verification', text)
|
|
output = Path(directory) / 'report.json'
|
|
export_json(output, data, assess(data))
|
|
self.assertTrue(json.loads(output.read_text())['inventory_result']['synthetic'])
|
|
|
|
def test_schema(self):
|
|
try:
|
|
import jsonschema
|
|
except ImportError:
|
|
self.skipTest('jsonschema not installed')
|
|
schema = json.loads(Path('schemas/inventory-v1.schema.json').read_text())
|
|
for data in [collect_demo(DEMO_HOST), json.loads(Path('tests/fixtures/synthetic_inventory.json').read_text())]:
|
|
jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker()).validate(data)
|