jarca-ledger-mcp/dashboard/authentication.py

from __future__ import annotations

from collections.abc import Callable as CallableABC, Mapping, MutableMapping
from dataclasses import dataclass
import hashlib
import hmac
import logging
import os
import secrets
import time
from typing import Callable

from itsdangerous import BadData, URLSafeSerializer
import streamlit as st

logger = logging.getLogger(__name__)

ADMIN_ID_ENV = "JARCA_DASHBOARD_ADMIN_ID"
ADMIN_PASSWORD_ENV = "JARCA_DASHBOARD_ADMIN_PASSWORD"
SIGNING_SECRET_ENV = "JARCA_DASHBOARD_SIGNING_SECRET"
CONFIGURATION_ERROR_MESSAGE = "관리자 인증 설정을 확인할 수 없습니다."
INVALID_CREDENTIALS_MESSAGE = "관리자 ID 또는 비밀번호가 올바르지 않습니다."
MINIMUM_FAILURE_DELAY_SECONDS = 1.0
AUTH_CORRELATION_SESSION_KEY = "_jarca_dashboard_auth_correlation_id"
AUTH_COOKIE_NAME = "__Host-jarca_dashboard_auth"
DEFAULT_TOKEN_TTL_SECONDS = 24 * 60 * 60
REMEMBER_ME_TOKEN_TTL_SECONDS = 30 * 24 * 60 * 60
MAX_TOKEN_LENGTH = 4096
CLOCK_SKEW_SECONDS = 30
_TOKEN_VERSION = 1
_TOKEN_SALT = "jarca-dashboard-persistent-auth-v1"
_TOKEN_CONTEXT = b"jarca-dashboard-persistent-auth-v1\x00"
_TOKEN_FIELDS = frozenset({"v", "sub", "iat", "exp", "jti"})
_TOKEN_SUBJECT = "dashboard-admin"


@dataclass(frozen=True)
class AdminCredentials:
    admin_id: str
    password: str


@dataclass(frozen=True)
class AuthSettings:
    admin_id: str
    admin_password: str
    signing_secret: str

    @property
    def credentials(self) -> AdminCredentials:
        return AdminCredentials(admin_id=self.admin_id, password=self.admin_password)


@dataclass(frozen=True)
class IssuedPersistentToken:
    value: str
    issued_at: int
    expires_at: int
    max_age: int


@dataclass(frozen=True)
class VerifiedPersistentToken:
    subject: str
    issued_at: int
    expires_at: int
    token_id: str


class AuthenticationConfigurationError(RuntimeError):
    """Raised without secret details when dashboard authentication is unavailable."""


def _read_required_setting(source: Mapping[str, str], name: str) -> str:
    try:
        value = source.get(name)
    except Exception as exc:
        raise AuthenticationConfigurationError(CONFIGURATION_ERROR_MESSAGE) from exc
    if not isinstance(value, str) or not value:
        raise AuthenticationConfigurationError(CONFIGURATION_ERROR_MESSAGE)
    return value


def load_admin_credentials(environ: Mapping[str, str] | None = None) -> AdminCredentials:
    source = os.environ if environ is None else environ
    return AdminCredentials(
        admin_id=_read_required_setting(source, ADMIN_ID_ENV),
        password=_read_required_setting(source, ADMIN_PASSWORD_ENV),
    )


def require_auth_settings(environ: Mapping[str, str] | None = None) -> AuthSettings:
    source = os.environ if environ is None else environ
    settings = AuthSettings(
        admin_id=_read_required_setting(source, ADMIN_ID_ENV),
        admin_password=_read_required_setting(source, ADMIN_PASSWORD_ENV),
        signing_secret=_read_required_setting(source, SIGNING_SECRET_ENV),
    )
    if len(settings.signing_secret.encode("utf-8")) < 32:
        raise AuthenticationConfigurationError(CONFIGURATION_ERROR_MESSAGE)
    return settings


def load_auth_settings(environ: Mapping[str, str] | None = None) -> AuthSettings | None:
    try:
        return require_auth_settings(environ)
    except AuthenticationConfigurationError:
        return None


def verify_credentials(
    credentials: AdminCredentials, submitted_id: str, submitted_password: str
) -> bool:
    if not isinstance(submitted_id, str) or not isinstance(submitted_password, str):
        return False
    id_matches = hmac.compare_digest(
        submitted_id.encode("utf-8"), credentials.admin_id.encode("utf-8")
    )
    password_matches = hmac.compare_digest(
        submitted_password.encode("utf-8"), credentials.password.encode("utf-8")
    )
    return id_matches and password_matches


