Skip to content
Draft
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
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,4 +1,12 @@
## [Unreleased]
### Fixed
- CI Trivy 파일시스템 보안 스캐너가 감지한 `next` 및 `sharp` 패키지의 심각한 보안 취약점(CVE-2026-75604, GHSA-rgj7-g3m4-5g8c 등)을 해결하기 위해, `frontend/package.json`에서 해당 의존성들을 안전한 최신 버전으로 업데이트했습니다.
### Added
- `backend/api/tools.py`에 다음 3개의 도구를 추가하고 등록했습니다.
- `url_extractor_handler`: 균형 잡힌 URL 괄호와 IPv6 대괄호는 보존하고 문장 끝의 짝 없는 닫힘 기호만 제거한 뒤 중복 URL을 제거합니다.
- `hash_generator_handler`: 선언된 SHA-256 기본값을 parameter validation 전에 적용하고 SHA-256/SHA-384/SHA-512 digest를 생성합니다.
- `json_validator_handler`: RFC 8259가 허용하지 않는 `NaN`, `Infinity`, `-Infinity`를 거부하면서 JSON 문자열을 검증·포맷팅합니다.
- 추가된 도구들에 대해 100% 테스트 커버리지를 보장하는 개별 테스트를 `backend/tests/test_tools_api.py`에 작성했습니다.
- 긴 이메일·첨부 본문을 의미 단위 청크로 임베딩한 뒤 기존 email/attachment 벡터 계약으로 평균화하고, 청크 요청·벡터 누적을 제한된 창으로 처리합니다. OpenAI `text-embedding-3-*`에는 저장 차원(`1536`)을 직접 요청하도록 보강했습니다. 합성 메일 fixture 5건(70청크)과 provider 요청 계약으로 1,536차원 벡터 경로를 검증했으며, 실행 시 선택한 임베딩 제공자에 본문·파싱된 첨부 텍스트를 전송할 수 있습니다. 회사 기밀 데이터는 fixture·commit·PR·log에 포함하지 않습니다.
- EmailDetail 테스트가 지원하지 않는 스레드 병합/분리 버튼을 `textContent`뿐 아니라 `aria-label`과 `title` 접근 가능 이름으로도 검출하도록 바꿔, 아이콘 전용 버튼 회귀를 놓치지 않습니다.

Expand Down
193 changes: 168 additions & 25 deletions backend/api/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,40 +149,50 @@ def get_all(self) -> List[ToolInfo]:
def get(self, code: str) -> Optional[ToolInfo]:
return self._tools.get(code)

async def invoke_tool(self, code: str, params: Dict[str, Any]) -> Any:
handler = self._handlers.get(code)
if not handler:
raise ValueError(f"No handler registered for tool {code}")
result = handler(self._validate_parameters(code, params))
if inspect.isawaitable(result):
return await result
return result

def _validate_parameters(self, code: str, params: Dict[str, Any]) -> Dict[str, Any]:
if not isinstance(params, dict):
async def invoke_tool(self, tool_code: str, tool_parameters: Dict[str, Any]) -> Any:
tool_handler = self._handlers.get(tool_code)
if not tool_handler:
raise ValueError(f"No handler registered for tool {tool_code}")
handler_result = tool_handler(
self._validate_parameters(tool_code, tool_parameters)
)
if inspect.isawaitable(handler_result):
return await handler_result
return handler_result

def _validate_parameters(
self, tool_code: str, tool_parameters: Dict[str, Any]
) -> Dict[str, Any]:
if not isinstance(tool_parameters, dict):
raise ValueError("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:
tool_info = self._tools.get(tool_code)
parameter_schema = tool_info.parameters if tool_info else None
if not parameter_schema:
if tool_parameters:
raise ValueError("Tool does not accept parameters")
return {}

unexpected_keys = set(params) - set(schema)
if unexpected_keys:
unexpected_parameters = set(tool_parameters) - set(parameter_schema)
if unexpected_parameters:
raise ValueError("Unexpected tool parameter")

validated: Dict[str, Any] = {}
for key, descriptor in schema.items():
if key not in params:
validated_parameters: Dict[str, Any] = {}
for parameter_name, parameter_descriptor in parameter_schema.items():
if parameter_name in tool_parameters:
parameter_value = tool_parameters[parameter_name]
elif (
isinstance(parameter_descriptor, dict)
and "default" in parameter_descriptor
):
parameter_value = parameter_descriptor["default"]
else:
raise ValueError("Missing required tool parameter")
value = params[key]
expected_type = _parameter_type_name(descriptor)
if not _parameter_matches_type(value, expected_type):
expected_type = _parameter_type_name(parameter_descriptor)
if not _parameter_matches_type(parameter_value, expected_type):
raise ValueError("Invalid tool parameter type")
validated[key] = value
return validated
validated_parameters[parameter_name] = parameter_value
return validated_parameters


registry = ToolRegistry()
Expand Down Expand Up @@ -706,6 +716,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:
Expand Down Expand Up @@ -769,6 +781,137 @@ async def uuid_v4_generator_handler(params: Dict[str, Any]) -> Dict[str, str]:
)


