"""Offline regression tests: no real containers, services, APT or credentials.""" from pathlib import Path import subprocess import tempfile import unittest SOURCE = Path(__file__).with_name('lxc-enroll.sh').read_text() WORKER = SOURCE.split("<<'WORKER'\n", 1)[1].split('\nWORKER\n', 1)[0] def bash(code, *args): return subprocess.run(['bash', '-c', code, '--', *args], capture_output=True, text=True) class EnrollTests(unittest.TestCase): def run_ui(self, action, answer='', rc=0, tool='whiptail', rows=24): lib = SOURCE.rsplit('\nmain\n', 1)[0] lib = lib.replace('set -Eeuo pipefail', 'set -Eeo pipefail', 1) lib = lib.replace("[[ $EUID == 0 ]] || fail 'Bitte auf dem Proxmox-Host als root starten.'", ':') lib = lib.replace('for tool in pct timeout python3; do command -v "$tool" >/dev/null || fail "$tool fehlt auf dem Host."; done', ':') with tempfile.TemporaryDirectory() as td: p = Path(td) mocks = r''' exec 3<>"$1/terminal" exec 5>"$1/dialog-args" TERM=xterm command() { if [[ $1 == -v && ( $2 == whiptail || $2 == dialog ) ]]; then [[ $2 == $MOCK_TOOL ]]; else builtin command "$@"; fi } stty() { printf '%s 100\n' "$MOCK_ROWS"; } whiptail() { printf '%s\n' "$@" >&5; printf '%s' "$MOCK_ANSWER" >&4; return "$MOCK_RC"; } dialog() { whiptail "$@"; } init_ui IDS=(101 103); NAMES[101]='first'; NAMES[103]='partial' TS[101]=0; WZ[101]=0; TUN[101]=0 TS[103]=1; WZ[103]=0; TUN[103]=1 ''' setup = 'MOCK_TOOL=$2; MOCK_ROWS=$3; MOCK_RC=$4; MOCK_ANSWER=$5\n' r = bash('UI_TEST_ARGS=("$@"); set --\n' + lib + '\nset -- "${UI_TEST_ARGS[@]}"\n' + setup + mocks + action, td, tool, str(rows), str(rc), answer) return r, (p/'dialog-args').read_text(), (p/'terminal').read_text() def test_checklist_captures_multiple_ids_and_shows_component_status(self): r, args, terminal = self.run_ui('choose_lxcs; echo "SELECTION=$SELECTION"', '101\n103\n') self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('SELECTION=101 103', r.stdout) self.assertIn('--separate-output', args) self.assertIn('--checklist', args) self.assertIn('partial | Tailscale: vorhanden | Wazuh: fehlt', args) self.assertEqual(args.splitlines().count('OFF'), 2) def test_cancel_or_empty_checklist_stops_selection(self): for rc, answer in ((1, ''), (255, ''), (0, '')): r, args, terminal = self.run_ui( 'if choose_lxcs; then echo CONTINUE; else echo CANCEL; fi', answer, rc) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('CANCEL', r.stdout) self.assertNotIn('CONTINUE', r.stdout) def test_yes_no_dialog_preserves_default_and_handles_escape(self): r, args, terminal = self.run_ui('yes_no "Sudo? [j/N]" 0; echo "ANSWER=$REPLY"', rc=1) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('ANSWER=0', r.stdout) self.assertIn('--defaultno', args) self.assertNotIn('[j/N]', args) r, args, terminal = self.run_ui('yes_no "SSH? [J/n]" 1; echo "ANSWER=$REPLY"') self.assertIn('ANSWER=1', r.stdout) self.assertNotIn('--defaultno', args) r, args, terminal = self.run_ui('yes_no "SSH?" 1; echo CONTINUE', rc=255) self.assertNotEqual(r.returncode, 0) self.assertNotIn('CONTINUE', r.stdout) def test_password_dialog_does_not_echo_secret_or_put_it_in_arguments(self): r, args, terminal = self.run_ui('secret "Key:"; echo "LENGTH=${#REPLY}"', 'fixture-private-key') self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('--passwordbox', args) self.assertNotIn('--insecure', args) self.assertNotIn('fixture-private-key', r.stdout + r.stderr + args + terminal) def test_input_dialog_and_dialog_fallback(self): r, args, terminal = self.run_ui('ask "Nutzer:" "existing"; echo "INPUT=$REPLY"', 'chosen', tool='dialog') self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('--inputbox', args) self.assertIn('existing', args) self.assertIn('INPUT=chosen', r.stdout) def test_no_tool_or_small_terminal_uses_text_without_installing_anything(self): for tool, rows in (('none', 24), ('whiptail', 12)): r, args, terminal = self.run_ui( 'ask() { REPLY="101 103"; }; choose_lxcs; echo "SELECTION=$SELECTION"', tool=tool, rows=rows) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('SELECTION=101 103', r.stdout) self.assertEqual(args, '') def test_syntax(self): for code in (SOURCE, WORKER): r = subprocess.run(['bash', '-n'], input=code, text=True, capture_output=True) self.assertEqual(r.returncode, 0, r.stderr) def test_inventory_excludes_complete_stopped_locked_and_unsupported(self): # Function-only host harness. Mac Bash 3 lacks modern nounset-array semantics. lib = SOURCE.rsplit('\nmain\n', 1)[0] lib = lib.replace('set -Eeuo pipefail', 'set -Eeo pipefail', 1) lib = lib.replace("[[ $EUID == 0 ]] || fail 'Bitte auf dem Proxmox-Host als root starten.'", ':') lib = lib.replace('for tool in pct timeout python3; do command -v "$tool" >/dev/null || fail "$tool fehlt auf dem Host."; done', ':') mocks = r''' timeout() { shift; "$@"; } pct() { case "$1" in list) printf 'VMID Status Lock Name\n102 running docker\n106 running adguard\n107 running complete\n108 stopped stopped\n109 running alpine\n112 running locked\n133 running manager\n';; config) if [[ $2 == 112 ]]; then echo 'lock: backup'; else echo 'hostname: fixture'; fi;; exec) [[ $4 == sh && $5 == -s ]] || return 99 cat >/dev/null case "$2" in 102) echo 'docker|debian|bookworm|1|0|0|1|1|done';; 106) echo 'adguard|debian|bookworm|0|1|0|0|1|done';; 107) echo 'complete|ubuntu|noble|1|1|0|1|1|done';; 109) echo 'alpine|alpine|unknown|0|0|0|0|0|done';; 133) echo 'manager|debian|bookworm|0|1|1|0|1|done';; esac;; *) echo "UNEXPECTED MUTATION: $*" >&2; return 99;; esac } scan printf 'CANDIDATES=%s\n' "${IDS[*]}" printf 'MANAGERS=%s\n' "${MANAGERS[*]}" ''' r = bash(lib + mocks) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('CANDIDATES=102 106 133', r.stdout) self.assertIn('MANAGERS=133', r.stdout) self.assertNotIn('UNEXPECTED MUTATION', r.stderr) def run_worker(self, want_ts, want_wz, ssh=0, installed=False, failure=False, dns=0, resume=0, ts_installed=None, state='NeedsLogin'): with tempfile.TemporaryDirectory() as td: p = Path(td) (p/'ossec/var/run').mkdir(parents=True) (p/'ossec/var/run/wazuh-agentd.state').write_text("status='connected'\n") (p/'os-release').write_text('NAME="Debian GNU/Linux"\nID=debian\nVERSION_CODENAME=bookworm\nVERSION="12 (bookworm)"\n') (p/'headscale.key').write_text('fixture-secret-not-for-logging') worker = WORKER for old, new in ( ('/var/ossec', str(p/'ossec')), ('/etc/os-release', str(p/'os-release')), ('/etc/apt', str(p/'apt')), ('/usr/share/keyrings', str(p/'keyrings')), ('/dev/net/tun', '/dev/null'), ): worker = worker.replace(old, new) mocks = r''' command() { if [[ $1 == -v && $2 == tailscale ]]; then [[ $MOCK_TS == 1 ]]; else builtin command "$@"; fi } dpkg-query() { case "${@: -1}" in tailscale) [[ $MOCK_TS != 1 ]] || printf 'install ok installed';; wazuh-agent) [[ $MOCK_INSTALLED != 1 ]] || printf 'install ok installed';; esac } timeout() { shift if [[ $1 == bash && $2 == -c ]]; then echo "CHECK TCP ${@: -1}"; return 0; fi "$@" } apt-get() { echo "APT $*" if [[ " $* " == *' wazuh-agent='* ]]; then echo "WAZUH NAME $WAZUH_AGENT_NAME"; fi if [[ $MOCK_FAILURE == 1 && " $* " == *' install '* ]]; then return 42; fi } apt-cache() { echo 'wazuh-agent | 4.14.8-1 | fixture'; } apt-mark() { echo "MARK $*"; } curl() { local last=''; for a in "$@"; do last=$a; done; printf 'fixture' >"$last"; } gpg() { local prev='' out=''; for a in "$@"; do [[ $prev != -o ]] || out=$a; prev=$a; done; printf 'fixture' >"$out"; } systemctl() { echo "SERVICE $*"; } tailscale() { if [[ $1 == status ]]; then printf '{"BackendState":"%s"}\n' "$MOCK_STATE"; return 0; fi echo "TAILSCALE $*" if [[ $1 == up ]]; then MOCK_STATE=Running; fi } id() { if [[ $1 == -u ]]; then echo 1000; else return 0; fi; } ''' mocks = 'MOCK_INSTALLED=%d; MOCK_TS=%d; MOCK_FAILURE=%d; MOCK_STATE=%s\n' % ( installed, installed if ts_installed is None else ts_installed, failure, state) + mocks r = bash(mocks + worker, td, 'manager.example.com', '4.14.8-1', 'https://headscale.example.com', 'fixture', str(want_ts), str(want_wz), str(ssh), 'fixtureuser', '0', '0', str(dns), str(resume)) self.assertNotIn('fixture-secret-not-for-logging', r.stdout+r.stderr) self.assertFalse((p/'headscale.key').exists(), 'secret cleanup did not run') return r def test_install_both_preserves_manager_version_after_os_release(self): r = self.run_worker(1, 1) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('wazuh-agent=4.14.8-1', r.stdout) self.assertIn('--hostname=fixture', r.stdout) self.assertIn('WAZUH NAME fixture', r.stdout) self.assertNotIn('Debian GNU/Linux', r.stdout) self.assertIn('--auth-key=file:', r.stdout) self.assertIn('--accept-dns=false', r.stdout) self.assertIn('mit dem Manager verbunden', r.stdout) self.assertLess(r.stdout.index('TAILSCALE up'), r.stdout.index('CHECK TCP 1514')) def test_existing_components_are_not_reinstalled(self): r = self.run_worker(1, 1, installed=True) self.assertEqual(r.returncode, 0, r.stderr) self.assertNotIn('APT ', r.stdout) self.assertNotIn('TAILSCALE up', r.stdout) def test_resume_installed_unregistered_tailscale_before_wazuh(self): r = self.run_worker(0, 1, ts_installed=True, resume=1, dns=1) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('TAILSCALE up', r.stdout) self.assertIn('--hostname=fixture', r.stdout) self.assertNotIn('install -y --no-install-recommends tailscale', r.stdout) self.assertIn('WAZUH NAME fixture', r.stdout) self.assertLess(r.stdout.index('TAILSCALE up'), r.stdout.index('CHECK TCP 1514')) def test_resume_preserves_registered_client(self): r = self.run_worker(0, 1, ts_installed=True, resume=1, state='Running') self.assertEqual(r.returncode, 0, r.stderr) self.assertNotIn('TAILSCALE up', r.stdout) self.assertIn('Registrierung bleibt erhalten', r.stdout) self.assertIn('wazuh-agent=4.14.8-1', r.stdout) def test_resume_rejects_ambiguous_state_without_reauthentication(self): r = self.run_worker(0, 1, ts_installed=True, resume=1, state='Stopped') self.assertNotEqual(r.returncode, 0) self.assertNotIn('TAILSCALE up', r.stdout) self.assertNotIn('APT ', r.stdout) def test_yes_no_accepts_numeric_answers_and_default(self): fn = SOURCE.split('yes_no() {', 1)[1].split('\n}\n', 1)[0] r = bash('fail() { exit 1; }; ask() { REPLY=$ANSWER; }; yes_no() {' + fn + '\n}\n' + 'for ANSWER in 1 0 j n ""; do yes_no prompt 1; echo "ANSWER=$REPLY"; done') self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(r.stdout.splitlines(), ['ANSWER=1', 'ANSWER=0', 'ANSWER=1', 'ANSWER=0', 'ANSWER=1']) def test_magicdns_enabled_before_manager_connection(self): r = self.run_worker(1, 1, dns=1) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('--accept-dns=true', r.stdout) self.assertLess(r.stdout.index('--accept-dns=true'), r.stdout.index('CHECK TCP 1514')) def test_existing_client_accepts_dns_without_reauthentication(self): r = self.run_worker(0, 0, ssh=1, installed=True, dns=1) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('TAILSCALE set --accept-dns=true', r.stdout) self.assertNotIn('TAILSCALE up', r.stdout) def test_tailscale_ssh_uses_no_openssh_server_or_public_key(self): r = self.run_worker(0, 0, ssh=1, installed=True) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn('TAILSCALE set --ssh=true', r.stdout) self.assertNotIn('openssh-server', r.stdout) self.assertNotIn('SERVICE enable --now ssh', r.stdout) self.assertNotIn('authorized_keys', WORKER) def test_install_failure_stops_and_cleans_secrets(self): r = self.run_worker(1, 1, failure=True) self.assertNotEqual(r.returncode, 0) self.assertIn('APT-Grundpakete (Status 42, Worker-Zeile', r.stderr) self.assertNotIn('TAILSCALE up', r.stdout) self.assertNotIn('MARK hold', r.stdout) def test_deployment_reports_worker_and_cleanup_failure_separately(self): lib = SOURCE.rsplit('\nmain\n', 1)[0] lib = lib.replace('set -Eeuo pipefail', 'set -Eeo pipefail', 1) lib = lib.replace("[[ $EUID == 0 ]] || fail 'Bitte auf dem Proxmox-Host als root starten.'", ':') lib = lib.replace('for tool in pct timeout python3; do command -v "$tool" >/dev/null || fail "$tool fehlt auf dem Host."; done', ':') mocks = r''' TS[144]=1; WZ[144]=0; TUN[144]=1; NAMES[144]=fixture TMP=/fixture; MANAGER=manager.example.com; VERSION=4.14.8-1 pct() { case "$1" in status) echo 'status: running';; config) echo 'hostname: fixture';; push) return 0;; exec) case "$4" in mktemp) echo /run/lxc-enroll.Fixture1;; bash) echo 'Wazuh connected; SSH enabled'; [[ $MOCK_STEP != worker ]] || return 23;; rm) [[ $MOCK_STEP != cleanup ]] || return 17;; *) return 99;; esac;; *) return 99;; esac } if deploy_one 144; then echo SUCCESS; else echo FAILED; fi ''' for step, status, label in (('worker', 23, 'Einrichtung im Container'), ('cleanup', 17, 'Temporaere Dateien entfernen'), ('none', 0, '')): r = bash(lib + '\nMOCK_STEP=' + step + '\n' + mocks) self.assertEqual(r.returncode, 0, r.stderr) if status: self.assertIn('FAILED', r.stdout) self.assertIn(f'LXC 144: {label} fehlgeschlagen (Status {status}).', r.stderr) else: self.assertIn('SUCCESS', r.stdout) self.assertEqual(r.stderr, '') def test_public_source_has_no_site_specific_values(self): for forbidden in ('faboi.de', '192.168.', '10.10.', 'tskey-auth-', 'hskey-auth-'): self.assertNotIn(forbidden, SOURCE) self.assertNotIn('source "$VARS_FILE"', SOURCE) self.assertNotIn('eval ', SOURCE) if __name__ == '__main__': unittest.main(verbosity=2)