vnctalk-prosody/tests/test_12_image_runtime.py

"""Image/runtime verification (test-improvement.md Phase 6).

  * mod_http_upload_external slot handshake + signed-URL PUT (live XMPP + mock)
  * healthcheck.sh exit 2 on a tampered cert (docker compose exec)
  * rendered config has no residual ${...} placeholders (docker compose exec)

The two docker-exec tests automate MANUAL §3.1 and the latent envsubst bug
documented in test-improvement.md §1.2. They skip when the compose stack is
not running.
"""
import asyncio
import os
import shutil
import subprocess
import uuid
import xml.etree.ElementTree as ET
from urllib.parse import urlparse

import pytest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))


def _compose_available():
    if not shutil.which("docker"):
        return False
    try:
        res = subprocess.run(
            ["docker", "compose", "ps", "--services", "--filter", "status=running"],
            cwd=REPO_ROOT, capture_output=True, text=True, timeout=20,
        )
    except (subprocess.SubprocessError, OSError):
        return False
    return res.returncode == 0 and "prosody" in res.stdout.split()


def _compose_exec(args, timeout=30, user=None):
    cmd = ["docker", "compose", "exec"]
    if user:
        cmd += ["--user", user]
    cmd += ["-T", "prosody"] + args
    return subprocess.run(cmd, cwd=REPO_ROOT, capture_output=True, text=True, timeout=timeout)


@pytest.mark.asyncio
class TestHttpUploadExternal:
    """mod_http_upload_external slot handshake (XEP-0363) + file PUT."""

    async def test_http_upload_slot_handshake(self, xmpp_client, xmpp_config, mock_client):
        domain = xmpp_config.get("domain")
        if not domain:
            pytest.skip("No XMPP domain configured")

        # XEP-0363 slot request is an IQ-get to the vhost (http_upload_external
        # is loaded on the VirtualHost and hooks iq/host/<ns>:request).
        iq = xmpp_client.make_iq_get(ito=domain)
        req = ET.SubElement(iq.xml, "{urn:xmpp:http:upload:0}request")
        req.set("filename", "smoke.txt")
        req.set("size", "4")
        res = await iq.send(timeout=10)

        slot = res.xml.find("{urn:xmpp:http:upload:0}slot")
        assert slot is not None, "no <slot> in http_upload response — module not loaded"
        put_el = slot.find("{urn:xmpp:http:upload:0}put")
        get_el = slot.find("{urn:xmpp:http:upload:0}get")
        assert put_el is not None and get_el is not None, "slot missing put/get urls"
        put_url = put_el.text or put_el.get("url")
        get_url = get_el.text or get_el.get("url")
        assert put_url and "/share/" in put_url, f"put_url not on mock share: {put_url}"

        # The returned URLs reference the internal `mocks` host. Re-PUT through
        # the mock_client session (same container network) by path so the test
        # works whether run from the tester container or the host.
        put_path = urlparse(put_url).path
        await mock_client.reset()
        async with mock_client.session.put(
            f"{mock_client.base_url}{put_path}", data=b"DATA"
        ) as r:
            assert r.status == 201, f"PUT to slot target failed: {r.status}"

        captured = await mock_client.wait_for(put_path, count=1, timeout=5)
        assert captured, "slot PUT was not recorded by the mock"
        assert captured[0]["method"] == "PUT"


class TestHealthcheckExecution:
    """Run healthcheck.sh inside the container with a tampered cert."""

    def test_healthcheck_exit2_on_cert_mismatch(self):
        if not _compose_available():
            pytest.skip("compose stack not running")
        # Backup current tls.crt (if any) so we can restore the original state.
        script = (
            "cp /etc/prosody/certs/prosody-ssl.pem /tmp/healthcheck_orig.pem 2>/dev/null; "
            "mkdir -p /etc/tls-update; "
            "echo tampered-cert-$(date +%s) > /etc/tls-update/tls.crt; "
            "/vnc/config/healthcheck.sh >/dev/null 2>&1; echo $?; "
            "cp /tmp/healthcheck_orig.pem /etc/tls-update/tls.crt 2>/dev/null"
        )
        res = _compose_exec(["sh", "-c", script], user="0:0", timeout=30)
        out = res.stdout.strip().splitlines()
        code = out[-1] if out else ""
        assert code == "2", (
            f"healthcheck exited {code!r} on tampered cert, expected 2. stdout={res.stdout!r}"
        )

    def test_healthcheck_not_exit2_after_restore(self):
        if not _compose_available():
            pytest.skip("compose stack not running")
        # Ensure tls.crt matches the loaded cert (copy ssl.pem into tls.crt),
        # then healthcheck must NOT exit 2 (it may exit 0/1 depending on telnet).
        script = (
            "mkdir -p /etc/tls-update; "
            "cp /etc/prosody/certs/prosody-ssl.pem /etc/tls-update/tls.crt; "
            "/vnc/config/healthcheck.sh >/dev/null 2>&1; echo $?"
        )
        res = _compose_exec(["sh", "-c", script], user="0:0", timeout=30)
        out = res.stdout.strip().splitlines()
        code = out[-1] if out else ""
        assert code != "2", (
            f"healthcheck exited 2 with matching certs, expected 0/1. stdout={res.stdout!r}"
        )


class TestConfigRendering:
    """No residual ${...} placeholders in the rendered config."""

    def test_config_rendering_no_residual_placeholders(self):
        if not _compose_available():
            pytest.skip("compose stack not running")
        res = _compose_exec(["sh", "-c", "grep -c '\\${' /etc/prosody/prosody.cfg.lua || true"], timeout=20)
        out = res.stdout.strip()
        count = int(out) if out.isdigit() else 0
        assert count == 0, (
            f"rendered config has {count} residual ${{...}} placeholders — "
            "an env var is missing or misnamed (see test-improvement.md §1.2)"
        )