nvidia-bug/auth.py

"""
NV-Log Analyzer - 인증 및 사용자 관리 모듈
역할: admin(관리자), editor(편집자), viewer(뷰어)
"""

import hashlib
import secrets
import sqlite3
from contextlib import contextmanager
from pathlib import Path

DB_PATH = Path(__file__).parent / "data" / "nv_log_history.db"

ROLES = {
    "admin": {
        "label": "관리자",
        "description": "모든 기능 사용 가능 + 사용자 관리",
        "can_upload": True,
        "can_view": True,
        "can_edit_memo": True,
        "can_delete": True,
        "can_manage_users": True,
    },
    "editor": {
        "label": "편집자",
        "description": "로그 업로드, 분석 결과 조회/수정 가능",
        "can_upload": True,
        "can_view": True,
        "can_edit_memo": True,
        "can_delete": False,
        "can_manage_users": False,
    },
    "viewer": {
        "label": "뷰어",
        "description": "분석 결과 조회만 가능",
        "can_upload": False,
        "can_view": True,
        "can_edit_memo": False,
        "can_delete": False,
        "can_manage_users": False,
    },
}


@contextmanager
def get_conn():
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def _hash_password(password: str, salt: str = None) -> tuple:
    """비밀번호 해싱 (SHA-256 + salt)"""
    if salt is None:
        salt = secrets.token_hex(16)
    hashed = hashlib.sha256((salt + password).encode()).hexdigest()
    return hashed, salt


def init_user_db():
    """사용자 테이블 초기화 + 기본 admin 계정 생성"""
    with get_conn() as conn:
        conn.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            username    TEXT    NOT NULL UNIQUE,
            password    TEXT    NOT NULL,
            salt        TEXT    NOT NULL,
            display_name TEXT   DEFAULT '',
            role        TEXT    NOT NULL DEFAULT 'viewer',
            status      TEXT    NOT NULL DEFAULT 'pending',
            created_at  TEXT    NOT NULL DEFAULT (datetime('now','localtime')),
            approved_by TEXT    DEFAULT NULL,
            last_login  TEXT    DEFAULT NULL
        );
        """)

        # 기본 admin 계정 (최초 1회만)
        existing = conn.execute("SELECT id FROM users WHERE username = 'admin'").fetchone()
        if not existing:
            hashed, salt = _hash_password("admin1234")
            conn.execute("""
                INSERT INTO users (username, password, salt, display_name, role, status)
                VALUES (?, ?, ?, ?, 'admin', 'approved')
            """, ("admin", hashed, salt, "시스템 관리자"))


def authenticate(username: str, password: str) -> dict:
    """로그인 인증. 성공 시 사용자 정보 dict 반환, 실패 시 None (대소문자 무시 체크)"""
    with get_conn() as conn:
        user = conn.execute(
            "SELECT * FROM users WHERE LOWER(username) = LOWER(?)", (username,)
        ).fetchone()

        if not user:
            return None

        hashed, _ = _hash_password(password, user["salt"])
        if hashed != user["password"]:
            return None

        if user["status"] != "approved":
            return {"error": "pending", "status": user["status"]}

        # 마지막 로그인 시간 업데이트
        conn.execute(
            "UPDATE users SET last_login = datetime('now','localtime') WHERE id = ?",
            (user["id"],)
        )

        return dict(user)


def register_user(username: str, password: str, display_name: str) -> dict:
    """회원가입. 성공 시 {"success": True}, 실패 시 에러 메시지 (대소문자 무시 중복 검사)"""
    if len(username) < 3:
        return {"success": False, "error": "아이디는 3자 이상이어야 합니다."}
    if len(password) < 6:
        return {"success": False, "error": "비밀번호는 6자 이상이어야 합니다."}

    with get_conn() as conn:
        existing = conn.execute(
            "SELECT id FROM users WHERE LOWER(username) = LOWER(?)", (username,)
        ).fetchone()
        if existing:
            return {"success": False, "error": "이미 사용 중인 아이디입니다."}

        hashed, salt = _hash_password(password)
        conn.execute("""
            INSERT INTO users (username, password, salt, display_name, role, status)
            VALUES (?, ?, ?, ?, 'viewer', 'pending')
        """, (username, hashed, salt, display_name))

    return {"success": True}


def get_pending_users() -> list:
    """승인 대기 중인 사용자 목록"""
    with get_conn() as conn:
        return [dict(r) for r in conn.execute(
            "SELECT * FROM users WHERE status = 'pending' ORDER BY created_at DESC"
        ).fetchall()]


def get_all_users() -> list:
    """전체 사용자 목록"""
    with get_conn() as conn:
        return [dict(r) for r in conn.execute(
            "SELECT * FROM users ORDER BY CASE status WHEN 'pending' THEN 0 ELSE 1 END, created_at DESC"
        ).fetchall()]


def approve_user(user_id: int, role: str, approved_by: str):
    """사용자 승인 + 역할 지정"""
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET status = 'approved', role = ?, approved_by = ? WHERE id = ?",
            (role, approved_by, user_id)
        )


def reject_user(user_id: int):
    """사용자 가입 거절"""
    with get_conn() as conn:
        conn.execute("UPDATE users SET status = 'rejected' WHERE id = ?", (user_id,))


def update_user_role(user_id: int, new_role: str):
    """사용자 역할 변경"""
    with get_conn() as conn:
        conn.execute("UPDATE users SET role = ? WHERE id = ?", (new_role, user_id))


def delete_user(user_id: int):
    """사용자 삭제"""
    with get_conn() as conn:
        conn.execute("DELETE FROM users WHERE id = ?", (user_id,))


def update_password(user_id: int, new_password: str) -> bool:
    """비밀번호 변경"""
    if len(new_password) < 6:
        return False
    hashed, salt = _hash_password(new_password)
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET password = ?, salt = ? WHERE id = ?",
            (hashed, salt, user_id)
        )
    return True


def get_user_permissions(role: str) -> dict:
    """역할별 권한 반환"""
    return ROLES.get(role, ROLES["viewer"])