364 lines
13 KiB
Python
364 lines
13 KiB
Python
#!/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)
|