Make deployed mobile dashboard reachable through authenticated readiness #1049
18
README.md
18
README.md
|
|
@ -288,6 +288,24 @@ export STACKCHAIN_PUSH_DB='/var/lib/stackchain-dashboard/push-subscriptions.sqli
|
|||
uvicorn src.main:app --host 127.0.0.1 --port 8000
|
||||
```
|
||||
|
||||
For a systemd deployment, keep those values in a root-readable environment file
|
||||
(`chmod 600`), reference it with `EnvironmentFile=`, and keep secrets out of the unit
|
||||
command line and repository. Liveness alone does not prove that operators can use the
|
||||
service: `/healthz` intentionally remains healthy when authentication is missing. After
|
||||
each restart or proxy change, run the complete public-subpath smoke journey with the
|
||||
operator secret supplied only through the process environment:
|
||||
|
||||
```bash
|
||||
STACKCHAIN_DASHBOARD_ACCESS_TOKEN='<operator-sign-in-secret>' \
|
||||
python3 scripts/verify_deployment.py \
|
||||
https://forge.example.com/dashboard/
|
||||
```
|
||||
|
||||
The verifier requires readiness, rendered sign-in, a manifest whose `scope` and
|
||||
`start_url` remain inside the supplied subpath, and an authenticated mobile Home with
|
||||
its New-issue entry. It emits only a small JSON result and never includes the operator
|
||||
secret in success or failure output.
|
||||
|
||||
Token and passkey sign-in failures are scoped to a hashed canonical client address and
|
||||
persisted across workers and restarts. Public passkey option issuance has a separate
|
||||
fixed-window admission budget in the same ledger, and live challenges are bounded per
|
||||
|
|
|
|||
137
scripts/verify_deployment.py
Normal file
137
scripts/verify_deployment.py
Normal file
|
|
@ -0,0 +1,137 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Verify that a deployed dashboard is usable, not merely alive."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import http.cookiejar
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
class DeploymentVerificationError(RuntimeError):
|
||||
"""A public deployment failed one observable user-flow check."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Response:
|
||||
status: int
|
||||
url: str
|
||||
content_type: str
|
||||
body: bytes
|
||||
|
||||
|
||||
def _request(opener: urllib.request.OpenerDirector, request: str | urllib.request.Request, label: str) -> Response:
|
||||
try:
|
||||
with opener.open(request, timeout=10) as response:
|
||||
return Response(
|
||||
status=response.status,
|
||||
url=response.geturl(),
|
||||
content_type=response.headers.get_content_type(),
|
||||
body=response.read(),
|
||||
)
|
||||
except urllib.error.HTTPError as error:
|
||||
raise DeploymentVerificationError(f"{label} returned HTTP {error.code}") from None
|
||||
except (OSError, urllib.error.URLError, TimeoutError):
|
||||
raise DeploymentVerificationError(f"{label} could not be reached") from None
|
||||
|
||||
|
||||
def verify_deployment(base_url: str, access_token: str) -> dict[str, str]:
|
||||
"""Exercise liveness, readiness, sign-in, PWA scope, and mobile Home."""
|
||||
parsed = urllib.parse.urlsplit(base_url)
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||||
raise DeploymentVerificationError("deployment URL must be absolute HTTP(S)")
|
||||
if not access_token:
|
||||
raise DeploymentVerificationError("operator access token is required")
|
||||
base_url = base_url.rstrip("/") + "/"
|
||||
base_path = urllib.parse.urlsplit(base_url).path
|
||||
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
||||
|
||||
health = _request(opener, urllib.parse.urljoin(base_url, "healthz"), "liveness")
|
||||
if health.status != 200:
|
||||
raise DeploymentVerificationError(f"liveness returned HTTP {health.status}")
|
||||
|
||||
readiness = _request(opener, urllib.parse.urljoin(base_url, "readyz"), "readiness")
|
||||
if readiness.status != 200:
|
||||
raise DeploymentVerificationError(f"readiness returned HTTP {readiness.status}")
|
||||
|
||||
public_entry = _request(opener, base_url, "public entry")
|
||||
public_text = public_entry.body.decode("utf-8", errors="replace")
|
||||
if public_entry.content_type != "text/html" or 'name="access_token"' not in public_text:
|
||||
raise DeploymentVerificationError("public entry did not render the operator sign-in form")
|
||||
|
||||
manifest_response = _request(
|
||||
opener, urllib.parse.urljoin(base_url, "manifest.webmanifest"), "manifest"
|
||||
)
|
||||
try:
|
||||
manifest = json.loads(manifest_response.body)
|
||||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||||
raise DeploymentVerificationError("manifest did not return valid JSON") from None
|
||||
manifest_scope = urllib.parse.urljoin(manifest_response.url, str(manifest.get("scope", "")))
|
||||
manifest_start = urllib.parse.urljoin(manifest_response.url, str(manifest.get("start_url", "")))
|
||||
expected = urllib.parse.urlsplit(base_url)
|
||||
resolved_scope = urllib.parse.urlsplit(manifest_scope)
|
||||
resolved_start = urllib.parse.urlsplit(manifest_start)
|
||||
if (
|
||||
(resolved_scope.scheme, resolved_scope.netloc, resolved_scope.path)
|
||||
!= (expected.scheme, expected.netloc, expected.path)
|
||||
or (resolved_start.scheme, resolved_start.netloc, resolved_start.path)
|
||||
!= (expected.scheme, expected.netloc, expected.path)
|
||||
):
|
||||
raise DeploymentVerificationError("manifest scope or start URL escaped the deployment subpath")
|
||||
|
||||
encoded = json.dumps({
|
||||
"access_token": access_token,
|
||||
"device_label": "Deployment smoke verifier",
|
||||
}).encode()
|
||||
login_request = urllib.request.Request(
|
||||
urllib.parse.urljoin(base_url, "api/v1/session"),
|
||||
data=encoded,
|
||||
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
authenticated = _request(opener, login_request, "operator sign-in")
|
||||
try:
|
||||
authenticated_payload = json.loads(authenticated.body)
|
||||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||||
raise DeploymentVerificationError("operator sign-in did not return valid JSON") from None
|
||||
if authenticated_payload.get("authenticated") is not True:
|
||||
raise DeploymentVerificationError("operator sign-in was not accepted")
|
||||
|
||||
home = _request(opener, base_url, "authenticated Home")
|
||||
home_text = home.body.decode("utf-8", errors="replace")
|
||||
if home.content_type != "text/html" or (
|
||||
'id="mobile-task-dock"' not in home_text or 'id="new-issue"' not in home_text
|
||||
):
|
||||
raise DeploymentVerificationError("operator sign-in did not reach the mobile dashboard Home")
|
||||
|
||||
return {
|
||||
"health": "ok",
|
||||
"readiness": "ready",
|
||||
"public_entry": "login",
|
||||
"manifest_scope": base_path,
|
||||
"authenticated_home": "mobile",
|
||||
}
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("url", help="Dashboard URL including its public subpath")
|
||||
args = parser.parse_args(argv)
|
||||
token = os.environ.get("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")
|
||||
try:
|
||||
result = verify_deployment(args.url, token)
|
||||
except DeploymentVerificationError as error:
|
||||
print(f"deployment verification failed: {error}", file=sys.stderr)
|
||||
return 1
|
||||
print(json.dumps(result, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
124
tests/test_verify_deployment.py
Normal file
124
tests/test_verify_deployment.py
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
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)
|
||||
Loading…
Reference in New Issue
Block a user