Protect Tabularium with persistent authenticator-code login

scripts/auth-admin.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+set -eu
+cd /opt/endorphin-tabularium-docs/current
+command=${1:-qr}
+compose() {
+  docker compose -p endorphin-tabularium-docs --env-file /etc/endorphin-tabularium-docs/production.env --env-file release.env -f compose.yml "$@"
+}
+case "$command" in
+  qr|token) compose exec docs node apps/docs/auth-cli.cjs "$command" ;;
+  qr-png|verify) compose exec -T docs node apps/docs/auth-cli.cjs "$command" ;;
+  rotate)
+    [ "${2:-}" = "--confirm" ] || { echo 'Use rotate --confirm to invalidate authenticators and sessions' >&2; exit 1; }
+    compose exec -T docs node apps/docs/auth-cli.cjs rotate --confirm ;;
+  *) echo 'Usage: auth qr | qr-png | token | verify | rotate --confirm' >&2; exit 1 ;;
+esac

scripts/check-container.sh

@@ -1,9 +1,11 @@
 #!/bin/sh
 set -eu
+export CI=true
 pnpm install --frozen-lockfile --store-dir /pnpm/store
 uv sync --project tools/figures --locked
 pnpm run lint
 pnpm run typecheck
+pnpm --filter @endor/docs run test:auth
 cp -R apps/docs/public/figures /tmp/expected-figures
 pnpm run figures
 uv run --locked --project tools/figures python -m unittest discover -s tools/figures/tests -v

scripts/ci.py

@@ -35,22 +35,30 @@ def fingerprint() -> str:
 
 
 def smoke_image(image: str, name: str) -> None:
-    from smoke import verify
-    run("docker", "run", "-d", "--name", name, "-p", "127.0.0.1::3000", image)
+    run("docker", "run", "-d", "--name", name, "--mount", "type=volume,dst=/auth", "-p", "127.0.0.1::3000", image)
     try:
-        port = run("docker", "port", name, "3000/tcp", capture=True).rsplit(":", 1)[1]
+        run("docker", "exec", name, "node", "apps/docs/auth-cli.cjs", "init")
         for _ in range(60):
             state = json.loads(run("docker", "inspect", name, capture=True))[0]["State"]
             if state.get("Health", {}).get("Status") == "healthy":
-                verify(f"http://127.0.0.1:{port}")
-                return
+                run("docker", "exec", name, "node", "apps/docs/auth-cli.cjs", "exercise")
+                run("docker", "restart", name)
+                # Counter/replay state is on the same durable mount after restart.
+                run("docker", "exec", name, "node", "apps/docs/auth-cli.cjs", "state-check")
+                for _ in range(30):
+                    state = json.loads(run("docker", "inspect", name, capture=True))[0]["State"]
+                    if state.get("Health", {}).get("Status") == "healthy":
+                        run("docker", "exec", name, "node", "apps/docs/auth-cli.cjs", "verify")
+                        return
+                    time.sleep(1)
+                raise RuntimeError("Authenticated image failed restart verification")
             if not state["Running"]:
                 break
             time.sleep(1)
         run("docker", "logs", "--tail", "80", name)
         raise RuntimeError("Production image did not become healthy")
     finally:
-        run("docker", "rm", "-f", name)
+        run("docker", "rm", "-fv", name)
 
 
 def main() -> None:

scripts/deploy.py

@@ -9,7 +9,6 @@ import socket
 import subprocess
 import urllib.request
 from ci import APP, ROOT, fingerprint, run
-from smoke import verify
 
 DEPLOY_ROOT = Path("/opt/endorphin-tabularium-docs")
 CONFIG = Path("/etc/endorphin-tabularium-docs/production.env")
@@ -101,6 +100,16 @@ def main() -> None:
         if not 1024 <= port <= 65535:
             raise ValueError("Invalid port")
         check_port(port)
+        auth_dir = DEPLOY_ROOT / "private-auth"
+        auth_dir.mkdir(mode=0o700, exist_ok=True)
+        auth_dir.chmod(0o700)
+        os.chown(auth_dir, 1000, 1000)
+        # Initialize once; normal updates must preserve the authenticator key.
+        run("docker", "run", "--rm", "--network", "none", "--mount",
+            f"type=bind,src={auth_dir},dst=/auth", actual,
+            "node", "apps/docs/auth-cli.cjs", "init")
+        shutil.copyfile(ROOT / "scripts/auth-admin.sh", DEPLOY_ROOT / "auth")
+        (DEPLOY_ROOT / "auth").chmod(0o750)
         current = DEPLOY_ROOT / "current"
         previous = current.resolve() if current.is_symlink() else None
         release = DEPLOY_ROOT / "releases" / sha
