oceans2alaska/Samsara
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Server Buddy desktop app with local Linux assessment and reporting

Browse Source
This commit is contained in:
Not Dave
2026-10-30 09:46:01 -07:00
commit 365f97942a
29 changed files with 1789 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
tests/fixtures/synthetic_inventory.json vendored Normal file
View File

@@ -0,0 +1,139 @@
{
"schema_version": "1.0",
"synthetic": true,
"collected_at": "2026-10-01T12:00:00Z",
"server": {
"hostname": "demo-web-01.example.test",
"operating_system": "Windows Server 2022 Standard",
"os_build": "20348",
"last_boot": "2026-09-25T06:30:00Z",
"ip_addresses": [
"192.0.2.20"
],
"network_adapters": [
"Synthetic Ethernet adapter"
],
"domain": "example.test"
},
"checks": {
"roles": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"services": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"network": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"tasks": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"software": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"storage": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"certificates": {
"status": "complete",
"message": "Synthetic fixture; no remote collection performed."
},
"ad": {
"status": "not_collected",
"message": "Optional Active Directory enrichment is not implemented."
}
},
"inventory": {
"roles": [
{
"name": "Web-Server",
"state": "Installed"
},
{
"name": "File-Services",
"state": "Installed"
}
],
"services": [
{
"name": "W3SVC",
"display_name": "World Wide Web Publishing Service",
"state": "Running",
"startup": "Automatic",
"account": "LocalSystem",
"executable": "C:\\Windows\\System32\\svchost.exe"
}
],
"network": [
{
"state": "Listen",
"local_address": "0.0.0.0",
"local_port": 443,
"remote_address": "",
"remote_port": 0,
"pid": 4,
"process": "System",
"direction": "Unknown"
},
{
"state": "Established",
"local_address": "192.0.2.20",
"local_port": 443,
"remote_address": "192.0.2.45",
"remote_port": 52104,
"pid": 4,
"process": "System",
"direction": "Unknown"
}
],
"tasks": [
{
"name": "Nightly export",
"path": "\\Operations\\",
"enabled": true,
"account": "EXAMPLE\\svc_export",
"executable": "C:\\Operations\\export.exe",
"arguments": "[omitted: potentially sensitive]",
"triggers": "Daily 02:00"
}
],
"software": [
{
"name": "Example Reporting Service",
"publisher": "Example Software",
"version": "4.2",
"installation_date": "2024-06-12"
}
],
"storage": [
{
"type": "Volume",
"name": "C:",
"path": "C:\\",
"capacity_gib": 160,
"free_gib": 72
},
{
"type": "SMB Share",
"name": "Reports",
"path": "D:\\Reports",
"capacity_gib": null,
"free_gib": null
}
],
"certificates": [
{
"subject": "CN=demo-web-01.example.test",
"issuer": "CN=Example Test CA",
"expires": "2027-06-01T00:00:00Z",
"thumbprint": "0123456789ABCDEF0123456789ABCDEF01234567"
}
]
}
}

55
tests/test_core.py Normal file
View File

@@ -0,0 +1,55 @@
import json
from pathlib import Path
import tempfile
import unittest
from server_buddy.core.validation import validate_hostname
from server_buddy.collectors.demo import collect_demo, DEMO_HOST
from server_buddy.analyzers.assessment import assess
from server_buddy.reporting.export 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'] == 'Confirmed observed activity' 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('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)

49
tests/test_import.py Normal file
View File

@@ -0,0 +1,49 @@
import json
from pathlib import Path
import tempfile
import unittest
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
from server_buddy.analyzers.assessment import assess
try:
from server_buddy.collectors.importer import load_inventory
except ImportError:
load_inventory = None
@unittest.skipIf(load_inventory is None, 'jsonschema not installed')
class ImportTests(unittest.TestCase):
def load(self, data):
with tempfile.TemporaryDirectory() as folder:
path = Path(folder) / 'inventory.json'
path.write_text(json.dumps(data))
return load_inventory(path)
def test_linux_inventory_and_nfs_evidence(self):
data = self.load(collect_demo(LINUX_DEMO_HOST))
self.assertEqual(data['server']['platform'], 'linux')
result = assess(data)
self.assertTrue(any('/srv/reports' in item['evidence'] for item in result['findings']))
self.assertTrue(any('certificates:' in item['evidence'] for item in result['findings']))
def test_report_wrapper_recomputed_and_sensitive_fields_removed(self):
data = collect_demo(LINUX_DEMO_HOST)
data['synthetic'] = False
data['inventory']['services'][0]['arguments'] = '--password=SECRET'
data['checks']['services']['message'] = 'SECRET command diagnostics'
imported = self.load({'inventory_result': data, 'assessment': {'level': 'safe'}})
self.assertFalse(imported['synthetic'])
self.assertNotIn('SECRET', json.dumps(imported))
self.assertNotEqual(assess(imported)['level'], 'safe')
def test_rejects_missing_checks_and_malformed_rows(self):
for mutation in ['check', 'row', 'version']:
data = collect_demo(LINUX_DEMO_HOST)
if mutation == 'check':
del data['checks']['services']
elif mutation == 'row':
data['inventory']['services'] = [{}]
else:
data['schema_version'] = '99'
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
self.load(data)

