Add loopback HTTPS smoke test

2026-10-03 14:02:11 +03:00
parent ceae822302
commit 052e600e32
3 changed files with 155 additions and 1 deletions
@@ -0,0 +1,152 @@
+"""Launch FastAPI on loopback with a disposable certificate and verify HTTPS."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import httpx
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def find_openssl() -> str:
+    found = shutil.which("openssl")
+    if found:
+        return found
+    candidates = (
+        Path(r"C:\Program Files\Git\usr\bin\openssl.exe"),
+        Path(r"C:\Program Files\Git\mingw64\bin\openssl.exe"),
+    )
+    for candidate in candidates:
+        if candidate.is_file():
+            return str(candidate)
+    raise RuntimeError("OpenSSL is required for this local TLS smoke test")
+
+
+def main() -> int:
+    openssl = find_openssl()
+    runtime_root = ROOT / ".test-runtime"
+    runtime_root.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix="sovereignai-tls-", dir=runtime_root) as temp:
+        temp_dir = Path(temp)
+        certificate = temp_dir / "localhost-cert.pem"
+        private_key = temp_dir / "localhost-key.pem"
+        subprocess.run(
+            [
+                openssl,
+                "req",
+                "-x509",
+                "-newkey",
+                "rsa:2048",
+                "-sha256",
+                "-nodes",
+                "-days",
+                "1",
+                "-keyout",
+                str(private_key),
+                "-out",
+                str(certificate),
+                "-subj",
+                "/CN=SovereignAI local TLS smoke",
+                "-addext",
+                "subjectAltName=DNS:localhost,IP:127.0.0.1",
+                "-addext",
+                "basicConstraints=critical,CA:TRUE",
+            ],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        with socket.socket() as listener:
+            listener.bind(("127.0.0.1", 0))
+            port = listener.getsockname()[1]
+        environment = os.environ.copy()
+        environment["SOVEREIGNAI_DATA_DIR"] = str(temp_dir / "data")
+        environment["SOVEREIGNAI_ALLOWED_WORKSPACES"] = str(ROOT)
+        environment["SOVEREIGNAI_WORKSPACE"] = str(ROOT)
+        environment["PYTHONPATH"] = str(ROOT)
+        server = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "uvicorn",
+                "app.main:app",
+                "--host",
+                "127.0.0.1",
+                "--port",
+                str(port),
+                "--ssl-certfile",
+                str(certificate),
+                "--ssl-keyfile",
+                str(private_key),
+                "--no-access-log",
+            ],
+            cwd=ROOT,
+            env=environment,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
+        try:
+            url = f"https://127.0.0.1:{port}/openapi.json"
+            response: httpx.Response | None = None
+            last_error: Exception | None = None
+            deadline = time.monotonic() + 25
+            while time.monotonic() < deadline:
+                if server.poll() is not None:
+                    break
+                try:
+                    response = httpx.get(
+                        url, verify=str(certificate), timeout=2, trust_env=False
+                    )
+                    break
+                except httpx.HTTPError as exc:
+                    last_error = exc
+                    time.sleep(0.2)
+            if response is None:
+                server.terminate()
+                server_output, _ = server.communicate(timeout=5)
+                raise RuntimeError(
+                    f"HTTPS server failed to become ready: {last_error};\n"
+                    f"{server_output.decode('utf-8', errors='replace')}"
+                )
+            if response.status_code != 200:
+                raise RuntimeError(f"HTTPS OpenAPI request returned {response.status_code}")
+            spec = response.json()
+            paths = spec.get("paths", {})
+            if "/v1/auth/password-reset/request" not in paths:
+                raise RuntimeError("Expected password reset route is missing from OpenAPI")
+            untrusted_certificate_rejected = False
+            try:
+                httpx.get(url, verify=True, timeout=2, trust_env=False)
+            except httpx.ConnectError:
+                untrusted_certificate_rejected = True
+            if not untrusted_certificate_rejected:
+                raise RuntimeError("The client unexpectedly trusted the disposable certificate")
+            print("tls_verified_with_explicit_local_certificate=true")
+            print(f"https_openapi_status={response.status_code}")
+            print("server_bind_host=127.0.0.1")
+            print("untrusted_certificate_rejected=true")
+            print("password_reset_route_published=true")
+        finally:
+            if server.poll() is None:
+                server.terminate()
+                try:
+                    server.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    server.kill()
+                    server.wait(timeout=5)
+            server.communicate(timeout=5)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())