picpeak/tests/nginx/test_request_logging.py

"""Run with python3 tests/nginx/test_request_logging.py; requires Docker and openssl."""

import http.client
import os
from pathlib import Path
import re
import ssl
import subprocess
import tempfile
import time
import unittest
import uuid


ROOT = Path(__file__).resolve().parents[2]
IMAGE = os.environ.get("NGINX_TEST_IMAGE", "nginx:1.30-alpine")
CONFIGS = (
    "nginx/nginx.conf",
    "nginx/nginx.conf.example",
    "frontend/nginx.conf",
    "frontend/nginx.dev.conf",
)
TOKEN = "NGINX_TEST_CAPABILITY_73cb40b17c"
FORWARDED_IP = "198.51.100.9"


def run(*args):
    return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT, timeout=120).strip()


class RequestLoggingTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.temporary = tempfile.TemporaryDirectory(prefix="picpeak-nginx-test-")
        cls.addClassCleanup(cls.temporary.cleanup)
        cls.directory = Path(cls.temporary.name)
        cls.network = "picpeak-nginx-test-" + uuid.uuid4().hex[:12]
        run("docker", "network", "create", cls.network)
        cls.addClassCleanup(run, "docker", "network", "rm", cls.network)
        certs = cls.directory / "certs"
        certs.mkdir()
        run("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
            "-keyout", str(certs / "privkey.pem"), "-out", str(certs / "fullchain.pem"),
            "-days", "1", "-subj", "/CN=your-domain.com",
            "-addext", "subjectAltName=IP:127.0.0.1,DNS:your-domain.com")

    def configuration(self, filename, critical, unsafe_control):
        source = (ROOT / filename).read_text()
        if unsafe_control:
            # Positive control: the same request must expose the fixture token
            # when native request errors are enabled, even at crit severity.
            source, count = re.subn(r"error_log\s+/dev/null\s*;",
                                   "error_log /var/log/nginx/error.log crit;", source)
            self.assertEqual(count, 1)

        # A source address absent from this disposable network makes bind()
        # fail at CRIT deterministically, without exhausting runner resources.
        bind = "proxy_bind 192.0.2.1;" if critical else ""
        if filename.startswith("frontend/"):
            (self.directory / "frontend.conf").write_text(source)
            source = """worker_processes 1;
error_log /var/log/nginx/error.log notice;
events { worker_connections 128; }
http {
    include /etc/nginx/mime.types;
    %s
    include /fixtures/frontend.conf;
}
""" % bind
        else:
            if "worker_processes auto;" in source:
                source = source.replace("worker_processes auto;", "worker_processes 1;")
            else:
                source = "worker_processes 1;\n" + source
            source = source.replace("http {", "http {\n    " + bind, 1)

        (self.directory / "nginx.conf").write_text(source)
        # The standalone main configuration intentionally delegates vhosts.
        (self.directory / "site.conf").write_text("""server {
    listen 80;
    server_name localhost;
    location /api { proxy_pass http://backend:3000; }
}
""")

    def request(self, port, tls, uri):
        # Trust only our fixture certificate while retaining hostname checks.
        context = ssl.create_default_context(cafile=str(self.directory / "certs" / "fullchain.pem"))
        for attempt in range(30):
            connection = (http.client.HTTPSConnection("127.0.0.1", port, context=context, timeout=3)
                          if tls else http.client.HTTPConnection("127.0.0.1", port, timeout=3))
            try:
                connection.request("GET", uri, headers={
                    "Host": "your-domain.com" if tls else "localhost",
                    "X-Forwarded-For": FORWARDED_IP,
                    "Referer": "https://example.invalid/gallery?token=" + TOKEN,
                })
                response = connection.getresponse()
                response.read()
                return response.status
            except (ConnectionError, OSError):
                if attempt == 29:
                    raise
                time.sleep(0.1)
            finally:
                connection.close()

    def exercise(self, filename, critical=False, unsafe_control=False):
        self.configuration(filename, critical, unsafe_control)
        name = "picpeak-nginx-test-" + uuid.uuid4().hex[:12]
        try:
            # DNS aliases cover both static and variable-based proxy_pass in
            # the supplied configs. Nothing listens on the fixture's port 3000.
            run("docker", "run", "-d", "--name", name, "--network", self.network,
                "--network-alias", "backend", "--network-alias", "frontend",
                "--sysctl", "net.ipv4.ip_nonlocal_bind=0",
                "-p", "127.0.0.1::80", "-p", "127.0.0.1::443",
                "-v", f"{self.directory}:/fixtures:ro",
                "-v", f"{self.directory}/site.conf:/etc/nginx/sites-enabled/test.conf:ro",
                "-v", f"{self.directory}/certs:/etc/letsencrypt/live/your-domain.com:ro",
                IMAGE, "nginx", "-c", "/fixtures/nginx.conf", "-g", "daemon off;")
            syntax = run("docker", "exec", name, "nginx", "-t", "-c", "/fixtures/nginx.conf")
            self.assertIn("test is successful", syntax)
            tls = filename.endswith(".example")
            port = int(run("docker", "port", name, "443/tcp" if tls else "80/tcp").split(":")[-1])
            paths = (
                f"/api/customer/auth/password-reset/{TOKEN}?token={TOKEN}",
                f"/api/secure-images/gallery/secure/1/{TOKEN}",
                f"/api/public/newsletter/unsubscribe/{TOKEN}",
            )
            for uri in paths:
                self.assertIn(self.request(port, tls, uri), (500, 502))
            # Wait for the final access entry to reach the container log pipe.
            for _ in range(30):
                logs = run("docker", "logs", name)
                if '"GET /api/public"' in logs:
                    break
                time.sleep(0.05)
            for surface in ("customer", "secure-images", "public"):
                self.assertRegex(logs, rf'"GET /api/{surface}" 50[02] \d+ \d+\.\d+')
            self.assertIn(FORWARDED_IP, logs)
            if unsafe_control:
                self.assertIn("[crit]", logs)
                self.assertIn("bind(192.0.2.1) failed", logs)
                self.assertIn(TOKEN, logs)
            else:
                self.assertNotIn(TOKEN, logs)
        finally:
            run("docker", "rm", "-f", name)

    def test_tokens_stay_out_of_logs_for_ordinary_and_critical_upstream_errors(self):
        for filename in CONFIGS:
            for scenario in ("ordinary", "critical", "unsafe-control"):
                with self.subTest(config=filename, scenario=scenario, image=IMAGE):
                    self.exercise(filename, critical=scenario != "ordinary",
                                  unsafe_control=scenario == "unsafe-control")


if __name__ == "__main__":
    unittest.main(verbosity=2)