diff --git a/line bot/bot_server.py b/line bot/bot_server.py index bcef2c0..4cbbf44 100644 --- a/line bot/bot_server.py +++ b/line bot/bot_server.py @@ -49,10 +49,12 @@ from backend.flex_builder import build_low_stock_flex, build_risk_events_flex from backend.chart_builder import build_carbon_trend_chart, build_finance_pie_chart from line_access import ( + build_line_gateway_response, build_line_tools, env_flag, is_line_tool_allowed, parse_line_user_ids, + resolve_line_role, ) # 確保資料庫初始化 @@ -98,42 +100,29 @@ def get_ngrok_url(): # Flex：單一 text 元件最多約 2000 字元；整則 Flex JSON 有大小上限，保守分段 _FLEX_CHUNK = 1400 _FLEX_MAX_CHUNKS = 5 +_DASHBOARD_AUTH_TOOLS = { + "low_stock": "get_low_stock_inventory", + "risk_events": "get_supply_chain_risk_events", + "chart_carbon": "get_carbon_emissions_by_month", + "chart_finance": "get_financial_overview", +} +# 僅用於建構模型可見的工具 Schema 白名單（LINE_TOOLS）。 +# 注意：此常數「不得」作為身分解析失敗時的執行期角色回退 —— 那會是 fail-open。 _LINE_GATEWAY_DEFAULT_ROLE = "warehouse" -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 - - -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", - "error": "LINE 入口僅允許唯讀或建議工具；寫入操作請由已登入的 Web 介面送審。", - }, - False, - ) - gw_result = gateway.call(tool_name, args or {}, role=role) - payload = gw_result.to_dict() +def _get_line_user_role(line_user_id: str) -> str | None: + """查詢 LINE 用戶的 ERP 角色（fail-closed）。 - if gw_result.is_ok(): - payload["result"] = gw_result.data - return payload, True + resolver 不可用、查詢例外、使用者不存在或角色空白時一律回傳 None； + 呼叫端必須拒絕，絕不回退 warehouse 或任何預設角色。 + """ + return resolve_line_role(line_user_id) - if gw_result.status == "pending": - payload["result"] = f"已送審批：{gw_result.message}" - return payload, False - payload["error"] = gw_result.message or f"Gateway returned status: {gw_result.status}" - return payload, False +def _build_gateway_function_response(tool_name: str, args: dict, role: str | None = None) -> tuple[dict, bool]: + """LINE 執行邊界：身分解析失敗（role=None）一律拒絕，不回退 warehouse。""" + return build_line_gateway_response(tool_name, args, registry, gateway, role) def _gateway_payload_to_reply(payload: dict) -> str: @@ -250,7 +239,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, 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, erp_role: str | None = None) -> tuple[str, list[str]]: from datetime import datetime today_str = datetime.now().strftime("%Y-%m-%d") current_year = datetime.now().year @@ -412,7 +401,9 @@ async def callback(request: Request, background_tasks: BackgroundTasks): raise HTTPException(status_code=400, detail="Invalid signature") return 'OK' -def _send_full_reply(event, line_bot_api, user_msg_log, reply_text, dashboards): +def _send_full_reply( + event, line_bot_api, user_msg_log, reply_text, dashboards, erp_role: str | None = None +): user_id = event.source.user_id try: profile = line_bot_api.get_profile(user_id) @@ -432,14 +423,18 @@ def _send_full_reply(event, line_bot_api, user_msg_log, reply_text, dashboards): reply_msgs = [reply_text_to_flex_message(reply_text)] # [防呆補強] 若 AI 偷懶沒呼叫工具（幻覺），只要字眼有關鍵字就強行加入視覺化圖表 - dashboards_set = set(dashboards) - if ("碳排" in user_msg_log or "探排" in user_msg_log or "ESG" in user_msg_log or "esg" in user_msg_log) and "chart_carbon" not in dashboards_set: + def dashboard_allowed(name: str) -> bool: + tool_name = _DASHBOARD_AUTH_TOOLS.get(name) + return bool(tool_name and is_line_tool_allowed(tool_name, registry, erp_role)) + + dashboards_set = {name for name in dashboards if dashboard_allowed(name)} + if dashboard_allowed("chart_carbon") and ("碳排" in user_msg_log or "探排" in user_msg_log or "ESG" in user_msg_log or "esg" in user_msg_log) and "chart_carbon" not in dashboards_set: dashboards_set.add("chart_carbon") - if ("財務" in user_msg_log or "資產" in user_msg_log) and "chart_finance" not in dashboards_set: + if dashboard_allowed("chart_finance") and ("財務" in user_msg_log or "資產" in user_msg_log) and "chart_finance" not in dashboards_set: dashboards_set.add("chart_finance") - if ("庫存" in user_msg_log or "補貨" in user_msg_log) and "low_stock" not in dashboards_set: + if dashboard_allowed("low_stock") and ("庫存" in user_msg_log or "補貨" in user_msg_log) and "low_stock" not in dashboards_set: dashboards_set.add("low_stock") - if ( + if dashboard_allowed("risk_events") and ( "供應鏈" in user_msg_log or "風險" in user_msg_log or "熱點" in user_msg_log @@ -526,7 +521,7 @@ def handle_text_message(event): reply_text = f"❌ 抱歉，系統運作發生錯誤：{e}" dashboards = [] - _send_full_reply(event, line_bot_api, user_msg, reply_text, dashboards) + _send_full_reply(event, line_bot_api, user_msg, reply_text, dashboards, erp_role) @handler.add(MessageEvent, message=AudioMessageContent) @@ -543,7 +538,7 @@ def handle_audio_message(event): reply_text = f"❌ 抱歉，語音處理發生錯誤：{e}" dashboards = [] - _send_full_reply(event, line_bot_api, "[語音訊息交辦]", reply_text, dashboards) + _send_full_reply(event, line_bot_api, "[語音訊息交辦]", reply_text, dashboards, erp_role) @handler.add(PostbackEvent) @@ -560,7 +555,7 @@ def handle_postback(event): reply_text = f"❌ 抱歉，快捷觸發發生錯誤：{e}" dashboards = [] - _send_full_reply(event, line_bot_api, f"[按鈕觸發] {data}", reply_text, dashboards) + _send_full_reply(event, line_bot_api, f"[按鈕觸發] {data}", reply_text, dashboards, erp_role) async def execute_morning_briefing(): diff --git a/line bot/line_access.py b/line bot/line_access.py index ab6ce58..a974bda 100644 --- a/line bot/line_access.py +++ b/line bot/line_access.py @@ -14,7 +14,7 @@ _BLOCKED_MODULES = {"hr", "finance"} -def is_line_tool_allowed(tool_name: str, registry, role: str) -> bool: +def is_line_tool_allowed(tool_name: str, registry, role: str | None) -> bool: """Apply the LINE boundary again at execution time, not only schema build.""" info = registry.get_tool_info(str(tool_name or "")) if not info: @@ -64,3 +64,73 @@ def build_line_tools(all_tools: Iterable, registry, role: str) -> list: continue allowed.append(tool) return allowed + + +_LINE_IDENTITY_DENIED_ERROR = ( + "LINE 使用者身分或角色無法解析，已拒絕本次操作；" + "請先完成 LINE 帳號與 ERP 角色綁定，或改由已登入的 Web 介面操作。" +) + + +def resolve_line_role(line_user_id: str | None, resolver=None) -> str | None: + """Resolve the LINE user's ERP role, fail-closed. + + Returns the stripped role string, or ``None`` when the identity is + missing/blank, the resolver cannot be imported, the lookup raises, or + the resolved role is missing/blank. Callers MUST deny on ``None`` — + there is no default-role fallback at the LINE boundary. + """ + if line_user_id is None or not str(line_user_id).strip(): + return None + if resolver is None: + try: + from backend.database import get_line_user_role + + resolver = get_line_user_role + except Exception: + return None + try: + role = resolver(str(line_user_id).strip()) + except Exception: + return None + if role is None or not str(role).strip(): + return None + return str(role).strip() + + +def build_line_gateway_response( + tool_name: str, args: dict, registry, gateway, role: str | None +) -> tuple[dict, bool]: + """Execute a LINE tool call through the Gateway, fail-closed on identity. + + ``role=None`` (or blank) means identity resolution failed: deny with a + consistent payload and never call the Gateway. Otherwise keep the + existing boundary — non-sensitive read-only/suggestion tools only, with + the Gateway as the final authorization check. + """ + if role is None or not str(role).strip(): + return ( + {"status": "denied", "error": _LINE_IDENTITY_DENIED_ERROR}, + False, + ) + if not is_line_tool_allowed(tool_name, registry, role): + return ( + { + "status": "denied", + "error": "LINE 入口僅允許唯讀或建議工具；寫入操作請由已登入的 Web 介面送審。", + }, + False, + ) + gw_result = gateway.call(tool_name, args or {}, role=role) + payload = gw_result.to_dict() + + if gw_result.is_ok(): + payload["result"] = gw_result.data + return payload, True + + if gw_result.status == "pending": + payload["result"] = f"已送審批：{gw_result.message}" + return payload, False + + payload["error"] = gw_result.message or f"Gateway returned status: {gw_result.status}" + return payload, False diff --git a/tests/test_line_access.py b/tests/test_line_access.py index 58af196..c54315d 100644 --- a/tests/test_line_access.py +++ b/tests/test_line_access.py @@ -1,7 +1,14 @@ from __future__ import annotations import importlib.util +import os +import socket +import sys +import tempfile from pathlib import Path +from types import SimpleNamespace + +import pytest MODULE_PATH = Path(__file__).resolve().parents[1] / "line bot" / "line_access.py" @@ -10,6 +17,46 @@ assert SPEC and SPEC.loader SPEC.loader.exec_module(line_access) +# ── bot_server 需要完整依賴（LINE SDK / google-genai / fastapi / backend）。 +# 強制覆寫為測試值，禁止沿用執行環境中的正式憑證。 +os.environ["LINE_CHANNEL_ACCESS_TOKEN"] = "test-token" +os.environ["LINE_CHANNEL_SECRET"] = "test-secret" +os.environ["GEMINI_API_KEY"] = "test-key" +_LINE_BOT_DIR = Path(__file__).resolve().parents[1] / "line bot" +if str(_LINE_BOT_DIR) not in sys.path: + sys.path.insert(0, str(_LINE_BOT_DIR)) + +# conftest 已在任何 backend import 前把 ERP_DB_PATH 指到暫存目錄。 +# 依賴或初始化若失敗，必須直接讓測試 collection 失敗，不可 skip 假綠。 +_IMPORT_NETWORK_ATTEMPTS = [] +_ORIGINAL_SOCKET_CONNECT = socket.socket.connect +_ORIGINAL_SOCKET_CONNECT_EX = socket.socket.connect_ex + + +def _deny_import_network(self, address): + _IMPORT_NETWORK_ATTEMPTS.append(address) + raise AssertionError(f"bot_server import attempted network access: {address!r}") + + +socket.socket.connect = _deny_import_network +socket.socket.connect_ex = _deny_import_network +try: + import bot_server +finally: + socket.socket.connect = _ORIGINAL_SOCKET_CONNECT + socket.socket.connect_ex = _ORIGINAL_SOCKET_CONNECT_EX + + +def test_bot_server_import_is_pinned_to_test_credentials_and_temp_db(): + from backend import database + + assert os.environ["LINE_CHANNEL_ACCESS_TOKEN"] == "test-token" + assert os.environ["LINE_CHANNEL_SECRET"] == "test-secret" + assert os.environ["GEMINI_API_KEY"] == "test-key" + assert Path(database.DB_FILE).resolve() == Path(os.environ["ERP_DB_PATH"]).resolve() + assert Path(database.DB_FILE).resolve().is_relative_to(Path(tempfile.gettempdir()).resolve()) + assert _IMPORT_NETWORK_ATTEMPTS == [] + def _tool(name): def tool(): @@ -37,6 +84,41 @@ def is_allowed(self, name, role): return role == "warehouse" and name != "sales_only" +class FakeGatewayResult: + def __init__(self, status: str, data=None, message: str = ""): + self.status = status + self.data = data + self.message = message + + def is_ok(self) -> bool: + return self.status == "ok" + + def to_dict(self) -> dict: + return { + "status": self.status, + "data": self.data, + "message": self.message, + "approval_id": "", + } + + +class SpyGateway: + """記錄每次 gateway.call；決不真正執行任何工具本體。""" + + def __init__(self, result: FakeGatewayResult | None = None): + self.calls: list[tuple] = [] + self.result = result or FakeGatewayResult("ok", data="SPY-RESULT") + + def call(self, tool_name: str, args: dict, role: str): + self.calls.append((tool_name, args, role)) + return self.result + + +# ───────────────────────────────────────────────────────────────────── +# 既有測試：LINE 白名單過濾（保留） +# ───────────────────────────────────────────────────────────────────── + + def test_line_tool_filter_is_fail_closed(): names = [ "inventory_read", @@ -69,9 +151,8 @@ def test_line_execution_guard_rejects_write_even_if_model_names_it(): def test_line_gateway_rechecks_execution_boundary(): - source = ( - Path(__file__).resolve().parents[1] / "line bot" / "bot_server.py" - ).read_text(encoding="utf-8") + # 執行期防線現在位於 line_access（bot_server 僅薄薄轉發）。 + source = MODULE_PATH.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 @@ -87,3 +168,221 @@ def test_briefing_flag_requires_explicit_truthy_value(): assert line_access.env_flag("ON") is True assert line_access.env_flag("false") is False assert line_access.env_flag("unexpected") is False + + +# ───────────────────────────────────────────────────────────────────── +# RED 批次：身分解析失敗路徑必須拒絕，不得回退 warehouse（fail-closed） +# ───────────────────────────────────────────────────────────────────── + + +def test_resolution_import_failure_denies_instead_of_falling_back_to_warehouse(monkeypatch): + from backend import database + + # 明確模擬 resolver 不存在，避免未來 database 新增同名函式後測試失去意義。 + monkeypatch.delattr(database, "get_line_user_role", raising=False) + assert bot_server._get_line_user_role("U1") is None + + +def test_resolution_query_exception_denies_instead_of_falling_back_to_warehouse(monkeypatch): + from backend import database + + def boom(line_user_id: str) -> str: + raise RuntimeError("db down") + + monkeypatch.setattr(database, "get_line_user_role", boom, raising=False) + # 目前實作 catch 例外後回退 "warehouse"（fail-open）；期望為 None（拒絕）。 + assert bot_server._get_line_user_role("U1") is None + + +def test_missing_or_blank_identity_denies_instead_of_falling_back_to_warehouse(): + # 缺失身分（None）與空白身分（""）都不應取得任何角色。 + assert bot_server._get_line_user_role("") is None + assert bot_server._get_line_user_role(None) is None + + +def test_empty_resolved_role_is_treated_as_unresolved(monkeypatch): + from backend import database + + monkeypatch.setattr(database, "get_line_user_role", lambda uid: "", raising=False) + # 目前實作直接回傳 ""；期望正規化為 None（拒絕）。 + assert bot_server._get_line_user_role("U1") is None + + +def test_nonempty_user_not_found_is_treated_as_unresolved(monkeypatch): + from backend import database + + monkeypatch.setattr(database, "get_line_user_role", lambda uid: None, raising=False) + assert bot_server._get_line_user_role("U-not-found") is None + + +def test_unresolved_role_never_enters_gateway(monkeypatch): + spy = SpyGateway() + monkeypatch.setattr(bot_server, "gateway", spy) + + payload, executed = bot_server._build_gateway_function_response( + "get_all_inventory", {}, role=None + ) + + # 目前實作把 None 回退為 warehouse 並呼叫 Gateway（fail-open）。 + assert spy.calls == [] + assert executed is False + + +def test_unresolved_role_returns_consistent_denied_payload(monkeypatch): + spy = SpyGateway() + monkeypatch.setattr(bot_server, "gateway", spy) + + payload, executed = bot_server._build_gateway_function_response( + "get_all_inventory", {}, role=None + ) + + # 目前實作會執行成功（status ok）；期望一致拒絕。 + assert executed is False + assert payload["status"] == "denied" + assert "身分" in payload.get("error", "") + assert spy.calls == [] + + +@pytest.mark.parametrize( + ("tool_name", "role"), + [ + ("inventory_read", "unknown-role"), + ("inventory_write", "warehouse"), + ("payroll", "warehouse"), + ("ledger", "warehouse"), + ("unknown", "warehouse"), + ], +) +def test_unknown_role_and_blocked_tools_never_enter_gateway(tool_name, role): + spy = SpyGateway() + + payload, executed = line_access.build_line_gateway_response( + tool_name, {}, FakeRegistry(), spy, role + ) + + assert payload["status"] == "denied" + assert executed is False + assert spy.calls == [] + + +def test_actual_registry_rejects_unknown_role_before_gateway(): + from backend.tool_registry import registry + + spy = SpyGateway() + payload, executed = line_access.build_line_gateway_response( + "get_all_inventory", {}, registry, spy, "unknown-role" + ) + + assert payload["status"] == "denied" + assert executed is False + assert spy.calls == [] + + +def test_get_ai_response_model_tool_call_with_unresolved_role_never_enters_gateway(monkeypatch): + class FakeModels: + def __init__(self): + self.calls = 0 + + def generate_content(self, **kwargs): + self.calls += 1 + if self.calls == 1: + function_call = SimpleNamespace(name="get_all_inventory", args={}) + return SimpleNamespace( + function_calls=[function_call], + text="", + candidates=[SimpleNamespace(content=SimpleNamespace(parts=[]))], + ) + return SimpleNamespace(function_calls=[], text="已拒絕未解析身分的工具呼叫") + + models = FakeModels() + spy = SpyGateway() + monkeypatch.setattr(bot_server, "client", SimpleNamespace(models=models)) + monkeypatch.setattr(bot_server, "gateway", spy) + monkeypatch.setattr(bot_server, "_write_line_dispatch_log", lambda *args: None) + + reply, dashboards = bot_server.get_ai_response( + "查詢全部庫存", user_id=None, erp_role=None + ) + + assert models.calls == 2 + assert spy.calls == [] + assert dashboards == [] + assert "拒絕" in reply + + +def test_valid_role_and_allowed_tool_still_execute_through_gateway(): + spy = SpyGateway() + + payload, executed = line_access.build_line_gateway_response( + "inventory_read", {"sku": "P001"}, FakeRegistry(), spy, "warehouse" + ) + + assert executed is True + assert payload["status"] == "ok" + assert payload["result"] == "SPY-RESULT" + assert spy.calls == [("inventory_read", {"sku": "P001"}, "warehouse")] + + +def test_unresolved_identity_does_not_trigger_keyword_dashboard_bypass(monkeypatch): + from backend import database + + dashboard_calls = [] + monkeypatch.setattr(database, "run_query", lambda *args, **kwargs: None) + monkeypatch.setattr( + bot_server, + "build_low_stock_flex", + lambda: dashboard_calls.append("low_stock"), + ) + + class FakeLineApi: + def get_profile(self, user_id): + return SimpleNamespace(display_name="Test") + + def reply_message_with_http_info(self, request): + return None + + event = SimpleNamespace( + source=SimpleNamespace(user_id="U-unresolved"), reply_token="reply-token" + ) + + # 舊路徑只看「庫存」關鍵字就直接呼叫 DB-backed dashboard builder， + # 即使身分解析失敗、Gateway 已拒絕也會繞過授權邊界。 + bot_server._send_full_reply( + event, FakeLineApi(), "查詢庫存", "身分解析失敗，已拒絕", [] + ) + + assert dashboard_calls == [] + + +def test_valid_warehouse_role_keeps_keyword_inventory_dashboard(monkeypatch): + from backend import database + + dashboard_calls = [] + monkeypatch.setattr(database, "run_query", lambda *args, **kwargs: None) + monkeypatch.setattr( + bot_server, + "build_low_stock_flex", + lambda: dashboard_calls.append("low_stock"), + ) + + class FakeLineApi: + def get_profile(self, user_id): + return SimpleNamespace(display_name="Test") + + def reply_message_with_http_info(self, request): + return None + + event = SimpleNamespace( + source=SimpleNamespace(user_id="U-warehouse"), reply_token="reply-token" + ) + + bot_server._send_full_reply( + event, + FakeLineApi(), + "查詢庫存", + "ok", + [], + erp_role="warehouse", + ) + + assert dashboard_calls == ["low_stock"]