196 lines
9.1 KiB
Python
196 lines
9.1 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 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.assertNotIn('TAILSCALE up', r.stdout)
|
|
self.assertNotIn('MARK hold', r.stdout)
|
|
|
|
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)
|