def enforce_failed_login_delay(
    *,
    started_at: float,
    minimum_seconds: float = MINIMUM_FAILURE_DELAY_SECONDS,
    monotonic: Callable[[], float] = time.monotonic,
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    remaining = minimum_seconds - (monotonic() - started_at)
    if remaining > 0:
        sleep(remaining)


def get_auth_correlation_id(
    session_state: MutableMapping[str, object],
    *,
    token_hex: CallableABC[[int], str] = secrets.token_hex,
) -> str:
    existing = session_state.get(AUTH_CORRELATION_SESSION_KEY)
    if isinstance(existing, str) and existing:
        return existing
    correlation_id = token_hex(8)
    session_state[AUTH_CORRELATION_SESSION_KEY] = correlation_id
    return correlation_id


def log_auth_event(event: str, correlation_id: str) -> None:
    logger.info("dashboard_auth event=%s correlation_id=%s", event, correlation_id)


def _derived_signing_key(settings: AuthSettings) -> bytes:
    return hmac.new(
        settings.signing_secret.encode("utf-8"),
        _TOKEN_CONTEXT + settings.admin_id.encode("utf-8") + b"\x00" + settings.admin_password.encode("utf-8"),
        hashlib.sha256,
    ).digest()


def _serializer_for(settings: AuthSettings) -> URLSafeSerializer:
    return URLSafeSerializer(
        secret_key=_derived_signing_key(settings),
        salt=_TOKEN_SALT,
        signer_kwargs={"digest_method": hashlib.sha256},
    )


def issue_persistent_token(
    settings: AuthSettings,
    *,
    remember_me: bool,
    now: float | None = None,
    token_id: str | None = None,
) -> IssuedPersistentToken:
    issued_at = int(time.time() if now is None else now)
    ttl = (
        REMEMBER_ME_TOKEN_TTL_SECONDS if remember_me else DEFAULT_TOKEN_TTL_SECONDS
    )
    expires_at = issued_at + ttl
    jti = secrets.token_urlsafe(16) if token_id is None else token_id
    payload = {
        "v": _TOKEN_VERSION,
        "sub": _TOKEN_SUBJECT,
        "iat": issued_at,
        "exp": expires_at,
        "jti": jti,
    }
    return IssuedPersistentToken(
        value=_serializer_for(settings).dumps(payload),
        issued_at=issued_at,
        expires_at=expires_at,
        max_age=ttl,
    )


def verify_persistent_token(
    settings: AuthSettings | None,
    token: object,
    *,
    now: float | None = None,
) -> VerifiedPersistentToken | None:
    if settings is None or not isinstance(token, str) or not token:
        return None
    if len(token) > MAX_TOKEN_LENGTH:
        return None
    try:
        payload = _serializer_for(settings).loads(token)
    except (BadData, TypeError, ValueError):
        return None
    if not isinstance(payload, dict) or frozenset(payload) != _TOKEN_FIELDS:
        return None

    version = payload.get("v")
    subject = payload.get("sub")
    issued_at = payload.get("iat")
    expires_at = payload.get("exp")
    token_id = payload.get("jti")
    if (
        type(version) is not int
        or version != _TOKEN_VERSION
        or subject != _TOKEN_SUBJECT
        or type(issued_at) is not int
        or type(expires_at) is not int
        or not isinstance(token_id, str)
        or not token_id
        or len(token_id) > 256
    ):
        return None
    current_time = int(time.time() if now is None else now)
    ttl = expires_at - issued_at
    if ttl not in {DEFAULT_TOKEN_TTL_SECONDS, REMEMBER_ME_TOKEN_TTL_SECONDS}:
        return None
    if issued_at > current_time + CLOCK_SKEW_SECONDS or expires_at <= current_time:
        return None

    return VerifiedPersistentToken(
        subject=subject,
        issued_at=issued_at,
        expires_at=expires_at,
        token_id=token_id,
    )


def _request_cookie() -> str | None:
    try:
        value = st.context.cookies.get(AUTH_COOKIE_NAME)
    except Exception:
        return None
    return value if isinstance(value, str) and value else None


def _query_error() -> str | None:
    try:
        value = st.query_params.get("auth_error")
    except Exception:
        return None
    return value if isinstance(value, str) else None


def _render_login(*, clear_cookie: bool = False) -> None:
    st.title("관리자 로그인")
    st.caption("Dashboard를 사용하려면 관리자 자격 증명이 필요합니다.")
    if _query_error() == "invalid":
        st.error(INVALID_CREDENTIALS_MESSAGE)
    elif _query_error() == "unavailable":
        st.error(CONFIGURATION_ERROR_MESSAGE)
    st.markdown(
        """
<form action="/api/auth/login" method="post">
  <label for="jarca-admin-id">관리자 ID</label>
  <input id="jarca-admin-id" name="admin_id" type="text" autocomplete="username" required>
  <label for="jarca-admin-password">비밀번호</label>
  <input id="jarca-admin-password" name="admin_password" type="password" autocomplete="current-password" required>
  <label><input name="remember_me" type="checkbox"> 로그인 상태 유지 (30일)</label>
  <button type="submit">로그인</button>
</form>
""",
        unsafe_allow_html=True,
    )
    if clear_cookie:
        st.markdown(
            "<meta http-equiv=\"refresh\" content=\"0;url=/api/auth/cleanup\">",
            unsafe_allow_html=True,
        )


def _render_logout() -> None:
    with st.sidebar:
        st.markdown(
            """
<form action="/api/auth/logout" method="post">
  <button type="submit">로그아웃</button>
</form>
""",
            unsafe_allow_html=True,
        )


def require_dashboard_authentication() -> bool:
    """Validate the server-signed cookie before allowing any dashboard access."""
    correlation_id = get_auth_correlation_id(st.session_state)
    cookie = _request_cookie()
    settings = load_auth_settings()
    if settings is None:
        log_auth_event("configuration_error", correlation_id)
        _render_login(clear_cookie=cookie is not None)
        st.error(CONFIGURATION_ERROR_MESSAGE)
        return False

    verified = verify_persistent_token(settings, cookie)
    if verified is None:
        log_auth_event("authentication_required", correlation_id)
        _render_login(clear_cookie=cookie is not None)
        return False

    # The configured server-side admin identity is the only actor source.  Do
    # not derive it from query parameters, client headers, or dashboard forms.
    st.session_state["authenticated_actor"] = settings.admin_id
    log_auth_event("token_accepted", correlation_id)
    _render_logout()
    return True


def get_authenticated_actor() -> str | None:
    """Return the actor established by successful server-side authentication."""
    actor = st.session_state.get("authenticated_actor")
    return actor if isinstance(actor, str) and actor.strip() else None