153 lines
5.2 KiB
Python
153 lines
5.2 KiB
Python
"""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())
|