bzuccaro/Mark-VIII
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(deploy): reliability API routes, atomic releases, version single-source

Browse Source
This commit is contained in:
bzuccaro
2026-08-22 15:20:39 -06:00
parent aba08e36c7
commit 1bb15de258
10 changed files with 586 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_build_common.py
+101
View File
@@ -0,0 +1,101 @@
"""Tests for scripts/build_common.py version stamping helpers."""
import json
import os
import shutil
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
import build_common
class BuildCommonTest(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
self.payload = os.path.join(self.dir, 'user', 'remote_access',
'pager-webui')
os.makedirs(self.payload)
with open(os.path.join(self.dir, '_hak5_manifest.json'), 'w') as f:
f.write('{"payload": "pager-webui", "version": "1.3.2"}')
with open(os.path.join(self.payload, 'payload.sh'), 'w') as f:
f.write('#!/bin/bash\n'
'# Title: Mark VIII\n'
'# Description: test payload\n'
'# Version: 1.3.2\n'
'# Category: Remote-Access\n'
'\n'
'echo hi\n')
with open(os.path.join(self.payload, 'server.py'), 'w') as f:
f.write('"""Mark VIII server."""\n'
'import os\n'
'\n'
'PORT = 8080\n')
def tearDown(self):
shutil.rmtree(self.dir, ignore_errors=True)
def _server_text(self):
with open(os.path.join(self.payload, 'server.py')) as f:
return f.read()
def test_stamp_version_updates_all_three_files(self):
stamped = sorted(build_common.stamp_version(self.dir, '1.4.0'))
expected = sorted([
os.path.join(self.dir, '_hak5_manifest.json'),
os.path.join(self.payload, 'payload.sh'),
os.path.join(self.payload, 'server.py'),
])
self.assertEqual(stamped, expected)
with open(os.path.join(self.dir, '_hak5_manifest.json')) as f:
self.assertEqual(json.load(f)['version'], '1.4.0')
with open(os.path.join(self.payload, 'payload.sh')) as f:
sh_text = f.read()
self.assertIn('# Version: 1.4.0', sh_text)
self.assertNotIn('# Version: 1.3.2', sh_text)
server_text = self._server_text()
self.assertIn("SERVER_VERSION = '1.4.0'", server_text)
self.assertEqual(server_text.count('SERVER_VERSION'), 1)
self.assertLess(server_text.index("SERVER_VERSION = '1.4.0'"),
server_text.index('\nimport os'))
def test_stamp_version_is_idempotent_and_upgrades(self):
build_common.stamp_version(self.dir, '1.4.0')
stamped = build_common.stamp_version(self.dir, '1.5.0')
self.assertEqual(len(stamped), 3)
server_text = self._server_text()
self.assertEqual(server_text.count('SERVER_VERSION'), 1)
self.assertIn("SERVER_VERSION = '1.5.0'", server_text)
with open(os.path.join(self.payload, 'payload.sh')) as f:
self.assertIn('# Version: 1.5.0', f.read())
def test_server_without_docstring_gets_top_injection(self):
path = os.path.join(self.payload, 'server.py')
with open(path, 'w') as f:
f.write('# comment header\n'
'\n'
'import os\n'
'PORT = 8080\n')
build_common.stamp_version(self.dir, '9.9.9')
text = self._server_text()
lines = text.splitlines(True)
idx = [i for i, l in enumerate(lines) if l.startswith('SERVER_VERSION')]
self.assertEqual(len(idx), 1)
self.assertLess(idx[0], [i for i, l in enumerate(lines)
if l.startswith('import os')][0])
def test_missing_files_are_tolerated(self):
empty = tempfile.mkdtemp()
try:
self.assertEqual(build_common.stamp_version(empty, '1.4.0'), [])
finally:
shutil.rmtree(empty, ignore_errors=True)
def test_invalid_version_is_rejected(self):
for bad in ("1.4'; import os", '', 'a' * 64, 'ver x'):
with self.assertRaises(ValueError):
build_common.stamp_version(self.dir, bad)
if __name__ == '__main__':
unittest.main()
tests/test_mk8_profiles.py
+23
View File
@@ -57,5 +57,28 @@ class ProfilesTest(unittest.TestCase):
self.state = {}
self.assertFalse(mk8_profiles.snapshot('empty'))
def test_path_rejects_traversal_and_bad_names(self):
for bad in ('../x', '..', 'a/b', '', 'a' * 65, './x', 'x/..',
'a b', 'a;b', None):
with self.assertRaises(ValueError):
mk8_profiles._path(bad)
def test_path_accepts_safe_names(self):
for good in ('p', 'pre-client_connect-123', 'lastknown-good',
'A.b-c_d', 'x' * 64, '0'):
path = mk8_profiles._path(good)
self.assertEqual(path, os.path.join(mk8_profiles.PROFILES_DIR,
good))
def test_snapshot_rejects_bad_name_without_side_effects(self):
with self.assertRaises(ValueError):
mk8_profiles.snapshot('../evil')
self.assertEqual(mk8_profiles.list_profiles(), [])
def test_list_profiles_skips_invalid_dirnames(self):
mk8_profiles.snapshot('good')
os.mkdir(os.path.join(mk8_profiles.PROFILES_DIR, 'bad name'))
self.assertEqual(mk8_profiles.list_profiles(), ['good'])
if __name__ == '__main__':
unittest.main()
tests/test_reliability_api.py
+129
View File
@@ -11,6 +11,11 @@ def setUpModule():
__import__('importlib').reload(server)
class _Ctx(object):
def __init__(self, body=None):
self.body = body
class ReliabilityApiTest(unittest.TestCase):
def setUp(self):
self.runs = []
@@ -58,6 +63,130 @@ class ReliabilityApiTest(unittest.TestCase):
os.unlink(marker)
server.BOOT_MARKER = old
def test_profile_routes_registered(self):
handler, _ = server.ROUTER.dispatch('GET',
'/api/reliability/profiles')
self.assertEqual(handler, server.h_profiles_get)
handler, _ = server.ROUTER.dispatch('POST', '/api/reliability/profile')
self.assertEqual(handler, server.h_profile_save)
handler, _ = server.ROUTER.dispatch('POST', '/api/reliability/restore')
self.assertEqual(handler, server.h_profile_restore)
handler, _ = server.ROUTER.dispatch('GET', '/api/reliability/nope')
self.assertIsNone(handler)
def test_h_profiles_get_lists_profiles(self):
import mk8_profiles
old = mk8_profiles.list_profiles
mk8_profiles.list_profiles = lambda: ['a', 'b']
try:
status, data = server.h_profiles_get(None)
finally:
mk8_profiles.list_profiles = old
self.assertEqual(status, 200)
self.assertEqual(data, {'profiles': ['a', 'b']})
def test_h_profile_save_validates_saves_and_journals(self):
import mk8_events
import mk8_profiles
calls = {'snapshots': []}
events = []
old_snapshot, old_log = mk8_profiles.snapshot, mk8_events.log_event
def fake_snapshot(name):
if not all(c.isalnum() or c in '._-' for c in name) \
or name in ('.', '..') or len(name) > 64:
raise ValueError('invalid profile name')
calls['snapshots'].append(name)
return True
def fake_log(kind, **kw):
events.append((kind, kw))
mk8_profiles.snapshot = fake_snapshot
mk8_events.log_event = fake_log
try:
status, data = server.h_profile_save(_Ctx({'name': ' pre-x-1 '}))
self.assertEqual(status, 200)
self.assertEqual(data, {'ok': True})
self.assertEqual(calls['snapshots'], ['pre-x-1'])
self.assertEqual(events[-1][0], 'profile_save')
status, data = server.h_profile_save(_Ctx({'name': ' '}))
self.assertEqual(status, 400)
self.assertIn('error', data)
status, data = server.h_profile_save(_Ctx({'name': '../evil'}))
self.assertEqual(status, 400)
self.assertIn('error', data)
self.assertEqual(calls['snapshots'], ['pre-x-1'])
status, data = server.h_profile_save(_Ctx({}))
self.assertEqual(status, 400)
def failed_snapshot(name):
calls['snapshots'].append(name)
return False
mk8_profiles.snapshot = failed_snapshot
status, data = server.h_profile_save(_Ctx({'name': 'p2'}))
self.assertEqual(status, 502)
self.assertFalse(data['ok'])
self.assertEqual(events[-1][0], 'profile_save')
self.assertEqual(events[-1][1].get('sev'), 'warn')
finally:
mk8_profiles.snapshot = old_snapshot
mk8_events.log_event = old_log
def test_h_profile_restore_gated_reload_journal(self):
import mk8_gate
import mk8_events
import mk8_profiles
calls = {'gate': [], 'events': []}
olds = (mk8_gate.enter, mk8_profiles.restore, mk8_events.log_event)
def fake_enter(op):
calls['gate'].append(op)
return 'snap-1'
def fake_restore(name):
calls['restored'] = name
return {'ok': True, 'restored': ['wireless']}
def fake_log(kind, **kw):
calls['events'].append((kind, kw))
mk8_gate.enter = fake_enter
mk8_profiles.restore = fake_restore
mk8_events.log_event = fake_log
try:
status, result = server.h_profile_restore(_Ctx({'name': 'p1'}))
self.assertEqual(status, 200)
self.assertEqual(result, {'ok': True, 'restored': ['wireless']})
self.assertEqual(calls['gate'], ['restore_profile'])
self.assertEqual(calls['restored'], 'p1')
reloads = [r for r in self.runs if r[0][:2] == ['wifi', 'reload']]
self.assertEqual(len(reloads), 1)
self.assertEqual(calls['events'][-1][0], 'profile_restore')
status, result = server.h_profile_restore(
_Ctx({'name': 'missing'}))
def missing_restore(name):
calls['restored'] = name
return {'ok': False, 'restored': [], 'error': 'not found'}
mk8_profiles.restore = missing_restore
status, result = server.h_profile_restore(
_Ctx({'name': 'missing'}))
self.assertEqual(status, 502)
self.assertFalse(result['ok'])
self.assertEqual(calls['events'][-1][0], 'profile_restore')
self.assertEqual(calls['events'][-1][1].get('sev'), 'warn')
status, result = server.h_profile_restore(_Ctx({'name': ''}))
self.assertEqual(status, 400)
finally:
(mk8_gate.enter, mk8_profiles.restore,
mk8_events.log_event) = olds
if __name__ == '__main__':
unittest.main()