@@ -109,7 +118,9 @@ def main() -> None:
         (release / "release.env").write_text(f"TABULARIUM_IMAGE={actual}\n")
         try:
             compose(release, "up", "-d", "--no-build", "--wait", "--wait-timeout", "120")
-            verify(f"http://127.0.0.1:{port}")
+            container = run("docker", "ps", "-q", "--filter", f"label=com.docker.compose.project={APP}",
+                            "--filter", "label=com.docker.compose.service=docs", capture=True)
+            run("docker", "exec", container, "node", "apps/docs/auth-cli.cjs", "verify")
         except Exception:
             if previous is not None:
                 print("Deployment failed; restoring previous verified release", flush=True)

scripts/smoke.py

@@ -1,48 +1,40 @@
-"""Exercise actual production routes and assets without third-party dependencies."""
-from html import unescape
+"""Check the public authentication boundary; auth-cli verify checks private content."""
 import json
-from pathlib import Path
-import re
 import sys
 import urllib.error
+import urllib.parse
 import urllib.request
-import xml.etree.ElementTree as ET
 
-ROOT = Path(__file__).resolve().parents[1]
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
 
 
 def verify(base: str) -> None:
-    def get(path: str) -> tuple[bytes, str]:
-        with urllib.request.urlopen(base.rstrip("/") + path, timeout=15) as response:
-            if response.status != 200:
-                raise RuntimeError(f"Unexpected status for {path}: {response.status}")
-            return response.read(), response.headers.get("Content-Type", "")
+    opener = urllib.request.build_opener(NoRedirect())
+    base = base.rstrip("/")
 
-    health, _ = get("/api/health")
-    assert json.loads(health) == {"status": "ok"}
-    home, _ = get("/")
-    assert b"TABULARIUM" in home
-    for file in sorted((ROOT / "content/docs").glob("*.md")):
-        page, _ = get(f"/docs/{file.stem}")
-        title = re.search(r"^# (.+)$", file.read_text(), re.M)
-        assert title is not None
-        text = unescape(re.sub(r"<[^>]+>", "", page.decode()))
-        assert title.group(1) in text, file.name
-    assets = set(re.findall(r'(?:src|href)="(/_next/static/[^" ]+)"', home.decode()))
-    assert assets, "No generated JavaScript/CSS assets found"
-    for asset in assets:
-        get(unescape(asset))
-    for file in sorted((ROOT / "apps/docs/public/figures").glob("*.svg")):
-        data, kind = get(f"/figures/{file.name}")
-        assert "image/svg+xml" in kind
-        assert ET.fromstring(data).tag.endswith("svg")
-    try:
-        get("/docs/this-document-does-not-exist")
-    except urllib.error.HTTPError as error:
-        assert error.code == 404
-    else:
-        raise RuntimeError("Unknown document did not return 404")
-    print(f"Smoke passed: {base} (health, documents, JS/CSS, SVG, 404)", flush=True)
+    def get(path):
+        try:
+            response = opener.open(base + path, timeout=15)
+        except urllib.error.HTTPError as error:
+            response = error
+        with response:
+            return response.code, response.headers, response.read()
+
+    status, _, health = get("/api/health")
+    assert status == 200 and json.loads(health) == {"status": "ok"}
+    status, _, login = get("/login")
+    assert status == 200 and "输入动态码" in login.decode()
+    for path in ("/", "/docs/00-endor-concepts-draft", "/figures/endor-architecture.svg"):
+        status, headers, _ = get(path)
+        assert status == 307, path
+        target = urllib.parse.urlparse(urllib.parse.urljoin(base, headers["Location"]))
+        assert target.netloc == urllib.parse.urlparse(base).netloc and target.path == "/login"
+        assert "no-store" in headers.get("Cache-Control", "")
+    assert get("/api/unknown")[0] == 401
+    print(f"Public auth boundary passed: {base} (login, health, protected pages/SVG/API)", flush=True)
 
 
 if __name__ == "__main__":