"""Small localhost authentication service for the HMM Streamlit portal. Nginx owns the public security boundary. This module validates the existing PBKDF2 login, issues a signed HttpOnly cookie, and answers Nginx auth_request subrequests. Authentication values are never accepted from a URL. """ from __future__ import annotations import base64 import binascii from collections import defaultdict, deque from dataclasses import dataclass from datetime import datetime, timezone from http import HTTPStatus from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import hashlib import hmac import html import json import logging import os import secrets import threading import time from typing import Deque from urllib.parse import parse_qs, urlsplit def _environment_value(*names: str, default: str = "") -> str: for name in names: value = os.environ.get(name) if value is not None and value.strip(): return value.strip() return default LOG = logging.getLogger("hmm_portal_auth") SESSION_COOKIE_NAME = "__Host-HMM_PORTAL_SESSION" CSRF_COOKIE_NAME = "__Host-HMM_LOGIN_CSRF" SESSION_TOKEN_VERSION = 2 MAX_REQUEST_BYTES = 8_192 MAX_FIELD_CHARS = 200 @dataclass(frozen=True) class AuthConfig: username: str password_pbkdf2: str cookie_secret: str bind_address: str = "127.0.0.1" port: int = 8621 session_seconds: int = 12 * 60 * 60 remember_seconds: int = 7 * 24 * 60 * 60 product_name: str = "HMM AI 업무 에이전트" login_title: str = "HMM AI 업무 에이전트" login_description: str = "사용자 인증 후 AI 업무 질의 기능을 이용할 수 있습니다." login_footer: str = "승인된 사용자만 접속할 수 있습니다." primary_color: str = "#004b87" @classmethod def from_environment(cls) -> "AuthConfig": config = cls( username=_environment_value( "AI_WEB_AGENT_CONSOLE_LOGIN_USER", "POC4_LOGIN_USER" ), password_pbkdf2=_environment_value( "AI_WEB_AGENT_CONSOLE_LOGIN_PASSWORD_PBKDF2", "POC4_LOGIN_PASSWORD_PBKDF2", ), cookie_secret=_environment_value( "AI_WEB_AGENT_CONSOLE_LOGIN_COOKIE_SECRET", "POC4_LOGIN_COOKIE_SECRET", ), bind_address=os.environ.get( "PORTAL_AUTH_BIND_ADDRESS", "127.0.0.1" ).strip(), port=int(os.environ.get("PORTAL_AUTH_PORT", "8621")), session_seconds=int( os.environ.get("PORTAL_AUTH_SESSION_SECONDS", str(12 * 60 * 60)) ), remember_seconds=int( os.environ.get( "PORTAL_AUTH_REMEMBER_SECONDS", str(7 * 24 * 60 * 60) ) ), product_name=os.environ.get( "AGENT_CONSOLE_NAME", "HMM AI 업무 에이전트" ).strip(), login_title=os.environ.get( "AGENT_CONSOLE_LOGIN_TITLE", "HMM AI 업무 에이전트" ).strip(), login_description=os.environ.get( "AGENT_CONSOLE_LOGIN_DESCRIPTION", "사용자 인증 후 AI 업무 질의 기능을 이용할 수 있습니다.", ).strip(), login_footer=os.environ.get( "AGENT_CONSOLE_LOGIN_FOOTER", "승인된 사용자만 접속할 수 있습니다.", ).strip(), primary_color=os.environ.get( "AGENT_CONSOLE_PRIMARY_COLOR", "#004b87" ).strip(), ) config.validate() return config def validate(self) -> None: if not self.username or not self.password_pbkdf2: raise ValueError("AI web agent console login credentials are not configured") if len(self.cookie_secret.encode("utf-8")) < 32: raise ValueError( "AI_WEB_AGENT_CONSOLE_LOGIN_COOKIE_SECRET must be at least 32 bytes" ) if self.bind_address not in {"127.0.0.1", "::1"}: raise ValueError("Portal authentication service must bind to loopback") if not 1 <= self.port <= 65535: raise ValueError("PORTAL_AUTH_PORT is invalid") if not 300 <= self.session_seconds <= 24 * 60 * 60: raise ValueError("PORTAL_AUTH_SESSION_SECONDS is outside the safe range") if not self.session_seconds <= self.remember_seconds <= 30 * 24 * 60 * 60: raise ValueError("PORTAL_AUTH_REMEMBER_SECONDS is outside the safe range") @dataclass(frozen=True) class AuthenticatedSession: username: str expires_at: int class SessionTokenCodec: def __init__(self, secret: str): self._secret = secret.encode("utf-8") def issue(self, username: str, lifetime_seconds: int, now: int | None = None) -> str: issued_at = int(time.time()) if now is None else now payload = { "v": SESSION_TOKEN_VERSION, "u": username, "i": issued_at, "e": issued_at + lifetime_seconds, "n": secrets.token_urlsafe(18), } encoded = _base64url_encode( json.dumps(payload, separators=(",", ":")).encode("utf-8") ) signature = hmac.new( self._secret, encoded.encode("ascii"), hashlib.sha256 ).hexdigest() return f"{encoded}.{signature}" def verify(self, token: str, expected_username: str, now: int | None = None) -> AuthenticatedSession | None: if not token or len(token) > 2048: return None current_time = int(time.time()) if now is None else now try: encoded, supplied_signature = token.split(".", 1) expected_signature = hmac.new( self._secret, encoded.encode("ascii"), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(supplied_signature, expected_signature): return None payload = json.loads(_base64url_decode(encoded).decode("utf-8")) version = int(payload["v"]) username = str(payload["u"]) issued_at = int(payload["i"]) expires_at = int(payload["e"]) except ( binascii.Error, KeyError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError, ): return None if version != SESSION_TOKEN_VERSION: return None if issued_at > current_time + 30 or expires_at <= current_time: return None if expires_at - issued_at > 30 * 24 * 60 * 60: return None if not hmac.compare_digest(username, expected_username): return None return AuthenticatedSession(username=username, expires_at=expires_at) class LoginAttemptLimiter: def __init__(self, maximum_failures: int = 5, window_seconds: int = 300): self._maximum_failures = maximum_failures self._window_seconds = window_seconds self._failures: dict[str, Deque[float]] = defaultdict(deque) self._lock = threading.Lock() def blocked(self, key: str, now: float | None = None) -> bool: current_time = time.monotonic() if now is None else now with self._lock: failures = self._failures[key] self._prune(failures, current_time) return len(failures) >= self._maximum_failures def record_failure(self, key: str, now: float | None = None) -> None: current_time = time.monotonic() if now is None else now with self._lock: failures = self._failures[key] self._prune(failures, current_time) failures.append(current_time) def reset(self, key: str) -> None: with self._lock: self._failures.pop(key, None) def _prune(self, failures: Deque[float], now: float) -> None: cutoff = now - self._window_seconds while failures and failures[0] < cutoff: failures.popleft() 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 session_cookie_header(token: str, max_age: int | None) -> str: attributes = [ f"{SESSION_COOKIE_NAME}={token}", "Path=/", "Secure", "HttpOnly", "SameSite=Lax", ] if max_age is not None: attributes.append(f"Max-Age={max_age}") return "; ".join(attributes) def clear_session_cookie_header() -> str: return ( f"{SESSION_COOKIE_NAME}=; Path=/; Max-Age=0; " "Secure; HttpOnly; SameSite=Lax" ) def csrf_cookie_header(value: str, max_age: int = 600) -> str: return ( f"{CSRF_COOKIE_NAME}={value}; Path=/; Max-Age={max_age}; " "Secure; HttpOnly; SameSite=Strict" ) def clear_csrf_cookie_header() -> str: return ( f"{CSRF_COOKIE_NAME}=; Path=/; Max-Age=0; " "Secure; HttpOnly; SameSite=Strict" ) def _base64url_encode(value: bytes) -> str: return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") def _base64url_decode(value: str) -> bytes: padded = value + "=" * (-len(value) % 4) return base64.urlsafe_b64decode(padded) def _cookie_value(cookie_header: str, name: str) -> str: try: cookies = SimpleCookie() cookies.load(cookie_header) morsel = cookies.get(name) return morsel.value if morsel is not None else "" except (KeyError, TypeError): return "" def _login_page(config: AuthConfig, csrf_value: str, error: str = "") -> bytes: error_html = ( f'' if error else "" ) return f""" {html.escape(config.product_name)}
HMM

{html.escape(config.login_title)}

{html.escape(config.login_description)}

{error_html}

{html.escape(config.login_footer)}

""".encode("utf-8") def build_handler(config: AuthConfig) -> type[BaseHTTPRequestHandler]: codec = SessionTokenCodec(config.cookie_secret) limiter = LoginAttemptLimiter() class PortalAuthHandler(BaseHTTPRequestHandler): server_version = "HMMPortalAuth/1.0" sys_version = "" def do_HEAD(self) -> None: self._route(send_body=False) def do_GET(self) -> None: self._route(send_body=True) def do_POST(self) -> None: path = urlsplit(self.path).path if path == "/auth/login": self._login() elif path == "/auth/logout": self._logout() else: self._send_text(HTTPStatus.NOT_FOUND, "Not found") def _route(self, send_body: bool) -> None: path = urlsplit(self.path).path if path == "/auth/check": self._check() elif path == "/auth/login": self._show_login(send_body=send_body) elif path == "/auth/logout": self._logout() elif path == "/auth/healthz": self._send_text(HTTPStatus.OK, "ok", send_body=send_body) else: self._send_text(HTTPStatus.NOT_FOUND, "Not found", send_body=send_body) def _check(self) -> None: session = self._session() if session is None: self._send_empty(HTTPStatus.UNAUTHORIZED) return self.send_response(HTTPStatus.NO_CONTENT) self._security_headers() self.send_header("X-Auth-User", session.username) self.send_header("X-Auth-Expires", str(session.expires_at)) self.end_headers() def _show_login(self, send_body: bool = True, error: str = "") -> None: if self._session() is not None and not error: self._redirect("/") return csrf_value = secrets.token_urlsafe(32) body = _login_page(config, csrf_value, error) self.send_response(HTTPStatus.OK) self._security_headers() self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Set-Cookie", csrf_cookie_header(csrf_value)) self.end_headers() if send_body: self.wfile.write(body) def _login(self) -> None: client_key = self._client_key() if limiter.blocked(client_key): self._show_login(error="로그인 시도가 잠시 제한되었습니다. 잠시 후 다시 시도해 주세요.") return try: content_length = int(self.headers.get("Content-Length", "0")) except ValueError: content_length = 0 if not 1 <= content_length <= MAX_REQUEST_BYTES: self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") return raw_body = self.rfile.read(content_length) try: form = parse_qs( raw_body.decode("utf-8"), keep_blank_values=True, strict_parsing=False, max_num_fields=8, ) except (UnicodeDecodeError, ValueError): self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") return username = _form_value(form, "username") password = _form_value(form, "password") csrf_form = _form_value(form, "csrf") csrf_cookie = _cookie_value( self.headers.get("Cookie", ""), CSRF_COOKIE_NAME ) if ( not csrf_form or not csrf_cookie or not hmac.compare_digest(csrf_form, csrf_cookie) ): self._send_text(HTTPStatus.BAD_REQUEST, "Invalid request") return valid_credentials = ( len(username) <= 80 and len(password) <= MAX_FIELD_CHARS and hmac.compare_digest(username.strip(), config.username) and password_matches(password, config.password_pbkdf2) ) if not valid_credentials: limiter.record_failure(client_key) self._show_login(error="사용자 ID 또는 비밀번호를 확인해 주세요.") return limiter.reset(client_key) remember = _form_value(form, "remember") == "yes" lifetime = ( config.remember_seconds if remember else config.session_seconds ) token = codec.issue(config.username, lifetime) self.send_response(HTTPStatus.SEE_OTHER) self._security_headers() self.send_header("Location", "/") self.send_header( "Set-Cookie", session_cookie_header(token, lifetime if remember else None), ) self.send_header("Set-Cookie", clear_csrf_cookie_header()) self.end_headers() def _logout(self) -> None: self.send_response(HTTPStatus.SEE_OTHER) self._security_headers() self.send_header("Location", "/auth/login") self.send_header("Set-Cookie", clear_session_cookie_header()) self.send_header("Set-Cookie", clear_csrf_cookie_header()) self.end_headers() def _session(self) -> AuthenticatedSession | None: token = _cookie_value( self.headers.get("Cookie", ""), SESSION_COOKIE_NAME ) return codec.verify(token, config.username) def _client_key(self) -> str: forwarded = self.headers.get("X-Real-IP", "").strip() return forwarded or self.client_address[0] def _redirect(self, location: str) -> None: self.send_response(HTTPStatus.SEE_OTHER) self._security_headers() self.send_header("Location", location) self.end_headers() def _send_empty(self, status: HTTPStatus) -> None: self.send_response(status) self._security_headers() self.send_header("Content-Length", "0") self.end_headers() def _send_text( self, status: HTTPStatus, message: str, send_body: bool = True, ) -> None: body = message.encode("utf-8") self.send_response(status) self._security_headers() self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() if send_body: self.wfile.write(body) def _security_headers(self) -> None: self.send_header("Cache-Control", "no-store") self.send_header("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; " "form-action 'self'; base-uri 'none'; frame-ancestors 'none'", ) def log_message(self, _format: str, *args: object) -> None: # Do not log query strings, cookies, form bodies, or tokens. LOG.info("%s %s", self.command, urlsplit(self.path).path) return PortalAuthHandler def _form_value(form: dict[str, list[str]], name: str) -> str: values = form.get(name) return values[0] if values else "" def main() -> None: logging.basicConfig( level=os.environ.get("PORTAL_AUTH_LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(name)s %(message)s", ) config = AuthConfig.from_environment() server = ThreadingHTTPServer( (config.bind_address, config.port), build_handler(config) ) LOG.info("HMM portal authentication service listening on loopback port %s", config.port) server.serve_forever() if __name__ == "__main__": main()