feat(dashboard): add administrator authentication

dashboard/app.py

@@ -13,6 +13,7 @@ if __package__ in (None, ""):
 st.set_page_config(page_title="Jarca Ledger", page_icon="💰", layout="wide")
 
 from dashboard.common import clear_cache, get_db_path, get_transfer_review_items, inject_app_css, render_sidebar_nav  # noqa: F401
+from dashboard.authentication import require_dashboard_authentication
 from dashboard.page_views.accounts import page_accounts
 from dashboard.page_views.assets import page_assets
 from dashboard.page_views.categories import page_categories
@@ -25,6 +26,9 @@ from dashboard.page_views.transactions import page_transactions
 from dashboard.page_views.uncategorized import page_uncategorized
 
 def main() -> None:
+    if not require_dashboard_authentication():
+        return
+
     db_path = get_db_path()
     inject_app_css()
     pages = ["월정산", "요약", "자산/마감", "거래내역", "확인필요 큐", "미분류", "분류 규칙", "카테고리", "계좌", "Import"]

dashboard/authentication.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from collections.abc import Callable as CallableABC, Mapping, MutableMapping
+from dataclasses import dataclass
+import hmac
+import logging
+import os
+import secrets
+import time
+from typing import Callable
+
+import streamlit as st
+
+logger = logging.getLogger(__name__)
+
+ADMIN_ID_ENV = "JARCA_DASHBOARD_ADMIN_ID"
+ADMIN_PASSWORD_ENV = "JARCA_DASHBOARD_ADMIN_PASSWORD"
+CONFIGURATION_ERROR_MESSAGE = "관리자 인증 설정을 확인할 수 없습니다."
+INVALID_CREDENTIALS_MESSAGE = "관리자 ID 또는 비밀번호가 올바르지 않습니다."
+MINIMUM_FAILURE_DELAY_SECONDS = 1.0
+AUTHENTICATED_SESSION_KEY = "_jarca_dashboard_authenticated"
+LOGIN_FAILED_SESSION_KEY = "_jarca_dashboard_login_failed"
+LOGIN_ID_SESSION_KEY = "_jarca_dashboard_login_id"
+LOGIN_PASSWORD_SESSION_KEY = "_jarca_dashboard_login_password"
+AUTH_CORRELATION_SESSION_KEY = "_jarca_dashboard_auth_correlation_id"
+
+
+@dataclass(frozen=True)
+class AdminCredentials:
+    admin_id: str
+    password: str
+
+
+class AuthenticationConfigurationError(RuntimeError):
+    """Raised without secret details when dashboard authentication is unavailable."""
+
+
+def load_admin_credentials(environ: Mapping[str, str] | None = None) -> AdminCredentials:
+    source = os.environ if environ is None else environ
+    try:
+        admin_id = source.get(ADMIN_ID_ENV)
+        password = source.get(ADMIN_PASSWORD_ENV)
+    except Exception as exc:
+        raise AuthenticationConfigurationError(CONFIGURATION_ERROR_MESSAGE) from exc
+
+    if not isinstance(admin_id, str) or not admin_id or not isinstance(password, str) or not password:
+        raise AuthenticationConfigurationError(CONFIGURATION_ERROR_MESSAGE)
+
+    return AdminCredentials(admin_id=admin_id, password=password)
+
+
+def verify_credentials(credentials: AdminCredentials, submitted_id: str, submitted_password: str) -> bool:
+    id_matches = hmac.compare_digest(submitted_id, credentials.admin_id)
+    password_matches = hmac.compare_digest(submitted_password, credentials.password)
+    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 _clear_submitted_credentials() -> None:
+    st.session_state.pop(LOGIN_ID_SESSION_KEY, None)
+    st.session_state.pop(LOGIN_PASSWORD_SESSION_KEY, None)
+
+
+def require_dashboard_authentication() -> bool:
+    """Render the authentication boundary and return whether access is allowed."""
+    correlation_id = get_auth_correlation_id(st.session_state)
+    try:
+        credentials = load_admin_credentials()
+    except AuthenticationConfigurationError:
+        st.session_state.pop(AUTHENTICATED_SESSION_KEY, None)
+        _clear_submitted_credentials()
+        log_auth_event("configuration_error", correlation_id)
+        st.title("관리자 로그인")
+        st.error(CONFIGURATION_ERROR_MESSAGE)
+        return False
+
+    if st.session_state.get(AUTHENTICATED_SESSION_KEY) is True:
+        _clear_submitted_credentials()
+        with st.sidebar:
+            if st.button("로그아웃", key="dashboard_logout", use_container_width=True):
+                st.session_state.pop(AUTHENTICATED_SESSION_KEY, None)
+                log_auth_event("logout", correlation_id)
+                st.rerun()
+        return True
+
+    st.title("관리자 로그인")
+    st.caption("Dashboard를 사용하려면 관리자 자격 증명이 필요합니다.")
+    show_login_error = st.session_state.pop(LOGIN_FAILED_SESSION_KEY, False) is True
+    if show_login_error:
+        _clear_submitted_credentials()
+    with st.form("dashboard_login", clear_on_submit=True):
+        submitted_id = st.text_input("관리자 ID", key=LOGIN_ID_SESSION_KEY)
+        submitted_password = st.text_input("비밀번호", type="password", key=LOGIN_PASSWORD_SESSION_KEY)
+        submitted = st.form_submit_button("로그인", use_container_width=True)
+
+    if show_login_error:
+        st.error(INVALID_CREDENTIALS_MESSAGE)
+
+    if not submitted:
+        return False
+
+    started_at = time.monotonic()
+    if verify_credentials(credentials, submitted_id, submitted_password):
+        st.session_state[AUTHENTICATED_SESSION_KEY] = True
+        log_auth_event("login_succeeded", correlation_id)
+        st.rerun()
+
+    enforce_failed_login_delay(started_at=started_at)
+    log_auth_event("login_failed", correlation_id)
+    st.session_state[LOGIN_FAILED_SESSION_KEY] = True
+    st.rerun()