125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
from scripts.verify_deployment import DeploymentVerificationError, verify_deployment
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def deployment_server(
|
|
*,
|
|
expected_token: str = "correct horse battery staple",
|
|
ready: int = 200,
|
|
relative_manifest: bool = False,
|
|
):
|
|
requests: list[tuple[str, str, str]] = []
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self): # noqa: N802
|
|
cookie = self.headers.get("Cookie", "")
|
|
requests.append(("GET", self.path, cookie))
|
|
if self.path == "/dashboard/healthz":
|
|
return self.respond(200, {"status": "ok"})
|
|
if self.path == "/dashboard/readyz":
|
|
return self.respond(ready, {"status": "ready" if ready == 200 else "not ready"})
|
|
if self.path == "/dashboard/manifest.webmanifest":
|
|
return self.respond(200, {
|
|
"name": "StackChain Dashboard",
|
|
"start_url": "./" if relative_manifest else "/dashboard/",
|
|
"scope": "./" if relative_manifest else "/dashboard/",
|
|
})
|
|
if self.path == "/dashboard/login":
|
|
return self.respond(200, "<html><form><input name=\"access_token\"></form></html>", "text/html")
|
|
if self.path == "/dashboard/":
|
|
if "stackchain_session=valid" in cookie:
|
|
return self.respond(200, "<html><nav id=\"mobile-task-dock\"></nav><button id=\"new-issue\">New issue</button></html>", "text/html")
|
|
self.send_response(303)
|
|
self.send_header("Location", "/dashboard/login")
|
|
self.end_headers()
|
|
return None
|
|
return self.respond(404, {"detail": "missing"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
body = self.rfile.read(length).decode()
|
|
requests.append(("POST", self.path, body))
|
|
supplied = json.loads(body).get("access_token", "")
|
|
if self.path == "/dashboard/api/v1/session" and supplied == expected_token:
|
|
payload = json.dumps({"authenticated": True}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Set-Cookie", "stackchain_session=valid; Path=/dashboard/; HttpOnly")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
return None
|
|
return self.respond(401, {"detail": "Invalid access token"})
|
|
|
|
def respond(self, status, body, content_type="application/json"):
|
|
payload = body.encode() if isinstance(body, str) else json.dumps(body).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
return None
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}/dashboard/", requests
|
|
finally:
|
|
server.shutdown()
|
|
thread.join()
|
|
|
|
|
|
def test_verifier_proves_public_login_manifest_and_authenticated_mobile_home():
|
|
with deployment_server() as (url, requests):
|
|
result = verify_deployment(url, "correct horse battery staple")
|
|
|
|
assert result == {
|
|
"health": "ok",
|
|
"readiness": "ready",
|
|
"public_entry": "login",
|
|
"manifest_scope": "/dashboard/",
|
|
"authenticated_home": "mobile",
|
|
}
|
|
assert any(
|
|
method == "POST" and path == "/dashboard/api/v1/session"
|
|
for method, path, _ in requests
|
|
)
|
|
|
|
|
|
def test_verifier_resolves_relative_manifest_scope_inside_public_subpath():
|
|
with deployment_server(relative_manifest=True) as (url, _):
|
|
result = verify_deployment(url, "correct horse battery staple")
|
|
|
|
assert result["manifest_scope"] == "/dashboard/"
|
|
|
|
|
|
def test_verifier_failure_redacts_operator_token():
|
|
secret = "never-print-this-secret"
|
|
with deployment_server(expected_token="different") as (url, _):
|
|
with pytest.raises(DeploymentVerificationError) as raised:
|
|
verify_deployment(url, secret)
|
|
|
|
assert secret not in str(raised.value)
|
|
|
|
|
|
def test_verifier_rejects_a_live_but_unready_deployment_before_login():
|
|
with deployment_server(ready=503) as (url, requests):
|
|
with pytest.raises(DeploymentVerificationError, match="readiness returned HTTP 503"):
|
|
verify_deployment(url, "correct horse battery staple")
|
|
|
|
assert not any(method == "POST" for method, _, _ in requests)
|