feat(tools): add bounded content checksum generator

backend/api/content_checksum_tool.py

@@ -0,0 +1,122 @@
+"""Bounded cryptographic checksums for exact UTF-8 text content.
+
+This module owns the checksum algorithm allowlist and registers the tool with
+Naruon's existing deterministic tool catalog. Its digests compare content
+bytes; they are not proof of sender identity or message authenticity.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Callable
+from typing import Any
+
+from api.tools import ToolInfo, registry
+
+MAX_CONTENT_BYTES = 1_048_576
+SECURITY_NOTE = (
+    "Use this digest to compare exact content bytes; it does not authenticate "
+    "the sender or replace a MAC/signature."
+)
+
+
+class ContentChecksumError(ValueError):
+    """Expected checksum validation failure with a stable machine error code."""
+
+    def __init__(self, message: str, *, error_code: str) -> None:
+        """Initialize a customer-safe validation failure and deterministic code."""
+        super().__init__(message)
+        self.error_code = error_code
+
+
+def _sha256(payload: bytes) -> str:
+    """Return the SHA-256 hexadecimal digest for ``payload``."""
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _sha3_256(payload: bytes) -> str:
+    """Return the SHA-3-256 hexadecimal digest for ``payload``."""
+    return hashlib.sha3_256(payload).hexdigest()
+
+
+def _blake2b_256(payload: bytes) -> str:
+    """Return the 256-bit BLAKE2b hexadecimal digest for ``payload``."""
+    return hashlib.blake2b(payload, digest_size=32).hexdigest()
+
+
+_HASHERS: dict[str, Callable[[bytes], str]] = {
+    "sha256": _sha256,
+    "sha3_256": _sha3_256,
+    "blake2b_256": _blake2b_256,
+}
+
+
+async def content_checksum_handler(params: dict[str, Any]) -> dict[str, Any]:
+    """Hash exact UTF-8 bytes with an allowlisted modern checksum algorithm.
+
+    The input is never Unicode-normalized, so the digest compares the exact
+    byte representation Naruon received. Inputs larger than one MiB after
+    UTF-8 encoding are rejected before hashing.
+
+    Args:
+        params: Validated tool parameters containing ``text`` and ``algorithm``.
+
+    Returns:
+        A deterministic checksum receipt with the algorithm, digest, byte
+        length, encoding, and an authenticity warning.
+
+    Raises:
+        ContentChecksumError: If the algorithm is not allowlisted, the text
+            cannot be represented as valid UTF-8, or the encoded content exceeds
+            one MiB. Each expected failure carries a stable machine-readable
+            ``error_code``.
+    """
+    text = params["text"]
+    algorithm = params["algorithm"]
+    if algorithm not in _HASHERS:
+        raise ContentChecksumError(
+            "Unsupported checksum algorithm; choose sha256, sha3_256, or blake2b_256",
+            error_code="unsupported_checksum_algorithm",
+        )
+
+    try:
+        payload = text.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        raise ContentChecksumError(
+            "Content must contain valid Unicode scalar values",
+            error_code="content_checksum_invalid_utf8",
+        ) from exc
+    if len(payload) > MAX_CONTENT_BYTES:
+        raise ContentChecksumError(
+            f"Content exceeds {MAX_CONTENT_BYTES} UTF-8 bytes",
+            error_code="content_checksum_payload_too_large",
+        )
+
+    return {
+        "algorithm_code": algorithm,
+        "digest_hex": _HASHERS[algorithm](payload),
+        "byte_length": len(payload),
+        "encoding_code": "utf-8",
+        "security_note": SECURITY_NOTE,
+    }
+
+
+def register_content_checksum_tool() -> None:
+    """Register the checksum generator once in Naruon's built-in tool catalog."""
+    if registry.get("content_checksum_generator") is not None:
+        return
+
+    registry.register(
+        ToolInfo(
+            code="content_checksum_generator",
+            name="Content checksum generator",
+            description=(
+                "Compare exact UTF-8 content using SHA-256, SHA-3-256, or "
+                "BLAKE2b-256. Choose an algorithm, then compare the returned "
+                "digest with the expected value."
+            ),
+            category="유틸리티",
+            parameters={"text": "string", "algorithm": "string"},
+        ),
+        content_checksum_handler,
+    )

backend/api/tools.py

@@ -20,7 +20,7 @@
 )
 from services.llm_provider_urls import build_pinned_https_async_client
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, SerializerFunctionWrapHandler, model_serializer
 
 router = APIRouter(prefix="/api", tags=["tools"])
 logger = logging.getLogger(__name__)
@@ -125,9 +125,24 @@ class ExecuteRequest(BaseModel):
 
 
 class ExecuteResponse(BaseModel):
+    """Stable public envelope returned by tool execution endpoints."""
+
     status: str = Field(..., description="실행 상태 (예: success, failed)")
     result: Any = Field(..., description="실행 결과 데이터")
     message: Optional[str] = Field(default=None, description="결과 메시지")
+    error_code: Optional[str] = Field(
+        default=None, description="예상된 실패의 안정적인 기계 판독 오류 코드"
+    )
+
+    @model_serializer(mode="wrap")
+    def _serialize_response(
+        self, handler: SerializerFunctionWrapHandler
+    ) -> dict[str, Any]:
+        """Omit an absent error code while preserving legacy null result fields."""
+        payload = handler(self)
+        if self.error_code is None:
+            payload.pop("error_code", None)
+        return payload
 
 
 class ToolRegistry:
@@ -706,6 +721,8 @@ async def base64_decoder_handler(params: Dict[str, Any]) -> Dict[str, str]:
         "합니다",
     }
 )
