From a6303b41280efffda2c7c26e85b4dba5716c1b90 Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 18:58:44 +0800 Subject: [PATCH 1/6] docs: add security foundation improvement plan This document outlines the Security Foundation Improvement Plan, detailing phases for enhancing identity, authorization, and LINE integration boundaries, along with acceptance criteria and out-of-scope items. --- docs/security-foundation-plan.md | 54 ++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 docs/security-foundation-plan.md diff --git a/docs/security-foundation-plan.md b/docs/security-foundation-plan.md new file mode 100644 index 0000000..87052cb --- /dev/null +++ b/docs/security-foundation-plan.md @@ -0,0 +1,54 @@ +# Security Foundation Improvement Plan + +## Purpose + +This document defines the first implementation phase for strengthening identity, authorization, and LINE integration boundaries before further AI decision features are added. + +## Scope + +### 1. Password storage and migration + +- Replace the current custom salted SHA-256 password scheme with Argon2id. +- Keep a version field for password hashes. +- On a successful legacy login, transparently re-hash the password with Argon2id. +- Add a rate limit and temporary lockout for repeated failed logins. +- Record successful logins, failures, lockouts, and password migration events in the audit trail. + +### 2. Backend authorization + +- Represent the authenticated actor with an explicit `Principal` object: user ID, organization ID, role, entitlements, authentication method, and session expiry. +- Require a principal and explicit capability check on every sensitive backend write. +- Reload membership and entitlement state from the database for sensitive operations. +- Fail closed when identity, organization membership, entitlement, or session validity is missing or revoked. +- Do not allow Streamlit runtime detection to bypass backend authorization. + +### 3. LINE identity boundary + +- Map LINE user IDs only to approved, active ERP identities. +- Deny access by default when a LINE identity is unknown, revoked, or cannot be looked up. +- Remove permissive role fallbacks such as defaulting lookup failures to a warehouse role. +- Return a generic, user-safe error message; write technical details only to server logs and audit events. + +### 4. Verification + +- Unit tests: Argon2id verification, legacy migration, rate limiting, invalid principals, revoked users, and unknown LINE users. +- Integration tests: sensitive writes reject unauthenticated or unauthorized requests. +- Regression tests: error replies do not disclose stack traces, database paths, tokens, or internal exception details. + +## Acceptance criteria + +- [ ] New passwords are stored with Argon2id. +- [ ] Legacy credentials upgrade on successful login without exposing passwords. +- [ ] Repeated failed logins are throttled and logged. +- [ ] Sensitive backend writes require a valid principal and capability. +- [ ] Unknown or revoked LINE users cannot access ERP tools. +- [ ] User-facing errors do not expose internal technical details. +- [ ] Automated tests cover the security behaviour above. + +## Out of scope + +- SSO or external IAM integration +- PostgreSQL migration +- Shared-database multi-tenancy +- AI decision-evidence records and explainability +- External ERP outbox or worker architecture From 0c793cf73805e4ba0066dd9a39827994199ca2dc Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 19:06:45 +0800 Subject: [PATCH 2/6] docs: translate security foundation plan to Traditional Chinese Translate the security foundation plan into Traditional Chinese while keeping necessary proper nouns. --- docs/security-foundation-plan.md | 78 ++++++++++++++++---------------- 1 file changed, 39 insertions(+), 39 deletions(-) diff --git a/docs/security-foundation-plan.md b/docs/security-foundation-plan.md index 87052cb..7ef5d1e 100644 --- a/docs/security-foundation-plan.md +++ b/docs/security-foundation-plan.md @@ -1,54 +1,54 @@ -# Security Foundation Improvement Plan +# 安全基礎改善計畫 -## Purpose +## 目的 -This document defines the first implementation phase for strengthening identity, authorization, and LINE integration boundaries before further AI decision features are added. +本文件定義第一階段的實作計畫：在持續新增 AI 決策功能前，先強化身分識別、授權與 LINE 整合的安全邊界。 -## Scope +## 範圍 -### 1. Password storage and migration +### 1. 密碼儲存與遷移 -- Replace the current custom salted SHA-256 password scheme with Argon2id. -- Keep a version field for password hashes. -- On a successful legacy login, transparently re-hash the password with Argon2id. -- Add a rate limit and temporary lockout for repeated failed logins. -- Record successful logins, failures, lockouts, and password migration events in the audit trail. +- 以 Argon2id 取代目前自訂的加鹽 SHA-256 密碼機制。 +- 為密碼雜湊保留版本欄位。 +- 舊版密碼成功登入後，透明地以 Argon2id 重新雜湊。 +- 為重複登入失敗新增節流與暫時鎖定機制。 +- 在稽核軌跡中記錄登入成功、登入失敗、鎖定與密碼遷移事件。 -### 2. Backend authorization +### 2. 後端授權 -- Represent the authenticated actor with an explicit `Principal` object: user ID, organization ID, role, entitlements, authentication method, and session expiry. -- Require a principal and explicit capability check on every sensitive backend write. -- Reload membership and entitlement state from the database for sensitive operations. -- Fail closed when identity, organization membership, entitlement, or session validity is missing or revoked. -- Do not allow Streamlit runtime detection to bypass backend authorization. +- 以明確的 `Principal` 物件表示已驗證的行為者：使用者 ID、組織 ID、角色、權限、驗證方式與工作階段到期時間。 +- 每個敏感的後端寫入操作都必須要求 Principal 與明確的權限檢查。 +- 對敏感操作，應從資料庫重新載入組織成員資格與權限狀態。 +- 身分、組織成員資格、權限或工作階段有效性缺失或已撤銷時，預設拒絕存取。 +- 不允許 Streamlit 執行環境判斷繞過後端授權。 -### 3. LINE identity boundary +### 3. LINE 身分識別邊界 -- Map LINE user IDs only to approved, active ERP identities. -- Deny access by default when a LINE identity is unknown, revoked, or cannot be looked up. -- Remove permissive role fallbacks such as defaulting lookup failures to a warehouse role. -- Return a generic, user-safe error message; write technical details only to server logs and audit events. +- LINE 使用者 ID 僅能對應到已核准且有效的 ERP 身分。 +- LINE 身分未知、已撤銷或無法查詢時，預設拒絕存取。 +- 移除寬鬆的角色備援行為，例如將查詢失敗預設成倉儲角色。 +- 對使用者回傳安全的一般錯誤；技術細節只寫入伺服器日誌與稽核事件。 -### 4. Verification +### 4. 驗證 -- Unit tests: Argon2id verification, legacy migration, rate limiting, invalid principals, revoked users, and unknown LINE users. -- Integration tests: sensitive writes reject unauthenticated or unauthorized requests. -- Regression tests: error replies do not disclose stack traces, database paths, tokens, or internal exception details. +- 單元測試：Argon2id 驗證、舊版密碼遷移、節流、無效 Principal、已撤銷使用者與未知 LINE 使用者。 +- 整合測試：敏感寫入操作必須拒絕未驗證或未授權的請求。 +- 迴歸測試：錯誤回覆不得洩漏堆疊追蹤、資料庫路徑、權杖或內部例外細節。 -## Acceptance criteria +## 驗收條件 -- [ ] New passwords are stored with Argon2id. -- [ ] Legacy credentials upgrade on successful login without exposing passwords. -- [ ] Repeated failed logins are throttled and logged. -- [ ] Sensitive backend writes require a valid principal and capability. -- [ ] Unknown or revoked LINE users cannot access ERP tools. -- [ ] User-facing errors do not expose internal technical details. -- [ ] Automated tests cover the security behaviour above. +- [ ] 新密碼以 Argon2id 儲存。 +- [ ] 舊版憑證在成功登入後升級，且不暴露密碼。 +- [ ] 重複登入失敗會受到節流並留下紀錄。 +- [ ] 敏感後端寫入操作需要有效的 Principal 與權限。 +- [ ] 未知或已撤銷的 LINE 使用者無法存取 ERP 工具。 +- [ ] 使用者可見的錯誤不暴露內部技術細節。 +- [ ] 自動化測試涵蓋上述安全行為。 -## Out of scope +## 不包含的範圍 -- SSO or external IAM integration -- PostgreSQL migration -- Shared-database multi-tenancy -- AI decision-evidence records and explainability -- External ERP outbox or worker architecture +- SSO 或外部 IAM 整合 +- PostgreSQL 遷移 +- 共用資料庫多租戶 +- AI 決策證據紀錄與可解釋性 +- 外部 ERP outbox 或 worker 架構 From 5dcee81af55c38cd2498eab731192b56fb45dfb0 Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 19:14:36 +0800 Subject: [PATCH 3/6] feat: migrate password storage to Argon2id --- backend/auth.py | 6 ++-- backend/passwords.py | 59 ++++++++++++++++++++++++++++++------- requirements.txt | 1 + tests/test_password_hash.py | 44 +++++++++++++++++++++++++-- 4 files changed, 94 insertions(+), 16 deletions(-) diff --git a/backend/auth.py b/backend/auth.py index 58a10a4..e94b50e 100644 --- a/backend/auth.py +++ b/backend/auth.py @@ -9,8 +9,8 @@ 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 rows = run_query( "SELECT password, role, name FROM users WHERE username=?", @@ -21,7 +21,7 @@ def check_login(username: str, password: str) -> dict | None: stored, role, name = rows[0] if not verify_password(password, stored or ""): 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) return {"role": role, "name": name} 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/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..3e18726 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(): @@ -44,6 +67,23 @@ 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_init_db_migrates_legacy_rows(): """init_db 的一次性遷移會把既有明文列升級（種子雜湊與此走同一 helper）。""" from backend.database import init_db From 8675f5e2db9efe5c071c9bed98d5d92313a4aa48 Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 19:19:35 +0800 Subject: [PATCH 4/6] feat: lock accounts after repeated login failures --- backend/auth.py | 131 ++++++++++++++++++++++++++++++++++++ backend/database.py | 12 ++++ tests/test_password_hash.py | 53 +++++++++++++++ 3 files changed, 196 insertions(+) diff --git a/backend/auth.py b/backend/auth.py index e94b50e..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。 密碼以 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 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..af02a23 100644 --- a/backend/database.py +++ b/backend/database.py @@ -59,6 +59,18 @@ 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 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/tests/test_password_hash.py b/tests/test_password_hash.py index 3e18726..39ef782 100644 --- a/tests/test_password_hash.py +++ b/tests/test_password_hash.py @@ -50,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() @@ -84,6 +93,50 @@ def test_check_login_upgrades_legacy_sha256_row(): 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 From 350fedf32a346a50b2305c532d45bdee04c06159 Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 19:26:42 +0800 Subject: [PATCH 5/6] feat: fail closed for LINE identities and capabilities --- backend/access_control.py | 40 ++++++++++++++-- backend/database.py | 5 ++ backend/tool_gateway.py | 10 ++-- line bot/bot_server.py | 96 ++++++++++++++++++++++--------------- tests/test_line_access.py | 5 +- tests/test_line_identity.py | 47 ++++++++++++++++++ 6 files changed, 156 insertions(+), 47 deletions(-) create mode 100644 tests/test_line_identity.py diff --git a/backend/access_control.py b/backend/access_control.py index e76536d..b9f7b49 100644 --- a/backend/access_control.py +++ b/backend/access_control.py @@ -94,7 +94,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 +107,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 +153,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], @@ -163,6 +168,33 @@ def _load(active_conn: sqlite3.Connection) -> AccessContext | None: return _load(owned_conn) +def resolve_line_principal( + line_user_id: str, *, conn: sqlite3.Connection | None = None +) -> Principal | None: + """將 LINE ID 對應為有效 Principal；任何缺漏都預設拒絕。""" + line_user_id = str(line_user_id or "").strip() + if not line_user_id: + return None + + def _resolve(active_conn: sqlite3.Connection) -> Principal | None: + try: + row = active_conn.execute( + """SELECT username FROM line_user_identities + WHERE line_user_id = ? AND enabled = 1""", + (line_user_id,), + ).fetchone() + except sqlite3.Error: + return None + if row is None: + return None + return load_principal(row[0], conn=active_conn) + + if conn is not None: + return _resolve(conn) + with sqlite3.connect(database.DB_FILE) as owned_conn: + return _resolve(owned_conn) + + def has_capability( username: str, capability: str, *, conn: sqlite3.Connection | None = None ) -> bool: diff --git a/backend/database.py b/backend/database.py index af02a23..bd5408a 100644 --- a/backend/database.py +++ b/backend/database.py @@ -59,6 +59,11 @@ 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, 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/line bot/bot_server.py b/line bot/bot_server.py index bcef2c0..4cdcb5d 100644 --- a/line bot/bot_server.py +++ b/line bot/bot_server.py @@ -98,22 +98,17 @@ def get_ngrok_url(): # Flex：單一 text 元件最多約 2000 字元；整則 Flex JSON 有大小上限，保守分段 _FLEX_CHUNK = 1400 _FLEX_MAX_CHUNKS = 5 -_LINE_GATEWAY_DEFAULT_ROLE = "warehouse" - - -def _get_line_user_role(line_user_id: str) -> str: - """查詢 LINE 用戶的 ERP 角色，預設為 warehouse（受限角色）""" +def _get_line_principal(line_user_id: str): + """LINE ID 必須對應到有效 ERP Principal；查詢失敗時拒絕。""" try: - from backend.database import get_line_user_role - return get_line_user_role(line_user_id) + from backend.access_control import resolve_line_principal + return resolve_line_principal(line_user_id) except Exception: - return _LINE_GATEWAY_DEFAULT_ROLE + return None -def _build_gateway_function_response(tool_name: str, args: dict, role: str = None) -> tuple[dict, bool]: - if role is None: - role = _LINE_GATEWAY_DEFAULT_ROLE - if not is_line_tool_allowed(tool_name, registry, role): +def _build_gateway_function_response(tool_name: str, args: dict, principal) -> tuple[dict, bool]: + if principal is None or not is_line_tool_allowed(tool_name, registry, principal.role): return ( { "status": "denied", @@ -121,7 +116,9 @@ def _build_gateway_function_response(tool_name: str, args: dict, role: str = Non }, False, ) - gw_result = gateway.call(tool_name, args or {}, role=role) + gw_result = gateway.call( + tool_name, args or {}, role=principal.role, actor=principal.username + ) payload = gw_result.to_dict() if gw_result.is_ok(): @@ -171,9 +168,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 @@ -250,7 +244,9 @@ def reply_text_to_flex_message(reply_text: str, title="進銷存助理") -> Flex return FlexMessage(alt_text=_flex_alt_text(reply_text), contents=bubble) -def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_prompt: str = "", user_id: str = None, erp_role: str = None) -> tuple[str, list[str]]: +def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_prompt: str = "", user_id: str = None, principal=None) -> tuple[str, list[str]]: + if principal is None: + return "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", [] from datetime import datetime today_str = datetime.now().strftime("%Y-%m-%d") current_year = datetime.now().year @@ -320,7 +316,7 @@ def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_promp contents=history, config=types.GenerateContentConfig( system_instruction=system_prompt, - tools=LINE_TOOLS, + tools=build_line_tools(ALL_TOOLS, registry, role=principal.role), temperature=0.2, # Match the Web orchestrator: the SDK must not execute # Python tools directly; every tool call goes through @@ -345,7 +341,7 @@ def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_promp try: args = dict(fc.args or {}) _write_line_dispatch_log(user_msg, fc.name, args) - payload, executed = _build_gateway_function_response(fc.name, args, role=erp_role) + payload, executed = _build_gateway_function_response(fc.name, args, principal) # 自動攔截有視覺化的工具，主動加入儀表板 if executed and fc.name in ["get_all_inventory", "get_low_stock_inventory", "calculate_smart_restocking"]: requested_dashboards.add("low_stock") @@ -498,7 +494,10 @@ def handle_text_message(event): line_bot_api = MessagingApi(api_client) user_msg = event.message.text.strip() user_id = event.source.user_id - erp_role = _get_line_user_role(user_id) + principal = _get_line_principal(user_id) + if principal is None: + _send_full_reply(event, line_bot_api, user_msg, "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) + return # 強制路由：查「全部/所有」庫存時，直接走後端工具，避免被 LLM 自由改寫成舊格式 direct_all_inventory = ( @@ -510,18 +509,18 @@ def handle_text_message(event): try: if direct_all_inventory: - payload, executed = _build_gateway_function_response("get_all_inventory", {}, role=erp_role) + payload, executed = _build_gateway_function_response("get_all_inventory", {}, principal) reply_text = _gateway_payload_to_reply(payload) dashboards = ["low_stock"] if executed else [] elif direct_smart_inventory: - low_stock_payload, low_stock_executed = _build_gateway_function_response("get_low_stock_inventory", {}, role=erp_role) - smart_payload, smart_executed = _build_gateway_function_response("calculate_smart_restocking", {}, role=erp_role) + low_stock_payload, low_stock_executed = _build_gateway_function_response("get_low_stock_inventory", {}, principal) + smart_payload, smart_executed = _build_gateway_function_response("calculate_smart_restocking", {}, principal) low_stock_text = _gateway_payload_to_reply(low_stock_payload) smart_text = _gateway_payload_to_reply(smart_payload) reply_text = f"{low_stock_text}\n\n🤖『AI 補貨建議』\n\n{smart_text}" dashboards = ["low_stock"] if (low_stock_executed or smart_executed) else [] else: - reply_text, dashboards = get_ai_response(user_msg, user_id=user_id, erp_role=erp_role) + reply_text, dashboards = get_ai_response(user_msg, user_id=user_id, principal=principal) except Exception as e: reply_text = f"❌ 抱歉，系統運作發生錯誤：{e}" dashboards = [] @@ -535,10 +534,13 @@ def handle_audio_message(event): line_bot_api = MessagingApi(api_client) blob_api = MessagingApiBlob(api_client) user_id = event.source.user_id - erp_role = _get_line_user_role(user_id) + principal = _get_line_principal(user_id) + if principal is None: + _send_full_reply(event, line_bot_api, "[語音訊息交辦]", "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) + return try: message_content = blob_api.get_message_content(event.message.id) - reply_text, dashboards = get_ai_response("", audio_bytes=message_content, user_id=user_id, erp_role=erp_role) + reply_text, dashboards = get_ai_response("", audio_bytes=message_content, user_id=user_id, principal=principal) except Exception as e: reply_text = f"❌ 抱歉，語音處理發生錯誤：{e}" dashboards = [] @@ -552,10 +554,13 @@ def handle_postback(event): line_bot_api = MessagingApi(api_client) data = event.postback.data user_id = event.source.user_id - erp_role = _get_line_user_role(user_id) + principal = _get_line_principal(user_id) + if principal is None: + _send_full_reply(event, line_bot_api, f"[按鈕觸發] {data}", "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) + return try: # 讓 AI 理解這是由按鈕觸發的指令 - reply_text, dashboards = get_ai_response(data, extra_system_prompt="使用者剛按下了一鍵觸發按鈕，請根據該按鈕指令執行相關作業並回報。", user_id=user_id, erp_role=erp_role) + reply_text, dashboards = get_ai_response(data, extra_system_prompt="使用者剛按下了一鍵觸發按鈕，請根據該按鈕指令執行相關作業並回報。", user_id=user_id, principal=principal) except Exception as e: reply_text = f"❌ 抱歉，快捷觸發發生錯誤：{e}" dashboards = [] @@ -574,20 +579,27 @@ 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 + skipped_unbound = 0 for user_id in LINE_BRIEFING_USER_IDS: + principal = _get_line_principal(user_id) + if principal is None: + skipped_unbound += 1 + print("Morning briefing skipped for an unbound or revoked LINE identity.") + continue try: + reply_text, _ = await asyncio.to_thread( + get_ai_response, prompt, user_id=user_id, principal=principal + ) + 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 +607,16 @@ 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}, skipped_unbound={skipped_unbound}." + ) + return { + "status": "completed", + "sent": sent, + "failed": failed, + "skipped_unbound": skipped_unbound, + } except Exception as e: print(f"Error in execute_morning_briefing: {e}") diff --git a/tests/test_line_access.py b/tests/test_line_access.py index 58af196..7bc0598 100644 --- a/tests/test_line_access.py +++ b/tests/test_line_access.py @@ -73,8 +73,9 @@ def test_line_gateway_rechecks_execution_boundary(): Path(__file__).resolve().parents[1] / "line bot" / "bot_server.py" ).read_text(encoding="utf-8") - assert "is_line_tool_allowed(tool_name, registry, role)" in source - assert "gateway.call(tool_name, args or {}, role=role)" in source + assert "is_line_tool_allowed(tool_name, registry, principal.role)" in source + assert "actor=principal.username" in source + assert "_get_line_principal(user_id)" in source def test_briefing_user_ids_are_trimmed_and_deduplicated(): diff --git a/tests/test_line_identity.py b/tests/test_line_identity.py new file mode 100644 index 0000000..7d598c4 --- /dev/null +++ b/tests/test_line_identity.py @@ -0,0 +1,47 @@ +"""LINE 身分必須映射到仍有效的 ERP Principal。""" + +import pytest + +from backend import database +from backend.access_control import Principal, resolve_line_principal + + +@pytest.fixture +def identity_db(tmp_path, monkeypatch): + monkeypatch.setattr(database, "DB_FILE", str(tmp_path / "line-identity.db")) + monkeypatch.setenv("ERP_DEMO_MODE", "1") + database.init_db() + + +def test_unbound_line_identity_is_denied(identity_db): + assert resolve_line_principal("U-unbound") is None + + +def test_line_identity_resolves_only_enabled_live_principal(identity_db): + database.run_query( + "INSERT INTO line_user_identities (line_user_id, username, enabled) VALUES (?, ?, 1)", + ("U-planner", "planner"), + fetch=False, + ) + + principal = resolve_line_principal("U-planner") + assert isinstance(principal, Principal) + assert principal.username == "planner" + + database.run_query( + "UPDATE line_user_identities SET enabled=0 WHERE line_user_id='U-planner'", + fetch=False, + ) + assert resolve_line_principal("U-planner") is None + + +def test_line_identity_denies_revoked_erp_membership(identity_db): + database.run_query( + "INSERT INTO line_user_identities (line_user_id, username, enabled) VALUES (?, ?, 1)", + ("U-viewer", "viewer"), + fetch=False, + ) + database.run_query( + "DELETE FROM user_organizations WHERE username='viewer'", fetch=False + ) + assert resolve_line_principal("U-viewer") is None From 3378d5f1e95324462d8b5d60d5629231d68672b5 Mon Sep 17 00:00:00 2001 From: evaeva12021202-sys Date: Sun, 13 Sep 2026 22:35:06 +0800 Subject: [PATCH 6/6] feat: add security audit page --- app.py | 4 ++ backend/access_control.py | 29 +------------ backend/security_audit.py | 63 ++++++++++++++++++++++++++++ frontend/access_navigation.py | 1 + frontend/page_security_audit.py | 57 +++++++++++++++++++++++++ line bot/bot_server.py | 73 +++++++++++++-------------------- tests/test_line_access.py | 5 +-- tests/test_line_identity.py | 47 --------------------- tests/test_security_audit.py | 72 ++++++++++++++++++++++++++++++++ tests/test_tier_navigation.py | 1 + 10 files changed, 231 insertions(+), 121 deletions(-) create mode 100644 backend/security_audit.py create mode 100644 frontend/page_security_audit.py delete mode 100644 tests/test_line_identity.py create mode 100644 tests/test_security_audit.py 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 b9f7b49..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) @@ -168,33 +170,6 @@ def _load(active_conn: sqlite3.Connection) -> Principal | None: return _load(owned_conn) -def resolve_line_principal( - line_user_id: str, *, conn: sqlite3.Connection | None = None -) -> Principal | None: - """將 LINE ID 對應為有效 Principal；任何缺漏都預設拒絕。""" - line_user_id = str(line_user_id or "").strip() - if not line_user_id: - return None - - def _resolve(active_conn: sqlite3.Connection) -> Principal | None: - try: - row = active_conn.execute( - """SELECT username FROM line_user_identities - WHERE line_user_id = ? AND enabled = 1""", - (line_user_id,), - ).fetchone() - except sqlite3.Error: - return None - if row is None: - return None - return load_principal(row[0], conn=active_conn) - - if conn is not None: - return _resolve(conn) - with sqlite3.connect(database.DB_FILE) as owned_conn: - return _resolve(owned_conn) - - def has_capability( username: str, capability: str, *, conn: sqlite3.Connection | None = None ) -> bool: 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/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 4cdcb5d..d814269 100644 --- a/line bot/bot_server.py +++ b/line bot/bot_server.py @@ -98,17 +98,18 @@ def get_ngrok_url(): # Flex：單一 text 元件最多約 2000 字元；整則 Flex JSON 有大小上限，保守分段 _FLEX_CHUNK = 1400 _FLEX_MAX_CHUNKS = 5 -def _get_line_principal(line_user_id: str): - """LINE ID 必須對應到有效 ERP Principal；查詢失敗時拒絕。""" - try: - from backend.access_control import resolve_line_principal - return resolve_line_principal(line_user_id) - except Exception: - return None +_LINE_GATEWAY_DEFAULT_ROLE = "warehouse" + + +def _get_line_user_role(line_user_id: str) -> str: + """LINE 入口不要求帳號綁定，固定使用受限的 warehouse 權限。""" + return _LINE_GATEWAY_DEFAULT_ROLE -def _build_gateway_function_response(tool_name: str, args: dict, principal) -> tuple[dict, bool]: - if principal is None or not is_line_tool_allowed(tool_name, registry, principal.role): +def _build_gateway_function_response(tool_name: str, args: dict, role: str = None) -> tuple[dict, bool]: + if role is None: + role = _LINE_GATEWAY_DEFAULT_ROLE + if not is_line_tool_allowed(tool_name, registry, role): return ( { "status": "denied", @@ -116,9 +117,7 @@ def _build_gateway_function_response(tool_name: str, args: dict, principal) -> t }, False, ) - gw_result = gateway.call( - tool_name, args or {}, role=principal.role, actor=principal.username - ) + gw_result = gateway.call(tool_name, args or {}, role=role) payload = gw_result.to_dict() if gw_result.is_ok(): @@ -133,6 +132,9 @@ def _build_gateway_function_response(tool_name: str, args: dict, principal) -> t 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 "") @@ -244,9 +246,7 @@ def reply_text_to_flex_message(reply_text: str, title="進銷存助理") -> Flex return FlexMessage(alt_text=_flex_alt_text(reply_text), contents=bubble) -def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_prompt: str = "", user_id: str = None, principal=None) -> tuple[str, list[str]]: - if principal is None: - return "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", [] +def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_prompt: str = "", user_id: str = None, erp_role: str = None) -> tuple[str, list[str]]: from datetime import datetime today_str = datetime.now().strftime("%Y-%m-%d") current_year = datetime.now().year @@ -316,7 +316,7 @@ def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_promp contents=history, config=types.GenerateContentConfig( system_instruction=system_prompt, - tools=build_line_tools(ALL_TOOLS, registry, role=principal.role), + tools=LINE_TOOLS, temperature=0.2, # Match the Web orchestrator: the SDK must not execute # Python tools directly; every tool call goes through @@ -341,7 +341,7 @@ def get_ai_response(user_msg: str, audio_bytes: bytes = None, extra_system_promp try: args = dict(fc.args or {}) _write_line_dispatch_log(user_msg, fc.name, args) - payload, executed = _build_gateway_function_response(fc.name, args, principal) + payload, executed = _build_gateway_function_response(fc.name, args, role=erp_role) # 自動攔截有視覺化的工具，主動加入儀表板 if executed and fc.name in ["get_all_inventory", "get_low_stock_inventory", "calculate_smart_restocking"]: requested_dashboards.add("low_stock") @@ -494,10 +494,7 @@ def handle_text_message(event): line_bot_api = MessagingApi(api_client) user_msg = event.message.text.strip() user_id = event.source.user_id - principal = _get_line_principal(user_id) - if principal is None: - _send_full_reply(event, line_bot_api, user_msg, "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) - return + erp_role = _get_line_user_role(user_id) # 強制路由：查「全部/所有」庫存時，直接走後端工具，避免被 LLM 自由改寫成舊格式 direct_all_inventory = ( @@ -509,18 +506,18 @@ def handle_text_message(event): try: if direct_all_inventory: - payload, executed = _build_gateway_function_response("get_all_inventory", {}, principal) + payload, executed = _build_gateway_function_response("get_all_inventory", {}, role=erp_role) reply_text = _gateway_payload_to_reply(payload) dashboards = ["low_stock"] if executed else [] elif direct_smart_inventory: - low_stock_payload, low_stock_executed = _build_gateway_function_response("get_low_stock_inventory", {}, principal) - smart_payload, smart_executed = _build_gateway_function_response("calculate_smart_restocking", {}, principal) + low_stock_payload, low_stock_executed = _build_gateway_function_response("get_low_stock_inventory", {}, role=erp_role) + smart_payload, smart_executed = _build_gateway_function_response("calculate_smart_restocking", {}, role=erp_role) low_stock_text = _gateway_payload_to_reply(low_stock_payload) smart_text = _gateway_payload_to_reply(smart_payload) reply_text = f"{low_stock_text}\n\n🤖『AI 補貨建議』\n\n{smart_text}" dashboards = ["low_stock"] if (low_stock_executed or smart_executed) else [] else: - reply_text, dashboards = get_ai_response(user_msg, user_id=user_id, principal=principal) + reply_text, dashboards = get_ai_response(user_msg, user_id=user_id, erp_role=erp_role) except Exception as e: reply_text = f"❌ 抱歉，系統運作發生錯誤：{e}" dashboards = [] @@ -534,13 +531,10 @@ def handle_audio_message(event): line_bot_api = MessagingApi(api_client) blob_api = MessagingApiBlob(api_client) user_id = event.source.user_id - principal = _get_line_principal(user_id) - if principal is None: - _send_full_reply(event, line_bot_api, "[語音訊息交辦]", "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) - return + erp_role = _get_line_user_role(user_id) try: message_content = blob_api.get_message_content(event.message.id) - reply_text, dashboards = get_ai_response("", audio_bytes=message_content, user_id=user_id, principal=principal) + reply_text, dashboards = get_ai_response("", audio_bytes=message_content, user_id=user_id, erp_role=erp_role) except Exception as e: reply_text = f"❌ 抱歉，語音處理發生錯誤：{e}" dashboards = [] @@ -554,13 +548,10 @@ def handle_postback(event): line_bot_api = MessagingApi(api_client) data = event.postback.data user_id = event.source.user_id - principal = _get_line_principal(user_id) - if principal is None: - _send_full_reply(event, line_bot_api, f"[按鈕觸發] {data}", "❌ 此 LINE 帳號尚未完成 ERP 身分綁定，無法存取系統資料。", []) - return + erp_role = _get_line_user_role(user_id) try: # 讓 AI 理解這是由按鈕觸發的指令 - reply_text, dashboards = get_ai_response(data, extra_system_prompt="使用者剛按下了一鍵觸發按鈕，請根據該按鈕指令執行相關作業並回報。", user_id=user_id, principal=principal) + reply_text, dashboards = get_ai_response(data, extra_system_prompt="使用者剛按下了一鍵觸發按鈕，請根據該按鈕指令執行相關作業並回報。", user_id=user_id, erp_role=erp_role) except Exception as e: reply_text = f"❌ 抱歉，快捷觸發發生錯誤：{e}" dashboards = [] @@ -584,16 +575,11 @@ async def execute_morning_briefing(): line_bot_api = MessagingApi(api_client) sent = 0 failed = 0 - skipped_unbound = 0 for user_id in LINE_BRIEFING_USER_IDS: - principal = _get_line_principal(user_id) - if principal is None: - skipped_unbound += 1 - print("Morning briefing skipped for an unbound or revoked LINE identity.") - continue try: reply_text, _ = await asyncio.to_thread( - get_ai_response, prompt, user_id=user_id, principal=principal + get_ai_response, prompt, user_id=user_id, + erp_role=_get_line_user_role(user_id) ) if not reply_text: reply_text = "今日暫無早報資訊可提供。" @@ -609,13 +595,12 @@ async def execute_morning_briefing(): print(f"Morning briefing push failed for one allowlisted user: {e}") print( "Morning briefing finished: " - f"sent={sent}, failed={failed}, skipped_unbound={skipped_unbound}." + f"sent={sent}, failed={failed}." ) return { "status": "completed", "sent": sent, "failed": failed, - "skipped_unbound": skipped_unbound, } except Exception as e: diff --git a/tests/test_line_access.py b/tests/test_line_access.py index 7bc0598..58af196 100644 --- a/tests/test_line_access.py +++ b/tests/test_line_access.py @@ -73,9 +73,8 @@ def test_line_gateway_rechecks_execution_boundary(): Path(__file__).resolve().parents[1] / "line bot" / "bot_server.py" ).read_text(encoding="utf-8") - assert "is_line_tool_allowed(tool_name, registry, principal.role)" in source - assert "actor=principal.username" in source - assert "_get_line_principal(user_id)" in source + assert "is_line_tool_allowed(tool_name, registry, role)" in source + assert "gateway.call(tool_name, args or {}, role=role)" in source def test_briefing_user_ids_are_trimmed_and_deduplicated(): diff --git a/tests/test_line_identity.py b/tests/test_line_identity.py deleted file mode 100644 index 7d598c4..0000000 --- a/tests/test_line_identity.py +++ /dev/null @@ -1,47 +0,0 @@ -"""LINE 身分必須映射到仍有效的 ERP Principal。""" - -import pytest - -from backend import database -from backend.access_control import Principal, resolve_line_principal - - -@pytest.fixture -def identity_db(tmp_path, monkeypatch): - monkeypatch.setattr(database, "DB_FILE", str(tmp_path / "line-identity.db")) - monkeypatch.setenv("ERP_DEMO_MODE", "1") - database.init_db() - - -def test_unbound_line_identity_is_denied(identity_db): - assert resolve_line_principal("U-unbound") is None - - -def test_line_identity_resolves_only_enabled_live_principal(identity_db): - database.run_query( - "INSERT INTO line_user_identities (line_user_id, username, enabled) VALUES (?, ?, 1)", - ("U-planner", "planner"), - fetch=False, - ) - - principal = resolve_line_principal("U-planner") - assert isinstance(principal, Principal) - assert principal.username == "planner" - - database.run_query( - "UPDATE line_user_identities SET enabled=0 WHERE line_user_id='U-planner'", - fetch=False, - ) - assert resolve_line_principal("U-planner") is None - - -def test_line_identity_denies_revoked_erp_membership(identity_db): - database.run_query( - "INSERT INTO line_user_identities (line_user_id, username, enabled) VALUES (?, ?, 1)", - ("U-viewer", "viewer"), - fetch=False, - ) - database.run_query( - "DELETE FROM user_organizations WHERE username='viewer'", fetch=False - ) - assert resolve_line_principal("U-viewer") is None 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"