refs #739: preserve Smilegate changes before repository layout migration

ai-web-agent-console/portal_auth_gateway.py

@@ -0,0 +1,128 @@
+"""Small same-origin authentication gateway for the Smilegate Streamlit portal.
+
+The gateway issues a signed HttpOnly cookie after validating the configured
+PBKDF2 password.  The Streamlit application verifies the signature and expiry
+from the incoming request, so browser refreshes and WebSocket reconnects do not
+require a new login.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs
+
+
+COOKIE_NAME = "poc4_portal_auth"
+MAX_BODY_BYTES = 8_192
+COOKIE_TTL_SECONDS = int(os.environ.get("POC4_LOGIN_COOKIE_TTL_SECONDS", "43200"))
+
+
+def _password_matches(password: str, encoded_password: str) -> bool:
+    try:
+        scheme, iterations_text, salt_hex, expected_hex = encoded_password.split("$", 3)
+        iterations = int(iterations_text)
+        salt = bytes.fromhex(salt_hex)
+        expected = bytes.fromhex(expected_hex)
+    except (TypeError, ValueError):
+        return False
+    if scheme != "pbkdf2_sha256" or not 100_000 <= iterations <= 2_000_000:
+        return False
+    candidate = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
+    return hmac.compare_digest(candidate, expected)
+
+
+def _cookie_value(username: str) -> str:
+    secret = os.environ["POC4_LOGIN_REMEMBER_SECRET"]
+    claims = {"v": 1, "u": username, "e": int(time.time()) + COOKIE_TTL_SECONDS}
+    encoded = base64.urlsafe_b64encode(
+        json.dumps(claims, separators=(",", ":")).encode("utf-8")
+    ).decode("ascii").rstrip("=")
+    signature = hmac.new(secret.encode("utf-8"), encoded.encode("ascii"), hashlib.sha256).hexdigest()
+    return f"{encoded}.{signature}"
+
+
+def _set_cookie(handler: BaseHTTPRequestHandler, value: str, max_age: int) -> None:
+    attributes = [
+        f"{COOKIE_NAME}={value}",
+        "Path=/",
+        f"Max-Age={max_age}",
+        "HttpOnly",
+        "Secure",
+        "SameSite=Lax",
+    ]
+    handler.send_header("Set-Cookie", "; ".join(attributes))
+
+
+class PortalAuthHandler(BaseHTTPRequestHandler):
+    server_version = "SmilegatePortalAuth/1.0"
+
+    def log_message(self, _format: str, *_args: object) -> None:
+        # Do not log form data or authentication details.
+        return
+
+    def _redirect(self, location: str, cookie_value: str | None = None, max_age: int = 0) -> None:
+        self.send_response(HTTPStatus.SEE_OTHER)
+        if cookie_value is not None:
+            _set_cookie(self, cookie_value, max_age)
+        self.send_header("Location", location)
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+
+    def do_GET(self) -> None:  # noqa: N802
+        if self.path == "/health":
+            self.send_response(HTTPStatus.OK)
+            self.send_header("Content-Type", "text/plain; charset=utf-8")
+            self.send_header("Cache-Control", "no-store")
+            self.end_headers()
+            self.wfile.write(b"ok\n")
+            return
+        if self.path == "/logout":
+            self._redirect("/", "", 0)
+            return
+        self.send_error(HTTPStatus.NOT_FOUND)
+
+    def do_POST(self) -> None:  # noqa: N802
+        if self.path != "/login":
+            self.send_error(HTTPStatus.NOT_FOUND)
+            return
+        try:
+            content_length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            content_length = 0
+        if content_length <= 0 or content_length > MAX_BODY_BYTES:
+            self._redirect("/?login=failed")
+            return
+        form = parse_qs(self.rfile.read(content_length).decode("utf-8"), keep_blank_values=True)
+        username = form.get("username", [""])[0].strip()
+        password = form.get("password", [""])[0]
+        expected_username = os.environ.get("POC4_LOGIN_USER", "").strip()
+        encoded_password = os.environ.get("POC4_LOGIN_PASSWORD_PBKDF2", "").strip()
+        if (
+            expected_username
+            and hmac.compare_digest(username, expected_username)
+            and _password_matches(password, encoded_password)
+        ):
+            self._redirect("/", _cookie_value(username), COOKIE_TTL_SECONDS)
+            return
+        self._redirect("/?login=failed")
+
+
+def main() -> None:
+    address = os.environ.get("POC4_AUTH_BIND", "127.0.0.1")
+    port = int(os.environ.get("POC4_AUTH_PORT", "8623"))
+    required = ("POC4_LOGIN_USER", "POC4_LOGIN_PASSWORD_PBKDF2", "POC4_LOGIN_REMEMBER_SECRET")
+    missing = [name for name in required if not os.environ.get(name, "").strip()]
+    if missing:
+        raise RuntimeError("missing required portal auth configuration")
+    ThreadingHTTPServer((address, port), PortalAuthHandler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()