89
tests/test_linux.py Normal file
View File

@@ -0,0 +1,89 @@
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from server_buddy.collectors.linux import (collect_local_linux, parse_connections,
parse_units, run_command, CollectionCancelled)
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
from server_buddy.analyzers.assessment import assess
from server_buddy.reporting.export import export_html
class LinuxTests(unittest.TestCase):
def test_service_states_and_startup(self):
rows = parse_units('nginx.service loaded active running Web server\nexample.service loaded inactive dead Example\n',
'nginx.service enabled enabled\nexample.service disabled disabled\n')
self.assertEqual(rows[0]['state'], 'Running')
self.assertEqual(rows[0]['startup'], 'enabled')
self.assertEqual(rows[1]['state'], 'inactive/dead')
def test_ipv4_ipv6_and_unprivileged_connections(self):
rows = parse_connections('LISTEN 0 128 0.0.0.0:22 0.0.0.0:*\nESTAB 0 0 [::1]:443 [::1]:51234 users:(("nginx",pid=123,fd=7))\n')
self.assertEqual(rows[0]['remote_port'], 0)
self.assertIsNone(rows[0]['pid'])
self.assertEqual(rows[1]['local_address'], '::1')
self.assertEqual(rows[1]['state'], 'Established')
self.assertEqual(rows[1]['process'], 'nginx')
self.assertEqual(rows[1]['direction'], 'Unknown')
@patch('server_buddy.collectors.linux.platform.system', return_value='Windows')
def test_wrong_execution_platform(self, system):
with self.assertRaises(ValueError):
collect_local_linux()
@patch('server_buddy.collectors.linux.platform.system', return_value='Linux')
def test_partial_failures_never_use_demo(self, system):
def unavailable(args, **kwargs):
raise RuntimeError('SECRET diagnostics must not be reported')
data = collect_local_linux(runner=unavailable)
self.assertFalse(data['synthetic'])
self.assertEqual(data['checks']['services']['status'], 'failed')
self.assertEqual(data['checks']['network']['status'], 'failed')
self.assertEqual(data['checks']['software']['status'], 'failed')
self.assertEqual(data['inventory']['services'], [])
self.assertNotIn('SECRET', json.dumps(data))
self.assertEqual(data['checks']['certificates']['status'], 'not_collected')
@patch('server_buddy.collectors.linux.platform.system', return_value='Linux')
def test_cancel(self, system):
with self.assertRaises(CollectionCancelled):
collect_local_linux(cancelled=lambda: True)
def test_mint_and_rhel_package_parsing(self):
for rpm in [False, True]:
def command(args, **kwargs):
if args[0] == 'ip':
return '[]'
if args[0] == 'rpm':
return 'nginx\t1.2-1\tExample vendor\n'
if args[0] == 'dpkg-query':
return 'installed\tnginx:amd64\t1.2\nconfig-files\tremoved\t1.0\n'
return ''
with self.subTest(rpm=rpm), patch('server_buddy.collectors.linux.platform.system', return_value='Linux'), patch('server_buddy.collectors.linux.Path.is_file', return_value=rpm):
data = collect_local_linux(runner=command)
self.assertEqual(len(data['inventory']['software']), 1)
self.assertIn('nginx', data['inventory']['software'][0]['name'])
self.assertEqual(data['checks']['software']['status'], 'partial')
def test_local_report_source_and_unclassified_services(self):
data = collect_demo(LINUX_DEMO_HOST)
data['synthetic'] = False
data['server']['collection_source'] = 'Local Linux read-only collection'
data['inventory']['services'] = [{'name': 'desktop-example.service', 'state': 'Running'}]
result = assess(data)
self.assertFalse(any(f['category'] == 'Active service' for f in result['findings']))
self.assertTrue(any(f['category'] == 'Unclassified services' for f in result['findings']))
with tempfile.TemporaryDirectory() as folder:
path = Path(folder) / 'report.html'
export_html(path, data, result)
self.assertIn('LOCAL LINUX INVENTORY', path.read_text())
self.assertNotIn('IMPORTED INVENTORY', path.read_text())
@unittest.skipUnless(__import__('sys').platform.startswith('linux'), 'Linux command test')
def test_command_timeout_and_cancellation(self):
with self.assertRaisesRegex(RuntimeError, 'timed out'):
run_command(['sleep', '1'], timeout=.05)
with self.assertRaises(CollectionCancelled):
run_command(['sleep', '1'], cancelled=lambda: True)