_URL_PATTERN = re.compile(
r"https?://[a-zA-Z0-9\-._~:/?#\[\]@!$&'()*+,;=%]+(?<![.,!?])",
re.IGNORECASE,
)


def _strip_unmatched_url_closers(extracted_url: str) -> str:
"""Remove terminal unmatched brackets while preserving balanced URL syntax."""
delimiter_pairs = {")": "(", "]": "["}
while extracted_url and extracted_url[-1] in delimiter_pairs:
unmatched_terminal = False
opening_delimiters: list[str] = []
for character_index, url_character in enumerate(extracted_url):
if url_character in delimiter_pairs.values():
opening_delimiters.append(url_character)
elif url_character in delimiter_pairs:
expected_opener = delimiter_pairs[url_character]
if opening_delimiters and opening_delimiters[-1] == expected_opener:
opening_delimiters.pop()
elif character_index == len(extracted_url) - 1:
unmatched_terminal = True
if not unmatched_terminal:
break
extracted_url = extracted_url[:-1]
return extracted_url


async def url_extractor_handler(tool_parameters: Dict[str, Any]) -> Dict[str, Any]:
"""Extract distinct HTTP(S) URLs from workspace text in encounter order."""
source_text = tool_parameters.get("text", "")
extracted_urls = _URL_PATTERN.findall(source_text)

seen_urls = set()
deduplicated_urls = []
for extracted_url in extracted_urls:
normalized_url = _strip_unmatched_url_closers(extracted_url)
if normalized_url not in seen_urls:
seen_urls.add(normalized_url)
deduplicated_urls.append(normalized_url)

return {"urls": deduplicated_urls, "url_count": len(deduplicated_urls)}


registry.register(
ToolInfo(
code="url_extractor",
name="URL 추출기 (URL Extractor)",
description="텍스트 본문에서 URL을 찾아 추출합니다.",
category="이메일 분석",
parameters={"text": "string"},
),
url_extractor_handler,
)


async def hash_generator_handler(tool_parameters: Dict[str, Any]) -> Dict[str, str]:
"""Generate a supported SHA digest for workspace text."""
source_text = tool_parameters.get("text", "")
digest_algorithm = tool_parameters.get("algorithm", "sha256")
if not digest_algorithm:
digest_algorithm = "sha256"
digest_algorithm = digest_algorithm.lower()

encoded_text = source_text.encode("utf-8")
if digest_algorithm == "sha256":
hash_digest = hashlib.sha256(encoded_text).hexdigest()
elif digest_algorithm == "sha384":
hash_digest = hashlib.sha384(encoded_text).hexdigest()
elif digest_algorithm == "sha512":
hash_digest = hashlib.sha512(encoded_text).hexdigest()
else:
raise ValueError("Unsupported algorithm. Supported: sha256, sha384, sha512")

return {"hash": hash_digest, "algorithm": digest_algorithm}


registry.register(
ToolInfo(
code="hash_generator",
name="해시 생성기 (Hash Generator)",
description="입력된 텍스트를 지정된 해시 알고리즘(SHA-256, SHA-384, SHA-512)으로 변환합니다.",
category="유틸리티",
parameters={
"text": "string",
"algorithm": {"type": "string", "default": "sha256"},
},
),
hash_generator_handler,
)


def _reject_non_standard_json_constant(constant_name: str) -> None:
"""Reject JavaScript numeric constants that RFC 8259 excludes from JSON."""
raise ValueError(f"Non-standard JSON constant: {constant_name}")


async def json_validator_handler(tool_parameters: Dict[str, Any]) -> Dict[str, Any]:
"""Validate and format an RFC 8259 JSON document."""
json_document = tool_parameters.get("json_string", "")
try:
parsed_json = json.loads(
json_document,
parse_constant=_reject_non_standard_json_constant,
)
formatted_json = json.dumps(parsed_json, indent=2, ensure_ascii=False)
return {
"is_valid": True,
"parsed": parsed_json,
"formatted_json": formatted_json,
"error_message": None,
}
except (json.JSONDecodeError, ValueError) as validation_error:
return {
"is_valid": False,
"parsed": None,
"formatted_json": None,
"error_message": str(validation_error),
}


registry.register(
ToolInfo(
code="json_validator",
name="JSON 검증기 (JSON Validator)",
description="주어진 JSON 문자열의 유효성을 검사하고, 유효한 경우 포맷팅된 결과를 반환합니다.",
category="유틸리티",
parameters={"json_string": "string"},
),
json_validator_handler,
)


@router.get("/tools", response_model=list[ToolInfo])
def get_tools() -> list[ToolInfo]:
Expand Down
Loading
Loading