Draft：強化驗證、授權與稽核安全

app.py

@@ -302,6 +302,7 @@ def update_submenu(item_key):
 from frontend.page_carbon import render as render_carbon
 from frontend.page_supply_chain_risk import render as render_supply_chain_risk
 from frontend.page_ai_assistant import render as render_ai
+from frontend.page_security_audit import render as render_security_audit
 
 if menu_selection == "📊 營運分析看板":
     render_dashboard()
@@ -342,3 +343,6 @@ def update_submenu(item_key):
         gemini_model=gemini_model,
         username=principal.username,
     )
+
+elif menu_selection == "🔐 安全管理":
+    render_security_audit(username=principal.username)

backend/access_control.py

@@ -25,6 +25,7 @@
 GLOBAL_APPROVAL_DECIDE = "approval.global.decide"
 ERP_EXCHANGE_EXPORT = "erp.exchange.export"
 ERP_EXCHANGE_RECONCILE = "erp.exchange.reconcile"
+SECURITY_AUDIT_READ = "security.audit.read"
 
 L1_MONITOR = "l1_monitor"
 L2_DECISION = "l2_decision"
@@ -43,6 +44,7 @@
     GLOBAL_APPROVAL_DECIDE: L3_GOVERNED_ACTION,
     ERP_EXCHANGE_EXPORT: L3_GOVERNED_ACTION,
     ERP_EXCHANGE_RECONCILE: L3_GOVERNED_ACTION,
+    SECURITY_AUDIT_READ: L3_GOVERNED_ACTION,
 }
 
 _ALL_CAPABILITIES = frozenset(_CAPABILITY_ENTITLEMENT)
@@ -94,7 +96,8 @@ def capabilities_for_role(role: str) -> set[str]:
 
 
 @dataclass(frozen=True)
-class AccessContext:
+class Principal:
+    """已驗證、且已重新載入組織與 capability 的後端身分。"""
     username: str
     role: str
     name: str
@@ -106,15 +109,19 @@ def can(self, capability: str) -> bool:
         return capability in self.capabilities
 
 
+# 保留舊名稱，避免既有前端／測試在這次安全遷移中中斷。
+AccessContext = Principal
+
+
 def load_principal(
     username: str, *, conn: sqlite3.Connection | None = None
-) -> AccessContext | None:
+) -> Principal | None:
     """Reload one principal from SQLite; missing identity or membership denies."""
     username = str(username or "").strip()
     if not username:
         return None
 
