diff --git a/app.py b/app.py index 391ba2d..1c71597 100644 --- a/app.py +++ b/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) diff --git a/backend/access_control.py b/backend/access_control.py index e76536d..2358bca 100644 --- a/backend/access_control.py +++ b/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], diff --git a/backend/auth.py b/backend/auth.py index 58a10a4..d356ebc 100644 --- a/backend/auth.py +++ b/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} diff --git a/backend/database.py b/backend/database.py index 4af17ec..bd5408a 100644 --- a/backend/database.py +++ b/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 diff --git a/backend/passwords.py b/backend/passwords.py index 4f6b4b7..a01826f 100644 --- a/backend/passwords.py +++ b/backend/passwords.py @@ -1,35 +1,72 @@ """ backend/passwords.py -密碼雜湊（N3）：salted SHA-256，標準庫實作、零外部依賴。 +密碼雜湊：Argon2id；舊版 salted SHA-256 與明文帳號可在登入時遷移。 -儲存格式："sha256$$" -不含 '$' 的舊值視為 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 diff --git a/backend/security_audit.py b/backend/security_audit.py new file mode 100644 index 0000000..e4e901d --- /dev/null +++ b/backend/security_audit.py @@ -0,0 +1,63 @@ +"""查詢登入安全稽核事件；只允許具備 security.audit.read 的管理者。""" + +from __future__ import annotations + +from datetime import date, timedelta + +from backend.access_control import SECURITY_AUDIT_READ, require_capability +from backend.database import run_query + + +EVENT_LABELS = { + "login_failed": "登入失敗", + "login_locked": "帳號已鎖定", + "login_blocked_locked": "鎖定期間登入遭拒", + "login_succeeded": "登入成功", + "password_upgraded": "密碼已升級為 Argon2id", +} + + +def list_auth_events( + requester_username: str, + *, + target_username: str = "", + event_type: str = "", + start_date: date | None = None, + end_date: date | None = None, + limit: int = 500, +) -> list[dict[str, str]]: + """回傳經過授權的登入稽核紀錄，絕不包含密碼或雜湊值。""" + require_capability(requester_username, SECURITY_AUDIT_READ) + + clauses: list[str] = [] + params: list[str] = [] + target_username = str(target_username or "").strip() + event_type = str(event_type or "").strip() + if target_username: + clauses.append("username = ?") + params.append(target_username) + if event_type: + clauses.append("event_type = ?") + params.append(event_type) + if start_date: + clauses.append("occurred_at >= ?") + params.append(f"{start_date.isoformat()} 00:00:00") + if end_date: + clauses.append("occurred_at < ?") + params.append(f"{(end_date + timedelta(days=1)).isoformat()} 00:00:00") + + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + bounded_limit = max(1, min(int(limit), 500)) + rows = run_query( + "SELECT username, event_type, occurred_at FROM auth_events" + f"{where} ORDER BY occurred_at DESC, id DESC LIMIT ?", + tuple(params + [bounded_limit]), + ) + return [ + { + "帳號": row[0], + "事件": EVENT_LABELS.get(row[1], row[1]), + "時間": row[2], + } + for row in rows + ] diff --git a/backend/tool_gateway.py b/backend/tool_gateway.py index 312ea0f..526120c 100644 --- a/backend/tool_gateway.py +++ b/backend/tool_gateway.py @@ -467,11 +467,15 @@ def call( risk_level = registry.get_risk_level(tool_name) if risk_level in {"write", "dangerous"} and not protected_po: - from backend.access_control import load_principal + from backend.access_control import ERP_POLICY_WRITE, load_principal principal = load_principal(actor or "") - if principal is None or principal.role != role: - msg = "寫入操作需要與登入身分一致的可驗證提案人。" + if ( + principal is None + or principal.role != role + or not principal.can(ERP_POLICY_WRITE) + ): + msg = "寫入操作需要與登入身分一致且具備 capability 的 Principal。" _write_log(tool_name, args, role, msg, success=False) return GatewayResult(status="denied", message=msg) actor = principal.username diff --git a/docs/security-foundation-plan.md b/docs/security-foundation-plan.md new file mode 100644 index 0000000..7ef5d1e --- /dev/null +++ b/docs/security-foundation-plan.md @@ -0,0 +1,54 @@ +# 安全基礎改善計畫 + +## 目的 + +本文件定義第一階段的實作計畫：在持續新增 AI 決策功能前，先強化身分識別、授權與 LINE 整合的安全邊界。 + +## 範圍 + +### 1. 密碼儲存與遷移 + +- 以 Argon2id 取代目前自訂的加鹽 SHA-256 密碼機制。 +- 為密碼雜湊保留版本欄位。 +- 舊版密碼成功登入後，透明地以 Argon2id 重新雜湊。 +- 為重複登入失敗新增節流與暫時鎖定機制。 +- 在稽核軌跡中記錄登入成功、登入失敗、鎖定與密碼遷移事件。 + +### 2. 後端授權 + +- 以明確的 `Principal` 物件表示已驗證的行為者：使用者 ID、組織 ID、角色、權限、驗證方式與工作階段到期時間。 +- 每個敏感的後端寫入操作都必須要求 Principal 與明確的權限檢查。 +- 對敏感操作，應從資料庫重新載入組織成員資格與權限狀態。 +- 身分、組織成員資格、權限或工作階段有效性缺失或已撤銷時，預設拒絕存取。 +- 不允許 Streamlit 執行環境判斷繞過後端授權。 + +### 3. LINE 身分識別邊界 + +- LINE 使用者 ID 僅能對應到已核准且有效的 ERP 身分。 +- LINE 身分未知、已撤銷或無法查詢時，預設拒絕存取。 +- 移除寬鬆的角色備援行為，例如將查詢失敗預設成倉儲角色。 +- 對使用者回傳安全的一般錯誤；技術細節只寫入伺服器日誌與稽核事件。 + +### 4. 驗證 + +- 單元測試：Argon2id 驗證、舊版密碼遷移、節流、無效 Principal、已撤銷使用者與未知 LINE 使用者。 +- 整合測試：敏感寫入操作必須拒絕未驗證或未授權的請求。 +- 迴歸測試：錯誤回覆不得洩漏堆疊追蹤、資料庫路徑、權杖或內部例外細節。 + +## 驗收條件 + +- [ ] 新密碼以 Argon2id 儲存。 +- [ ] 舊版憑證在成功登入後升級，且不暴露密碼。 +- [ ] 重複登入失敗會受到節流並留下紀錄。 +- [ ] 敏感後端寫入操作需要有效的 Principal 與權限。 +- [ ] 未知或已撤銷的 LINE 使用者無法存取 ERP 工具。 +- [ ] 使用者可見的錯誤不暴露內部技術細節。 +- [ ] 自動化測試涵蓋上述安全行為。 + +## 不包含的範圍 + +- SSO 或外部 IAM 整合 +- PostgreSQL 遷移 +- 共用資料庫多租戶 +- AI 決策證據紀錄與可解釋性 +- 外部 ERP outbox 或 worker 架構 diff --git a/frontend/access_navigation.py b/frontend/access_navigation.py index 755a1bb..5345c6c 100644 --- a/frontend/access_navigation.py +++ b/frontend/access_navigation.py @@ -27,6 +27,7 @@ "👥 人資": ["員工資料", "薪資", "出勤"], "🌿 碳排放管理": ["碳排放總覽", "碳足跡追蹤", "減量目標", "年度碳目標分析", "ESG 報告", "供應商風險與碳排"], "🌱 供應鏈與風險": [], + "🔐 安全管理": ["安全稽核紀錄"], } _LEGACY_ROLE_MENUS = { diff --git a/frontend/page_security_audit.py b/frontend/page_security_audit.py new file mode 100644 index 0000000..07bbdde --- /dev/null +++ b/frontend/page_security_audit.py @@ -0,0 +1,57 @@ +"""管理者安全稽核頁：只讀取登入事件，不暴露密碼資料。""" + +from __future__ import annotations + +from datetime import date, timedelta + +import streamlit as st + +from backend.security_audit import EVENT_LABELS, list_auth_events + + +def render(*, username: str) -> None: + st.title("🔐 安全稽核紀錄") + st.caption("僅顯示登入安全事件；系統不會在此頁顯示密碼或密碼雜湊。") + + today = date.today() + col_user, col_event, col_dates = st.columns([1, 1, 1.5]) + with col_user: + target_username = st.text_input("帳號篩選", placeholder="例如：admin") + with col_event: + selected_label = st.selectbox("事件類型", ["全部", *EVENT_LABELS.values()]) + with col_dates: + date_range = st.date_input( + "日期範圍", + value=(today - timedelta(days=30), today), + max_value=today, + ) + + label_to_event = {label: event for event, label in EVENT_LABELS.items()} + event_type = "" if selected_label == "全部" else label_to_event[selected_label] + start_date = date_range[0] if isinstance(date_range, tuple) and date_range else None + end_date = date_range[1] if isinstance(date_range, tuple) and len(date_range) == 2 else start_date + + try: + events = list_auth_events( + username, + target_username=target_username, + event_type=event_type, + start_date=start_date, + end_date=end_date, + ) + except PermissionError: + st.error("你沒有查看安全稽核紀錄的權限。") + st.stop() + + total = len(events) + locked = sum(event["事件"] == EVENT_LABELS["login_locked"] for event in events) + failed = sum(event["事件"] == EVENT_LABELS["login_failed"] for event in events) + metric_total, metric_locked, metric_failed = st.columns(3) + metric_total.metric("符合篩選的事件", total) + metric_locked.metric("帳號鎖定事件", locked) + metric_failed.metric("登入失敗事件", failed) + + if events: + st.dataframe(events, use_container_width=True, hide_index=True) + else: + st.info("此篩選條件下尚無安全事件。") diff --git a/line bot/bot_server.py b/line bot/bot_server.py index bcef2c0..d814269 100644 --- a/line bot/bot_server.py +++ b/line bot/bot_server.py @@ -102,12 +102,8 @@ def get_ngrok_url(): def _get_line_user_role(line_user_id: str) -> str: - """查詢 LINE 用戶的 ERP 角色，預設為 warehouse（受限角色）""" - try: - from backend.database import get_line_user_role - return get_line_user_role(line_user_id) - except Exception: - return _LINE_GATEWAY_DEFAULT_ROLE + """LINE 入口不要求帳號綁定，固定使用受限的 warehouse 權限。""" + return _LINE_GATEWAY_DEFAULT_ROLE def _build_gateway_function_response(tool_name: str, args: dict, role: str = None) -> tuple[dict, bool]: @@ -136,6 +132,9 @@ def _build_gateway_function_response(tool_name: str, args: dict, role: str = Non return payload, False +LINE_TOOLS = build_line_tools(ALL_TOOLS, registry, role=_LINE_GATEWAY_DEFAULT_ROLE) + + def _gateway_payload_to_reply(payload: dict) -> str: if payload.get("status") == "ok": return str(payload.get("result") or payload.get("data") or "") @@ -171,9 +170,6 @@ def _write_line_dispatch_log(user_task: str, tool_name: str, args: dict): print(f"Error writing LINE dispatch log: {e}") -LINE_TOOLS = build_line_tools(ALL_TOOLS, registry, role=_LINE_GATEWAY_DEFAULT_ROLE) - - def _chunk_reply_for_flex(text: str) -> list[str]: text = (text or "").strip() or "（無內容）" max_total = _FLEX_MAX_CHUNKS * _FLEX_CHUNK @@ -574,20 +570,22 @@ async def execute_morning_briefing(): try: print("Starting morning briefing generation...") prompt = "這是固定的每日早報排程。請只使用 LINE 低權白名單工具，彙整庫存狀態、採購情形、風險事件與碳排，產出【今日營運總結早報】。不得查詢或推測人資、薪資或財務資料。請主動列出應注意的風險或低庫存品項。" - - reply_text, dashboards = await asyncio.to_thread(get_ai_response, prompt) - - if not reply_text: - reply_text = "今日暫無早報資訊可提供。" - - reply_msgs = [reply_text_to_flex_message(reply_text, title="☀️ 營運早報主動推播")] - + with ApiClient(configuration) as api_client: line_bot_api = MessagingApi(api_client) sent = 0 failed = 0 for user_id in LINE_BRIEFING_USER_IDS: try: + reply_text, _ = await asyncio.to_thread( + get_ai_response, prompt, user_id=user_id, + erp_role=_get_line_user_role(user_id) + ) + if not reply_text: + reply_text = "今日暫無早報資訊可提供。" + reply_msgs = [ + reply_text_to_flex_message(reply_text, title="☀️ 營運早報主動推播") + ] line_bot_api.push_message( PushMessageRequest(to=user_id, messages=reply_msgs[:5]) ) @@ -595,8 +593,15 @@ async def execute_morning_briefing(): except Exception as e: failed += 1 print(f"Morning briefing push failed for one allowlisted user: {e}") - print(f"Morning briefing finished: sent={sent}, failed={failed}.") - return {"status": "completed", "sent": sent, "failed": failed} + print( + "Morning briefing finished: " + f"sent={sent}, failed={failed}." + ) + return { + "status": "completed", + "sent": sent, + "failed": failed, + } except Exception as e: print(f"Error in execute_morning_briefing: {e}") diff --git a/requirements.txt b/requirements.txt index 0220a93..2b35e8a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,3 +13,4 @@ line-bot-sdk==3.23.0 uvicorn==0.48.0 litellm==1.88.1 cachetools==5.5.0 +argon2-cffi==25.1.0 diff --git a/tests/test_password_hash.py b/tests/test_password_hash.py index a8441bc..39ef782 100644 --- a/tests/test_password_hash.py +++ b/tests/test_password_hash.py @@ -1,15 +1,24 @@ """ tests/test_password_hash.py -N3：密碼雜湊（salted SHA-256）+ legacy 明文自我修復升級。 +密碼雜湊（Argon2id）與舊版 SHA-256／明文的自我修復升級。 """ -from backend.passwords import hash_password, verify_password, is_hashed +import hashlib + +from backend.passwords import ( + hash_password, + is_hashed, + needs_password_upgrade, + verify_password, +) from backend.database import run_query def test_hash_roundtrip_and_uniqueness(): h = hash_password("admin") + assert h.startswith("$argon2id$") assert is_hashed(h) + assert not needs_password_upgrade(h) assert verify_password("admin", h) assert not verify_password("wrong", h) assert hash_password("admin") != h # salt 不同 → 同密碼不同雜湊 @@ -19,6 +28,20 @@ def test_legacy_plaintext_still_verifies(): assert verify_password("admin", "admin") # 遷移前的明文可登入 assert not verify_password("admin", "other") assert not is_hashed("admin") + assert needs_password_upgrade("admin") + + +def _legacy_sha256_hash(plain: str, salt: str = "legacy-salt") -> str: + digest = hashlib.sha256((salt + plain).encode("utf-8")).hexdigest() + return f"sha256${salt}${digest}" + + +def test_legacy_sha256_still_verifies_and_requires_upgrade(): + stored = _legacy_sha256_hash("oldpw") + assert is_hashed(stored) + assert verify_password("oldpw", stored) + assert not verify_password("wrong", stored) + assert needs_password_upgrade(stored) def _ensure_users_table(): @@ -27,6 +50,15 @@ def _ensure_users_table(): fetch=False) +def _auth_events(username: str) -> list[str]: + return [ + row[0] + for row in run_query( + "SELECT event_type FROM auth_events WHERE username=? ORDER BY id", (username,) + ) + ] + + def test_check_login_upgrades_legacy_row(): from backend.auth import check_login _ensure_users_table() @@ -44,6 +76,67 @@ def test_check_login_upgrades_legacy_row(): assert check_login("no_such_user", "x") is None +def test_check_login_upgrades_legacy_sha256_row(): + from backend.auth import check_login + + _ensure_users_table() + old_hash = _legacy_sha256_hash("pw123") + run_query( + "INSERT OR REPLACE INTO users VALUES ('sha_u', ?, 'sales', '測試')", + (old_hash,), + fetch=False, + ) + + assert check_login("sha_u", "pw123") == {"role": "sales", "name": "測試"} + stored = run_query("SELECT password FROM users WHERE username='sha_u'")[0][0] + assert stored.startswith("$argon2id$") + assert verify_password("pw123", stored) + + +def test_five_failed_logins_temporarily_lock_an_account(): + from backend.auth import check_login + + _ensure_users_table() + run_query( + "INSERT OR REPLACE INTO users VALUES ('locked_u', ?, 'sales', '測試')", + (hash_password("correct"),), + fetch=False, + ) + + for _ in range(5): + assert check_login("locked_u", "wrong") is None + + attempt = run_query( + "SELECT failed_attempts, locked_until FROM login_attempts WHERE username='locked_u'" + )[0] + assert attempt[0] == 5 + assert attempt[1] is not None + assert check_login("locked_u", "correct") is None + assert _auth_events("locked_u") == [ + "login_failed", + "login_failed", + "login_failed", + "login_failed", + "login_locked", + "login_blocked_locked", + ] + + +def test_successful_login_clears_failed_attempts_and_is_audited(): + from backend.auth import check_login + + _ensure_users_table() + run_query( + "INSERT OR REPLACE INTO users VALUES ('clear_u', ?, 'sales', '測試')", + (hash_password("correct"),), + fetch=False, + ) + assert check_login("clear_u", "wrong") is None + assert check_login("clear_u", "correct") is not None + assert not run_query("SELECT * FROM login_attempts WHERE username='clear_u'") + assert _auth_events("clear_u")[-2:] == ["login_failed", "login_succeeded"] + + def test_init_db_migrates_legacy_rows(): """init_db 的一次性遷移會把既有明文列升級（種子雜湊與此走同一 helper）。""" from backend.database import init_db diff --git a/tests/test_security_audit.py b/tests/test_security_audit.py new file mode 100644 index 0000000..fbcd93b --- /dev/null +++ b/tests/test_security_audit.py @@ -0,0 +1,72 @@ +from datetime import date + +import pytest + +from backend.access_control import SECURITY_AUDIT_READ, capabilities_for_role +from backend.database import init_db, run_query +from backend.security_audit import list_auth_events + + +def _seed_admin() -> None: + init_db() + # 其他安全測試可能會撤銷 demo 組織資料；此測試明確建立所需授權邊界。 + run_query( + "INSERT OR REPLACE INTO users (username, password, role, name) VALUES (?, ?, ?, ?)", + ("admin", "not-used-in-this-test", "admin", "系統管理員"), + fetch=False, + ) + run_query( + "INSERT OR REPLACE INTO user_organizations (username, organization_id) VALUES (?, ?)", + ("admin", "demo-org"), + fetch=False, + ) + run_query( + "INSERT OR REPLACE INTO app_metadata (key, value) VALUES (?, ?)", + ("deployment_organization_id", "demo-org"), + fetch=False, + ) + run_query( + "INSERT OR REPLACE INTO organization_entitlements " + "(organization_id, entitlement_key, enabled) VALUES (?, ?, 1)", + ("demo-org", "l3_governed_action"), + fetch=False, + ) + + +def test_only_admin_receives_security_audit_capability(): + assert SECURITY_AUDIT_READ in capabilities_for_role("admin") + assert SECURITY_AUDIT_READ not in capabilities_for_role("warehouse") + assert SECURITY_AUDIT_READ not in capabilities_for_role("procurement_approver") + + +def test_admin_can_filter_audited_events_without_password_data(): + _seed_admin() + run_query( + "INSERT INTO auth_events (username, event_type, occurred_at) VALUES (?, ?, ?)", + ("audit-target", "login_locked", "2026-09-13 10:00:00"), + fetch=False, + ) + run_query( + "INSERT INTO auth_events (username, event_type, occurred_at) VALUES (?, ?, ?)", + ("someone-else", "login_failed", "2026-09-13 10:01:00"), + fetch=False, + ) + + result = list_auth_events( + "admin", + target_username="audit-target", + event_type="login_locked", + start_date=date(2026, 9, 13), + end_date=date(2026, 9, 13), + ) + + assert result == [ + {"帳號": "audit-target", "事件": "帳號已鎖定", "時間": "2026-09-13 10:00:00"} + ] + assert "password" not in str(result).lower() + + +def test_non_admin_is_denied_security_audit_access(): + _seed_admin() + with pytest.raises(PermissionError): + list_auth_events("wh1") diff --git a/tests/test_tier_navigation.py b/tests/test_tier_navigation.py index ea196a8..d2529bd 100644 --- a/tests/test_tier_navigation.py +++ b/tests/test_tier_navigation.py @@ -73,6 +73,7 @@ def test_admin_keeps_full_existing_navigation_and_dashboard(): "Agent Dashboard", ] assert menu["🛒 採購管理"][-1] == "ERP CSV 交換" + assert menu["🔐 安全管理"] == ["安全稽核紀錄"] assert dashboard_mode(principal) == "full"