314 lines
15 KiB
Python
314 lines
15 KiB
Python
"""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)
|