129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""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()
|