Isolate external monitor SMTP credentials
This commit is contained in:
parent
360c7a567e
commit
f672e9cc90
|
|
@ -742,6 +742,50 @@ guarantees. Re-running the installer refreshes the exact script, mode-`0600`
|
|||
configuration, and user units, then performs one check before enabling the
|
||||
timer.
|
||||
|
||||
By default the installer mirrors the production application's SMTP credential.
|
||||
For the pilot, use an independently revocable monitor SMTP key so revoking or
|
||||
rotating the application key does not also disable operational alerts. Keep the
|
||||
override outside the repository in a mode-`0600` (or read-only mode-`0400`)
|
||||
file containing exactly these keys:
|
||||
|
||||
```dotenv
|
||||
SMTP_RELAY=smtp-relay.example
|
||||
SMTP_PORT=587
|
||||
SMTP_USERNAME=independent-monitor-login
|
||||
SMTP_PASSWORD=secret-from-the-provider
|
||||
SMTP_TLS=always
|
||||
SMTP_SSL=false
|
||||
EMAIL_FROM_ADDRESS=monitor@whoneedhelp.com
|
||||
EMAIL_FROM_NAME=Who Need Help monitor
|
||||
SUPPORT_INBOX_ADDRESS=monitored-operator@example.com
|
||||
```
|
||||
|
||||
Create that file through the password manager or an editor that does not place
|
||||
the secret in shell history. When the override is present, the installer reads
|
||||
only `METRICS_TOKEN` from production and never copies the application's SMTP
|
||||
credential into its local staging configuration. Then reinstall the monitor
|
||||
with the scoped override and send one test notification before revoking any
|
||||
previous key:
|
||||
|
||||
```bash
|
||||
MONITOR_SMTP_VALUES_FILE="$HOME/.config/who-need-help/monitor-smtp.env" \
|
||||
./scripts/install-production-external-monitor.sh
|
||||
ssh buyvm-maya \
|
||||
'/usr/bin/python3 ~/.local/lib/who-need-help/production-external-monitor.py \
|
||||
--config ~/.config/who-need-help/monitor.json \
|
||||
--state ~/.local/state/who-need-help/monitor.json \
|
||||
send-test-notification'
|
||||
```
|
||||
|
||||
Verify receipt in the monitored mailbox. Only then revoke the former monitor
|
||||
key. The application SMTP key is a separate rotation: validate the new key,
|
||||
atomically update the production `.env`, recreate only the application
|
||||
services so they read the new runtime value, verify an application-generated
|
||||
authentication message and readiness, and revoke the former application key
|
||||
last. If any check fails before revocation, restore the prior mode-`0600`
|
||||
environment and recreate the same services; do not leave application and
|
||||
monitor configurations half-updated.
|
||||
|
||||
## Local external-service boundary drill
|
||||
|
||||
Run the OAuth, SMTP, and provider-neutral push protocol checks without public
|
||||
|
|
|
|||
|
|
@ -3,6 +3,34 @@
|
|||
Observed through 2026-08-09 in the local workspace. This report separates observed
|
||||
results from product limits and unknown production properties.
|
||||
|
||||
## External-monitor SMTP isolation proof on 2026-08-09
|
||||
|
||||
- The local change set based on revision `360c7a5` adds an optional,
|
||||
independently revocable SMTP credential set for the off-host production
|
||||
monitor. In override mode the installer reads only `METRICS_TOKEN` from the
|
||||
production environment; application SMTP values are neither requested nor
|
||||
copied into the local staging configuration. The existing application-SMTP
|
||||
mode remains available for a controlled transition.
|
||||
- Six focused tests passed in isolated user-systemd scope
|
||||
`codex-heavy-wnh-monitor-smtp-focused-20260809-224809-2206905.service`.
|
||||
They covered independent and legacy installer modes, exact-key parsing,
|
||||
restrictive source-file modes, transport validation, atomic replacement,
|
||||
preservation of non-SMTP monitor settings, and absence of both application
|
||||
and monitor passwords from command output. The modified installer also
|
||||
passed ShellCheck 0.11.0 in isolated scope
|
||||
`codex-heavy-wnh-monitor-smtp-shellcheck-20260809-224809-2206903.service`.
|
||||
- The complete isolated `scripts/quality.sh` pipeline passed in
|
||||
`codex-heavy-wnh-quality-monitor-smtp-final-20260809-224827-2217374.service`.
|
||||
It completed successfully in 2 minutes 21.914 seconds with a measured
|
||||
218.3 MiB memory peak, passed all configured quality and security gates,
|
||||
458 ExUnit tests, fourteen browser-asset tests, the monitor suites, and the
|
||||
final Debian 13.6 image scan with zero detected vulnerabilities.
|
||||
- These checks were local. They did not install or reconfigure the external
|
||||
monitor, rotate a provider credential, deploy production, change the frozen
|
||||
hackathon-test environment, or push the public Git remote. Provider-side key
|
||||
creation and the monitored mailbox receipt check remain explicit external
|
||||
operations.
|
||||
|
||||
## Current pilot-candidate recheck on 2026-08-09
|
||||
|
||||
- Local revision `beb5de5eda5e7490cf8b757810bf55787cce211f` passed the
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
umask 077
|
||||
|
||||
ROOT=$(CDPATH='' cd -- "$(dirname -- "$0")/.." && pwd)
|
||||
|
|
@ -9,22 +9,58 @@ production_env=${PRODUCTION_ENV_PATH:-/srv/who_need_help-production/.env}
|
|||
remote_root=${MONITOR_REMOTE_ROOT:-/home/simple/.local/lib/who-need-help}
|
||||
remote_config=${MONITOR_REMOTE_CONFIG:-/home/simple/.config/who-need-help/monitor.json}
|
||||
remote_state=${MONITOR_REMOTE_STATE:-/home/simple/.local/state/who-need-help/monitor.json}
|
||||
monitor_smtp_values_file=${MONITOR_SMTP_VALUES_FILE:-}
|
||||
monitor_install_work_root=${MONITOR_INSTALL_WORK_ROOT:-}
|
||||
monitor_url=${MONITOR_URL:-https://whoneedhelp.com/healthz/ready}
|
||||
metrics_url=${MONITOR_METRICS_URL:-https://whoneedhelp.com/metrics}
|
||||
monitor_calendar=${MONITOR_ON_CALENDAR:-'*:0/1'}
|
||||
health_timeout=${MONITOR_HEALTH_TIMEOUT_SECONDS:-3}
|
||||
metrics_timeout=${MONITOR_METRICS_TIMEOUT_SECONDS:-3}
|
||||
|
||||
for command in mktemp scp ssh; do
|
||||
for command in awk install mktemp python3 realpath scp ssh stat; do
|
||||
command -v "$command" >/dev/null 2>&1 || {
|
||||
echo "Required command is unavailable: $command" >&2
|
||||
exit 2
|
||||
}
|
||||
done
|
||||
|
||||
if [ -z "$monitor_install_work_root" ]; then
|
||||
monitor_install_work_root=$ROOT/tmp/production-operations
|
||||
install -d -m 700 "$monitor_install_work_root"
|
||||
elif [ ! -d "$monitor_install_work_root" ]; then
|
||||
echo "MONITOR_INSTALL_WORK_ROOT must be an existing directory when supplied." >&2
|
||||
exit 2
|
||||
fi
|
||||
monitor_install_work_root=$(realpath "$monitor_install_work_root")
|
||||
if [ ! -w "$monitor_install_work_root" ]; then
|
||||
echo "MONITOR_INSTALL_WORK_ROOT must be an existing writable directory." >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
if [ -n "$monitor_smtp_values_file" ]; then
|
||||
if [ ! -f "$monitor_smtp_values_file" ]; then
|
||||
echo "MONITOR_SMTP_VALUES_FILE must be an existing regular file." >&2
|
||||
exit 2
|
||||
fi
|
||||
monitor_smtp_values_file=$(realpath "$monitor_smtp_values_file")
|
||||
case "$(stat -c '%a' "$monitor_smtp_values_file")" in
|
||||
400 | 600) ;;
|
||||
*)
|
||||
echo "MONITOR_SMTP_VALUES_FILE must have mode 0400 or 0600." >&2
|
||||
exit 2
|
||||
;;
|
||||
esac
|
||||
fi
|
||||
|
||||
if [ -n "$monitor_smtp_values_file" ]; then
|
||||
smtp_source=override
|
||||
else
|
||||
smtp_source=application
|
||||
fi
|
||||
|
||||
source_host=$(ssh -G "$source_target" | awk '$1 == "hostname" {print $2; exit}')
|
||||
monitor_host=$(ssh -G "$monitor_target" | awk '$1 == "hostname" {print $2; exit}')
|
||||
if [[ -z "$source_host" || -z "$monitor_host" || "$source_host" == "$monitor_host" ]]; then
|
||||
if [ -z "$source_host" ] || [ -z "$monitor_host" ] || [ "$source_host" = "$monitor_host" ]; then
|
||||
echo "The external monitor must resolve and run on a host other than production." >&2
|
||||
exit 2
|
||||
fi
|
||||
|
|
@ -39,24 +75,25 @@ case "$metrics_timeout" in
|
|||
0) echo "MONITOR_METRICS_TIMEOUT_SECONDS must be a positive integer." >&2; exit 2 ;;
|
||||
esac
|
||||
|
||||
work_dir=$(mktemp -d "$ROOT/tmp/production-operations/monitor-install.XXXXXX")
|
||||
work_dir=$(mktemp -d "$monitor_install_work_root/monitor-install.XXXXXX")
|
||||
config="$work_dir/monitor.json"
|
||||
service="$work_dir/who-need-help-production-monitor.service"
|
||||
timer="$work_dir/who-need-help-production-monitor.timer"
|
||||
|
||||
cleanup() {
|
||||
trap - 0 HUP INT TERM
|
||||
rm -rf "$work_dir"
|
||||
}
|
||||
trap cleanup EXIT HUP INT TERM
|
||||
trap cleanup 0 HUP INT TERM
|
||||
|
||||
ssh -o BatchMode=yes "$source_target" \
|
||||
"python3 - '$production_env' '$monitor_url' '$metrics_url' '$health_timeout' '$metrics_timeout'" >"$config" <<'PY'
|
||||
"python3 - '$production_env' '$monitor_url' '$metrics_url' '$health_timeout' '$metrics_timeout' '$smtp_source'" >"$config" <<'PY'
|
||||
import json
|
||||
import shlex
|
||||
import sys
|
||||
|
||||
path, health_url, metrics_url, health_timeout, metrics_timeout = sys.argv[1:]
|
||||
wanted = {
|
||||
path, health_url, metrics_url, health_timeout, metrics_timeout, smtp_source = sys.argv[1:]
|
||||
smtp_keys = {
|
||||
"SMTP_RELAY",
|
||||
"SMTP_PORT",
|
||||
"SMTP_USERNAME",
|
||||
|
|
@ -66,8 +103,13 @@ wanted = {
|
|||
"EMAIL_FROM_ADDRESS",
|
||||
"EMAIL_FROM_NAME",
|
||||
"SUPPORT_INBOX_ADDRESS",
|
||||
"METRICS_TOKEN",
|
||||
}
|
||||
wanted = {"METRICS_TOKEN"}
|
||||
if smtp_source == "application":
|
||||
wanted.update(smtp_keys)
|
||||
elif smtp_source != "override":
|
||||
raise SystemExit("Unknown monitor SMTP source")
|
||||
|
||||
values = {}
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
for raw_line in handle:
|
||||
|
|
@ -82,15 +124,11 @@ with open(path, encoding="utf-8") as handle:
|
|||
|
||||
missing = sorted(key for key in wanted if not values.get(key))
|
||||
if missing:
|
||||
raise SystemExit("Missing production mail settings: " + ", ".join(missing))
|
||||
raise SystemExit("Missing production monitor settings: " + ", ".join(missing))
|
||||
|
||||
payload = {
|
||||
"health_timeout_seconds": int(health_timeout),
|
||||
"health_url": health_url,
|
||||
"metrics_timeout_seconds": int(metrics_timeout),
|
||||
"metrics_token": values["METRICS_TOKEN"],
|
||||
"metrics_url": metrics_url,
|
||||
"smtp": {
|
||||
smtp = {}
|
||||
if smtp_source == "application":
|
||||
smtp = {
|
||||
"from_address": values["EMAIL_FROM_ADDRESS"],
|
||||
"from_name": values["EMAIL_FROM_NAME"],
|
||||
"implicit_ssl": values["SMTP_SSL"].lower() == "true",
|
||||
|
|
@ -100,13 +138,26 @@ payload = {
|
|||
"relay": values["SMTP_RELAY"],
|
||||
"starttls": values["SMTP_TLS"].lower() == "always",
|
||||
"username": values["SMTP_USERNAME"],
|
||||
},
|
||||
}
|
||||
|
||||
payload = {
|
||||
"health_timeout_seconds": int(health_timeout),
|
||||
"health_url": health_url,
|
||||
"metrics_timeout_seconds": int(metrics_timeout),
|
||||
"metrics_token": values["METRICS_TOKEN"],
|
||||
"metrics_url": metrics_url,
|
||||
"smtp": smtp,
|
||||
}
|
||||
json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
|
||||
sys.stdout.write("\n")
|
||||
PY
|
||||
chmod 600 "$config"
|
||||
|
||||
if [ -n "$monitor_smtp_values_file" ]; then
|
||||
"$ROOT/scripts/override-production-monitor-smtp.py" \
|
||||
"$config" "$monitor_smtp_values_file"
|
||||
fi
|
||||
|
||||
cat >"$service" <<EOF
|
||||
[Unit]
|
||||
Description=Who Need Help independent production operations monitor
|
||||
|
|
@ -148,3 +199,8 @@ printf 'Metrics URL: %s\n' "$metrics_url"
|
|||
printf 'Schedule: %s; health timeout: %ss; metrics timeout: %ss.\n' \
|
||||
"$monitor_calendar" "$health_timeout" "$metrics_timeout"
|
||||
echo "The SMTP and metrics credentials are stored only in a mode-0600 configuration on the external monitor host."
|
||||
if [ -n "$monitor_smtp_values_file" ]; then
|
||||
echo "The external monitor uses the independently supplied SMTP credential set."
|
||||
else
|
||||
echo "The external monitor currently mirrors the production application SMTP credential set."
|
||||
fi
|
||||
|
|
|
|||
140
scripts/override-production-monitor-smtp.py
Executable file
140
scripts/override-production-monitor-smtp.py
Executable file
|
|
@ -0,0 +1,140 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Atomically replace only the SMTP section of a monitor configuration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import stat
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
EXPECTED_KEYS = {
|
||||
"EMAIL_FROM_ADDRESS",
|
||||
"EMAIL_FROM_NAME",
|
||||
"SMTP_PASSWORD",
|
||||
"SMTP_PORT",
|
||||
"SMTP_RELAY",
|
||||
"SMTP_SSL",
|
||||
"SMTP_TLS",
|
||||
"SMTP_USERNAME",
|
||||
"SUPPORT_INBOX_ADDRESS",
|
||||
}
|
||||
|
||||
|
||||
def parse_values(path: Path) -> dict[str, str]:
|
||||
mode = stat.S_IMODE(path.stat().st_mode)
|
||||
if mode not in {0o400, 0o600}:
|
||||
raise ValueError("SMTP values file must have mode 0400 or 0600")
|
||||
|
||||
values: dict[str, str] = {}
|
||||
with path.open(encoding="utf-8") as handle:
|
||||
for line_number, raw_line in enumerate(handle, start=1):
|
||||
line = raw_line.rstrip("\r\n")
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" not in line:
|
||||
raise ValueError(f"Malformed SMTP value on line {line_number}")
|
||||
key, raw_value = line.split("=", 1)
|
||||
if key in values:
|
||||
raise ValueError(f"Duplicate SMTP value: {key}")
|
||||
parsed = shlex.split(raw_value, comments=False, posix=True)
|
||||
if len(parsed) != 1 or not parsed[0]:
|
||||
raise ValueError(f"SMTP value must contain one non-empty token: {key}")
|
||||
values[key] = parsed[0]
|
||||
|
||||
missing = sorted(EXPECTED_KEYS - values.keys())
|
||||
extra = sorted(values.keys() - EXPECTED_KEYS)
|
||||
if missing or extra:
|
||||
details: list[str] = []
|
||||
if missing:
|
||||
details.append("missing " + ", ".join(missing))
|
||||
if extra:
|
||||
details.append("unexpected " + ", ".join(extra))
|
||||
raise ValueError("Invalid SMTP values file: " + "; ".join(details))
|
||||
return values
|
||||
|
||||
|
||||
def parse_bool(value: str, name: str) -> bool:
|
||||
normalized = value.lower()
|
||||
if normalized in {"true", "1"}:
|
||||
return True
|
||||
if normalized in {"false", "0"}:
|
||||
return False
|
||||
raise ValueError(f"{name} must be true, false, 1, or 0")
|
||||
|
||||
|
||||
def build_smtp(values: dict[str, str]) -> dict[str, Any]:
|
||||
try:
|
||||
port = int(values["SMTP_PORT"])
|
||||
except ValueError as error:
|
||||
raise ValueError("SMTP_PORT must be an integer") from error
|
||||
if not 1 <= port <= 65535:
|
||||
raise ValueError("SMTP_PORT must be between 1 and 65535")
|
||||
|
||||
tls = values["SMTP_TLS"].lower()
|
||||
if tls not in {"always", "never"}:
|
||||
raise ValueError("SMTP_TLS must be always or never for the external monitor")
|
||||
implicit_ssl = parse_bool(values["SMTP_SSL"], "SMTP_SSL")
|
||||
if implicit_ssl and tls != "never":
|
||||
raise ValueError("SMTP_TLS must be never when SMTP_SSL enables implicit TLS")
|
||||
|
||||
return {
|
||||
"from_address": values["EMAIL_FROM_ADDRESS"],
|
||||
"from_name": values["EMAIL_FROM_NAME"],
|
||||
"implicit_ssl": implicit_ssl,
|
||||
"password": values["SMTP_PASSWORD"],
|
||||
"port": port,
|
||||
"recipient": values["SUPPORT_INBOX_ADDRESS"],
|
||||
"relay": values["SMTP_RELAY"],
|
||||
"starttls": tls == "always",
|
||||
"username": values["SMTP_USERNAME"],
|
||||
}
|
||||
|
||||
|
||||
def read_config(path: Path) -> dict[str, Any]:
|
||||
with path.open(encoding="utf-8") as handle:
|
||||
config = json.load(handle)
|
||||
if not isinstance(config, dict) or not isinstance(config.get("smtp"), dict):
|
||||
raise ValueError("Monitor configuration must contain an SMTP object")
|
||||
return config
|
||||
|
||||
|
||||
def write_atomic(path: Path, config: dict[str, Any]) -> None:
|
||||
descriptor, temporary_name = tempfile.mkstemp(
|
||||
dir=path.parent, prefix=f".{path.name}.smtp."
|
||||
)
|
||||
temporary = Path(temporary_name)
|
||||
try:
|
||||
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
||||
json.dump(config, handle, ensure_ascii=False, indent=2, sort_keys=True)
|
||||
handle.write("\n")
|
||||
temporary.chmod(0o600)
|
||||
temporary.replace(path)
|
||||
finally:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("config", type=Path)
|
||||
parser.add_argument("values", type=Path)
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
config = read_config(args.config)
|
||||
config["smtp"] = build_smtp(parse_values(args.values))
|
||||
write_atomic(args.config, config)
|
||||
except (OSError, ValueError, json.JSONDecodeError) as error:
|
||||
parser.error(str(error))
|
||||
|
||||
print("External monitor SMTP configuration updated without printing credentials.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
@ -1546,6 +1546,14 @@ docker run --rm \
|
|||
--volume "$ROOT:/src:ro" \
|
||||
--workdir /src \
|
||||
"$PYTHON_IMAGE" python test/scripts/production_external_monitor_test.py
|
||||
docker run --rm \
|
||||
--volume "$ROOT:/src:ro" \
|
||||
--workdir /src \
|
||||
"$PYTHON_IMAGE" python test/scripts/override_production_monitor_smtp_test.py
|
||||
docker run --rm \
|
||||
--volume "$ROOT:/src:ro" \
|
||||
--workdir /src \
|
||||
"$PYTHON_IMAGE" python test/scripts/install_production_external_monitor_test.py
|
||||
docker run --rm \
|
||||
--volume "$ROOT/ops/external-boundaries/mock_server.py:/src/mock_server.py:ro" \
|
||||
"$PYTHON_IMAGE" python -c \
|
||||
|
|
|
|||
170
test/scripts/install_production_external_monitor_test.py
Normal file
170
test/scripts/install_production_external_monitor_test.py
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import textwrap
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
INSTALLER = ROOT / "scripts" / "install-production-external-monitor.sh"
|
||||
|
||||
|
||||
class InstallProductionExternalMonitorTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.tempdir.name)
|
||||
self.fake_bin = self.root / "bin"
|
||||
self.fake_bin.mkdir()
|
||||
self.capture = self.root / "captured-monitor.json"
|
||||
self.work_root = self.root / "work"
|
||||
self.work_root.mkdir()
|
||||
self.production_env = self.root / "production.env"
|
||||
self.monitor_values = self.root / "monitor-smtp.env"
|
||||
|
||||
self.production_env.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
METRICS_TOKEN=metrics-secret
|
||||
SMTP_RELAY=application-relay.example.test
|
||||
SMTP_PORT=587
|
||||
SMTP_USERNAME=application-user
|
||||
SMTP_PASSWORD=application-secret
|
||||
SMTP_TLS=always
|
||||
SMTP_SSL=false
|
||||
EMAIL_FROM_ADDRESS=application@example.test
|
||||
EMAIL_FROM_NAME='Application sender'
|
||||
SUPPORT_INBOX_ADDRESS=application-ops@example.test
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
self.production_env.chmod(0o600)
|
||||
self.monitor_values.write_text(
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
SMTP_RELAY=monitor-relay.example.test
|
||||
SMTP_PORT=587
|
||||
SMTP_USERNAME=monitor-user
|
||||
SMTP_PASSWORD=monitor-secret
|
||||
SMTP_TLS=always
|
||||
SMTP_SSL=false
|
||||
EMAIL_FROM_ADDRESS=monitor@example.test
|
||||
EMAIL_FROM_NAME='Who Need Help monitor'
|
||||
SUPPORT_INBOX_ADDRESS=monitor-ops@example.test
|
||||
"""
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
self.monitor_values.chmod(0o600)
|
||||
self.write_fake_commands()
|
||||
|
||||
def tearDown(self):
|
||||
self.tempdir.cleanup()
|
||||
|
||||
def write_executable(self, name, content):
|
||||
path = self.fake_bin / name
|
||||
path.write_text(content, encoding="utf-8")
|
||||
path.chmod(0o755)
|
||||
|
||||
def write_fake_commands(self):
|
||||
self.write_executable(
|
||||
"ssh",
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
#!/usr/bin/env python3
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
if len(sys.argv) == 3 and sys.argv[1] == "-G":
|
||||
print(f"hostname {sys.argv[2]}.example.test")
|
||||
raise SystemExit(0)
|
||||
|
||||
command = sys.argv[-1]
|
||||
if command.startswith("python3 - "):
|
||||
result = subprocess.run(
|
||||
shlex.split(command),
|
||||
input=sys.stdin.read(),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
sys.stdout.write(result.stdout)
|
||||
sys.stderr.write(result.stderr)
|
||||
raise SystemExit(result.returncode)
|
||||
raise SystemExit(0)
|
||||
"""
|
||||
),
|
||||
)
|
||||
self.write_executable(
|
||||
"scp",
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sources = [Path(value) for value in sys.argv[1:-1] if not value.startswith("-")]
|
||||
for source in sources:
|
||||
if source.name == "monitor.json":
|
||||
shutil.copyfile(source, os.environ["FAKE_MONITOR_CAPTURE"])
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
def run_installer(self, *, override):
|
||||
self.capture.unlink(missing_ok=True)
|
||||
env = os.environ.copy()
|
||||
env.update(
|
||||
{
|
||||
"PATH": f"{self.fake_bin}:{env['PATH']}",
|
||||
"PRODUCTION_SSH_TARGET": "production",
|
||||
"MONITOR_SSH_TARGET": "monitor",
|
||||
"PRODUCTION_ENV_PATH": str(self.production_env),
|
||||
"FAKE_MONITOR_CAPTURE": str(self.capture),
|
||||
"MONITOR_INSTALL_WORK_ROOT": str(self.work_root),
|
||||
}
|
||||
)
|
||||
if override:
|
||||
env["MONITOR_SMTP_VALUES_FILE"] = str(self.monitor_values)
|
||||
else:
|
||||
env.pop("MONITOR_SMTP_VALUES_FILE", None)
|
||||
return subprocess.run(
|
||||
[str(INSTALLER)],
|
||||
cwd=ROOT,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_independent_smtp_reaches_monitor_without_application_smtp_secret(self):
|
||||
result = self.run_installer(override=True)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
captured = json.loads(self.capture.read_text(encoding="utf-8"))
|
||||
self.assertEqual(captured["metrics_token"], "metrics-secret")
|
||||
self.assertEqual(captured["smtp"]["password"], "monitor-secret")
|
||||
self.assertEqual(captured["smtp"]["relay"], "monitor-relay.example.test")
|
||||
combined_output = result.stdout + result.stderr
|
||||
self.assertNotIn("application-secret", combined_output)
|
||||
self.assertNotIn("monitor-secret", combined_output)
|
||||
self.assertIn("independently supplied SMTP credential", result.stdout)
|
||||
|
||||
def test_default_mode_still_uses_application_smtp(self):
|
||||
result = self.run_installer(override=False)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
captured = json.loads(self.capture.read_text(encoding="utf-8"))
|
||||
self.assertEqual(captured["smtp"]["password"], "application-secret")
|
||||
self.assertEqual(captured["smtp"]["relay"], "application-relay.example.test")
|
||||
self.assertNotIn("application-secret", result.stdout + result.stderr)
|
||||
self.assertIn("mirrors the production application SMTP", result.stdout)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
122
test/scripts/override_production_monitor_smtp_test.py
Normal file
122
test/scripts/override_production_monitor_smtp_test.py
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SCRIPT = ROOT / "scripts" / "override-production-monitor-smtp.py"
|
||||
|
||||
|
||||
class OverrideProductionMonitorSmtpTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.tempdir.name)
|
||||
self.config = self.root / "monitor.json"
|
||||
self.values = self.root / "smtp.env"
|
||||
self.original = {
|
||||
"health_url": "https://example.test/healthz/ready",
|
||||
"metrics_token": "metrics-token",
|
||||
"metrics_url": "https://example.test/metrics",
|
||||
"smtp": {
|
||||
"from_address": "old@example.test",
|
||||
"from_name": "Old sender",
|
||||
"implicit_ssl": False,
|
||||
"password": "old-password",
|
||||
"port": 587,
|
||||
"recipient": "old-ops@example.test",
|
||||
"relay": "old-relay.example.test",
|
||||
"starttls": True,
|
||||
"username": "old-user",
|
||||
},
|
||||
}
|
||||
self.config.write_text(json.dumps(self.original), encoding="utf-8")
|
||||
self.config.chmod(0o600)
|
||||
|
||||
def tearDown(self):
|
||||
self.tempdir.cleanup()
|
||||
|
||||
def write_values(self, content):
|
||||
self.values.write_text(content, encoding="utf-8")
|
||||
self.values.chmod(0o600)
|
||||
|
||||
def run_script(self):
|
||||
return subprocess.run(
|
||||
[str(SCRIPT), str(self.config), str(self.values)],
|
||||
cwd=ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def valid_values(self):
|
||||
return "\n".join(
|
||||
[
|
||||
"SMTP_RELAY=smtp.example.test",
|
||||
"SMTP_PORT=587",
|
||||
"SMTP_USERNAME=monitor-user",
|
||||
"SMTP_PASSWORD='new secret with spaces'",
|
||||
"SMTP_TLS=always",
|
||||
"SMTP_SSL=false",
|
||||
"EMAIL_FROM_ADDRESS=monitor@example.test",
|
||||
"EMAIL_FROM_NAME='Who Need Help monitor'",
|
||||
"SUPPORT_INBOX_ADDRESS=ops@example.test",
|
||||
"",
|
||||
]
|
||||
)
|
||||
|
||||
def test_replaces_only_smtp_atomically_without_printing_secret(self):
|
||||
self.write_values(self.valid_values())
|
||||
|
||||
result = self.run_script()
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertNotIn("new secret", result.stdout + result.stderr)
|
||||
updated = json.loads(self.config.read_text(encoding="utf-8"))
|
||||
self.assertEqual(updated["health_url"], self.original["health_url"])
|
||||
self.assertEqual(updated["metrics_token"], self.original["metrics_token"])
|
||||
self.assertEqual(updated["smtp"]["password"], "new secret with spaces")
|
||||
self.assertEqual(updated["smtp"]["from_name"], "Who Need Help monitor")
|
||||
self.assertEqual(updated["smtp"]["port"], 587)
|
||||
self.assertTrue(updated["smtp"]["starttls"])
|
||||
self.assertFalse(updated["smtp"]["implicit_ssl"])
|
||||
self.assertEqual(os.stat(self.config).st_mode & 0o777, 0o600)
|
||||
|
||||
def test_rejects_missing_or_extra_keys_without_modifying_config(self):
|
||||
self.write_values(
|
||||
self.valid_values().replace("SMTP_USERNAME=monitor-user\n", "")
|
||||
+ "UNEXPECTED=value\n"
|
||||
)
|
||||
before = self.config.read_bytes()
|
||||
|
||||
result = self.run_script()
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertEqual(self.config.read_bytes(), before)
|
||||
|
||||
def test_rejects_insecure_values_file_mode(self):
|
||||
self.write_values(self.valid_values())
|
||||
self.values.chmod(0o644)
|
||||
before = self.config.read_bytes()
|
||||
|
||||
result = self.run_script()
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("0400 or 0600", result.stderr)
|
||||
self.assertEqual(self.config.read_bytes(), before)
|
||||
|
||||
def test_rejects_conflicting_tls_modes(self):
|
||||
self.write_values(self.valid_values().replace("SMTP_SSL=false", "SMTP_SSL=true"))
|
||||
before = self.config.read_bytes()
|
||||
|
||||
result = self.run_script()
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("SMTP_TLS must be never", result.stderr)
|
||||
self.assertEqual(self.config.read_bytes(), before)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user