Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
## [Unreleased]
### 새로운 유틸리티 도구 추가 (UUID/해시 생성기)

- `backend/api/tools.py`에 `uuid_generator`와 `hash_generator` 두 가지 유틸리티 도구를 추가했습니다. UUID(버전 1, 4) 생성 및 해시(MD5, SHA1, SHA256, SHA512) 생성을 지원하며, 관련된 테스트 코드를 `backend/tests/test_tools_api.py`에 추가하여 100% 커버리지를 달성했습니다.

### 보안 패치 (CodeQL extended current-head)

- CodeQL `extended` 기본 설정이 current `develop`에서 확인한 Critical 8건·High 21건·Medium 1건을 코드 경계에서 제거합니다. 서버 요청은 검증된 loopback/HTTPS origin, 동일 OIDC issuer origin, 허용 API 경로·쿼리만 재구성하고 redirect를 자동 추종하지 않으며, 공개 IPv6 authority를 보존합니다. UI smoke는 고정 Node/Next 실행 파일과 인자, localhost:3001 allowlist, private `mkdtemp` artifact 디렉터리 및 containment 검사만 사용합니다.
Expand Down
153 changes: 142 additions & 11 deletions backend/api/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,10 @@
import json
import logging
import re
import secrets
import unicodedata
import urllib.parse
import uuid
from collections import Counter
from collections.abc import Callable
from typing import Any, Dict, List, Optional
Expand All @@ -27,6 +29,22 @@
MAX_TOOL_FAILURE_MESSAGE_CHARS = 500


class ToolOptionError(ValueError):
"""A tool option failure with a stable machine-readable code."""

def __init__(self, error_code: str, message: str):
super().__init__(message)
self.error_code = error_code


class ToolValidationError(ValueError):
"""A tool request validation failure with a stable machine-readable code."""

def __init__(self, error_code: str, message: str):
super().__init__(message)
self.error_code = error_code


def _tool_code_fingerprint(code: str) -> str:
"""Return a stable non-reversible identifier for correlating tool failures."""
return hashlib.sha256(code.encode("utf-8", errors="replace")).hexdigest()[:12]
Expand Down Expand Up @@ -125,8 +143,11 @@ class ExecuteRequest(BaseModel):

class ExecuteResponse(BaseModel):
status: str = Field(..., description="실행 상태 (예: success, failed)")
result: Any = Field(..., description="실행 결과 데이터")
result: Any = Field(default=None, description="실행 결과 데이터")
message: Optional[str] = Field(default=None, description="결과 메시지")
error_code: Optional[str] = Field(
default=None, description="실패 유형을 나타내는 안정적인 오류 코드"
)


class ToolRegistry:
Expand Down Expand Up @@ -159,27 +180,42 @@ async def invoke_tool(self, code: str, params: Dict[str, Any]) -> Any:

def _validate_parameters(self, code: str, params: Dict[str, Any]) -> Dict[str, Any]:
if not isinstance(params, dict):
raise ValueError("Tool parameters must be an object")
raise ToolValidationError(
"invalid_tool_parameters",
"Tool parameters must be an object",
)

tool_info = self._tools.get(code)
schema = tool_info.parameters if tool_info else None
if not schema:
if params:
raise ValueError("Tool does not accept parameters")
raise ToolValidationError(
"tool_parameters_not_supported",
"Tool does not accept parameters",
)
return {}

unexpected_keys = set(params) - set(schema)
if unexpected_keys:
raise ValueError("Unexpected tool parameter")
raise ToolValidationError(
"unexpected_tool_parameter",
"Unexpected tool parameter",
)

validated: Dict[str, Any] = {}
for key, descriptor in schema.items():
if key not in params:
raise ValueError("Missing required tool parameter")
raise ToolValidationError(
"missing_tool_parameter",
"Missing required tool parameter",
)
value = params[key]
expected_type = _parameter_type_name(descriptor)
if not _parameter_matches_type(value, expected_type):
raise ValueError("Invalid tool parameter type")
raise ToolValidationError(
"invalid_tool_parameter_type",
"Invalid tool parameter type",
)
validated[key] = value
return validated

