mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-07-31 15:59:27 +00:00
331 lines
13 KiB
Python
331 lines
13 KiB
Python
"""Regression tests for wheel contents and hardened service deployment policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from scripts.migrate_service_layout import migrate_service_layout
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_service_units_do_not_make_code_tree_writable():
|
|
unit = (REPO_ROOT / "meshcore-bot.service").read_text(encoding="utf-8")
|
|
assert "--config /etc/meshcore-bot/config.ini" in unit
|
|
assert "ReadWritePaths=/opt/meshcore-bot" not in unit
|
|
assert "ReadWritePaths=/var/lib/meshcore-bot" in unit
|
|
assert "ReadWritePaths=/etc/meshcore-bot" in unit
|
|
assert "UMask=0077" in unit
|
|
|
|
deb_builder = (REPO_ROOT / "scripts/build-deb.sh").read_text(encoding="utf-8")
|
|
assert 'chown -R "${SERVICE_USER}:${SERVICE_USER}" "${INSTALL_ROOT}"' not in deb_builder
|
|
assert "migrate_service_layout.py" in deb_builder
|
|
assert 'chmod 0600 "${BUILD_DIR}${CONF_DIR}/config.ini"' in deb_builder
|
|
assert (
|
|
'chown -R "${SERVICE_USER}:${SERVICE_USER}" "${CONF_DIR}" "${DATA_DIR}" "${LOG_DIR}"'
|
|
in deb_builder
|
|
)
|
|
assert 'find "${CONF_DIR}" "${DATA_DIR}" -type f -exec chmod 0600' in deb_builder
|
|
assert 'VENV_BUILD="${INSTALL_ROOT}/.venv-build-$$"' in deb_builder
|
|
assert 'if [ ! -d "${INSTALL_ROOT}/venv" ]' not in deb_builder
|
|
assert 'scripts/debian_service_state.sh" "${BUILD_DIR}/DEBIAN/preinst"' in deb_builder
|
|
assert 'debian_service_state.sh" restore "${2:-}"' in deb_builder
|
|
assert 'debian_service_state.sh preremove "${1:-}"' in deb_builder
|
|
|
|
|
|
def test_standalone_installer_separates_code_and_private_state():
|
|
installer = (REPO_ROOT / "install-service.sh").read_text(encoding="utf-8")
|
|
assert 'chown -R "root:$CODE_GROUP" "$INSTALL_DIR"' in installer
|
|
assert 'chown -R "$SERVICE_USER:$SERVICE_GROUP" "$CONF_DIR" "$STATE_DIR"' in installer
|
|
assert 'find "$CONF_DIR" "$STATE_DIR" -type f -exec chmod 600' in installer
|
|
assert 'systemctl stop "$SERVICE_NAME"' in installer
|
|
assert "migrate_service_layout.py" in installer
|
|
assert "rsync -a --delete" in installer
|
|
assert "rsync -a --update" not in installer
|
|
assert "source-authoritative secure install" in installer
|
|
assert 'VENV_BUILD="$INSTALL_DIR/.venv-build-$$"' in installer
|
|
assert 'Preserving existing virtual environment' not in installer
|
|
assert "Python 3.10+ installed" in installer
|
|
assert "sys.version_info < (3, 10)" in installer
|
|
|
|
|
|
def _service_state_test_environment(tmp_path: Path) -> tuple[dict[str, str], Path, Path]:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_state = tmp_path / "systemctl-state"
|
|
maint_state = tmp_path / "maint-state"
|
|
fake_bin.mkdir()
|
|
fake_state.mkdir()
|
|
maint_state.mkdir()
|
|
systemctl = fake_bin / "systemctl"
|
|
systemctl.write_text(
|
|
"""#!/bin/bash
|
|
set -e
|
|
state="${FAKE_SYSTEMCTL_STATE:?}"
|
|
case "${1:-}" in
|
|
is-active) [ -f "$state/active" ] ;;
|
|
is-enabled) [ -f "$state/enabled" ] ;;
|
|
start) touch "$state/active" ;;
|
|
stop) rm -f "$state/active" ;;
|
|
enable) touch "$state/enabled" ;;
|
|
disable) rm -f "$state/enabled" ;;
|
|
daemon-reload) : ;;
|
|
*) exit 2 ;;
|
|
esac
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
systemctl.chmod(0o755)
|
|
env = os.environ.copy()
|
|
env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}"
|
|
env["FAKE_SYSTEMCTL_STATE"] = str(fake_state)
|
|
env["MESHCORE_MAINT_STATE_DIR"] = str(maint_state)
|
|
env["MESHCORE_MAINT_POLICY_MARKER"] = str(tmp_path / "policy-marker")
|
|
return env, fake_state, maint_state
|
|
|
|
|
|
def _run_service_state(script: Path, *args: str, env: dict[str, str]) -> None:
|
|
subprocess.run(
|
|
["bash", str(script), *args],
|
|
check=True,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
def test_first_hardened_deb_upgrade_survives_legacy_prerm(tmp_path: Path) -> None:
|
|
"""Model Debian's old prerm -> new preinst -> new postinst ordering."""
|
|
env, service_state, maint_state = _service_state_test_environment(tmp_path)
|
|
helper = REPO_ROOT / "scripts/debian_service_state.sh"
|
|
preinst = tmp_path / "preinst"
|
|
shutil.copy2(helper, preinst)
|
|
|
|
# The baseline package's old prerm destroys state before new code runs.
|
|
# The one-time migration policy favors availability and records completion.
|
|
(service_state / "enabled").touch()
|
|
(service_state / "active").touch()
|
|
(service_state / "enabled").unlink()
|
|
(service_state / "active").unlink()
|
|
_run_service_state(preinst, "upgrade", "0.9.2", env=env)
|
|
assert not list(maint_state.iterdir())
|
|
|
|
_run_service_state(helper, "restore", "0.9.2", env=env)
|
|
|
|
assert (service_state / "enabled").is_file()
|
|
assert (service_state / "active").is_file()
|
|
assert not list(maint_state.iterdir())
|
|
assert Path(env["MESHCORE_MAINT_POLICY_MARKER"]).is_file()
|
|
|
|
|
|
def test_deb_upgrade_preserves_intentionally_disabled_stopped_service(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env, service_state, maint_state = _service_state_test_environment(tmp_path)
|
|
helper = REPO_ROOT / "scripts/debian_service_state.sh"
|
|
preinst = tmp_path / "preinst"
|
|
shutil.copy2(helper, preinst)
|
|
Path(env["MESHCORE_MAINT_POLICY_MARKER"]).touch()
|
|
|
|
# Hardened old prerm runs first and writes no state markers because this
|
|
# service was intentionally disabled/stopped. New preinst preserves that.
|
|
_run_service_state(helper, "preremove", "upgrade", env=env)
|
|
_run_service_state(preinst, "upgrade", "0.9.2", env=env)
|
|
_run_service_state(helper, "restore", "0.9.2", env=env)
|
|
|
|
assert not (service_state / "enabled").exists()
|
|
assert not (service_state / "active").exists()
|
|
assert not list(maint_state.iterdir())
|
|
|
|
|
|
def test_later_deb_upgrade_preserves_enabled_active_service(tmp_path: Path) -> None:
|
|
env, service_state, maint_state = _service_state_test_environment(tmp_path)
|
|
helper = REPO_ROOT / "scripts/debian_service_state.sh"
|
|
preinst = tmp_path / "preinst"
|
|
shutil.copy2(helper, preinst)
|
|
Path(env["MESHCORE_MAINT_POLICY_MARKER"]).touch()
|
|
(service_state / "enabled").touch()
|
|
(service_state / "active").touch()
|
|
|
|
# Real Debian order: hardened old prerm, new preinst, new postinst.
|
|
_run_service_state(helper, "preremove", "upgrade", env=env)
|
|
assert not (service_state / "active").exists()
|
|
assert (maint_state / "meshcore-bot-was-enabled").is_file()
|
|
assert (maint_state / "meshcore-bot-was-active").is_file()
|
|
_run_service_state(preinst, "upgrade", "0.9.2", env=env)
|
|
_run_service_state(helper, "restore", "0.9.2", env=env)
|
|
|
|
assert (service_state / "enabled").is_file()
|
|
assert (service_state / "active").is_file()
|
|
assert not list(maint_state.iterdir())
|
|
|
|
|
|
def test_relative_legacy_paths_are_migrated_without_losing_data(tmp_path):
|
|
# URI-significant characters prove the read-only SQLite URI is encoded.
|
|
legacy = tmp_path / "legacy #1?"
|
|
state = tmp_path / "state"
|
|
logs = tmp_path / "logs"
|
|
config_dir = tmp_path / "config"
|
|
legacy.mkdir()
|
|
config_dir.mkdir()
|
|
(legacy / "local" / "commands").mkdir(parents=True)
|
|
(legacy / "local" / "commands" / "custom.py").write_text("VALUE = 1\n", encoding="utf-8")
|
|
(legacy / "meshcore_bot.log").write_text("old log\n", encoding="utf-8")
|
|
|
|
db_path = legacy / "meshcore_bot.db"
|
|
connection = sqlite3.connect(db_path)
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("CREATE TABLE messages (body TEXT)")
|
|
connection.execute("INSERT INTO messages VALUES ('preserved')")
|
|
connection.commit()
|
|
|
|
config_path = config_dir / "config.ini"
|
|
config_path.write_text(
|
|
"# keep this comment\n"
|
|
"[Bot]\n"
|
|
"db_path = meshcore_bot.db\n"
|
|
"local_dir_path = local\n"
|
|
"\n[Logging]\n"
|
|
"log_file = meshcore_bot.log\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
try:
|
|
updates = migrate_service_layout(config_path, legacy, state, logs)
|
|
finally:
|
|
connection.close()
|
|
|
|
assert set(updates) == {"Bot", "Logging"}
|
|
migrated = sqlite3.connect(state / "meshcore_bot.db")
|
|
try:
|
|
assert migrated.execute("SELECT body FROM messages").fetchall() == [("preserved",)]
|
|
finally:
|
|
migrated.close()
|
|
assert (state / "local" / "commands" / "custom.py").is_file()
|
|
assert (logs / "meshcore_bot.log").read_text(encoding="utf-8") == "old log\n"
|
|
assert config_path.read_text(encoding="utf-8").startswith("# keep this comment\n")
|
|
|
|
parsed = configparser.ConfigParser()
|
|
parsed.read(config_path)
|
|
assert parsed.get("Bot", "db_path") == str((state / "meshcore_bot.db").resolve())
|
|
assert parsed.get("Bot", "local_dir_path") == str((state / "local").resolve())
|
|
assert parsed.get("Logging", "log_file") == str((logs / "meshcore_bot.log").resolve())
|
|
|
|
|
|
def test_absolute_custom_paths_are_preserved(tmp_path):
|
|
config_path = tmp_path / "config.ini"
|
|
config_path.write_text(
|
|
"[Bot]\n"
|
|
f"db_path = {tmp_path / 'custom.db'}\n"
|
|
f"local_dir_path = {tmp_path / 'custom-local'}\n"
|
|
"[Logging]\nlog_file = \n",
|
|
encoding="utf-8",
|
|
)
|
|
before = config_path.read_text(encoding="utf-8")
|
|
|
|
updates = migrate_service_layout(
|
|
config_path,
|
|
tmp_path / "legacy",
|
|
tmp_path / "state",
|
|
tmp_path / "logs",
|
|
)
|
|
|
|
assert updates == {}
|
|
assert config_path.read_text(encoding="utf-8") == before
|
|
|
|
|
|
def test_relative_legacy_database_replaces_but_preserves_stale_target(tmp_path):
|
|
legacy = tmp_path / "legacy"
|
|
state = tmp_path / "state"
|
|
logs = tmp_path / "logs"
|
|
legacy.mkdir()
|
|
state.mkdir()
|
|
logs.mkdir()
|
|
|
|
def create_db(path: Path, marker: str) -> None:
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("CREATE TABLE marker (value TEXT)")
|
|
conn.execute("INSERT INTO marker VALUES (?)", (marker,))
|
|
conn.commit()
|
|
|
|
create_db(legacy / "meshcore_bot.db", "current-legacy")
|
|
create_db(state / "meshcore_bot.db", "stale-target")
|
|
config_path = tmp_path / "config.ini"
|
|
config_path.write_text("[Bot]\ndb_path = meshcore_bot.db\n", encoding="utf-8")
|
|
|
|
migrate_service_layout(config_path, legacy, state, logs)
|
|
|
|
with sqlite3.connect(state / "meshcore_bot.db") as conn:
|
|
assert conn.execute("SELECT value FROM marker").fetchone() == ("current-legacy",)
|
|
backups = list(state.glob("meshcore_bot.pre-layout-*.db"))
|
|
assert len(backups) == 1
|
|
with sqlite3.connect(backups[0]) as conn:
|
|
assert conn.execute("SELECT value FROM marker").fetchone() == ("stale-target",)
|
|
|
|
|
|
def test_split_bot_and_viewer_databases_migrate_without_filename_collision(tmp_path):
|
|
legacy = tmp_path / "legacy"
|
|
state = tmp_path / "state"
|
|
logs = tmp_path / "logs"
|
|
(legacy / "bot").mkdir(parents=True)
|
|
(legacy / "viewer").mkdir(parents=True)
|
|
|
|
for path, marker in (
|
|
(legacy / "bot" / "shared.db", "bot"),
|
|
(legacy / "viewer" / "shared.db", "viewer"),
|
|
):
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("CREATE TABLE marker (value TEXT)")
|
|
conn.execute("INSERT INTO marker VALUES (?)", (marker,))
|
|
conn.commit()
|
|
|
|
config_path = tmp_path / "config.ini"
|
|
config_path.write_text(
|
|
"[Bot]\n"
|
|
"db_path = bot/shared.db\n"
|
|
"[Web_Viewer]\n"
|
|
"db_path = viewer/shared.db\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
migrate_service_layout(config_path, legacy, state, logs)
|
|
|
|
parsed = configparser.ConfigParser()
|
|
parsed.read(config_path)
|
|
bot_target = Path(parsed.get("Bot", "db_path"))
|
|
viewer_target = Path(parsed.get("Web_Viewer", "db_path"))
|
|
assert bot_target != viewer_target
|
|
with sqlite3.connect(bot_target) as conn:
|
|
assert conn.execute("SELECT value FROM marker").fetchone() == ("bot",)
|
|
with sqlite3.connect(viewer_target) as conn:
|
|
assert conn.execute("SELECT value FROM marker").fetchone() == ("viewer",)
|
|
|
|
|
|
def test_legacy_local_plugins_replace_and_preserve_stale_target_tree(tmp_path):
|
|
legacy = tmp_path / "legacy"
|
|
state = tmp_path / "state"
|
|
logs = tmp_path / "logs"
|
|
(legacy / "local" / "commands").mkdir(parents=True)
|
|
(legacy / "local" / "commands" / "current.py").write_text("CURRENT = 1\n")
|
|
(state / "local" / "commands").mkdir(parents=True)
|
|
(state / "local" / "commands" / "stale.py").write_text("STALE = 1\n")
|
|
config_path = tmp_path / "config.ini"
|
|
config_path.write_text(
|
|
"[Bot]\n"
|
|
f"db_path = {tmp_path / 'absolute.db'}\n"
|
|
"local_dir_path = local\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
migrate_service_layout(config_path, legacy, state, logs)
|
|
|
|
assert (state / "local" / "commands" / "current.py").is_file()
|
|
assert not (state / "local" / "commands" / "stale.py").exists()
|
|
backups = list(state.glob("local.pre-layout-*"))
|
|
assert len(backups) == 1
|
|
assert (backups[0] / "commands" / "stale.py").is_file()
|