diff --git a/CHANGELOG.md b/CHANGELOG.md index 997773327..b4e7bf168 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,5 @@ ## [Unreleased] +- 텍스트 본문에서 HTTP 및 HTTPS URL을 추출하여 중복 없이 반환하는 유틸리티 도구인 `url_extractor` (URL 추출기)를 추가했습니다. - 분석·유틸리티 도구 2종(`hash_generator`, `email_phone_masker`)을 추가했습니다. 해시 도구는 MD5·SHA-1 호환 fingerprint와 SHA-256을 구분하고, 연락처 도구는 제한된 길이 안에서 이메일 주소와 전화번호를 단순 마스킹합니다. ### Source-bound 요약·업무·관계·일정 경계 diff --git a/backend/api/tools.py b/backend/api/tools.py index b1f2e6070..b13e8dd11 100644 --- a/backend/api/tools.py +++ b/backend/api/tools.py @@ -655,6 +655,69 @@ async def uuid_v4_generator_handler(params: Dict[str, Any]) -> Dict[str, str]: ) +_URL_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE) +_PROSE_TRAILING_PUNCTUATION = ".,;:!?" + + +def _trim_url_candidate(candidate: str, wrapping_openers: str) -> str: + """Remove only closing delimiters proven by adjacent opening wrappers.""" + delimiters = (("(", ")"), ("[", "]"), ("{", "}")) + excess = { + closer: min( + wrapping_openers.count(opener), + max(0, candidate.count(closer) - candidate.count(opener)), + ) + for opener, closer in delimiters + } + without_prose = candidate.rstrip(_PROSE_TRAILING_PUNCTUATION) + end = len(without_prose) + while end and excess.get(without_prose[end - 1], 0): + excess[without_prose[end - 1]] -= 1 + end -= 1 + return without_prose[:end] if end < len(without_prose) else candidate + +async def url_extractor_handler(params: Dict[str, Any]) -> Dict[str, list[str]]: + text = params["text"] + if len(text) > ANALYSIS_TEXT_MAX_CHARS: + raise ValueError( + f"Analysis text must not exceed {ANALYSIS_TEXT_MAX_CHARS} characters" + ) + urls: list[str] = [] + seen: set[str] = set() + for match in _URL_PATTERN.finditer(text): + wrapper_start = match.start() + while wrapper_start and text[wrapper_start - 1].isspace(): + wrapper_start -= 1 + wrapper_end = wrapper_start + while wrapper_start and text[wrapper_start - 1] in "([{": + wrapper_start -= 1 + candidate = _trim_url_candidate( + match.group(), text[wrapper_start:wrapper_end] + ) + try: + parsed = urllib.parse.urlsplit(candidate) + _ = parsed.port # validate a declared port without requiring one + valid = parsed.hostname is not None + except ValueError: + valid = False + if valid and candidate not in seen: + seen.add(candidate) + urls.append(candidate) + return {"urls": urls} + + +registry.register( + ToolInfo( + code="url_extractor", + name="URL 추출기 (URL Extractor)", + description="텍스트 본문에서 HTTP 및 HTTPS 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_tools_api.py b/backend/tests/test_tools_api.py index e85e8020b..3e1ef1a78 100644 --- a/backend/tests/test_tools_api.py +++ b/backend/tests/test_tools_api.py @@ -15,6 +15,7 @@ os.environ.setdefault("AUTH_SESSION_HMAC_SECRET", secrets.token_urlsafe(48)) from api.tools import ( + ANALYSIS_TEXT_MAX_CHARS, MAX_TOOL_FAILURE_MESSAGE_CHARS, ExecuteRequest, ToolInfo, @@ -417,6 +418,167 @@ async def error_handler(params): assert "Simulated error" in data["message"] +def test_execute_url_extractor(): + text = ( + "URLs: https://example.com:8443/foo/bar?q=a%20b#section " + "(http://foo.com?next=/a) https://example.com:8443/foo/bar?q=a%20b#section." + ) + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": text}}, + ) + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + assert data["result"]["urls"] == [ + "https://example.com:8443/foo/bar?q=a%20b#section", + "http://foo.com?next=/a", + "https://example.com:8443/foo/bar?q=a%20b#section.", + ] + + +@pytest.mark.parametrize( + "url", + ( + "https://example.com/a,b,", + "https://example.com?q=what?", + "https://example.com#frag!", + ), +) +def test_execute_url_extractor_preserves_valid_terminal_punctuation(url): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": url}}, + ) + + assert response.json()["result"]["urls"] == [url] + + +@pytest.mark.parametrize("suffix", (").", "),", ")!")) +def test_execute_url_extractor_removes_wrapped_prose_delimiters(suffix): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": f"(https://example.com/path{suffix}"}}, + ) + + assert response.json()["result"]["urls"] == ["https://example.com/path"] + + +@pytest.mark.parametrize( + ("text", "url"), + ( + ("(https://example.com/path!),", "https://example.com/path!"), + ("([https://example.com/path]).", "https://example.com/path"), + ), +) +def test_execute_url_extractor_handles_punctuated_nested_wrappers(text, url): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": text}}, + ) + + assert response.json()["result"]["urls"] == [url] + + +def test_execute_url_extractor_handles_spaced_nested_wrappers(): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": "([ https://example.com/path])."}}, + ) + + assert response.json()["result"]["urls"] == ["https://example.com/path"] + + +def test_execute_url_extractor_does_not_borrow_distant_wrapper(): + url = "https://example.com/path)." + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": f"(see {url}"}}, + ) + + assert response.json()["result"]["urls"] == [url] + + +@pytest.mark.parametrize( + "url", + ( + "https://example.com/a(b)", + "https://example.com/a[b]", + "https://example.com/a{b}", + ), +) +def test_execute_url_extractor_preserves_balanced_url_delimiters(url): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": f"(\n{url}"}}, + ) + + assert response.json()["result"]["urls"] == [url] + + +@pytest.mark.parametrize( + "url", + ( + "https://example.com/a).", + "https://example.com/a],", + "https://example.com/a}!", + ), +) +def test_execute_url_extractor_preserves_unwrapped_delimiter_suffixes(url): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": url}}, + ) + + assert response.json()["result"]["urls"] == [url] + + +def test_execute_url_extractor_handles_many_unmatched_delimiters_linearly(): + wrapper_count = 25_000 + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={ + "parameters": { + "text": "(" * wrapper_count + + "https://example.com/" + + ")" * wrapper_count + } + }, + ) + + assert response.json()["result"]["urls"] == ["https://example.com/"] + + +def test_execute_url_extractor_rejects_oversized_text(): + with TestClient(app) as client: + response = client.post( + "/api/tools/url_extractor/execute", + headers={"Authorization": f"Bearer {_signed_session_token()}"}, + json={"parameters": {"text": "x" * (ANALYSIS_TEXT_MAX_CHARS + 1)}}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "failed" + + @pytest.mark.asyncio async def test_execute_tool_failure_log_does_not_include_user_controlled_lines(caplog): hostile_code = "error_tool\r\nforged_event=true"