315 lines
13 KiB
Python
315 lines
13 KiB
Python
import ast
|
|
import re
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _read(name: str) -> str:
|
|
return (ROOT / name).read_text(encoding="utf-8")
|
|
|
|
|
|
def _docker_logical_lines(text: str) -> list[str]:
|
|
logical: list[str] = []
|
|
pending = ""
|
|
for raw_line in text.splitlines():
|
|
stripped = raw_line.strip()
|
|
if not stripped or (not pending and stripped.startswith("#")):
|
|
continue
|
|
pending = f"{pending} {stripped}".strip()
|
|
if pending.endswith("\\"):
|
|
pending = pending[:-1].rstrip()
|
|
continue
|
|
logical.append(pending)
|
|
pending = ""
|
|
if pending:
|
|
logical.append(pending)
|
|
return logical
|
|
|
|
|
|
def _environment_map(value: Any) -> dict[str, str | None]:
|
|
if isinstance(value, dict):
|
|
return {str(key): None if item is None else str(item) for key, item in value.items()}
|
|
result: dict[str, str | None] = {}
|
|
for entry in value or []:
|
|
key, separator, item = str(entry).partition("=")
|
|
result[key] = item if separator else None
|
|
return result
|
|
|
|
|
|
def _volume_mounts(value: Any) -> list[tuple[str | None, str, bool]]:
|
|
mounts: list[tuple[str | None, str, bool]] = []
|
|
for entry in value or []:
|
|
if isinstance(entry, dict):
|
|
target = str(entry.get("target") or "")
|
|
if target:
|
|
mounts.append(
|
|
(
|
|
None if entry.get("source") is None else str(entry["source"]),
|
|
target,
|
|
bool(entry.get("read_only", False)),
|
|
)
|
|
)
|
|
continue
|
|
# Greedy source matching tolerates Compose substitutions containing
|
|
# `:-` or `:?` while keeping the absolute container target distinct.
|
|
match = re.match(r"^(.*):(/[^:]+)(?::([^:]+))?$", str(entry))
|
|
if match:
|
|
mode = match.group(3) or ""
|
|
mounts.append((match.group(1), match.group(2), "ro" in mode.split(",")))
|
|
return mounts
|
|
|
|
|
|
def _secret_targets(value: Any) -> set[str]:
|
|
targets: set[str] = set()
|
|
for entry in value or []:
|
|
if isinstance(entry, dict):
|
|
source = str(entry.get("source") or "")
|
|
target = str(entry.get("target") or source)
|
|
else:
|
|
source = str(entry)
|
|
target = source
|
|
if not target:
|
|
continue
|
|
targets.add(target if target.startswith("/") else f"/run/secrets/{target}")
|
|
return targets
|
|
|
|
|
|
class RequirementPinTests(unittest.TestCase):
|
|
def test_runtime_requirements_are_exactly_pinned(self):
|
|
path = ROOT / "requirements.txt"
|
|
self.assertTrue(path.is_file(), "requirements.txt is missing")
|
|
requirements: dict[str, str] = {}
|
|
pattern = re.compile(
|
|
r"^([A-Za-z0-9_.-]+)(?:\[[A-Za-z0-9_,.-]+\])?==([^;\s]+)(?:\s*;.*)?$"
|
|
)
|
|
for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
line = raw_line.split("#", 1)[0].strip()
|
|
if not line:
|
|
continue
|
|
match = pattern.fullmatch(line)
|
|
self.assertIsNotNone(
|
|
match,
|
|
f"requirements.txt:{line_number} is not an exact == pin: {line!r}",
|
|
)
|
|
assert match is not None
|
|
normalized = re.sub(r"[-_.]+", "-", match.group(1)).lower()
|
|
requirements[normalized] = match.group(2)
|
|
|
|
for project in ("flask", "gunicorn", "paho-mqtt", "pyyaml", "yt-dlp"):
|
|
self.assertIn(project, requirements, f"missing runtime dependency {project}")
|
|
self.assertNotIn("*", requirements[project])
|
|
|
|
|
|
class DockerfileTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.path = ROOT / "Dockerfile"
|
|
self.assertTrue(self.path.is_file(), "Dockerfile is missing")
|
|
self.text = self.path.read_text(encoding="utf-8")
|
|
self.lines = _docker_logical_lines(self.text)
|
|
|
|
def test_image_copies_the_v2_runtime(self):
|
|
copies = "\n".join(
|
|
line for line in self.lines if line.upper().startswith("COPY ")
|
|
)
|
|
for required in (
|
|
"controller",
|
|
"config.yaml",
|
|
"app.py",
|
|
"wsgi.py",
|
|
"gunicorn.conf.py",
|
|
):
|
|
self.assertIn(required, copies, f"Dockerfile does not copy {required}")
|
|
|
|
def test_image_runs_non_root_on_the_controller_port(self):
|
|
users = [line.split(None, 1)[1] for line in self.lines if line.upper().startswith("USER ")]
|
|
self.assertTrue(users, "Dockerfile must set a non-root USER")
|
|
self.assertNotIn(users[-1].strip().lower(), {"root", "0", "0:0"})
|
|
self.assertRegex(self.text, r"(?im)^\s*EXPOSE\s+7531(?:/tcp)?\s*$")
|
|
|
|
def test_image_has_an_http_healthcheck_and_runs_gunicorn(self):
|
|
self.assertRegex(self.text, r"(?im)^\s*HEALTHCHECK\b")
|
|
self.assertIn("/healthz", self.text)
|
|
launch = "\n".join(
|
|
line
|
|
for line in self.lines
|
|
if line.upper().startswith(("CMD ", "ENTRYPOINT "))
|
|
).lower()
|
|
self.assertIn("gunicorn", launch)
|
|
|
|
def test_image_has_no_unbounded_yt_dlp_upgrade_or_hidden_global_config(self):
|
|
self.assertNotRegex(
|
|
self.text,
|
|
r"(?i)(?:--upgrade|-U)\s+[\"']?(?:yt-dlp|yt_dlp)",
|
|
)
|
|
self.assertNotIn("/etc/yt-dlp.conf", self.text)
|
|
|
|
def test_image_verifies_a_supported_node_runtime_for_yt_dlp(self):
|
|
self.assertRegex(
|
|
self.text,
|
|
r"(?im)^FROM\s+node:22\.\d+\.\d+-bookworm-slim\s+AS\s+",
|
|
)
|
|
self.assertIn("NodeJsRuntime", self.text)
|
|
self.assertIn("info.supported", self.text)
|
|
|
|
|
|
class GunicornConfigTests(unittest.TestCase):
|
|
def test_gunicorn_uses_exactly_one_worker_without_preload(self):
|
|
path = ROOT / "gunicorn.conf.py"
|
|
self.assertTrue(path.is_file(), "gunicorn.conf.py is missing")
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
assignments: dict[str, Any] = {}
|
|
for node in tree.body:
|
|
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
|
continue
|
|
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
|
try:
|
|
value = ast.literal_eval(node.value)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
for target in targets:
|
|
if isinstance(target, ast.Name):
|
|
assignments[target.id] = value
|
|
|
|
self.assertEqual(assignments.get("workers"), 1)
|
|
self.assertIs(assignments.get("preload_app"), False)
|
|
self.assertIs(assignments.get("control_socket_disable"), True)
|
|
self.assertEqual(assignments.get("worker_tmp_dir"), "/tmp")
|
|
|
|
|
|
class ComposeRuntimeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.path = ROOT / "docker-compose.yml"
|
|
self.assertTrue(self.path.is_file(), "docker-compose.yml is missing")
|
|
self.text = self.path.read_text(encoding="utf-8")
|
|
document = yaml.safe_load(self.text)
|
|
self.assertIsInstance(document, dict)
|
|
self.document = document
|
|
self.service = document.get("services", {}).get("controller")
|
|
self.assertIsInstance(self.service, dict, "Compose controller service is missing")
|
|
|
|
def test_compose_requires_a_real_flask_secret_and_publishes_port_7531(self):
|
|
environment = _environment_map(self.service.get("environment"))
|
|
secret = environment.get("FLASK_SECRET_KEY")
|
|
self.assertIsNotNone(secret, "FLASK_SECRET_KEY environment setting is missing")
|
|
assert secret is not None
|
|
self.assertTrue(secret.startswith("${FLASK_SECRET_KEY"))
|
|
self.assertIn("?", secret, "FLASK_SECRET_KEY must use required Compose substitution")
|
|
self.assertNotIn("change-me", secret.lower())
|
|
self.assertNotIn("dev-secret", secret.lower())
|
|
|
|
published_targets: set[str] = set()
|
|
for entry in self.service.get("ports") or []:
|
|
if isinstance(entry, dict):
|
|
published_targets.add(str(entry.get("target")))
|
|
else:
|
|
match = re.search(r":(\d+)(?:/(?:tcp|udp))?$", str(entry))
|
|
if match:
|
|
published_targets.add(match.group(1))
|
|
self.assertIn("7531", published_targets)
|
|
|
|
def test_compose_passes_all_supported_mqtt_environment_overrides(self):
|
|
environment = _environment_map(self.service.get("environment"))
|
|
for name in (
|
|
"MQTT_HOST",
|
|
"MQTT_PORT",
|
|
"MQTT_USERNAME",
|
|
"MQTT_PASSWORD",
|
|
"MQTT_CLIENT_ID",
|
|
):
|
|
with self.subTest(name=name):
|
|
self.assertIn(name, environment)
|
|
|
|
def test_compose_uses_dedicated_data_media_and_cookie_mounts(self):
|
|
mounts = _volume_mounts(self.service.get("volumes"))
|
|
by_target = {target: (source, read_only) for source, target, read_only in mounts}
|
|
self.assertIn("/controller-data", by_target)
|
|
self.assertIn("/mnt/Media/YTPlayer", by_target)
|
|
self.assertNotEqual(by_target["/controller-data"][0], "/home/media/Scripts/Video-Player")
|
|
|
|
cookie_path = "/run/secrets/yt-dlp-cookies.txt"
|
|
mounted_cookie = cookie_path in by_target and by_target[cookie_path][1]
|
|
mounted_secret_directory = (
|
|
"/run/secrets" in by_target and by_target["/run/secrets"][1]
|
|
)
|
|
secret_cookie = cookie_path in _secret_targets(self.service.get("secrets"))
|
|
self.assertTrue(
|
|
mounted_cookie or mounted_secret_directory or secret_cookie,
|
|
f"cookies must be read-only at exactly {cookie_path}",
|
|
)
|
|
self.assertNotIn("/cookies_ro.txt", self.text)
|
|
self.assertNotRegex(
|
|
self.text,
|
|
r"/home/media/Scripts/Video-Player(?=[:\s\"'])",
|
|
)
|
|
|
|
def test_compose_enables_container_hardening_and_bounded_operations(self):
|
|
self.assertIs(self.service.get("read_only"), True)
|
|
self.assertIs(self.service.get("init"), True)
|
|
|
|
cap_drop = self.service.get("cap_drop") or []
|
|
if isinstance(cap_drop, str):
|
|
cap_drop = [cap_drop]
|
|
self.assertIn("ALL", {str(value).upper() for value in cap_drop})
|
|
|
|
security_options = self.service.get("security_opt") or []
|
|
if isinstance(security_options, str):
|
|
security_options = [security_options]
|
|
self.assertTrue(
|
|
any("no-new-privileges" in str(value).lower() for value in security_options)
|
|
)
|
|
|
|
tmpfs = self.service.get("tmpfs") or []
|
|
if isinstance(tmpfs, str):
|
|
tmpfs = [tmpfs]
|
|
self.assertTrue(any(str(value).split(":", 1)[0] == "/tmp" for value in tmpfs))
|
|
self.assertTrue(any("noexec" in str(value) for value in tmpfs))
|
|
|
|
healthcheck = self.service.get("healthcheck")
|
|
self.assertIsInstance(healthcheck, dict)
|
|
self.assertIn("/healthz", str(healthcheck.get("test")))
|
|
|
|
logging_config = self.service.get("logging")
|
|
self.assertIsInstance(logging_config, dict)
|
|
options = logging_config.get("options") or {}
|
|
self.assertTrue(options.get("max-size"), "logging max-size is missing")
|
|
self.assertTrue(options.get("max-file"), "logging max-file is missing")
|
|
grace = str(self.service.get("stop_grace_period") or "").rstrip("s")
|
|
self.assertGreaterEqual(float(grace), 30.0)
|
|
|
|
def test_compose_preserves_the_external_tabernet_network(self):
|
|
service_networks = self.service.get("networks") or []
|
|
if isinstance(service_networks, dict):
|
|
service_networks = service_networks.keys()
|
|
self.assertIn("tabernet", service_networks)
|
|
tabernet = self.document.get("networks", {}).get("tabernet")
|
|
self.assertIsInstance(tabernet, dict)
|
|
self.assertIs(tabernet.get("external"), True)
|
|
|
|
|
|
class DockerIgnoreTests(unittest.TestCase):
|
|
def test_dockerignore_excludes_secrets_and_python_caches(self):
|
|
path = ROOT / ".dockerignore"
|
|
self.assertTrue(path.is_file(), ".dockerignore is missing")
|
|
patterns = {
|
|
line.strip().lower()
|
|
for line in path.read_text(encoding="utf-8").splitlines()
|
|
if line.strip() and not line.lstrip().startswith("#")
|
|
}
|
|
self.assertTrue(any("__pycache__" in pattern for pattern in patterns))
|
|
self.assertTrue(
|
|
any("pyc" in pattern or "py[cod]" in pattern for pattern in patterns)
|
|
)
|
|
self.assertTrue(any(pattern in {".env", ".env*", "*.env"} for pattern in patterns))
|
|
self.assertTrue(any("secret" in pattern for pattern in patterns))
|
|
self.assertTrue(any("cookie" in pattern for pattern in patterns))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|