#!/usr/bin/env python3 """Isolated fail-closed SRS service status / listener / verification tests. Only extracts shell functions and read-only verification code from deactivate-srs.sh. All calls to systemctl and ss use mocks; no system service is modified. Run from repository root: python3 tests/test_deactivation_status_mock.py """ import re import subprocess import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] / "deactivate-srs.sh" TEXT = SOURCE.read_text(encoding="utf-8") FUNCTIONS = ( "read_srs_service_state", "service_active", "service_enabled_state", "service_enabled", "snapshot_service_states", "restore_service_states", "port_is_listening", "master_srs_forward_present", ) def extract_function(name): match = re.search(rf"^{re.escape(name)}\(\) \{{\n.*?^\}}", TEXT, re.M | re.S) if not match: raise ValueError(f"Missing shell function {name}") return match.group(0) FUNCTION_TEXT = "\n".join(extract_function(name) for name in FUNCTIONS) def section(start, end): begin = TEXT.index(start) finish = TEXT.index(end, begin) return TEXT[begin:finish] PREFLIGHT = section('services_exist=true\n', '\n\n# ----------\n# Check Postfix integration') FINAL = section('verification_ok=true\n', 'if ! $verification_ok ; then') STATE_CLASSIFICATION = section('srs_integration_enabled=false\n', '\n\n# ----------\n# Summary') class StatusMockTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="srs-status-mock-") self.addCleanup(self.tmp.cleanup) self.base = Path(self.tmp.name) self.master = self.base / 'master.cf' self.master.write_text('smtp inet n - y - - smtpd\n') self.calls = self.base / 'calls' self.log = self.base / 'log' def run_shell(self, body, config='', functions=True): script = f'''#!/bin/bash set -u log_file={str(self.log)!r} CALLS={str(self.calls)!r} POSTFIX_MASTER_CF={str(self.master)!r} POSTSRSD_SERVICE=postsrsd-srs.service SRS_REVERSE_MAP_SERVICE=srs-reverse-map.service SRS_DELIVERY_SERVICE=postfix@srs-delivery.service declare -A SERVICE_WAS_ACTIVE=() declare -A SERVICE_WAS_ENABLED=() warn() {{ :; }} echononl() {{ :; }} echo_ok() {{ :; }} echo_failed() {{ :; }} echo_inactive() {{ :; }} fatal() {{ echo "FATAL: $*" >&2; exit 70; }} systemctl() {{ printf 'systemctl %s\\n' "$*" >> "$CALLS" case "$1" in show) [[ "${{SHOW_FAIL_UNIT:-}}" == "$2" ]] && return 17 if [[ "${{SHOW_BAD_UNIT:-}}" == "$2" ]]; then printf 'LoadState=not-found\\nActiveState=inactive\\nUnitFileState=disabled\\n' return 0 fi printf '%s\\n' "${{SHOW_DATA:-$'LoadState=loaded\\nActiveState=active\\nUnitFileState=enabled'}}" ;; enable|disable|start|stop) [[ "${{FAIL_ACTION:-}}" == "$1" ]] && return 11 return 0 ;; *) return 99 ;; esac }} ss() {{ [[ "${{SS_FAIL:-0}}" == 1 ]] && return 7 printf '%s\\n' "${{SS_DATA:-}}" }} {FUNCTION_TEXT if functions else ''} {config} {body} ''' return subprocess.run(['bash', '-c', script], capture_output=True, text=True, timeout=15) def assert_exit(self, body, rc=0, config='', functions=True): p = self.run_shell(body, config=config, functions=functions) self.assertEqual(p.returncode, rc, f"stdout={p.stdout!r}\nstderr={p.stderr!r}") return p def test_loaded_active_enabled_and_runtime_enabled(self): self.assert_exit(''' read_srs_service_state postsrsd-srs.service || exit 90 [[ "$SRS_SERVICE_ACTIVE_STATE" == active && "$SRS_SERVICE_ENABLED_STATE" == enabled ]] || exit 91 service_active postsrsd-srs.service || exit 92 service_enabled postsrsd-srs.service || exit 93 ''') self.assert_exit(''' SHOW_DATA=$'LoadState=loaded\\nActiveState=active\\nUnitFileState=enabled-runtime' service_enabled postsrsd-srs.service || exit 94 [[ "$(service_enabled_state postsrsd-srs.service)" == enabled-runtime ]] || exit 95 ''') def test_inactive_disabled_is_valid_and_has_nonzero_status(self): self.assert_exit(''' SHOW_DATA=$'LoadState=loaded\\nActiveState=inactive\\nUnitFileState=disabled' service_active postsrsd-srs.service; [[ $? -eq 1 ]] || exit 1 service_enabled postsrsd-srs.service; [[ $? -eq 1 ]] || exit 2 [[ "$(service_enabled_state postsrsd-srs.service)" == disabled ]] || exit 3 ''') def test_systemctl_error_and_not_found_return_two(self): self.assert_exit(''' SHOW_FAIL_UNIT=postsrsd-srs.service service_active postsrsd-srs.service; [[ $? -eq 2 ]] || exit 1 service_enabled postsrsd-srs.service; [[ $? -eq 2 ]] || exit 2 service_enabled_state postsrsd-srs.service >/dev/null; [[ $? -eq 2 ]] || exit 3 unset SHOW_FAIL_UNIT SHOW_BAD_UNIT=postsrsd-srs.service read_srs_service_state postsrsd-srs.service; [[ $? -eq 2 ]] || exit 4 ''') def test_invalid_and_transient_states_are_rejected(self): for state in [ 'LoadState=loaded\nActiveState=failed\nUnitFileState=enabled', 'LoadState=loaded\nActiveState=activating\nUnitFileState=enabled', 'LoadState=loaded\nActiveState=inactive\nUnitFileState=masked', 'LoadState=loaded\nActiveState=inactive\nUnitFileState=static', 'LoadState=loaded\nActiveState=inactive', 'LoadState=loaded\nActiveState=inactive\nUnitFileState=disabled\nUnexpected=yes', ]: with self.subTest(state=state): # Quote shell strings safely by passing bash's single-quoted, escaped data. import shlex self.assert_exit(''' read_srs_service_state postsrsd-srs.service [[ $? -eq 2 ]] ''', config=f'SHOW_DATA={shlex.quote(state)}\n') def test_snapshot_complete_and_fail_atomic(self): self.assert_exit(''' snapshot_service_states || exit 1 for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do [[ "${SERVICE_WAS_ACTIVE[$unit]}" == true ]] || exit 2 [[ "${SERVICE_WAS_ENABLED[$unit]}" == enabled ]] || exit 3 done ''') self.assert_exit(''' SHOW_FAIL_UNIT=srs-reverse-map.service snapshot_service_states; [[ $? -eq 1 ]] || exit 1 [[ "${#SERVICE_WAS_ACTIVE[@]}" -eq 0 ]] || exit 2 [[ "${#SERVICE_WAS_ENABLED[@]}" -eq 0 ]] || exit 3 ''') def test_snapshot_rejects_state_change_after_enabled_preflight(self): self.assert_exit(''' SHOW_DATA=$'LoadState=loaded\nActiveState=inactive\nUnitFileState=disabled' snapshot_service_states; [[ $? -eq 1 ]] || exit 1 [[ "${#SERVICE_WAS_ACTIVE[@]}" -eq 0 ]] || exit 2 ''') def test_restore_exact_states(self): self.assert_exit(''' SERVICE_WAS_ENABLED[postsrsd-srs.service]=enabled SERVICE_WAS_ACTIVE[postsrsd-srs.service]=true SERVICE_WAS_ENABLED[srs-reverse-map.service]=enabled-runtime SERVICE_WAS_ACTIVE[srs-reverse-map.service]=true SERVICE_WAS_ENABLED[postfix@srs-delivery.service]=disabled SERVICE_WAS_ACTIVE[postfix@srs-delivery.service]=false restore_service_states || exit 1 ''') calls = self.calls.read_text() for expected in ( 'systemctl enable postsrsd-srs.service', 'systemctl enable --runtime srs-reverse-map.service', 'systemctl disable postfix@srs-delivery.service', 'systemctl start postsrsd-srs.service', 'systemctl stop postfix@srs-delivery.service', ): self.assertIn(expected, calls) def test_restore_rejects_invalid_snapshot_before_actions(self): self.assert_exit(''' SERVICE_WAS_ENABLED[postsrsd-srs.service]=enabled SERVICE_WAS_ACTIVE[postsrsd-srs.service]=true SERVICE_WAS_ENABLED[srs-reverse-map.service]=bad-value SERVICE_WAS_ACTIVE[srs-reverse-map.service]=true restore_service_states; [[ $? -eq 1 ]] || exit 1 [[ ! -e "$CALLS" ]] || exit 2 ''') def test_restore_propagates_action_failure(self): self.assert_exit(''' for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do SERVICE_WAS_ENABLED["$unit"]=enabled SERVICE_WAS_ACTIVE["$unit"]=true done FAIL_ACTION=start restore_service_states; [[ $? -eq 1 ]] ''') def test_ss_matching_nonmatching_and_command_failure(self): self.assert_exit(''' SS_DATA='LISTEN 0 128 127.0.0.1:10003 0.0.0.0:*' port_is_listening 127.0.0.1 10003; [[ $? -eq 0 ]] || exit 1 port_is_listening 127.0.0.1 10004; [[ $? -eq 1 ]] || exit 2 SS_FAIL=1 port_is_listening 127.0.0.1 10003; [[ $? -eq 2 ]] || exit 3 ''') def test_master_service_present_absent_and_read_error(self): self.assert_exit(''' master_srs_forward_present; [[ $? -eq 1 ]] || exit 1 printf 'srs-forward unix - n n - - pipe\\n argv=/usr/local/sbin/srs-forward.py\\n' > "$POSTFIX_MASTER_CF" master_srs_forward_present; [[ $? -eq 0 ]] || exit 2 rm "$POSTFIX_MASTER_CF" master_srs_forward_present 2>/dev/null; [[ $? -eq 2 ]] || exit 3 ''') def verification_script(self): return FINAL + '\n[[ "$verification_ok" == "${EXPECTED}" ]]\n' def verification_config(self): return ''' check_managed_block_markers() { return 1; } master_srs_forward_present() { return "${MASTER_RC:-1}"; } state_main_cf_matches() { return 0; } state_master_option_matches() { return 0; } srs_alias_tool() { echo absent; } service_active() { return "${ACTIVE_RC:-1}"; } service_enabled() { return "${ENABLED_RC:-1}"; } port_is_listening() { return "${PORT_RC:-1}"; } postfix() { return 0; } POSTFIX_MAIN_CF="${CALLS}.main" POSTFIX_MASTER_CF="${CALLS}.master" POSTFIX_SRS_DROPIN_FILE="${CALLS}.dropin1" SRS_DELIVERY_DROPIN_FILE="${CALLS}.dropin2" SRS_MAIN_CF_BEGIN_MARKER='begin' SRS_MAIN_CF_END_MARKER='end' SRS_MASTER_CF_BEGIN_MARKER='begin' SRS_MASTER_CF_END_MARKER='end' _PRE_SRS_DEFAULT_TRANSPORT_SET=false _PRE_SRS_DEFAULT_TRANSPORT='' _PRE_SRS_RECIPIENT_CANONICAL_MAPS_SET=false _PRE_SRS_RECIPIENT_CANONICAL_MAPS='' _PRE_SRS_SRS_FORWARD_DESTINATION_RECIPIENT_LIMIT_SET=false _PRE_SRS_SRS_FORWARD_DESTINATION_RECIPIENT_LIMIT='' _PRE_SRS_SMTP_RECEIVE_OVERRIDE_OPTIONS_SET=false _PRE_SRS_SMTP_RECEIVE_OVERRIDE_OPTIONS='' _PRE_SRS_AMAVIS_RECEIVE_OVERRIDE_OPTIONS_SET=false _PRE_SRS_AMAVIS_RECEIVE_OVERRIDE_OPTIONS='' POSTSRSD_SOCKET_HOST=127.0.0.1 POSTSRSD_SOCKET_PORT=10003 SRS_REVERSE_SOCKET_HOST=127.0.0.1 SRS_REVERSE_SOCKET_PORT=10004 SRS_REINJECT_HOST=127.0.0.1 SRS_REINJECT_PORT=10028 SRS_REVERSE_HOST=127.0.0.1 SRS_REVERSE_PORT=10029 ''' def test_final_verification_accepts_only_known_absence(self): for var in ('MASTER_RC', 'ACTIVE_RC', 'ENABLED_RC', 'PORT_RC'): with self.subTest(var=var): self.assert_exit(self.verification_script(), config=self.verification_config() + f'EXPECTED=false\n{var}=2\n') self.assert_exit(self.verification_script(), config=self.verification_config() + 'EXPECTED=true\n') def test_disabled_requires_every_service_and_listener_absent(self): """Mixed runtime states must never be classified as DISABLED.""" mock = ''' service_exists() { return 0; } service_active() { [[ "$1" == "${ACTIVE_UNIT:-}" ]]; } service_enabled() { [[ "$1" == "${ENABLED_UNIT:-}" ]]; } port_is_listening() { [[ "$2" == "${LISTENER_PORT:-}" ]]; } systemd_dropins_match_expected() { return 1; } POSTFIX_SRS_DROPIN_FILE="$CALLS.dropin1" SRS_DELIVERY_DROPIN_FILE="$CALLS.dropin2" POSTSRSD_SOCKET_HOST=127.0.0.1 POSTSRSD_SOCKET_PORT=10003 SRS_REVERSE_SOCKET_HOST=127.0.0.1 SRS_REVERSE_SOCKET_PORT=10004 SRS_REINJECT_HOST=127.0.0.1 SRS_REINJECT_PORT=10028 SRS_REVERSE_HOST=127.0.0.1 SRS_REVERSE_PORT=10029 main_block_status=absent master_block_status=absent pre_srs_main_ok=true pre_srs_master_ok=true master_srs_forward_ok=false main_srs_values_ok=false smtp_override_ok=false amavis_override_ok=false ''' test_body = PREFLIGHT + STATE_CLASSIFICATION + '\nprintf "%s\\n" "$srs_state"\n' for setting, expected in ( ('', 'disabled'), ('ACTIVE_UNIT=postsrsd-srs.service\n', 'inconsistent'), ('ENABLED_UNIT=srs-reverse-map.service\n', 'inconsistent'), ('LISTENER_PORT=10003\n', 'inconsistent'), ): with self.subTest(setting=setting or 'none'): result = self.run_shell(test_body, config=mock + setting) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip().splitlines()[-1], expected) def test_initial_preflight_aborts_on_query_errors(self): config = ''' service_exists() { return 0; } service_active() { return "${ACTIVE_RC:-1}"; } service_enabled() { return "${ENABLED_RC:-1}"; } port_is_listening() { return "${PORT_RC:-1}"; } systemd_dropins_match_expected() { return 1; } POSTFIX_SRS_DROPIN_FILE="$CALLS.none1" SRS_DELIVERY_DROPIN_FILE="$CALLS.none2" POSTSRSD_SOCKET_HOST=127.0.0.1 POSTSRSD_SOCKET_PORT=10003 SRS_REVERSE_SOCKET_HOST=127.0.0.1 SRS_REVERSE_SOCKET_PORT=10004 SRS_REINJECT_HOST=127.0.0.1 SRS_REINJECT_PORT=10028 SRS_REVERSE_HOST=127.0.0.1 SRS_REVERSE_PORT=10029 ''' for variable in ('ACTIVE_RC', 'ENABLED_RC', 'PORT_RC'): with self.subTest(variable=variable): self.assert_exit(PREFLIGHT, rc=70, config=config + f'{variable}=2\n') self.assert_exit(PREFLIGHT, config=config) self.assert_exit(PREFLIGHT, rc=70, config=config + ''' POSTFIX_SRS_DROPIN_FILE="$CALLS.bad-dropin" SRS_DELIVERY_DROPIN_FILE="$CALLS.other-dropin" touch "$POSTFIX_SRS_DROPIN_FILE" ''') if __name__ == '__main__': unittest.main(verbosity=2)