diff --git a/CHANGELOG.md b/CHANGELOG.md index b0f3cf8c3..47b749449 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,14 @@ ## [Unreleased] +### 도구 변경 경계 (Tool Mutation Boundary) + +- 프로세스 전역·비영속 레지스트리를 모든 인증 사용자가 변경할 수 있었던 + `POST /api/tools`, `PATCH /api/tools/{code}`, `DELETE /api/tools/{code}`를 + OpenAPI에서 숨긴 fail-closed tombstone으로 전환했습니다. 세 경로는 인증 후 + `501 tool_mutation_not_supported`를 반환하며, 요청 body를 검증하거나 레지스트리를 + 변경하거나 webhook DNS/egress를 시작하지 않습니다. webhook이 없는 사용자 정의 + 도구에 실제 작업 없이 성공을 반환하던 mock handler도 제거했습니다. 도구 목록·상세 + 조회와 기존 내장 도구 실행 계약은 변경하지 않았습니다. + - Starlette `TestClient`의 기존 `httpx2==2.5.0` pin을 core 개발·테스트 의존성으로 승격하고, deprecated `httpx` fallback 경고 억제를 제거했습니다. - 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다. - EmailDetail 테스트가 지원하지 않는 스레드 병합/분리 버튼을 `textContent`뿐 아니라 `aria-label`과 `title` 접근 가능 이름으로도 검출하도록 바꿔, 아이콘 전용 버튼 회귀를 놓치지 않습니다. diff --git a/backend/api/tools.py b/backend/api/tools.py index bd15abfac..d2d7d2a51 100644 --- a/backend/api/tools.py +++ b/backend/api/tools.py @@ -1,7 +1,6 @@ import base64 import hashlib import inspect -import json import logging import re import unicodedata @@ -9,7 +8,7 @@ import uuid from collections import Counter from collections.abc import Callable -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, NoReturn, Optional import httpx from core.url_validation import ( @@ -26,6 +25,13 @@ logger = logging.getLogger(__name__) ToolHandler = Callable[[Dict[str, Any]], Any] MAX_TOOL_FAILURE_MESSAGE_CHARS = 500 +TOOL_MUTATION_NOT_SUPPORTED_DETAIL = { + "error_code": "tool_mutation_not_supported", + "message": ( + "Dynamic tool mutations are disabled until tenant-scoped persistent " + "storage and administrative authorization are implemented." + ), +} def _tool_code_fingerprint(code: str) -> str: @@ -89,35 +95,6 @@ class ToolInfo(BaseModel): ) -class ToolCreate(BaseModel): - code: str = Field(..., description="도구의 고유 식별 코드") - name: str = Field(..., description="도구의 이름") - description: str = Field(..., description="도구에 대한 상세 설명") - category: str = Field(..., description="도구의 분류 (예: 이메일, 일정, 분석 등)") - parameters: Optional[Dict[str, Any]] = Field( - default=None, description="도구 실행에 필요한 파라미터 스키마" - ) - is_active: bool = Field(default=True, description="도구의 활성화 여부") - webhook_url: Optional[str] = Field( - default=None, description="도구 실행을 위한 외부 웹훅 URL" - ) - - -class ToolUpdate(BaseModel): - name: Optional[str] = Field(default=None, description="도구의 이름") - description: Optional[str] = Field( - default=None, description="도구에 대한 상세 설명" - ) - category: Optional[str] = Field(default=None, description="도구의 분류") - parameters: Optional[Dict[str, Any]] = Field( - default=None, description="도구 실행에 필요한 파라미터 스키마" - ) - is_active: Optional[bool] = Field(default=None, description="도구의 활성화 여부") - webhook_url: Optional[str] = Field( - default=None, description="도구 실행을 위한 외부 웹훅 URL" - ) - - class ExecuteRequest(BaseModel): parameters: Dict[str, Any] = Field( default_factory=dict, description="실행 파라미터" @@ -191,11 +168,6 @@ def _validate_parameters(self, code: str, params: Dict[str, Any]) -> Dict[str, A # Initialize default tools -async def mock_handler(params: Dict[str, Any]) -> str: - encoded = json.dumps(params, ensure_ascii=False, sort_keys=True) - return f"Mock execution successful with params: {encoded}" - - async def thread_summarizer_handler(params: Dict[str, Any]) -> Any: thread_id = params.get("thread_id", "") return { @@ -247,7 +219,6 @@ async def tone_analyzer_handler(params: Dict[str, Any]) -> Any: "tone_score": 85, } - def _detect_text_language(text: str) -> str: if any("\uac00" <= char <= "\ud7a3" for char in text): return "ko" @@ -275,10 +246,7 @@ async def email_translator_handler(params: Dict[str, Any]) -> Any: ] translated_terms: list[str] = [] for source_phrase, translated_phrase in phrase_map: - if ( - source_phrase in lowered_text - and translated_phrase not in translated_terms - ): + if source_phrase in lowered_text and translated_phrase not in translated_terms: translated_terms.append(translated_phrase) translated_text = " ".join(translated_terms) if translated_terms else text confidence = 0.9 if translated_terms else 0.45 @@ -297,9 +265,7 @@ async def spam_phishing_detector_handler(params: Dict[str, Any]) -> Any: normalized_domain = sender_domain.lower() phishing_terms = {"password", "bank", "login", "verify", "account", "credential"} spam_terms = {"urgent", "now", "free", "winner", "click", "limited"} - phishing_hits = sorted( - term for term in phishing_terms if term in normalized_content - ) + phishing_hits = sorted(term for term in phishing_terms if term in normalized_content) spam_hits = sorted(term for term in spam_terms if term in normalized_content) suspicious_domain = ( normalized_domain.endswith((".ru", ".zip", ".tk")) @@ -322,9 +288,7 @@ async def spam_phishing_detector_handler(params: Dict[str, Any]) -> Any: warnings.append(f"sender domain looks suspicious: {sender_domain}") return { "is_spam": bool(spam_hits or suspicious_domain), - "is_phishing": bool( - len(phishing_hits) >= 2 or (phishing_hits and suspicious_domain) - ), + "is_phishing": bool(len(phishing_hits) >= 2 or (phishing_hits and suspicious_domain)), "risk_score": risk_score, "warnings": warnings, } @@ -349,15 +313,7 @@ async def sentiment_analyzer_handler(params: Dict[str, Any]) -> Any: text = params.get("text", "") normalized_text = text.lower() positive_terms = {"thank", "thanks", "great", "good", "excellent", "감사", "좋"} - negative_terms = { - "disappointed", - "urgent", - "issue", - "problem", - "bad", - "불만", - "문제", - } + negative_terms = {"disappointed", "urgent", "issue", "problem", "bad", "불만", "문제"} positive_hits = [term for term in positive_terms if term in normalized_text] negative_hits = [term for term in negative_terms if term in normalized_text] if negative_hits and len(negative_hits) >= len(positive_hits): @@ -551,7 +507,6 @@ def _parameter_matches_type(value: Any, expected_type: str) -> bool: tone_analyzer_handler, ) - async def text_analyzer_handler(params: Dict[str, Any]) -> Dict[str, int]: text = params.get("text", "") char_count = len(text) @@ -564,7 +519,6 @@ async def text_analyzer_handler(params: Dict[str, Any]) -> Dict[str, int]: "word_count": len(text.split()), } - registry.register( ToolInfo( code="text_analyzer", @@ -754,6 +708,7 @@ async def keyword_extractor_handler(params: Dict[str, Any]) -> Any: async def uuid_v4_generator_handler(params: Dict[str, Any]) -> Dict[str, str]: + """Generate one RFC 9562 UUID version 4 for the retained built-in utility.""" return {"uuid": str(uuid.uuid4())} @@ -769,7 +724,6 @@ async def uuid_v4_generator_handler(params: Dict[str, Any]) -> Dict[str, str]: ) - @router.get("/tools", response_model=list[ToolInfo]) def get_tools() -> list[ToolInfo]: """ @@ -778,30 +732,17 @@ def get_tools() -> list[ToolInfo]: return registry.get_all() -@router.post("/tools", response_model=ToolInfo, status_code=201) -def create_tool(tool_data: ToolCreate) -> ToolInfo: - """ - 새로운 도구를 등록합니다. - """ - if registry.get(tool_data.code): - raise HTTPException( - status_code=400, detail="Tool with this code already exists" - ) +def _reject_tool_mutation() -> NoReturn: + raise HTTPException( + status_code=501, + detail=TOOL_MUTATION_NOT_SUPPORTED_DETAIL, + ) - tool_info = ToolInfo(**tool_data.model_dump()) - if tool_info.webhook_url: - try: - handler = make_webhook_handler(tool_info.webhook_url) - except ValueError as e: - raise HTTPException( - status_code=400, detail=f"Invalid or unsafe webhook URL: {e}" - ) - else: - handler = mock_handler - - registry.register(tool_info, handler) - return tool_info +@router.post("/tools", include_in_schema=False, response_model=None) +def create_tool() -> NoReturn: + """Fail closed until custom tools have durable tenant-scoped ownership.""" + _reject_tool_mutation() @router.get("/tools/{code}", response_model=ToolInfo) @@ -815,49 +756,16 @@ def get_tool(code: str) -> ToolInfo: return tool -@router.patch("/tools/{code}", response_model=ToolInfo) -def update_tool(code: str, tool_data: ToolUpdate) -> ToolInfo: - """ - 특정 도구의 정보를 업데이트합니다. - """ - tool = registry.get(code) - if not tool: - raise HTTPException(status_code=404, detail="Tool not found") +@router.patch("/tools/{code}", include_in_schema=False, response_model=None) +def update_tool(code: str) -> NoReturn: + """Fail closed without mutating a process-global tool.""" + _reject_tool_mutation() - update_data = tool_data.model_dump(exclude_unset=True) - # Validate webhook URL first to avoid state inconsistency - handler = None - if "webhook_url" in update_data: - if update_data["webhook_url"]: - try: - handler = make_webhook_handler(update_data["webhook_url"]) - except ValueError as e: - raise HTTPException( - status_code=400, detail=f"Invalid or unsafe webhook URL: {e}" - ) - else: - handler = mock_handler - - # Apply updates safely - for key, value in update_data.items(): - setattr(tool, key, value) - - if handler: - registry.register(tool, handler) - - return tool - - -@router.delete("/tools/{code}", status_code=204) -def delete_tool(code: str) -> None: - """ - 특정 도구를 삭제(등록 해제)합니다. - """ - tool = registry.get(code) - if not tool: - raise HTTPException(status_code=404, detail="Tool not found") - registry.unregister(code) +@router.delete("/tools/{code}", include_in_schema=False, response_model=None) +def delete_tool(code: str) -> NoReturn: + """Fail closed without unregistering a process-global tool.""" + _reject_tool_mutation() @router.post("/tools/{code}/execute", response_model=ExecuteResponse) diff --git a/backend/tests/test_tools_api.py b/backend/tests/test_tools_api.py index 8e537cef7..a4308d240 100644 --- a/backend/tests/test_tools_api.py +++ b/backend/tests/test_tools_api.py @@ -5,10 +5,11 @@ import os import secrets import time -from unittest.mock import AsyncMock, patch +from unittest.mock import MagicMock, patch import httpx import pytest +from fastapi import FastAPI from fastapi.testclient import TestClient os.environ.setdefault("AUTH_SESSION_HMAC_SECRET", secrets.token_urlsafe(48)) @@ -62,6 +63,20 @@ def _signed_session_token() -> str: return f"{signing_input}.{_base64url_encode(signature)}" +def _assert_tool_mutation_not_supported(response) -> None: + assert response.status_code == 501 + assert response.json() == { + "detail": { + "error_code": "tool_mutation_not_supported", + "message": ( + "Dynamic tool mutations are disabled until tenant-scoped " + "persistent storage and administrative authorization are " + "implemented." + ), + } + } + + def test_tools_rejects_missing_signed_session(): with TestClient(app) as client: response = client.get("/api/tools") @@ -414,10 +429,9 @@ def error_handler(_params): assert records[0].exception_type == "ValueError" assert len(records[0].exception_traceback_fingerprint) == 12 int(records[0].exception_traceback_fingerprint, 16) - assert ( - records[0].tool_code_fingerprint - == hashlib.sha256(hostile_code.encode("utf-8")).hexdigest()[:12] - ) + assert records[0].tool_code_fingerprint == hashlib.sha256( + hostile_code.encode("utf-8") + ).hexdigest()[:12] assert response.message == r"failure\r\nforged_exception=true" assert "\r" not in response.message assert "\n" not in response.message @@ -519,30 +533,6 @@ async def test_text_analyzer_tool_success(): assert result["word_count"] == 6 -@pytest.mark.asyncio -async def test_uuid_v4_generator_tool_success(): - with TestClient(app) as client: - response = client.post( - "/api/tools/uuid_v4_generator/execute", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"parameters": {}}, - ) - assert response.status_code == 200 - data = response.json() - assert data["status"] == "success" - result = data["result"] - - # Check if the result has 'uuid' key - assert "uuid" in result - - # Validate UUID v4 format - import uuid - - generated_uuid = result["uuid"] - parsed_uuid = uuid.UUID(generated_uuid) - assert parsed_uuid.version == 4 - - @pytest.mark.asyncio async def test_base64_encoder_tool_success(): with TestClient(app) as client: @@ -588,14 +578,15 @@ async def test_base64_decoder_tool_invalid_input(): assert "Invalid Base64 string" in data["message"] -def test_create_tool_success(): +def test_create_tool_mutation_fails_closed_without_registry_write(): + code = "new_custom_tool" try: with TestClient(app) as client: response = client.post( "/api/tools", headers={"Authorization": f"Bearer {_signed_session_token()}"}, json={ - "code": "new_custom_tool", + "code": code, "name": "Custom Tool", "description": "Custom Description", "category": "Custom Category", @@ -603,257 +594,197 @@ def test_create_tool_success(): "is_active": True, }, ) - assert response.status_code == 201 - data = response.json() - assert data["code"] == "new_custom_tool" + _assert_tool_mutation_not_supported(response) + assert registry.get(code) is None + finally: + registry.unregister(code) + + +def test_create_tool_mutation_fails_closed_even_with_safe_webhook(): + code = "webhook_custom_tool" + try: + with patch("api.tools._resolve_global_addresses") as resolve_addresses: + with TestClient(app) as client: + response = client.post( + "/api/tools", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={ + "code": code, + "name": "Webhook Tool", + "description": "Calls an external webhook", + "category": "Custom Category", + "parameters": {"input": "string"}, + "webhook_url": "https://example.com/webhook", + }, + ) - tool = registry.get("new_custom_tool") - assert tool is not None - assert tool.name == "Custom Tool" + _assert_tool_mutation_not_supported(response) + assert registry.get(code) is None + resolve_addresses.assert_not_called() finally: - registry.unregister("new_custom_tool") + registry.unregister(code) -def test_create_tool_already_exists(): - with TestClient(app) as client: - response = client.post( - "/api/tools", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={ - "code": "thread_summarizer", - "name": "Should Fail", - "description": "Should Fail", - "category": "Test", - }, - ) - assert response.status_code == 400 - assert response.json() == {"detail": "Tool with this code already exists"} +def test_update_tool_mutation_fails_closed_without_registry_change(): + code = "update_tool" + def handler(_params): + return "ok" -def test_update_tool_success(): + original = ToolInfo( + code=code, + name="Old Name", + description="Old Desc", + category="Test", + ) + original_snapshot = original.model_copy(deep=True) try: - registry.register( - ToolInfo( - code="update_tool", - name="Old Name", - description="Old Desc", - category="Test", - ), - lambda p: "ok", - ) + registry.register(original, handler) with TestClient(app) as client: response = client.patch( - "/api/tools/update_tool", + f"/api/tools/{code}", headers={"Authorization": f"Bearer {_signed_session_token()}"}, json={"name": "New Name", "is_active": False}, ) - assert response.status_code == 200 - data = response.json() - assert data["name"] == "New Name" - assert data["is_active"] is False - - tool = registry.get("update_tool") - assert tool.name == "New Name" - assert tool.is_active is False + _assert_tool_mutation_not_supported(response) + assert registry.get(code) == original_snapshot + assert registry._handlers[code] is handler finally: - registry.unregister("update_tool") + registry.unregister(code) -def test_update_tool_with_webhook(): - try: - registry.register( - ToolInfo( - code="webhook_update_tool", - name="Old", - description="Old", - category="Test", - ), - lambda p: "ok", - ) +def test_delete_tool_mutation_fails_closed_without_registry_change(): + code = "delete_tool" - with patch( - "api.tools._resolve_global_addresses", - return_value=("93.184.216.34",), - ): - with TestClient(app) as client: - response = client.patch( - "/api/tools/webhook_update_tool", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"webhook_url": "https://example.com/webhook"}, - ) - - assert response.status_code == 200 - tool = registry.get("webhook_update_tool") - assert tool.webhook_url == "https://example.com/webhook" - finally: - registry.unregister("webhook_update_tool") + def handler(_params): + return "ok" - -def test_update_tool_remove_webhook(): + original = ToolInfo( + code=code, + name="Do Not Delete", + description="Do Not Delete", + category="Test", + ) try: - registry.register( - ToolInfo( - code="webhook_remove_tool", - name="Old", - description="Old", - category="Test", - webhook_url="https://example.com/webhook", - ), - lambda p: "ok", - ) + registry.register(original, handler) with TestClient(app) as client: - response = client.patch( - "/api/tools/webhook_remove_tool", + response = client.delete( + f"/api/tools/{code}", headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"webhook_url": None}, ) - assert response.status_code == 200 - tool = registry.get("webhook_remove_tool") - assert tool.webhook_url is None + _assert_tool_mutation_not_supported(response) + assert registry.get(code) == original + assert registry._handlers[code] is handler finally: - registry.unregister("webhook_remove_tool") + registry.unregister(code) -def test_update_tool_not_found(): +@pytest.mark.parametrize( + ("method", "path", "payload"), + [ + ( + "POST", + "/api/tools", + { + "code": "unauthorized_tool", + "name": "Unauthorized Tool", + "description": "Must not be registered", + "category": "Test", + }, + ), + ("PATCH", "/api/tools/thread_summarizer", {"name": "Unauthorized"}), + ("DELETE", "/api/tools/thread_summarizer", None), + ], +) +def test_tool_mutation_routes_require_signed_session(method, path, payload): with TestClient(app) as client: - response = client.patch( - "/api/tools/non_existent_tool", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"name": "New Name"}, - ) - assert response.status_code == 404 + response = client.request(method, path, json=payload) + assert response.status_code == 401 + assert response.json() == {"detail": "Authentication required"} -def test_delete_tool_success(): - try: - registry.register( - ToolInfo( - code="delete_tool", - name="To Delete", - description="To Delete", - category="Test", - ), - lambda p: "ok", + +@pytest.mark.parametrize( + ("method", "path"), + [ + ("POST", "/api/tools"), + ("PATCH", "/api/tools/non_existent_tool"), + ], +) +def test_tool_mutation_tombstones_do_not_validate_request_models(method, path): + with TestClient(app) as client: + response = client.request( + method, + path, + headers={ + "Authorization": f"Bearer {_signed_session_token()}", + "Content-Type": "application/json", + }, + content="{not-json", ) - with TestClient(app) as client: - response = client.delete( - "/api/tools/delete_tool", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - ) + _assert_tool_mutation_not_supported(response) - assert response.status_code == 204 - assert registry.get("delete_tool") is None - finally: - registry.unregister("delete_tool") +def test_tool_mutation_routes_are_hidden_from_openapi(): + from api.tools import router as tools_router -def test_delete_tool_not_found(): - with TestClient(app) as client: - response = client.delete( - "/api/tools/non_existent_tool", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - ) - assert response.status_code == 404 + schema_app = FastAPI() + schema_app.include_router(tools_router) + paths = schema_app.openapi()["paths"] + + assert "post" not in paths["/api/tools"] + assert "patch" not in paths["/api/tools/{code}"] + assert "delete" not in paths["/api/tools/{code}"] @pytest.mark.asyncio async def test_webhook_handler_success(): - try: - with patch( - "api.tools._resolve_global_addresses", - return_value=("93.184.216.34",), - ): - with TestClient(app) as client: - client.post( - "/api/tools", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={ - "code": "webhook_tool", - "name": "Webhook Tool", - "description": "Calls external webhook", - "category": "Test", - "parameters": {"input": "string"}, - "webhook_url": "https://example.com/webhook", - }, - ) - - with patch("httpx.AsyncClient.post") as mock_post: - mock_response = AsyncMock() - mock_response.json.return_value = { - "webhook_success": True - } # json() is sync, return_value returns coroutine from AsyncMock, wait... - from unittest.mock import MagicMock - - mock_response.json = MagicMock(return_value={"webhook_success": True}) - mock_response.raise_for_status = lambda: None - mock_post.return_value = mock_response - - with TestClient(app) as client: - response = client.post( - "/api/tools/webhook_tool/execute", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"parameters": {"input": "hello"}}, - ) - - assert response.status_code == 200 - data = response.json() - assert data["status"] == "success" - assert data["result"] == {"webhook_success": True} - - mock_post.assert_called_once() - args, kwargs = mock_post.call_args - assert args[0] == "https://example.com/webhook" - assert kwargs["json"] == {"parameters": {"input": "hello"}} + from api.tools import make_webhook_handler - finally: - registry.unregister("webhook_tool") + with patch( + "api.tools._resolve_global_addresses", + return_value=("93.184.216.34",), + ): + handler = make_webhook_handler("https://example.com/webhook") + with patch("httpx.AsyncClient.post") as mock_post: + mock_response = MagicMock() + mock_response.json.return_value = {"webhook_success": True} + mock_response.raise_for_status.return_value = None + mock_post.return_value = mock_response + + result = await handler({"input": "hello"}) + + assert result == {"webhook_success": True} + mock_post.assert_awaited_once_with( + "https://example.com/webhook", + json={"parameters": {"input": "hello"}}, + timeout=10.0, + ) @pytest.mark.asyncio async def test_webhook_handler_http_error(): - try: + from api.tools import make_webhook_handler + + with patch( + "api.tools._resolve_global_addresses", + return_value=("93.184.216.34",), + ): + handler = make_webhook_handler("https://example.com/webhook") with patch( - "api.tools._resolve_global_addresses", - return_value=("93.184.216.34",), + "httpx.AsyncClient.post", + side_effect=httpx.HTTPError("Simulated HTTP Error"), ): - with TestClient(app) as client: - client.post( - "/api/tools", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={ - "code": "webhook_fail_tool", - "name": "Webhook Fail Tool", - "description": "Calls external webhook", - "category": "Test", - "parameters": {"input": "string"}, - "webhook_url": "https://example.com/webhook", - }, - ) - - with patch("httpx.AsyncClient.post") as mock_post: - mock_post.side_effect = httpx.HTTPError("Simulated HTTP Error") - - with TestClient(app) as client: - response = client.post( - "/api/tools/webhook_fail_tool/execute", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"parameters": {"input": "hello"}}, - ) - - assert response.status_code == 200 - data = response.json() - assert data["status"] == "failed" - assert ( - "Webhook execution failed: Simulated HTTP Error" in data["message"] - ) - - finally: - registry.unregister("webhook_fail_tool") + with pytest.raises( + ValueError, + match="Webhook execution failed: Simulated HTTP Error", + ): + await handler({"input": "hello"}) def test_tool_registry_execute_no_handler(): @@ -924,46 +855,6 @@ def test_validate_parameters_not_dict(): registry._validate_parameters("some_code", "not a dict") # type: ignore -def test_create_tool_unsafe_webhook(): - with TestClient(app) as client: - response = client.post( - "/api/tools", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={ - "code": "unsafe_tool", - "name": "Unsafe", - "description": "Unsafe", - "category": "Test", - "webhook_url": "http://localhost:8080/admin", - }, - ) - assert response.status_code == 400 - assert "Invalid or unsafe webhook URL" in response.json()["detail"] - - -def test_update_tool_unsafe_webhook(): - try: - registry.register( - ToolInfo( - code="unsafe_update_tool", - name="Safe", - description="Safe", - category="Test", - ), - lambda p: "ok", - ) - with TestClient(app) as client: - response = client.patch( - "/api/tools/unsafe_update_tool", - headers={"Authorization": f"Bearer {_signed_session_token()}"}, - json={"webhook_url": "http://169.254.169.254/latest/meta-data/"}, - ) - assert response.status_code == 400 - assert "Invalid or unsafe webhook URL" in response.json()["detail"] - finally: - registry.unregister("unsafe_update_tool") - - def test_is_safe_webhook_url_coverage(): from api.tools import is_safe_webhook_url @@ -1078,14 +969,6 @@ def test_execute_grammar_checker(): assert data["result"]["errors_found"] == 3 -@pytest.mark.asyncio -async def test_mock_handler(): - from api.tools import mock_handler - - res = await mock_handler({"test": 123}) - assert "123" in res - - def test_validate_webhook_url_no_host(): from api.tools import validate_webhook_url diff --git a/backend/tests/test_tools_uuid_generator_contract.py b/backend/tests/test_tools_uuid_generator_contract.py new file mode 100644 index 000000000..f15ca254d --- /dev/null +++ b/backend/tests/test_tools_uuid_generator_contract.py @@ -0,0 +1,21 @@ +"""Regression contract for the retained UUID v4 built-in tool.""" + +from __future__ import annotations + +import uuid + +import pytest + +from api.tools import registry + + +@pytest.mark.asyncio +async def test_uuid_v4_generator_remains_available_after_mutation_freeze() -> None: + """Keep the safe built-in utility while disabling only dynamic mutations.""" + tool = registry.get("uuid_v4_generator") + + assert tool is not None + assert tool.parameters == {} + result = await registry.invoke_tool("uuid_v4_generator", {}) + generated_uuid = uuid.UUID(result["uuid"]) + assert generated_uuid.version == 4