+
+
 def _normalize_analysis_text(value: str) -> str:
     """Normalize user text for deterministic, multilingual rule matching."""
     if len(value) > ANALYSIS_TEXT_MAX_CHARS:
@@ -769,7 +786,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]:
     """
@@ -889,4 +905,5 @@ async def execute_tool(code: str, request: ExecuteRequest) -> ExecuteResponse:
             status="failed",
             result=None,
             message=_safe_tool_failure_message(e),
+            error_code=getattr(e, "error_code", None),
         )

backend/main.py

@@ -7,6 +7,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from api.auth import get_auth_context, preload_oidc_jwks
+from api.content_checksum_tool import register_content_checksum_tool
 from api.search import router as search_router
 from api.llm import router as llm_router
 from api.calendar import router as calendar_router
@@ -42,6 +43,8 @@
 from services.reply_sla_scheduler import ReplySlaScheduler
 from prometheus_fastapi_instrumentator import Instrumentator
 
+register_content_checksum_tool()
+
 imap_worker = ImapSyncWorker()
 pop3_worker = Pop3SyncWorker()
 reply_sla_scheduler = ReplySlaScheduler()

backend/tests/test_content_checksum_api.py

@@ -0,0 +1,149 @@
+"""Public API contract tests for the bounded content-checksum tool."""
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import time
+
+from fastapi.testclient import TestClient
+
+os.environ.setdefault("AUTH_SESSION_HMAC_SECRET", secrets.token_urlsafe(48))
+
+from main import app
+
+
+EXPECTED_SHA256_ABC = "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
+
+
+def _base64url_encode(raw: bytes) -> str:
+    """Encode one JWT segment without padding."""
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
+
+
+def _signed_session_token() -> str:
+    """Create a real short-lived HMAC session accepted by the private tools API."""
+    now = int(time.time())
+    header_segment = _base64url_encode(
+        json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode(
+            "utf-8"
+        )
+    )
+    payload_segment = _base64url_encode(
+        json.dumps(
+            {
+                "ver": 1,
+                "iss": "naruon-control-plane",
+                "aud": "naruon-api",
+                "sub": "checksum-contract-user",
+                "role": "member",
+                "org": "checksum-contract-org",
+                "groups": [],
+                "workspace": "workspace-checksum-contract-org",
+                "iat": now,
+                "exp": now + 300,
+            },
+            separators=(",", ":"),
+            sort_keys=True,
+        ).encode("utf-8")
+    )
+    signing_input = f"{header_segment}.{payload_segment}"
+    signature = hmac.new(
+        os.environ["AUTH_SESSION_HMAC_SECRET"].encode("utf-8"),
+        signing_input.encode("ascii"),
+        hashlib.sha256,
+    ).digest()
+    return f"{signing_input}.{_base64url_encode(signature)}"
+
+
+def _tampered_session_token() -> str:
+    """Return a structurally valid session token with a deliberately forged signature."""
+    token = _signed_session_token()
+    header_segment, payload_segment, signature_segment = token.split(".")
+    signature_padding = "=" * (-len(signature_segment) % 4)
+    signature = bytearray(
+        base64.urlsafe_b64decode(signature_segment + signature_padding)
+    )
+    signature[0] ^= 0x01
+    return f"{header_segment}.{payload_segment}.{_base64url_encode(bytes(signature))}"
+
+
+def test_content_checksum_api_executes_authenticated_request() -> None:
+    """Startup registration and the authenticated execute route must work together."""
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/tools/content_checksum_generator/execute",
+            headers={"Authorization": f"Bearer {_signed_session_token()}"},
+            json={"parameters": {"text": "abc", "algorithm": "sha256"}},
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "success"
+    assert payload["result"]["digest_hex"] == EXPECTED_SHA256_ABC
+    assert payload["result"]["byte_length"] == 3
+    assert payload["message"] == "Execution successful"
+
+
+def test_content_checksum_api_rejects_unauthenticated_request() -> None:
+    """The checksum execute route must retain the generic tools auth boundary."""
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/tools/content_checksum_generator/execute",
+            json={"parameters": {"text": "abc", "algorithm": "sha256"}},
+        )
+
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Authentication required"}
+
+
+def test_content_checksum_api_rejects_forged_signed_session() -> None:
+    """The checksum route must reject a structurally valid token with a forged signature."""
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/tools/content_checksum_generator/execute",
+            headers={"Authorization": f"Bearer {_tampered_session_token()}"},
+            json={"parameters": {"text": "abc", "algorithm": "sha256"}},
+        )
+
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Authentication required"}
+
+
+def test_content_checksum_api_maps_invalid_algorithm_to_execute_failure() -> None:
+    """Invalid tool input must expose a stable machine-readable failure code."""
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/tools/content_checksum_generator/execute",
+            headers={"Authorization": f"Bearer {_signed_session_token()}"},
+            json={"parameters": {"text": "abc", "algorithm": "md5"}},
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "failed"
+    assert payload["result"] is None
+    assert payload["error_code"] == "unsupported_checksum_algorithm"
+    assert "Unsupported checksum algorithm" in payload["message"]
+
+
+def test_content_checksum_api_maps_invalid_utf8_to_execute_failure() -> None:
+    """Escaped lone surrogates must reach the handler and fail with a stable code."""
+    raw_body = b'{"parameters":{"text":"\\ud800","algorithm":"sha256"}}'
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/tools/content_checksum_generator/execute",
+            headers={
+                "Authorization": f"Bearer {_signed_session_token()}",
+                "Content-Type": "application/json",
+            },
+            content=raw_body,
+        )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["status"] == "failed"
+    assert payload["result"] is None
+    assert payload["error_code"] == "content_checksum_invalid_utf8"