feat: verify authenticated deployment readiness (Closes #1048)
This commit is contained in:
1 parent
a757bfe587
commit
e8817056f7
3 files changed
+279
No files matched your search
@@ -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())
|
||||
Reference in new issue
Block a user