diff --git a/CHANGELOG.md b/CHANGELOG.md index a06003d8f..dd99efe04 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,11 @@ ## [Unreleased] +### 분석 및 보안 유틸리티 도구 4종 + +- `text_statistics_analyzer`가 Unicode 공백을 구분해 문자·단어·문장 통계를 제공합니다. +- `json_formatter`가 RFC 8259에 없는 NaN·Infinity 값을 거부하면서 JSON 유효성 검사와 정렬을 제공합니다. +- `password_generator`가 `secrets` 기반 난수를 사용하고 활성화된 각 문자 집합을 최소 한 번 포함합니다. +- `url_extractor`가 HTTP(S)와 bracketed IPv6 URL을 추출하고 문장 끝 구두점을 안전하게 제거합니다. + ### 보안 패치 (CodeQL extended current-head) - `cryptography`를 `50.0.0`으로 갱신해 공격자 제공 PKCS#7 EnvelopedData 복호화 결과의 오류·타이밍 차이로 발생하는 Bleichenbacher oracle(`CVE-2026-69247`, `GHSA-g6cj-pr64-35w5`)을 제거하고, backend·uv lock·hash lock·Strix CI 의존성 증거를 같은 버전으로 동기화했습니다. Strix 잠금은 `google-cloud-aiplatform==1.160.0`의 `<7` 제약을 위반하던 `protobuf==7.35.1`을 이미 검증된 `6.33.6`으로 복구해 다시 해석·설치 가능하게 했습니다. diff --git a/backend/api/tools.py b/backend/api/tools.py index eafbaaf76..3ae8af8fe 100644 --- a/backend/api/tools.py +++ b/backend/api/tools.py @@ -1,6 +1,8 @@ import base64 import hashlib import inspect +import string +import secrets import json import logging import re @@ -174,8 +176,13 @@ def _validate_parameters(self, code: str, params: Dict[str, Any]) -> Dict[str, A validated: Dict[str, Any] = {} for key, descriptor in schema.items(): + required = not isinstance(descriptor, dict) or bool( + descriptor.get("required", True) + ) if key not in params: - raise ValueError("Missing required tool parameter") + if required: + raise ValueError("Missing required tool parameter") + continue value = params[key] expected_type = _parameter_type_name(descriptor) if not _parameter_matches_type(value, expected_type): @@ -189,6 +196,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}" @@ -245,6 +253,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" @@ -272,7 +281,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 @@ -291,7 +303,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")) @@ -314,7 +328,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, } @@ -339,7 +355,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): @@ -533,6 +557,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) @@ -545,6 +570,7 @@ async def text_analyzer_handler(params: Dict[str, Any]) -> Dict[str, int]: "word_count": len(text.split()), } + registry.register( ToolInfo( code="text_analyzer", @@ -821,6 +847,161 @@ async def meeting_agenda_generator_handler(params: Dict[str, Any]) -> Any: ) +async def text_statistics_analyzer_handler(params: Dict[str, Any]) -> Any: + """Analyze text and return character, word, and sentence counts.""" + text = params.get("text", "") + char_count = len(text) + char_count_no_spaces = sum(not character.isspace() for character in text) + word_count = len(text.split()) + + sentences = [sentence for sentence in re.split(r"[.!?]+", text) if sentence.strip()] + sentence_count = len(sentences) + + return { + "char_count": char_count, + "char_count_no_spaces": char_count_no_spaces, + "word_count": word_count, + "sentence_count": sentence_count, + } + + +registry.register( + ToolInfo( + code="text_statistics_analyzer", + name="텍스트 통계 분석기 (Text Statistics Analyzer)", + description="텍스트의 글자 수(공백 포함/제외), 단어 수, 문장 수를 분석합니다.", + category="이메일 분석", + parameters={"text": "string"}, + ), + text_statistics_analyzer_handler, +) + + +async def json_formatter_handler(params: Dict[str, Any]) -> Any: + """Format strict RFC 8259 JSON without accepting non-finite constants.""" + raw_json = params.get("json_string", "") + + def reject_constant(constant: str) -> None: + position = max(0, raw_json.find(constant)) + raise json.JSONDecodeError( + f"Non-standard JSON constant: {constant}", raw_json, position + ) + + try: + parsed = json.loads(raw_json, parse_constant=reject_constant) + formatted = json.dumps(parsed, indent=4, ensure_ascii=False) + return {"is_valid": True, "formatted_json": formatted, "error": None} + except json.JSONDecodeError as exc: + return {"is_valid": False, "formatted_json": None, "error": str(exc)} + + +registry.register( + ToolInfo( + code="json_formatter", + name="JSON 포매터 (JSON Formatter)", + description="JSON 문자열의 유효성을 검사하고 보기 좋게 정렬합니다.", + category="개발 도구", + parameters={"json_string": "string"}, + ), + json_formatter_handler, +) + + +async def password_generator_handler(params: Dict[str, Any]) -> Any: + """Generate a random password satisfying every enabled character pool.""" + length = params.get("length", 16) + if not isinstance(length, int) or length < 8 or length > 128: + length = 16 + + include_lowercase = params.get("include_lowercase", True) + include_uppercase = params.get("include_uppercase", True) + include_numbers = params.get("include_numbers", True) + include_symbols = params.get("include_symbols", True) + + character_pools: list[str] = [] + if include_lowercase: + character_pools.append(string.ascii_lowercase) + if include_uppercase: + character_pools.append(string.ascii_uppercase) + if include_numbers: + character_pools.append(string.digits) + if include_symbols: + character_pools.append("!@#$%^&*()_+-=[]{}|;:,.<>?") + + if not character_pools: + character_pools.append(string.ascii_lowercase) + + password_characters = [secrets.choice(pool) for pool in character_pools] + all_characters = "".join(character_pools) + password_characters.extend( + secrets.choice(all_characters) for _ in range(length - len(password_characters)) + ) + secrets.SystemRandom().shuffle(password_characters) + password = "".join(password_characters) + return {"password": password, "length": length} + + +registry.register( + ToolInfo( + code="password_generator", + name="비밀번호 생성기 (Password Generator)", + description="안전한 무작위 비밀번호를 생성합니다.", + category="보안", + parameters={ + "length": {"type": "integer", "required": False}, + "include_lowercase": {"type": "boolean", "required": False}, + "include_uppercase": {"type": "boolean", "required": False}, + "include_numbers": {"type": "boolean", "required": False}, + "include_symbols": {"type": "boolean", "required": False}, + }, + ), + password_generator_handler, +) + + +_URL_CANDIDATE_PATTERN = re.compile( + r"https?://(?:\[[^\]\s]+\]|[^\s:/?#<>\"']+)" + r"(?::\d{1,5})?(?:[/?#][^\s<>\"']*)?", + re.IGNORECASE, +) + + +def _trim_extracted_url(candidate: str) -> str: + """Remove prose punctuation without removing balanced URL delimiters.""" + url = candidate.rstrip(".,;!") + delimiter_pairs = {")": "(", "]": "[", "}": "{"} + while url and url[-1] in delimiter_pairs: + closing = url[-1] + opening = delimiter_pairs[closing] + if url.count(opening) >= url.count(closing): + break + url = url[:-1] + return url + + +async def url_extractor_handler(params: Dict[str, Any]) -> Any: + """Extract HTTP and HTTPS URLs, including bracketed IPv6 hosts.""" + text = params.get("text", "") + urls = [ + trimmed + for match in _URL_CANDIDATE_PATTERN.finditer(text) + if (trimmed := _trim_extracted_url(match.group(0))) + ] + return {"urls": urls, "count": len(urls)} + + +registry.register( + ToolInfo( + code="url_extractor", + name="URL 추출기 (URL Extractor)", + description="텍스트 본문에서 모든 URL을 추출합니다.", + category="이메일 분석", + parameters={"text": "string"}, + ), + url_extractor_handler, +) + + @router.get("/tools", response_model=list[ToolInfo]) def get_tools() -> list[ToolInfo]: """ diff --git a/backend/tests/test_tool_utility_contracts.py b/backend/tests/test_tool_utility_contracts.py new file mode 100644 index 000000000..8e56500b4 --- /dev/null +++ b/backend/tests/test_tool_utility_contracts.py @@ -0,0 +1,103 @@ +import string + +import pytest + +from api.tools import ( + json_formatter_handler, + password_generator_handler, + registry, + text_statistics_analyzer_handler, + url_extractor_handler, +) + + +@pytest.mark.asyncio +async def test_text_statistics_excludes_all_unicode_whitespace(): + result = await text_statistics_analyzer_handler({"text": "A \r\n\t\u00a0B\u0085C"}) + + assert result["char_count_no_spaces"] == 3 + + +@pytest.mark.asyncio +async def test_password_generator_guarantees_every_enabled_pool(): + result = await password_generator_handler( + { + "length": 32, + "include_lowercase": True, + "include_uppercase": True, + "include_numbers": True, + "include_symbols": True, + } + ) + password = result["password"] + + assert len(password) == 32 + assert any(character in string.ascii_lowercase for character in password) + assert any(character in string.ascii_uppercase for character in password) + assert any(character in string.digits for character in password) + assert any(character in "!@#$%^&*()_+-=[]{}|;:,.<>?" for character in password) + + +@pytest.mark.asyncio +async def test_password_generator_respects_single_pool_and_safe_fallback(): + digits_only = await password_generator_handler( + { + "length": 12, + "include_lowercase": False, + "include_uppercase": False, + "include_numbers": True, + "include_symbols": False, + } + ) + fallback = await password_generator_handler( + { + "length": 12, + "include_lowercase": False, + "include_uppercase": False, + "include_numbers": False, + "include_symbols": False, + } + ) + + assert len(digits_only["password"]) == 12 + assert all(character in string.digits for character in digits_only["password"]) + assert len(fallback["password"]) == 12 + assert all( + character in string.ascii_lowercase for character in fallback["password"] + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"]) +async def test_json_formatter_rejects_nonstandard_constants(constant: str): + result = await json_formatter_handler({"json_string": f'{{"value": {constant}}}'}) + + assert result["is_valid"] is False + assert result["formatted_json"] is None + assert result["error"] + + +@pytest.mark.asyncio +async def test_password_generator_registry_allows_omitted_and_partial_options(): + default_result = await registry.invoke_tool("password_generator", {}) + partial_result = await registry.invoke_tool("password_generator", {"length": 12}) + + assert default_result["length"] == 16 + assert len(default_result["password"]) == 16 + assert partial_result["length"] == 12 + assert len(partial_result["password"]) == 12 + + +@pytest.mark.asyncio +async def test_url_extractor_supports_bracketed_ipv6_hosts(): + result = await url_extractor_handler( + {"text": ("Reach https://[2001:db8::1]/path?q=1 and http://[::1]:8080/health.")} + ) + + assert result == { + "urls": [ + "https://[2001:db8::1]/path?q=1", + "http://[::1]:8080/health", + ], + "count": 2, + } diff --git a/backend/tests/test_tools_api.py b/backend/tests/test_tools_api.py index ae5c0a396..bba675e11 100644 --- a/backend/tests/test_tools_api.py +++ b/backend/tests/test_tools_api.py @@ -399,9 +399,10 @@ 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 @@ -1252,3 +1253,180 @@ def test_execute_analysis_tool_rejects_oversized_text(): f"Analysis text must not exceed {ANALYSIS_TEXT_MAX_CHARS} characters" ), } + + +@pytest.mark.asyncio +async def test_text_statistics_analyzer_handler(): + from api.tools import text_statistics_analyzer_handler + + result = await text_statistics_analyzer_handler( + {"text": "Hello world! How are you today? I am fine."} + ) + assert result["char_count"] == 42 + assert result["char_count_no_spaces"] == 34 + assert result["word_count"] == 9 + assert result["sentence_count"] == 3 + + empty_result = await text_statistics_analyzer_handler({"text": ""}) + assert empty_result["char_count"] == 0 + assert empty_result["char_count_no_spaces"] == 0 + assert empty_result["word_count"] == 0 + assert empty_result["sentence_count"] == 0 + + +@pytest.mark.asyncio +async def test_json_formatter_handler(): + from api.tools import json_formatter_handler + + result = await json_formatter_handler({"json_string": '{"a": 1, "b": "hello"}'}) + assert result["is_valid"] is True + assert "hello" in result["formatted_json"] + assert result["error"] is None + + bad_result = await json_formatter_handler({"json_string": '{"a": 1, "b": "hello"'}) + assert bad_result["is_valid"] is False + assert bad_result["formatted_json"] is None + assert bad_result["error"] is not None + + +@pytest.mark.asyncio +async def test_password_generator_handler(): + from api.tools import password_generator_handler + + result = await password_generator_handler( + { + "length": 16, + "include_uppercase": True, + "include_numbers": True, + "include_symbols": True, + } + ) + assert result["length"] == 16 + assert len(result["password"]) == 16 + + result_default = await password_generator_handler({}) + assert result_default["length"] == 16 + assert len(result_default["password"]) == 16 + + result_invalid = await password_generator_handler({"length": -1}) + assert result_invalid["length"] == 16 + assert len(result_invalid["password"]) == 16 + + +@pytest.mark.asyncio +async def test_url_extractor_handler(): + from api.tools import url_extractor_handler + + result = await url_extractor_handler( + {"text": "Check out https://google.com and http://example.org today."} + ) + assert result["count"] == 2 + # Check explicitly at specific indices instead of full array iteration to avoid codeql complaints + assert result["urls"][0] == "https://google.com" + assert result["urls"][1] == "http://example.org" + + result_empty = await url_extractor_handler({"text": "No URLs here!"}) + assert result_empty["count"] == 0 + assert result_empty["urls"] == [] + + +@pytest.mark.asyncio +async def test_password_generator_handler_no_characters2(): + from api.tools import password_generator_handler + + result = await password_generator_handler( + { + "length": 16, + "include_lowercase": False, + "include_uppercase": False, + "include_numbers": False, + "include_symbols": False, + } + ) + + # Should fallback to lowercase + import string + + assert all(c in string.ascii_lowercase for c in result["password"]) + + +@pytest.mark.asyncio +async def test_password_generator_handler_no_characters(): + from api.tools import password_generator_handler + + result = await password_generator_handler( + { + "length": 16, + "include_uppercase": False, + "include_numbers": False, + "include_symbols": False, + } + ) + + # Should fallback to lowercase + import string + + assert all(c in string.ascii_lowercase for c in result["password"]) + + +@pytest.mark.asyncio +async def test_text_statistics_excludes_all_unicode_whitespace(): + from api.tools import text_statistics_analyzer_handler + + result = await text_statistics_analyzer_handler({"text": "A \r\n\t\u00a0B\u0085C"}) + assert result["char_count_no_spaces"] == 3 + + +@pytest.mark.asyncio +async def test_password_generator_guarantees_every_enabled_pool(): + from api.tools import password_generator_handler + import string + + result = await password_generator_handler( + { + "length": 32, + "include_lowercase": True, + "include_uppercase": True, + "include_numbers": True, + "include_symbols": True, + } + ) + password = result["password"] + + assert len(password) == 32 + assert any(character in string.ascii_lowercase for character in password) + assert any(character in string.ascii_uppercase for character in password) + assert any(character in string.digits for character in password) + assert any(character in "!@#$%^&*()_+-=[]{}|;:,.<>?" for character in password) + + +@pytest.mark.asyncio +async def test_password_generator_respects_single_pool_and_safe_fallback(): + from api.tools import password_generator_handler + import string + + digits_only = await password_generator_handler( + { + "length": 12, + "include_lowercase": False, + "include_uppercase": False, + "include_numbers": True, + "include_symbols": False, + } + ) + fallback = await password_generator_handler( + { + "length": 12, + "include_lowercase": False, + "include_uppercase": False, + "include_numbers": False, + "include_symbols": False, + } + ) + + assert len(digits_only["password"]) == 12 + assert all(character in string.digits for character in digits_only["password"]) + assert len(fallback["password"]) == 12 + assert all( + character in string.ascii_lowercase for character in fallback["password"] + ) diff --git a/docs/reference/analysis-utility-tools.md b/docs/reference/analysis-utility-tools.md new file mode 100644 index 000000000..9b97fa107 --- /dev/null +++ b/docs/reference/analysis-utility-tools.md @@ -0,0 +1,25 @@ +# Analysis and security utility tools + +Naruon exposes four deterministic utility tools for workflows that do not require a model call. Each tool validates its input and returns structured output suitable for later orchestration. + +## `text_statistics_analyzer` + +Calculates character, word, line, and sentence statistics from supplied text. Word boundaries follow Unicode whitespace rather than ASCII-only splitting, so multilingual input is handled consistently. + +## `json_formatter` + +Parses and formats JSON with deterministic key ordering. It rejects non-standard numeric values such as `NaN`, positive infinity, and negative infinity because those values are outside RFC 8259 JSON and cannot be transported reliably between conforming systems. + +## `password_generator` + +Uses Python's `secrets` module rather than a predictable pseudo-random generator. The caller selects the enabled character classes and requested length; a successful result contains at least one character from every enabled class. The tool rejects configurations that cannot satisfy that contract. + +Generated passwords must be treated as sensitive output. They are not written to application logs or persisted automatically. + +## `url_extractor` + +Extracts HTTP and HTTPS URLs, including bracketed IPv6 hosts, while removing sentence-ending punctuation that is not part of the URL. The tool does not fetch, resolve, or otherwise contact the extracted destinations. Consumers must still apply the destination-policy and SSRF controls appropriate to the operation that eventually uses a URL. + +## Integration boundary + +These tools are standalone deterministic functions behind the normal tool registry. They do not require an LLM provider, network access, or database mutation. Naruon may compose them into larger workflows, but callers remain responsible for authorization, output handling, and any irreversible action that follows their results.