17
tests/test_startup.py Normal file
View File

@@ -0,0 +1,17 @@
import unittest
from unittest.mock import patch
from server_buddy.core.startup import desktop_error
class StartupTests(unittest.TestCase):
@patch('server_buddy.core.startup.sys.platform', 'linux')
@patch.dict('os.environ', {'DISPLAY': ':0', 'QT_QPA_PLATFORM': 'xcb'}, clear=True)
@patch('server_buddy.core.startup.ctypes.CDLL', side_effect=OSError)
def test_missing_xcb_returns_install_instruction(self, loader):
self.assertIn('sudo apt install libxcb-cursor0', desktop_error())
@patch('server_buddy.core.startup.sys.platform', 'linux')
@patch.dict('os.environ', {'QT_QPA_PLATFORM': 'offscreen'}, clear=True)
@patch('server_buddy.core.startup.ctypes.CDLL')
def test_offscreen_does_not_require_xcb(self, loader):
self.assertIsNone(desktop_error())
loader.assert_not_called()

87
tests/test_ui.py Normal file
View File

@@ -0,0 +1,87 @@
import os
os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
import time
import unittest
try:
from PySide6.QtWidgets import QApplication
from server_buddy.ui.window import MainWindow
except ImportError:
QApplication = None
@unittest.skipIf(QApplication is None, 'PySide6 not installed')
class UITests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = QApplication.instance() or QApplication([])
def test_navigation_collection_and_environment_errors(self):
window = MainWindow()
window.show()
self.assertEqual(window.tabs.count(), 10)
for index in range(6):
window.nav.setCurrentRow(index)
self.assertEqual(window.pages.currentIndex(), index)
window.hostname.setText('host;bad')
window.start_collection()
self.assertIn('Enter a hostname', window.status.text())
window.hostname.setText('demo-web-01.example.test')
window.environment.setCurrentIndex(1)
window.start_collection()
self.assertIn('not implemented', window.status.text())
window.environment.setCurrentIndex(0)
window.start_collection()
deadline = time.monotonic() + 5
while window.job is not None and time.monotonic() < deadline:
self.app.processEvents()
time.sleep(.01)
self.assertIsNone(window.job)
self.assertEqual(len(window.history), 2)
self.assertTrue(window.analyze_button.isEnabled())
self.assertIn('Complete', window.status.text())
window.close()
def test_linux_demo_collection(self):
window = MainWindow()
window.target.setCurrentIndex(1)
self.assertEqual(window.hostname.text(), 'demo-rhel-01.example.test')
window.start_collection()
deadline = time.monotonic() + 5
while window.job is not None and time.monotonic() < deadline:
self.app.processEvents()
time.sleep(.01)
self.assertIsNone(window.job)
self.assertEqual(window.data['server']['platform'], 'linux')
self.assertIn('SYNTHETIC', window.banner.text())
self.assertEqual(window.tabs.count(), 10)
window.close()
def test_local_collection_source_and_report(self):
from unittest.mock import patch
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
data = collect_demo(LINUX_DEMO_HOST)
data['synthetic'] = False
data['server']['collection_source'] = 'Local Linux read-only collection'
with patch('server_buddy.ui.window.platform.system', return_value='Linux'), patch('server_buddy.ui.window.collect_local_linux', return_value=data):
window = MainWindow()
# Local collection does not validate or contact the hostname field.
window.hostname.setText('not-a-remote-target;')
window.local_button.click()
self.assertFalse(window.analyze_button.isEnabled())
self.assertTrue(window.cancel_button.isEnabled())
deadline = time.monotonic() + 5
while window.job is not None and time.monotonic() < deadline:
self.app.processEvents()
time.sleep(.01)
self.assertIsNone(window.job)
self.assertIn('LOCAL LINUX', window.banner.text())
self.assertEqual(window.history[-1]['source'], 'Local Linux read-only collection')
self.assertTrue(window.local_button.isEnabled())
self.assertFalse(window.cancel_button.isEnabled())
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as folder:
report = Path(folder) / 'local.html'
with patch('server_buddy.ui.window.QFileDialog.getSaveFileName', return_value=(str(report), '')), patch('server_buddy.ui.window.QMessageBox.information'):
window.save_report('html')
self.assertIn('LOCAL LINUX INVENTORY', report.read_text())
window.close()