install/mailsystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add selective Postfix SRS installation and management

Browse Source
This commit is contained in:
chris
2026-10-25 22:20:12 +01:00
parent e732e6d9d8
commit efaa58fcf9
28 changed files with 2325 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_activation_status_mock.py
+275
View File
@@ -0,0 +1,275 @@
#!/usr/bin/env python3
"""Isolated status, preflight, snapshot and rollback tests for activate-srs.sh.
All systemctl and ss operations are mocked. No real services or Postfix
configuration are modified. Run from the repository root:
python3 tests/test_activation_status_mock.py
"""
import re
import subprocess
import tempfile
import unittest
from pathlib import Path
SOURCE = Path(__file__).resolve().parents[1] / 'activate-srs.sh'
TEXT = SOURCE.read_text(encoding='utf-8')
def function(name):
match = re.search(rf'^{re.escape(name)}\(\) \{{\n.*?^\}}', TEXT, re.M | re.S)
if not match:
raise ValueError(f'Missing function {name}')
return match.group(0)
FUNCTIONS = '\n'.join(function(n) for n in (
'master_srs_forward_present', 'port_is_listening', 'service_exists',
'read_srs_service_state', 'service_active', 'service_enabled_state',
'snapshot_service_states', 'restore_service_states', 'rollback_activation',
))
def section(start, end):
return TEXT[TEXT.index(start):TEXT.index(end, TEXT.index(start))]
PREFLIGHT = section('services_exist=true\n', '# ----------\n# Determine main.cf state')
CLASSIFICATION = section('srs_integration_enabled=false\n', '# ----------\n# Summary')
class ActivationMockTests(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory(prefix='srs-activation-status-')
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
self.master = self.root / 'master.cf'
self.master.write_text('smtp inet n - y - - smtpd\n')
self.calls = self.root / 'calls'
self.log = self.root / 'log'
self.drop1 = self.root / 'postfix.srs.conf'
self.drop2 = self.root / 'delivery.srs.conf'
def run_shell(self, commands, *, extra='', include_funcs=True):
sh = f'''#!/bin/bash
set -u
POSTFIX_MASTER_CF={str(self.master)!r}
POSTFIX_SRS_DROPIN_FILE={str(self.drop1)!r}
SRS_DELIVERY_DROPIN_FILE={str(self.drop2)!r}
POSTSRSD_SERVICE=postsrsd-srs.service
SRS_REVERSE_MAP_SERVICE=srs-reverse-map.service
SRS_DELIVERY_SERVICE=postfix@srs-delivery.service
POSTFIX_SERVICE=postfix.service
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
CALLS={str(self.calls)!r}
log_file={str(self.log)!r}
ACTIVATION_CHANGES_STARTED=true
ACTIVATION_BACKUP_DIR={str(self.root)!r}
declare -A SERVICE_WAS_ACTIVE=() SERVICE_WAS_ENABLED=()
blank_line() {{ :; }}
warn() {{ :; }}
error() {{ :; }}
info() {{ :; }}
echononl() {{ :; }}
echo_ok() {{ :; }}
echo_failed() {{ :; }}
postfix() {{ return 0; }}
restore_systemd_dropins() {{ return 0; }}
systemctl() {{
printf 'systemctl %s\\n' "$*" >> "$CALLS"
case "$1" in
cat) return 0 ;;
show)
[[ "${{SHOW_FAIL_UNIT:-}}" == "$2" ]] && return 4
if [[ "${{BAD_STATE_UNIT:-}}" == "$2" ]]; then
printf 'LoadState=loaded\\nActiveState=failed\\nUnitFileState=enabled\\n'
elif [[ "${{ACTIVE_UNIT:-}}" == "$2" ]]; then
printf 'LoadState=loaded\\nActiveState=active\\nUnitFileState=enabled\\n'
elif [[ "${{ENABLED_UNIT:-}}" == "$2" ]]; then
printf 'LoadState=loaded\\nActiveState=inactive\\nUnitFileState=enabled\\n'
else
printf '%s\\n' "${{SHOW_DATA:-$'LoadState=loaded\\nActiveState=inactive\\nUnitFileState=disabled'}}"
fi ;;
is-active) [[ "${{MAIN_INACTIVE:-0}}" != 1 ]] ;;
enable|disable|stop|start|reload)
[[ "${{FAIL_ACTION:-}}" != "$1" ]] ;;
*) return 88 ;;
esac
}}
ss() {{
[[ "${{SS_FAIL:-0}}" == 1 ]] && return 7
printf '%s\\n' "${{SS_DATA:-}}"
}}
{FUNCTIONS if include_funcs else ''}
{extra}
{commands}
'''
return subprocess.run(['bash','-c',sh],capture_output=True,text=True,timeout=10)
def check(self, commands, extra=''):
result = self.run_shell(commands, extra=extra)
self.assertEqual(result.returncode, 0, f'{result.stdout}\n{result.stderr}')
return result
def test_valid_disabled_preflight(self):
self.check(PREFLIGHT + '''
[[ "$services_disabled" == true && "$listeners_absent" == true && "$dropins_absent" == true && "$services_status_known" == true && "$primary_postfix_active" == true ]]
''')
def test_active_srs_service_is_not_disabled(self):
self.check(PREFLIGHT + '[[ "$services_disabled" == false ]]',
extra='ACTIVE_UNIT=postsrsd-srs.service')
def test_enabled_but_inactive_srs_service_is_not_disabled(self):
self.check(PREFLIGHT + '[[ "$services_disabled" == false ]]',
extra='ENABLED_UNIT=postsrsd-srs.service')
def test_systemctl_query_error_marks_unknown(self):
self.check(PREFLIGHT + '[[ "$services_status_known" == false && "$services_disabled" == false ]]',
extra='SHOW_FAIL_UNIT=srs-reverse-map.service')
def test_transient_systemctl_state_marks_unknown(self):
self.check(PREFLIGHT + '[[ "$services_status_known" == false ]]',
extra='BAD_STATE_UNIT=postsrsd-srs.service')
def test_active_port_not_disabled(self):
self.check(PREFLIGHT + '[[ "$listeners_absent" == false ]]',
extra="SS_DATA='LISTEN 0 5 127.0.0.1:10003 0.0.0.0:*'")
def test_ss_failure_marks_unknown(self):
self.check(PREFLIGHT + '[[ "$listeners_status_known" == false && "$listeners_absent" == false ]]',
extra='SS_FAIL=1')
def test_main_postfix_inactive_is_not_ready(self):
self.check(PREFLIGHT + '[[ "$primary_postfix_active" == false ]]',
extra='MAIN_INACTIVE=1')
def test_stray_dropin_not_disabled(self):
self.drop1.write_text('[Unit]\n')
self.check(PREFLIGHT + '[[ "$dropins_absent" == false ]]')
def test_broken_symlink_dropin_not_disabled(self):
self.drop1.symlink_to(self.root/'not-existing')
self.check(PREFLIGHT + '[[ "$dropins_absent" == false ]]')
def test_master_cf_unreadable_is_query_error(self):
self.master.unlink()
self.check('master_srs_forward_present; [[ $? -eq 2 ]]')
def test_master_cf_absent_and_present(self):
self.check('master_srs_forward_present; [[ $? -eq 1 ]]')
self.master.write_text('srs-forward unix - n n - - pipe\n argv=/usr/local/sbin/srs-forward.py ${sender}\n')
self.check('master_srs_forward_present; [[ $? -eq 0 ]]')
def test_atomic_snapshot_disabled(self):
self.check('''snapshot_service_states || exit 1
for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
[[ "${SERVICE_WAS_ACTIVE[$unit]}" == false ]] || exit 2
[[ "${SERVICE_WAS_ENABLED[$unit]}" == disabled ]] || exit 3
done''')
self.check('''SHOW_FAIL_UNIT=srs-reverse-map.service
snapshot_service_states; [[ $? -eq 1 ]] || exit 1
[[ "${#SERVICE_WAS_ACTIVE[@]}" -eq 0 ]] || exit 2''')
def test_snapshot_rejects_changed_state(self):
self.check('''ACTIVE_UNIT=postsrsd-srs.service
snapshot_service_states; [[ $? -eq 1 ]] || exit 1
[[ "${#SERVICE_WAS_ENABLED[@]}" -eq 0 ]] || exit 2''')
def test_restore_rejects_invalid_snapshot_without_actions(self):
self.check('''for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
SERVICE_WAS_ACTIVE[$unit]=false
SERVICE_WAS_ENABLED[$unit]=disabled
done
SERVICE_WAS_ENABLED[$SRS_REVERSE_MAP_SERVICE]=masked
restore_service_states; [[ $? -eq 1 ]]''')
self.assertFalse(self.calls.exists() and any(
part in self.calls.read_text() for part in ('systemctl start ', 'systemctl stop ', 'systemctl enable ', 'systemctl disable ')
))
def test_restore_respects_enabled_runtime(self):
self.check('''for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
SERVICE_WAS_ACTIVE[$unit]=false
SERVICE_WAS_ENABLED[$unit]=disabled
done
SERVICE_WAS_ACTIVE[$POSTSRSD_SERVICE]=true
SERVICE_WAS_ENABLED[$POSTSRSD_SERVICE]=enabled-runtime
restore_service_states''')
self.assertIn('systemctl enable --runtime postsrsd-srs.service',self.calls.read_text())
def test_rollback_inactive_primary_postfix_is_failure(self):
(self.root/'main.cf').write_text('before\n')
(self.root/'master.cf').write_text('before\n')
self.check('''POSTFIX_MAIN_CF=${ACTIVATION_BACKUP_DIR}/working-main.cf
POSTFIX_MASTER_CF=${ACTIVATION_BACKUP_DIR}/working-master.cf
for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
SERVICE_WAS_ACTIVE[$unit]=false
SERVICE_WAS_ENABLED[$unit]=disabled
done
MAIN_INACTIVE=1
rollback_activation; [[ $? -eq 1 ]]''')
self.assertNotIn('systemctl reload postfix.service',self.calls.read_text())
def test_enabled_classification_still_works(self):
baseline = '''main_block_status=present
master_block_status=present
master_srs_forward_known=true
main_srs_values_ok=true
master_srs_forward_ok=true
smtp_override_ok=true
amavis_override_ok=true
pre_srs_main_ok=false
pre_srs_master_ok=false
installation_ok=true
services_exist=true
services_status_known=true
services_disabled=false
services_enabled=true
services_active=true
primary_postfix_active=true
listeners_status_known=true
listeners_absent=false
listeners_ok=true
dropins_absent=false
'''
self.check(CLASSIFICATION + '[[ "$srs_state" == enabled ]]', extra=baseline)
def test_disabled_classification_requires_every_prerequisite(self):
baseline = '''main_block_status=absent
master_block_status=absent
master_srs_forward_known=true
main_srs_values_ok=false
master_srs_forward_ok=false
smtp_override_ok=false
amavis_override_ok=false
pre_srs_main_ok=true
pre_srs_master_ok=true
installation_ok=true
services_exist=true
services_status_known=true
services_disabled=true
services_enabled=false
services_active=false
primary_postfix_active=true
listeners_status_known=true
listeners_absent=true
listeners_ok=false
dropins_absent=true
'''
self.check(CLASSIFICATION+'[[ "$srs_state" == disabled ]]',extra=baseline)
for name in ('services_status_known','services_disabled','primary_postfix_active',
'listeners_status_known','listeners_absent','dropins_absent',
'master_srs_forward_known'):
with self.subTest(name=name):
self.check(CLASSIFICATION+'[[ "$srs_state" == inconsistent ]]',
extra=baseline+f'{name}=false\n')
if __name__ == '__main__':
unittest.main(verbosity=2)
tests/test_deactivation_managed_block_mock.py
+79
View File
@@ -0,0 +1,79 @@
#!/usr/bin/env python3
"""Isolated byte-exact managed-block removal tests; no Postfix services used.
Run from repository: python3 tests/test_deactivation_managed_block_mock.py
"""
import re
import subprocess
import tempfile
import unittest
from pathlib import Path
SOURCE = Path(__file__).resolve().parents[1] / "deactivate-srs.sh"
NAME = "remove_managed_block"
BEGIN = "# BEGIN OOPEN-SRS-MANAGED"
END = "# END OOPEN-SRS-MANAGED"
BLOCK = "\n" + BEGIN + "\n# managed block\n" + END + "\n"
def extract_function():
source = SOURCE.read_text(encoding="utf-8")
m = re.search(rf"^{NAME}\(\) \{{\n.*?^\}}", source, re.M | re.S)
if not m:
raise RuntimeError(f"Missing {NAME}() in {SOURCE}")
return m.group(0)
class ManagedBlockMockTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.function = extract_function()
def run_remove(self, data, begin=BEGIN, end=END):
with tempfile.TemporaryDirectory(prefix="srs-managed-block-") as temp:
path = Path(temp) / "main.cf"
path.write_bytes(data)
# Call the real Bash function, which executes its embedded Python.
script = (self.function + "\n" +
'remove_managed_block "$1" "$2" "$3"\n')
result = subprocess.run(
["bash", "-c", script, "bash", str(path), begin, end],
capture_output=True, text=True, timeout=10,
)
return result, path.read_bytes()
def test_exact_restoration_for_zero_to_five_existing_blank_lines(self):
for blanks in range(6):
with self.subTest(original_blank_lines=blanks):
original = b"myhostname = e.mx.oopen.de\n" + b"\n" * blanks
result, after = self.run_remove(original + BLOCK.encode())
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(after, original)
def test_empty_file_is_restored(self):
result, after = self.run_remove(BLOCK.encode())
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(after, b"")
def test_missing_begin_marker_does_not_modify_file(self):
original = b"myhostname = e.mx.oopen.de\n" + ("\n# wrong begin\n" + END + "\n").encode()
result, after = self.run_remove(original)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(after, original)
def test_duplicate_begin_marker_does_not_modify_file(self):
original = ("myhostname = e.mx.oopen.de\n" + BLOCK + BLOCK).encode()
result, after = self.run_remove(original)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(after, original)
def test_reversed_markers_does_not_modify_file(self):
original = (END + "\n" + BEGIN + "\n").encode()
result, after = self.run_remove(original)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(after, original)
if __name__ == "__main__":
unittest.main(verbosity=2)
tests/test_deactivation_rollback_mock.py
+180
View File
@@ -0,0 +1,180 @@
#!/usr/bin/env python3
"""Isolated tests of functions extracted from deactivate-srs.sh.
No mail services are started/stopped and no files outside TemporaryDirectory
are modified. Commands such as systemctl and postfix are shell mocks.
Run from the repository:
python3 tests/test_deactivation_rollback_mock.py
"""
import argparse
import re
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
SOURCE = Path(__file__).resolve().parents[1] / "deactivate-srs.sh"
FUNCTIONS = (
"rollback_deactivation",
"wait_for_srs_forward_drain",
)
def extract_functions(source):
result = {}
for name in FUNCTIONS:
match = re.search(rf"^{re.escape(name)}\(\) \{{\n.*?^\}}", source, re.M | re.S)
if not match:
raise ValueError(f"Function {name}() not found in supplied deactivate-srs.sh")
result[name] = match.group(0)
return result
class IsolatedShellTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.functions = extract_functions(SOURCE.read_text(encoding="utf-8"))
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix="srs-rollback-mock-")
self.addCleanup(self.tmp.cleanup)
self.base = Path(self.tmp.name)
self.backup = self.base / "backup"
self.backup.mkdir()
(self.backup / "main.cf").write_text("original main\n")
(self.backup / "master.cf").write_text("original master\n")
self.main = self.base / "main.cf"
self.master = self.base / "master.cf"
self.main.write_text("modified main\n")
self.master.write_text("modified master\n")
self.calls = self.base / "calls"
self.log = self.base / "log"
def run_bash(self, *, function, setup="", fail="", run_args=""):
self.assertIn(function, self.functions)
script = f'''#!/bin/bash
set -u
POSTFIX_MAIN_CF={str(self.main)!r}
POSTFIX_MASTER_CF={str(self.master)!r}
DEACTIVATION_BACKUP_DIR={str(self.backup)!r}
POSTFIX_SERVICE=postfix.service
log_file={str(self.log)!r}
CALLS={str(self.calls)!r}
DEACTIVATION_CHANGES_STARTED=true
blank_line() {{ :; }}
warn() {{ :; }}
error() {{ :; }}
info() {{ :; }}
echononl() {{ :; }}
echo_ok() {{ :; }}
echo_failed() {{ :; }}
cp() {{
printf 'cp %s\\n' "${{@: -1}}" >> "$CALLS"
if [[ "${{FAIL_CP_MAIN:-0}}" == 1 && "${{@: -1}}" == "$POSTFIX_MAIN_CF" ]]; then return 1; fi
command cp "$@"
}}
restore_systemd_dropins() {{
echo restore-dropins >> "$CALLS"
[[ "${{FAIL_DROPINS:-0}}" != 1 ]]
}}
restore_service_states() {{
echo restore-services >> "$CALLS"
[[ "${{FAIL_SERVICES:-0}}" != 1 ]]
}}
postfix() {{
printf 'postfix %s\\n' "$*" >> "$CALLS"
[[ "${{FAIL_POSTFIX_CHECK:-0}}" != 1 ]]
}}
systemctl() {{
printf 'systemctl %s\\n' "$*" >> "$CALLS"
case "$1" in
is-active) [[ "${{MAIN_INACTIVE:-0}}" != 1 ]] ;;
reload) [[ "${{FAIL_RELOAD:-0}}" != 1 ]] ;;
*) return 1 ;;
esac
}}
srs_old_queue_manager_is_running() {{ return "${{QMGR_RC:-1}}"; }}
srs_forward_worker_is_running() {{ return "${{WORKER_RC:-1}}"; }}
srs_delivery_queue_is_empty() {{ return "${{QUEUE_RC:-0}}"; }}
sleep() {{ SECONDS=$((SECONDS + 181)); }}
{self.functions[function]}
{setup}
{fail}
{function} {run_args}
'''
return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
timeout=10)
def test_rollback_restores_config_and_reloads(self):
result = self.run_bash(function="rollback_deactivation")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.main.read_text(), "original main\n")
self.assertEqual(self.master.read_text(), "original master\n")
calls = self.calls.read_text()
for expected in ("restore-dropins", "restore-services", "postfix check",
"systemctl reload postfix.service"):
self.assertIn(expected, calls)
def test_rollback_cp_error_is_reported(self):
result = self.run_bash(function="rollback_deactivation", fail="FAIL_CP_MAIN=1")
self.assertNotEqual(result.returncode, 0)
self.assertEqual(self.main.read_text(), "modified main\n")
self.assertEqual(self.master.read_text(), "original master\n")
def test_rollback_dropin_error_is_reported(self):
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
fail="FAIL_DROPINS=1").returncode, 0)
def test_rollback_service_restore_error_is_reported(self):
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
fail="FAIL_SERVICES=1").returncode, 0)
def test_rollback_postfix_check_error_is_reported(self):
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
fail="FAIL_POSTFIX_CHECK=1").returncode, 0)
def test_rollback_reload_error_is_reported(self):
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
fail="FAIL_RELOAD=1").returncode, 0)
def test_rollback_inactive_main_postfix_is_not_silently_successful(self):
"""Safety regression: an inactive primary Postfix must not mean success."""
result = self.run_bash(function="rollback_deactivation", fail="MAIN_INACTIVE=1")
self.assertNotEqual(result.returncode, 0,
"BUG: inactive primary Postfix incorrectly reported rollback success")
self.assertNotIn("systemctl reload postfix.service", self.calls.read_text())
def test_drain_succeeds_when_qmgr_workers_and_queue_clear(self):
result = self.run_bash(function="wait_for_srs_forward_drain",
setup="sleep() { :; }")
self.assertEqual(result.returncode, 0, result.stderr)
def test_drain_times_out_on_running_old_qmgr(self):
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
fail="QMGR_RC=0").returncode, 1)
def test_drain_rejects_process_check_error(self):
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
fail="WORKER_RC=2").returncode, 2)
def test_drain_rejects_queue_read_error(self):
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
fail="QUEUE_RC=2").returncode, 2)
def test_drain_times_out_if_queue_remains_nonempty(self):
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
fail="QUEUE_RC=1").returncode, 1)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source", type=Path, default=SOURCE,
help="Path to deactivate-srs.sh (defaults to sibling repo script)")
args, unittest_args = parser.parse_known_args()
SOURCE = args.source.resolve()
if not SOURCE.is_file():
parser.error(f"Source script not found: {SOURCE}")
unittest.main(argv=[sys.argv[0], *unittest_args], verbosity=2)
tests/test_deactivation_status_mock.py
+363
View File
@@ -0,0 +1,363 @@
#!/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)
tests/test_installer_mock.py
Executable
+263
View File
@@ -0,0 +1,263 @@
#!/usr/bin/env python3
"""Offline, path-relocated installer contract tests. All external commands mocked.
Do not equate these tests with genuine Postfix/PostSRSd integration.
"""
from pathlib import Path
import os, subprocess, tempfile, zipfile, re, sys, shutil
SOURCE = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else Path(__file__).resolve().parents[1]
def putexe(file, content):
file.parent.mkdir(parents=True, exist_ok=True)
file.write_text('#!/usr/bin/env bash\n' + content)
file.chmod(0o755)
def setup(root):
root.mkdir(parents=True, exist_ok=True)
repo = root / 'repo'
repo.mkdir()
if SOURCE.is_file():
with zipfile.ZipFile(SOURCE) as z:
z.extractall(repo)
else:
for source_file in SOURCE.rglob('*'):
if source_file.is_file():
dst = repo / source_file.relative_to(SOURCE)
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_file, dst)
for path in (repo/'install-update-srs.sh', repo/'helpers/srs-common.sh',
*list((repo/'files/srs').glob('*.py')),
*list((repo/'files/srs/templates').glob('*'))):
s = path.read_text()
# Keep all managed install/config/queue paths in an unprivileged fixture tree.
# The postconf absolute command must also be relocated.
s = s.replace('/usr/sbin/postconf', str(root/'bin/postconf'))
for prefix in ['/etc/', '/usr/local/', '/var/', '/run/lock/', '/root/srs-installer-backups']:
new = str(root) + prefix
s = s.replace(prefix, new)
# prevent reading the real original root Postfix config in our fixtures
path.write_text(s)
(repo/'install-update-srs.sh').chmod(0o755)
for p in ['etc/postfix','usr/local/sbin','run/lock','var/spool','var/lib','etc/systemd/system','flags']:
(root/p).mkdir(parents=True, exist_ok=True)
(root/'etc/postfix/main.cf').write_text('myhostname = mail.example.org\n')
(root/'etc/postfix/master.cf').write_text('smtp inet n - y - - smtpd\nlocalhost:10025 inet n - y - - smtpd\n -o receive_override_options=no_unknown_recipient_checks\n')
(repo/'conf').mkdir(exist_ok=True)
# All expected lookup paths are extracted from the (relocated) reference source.
alias = (repo/'files/srs/srs-alias-decision.py').read_text()
for typ,name in set(re.findall(r'"(pgsql|btree):([^" ]+)"', alias)):
f = Path(name + ('.db' if typ == 'btree' else ''))
f.parent.mkdir(parents=True,exist_ok=True)
f.touch()
bin = root/'bin'
bin.mkdir()
putexe(bin/'postconf', '''
name="${@: -1}"
# e.mx.oopen.de has two pre-existing unrelated warnings in main.cf.
# They are emitted on every postconf call while the requested value is valid.
echo "postconf: warning: ${SRS_TEST_ROOT}/etc/postfix/main.cf: unused parameter: policyd-spf_time_limit=3600" >&2
# The warning path matches the source main.cf in the relocated fixture.
# Existing warnings must be distinguished from queried-param errors.
if [[ -f "${SRS_TEST_ROOT}/flags/unknown_parameter_warning" ]]; then
echo "postconf: warning: smtp_tls_cafile: unknown parameter" >&2
fi
case "$name" in
myhostname) echo mail.example.org;;
mydomain) echo example.org;;
myorigin) echo /etc/mailname;;
mydestination) echo 'mail.example.org localhost';;
relay_domains) echo '\u0024mydestination';;
virtual_alias_maps) echo 'pgsql:some-alias';;
virtual_alias_domains) echo 'btree:some-domain';;
virtual_mailbox_maps) echo 'pgsql:some-mailbox';;
virtual_mailbox_domains) echo 'pgsql:some-domain';;
smtp_bind_address) echo 192.0.2.22;;
smtp_bind_address6) echo 2001:db8::22;;
smtp_dns_support_level) echo dnssec;;
smtp_host_lookup) echo dns;;
smtp_tls_CAfile) echo /etc/ssl/certs/ca-certificates.crt;;
smtp_tls_mandatory_protocols|smtp_tls_protocols) echo '>=TLSv1.2';;
smtp_tls_security_level) echo dane;;
*) echo "unknown $name" >&2; exit 1;;
esac
''')
putexe(bin/'systemctl', '''
if [[ "$1" == is-active ]]; then exit 3; fi
exit 0
''')
putexe(bin/'git', '''
if [[ "$1" == clone ]]; then
mkdir -p "${@: -1}"; exit 0
fi
exec /usr/bin/git "$@"
''')
putexe(bin/'cmake', '''
if [[ "$1" == --version ]]; then echo 'cmake version 3.31.6'; exit 0; fi
if [[ "$1" == -S ]]; then
for ((i=1; i<=$#; i++)); do
arg="${!i}"
if [[ "$arg" == -B ]]; then n=$((i+1)); build="${!n}"; fi
if [[ "$arg" == -DCMAKE_INSTALL_PREFIX=* ]]; then prefix="${arg#*=}"; fi
done
mkdir -p "$build"; printf '%s\\n' "$prefix" > "$build/prefix"; exit 0
fi
if [[ "$1" == --build ]]; then exit 0; fi
if [[ "$1" == --install ]]; then
prefix="$(cat "$2/prefix")"
mkdir -p "${DESTDIR}${prefix}/sbin"
printf '#!/bin/sh\\nexit 0\\n' > "${DESTDIR}${prefix}/sbin/postsrsd"
chmod +x "${DESTDIR}${prefix}/sbin/postsrsd"
exit 0
fi
exit 9
''')
putexe(bin/'ctest','exit 0\n')
putexe(bin/'dig','exit 0\n')
putexe(bin/'postmap','''
case "$1" in
btree:*) dest="${1#btree:}"; touch "${dest}.db";;
*) exit 1;;
esac
if [[ -f "${SRS_TEST_ROOT}/flags/fail_postmap" ]]; then exit 1; fi
exit 0
''')
putexe(bin/'postfix','''
if [[ "$*" == *check* ]]; then
mkdir -p "${SRS_TEST_ROOT}/var/spool/postfix-srs-delivery/private" "${SRS_TEST_ROOT}/var/lib/postfix-srs-delivery"
if [[ -f "${SRS_TEST_ROOT}/flags/fail_postfix" ]]; then echo 'forced postfix check error' >&2; exit 1; fi
fi
exit 0
''')
(repo/'helpers/srs-dns-check.py').write_text('print(" Simulated authoritative DNS: OK")\n')
return repo
def run(root, repo, *, fail=None, version=None):
if fail: (root/'flags'/('fail_'+fail)).touch()
cmd = f"{repo/'install-update-srs.sh'} --install"
if version: cmd += f' --version {version}'
env = os.environ.copy()
env.update({'PATH':str(root/'bin') + ':' + env['PATH'], 'SRS_TEST_ROOT':str(root), 'LC_ALL':'C'})
# Running inside util-linux script supplies a real PTY for init_srs_runtime.
result = subprocess.run(['script','-q','-e','-c',cmd,'/dev/null'], input='srs.example.org\nyes\nyes\n',
text=True, capture_output=True, timeout=35, env=env)
return result
def assert_ok(condition, msg):
if not condition: raise AssertionError(msg)
with tempfile.TemporaryDirectory(prefix='oopen-srs-test-') as tmp:
base=Path(tmp)
os.chmod(base, 0o2770) # Deliberately inherit SGID on fixture subdirectories.
root=base/'initial'
repo=setup(root)
result=run(root,repo)
print('initial install', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'initial installation failed')
assert_ok((root/'usr/local/postsrsd').is_symlink(), 'missing symlink')
assert_ok((root/'usr/local/sbin/srs-forward.py').exists(), 'missing srs forward')
assert_ok((root/'etc/postsrsd-srs/secret').stat().st_mode & 0o777 == 0o600, 'secret mode')
assert_ok('smtp_tls_CAfile = /etc/ssl/certs/ca-certificates.crt' in (root/'etc/postfix-srs/delivery/main.cf').read_text(),'rendered TLS CA')
print(" secret dir mode:", oct((root/"etc/postsrsd-srs").stat().st_mode & 0o777), "uid", (root/"etc/postsrsd-srs").stat().st_uid)
print(" stat check:", subprocess.check_output(["stat", "-c", "%u:%g:%a", str(root/"etc/postsrsd-srs")], text=True).strip())
print(' initial install: state, secret, config, symlink, CAfile verified')
original_secret=(root/'etc/postsrsd-srs/secret').read_bytes()
original_forward=(root/'usr/local/sbin/srs-forward.py').read_bytes()
original_link=os.readlink(root/'usr/local/postsrsd')
result=run(root,repo)
print('maintenance update', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'maintenance update failed')
assert_ok((root/'etc/postsrsd-srs/secret').read_bytes()==original_secret,'secret changed')
assert_ok(os.readlink(root/'usr/local/postsrsd')==original_link,'link changed')
print(' maintenance update: original secret and binary retained')
# Force late failure, after Postfix configuration and all units have been deployed.
source=repo/'files/srs/srs-forward.py'
source.write_text(source.read_text()+'\n# injected test change\n')
result=run(root,repo,fail='postfix')
print('failed maintenance rollback', result.returncode)
if result.returncode==0: print(result.stdout[-2500:])
assert_ok(result.returncode != 0,'injected failure not detected')
assert_ok((root/'usr/local/sbin/srs-forward.py').read_bytes()==original_forward,'rollback did not restore file')
assert_ok((root/'etc/postsrsd-srs/secret').read_bytes()==original_secret,'rollback changed secret')
assert_ok(os.readlink(root/'usr/local/postsrsd')==original_link,'rollback changed symlink')
print(' failure rollback: file restored; secret and symlink preserved')
(root/'flags/fail_postfix').unlink()
result=run(root,repo,version='2.4.2')
print('version upgrade', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'version upgrade failed')
assert_ok((root/'usr/local/postsrsd-2.4.1').exists(),'prior prefix missing')
assert_ok(os.readlink(root/'usr/local/postsrsd')=='postsrsd-2.4.2','new link incorrect')
print(' version upgrade: previous prefix retained; link updated')
(root/'flags/fail_postfix').touch()
result=run(root,repo,version='2.4.3')
print('failed version upgrade rollback', result.returncode)
assert_ok(result.returncode != 0,'injected failure not detected')
assert_ok(os.readlink(root/'usr/local/postsrsd')=='postsrsd-2.4.2','previous version link not restored')
assert_ok(not (root/'usr/local/postsrsd-2.4.3').exists(),'failed version prefix not removed')
print(' failed version upgrade: symlink restored, new prefix removed')
# Active SRS configuration must prohibit any maintenance update before writes.
(root/'flags/fail_postfix').unlink()
active_marker = '# BEGIN OOPEN-SRS-MANAGED'
existing_main=(root/'etc/postfix/main.cf')
original_main=existing_main.read_text()
existing_main.write_text(original_main + active_marker + '\n')
before=(root/'usr/local/sbin/srs-forward.py').read_bytes()
result=run(root,repo,version='2.4.2')
print('active integration guard', result.returncode)
assert_ok(result.returncode != 0,'active SRS was not refused')
assert_ok((root/'usr/local/sbin/srs-forward.py').read_bytes() == before, 'active check changed scripts')
existing_main.write_text(original_main)
print(' active integration guard: blocked before file changes')
# Independent initial installation with late failure. New SRS files must be removed.
root=base/'failed-first'
repo=setup(root)
result=run(root,repo,fail='postfix')
print('failed first install', result.returncode)
assert_ok(result.returncode != 0,'injected first-install failure not detected')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'first-install rollback left symlink')
assert_ok(not (root/'usr/local/postsrsd-2.4.1').exists(), 'first-install rollback left prefix')
assert_ok(not (root/'etc/postsrsd-srs/secret').exists(), 'first-install rollback left secret')
assert_ok(not (root/'etc/postfix-srs/delivery/main.cf').exists(), 'first-install rollback left config')
assert_ok((root/'var/spool/postfix-srs-delivery/private').is_dir(), 'test Postfix runtime queue unexpectedly deleted')
print(' failed first install: tracked files removed; Postfix runtime directory safely preserved')
# Fail at postmap, before systemd and Postfix checks.
root=base/'failed-map'
repo=setup(root)
result=run(root,repo,fail='postmap')
print('failed postmap rollback', result.returncode)
assert_ok(result.returncode != 0, 'injected postmap failure not detected')
assert_ok(not (root/'etc/postfix-srs/delivery/transport.db').exists(),'failed postmap index remains')
assert_ok(not (root/'etc/postsrsd-srs/secret').exists(),'failed postmap secret remains')
assert_ok(not (root/'usr/local/postsrsd').exists(),'failed postmap link remains')
print(' failed postmap: generated index and secret removed')
# Unexpected warnings must never be silently swallowed.
root=base/'unexpected-warning'
repo=setup(root)
(root/'flags/unknown_parameter_warning').touch()
result=run(root, repo)
print('unexpected postconf warning', result.returncode)
assert_ok(result.returncode != 0, 'unexpected postconf warning accepted')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'unexpected warning allowed installation')
print(' unrelated existing warnings tolerated, queried-parameter warning rejected')
# A template referencing an unknown Postfix parameter must fail closed.
root=base/'bad-template-parameter'
repo=setup(root)
template=root/'repo/files/srs/templates/delivery-main.cf.in'
contents=template.read_text()
assert_ok('smtp_tls_CAfile = @SMTP_TLS_CAFILE@' in contents, 'missing expected template parameter')
template.write_text(contents.replace('smtp_tls_CAfile = @SMTP_TLS_CAFILE@',
'smtp_tls_cafile = @SMTP_TLS_CAFILE@'))
result=run(root, repo)
print('invalid template parameter', result.returncode)
assert_ok(result.returncode != 0, 'unknown template Postfix parameter accepted')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'invalid template allowed installation')
print(' unknown requested Postfix parameter rejected')
print('ALL OFFLINE MOCK CONTRACT TESTS PASSED')
tests/test_installer_mock_safe.py
Executable
+303
View File
@@ -0,0 +1,303 @@
#!/usr/bin/env python3
"""Offline, path-relocated installer contract tests. All external commands mocked.
Do not equate these tests with genuine Postfix/PostSRSd integration.
"""
from pathlib import Path
import os, subprocess, tempfile, re, sys, shutil, signal, types
SOURCE = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else Path(__file__).resolve().parents[1]
def putexe(file, content):
file.parent.mkdir(parents=True, exist_ok=True)
file.write_text('#!/usr/bin/env bash\n' + content)
file.chmod(0o755)
def setup(root):
root.mkdir(parents=True, exist_ok=True)
repo = root / 'repo'
repo.mkdir()
# Copy only public SRS source files needed by the test. Never copy the
# complete mailserver repository, .git, credentials, local conf/srs.conf,
# or unrelated applications into an executable test fixture.
if not SOURCE.is_dir():
raise RuntimeError(f'SOURCE must be a repository directory: {SOURCE}')
required = [
'install-update-srs.sh',
'helpers/srs-common.sh',
'helpers/srs-dns-check.py',
'helpers/srs-virtual-alias.py',
'files/srs/srs-alias-decision.py',
'files/srs/srs-forward.py',
'files/srs/srs-reverse-map.py',
'files/srs/srs-reverse-policy.py',
]
templates = SOURCE/'files/srs/templates'
required += [str(f.relative_to(SOURCE)) for f in templates.iterdir() if f.is_file() and not f.is_symlink()]
for rel in required:
src = SOURCE/rel
if not src.is_file() or src.is_symlink():
raise RuntimeError(f'Missing or symlinked source file: {src}')
dst = repo/rel
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
for path in (repo/'install-update-srs.sh', repo/'helpers/srs-common.sh',
*list((repo/'files/srs').glob('*.py')),
*list((repo/'files/srs/templates').glob('*'))):
s = path.read_text()
# Keep all managed install/config/queue paths in an unprivileged fixture tree.
# The postconf absolute command must also be relocated.
s = s.replace('/usr/sbin/postconf', str(root/'bin/postconf'))
for prefix in ['/etc/', '/usr/local/', '/var/', '/run/lock/', '/root/srs-installer-backups']:
new = str(root) + prefix
s = s.replace(prefix, new)
# prevent reading the real original root Postfix config in our fixtures
if path == repo/'helpers/srs-common.sh':
# The test runner uses a pipe for scripted confirmations;
# allow that only inside the copied, isolated helper fixture.
needle = 'if [[ -t 1 ]]; then'
if s.count(needle) != 1:
raise RuntimeError('Unexpected terminal gate in helper')
s = s.replace(needle, 'if true; then', 1)
path.write_text(s)
(repo/'install-update-srs.sh').chmod(0o755)
for p in ['etc/postfix','usr/local/sbin','run/lock','var/spool','var/lib','etc/systemd/system','flags']:
(root/p).mkdir(parents=True, exist_ok=True)
(root/'etc/postfix/main.cf').write_text('myhostname = mail.example.org\n')
(root/'etc/postfix/master.cf').write_text('smtp inet n - y - - smtpd\nlocalhost:10025 inet n - y - - smtpd\n -o receive_override_options=no_unknown_recipient_checks\n')
(repo/'conf').mkdir(exist_ok=True)
# All expected lookup paths are extracted from the (relocated) reference source.
alias = (repo/'files/srs/srs-alias-decision.py').read_text()
for typ,name in set(re.findall(r'"(pgsql|btree):([^" ]+)"', alias)):
f = Path(name + ('.db' if typ == 'btree' else ''))
f.parent.mkdir(parents=True,exist_ok=True)
f.touch()
bin = root/'bin'
bin.mkdir()
putexe(bin/'postconf', '''
name="${@: -1}"
# e.mx.oopen.de has two pre-existing unrelated warnings in main.cf.
# They are emitted on every postconf call while the requested value is valid.
echo "postconf: warning: ${SRS_TEST_ROOT}/etc/postfix/main.cf: unused parameter: policyd-spf_time_limit=3600" >&2
# The warning path matches the source main.cf in the relocated fixture.
# Existing warnings must be distinguished from queried-param errors.
if [[ -f "${SRS_TEST_ROOT}/flags/unknown_parameter_warning" ]]; then
echo "postconf: warning: smtp_tls_cafile: unknown parameter" >&2
fi
case "$name" in
mail_owner) echo postfix;;
myhostname) echo mail.example.org;;
mydomain) echo example.org;;
myorigin) echo /etc/mailname;;
mydestination) echo 'mail.example.org localhost';;
relay_domains) echo '\u0024mydestination';;
virtual_alias_maps) echo 'pgsql:some-alias';;
virtual_alias_domains) echo 'btree:some-domain';;
virtual_mailbox_maps) echo 'pgsql:some-mailbox';;
virtual_mailbox_domains) echo 'pgsql:some-domain';;
smtp_bind_address) echo 192.0.2.22;;
smtp_bind_address6) echo 2001:db8::22;;
smtp_dns_support_level) echo dnssec;;
smtp_host_lookup) echo dns;;
smtp_tls_CAfile) echo /etc/ssl/certs/ca-certificates.crt;;
smtp_tls_mandatory_protocols|smtp_tls_protocols) echo '>=TLSv1.2';;
smtp_tls_security_level) echo dane;;
*) echo "unknown $name" >&2; exit 1;;
esac
''')
putexe(bin/'id', '''
if [[ "$1" == -u && "$2" == postfix ]]; then echo 123; exit 0; fi
if [[ "$1" == -g && "$2" == postfix ]]; then echo 123; exit 0; fi
exec /usr/bin/id "$@"
''')
putexe(bin/'systemctl', '''
if [[ "$1" == is-active ]]; then exit 3; fi
exit 0
''')
putexe(bin/'git', '''
if [[ "$1" == clone ]]; then
mkdir -p "${@: -1}"; exit 0
fi
exec /usr/bin/git "$@"
''')
putexe(bin/'cmake', '''
if [[ "$1" == --version ]]; then echo 'cmake version 3.31.6'; exit 0; fi
if [[ "$1" == -S ]]; then
for ((i=1; i<=$#; i++)); do
arg="${!i}"
if [[ "$arg" == -B ]]; then n=$((i+1)); build="${!n}"; fi
if [[ "$arg" == -DCMAKE_INSTALL_PREFIX=* ]]; then prefix="${arg#*=}"; fi
done
mkdir -p "$build"; printf '%s\\n' "$prefix" > "$build/prefix"; exit 0
fi
if [[ "$1" == --build ]]; then exit 0; fi
if [[ "$1" == --install ]]; then
prefix="$(cat "$2/prefix")"
mkdir -p "${DESTDIR}${prefix}/sbin"
printf '#!/bin/sh\\nexit 0\\n' > "${DESTDIR}${prefix}/sbin/postsrsd"
chmod +x "${DESTDIR}${prefix}/sbin/postsrsd"
exit 0
fi
exit 9
''')
putexe(bin/'ctest','exit 0\n')
putexe(bin/'dig','exit 0\n')
putexe(bin/'postmap','''
case "$1" in
btree:*) dest="${1#btree:}"; touch "${dest}.db";;
*) exit 1;;
esac
if [[ -f "${SRS_TEST_ROOT}/flags/fail_postmap" ]]; then exit 1; fi
exit 0
''')
putexe(bin/'postfix','''
if [[ "$*" == *check* ]]; then
mkdir -p "${SRS_TEST_ROOT}/var/spool/postfix-srs-delivery/private" "${SRS_TEST_ROOT}/var/lib/postfix-srs-delivery"
if [[ -f "${SRS_TEST_ROOT}/flags/fail_postfix" ]]; then echo 'forced postfix check error' >&2; exit 1; fi
fi
exit 0
''')
(repo/'helpers/srs-dns-check.py').write_text('print(" Simulated authoritative DNS: OK")\n')
return repo
def run(root, repo, *, fail=None, version=None):
if fail: (root/'flags'/('fail_'+fail)).touch()
argv = [str(repo/'install-update-srs.sh'), '--install']
if version: argv.extend(['--version', version])
env = os.environ.copy()
env.update({'PATH':str(root/'bin') + ':' + env['PATH'], 'SRS_TEST_ROOT':str(root), 'LC_ALL':'C'})
# No real PTY is needed: the terminal requirement is disabled only in the
# relocated helper copy, never in the actual repository.
# File-backed capture prevents subprocess pipe hangs from descendant
# processes retaining output fds. A timeout kills the entire fixture group.
with tempfile.TemporaryFile(mode='w+t') as out:
proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=out,
stderr=subprocess.STDOUT, text=True, env=env,
start_new_session=True)
try:
proc.communicate(input='srs.example.org\nyes\nyes\n', timeout=20)
except subprocess.TimeoutExpired:
os.killpg(proc.pid, signal.SIGKILL)
proc.wait(timeout=5)
raise RuntimeError('Mock installer timed out')
out.seek(0)
return types.SimpleNamespace(returncode=proc.returncode, stdout=out.read())
def assert_ok(condition, msg):
if not condition: raise AssertionError(msg)
with tempfile.TemporaryDirectory(prefix='oopen-srs-test-') as tmp:
base=Path(tmp)
os.chmod(base, 0o2700) # Test SGID inheritance without granting group access.
root=base/'initial'
repo=setup(root)
result=run(root,repo)
print('initial install', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'initial installation failed')
assert_ok((root/'usr/local/postsrsd').is_symlink(), 'missing symlink')
assert_ok((root/'usr/local/sbin/srs-forward.py').exists(), 'missing srs forward')
assert_ok((root/'etc/postsrsd-srs/secret').stat().st_mode & 0o777 == 0o600, 'secret mode')
assert_ok('smtp_tls_CAfile = /etc/ssl/certs/ca-certificates.crt' in (root/'etc/postfix-srs/delivery/main.cf').read_text(),'rendered TLS CA')
print(" secret dir mode:", oct((root/"etc/postsrsd-srs").stat().st_mode & 0o777), "uid", (root/"etc/postsrsd-srs").stat().st_uid)
print(" stat check:", subprocess.check_output(["stat", "-c", "%u:%g:%a", str(root/"etc/postsrsd-srs")], text=True).strip())
print(' initial install: state, secret, config, symlink, CAfile verified')
original_secret=(root/'etc/postsrsd-srs/secret').read_bytes()
original_forward=(root/'usr/local/sbin/srs-forward.py').read_bytes()
original_link=os.readlink(root/'usr/local/postsrsd')
result=run(root,repo)
print('maintenance update', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'maintenance update failed')
assert_ok((root/'etc/postsrsd-srs/secret').read_bytes()==original_secret,'secret changed')
assert_ok(os.readlink(root/'usr/local/postsrsd')==original_link,'link changed')
print(' maintenance update: original secret and binary retained')
# Force late failure, after Postfix configuration and all units have been deployed.
source=repo/'files/srs/srs-forward.py'
source.write_text(source.read_text()+'\n# injected test change\n')
result=run(root,repo,fail='postfix')
print('failed maintenance rollback', result.returncode)
if result.returncode==0: print(result.stdout[-2500:])
assert_ok(result.returncode != 0,'injected failure not detected')
assert_ok((root/'usr/local/sbin/srs-forward.py').read_bytes()==original_forward,'rollback did not restore file')
assert_ok((root/'etc/postsrsd-srs/secret').read_bytes()==original_secret,'rollback changed secret')
assert_ok(os.readlink(root/'usr/local/postsrsd')==original_link,'rollback changed symlink')
print(' failure rollback: file restored; secret and symlink preserved')
(root/'flags/fail_postfix').unlink()
result=run(root,repo,version='2.4.2')
print('version upgrade', result.returncode)
if result.returncode: print(result.stdout[-2500:])
assert_ok(result.returncode == 0,'version upgrade failed')
assert_ok((root/'usr/local/postsrsd-2.4.1').exists(),'prior prefix missing')
assert_ok(os.readlink(root/'usr/local/postsrsd')=='postsrsd-2.4.2','new link incorrect')
print(' version upgrade: previous prefix retained; link updated')
(root/'flags/fail_postfix').touch()
result=run(root,repo,version='2.4.3')
print('failed version upgrade rollback', result.returncode)
assert_ok(result.returncode != 0,'injected failure not detected')
assert_ok(os.readlink(root/'usr/local/postsrsd')=='postsrsd-2.4.2','previous version link not restored')
assert_ok(not (root/'usr/local/postsrsd-2.4.3').exists(),'failed version prefix not removed')
print(' failed version upgrade: symlink restored, new prefix removed')
# Active SRS configuration must prohibit any maintenance update before writes.
(root/'flags/fail_postfix').unlink()
active_marker = '# BEGIN OOPEN-SRS-MANAGED'
existing_main=(root/'etc/postfix/main.cf')
original_main=existing_main.read_text()
existing_main.write_text(original_main + active_marker + '\n')
before=(root/'usr/local/sbin/srs-forward.py').read_bytes()
result=run(root,repo,version='2.4.2')
print('active integration guard', result.returncode)
assert_ok(result.returncode != 0,'active SRS was not refused')
assert_ok((root/'usr/local/sbin/srs-forward.py').read_bytes() == before, 'active check changed scripts')
existing_main.write_text(original_main)
print(' active integration guard: blocked before file changes')
# Independent initial installation with late failure. New SRS files must be removed.
root=base/'failed-first'
repo=setup(root)
result=run(root,repo,fail='postfix')
print('failed first install', result.returncode)
assert_ok(result.returncode != 0,'injected first-install failure not detected')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'first-install rollback left symlink')
assert_ok(not (root/'usr/local/postsrsd-2.4.1').exists(), 'first-install rollback left prefix')
assert_ok(not (root/'etc/postsrsd-srs/secret').exists(), 'first-install rollback left secret')
assert_ok(not (root/'etc/postfix-srs/delivery/main.cf').exists(), 'first-install rollback left config')
assert_ok((root/'var/spool/postfix-srs-delivery/private').is_dir(), 'test Postfix runtime queue unexpectedly deleted')
print(' failed first install: tracked files removed; Postfix runtime directory safely preserved')
# Fail at postmap, before systemd and Postfix checks.
root=base/'failed-map'
repo=setup(root)
result=run(root,repo,fail='postmap')
print('failed postmap rollback', result.returncode)
assert_ok(result.returncode != 0, 'injected postmap failure not detected')
assert_ok(not (root/'etc/postfix-srs/delivery/transport.db').exists(),'failed postmap index remains')
assert_ok(not (root/'etc/postsrsd-srs/secret').exists(),'failed postmap secret remains')
assert_ok(not (root/'usr/local/postsrsd').exists(),'failed postmap link remains')
print(' failed postmap: generated index and secret removed')
# Unexpected warnings must never be silently swallowed.
root=base/'unexpected-warning'
repo=setup(root)
(root/'flags/unknown_parameter_warning').touch()
result=run(root, repo)
print('unexpected postconf warning', result.returncode)
assert_ok(result.returncode != 0, 'unexpected postconf warning accepted')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'unexpected warning allowed installation')
print(' unrelated existing warnings tolerated, queried-parameter warning rejected')
# A template referencing an unknown Postfix parameter must fail closed.
root=base/'bad-template-parameter'
repo=setup(root)
template=root/'repo/files/srs/templates/delivery-main.cf.in'
contents=template.read_text()
assert_ok('smtp_tls_CAfile = @SMTP_TLS_CAFILE@' in contents, 'missing expected template parameter')
template.write_text(contents.replace('smtp_tls_CAfile = @SMTP_TLS_CAFILE@',
'smtp_tls_cafile = @SMTP_TLS_CAFILE@'))
result=run(root, repo)
print('invalid template parameter', result.returncode)
assert_ok(result.returncode != 0, 'unknown template Postfix parameter accepted')
assert_ok(not (root/'usr/local/postsrsd').exists(), 'invalid template allowed installation')
print(' unknown requested Postfix parameter rejected')
print('ALL OFFLINE MOCK CONTRACT TESTS PASSED')
tests/test_installer_mx_host_mock.py
+185
View File
@@ -0,0 +1,185 @@
#!/usr/bin/env python3
"""Isolated, non-root tests for the public MX host installer addition.
No Postfix services, network DNS records or on-host config files are changed.
"""
import os
import pathlib
import re
import subprocess
import sys
import tempfile
import unittest
ROOT = pathlib.Path(__file__).resolve().parents[1]
INSTALLER = (ROOT / 'install-update-srs.sh').read_text()
COMMON = (ROOT / 'helpers/srs-common.sh').read_text()
SAMPLE = (ROOT / 'conf/srs.conf.sample').read_text()
def slice_script(source, start, end):
return source[source.index(start):source.index(end)]
PROMPTS = slice_script(INSTALLER, 'read_config_domain_hints() {', '# Consistent installer-only confirmation prompts.')
DNS = slice_script(INSTALLER, 'verify_srs_domain_dns() {', '# Compact display of APT simulations;')
LOAD = slice_script(COMMON, 'load_srs_config() {', '# ----------\n# Persistent pre-SRS state')
MIGRATION = re.search(
r"<<'PY_MX_MIGRATE'[^\n]*\n(.*?)\nPY_MX_MIGRATE", INSTALLER, re.S
)
assert MIGRATION is not None
class InstallerMxHostTests(unittest.TestCase):
def run_prompt(self, config=None, hostname='o13-mail.oopen.de', choices='\n\n'):
with tempfile.TemporaryDirectory() as temp:
conf = pathlib.Path(temp) / 'srs.conf'
if config is not None:
conf.write_text(config)
script = f'''SRS_CONF_FILE={str(conf)!r}
CONFIG_DOMAIN_HINT=''
CONFIG_MX_HINT=''
SELECTED_MX_HOST=''
SELECTED_SRS_DOMAIN=''
hostname() {{ printf '%s\\n' {hostname!r}; }}
{PROMPTS}
prompt_srs_domain
printf 'RESULT_MX=%s\\nRESULT_SRS=%s\\n' "$SELECTED_MX_HOST" "$SELECTED_SRS_DOMAIN"
'''
return subprocess.run(
['bash', '-c', script], input=choices, text=True, capture_output=True
)
def test_first_install_defaults_from_hostname(self):
result = self.run_prompt()
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('RESULT_MX=o13-mail.oopen.de', result.stdout)
self.assertIn('RESULT_SRS=srs.o13-mail.oopen.de', result.stdout)
def test_public_mx_override_changes_srs_suggestion(self):
result = self.run_prompt(choices='mail.interventionistische-linke.org\n\n')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('RESULT_MX=mail.interventionistische-linke.org', result.stdout)
self.assertIn('RESULT_SRS=srs.mail.interventionistische-linke.org', result.stdout)
def test_configured_values_take_precedence(self):
config = '_MX_HOST="mail.example.org"\n_SRS_DOMAIN="srs.mail.example.org"\n'
result = self.run_prompt(config=config)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('RESULT_MX=mail.example.org', result.stdout)
self.assertIn('RESULT_SRS=srs.mail.example.org', result.stdout)
def test_legacy_srs_domain_preserved(self):
config = '_SRS_DOMAIN="srs.original.example.org"\n'
result = self.run_prompt(config=config, choices='mail.example.org\n\n')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('RESULT_MX=mail.example.org', result.stdout)
self.assertIn('RESULT_SRS=srs.original.example.org', result.stdout)
def test_invalid_host_reprompted(self):
result = self.run_prompt(choices='bad_host\nmail.example.org\n\n')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('[ invalid ]', result.stderr)
self.assertIn('RESULT_MX=mail.example.org', result.stdout)
def test_prompt_eof_fails_without_defaulting(self):
result = self.run_prompt(choices='')
self.assertEqual(result.returncode, 2)
def test_dns_preflight_passes_selected_mx_not_postfix_myhostname(self):
with tempfile.TemporaryDirectory() as temp:
binpath = pathlib.Path(temp)
(binpath / 'dig').write_text('#!/bin/sh\nexit 0\n')
(binpath / 'dig').chmod(0o755)
postconf = binpath / 'postconf'
postconf.write_text('''#!/bin/sh
case "$2" in
smtp_bind_address) echo 83.223.90.231 ;;
smtp_bind_address6) echo 2a01:31:0:2::ad52:a231 ;;
*) echo 'unexpected postconf query' >&2; exit 4 ;;
esac
''')
postconf.chmod(0o755)
checker = binpath / 'check.py'
checker.write_text('''import sys
print('ARGS=' + ';'.join(sys.argv[1:]))
''')
script = f'''SELECTED_MX_HOST=mail.interventionistische-linke.org
SELECTED_SRS_DOMAIN=srs.mail.interventionistische-linke.org
SRS_DNS_CHECKER={str(checker)!r}
{DNS}
verify_srs_domain_dns
'''
env = dict(os.environ, PATH=f'{temp}:' + os.environ['PATH'])
proc = subprocess.run(['bash','-c',script], capture_output=True, text=True, env=env)
self.assertEqual(proc.returncode, 0, proc.stderr + proc.stdout)
self.assertIn('--hostname;mail.interventionistische-linke.org', proc.stdout)
self.assertIn('--ipv4;83.223.90.231', proc.stdout)
self.assertIn('--ipv6;2a01:31:0:2::ad52:a231', proc.stdout)
def test_existing_config_migration_is_append_only(self):
with tempfile.TemporaryDirectory() as tmp:
src = pathlib.Path(tmp) / 'old.conf'
dst = pathlib.Path(tmp) / 'new.conf'
content = b'# existing comments\n_SRS_DOMAIN="srs.example.org"\n\n'
src.write_bytes(content)
proc = subprocess.run([sys.executable, '-', str(src), str(dst), 'mail.example.org'],
input=MIGRATION.group(1), capture_output=True, text=True)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertTrue(dst.read_bytes().startswith(content))
self.assertIn(b'_MX_HOST="mail.example.org"\n', dst.read_bytes())
self.assertEqual(src.read_bytes(), content)
def test_migration_rejects_existing_mx_assignment(self):
with tempfile.TemporaryDirectory() as tmp:
src = pathlib.Path(tmp) / 'old.conf'
dst = pathlib.Path(tmp) / 'new.conf'
src.write_text('_MX_HOST="other.example.org"\n')
proc = subprocess.run([sys.executable, '-', str(src), str(dst), 'mail.example.org'],
input=MIGRATION.group(1), capture_output=True, text=True)
self.assertNotEqual(proc.returncode, 0)
self.assertFalse(dst.exists())
def run_config_load(self, content):
with tempfile.TemporaryDirectory() as tmp:
conf = pathlib.Path(tmp) / 'srs.conf'
conf.write_text(content)
script = f'''SRS_CONF_FILE={str(conf)!r}
SRS_CONF_SAMPLE_FILE=unused
check_secure_source_file() {{ :; }}
echononl() {{ :; }}
echo_failed() {{ :; }}
echo_ok() {{ :; }}
fatal() {{ echo "FATAL: $*" >&2; exit 1; }}
{LOAD}
load_srs_config
printf 'MX_IS_SET=%s\\n' "${{_MX_HOST+x}}"
'''
return subprocess.run(['bash','-c',script],text=True,capture_output=True)
def test_load_config_accepts_legacy_without_mx_host(self):
result = self.run_config_load('_SRS_DOMAIN="srs.example.org"\n')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('MX_IS_SET=', result.stdout)
def test_load_config_accepts_valid_mx_host(self):
result = self.run_config_load('_MX_HOST="mail.example.org"\n_SRS_DOMAIN="srs.example.org"\n')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('MX_IS_SET=x', result.stdout)
def test_load_config_rejects_invalid_mx_host(self):
result = self.run_config_load('_MX_HOST="bad_host"\n_SRS_DOMAIN="srs.example.org"\n')
self.assertEqual(result.returncode, 1)
self.assertIn('Invalid public MX hostname', result.stderr)
def test_templates_and_security_contract(self):
self.assertIn('_MX_HOST=""', SAMPLE)
self.assertIn('_SRS_DOMAIN=""', SAMPLE)
self.assertIn('unset _MX_HOST', COMMON)
self.assertIn('if [[ -n "${_MX_HOST+x}" ]]', COMMON)
self.assertIn('"${_MX_HOST,,}" != "$SELECTED_MX_HOST"', INSTALLER)
self.assertIn('install_regular_file "${STAGE_DIR}/rendered/srs.conf" "$SRS_CONF_FILE"', INSTALLER)
if __name__ == '__main__':
unittest.main(verbosity=2)
tests/test_srs_dns.py
+106
View File
@@ -0,0 +1,106 @@
#!/usr/bin/env python3
"""Offline regression tests for the selective SRS DNS preflight helper.
All DNS responses are mocked. No live DNS queries, Postfix changes,
or SRS installation operations are performed.
"""
import contextlib
import io
import runpy
import unittest
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
HELPER = ROOT / "helpers/srs-dns-check.py"
module = runpy.run_path(str(HELPER))
check_server = module["check_server"]
class TestSrsDnsSpf(unittest.TestCase):
DOMAIN = "srs.example.org"
HOSTNAME = "mail.example.org"
IPV4 = "192.0.2.25"
IPV6 = "2001:db8::25"
def check_with_spf(self, txt_records):
answers = {
(self.DOMAIN, "TXT"): txt_records,
(self.DOMAIN, "MX"): [
f"10 {self.HOSTNAME}."
],
(self.HOSTNAME, "A"): [
self.IPV4
],
(self.HOSTNAME, "AAAA"): [
self.IPV6
],
}
def fake_dig(server, name, kind, authoritative=False):
self.assertEqual(server, "192.0.2.53")
self.assertTrue(
authoritative,
"DNS checks must query authoritative nameservers"
)
return answers.get((name, kind), [])
output = io.StringIO()
with patch.dict(
check_server.__globals__,
{"dig": fake_dig}
):
with contextlib.redirect_stdout(output):
result = check_server(
"ns.example.org",
"192.0.2.53",
self.DOMAIN,
self.HOSTNAME,
self.IPV4,
self.IPV6,
)
return result, output.getvalue()
def test_valid_spf(self):
result, output = self.check_with_spf(
['"v=spf1 mx -all"']
)
self.assertEqual(result, (True, True))
self.assertIn("[ ok ]", output)
def test_missing_spf(self):
result, output = self.check_with_spf([])
self.assertEqual(result, (False, False))
self.assertIn("SPF is not the approved", output)
self.assertIn("[ failed ]", output)
def test_invalid_spf(self):
result, output = self.check_with_spf(
['"v=spf1 -all"']
)
self.assertEqual(result, (False, False))
self.assertIn("SPF is not the approved", output)
self.assertIn("[ failed ]", output)
def test_multiple_spf_records(self):
result, output = self.check_with_spf([
'"v=spf1 mx -all"',
'"v=spf1 ip4:192.0.2.1 -all"',
])
self.assertEqual(result, (False, False))
self.assertIn("SPF is not the approved", output)
self.assertIn("[ failed ]", output)
if __name__ == "__main__":
unittest.main(verbosity=2)