140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
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()
|