-    def _load(active_conn: sqlite3.Connection) -> AccessContext | None:
+    def _load(active_conn: sqlite3.Connection) -> Principal | None:
         row = active_conn.execute(
             """
             SELECT u.username, u.role, u.name, membership.organization_id
@@ -148,7 +155,7 @@ def _load(active_conn: sqlite3.Connection) -> AccessContext | None:
             for capability in capabilities_for_role(row[1])
             if _CAPABILITY_ENTITLEMENT.get(capability) in entitlements
         )
-        return AccessContext(
+        return Principal(
             username=row[0],
             role=row[1],
             name=row[2],

backend/auth.py

@@ -3,27 +3,158 @@
 使用者驗證與角色型存取控制 (RBAC)
 """
 
+from datetime import datetime, timedelta, timezone
+
 import streamlit as st
 from .database import run_query
 
 
+MAX_FAILED_LOGIN_ATTEMPTS = 5
+LOGIN_ATTEMPT_WINDOW = timedelta(minutes=15)
+LOGIN_LOCK_DURATION = timedelta(minutes=15)
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _timestamp(value: datetime | None = None) -> str:
+    return (value or _now()).isoformat()
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(value)
+        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+    except ValueError:
+        return None
+
+
+def _ensure_auth_tables() -> None:
+    """支援尚未跑過 init_db 的既有部署與獨立驗證測試。"""
+    run_query(
+        """CREATE TABLE IF NOT EXISTS login_attempts (
+            username TEXT PRIMARY KEY,
+            failed_attempts INTEGER NOT NULL DEFAULT 0,
+            window_started_at TEXT,
+            locked_until TEXT
+        )""",
+        fetch=False,
+    )
+    run_query(
+        """CREATE TABLE IF NOT EXISTS auth_events (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT NOT NULL,
+            event_type TEXT NOT NULL,
+            occurred_at TEXT NOT NULL
+        )""",
+        fetch=False,
+    )
+
+
+def _record_auth_event(username: str, event_type: str) -> None:
+    run_query(
+        "INSERT INTO auth_events (username, event_type, occurred_at) VALUES (?, ?, ?)",
+        (username, event_type, _timestamp()),
+        fetch=False,
+    )
+
+
+def _is_login_locked(username: str) -> bool:
+    rows = run_query(
+        "SELECT locked_until FROM login_attempts WHERE username=?", (username,)
+    )
+    if not rows:
+        return False
+    locked_until = _parse_timestamp(rows[0][0])
+    if locked_until and locked_until > _now():
+        return True
+    if locked_until:
+        # 鎖定期結束後，重新開始計算新的失敗嘗試視窗。
+        run_query(
+            "UPDATE login_attempts SET failed_attempts=0, window_started_at=NULL, "
+            "locked_until=NULL WHERE username=?",
+            (username,),
+            fetch=False,
+        )
+    return False
+
+
+def _record_failed_login(username: str) -> None:
+    rows = run_query(
+        "SELECT failed_attempts, window_started_at FROM login_attempts WHERE username=?",
+        (username,),
+    )
+    now = _now()
+    attempts = 1
+    window_started_at = now
+    if rows:
+        previous_attempts, previous_window = rows[0]
+        parsed_window = _parse_timestamp(previous_window)
+        if parsed_window and now - parsed_window < LOGIN_ATTEMPT_WINDOW:
+            attempts = int(previous_attempts) + 1
+            window_started_at = parsed_window
+
+    if attempts >= MAX_FAILED_LOGIN_ATTEMPTS:
+        run_query(
+            """INSERT INTO login_attempts
+               (username, failed_attempts, window_started_at, locked_until)
+               VALUES (?, ?, ?, ?)
+               ON CONFLICT(username) DO UPDATE SET
+                   failed_attempts=excluded.failed_attempts,
+                   window_started_at=excluded.window_started_at,
+                   locked_until=excluded.locked_until""",
+            (username, attempts, _timestamp(window_started_at), _timestamp(now + LOGIN_LOCK_DURATION)),
+            fetch=False,
+        )
+        _record_auth_event(username, "login_locked")
+        return
+
+    run_query(
+        """INSERT INTO login_attempts
+           (username, failed_attempts, window_started_at, locked_until)
+           VALUES (?, ?, ?, NULL)
+           ON CONFLICT(username) DO UPDATE SET
+               failed_attempts=excluded.failed_attempts,
+               window_started_at=excluded.window_started_at,
+               locked_until=NULL""",
+        (username, attempts, _timestamp(window_started_at)),
+        fetch=False,
+    )
+    _record_auth_event(username, "login_failed")
+
+
+def _clear_failed_logins(username: str) -> None:
+    run_query("DELETE FROM login_attempts WHERE username=?", (username,), fetch=False)
+
+
 def check_login(username: str, password: str) -> dict | None:
     """驗證帳號密碼，成功回傳 {role, name}，失敗回傳 None。
-    （N3）密碼以 salted hash 比對；遇到 legacy 明文則於登入成功時就地升級。"""
-    from backend.passwords import verify_password, is_hashed, hash_password
+    密碼以 Argon2id 比對；舊 SHA-256 與明文格式會在成功登入時就地升級。"""
+    from backend.passwords import hash_password, needs_password_upgrade, verify_password
 
+    _ensure_auth_tables()
     rows = run_query(
         "SELECT password, role, name FROM users WHERE username=?",
         (username,),
     )
     if not rows:
         return None
+    if _is_login_locked(username):
+        _record_auth_event(username, "login_blocked_locked")
+        return None
     stored, role, name = rows[0]
     if not verify_password(password, stored or ""):
+        _record_failed_login(username)
         return None
-    if not is_hashed(stored or ""):  # legacy 明文 → 自我修復式升級
+    if needs_password_upgrade(stored or ""):
         run_query("UPDATE users SET password=? WHERE username=?",
                   (hash_password(password), username), fetch=False)
+        _record_auth_event(username, "password_upgraded")
+    _clear_failed_logins(username)
+    _record_auth_event(username, "login_succeeded")
     return {"role": role, "name": name}
 
 

backend/database.py

@@ -59,6 +59,23 @@ def init_db():
 
     # 使用者與權限
     c.execute('''CREATE TABLE IF NOT EXISTS users (username TEXT PRIMARY KEY, password TEXT, role TEXT, name TEXT)''')
+    c.execute('''CREATE TABLE IF NOT EXISTS line_user_identities (
+        line_user_id TEXT PRIMARY KEY,
+        username TEXT NOT NULL UNIQUE,
+        enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1))
+    )''')
+    c.execute('''CREATE TABLE IF NOT EXISTS login_attempts (
+        username TEXT PRIMARY KEY,
+        failed_attempts INTEGER NOT NULL DEFAULT 0,
+        window_started_at TEXT,
+        locked_until TEXT
+    )''')
+    c.execute('''CREATE TABLE IF NOT EXISTS auth_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        username TEXT NOT NULL,
+        event_type TEXT NOT NULL,
+        occurred_at TEXT NOT NULL
+    )''')
     c.execute('''CREATE TABLE IF NOT EXISTS user_organizations (
         username TEXT PRIMARY KEY,
         organization_id TEXT NOT NULL

backend/passwords.py

@@ -1,35 +1,72 @@
 """
 backend/passwords.py
-密碼雜湊（N3）：salted SHA-256，標準庫實作、零外部依賴。
+密碼雜湊：Argon2id；舊版 salted SHA-256 與明文帳號可在登入時遷移。
 
-儲存格式："sha256$<hex_salt>$<hex_digest>"
-不含 '$' 的舊值視為 legacy 明文 —— verify 時直接比對，
-並由 auth.check_login / database.init_db 在適當時機就地升級。
+新儲存格式為 argon2-cffi 的標準編碼（例如
+"$argon2id$v=19$m=65536,t=3,p=4$..."）。它含有演算法版本、成本參數與 salt，
+因此不需另行維護格式版本欄位。
 """
 
 import hashlib
-import os
+import hmac
+
+from argon2 import PasswordHasher
+from argon2.exceptions import InvalidHashError, VerificationError
+from argon2.low_level import Type
+
+
+# OWASP 的一般用途密碼雜湊建議：64 MiB 記憶體、3 次迭代、4 個平行度。
+_PASSWORD_HASHER = PasswordHasher(
+    time_cost=3,
+    memory_cost=65536,
+    parallelism=4,
+    hash_len=32,
+    salt_len=16,
+    type=Type.ID,
+)
 
 
 def hash_password(plain: str) -> str:
-    salt = os.urandom(16).hex()
-    digest = hashlib.sha256((salt + plain).encode("utf-8")).hexdigest()
-    return f"sha256${salt}${digest}"
+    """以目前的 Argon2id 參數雜湊密碼。"""
+    return _PASSWORD_HASHER.hash(plain)
 
 
 def verify_password(plain: str, stored: str) -> bool:
+    """驗證目前與舊版格式；呼叫端須在成功後檢查是否需要升級。"""
     if not stored:
         return False
+    if stored.startswith("$argon2id$"):
+        try:
+            return _PASSWORD_HASHER.verify(stored, plain)
+        except (InvalidHashError, VerificationError):
+            return False
+
     if "$" not in stored:  # legacy 明文（遷移前）
-        return plain == stored
+        return hmac.compare_digest(plain, stored)
     try:
         algo, salt, digest = stored.split("$", 2)
     except ValueError:
         return False
     if algo != "sha256":
         return False
-    return hashlib.sha256((salt + plain).encode("utf-8")).hexdigest() == digest
+    candidate = hashlib.sha256((salt + plain).encode("utf-8")).hexdigest()
+    return hmac.compare_digest(candidate, digest)
 
 
 def is_hashed(stored: str) -> bool:
-    return bool(stored) and stored.startswith("sha256$") and stored.count("$") == 2
+    """回傳值是否已是任一已知雜湊格式（供資料庫初始化保護舊資料）。"""
+    return bool(stored) and (
+        stored.startswith("$argon2id$")
+        or (stored.startswith("sha256$") and stored.count("$") == 2)
+    )
+
+
+def needs_password_upgrade(stored: str) -> bool:
+    """成功驗證後，這筆密碼是否應以目前 Argon2id 參數重寫。"""
+    if not stored.startswith("$argon2id$"):
+        return True
+    try:
+        return _PASSWORD_HASHER.check_needs_rehash(stored)
+    except InvalidHashError:
+        # 正常流程中此情形不會通過 verify；保守地要求升級。
+        return True