feat: merge Xid-Catalog.xlsx, fix dmesg regex, and improve history UI
This commit is contained in:
@@ -0,0 +1,204 @@
|
||||
"""
|
||||
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"])
|
||||
Reference in New Issue
Block a user