Expand All @@ -189,6 +225,7 @@ 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}"
Expand Down Expand Up @@ -245,6 +282,7 @@ 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"
Expand Down Expand Up @@ -272,7 +310,10 @@ 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
Expand All @@ -291,7 +332,9 @@ 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"))
Expand All @@ -314,7 +357,9 @@ 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,
}
Expand All @@ -339,7 +384,15 @@ 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):
Expand Down Expand Up @@ -533,6 +586,7 @@ 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)
Expand All @@ -545,6 +599,7 @@ async def text_analyzer_handler(params: Dict[str, Any]) -> Dict[str, int]:
"word_count": len(text.split()),
}


registry.register(
ToolInfo(
code="text_analyzer",
Expand Down Expand Up @@ -821,6 +876,77 @@ async def meeting_agenda_generator_handler(params: Dict[str, Any]) -> Any:
)


async def uuid_generator_handler(params: Dict[str, Any]) -> Any:
"""
Generates a UUID based on the specified version.
Supports UUIDv4 (random) and UUIDv1 (timestamp-based).
For UUIDv1, the node (MAC address) is randomized to ensure privacy.
"""
version = params.get("version", 4)
if version == 1:
random_multicast_node = secrets.randbits(48) | (1 << 40)
return {
"uuid": str(uuid.uuid1(node=random_multicast_node)) # nosemgrep
}
if version == 4:
return {"uuid": str(uuid.uuid4())}
raise ToolOptionError(
"unsupported_uuid_version",
f"Unsupported UUID version: {version}",
)


async def hash_generator_handler(params: Dict[str, Any]) -> Any:
"""
Generates a hash for the provided text using the specified algorithm.
Supported algorithms: MD5, SHA1, SHA256, SHA512.
Note: MD5 and SHA1 are included for interoperability purposes only and should not be used for security.
"""
text = params.get("text", "")
algorithm = params.get("algorithm", "sha256").lower()

if algorithm == "sha256":
hash_obj = hashlib.sha256(text.encode("utf-8"))
elif algorithm == "md5":
hash_obj = hashlib.md5(text.encode("utf-8"), usedforsecurity=False)
elif algorithm == "sha1":
# fmt: off
hash_obj = hashlib.sha1(text.encode("utf-8"), usedforsecurity=False) # nosemgrep: python.lang.security.insecure-hash-algorithms.insecure-hash-algorithm-sha1 -- interoperability-only digest
# fmt: on
elif algorithm == "sha512":
hash_obj = hashlib.sha512(text.encode("utf-8"))
else:
raise ToolOptionError(
"unsupported_hash_algorithm",
f"Unsupported hash algorithm: {algorithm}",
)

return {"hash": hash_obj.hexdigest()}


registry.register(
ToolInfo(
code="uuid_generator",
name="UUID 생성기",
description="지정된 버전(1 또는 4)의 UUID를 생성합니다.",
category="유틸리티",
parameters={"version": "integer"},
),
uuid_generator_handler,
)

registry.register(
ToolInfo(
code="hash_generator",
name="해시 생성기",
description="입력된 텍스트에 대해 지정된 알고리즘(MD5, SHA1, SHA256, SHA512)으로 해시 값을 생성합니다.",
category="유틸리티",
parameters={"text": "string", "algorithm": "string"},
),
hash_generator_handler,
)


@router.get("/tools", response_model=list[ToolInfo])
def get_tools() -> list[ToolInfo]:
"""
Expand Down Expand Up @@ -911,7 +1037,11 @@ def delete_tool(code: str) -> None:
registry.unregister(code)


@router.post("/tools/{code}/execute", response_model=ExecuteResponse)
@router.post(
"/tools/{code}/execute",
response_model=ExecuteResponse,
response_model_exclude_none=True,
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
async def execute_tool(code: str, request: ExecuteRequest) -> ExecuteResponse:
"""
특정 도구를 실행합니다.
Expand Down Expand Up @@ -940,4 +1070,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),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
)
Loading
Loading