"""JWT authentication and SQLite-backed administrator-managed users.""" import json import logging import os import secrets from datetime import datetime, timedelta, timezone from pathlib import Path import bcrypt import jwt from dotenv import load_dotenv from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from backend.database import database PROJECT_ROOT = Path(__file__).parent.parent load_dotenv(PROJECT_ROOT / ".env") def _load_jwt_secret() -> str: configured = os.getenv("JWT_SECRET_KEY") if configured: return configured secret_path = Path( os.getenv("JWT_SECRET_FILE", str(PROJECT_ROOT / "data" / ".jwt-secret")) ) secret_path.parent.mkdir(parents=True, exist_ok=True) try: secret = secret_path.read_text(encoding="utf-8").strip() if secret: return secret except FileNotFoundError: pass secret = secrets.token_urlsafe(48) try: descriptor = os.open(secret_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: existing = secret_path.read_text(encoding="utf-8").strip() if existing: return existing raise RuntimeError(f"JWT secret file is empty: {secret_path}") with os.fdopen(descriptor, "w", encoding="utf-8") as secret_file: secret_file.write(secret) logging.getLogger(__name__).warning("Created JWT secret file at %s", secret_path) return secret JWT_SECRET_KEY = _load_jwt_secret() JWT_ALGORITHM = "HS256" JWT_EXPIRE_HOURS = int(os.getenv("JWT_EXPIRE_HOURS", "24")) security = HTTPBearer(auto_error=False) def _migrate_users() -> None: """Import users.json exactly once, then optionally bootstrap an administrator.""" logger = logging.getLogger(__name__) with database._lock, database.connect() as connection: marker = connection.execute( "SELECT value FROM schema_meta WHERE key = 'users_json_v1'" ).fetchone() if marker is None: config_path = Path( os.getenv("USERS_CONFIG_PATH", str(PROJECT_ROOT / "config" / "users.json")) ) records: list[tuple[str, str, int]] = [] sanitized_users: list[dict] = [] contained_plaintext = False try: raw = json.loads(config_path.read_text(encoding="utf-8")) if not isinstance(raw, list): raise ValueError("users.json must contain a list") seen: set[str] = set() for item in raw: user_id = str(item.get("id") or "").strip() if not user_id or user_id in seen: raise ValueError("users.json contains an empty or duplicate id") seen.add(user_id) password_hash = item.get("password") if not password_hash and item.get("password_plain"): contained_plaintext = True password_hash = bcrypt.hashpw( str(item["password_plain"]).encode(), bcrypt.gensalt() ).decode() if not password_hash: raise ValueError(f"missing password for user {user_id}") try: bcrypt.checkpw(b"hash-validation", password_hash.encode()) except ValueError as exc: raise ValueError(f"invalid bcrypt hash for user {user_id}") from exc records.append( (user_id, password_hash, int(bool(item.get("is_admin") or user_id == "admin"))) ) sanitized = dict(item) sanitized.pop("password_plain", None) sanitized["password"] = password_hash sanitized_users.append(sanitized) except FileNotFoundError: records = [] except (json.JSONDecodeError, ValueError) as exc: raise RuntimeError(f"Cannot migrate users from {config_path}: {exc}") from exc if contained_plaintext: temporary_path = config_path.with_suffix(config_path.suffix + ".tmp") try: temporary_path.write_text( json.dumps(sanitized_users, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) os.replace(temporary_path, config_path) except OSError as exc: temporary_path.unlink(missing_ok=True) raise RuntimeError( f"Cannot remove plaintext passwords from {config_path}; " "make the file writable for the first migration" ) from exc now = datetime.now(timezone.utc).isoformat() connection.execute("BEGIN IMMEDIATE") try: for user_id, password_hash, is_admin in records: connection.execute( "INSERT OR IGNORE INTO users(id, password_hash, is_admin, is_active, " "token_version, created_at, updated_at) VALUES(?, ?, ?, 1, 0, ?, ?)", (user_id, password_hash, is_admin, now, now), ) connection.execute( "INSERT OR IGNORE INTO schema_meta(key, value) VALUES('users_json_v1', ?)", (now,), ) connection.execute("COMMIT") except Exception: connection.execute("ROLLBACK") raise logger.info("Imported %d user(s) from users.json", len(records)) if database.count_admins() == 0: admin_id = os.getenv("BOOTSTRAP_ADMIN_ID") admin_password = os.getenv("BOOTSTRAP_ADMIN_PASSWORD") if admin_id and admin_password: password_hash = bcrypt.hashpw(admin_password.encode(), bcrypt.gensalt()).decode() if database.get_user(admin_id): database.update_user( admin_id, password_hash=password_hash, is_admin=True, is_active=True, ) else: database.create_user(admin_id, password_hash, is_admin=True) logger.warning("Created bootstrap administrator %s", admin_id) else: raise RuntimeError( "No active administrator exists. Mark a users.json account as admin or " "provide BOOTSTRAP_ADMIN_ID and BOOTSTRAP_ADMIN_PASSWORD." ) _migrate_users() def get_user_by_username(username: str): return database.get_user(username) def create_access_token(data: dict) -> str: payload = data.copy() payload["exp"] = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS) return jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM) def decode_access_token(token: str) -> dict | None: try: return jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM]) except (jwt.ExpiredSignatureError, jwt.InvalidTokenError): return None def _authenticated_user(credentials: HTTPAuthorizationCredentials | None) -> dict: if credentials is None: raise HTTPException(status_code=401, detail="인증이 필요합니다") payload = decode_access_token(credentials.credentials) user_id = payload.get("user_id") if payload else None user = database.get_user(user_id) if isinstance(user_id, str) else None if ( user is None or not user["is_active"] or payload.get("token_version") != user["token_version"] ): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="유효하지 않거나 만료된 토큰입니다", ) return user async def get_current_user( credentials: HTTPAuthorizationCredentials | None = Depends(security), ) -> str: return str(_authenticated_user(credentials)["id"]) async def require_admin( credentials: HTTPAuthorizationCredentials | None = Depends(security), ) -> str: user = _authenticated_user(credentials) if not user["is_admin"]: raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다") return str(user["id"]) def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() __all__ = [ "create_access_token", "get_current_user", "get_user_by_username", "hash_password", "require_admin", ]