From 1bb15de2583d8a9631463c6a2fc2bbe0ed85bae4 Mon Sep 17 00:00:00 2001 From: c4ch3c4d3 Date: Sat, 22 Aug 2026 15:20:39 -0600 Subject: [PATCH] feat(deploy): reliability API routes, atomic releases, version single-source --- VERSION | 1 + .../remote_access/pager-webui/mk8_profiles.py | 19 ++- .../user/remote_access/pager-webui/server.py | 57 ++++++++ .../remote_access/pager-webui/www/index.html | 2 +- .../remote_access/pager-webui/www/js/views.js | 37 +++++ scripts/build_common.py | 124 ++++++++++++++++ scripts/deploy.sh | 133 +++++++++++++----- tests/test_build_common.py | 101 +++++++++++++ tests/test_mk8_profiles.py | 23 +++ tests/test_reliability_api.py | 129 +++++++++++++++++ 10 files changed, 586 insertions(+), 40 deletions(-) create mode 100644 VERSION create mode 100644 scripts/build_common.py create mode 100644 tests/test_build_common.py diff --git a/VERSION b/VERSION new file mode 100644 index 0000000..88c5fb8 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +1.4.0 diff --git a/payload/user/remote_access/pager-webui/mk8_profiles.py b/payload/user/remote_access/pager-webui/mk8_profiles.py index bc4abfa..b360f70 100644 --- a/payload/user/remote_access/pager-webui/mk8_profiles.py +++ b/payload/user/remote_access/pager-webui/mk8_profiles.py @@ -1,8 +1,9 @@ """UCI profile snapshots under /mmc/mk8/profiles//{pineapd,wireless,network}""" -import os, time +import os, re, time PROFILES_DIR = '/mmc/mk8/profiles' CONFIGS = ('pineapd', 'wireless', 'network') +NAME_RE = re.compile(r'^[A-Za-z0-9._-]{1,64}$') def run_cmd(args, timeout=20, input_data=None): @@ -12,6 +13,12 @@ def run_cmd(args, timeout=20, input_data=None): def _path(name): + """Resolve a profile name to its directory. HTTP-supplied names are never + trusted: reject anything but [A-Za-z0-9._-]{1,64} and explicitly refuse + '.'/'..' so traversal can never escape PROFILES_DIR.""" + if not isinstance(name, str) or not NAME_RE.match(name) \ + or name in ('.', '..'): + raise ValueError('invalid profile name') return os.path.join(PROFILES_DIR, name) @@ -38,8 +45,14 @@ def auto_name(op): def list_profiles(): try: - return sorted(d for d in os.listdir(PROFILES_DIR) - if os.path.isdir(_path(d))) + out = [] + for d in os.listdir(PROFILES_DIR): + try: + if os.path.isdir(_path(d)): + out.append(d) + except ValueError: + continue + return sorted(out) except OSError: return [] diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index bd5736b..a0d80e3 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -4863,6 +4863,60 @@ def h_rfplan_post(ctx): return (200, result) if ok else (502, result) +def h_profiles_get(ctx): + import mk8_profiles + return 200, {'profiles': mk8_profiles.list_profiles()} + + +def h_profile_save(ctx): + import mk8_events + import mk8_profiles + body = ctx.body or {} + name = (body.get('name') or '').strip() + if not name: + return 400, {'error': 'profile name is required'} + try: + ok = bool(mk8_profiles.snapshot(name)) + except ValueError as exc: + return 400, {'error': str(exc)} + except OSError: + ok = False + try: + mk8_events.log_event('profile_save', sev='info' if ok else 'warn', + msg='profile %s %s' + % (name, 'saved' if ok else 'save failed'), + meta={'name': name, 'ok': ok}) + except Exception: + pass + return (200, {'ok': True}) if ok else (502, {'ok': False}) + + +def h_profile_restore(ctx): + import mk8_events + import mk8_gate + import mk8_profiles + body = ctx.body or {} + name = (body.get('name') or '').strip() + if not name: + return 400, {'error': 'profile name is required'} + mk8_gate.enter('restore_profile') + try: + result = mk8_profiles.restore(name) + except ValueError as exc: + return 400, {'error': str(exc)} + device_run(['wifi', 'reload'], timeout=45) + ok = bool(result.get('ok')) + try: + mk8_events.log_event('profile_restore', + sev='info' if ok else 'warn', + msg='profile %s %s' % (name, 'restored' if ok + else 'restore failed'), + meta=result) + except Exception: + pass + return (200, result) if ok else (502, result) + + # -------------------------------------------------------------------------- # Startup environment check: run once at service startup (and via # ``server.py --env-check`` on the payload screen) to make the device match @@ -7247,6 +7301,9 @@ ROUTER.add('GET', r'/api/mode', h_mode_get) ROUTER.add('POST', r'/api/mode/release', h_mode_release) ROUTER.add('GET', r'/api/rfplan', h_rfplan_get) ROUTER.add('POST', r'/api/rfplan/role', h_rfplan_post) +ROUTER.add('GET', r'/api/reliability/profiles', h_profiles_get) +ROUTER.add('POST', r'/api/reliability/profile', h_profile_save) +ROUTER.add('POST', r'/api/reliability/restore', h_profile_restore) def serve(): diff --git a/payload/user/remote_access/pager-webui/www/index.html b/payload/user/remote_access/pager-webui/www/index.html index 9b6bd57..3c1efd8 100644 --- a/payload/user/remote_access/pager-webui/www/index.html +++ b/payload/user/remote_access/pager-webui/www/index.html @@ -269,7 +269,7 @@ - + diff --git a/payload/user/remote_access/pager-webui/www/js/views.js b/payload/user/remote_access/pager-webui/www/js/views.js index f16443b..d2dcc9a 100644 --- a/payload/user/remote_access/pager-webui/www/js/views.js +++ b/payload/user/remote_access/pager-webui/www/js/views.js @@ -3920,6 +3920,43 @@ views.settings = (root) => { overlay.appendChild(restoreBtn); loadOverlay(); + const profiles = settingsCard(box, 'Reliability Profiles', + 'Snapshots of the PineAP, wireless, and network UCI config. Save one before risky changes; restoring rolls the config back and reloads the radios.'); + const profBody = h('div', { class: 'settings-table-wrap', text: 'Loading…' }); + profiles.appendChild(profBody); + function loadProfiles() { + PagerAPI.get('/api/reliability/profiles').then((r) => { + const names = (r.data || {}).profiles || []; + profBody.innerHTML = ''; + if (!names.length) { + profBody.appendChild(h('div', { class: 'empty', text: 'No saved profiles yet.' })); + return; + } + profBody.appendChild(table([ + { label: 'Profile', key: 'name' }, + { label: '', render: (row) => { + const doRestore = btn('Restore', () => runAction(doRestore, + () => PagerAPI.post('/api/reliability/restore', { name: row.name }) + .then(() => App.toast('Profile restored; radios reloaded')), 'Restoring…'), 'ghost'); + return doRestore; + } } + ], names.map((name) => ({ name })))); + }).catch(() => { profBody.textContent = 'Unable to load profiles.'; }); + } + const profName = h('input', { placeholder: 'e.g. pre-evilwpa', maxlength: '64', + autocomplete: 'off' }); + const profSave = btn('Save Profile', () => { + const name = profName.value.trim(); + if (!name) { App.toast('Profile name is required', 'error'); return; } + return runAction(profSave, () => PagerAPI.post('/api/reliability/profile', { name }) + .then(() => { profName.value = ''; App.toast('Config profile saved'); loadProfiles(); }), 'Saving…'); + }); + profiles.appendChild(h('div', { class: 'settings-form-grid' }, + h('label', {}, 'Profile Name', profName))); + profiles.appendChild(h('div', { class: 'settings-actions' }, profSave, + btn('Refresh', () => { loadProfiles(); }, 'ghost'))); + loadProfiles(); + settingsCard(box, 'Button Script', 'The Mark VII button script has no safe Pager equivalent. Pager buttons remain managed by the native input and payload-launcher system.'); const resources = settingsCard(box, 'Resources'); diff --git a/scripts/build_common.py b/scripts/build_common.py new file mode 100644 index 0000000..6bf16d6 --- /dev/null +++ b/scripts/build_common.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Build helpers shared by Mark VIII deploy scripts. + +stamp_version() stamps release metadata into BUILD COPIES ONLY: callers +always pass a staging/build directory, never the source tree, so the repo +stays clean while every deployed artifact reports the same VERSION. +""" + +import json +import os +import re +import sys + +VERSION_RE = re.compile(r'^[0-9][A-Za-z0-9._-]{0,31}$') +SERVER_VERSION_LINE = "SERVER_VERSION = '%s'\n" + + +def _stamp_manifest(path, version): + try: + with open(path, encoding='utf-8') as f: + data = json.load(f) + except (OSError, ValueError): + return False + if not isinstance(data, dict): + return False + data['version'] = version + with open(path, 'w', encoding='ascii') as f: + json.dump(data, f, indent=2) + f.write('\n') + return True + + +def _stamp_payload_sh(path, version): + try: + with open(path, encoding='utf-8') as f: + lines = f.readlines() + except OSError: + return False + out = [] + replaced = False + for line in lines: + m = None if replaced \ + else re.match(r'^(\s*#\s*[Vv]ersion:).*$', line) + if m: + out.append(m.group(1) + ' ' + version + '\n') + replaced = True + else: + out.append(line) + if not replaced: + insert = 1 if lines and lines[0].startswith('#!') else 0 + out.insert(insert, '# Version: %s\n' % version) + with open(path, 'w', encoding='utf-8') as f: + f.writelines(out) + return True + + +def _server_insert_index(lines): + """Index just past any shebang/comments/blanks and module docstring.""" + i = 0 + n = len(lines) + if i < n and lines[i].startswith('#!'): + i += 1 + while i < n and (lines[i].strip().startswith('#') + or not lines[i].strip()): + i += 1 + if i < n: + stripped = lines[i].lstrip() + quote = stripped[:3] + if quote in ('"""', "'''"): + closed_here = quote in stripped[3:] + i += 1 + if not closed_here: + while i < n and quote not in lines[i]: + i += 1 + i += 1 + return min(i, len(lines)) + + +def _stamp_server_py(path, version): + try: + with open(path, encoding='utf-8') as f: + lines = f.readlines() + except OSError: + return False + lines = [l for l in lines if not l.startswith('SERVER_VERSION')] + idx = _server_insert_index(lines) + lines.insert(idx, SERVER_VERSION_LINE % version) + with open(path, 'w', encoding='utf-8') as f: + f.writelines(lines) + return True + + +def stamp_version(build_dir, version): + """Stamp into build copies found under build_dir. + + Updates every ``_hak5_manifest.json`` (version field), ``payload.sh`` + (header Version line) and ``server.py`` (injected SERVER_VERSION + constant near the top). Idempotent: re-running never duplicates the + injected constant or header. Returns the list of stamped paths.""" + if not VERSION_RE.match(str(version)): + raise ValueError('invalid version string: %r' % (version,)) + stamped = [] + for root, dirs, files in os.walk(build_dir): + for fname in ('_hak5_manifest.json', 'payload.sh', 'server.py'): + if fname in files: + path = os.path.join(root, fname) + if fname == '_hak5_manifest.json': + ok = _stamp_manifest(path, version) + elif fname == 'payload.sh': + ok = _stamp_payload_sh(path, version) + else: + ok = _stamp_server_py(path, version) + if ok: + stamped.append(path) + return sorted(stamped) + + +if __name__ == '__main__': + if len(sys.argv) != 3: + print('usage: build_common.py ', + file=sys.stderr) + sys.exit(2) + for path in stamp_version(sys.argv[1], sys.argv[2]): + print('stamped: %s' % path) diff --git a/scripts/deploy.sh b/scripts/deploy.sh index fdb6afb..e90da96 100755 --- a/scripts/deploy.sh +++ b/scripts/deploy.sh @@ -21,6 +21,11 @@ Options: --no-portal-refresh Skip the best-effort portal refresh -h, --help Show this help +Deploys to /mmc/mk8/releases// with an atomically repointed 'current' +symlink, mirrors the payload into the legacy /root/payloads location the +init scripts run from, verifies sha256 of the upload, and polls the local +API after start; on failure it rolls the symlink + legacy dir back. + If neither --password nor --ssh-key is supplied, ssh/scp prompt normally. EOF } @@ -61,6 +66,17 @@ STAGE="$OUT_DIR/stage" printf 'Payload directory not found: %s\n' "$PAYLOAD_DIR" >&2 exit 1 } +VERSION_FILE="$ROOT/VERSION" +[[ -f "$VERSION_FILE" ]] || { + printf 'VERSION file not found: %s\n' "$VERSION_FILE" >&2 + exit 1 +} +VERSION="$(tr -d '[:space:]' < "$VERSION_FILE")" +[[ -n "$VERSION" ]] || { + printf 'VERSION file is empty.\n' >&2 + exit 1 +} +printf 'Deploying Mark VIII version %s\n' "$VERSION" mkdir -p "$OUT_DIR" rm -rf "$STAGE" @@ -68,6 +84,10 @@ mkdir -p "$STAGE/user/$PAYLOAD_CATEGORY" cp -R "$PAYLOAD_DIR" "$STAGE/user/$PAYLOAD_CATEGORY/$PAYLOAD_KEY" find "$STAGE" \( -type d -name __pycache__ -o -type f -name '*.pyc' \) -prune -exec rm -rf {} + +# Stamp build copies only (never the source tree): payload.sh header, +# staged server.py SERVER_VERSION constant; manifest is stamped below. +python3 "$ROOT/scripts/build_common.py" "$STAGE" "$VERSION" + B64_KEY="$(python3 -c 'import base64; print(base64.urlsafe_b64encode(b"pager-webui").decode().rstrip("="))')" ZIP_NAME="payload-$B64_KEY.zip" ZIP_PATH="$OUT_DIR/$ZIP_NAME" @@ -94,7 +114,9 @@ with open(destination, 'w', encoding='ascii') as handle: json.dump(manifest, handle, indent=2) handle.write('\n') PY -printf 'Built: %s\n' "$ZIP_PATH" +# Single-source version: stamp the generated manifest copy too. +python3 "$ROOT/scripts/build_common.py" "$MANIFEST_PATH" "$VERSION" >/dev/null +printf 'Built: %s (sha256 %s)\n' "$ZIP_PATH" "$HASH" TARGET="$PAGER_USER@$PAGER_HOST" SSH_OPTS=(-o StrictHostKeyChecking=accept-new) @@ -166,49 +188,88 @@ echo PYTHON_OK' install_python3 -run_scp "$ZIP_PATH" "$MANIFEST_PATH" "$TARGET:/tmp/" +run_ssh "$TARGET" 'mkdir -p /tmp/mk8-stage && rm -rf /tmp/mk8-stage/*' +run_scp "$ZIP_PATH" "$MANIFEST_PATH" "$TARGET:/tmp/mk8-stage/" REMOTE_PAYLOAD_DIR="user/$PAYLOAD_CATEGORY/$PAYLOAD_KEY" -LEGACY_PAYLOAD_DIR="user/general/$PAYLOAD_KEY" +RELEASE_TS="$(date +%Y%m%d-%H%M%S)" REMOTE_COMMAND="set -e -cd /root/payloads -stage='.pager-webui.deploy.\$\$' -backup='.pager-webui.backup.\$\$' -trap 'rm -rf \"\$stage\" \"\$backup\"' EXIT -mkdir -p \"\$stage\" -cd \"\$stage\" -unzip -q '/tmp/$ZIP_NAME' -new=\"\$PWD/$REMOTE_PAYLOAD_DIR\" -[ -f \"\$new/server.py\" ] && [ -f \"\$new/payload.sh\" ] && [ -d \"\$new/www\" ] -cp /tmp/_hak5_manifest.json \"\$new/_hak5_manifest.json\" -chmod +x \"\$new/payload.sh\" \"\$new/pagerwebui.init\" -chmod -R 755 \"\$new/www\" -cd /root/payloads -if [ -d '$REMOTE_PAYLOAD_DIR' ]; then - mkdir -p \"\$(dirname \"\$backup\")\" - mv '$REMOTE_PAYLOAD_DIR' \"\$backup\" -fi -if mv \"\$new\" '$REMOTE_PAYLOAD_DIR'; then - rm -rf \"\$backup\" '$LEGACY_PAYLOAD_DIR' -else - [ ! -d \"\$backup\" ] || mv \"\$backup\" '$REMOTE_PAYLOAD_DIR' +STAGE_DIR='/tmp/mk8-stage' +ZIP='\$STAGE_DIR/$ZIP_NAME' +RELDIR='/mmc/mk8/releases/$RELEASE_TS' +PAYDIR='$REMOTE_PAYLOAD_DIR' +LIVE=\"/root/payloads/\$PAYDIR\" +BACKUP=\"/root/payloads/.pager-webui.backup.\$\$\" +CURRENT='/mmc/mk8/releases/current' +PREV=\$(readlink \$CURRENT 2>/dev/null || true) +cleanup() { rm -rf \"\$STAGE_DIR\"; } +trap cleanup EXIT + +# Upload integrity gate: remote sha256 must match the local build hash. +GOT=\$(sha256sum \"\$ZIP\" | awk '{print \$1}') +[ \"\$GOT\" = '$HASH' ] || { echo 'sha256 mismatch on uploaded zip' >&2; exit 1; } +[ -f \"\$STAGE_DIR/_hak5_manifest.json\" ] || { echo 'manifest missing' >&2; exit 1; } + +/etc/init.d/pagerwebui stop >/dev/null 2>&1 || true +mkdir -p \"\$RELDIR\" +unzip -q \"\$ZIP\" -d \"\$RELDIR\" +NEW=\"\$RELDIR/\$PAYDIR\" +[ -f \"\$NEW/server.py\" ] && [ -f \"\$NEW/payload.sh\" ] && [ -d \"\$NEW/www\" ] || { + echo 'release payload incomplete' >&2 + rm -rf \"\$RELDIR\" + exit 1 +} +cp \"\$STAGE_DIR/_hak5_manifest.json\" \"\$NEW/_hak5_manifest.json\" +chmod +x \"\$NEW/payload.sh\" \"\$NEW/pagerwebui.init\" +chmod -R 755 \"\$NEW/www\" +ln -sfn \"\$RELDIR\" \"\$CURRENT\" + +# Mirror into the legacy /root/payloads path the init scripts execute. +rollback_install() { + rm -rf \"\$LIVE\" + [ ! -d \"\$BACKUP\" ] || mv \"\$BACKUP\" \"\$LIVE\" + [ -z \"\$PREV\" ] || ln -sfn \"\$PREV\" \"\$CURRENT\" +} +if [ -d \"\$LIVE\" ]; then mv \"\$LIVE\" \"\$BACKUP\"; fi +if ! mkdir -p \"\$LIVE\" || ! cp -a \"\$NEW/.\" \"\$LIVE/\"; then + rollback_install exit 1 fi -rm -f '/tmp/$ZIP_NAME' /tmp/_hak5_manifest.json -cp '$REMOTE_PAYLOAD_DIR/pagerwebui.init' /etc/init.d/pagerwebui +cp -f \"\$LIVE/pagerwebui.init\" /etc/init.d/pagerwebui chmod +x /etc/init.d/pagerwebui /etc/init.d/pagerwebui enable -if /etc/init.d/pagerwebui running >/dev/null 2>&1; then - /etc/init.d/pagerwebui restart -else - /etc/init.d/pagerwebui start -fi -cp -f '$REMOTE_PAYLOAD_DIR/mk8-guard.init' /etc/init.d/mk8-guard +cp -f \"\$LIVE/mk8-guard.init\" /etc/init.d/mk8-guard chmod 755 /etc/init.d/mk8-guard /etc/init.d/mk8-guard enable -echo EXTRACT_OK" -run_ssh "$TARGET" "$REMOTE_COMMAND" -printf 'Installed to /root/payloads/%s/\n' "$REMOTE_PAYLOAD_DIR" +/etc/init.d/pagerwebui start + +# Post-deploy verification: API must answer within 60s or we roll back. +i=0 +HEALTH_OK='' +while [ \$i -lt 60 ]; do + if curl -fsS -m 3 http://127.0.0.1:8080/api/api_ping >/dev/null 2>&1; then + HEALTH_OK=1 + break + fi + sleep 1 + i=\$((i+1)) +done +if [ -z \"\$HEALTH_OK\" ]; then + echo 'post-deploy health check failed; rolling back' >&2 + /etc/init.d/pagerwebui stop >/dev/null 2>&1 || true + rollback_install + rm -rf \"\$RELDIR\" + /etc/init.d/pagerwebui start + exit 1 +fi +rm -rf \"\$BACKUP\" +echo RELEASE_OK@\"\$RELDIR\"" +if run_ssh "$TARGET" "$REMOTE_COMMAND"; then + printf 'Release active: /mmc/mk8/releases/%s\n' "$RELEASE_TS" +else + printf 'Deployment failed; previous release restored on the pager.\n' >&2 + exit 1 +fi if $PORTAL_REFRESH && [[ -n "$PASSWORD" ]]; then PASSWORD_B64="$(printf '%s' "$PASSWORD" | base64)" @@ -235,4 +296,4 @@ elif $PORTAL_REFRESH; then printf 'Skipping portal refresh without --password; payload installation is complete.\n' fi -printf 'Deploy complete. Browse http://%s:8080/\n' "$PAGER_HOST" +printf 'Deploy complete (v%s). Browse http://%s:8080/\n' "$VERSION" "$PAGER_HOST" diff --git a/tests/test_build_common.py b/tests/test_build_common.py new file mode 100644 index 0000000..529934e --- /dev/null +++ b/tests/test_build_common.py @@ -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() diff --git a/tests/test_mk8_profiles.py b/tests/test_mk8_profiles.py index 9649e1d..c82667f 100644 --- a/tests/test_mk8_profiles.py +++ b/tests/test_mk8_profiles.py @@ -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() diff --git a/tests/test_reliability_api.py b/tests/test_reliability_api.py index cf53d30..9d8307c 100644 --- a/tests/test_reliability_api.py +++ b/tests/test_reliability_api.py @@ -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()