From 35675424d9b0d1e5e72c0206670ecf798bd65ee7 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 11 Sep 2026 09:04:22 +0900 Subject: [PATCH 1/4] feat(tools): add bounded URL evidence and contact redaction components Implements the pure detector layer of the data-hygiene utility suite behind issue #1247: non-fetching HTTP(S) URL evidence with normalization, Unicode source offsets, duplicate-location preservation and fail-closed userinfo/malformed-authority warnings, plus an explicit, versioned email/telephone redactor that never retains detected values and leaves unsupported PII untouched. Registry and API wiring stay with the tools owner. --- backend/services/contact_data_redactor.py | 130 ++++++++++++++ backend/services/url_evidence.py | 185 ++++++++++++++++++++ backend/tests/test_contact_data_redactor.py | 52 ++++++ backend/tests/test_url_evidence.py | 51 ++++++ 4 files changed, 418 insertions(+) create mode 100644 backend/services/contact_data_redactor.py create mode 100644 backend/services/url_evidence.py create mode 100644 backend/tests/test_contact_data_redactor.py create mode 100644 backend/tests/test_url_evidence.py diff --git a/backend/services/contact_data_redactor.py b/backend/services/contact_data_redactor.py new file mode 100644 index 000000000..d40b7fd38 --- /dev/null +++ b/backend/services/contact_data_redactor.py @@ -0,0 +1,130 @@ +"""Bounded redaction for the explicitly supported contact-data classes. + +This module is not a general PII anonymizer. It detects email addresses and +telephone numbers only; unsupported personal-data classes remain untouched. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +DETECTOR_VERSION = "contact-data-redactor.v1" +_DEFAULT_MAX_INPUT_CHARS = 1_048_576 +_EMAIL_PATTERN = re.compile( + r"(? bool: + if _UNSUPPORTED_ID_PATTERN.fullmatch(value.strip()): + return False + digits = re.sub(r"\D", "", value) + if not 8 <= len(digits) <= 15: + return False + # International candidates must carry an explicit country prefix. This + # avoids treating arbitrary long numbers in prose as contact data. + if value.lstrip().startswith("+"): + return len(digits) >= 8 and len(digits) <= 15 + + # Korean mobile, landline, and business/service numbers. Separators are + # permitted by the candidate regex, but the digit structure stays strict. + if digits.startswith("01"): + return len(digits) in {10, 11} + if digits.startswith("02"): + return len(digits) in {9, 10} + if digits.startswith("0") and len(digits) in {10, 11}: + return True + return digits[:2] in {"15", "16", "18"} and len(digits) == 8 + + +def _candidate_spans(text: str) -> list[tuple[int, int, str]]: + candidates = [(match.start(), match.end(), "email") for match in _EMAIL_PATTERN.finditer(text)] + candidates.extend( + (match.start(), match.end(), "phone") + for match in _PHONE_PATTERN.finditer(text) + if _phone_is_supported(match.group(0)) + ) + # Email wins when a malformed phone-like candidate overlaps it. The stable + # sort also makes output deterministic for equal boundaries. + candidates.sort(key=lambda item: (item[0], item[1], 0 if item[2] == "email" else 1)) + selected: list[tuple[int, int, str]] = [] + for candidate in candidates: + if selected and candidate[0] < selected[-1][1]: + continue + selected.append(candidate) + return selected + + +def redact_contact_data( + text: str, + *, + placeholders: bool = False, + max_input_chars: int = _DEFAULT_MAX_INPUT_CHARS, +) -> ContactRedactionResult: + """Redact supported email and telephone forms without retaining values.""" + if not isinstance(text, str): + raise TypeError("text must be a string") + if max_input_chars <= 0: + raise ValueError("max_input_chars must be positive") + if len(text) > max_input_chars: + raise ValueError("input exceeds the contact redaction limit") + + output: list[str] = [] + matches: list[ContactMatch] = [] + counts = {"email": 0, "phone": 0} + source_cursor = 0 + output_cursor = 0 + for start, end, data_class in _candidate_spans(text): + output.append(text[source_cursor:start]) + output_cursor += start - source_cursor + counts[data_class] += 1 + replacement = ( + f"[{data_class.upper()}_{counts[data_class]}]" + if placeholders + else f"[REDACTED_{data_class.upper()}]" + ) + replacement_start = output_cursor + output.append(replacement) + output_cursor += len(replacement) + matches.append( + ContactMatch(data_class, start, end, replacement_start, output_cursor) + ) + source_cursor = end + output.append(text[source_cursor:]) + return ContactRedactionResult( + redacted_text="".join(output), + matches=tuple(matches), + match_counts={key: value for key, value in counts.items() if value}, + warnings=("unsupported_pii_classes_not_removed",), + ) + + +__all__ = ["ContactMatch", "ContactRedactionResult", "DETECTOR_VERSION", "redact_contact_data"] diff --git a/backend/services/url_evidence.py b/backend/services/url_evidence.py new file mode 100644 index 000000000..70231173e --- /dev/null +++ b/backend/services/url_evidence.py @@ -0,0 +1,185 @@ +"""Bounded, source-grounded extraction of absolute HTTP(S) URLs. + +The extractor is deliberately local-only: it identifies and validates syntax, +but never resolves hosts or performs network requests. ``source_locations`` +keeps every occurrence when normalized URLs are deduplicated. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from urllib.parse import SplitResult, urlsplit, urlunsplit + +_CANDIDATE = re.compile(r"(?i)https?://[^\s<>\"']+") +_TRAILING_PUNCTUATION = ".,;:!?" +_DEFAULT_MAX_INPUT_CHARS = 100_000 +_DEFAULT_MAX_MATCHES = 100 +_DEFAULT_MAX_MATCH_CHARS = 2_048 + + +@dataclass(frozen=True, slots=True) +class UrlEvidence: + """One normalized URL and every source location where it occurred.""" + + raw_value: str + normalized_value: str | None + source_start: int + source_end: int + source_locations: tuple[tuple[int, int], ...] + scheme_code: str + host_value: str | None + contains_userinfo: bool + validation_status: str + warning_codes: tuple[str, ...] + + +def _trim_candidate(candidate: str) -> str: + value = candidate.rstrip(_TRAILING_PUNCTUATION) + while value.endswith(")") and value.count(")") > value.count("("): + value = value[:-1] + return value + + +def _parsed_host(parsed: SplitResult) -> str | None: + try: + return parsed.hostname + except ValueError: + return None + + +def _normalize(parsed: SplitResult, host: str) -> str: + # Rebuild from parsed components so equivalent casing in the scheme/host + # does not create separate evidence records. Other source spelling stays. + username = parsed.username + password = parsed.password + userinfo = "" + if username is not None: + userinfo = username + if password is not None: + userinfo += f":{password}" + userinfo += "@" + authority_host = f"[{host}]" if ":" in host and not host.startswith("[") else host + port = f":{parsed.port}" if parsed.port is not None else "" + return urlunsplit( + (parsed.scheme.lower(), f"{userinfo}{authority_host}{port}", parsed.path, parsed.query, parsed.fragment) + ) + + +def extract_url_evidence( + text: str, + *, + max_input_chars: int = _DEFAULT_MAX_INPUT_CHARS, + max_matches: int = _DEFAULT_MAX_MATCHES, + max_match_chars: int = _DEFAULT_MAX_MATCH_CHARS, +) -> tuple[UrlEvidence, ...]: + """Extract bounded HTTP(S) evidence without fetching any URL. + + Repeated normalized URLs are represented once, with all source spans in + ``source_locations``. Offsets use Python's Unicode string indexing. + """ + if not isinstance(text, str): + raise TypeError("text must be a string") + if max_input_chars < 0 or max_matches < 0 or max_match_chars <= 0: + raise ValueError("extraction limits must be non-negative and finite") + if len(text) > max_input_chars: + raise ValueError("input exceeds the URL evidence limit") + + records: dict[str, UrlEvidence] = {} + for index, match in enumerate(_CANDIDATE.finditer(text)): + if index >= max_matches: + break + raw_value = _trim_candidate(match.group(0)) + start = match.start() + end = start + len(raw_value) + warnings: list[str] = [] + if len(raw_value) > max_match_chars: + raw_value = raw_value[:max_match_chars] + end = start + len(raw_value) + warnings.append("match_too_long") + + try: + parsed = urlsplit(raw_value) + except ValueError: + key = f"invalid:{raw_value}" + existing = records.get(key) + if existing is not None: + records[key] = UrlEvidence( + raw_value=existing.raw_value, + normalized_value=None, + source_start=existing.source_start, + source_end=existing.source_end, + source_locations=existing.source_locations + ((start, end),), + scheme_code=existing.scheme_code, + host_value=None, + contains_userinfo=existing.contains_userinfo, + validation_status="invalid", + warning_codes=existing.warning_codes, + ) + continue + records[key] = UrlEvidence( + raw_value=raw_value, + normalized_value=None, + source_start=start, + source_end=end, + source_locations=((start, end),), + scheme_code=raw_value.split(":", 1)[0].lower(), + host_value=None, + contains_userinfo="@" in raw_value, + validation_status="invalid", + warning_codes=("invalid_host",), + ) + continue + host = _parsed_host(parsed) + contains_userinfo = parsed.username is not None or parsed.password is not None + if contains_userinfo: + warnings.append("userinfo_present") + if parsed.scheme.lower() not in {"http", "https"}: + warnings.append("unsupported_scheme") + if host is None or not parsed.netloc: + warnings.append("missing_host") + try: + parsed.port + except ValueError: + warnings.append("invalid_port") + if "[" in parsed.netloc and "]" not in parsed.netloc: + warnings.append("invalid_host") + if not warnings: + normalized = _normalize(parsed, host) + status = "warning" if contains_userinfo else "valid" + else: + normalized = None + status = "invalid" + key = normalized or f"invalid:{raw_value}" + existing = records.get(key) + location = (start, end) + if existing is not None: + records[key] = UrlEvidence( + raw_value=existing.raw_value, + normalized_value=existing.normalized_value, + source_start=existing.source_start, + source_end=existing.source_end, + source_locations=existing.source_locations + (location,), + scheme_code=existing.scheme_code, + host_value=existing.host_value, + contains_userinfo=existing.contains_userinfo, + validation_status=existing.validation_status, + warning_codes=existing.warning_codes, + ) + continue + records[key] = UrlEvidence( + raw_value=raw_value, + normalized_value=normalized, + source_start=start, + source_end=end, + source_locations=(location,), + scheme_code=parsed.scheme.lower(), + host_value=host, + contains_userinfo=contains_userinfo, + validation_status=status, + warning_codes=tuple(dict.fromkeys(warnings)), + ) + return tuple(records.values()) + + +__all__ = ["UrlEvidence", "extract_url_evidence"] diff --git a/backend/tests/test_contact_data_redactor.py b/backend/tests/test_contact_data_redactor.py new file mode 100644 index 000000000..65e1e9df1 --- /dev/null +++ b/backend/tests/test_contact_data_redactor.py @@ -0,0 +1,52 @@ +import pytest + +from services.contact_data_redactor import DETECTOR_VERSION, redact_contact_data + + +def test_redacts_email_and_korean_phone_with_output_spans(): + result = redact_contact_data("문의: user@example.com, 010-1234-5678") + + assert result.redacted_text == "문의: [REDACTED_EMAIL], [REDACTED_PHONE]" + assert result.match_counts == {"email": 1, "phone": 1} + assert [match.data_class for match in result.matches] == ["email", "phone"] + assert all(match.detector_version == DETECTOR_VERSION for match in result.matches) + for match in result.matches: + assert result.redacted_text[match.replacement_start : match.replacement_end].startswith("[REDACTED_") + + +def test_placeholders_are_deterministic_and_do_not_expose_values(): + result = redact_contact_data( + "a@example.com a@example.com +82 10 1234 5678", placeholders=True + ) + + assert result.redacted_text == "[EMAIL_1] [EMAIL_2] [PHONE_1]" + assert "a@example.com" not in result.redacted_text + assert "+82" not in result.redacted_text + + +def test_avoids_phone_near_misses_and_warns_about_unsupported_classes(): + result = redact_contact_data( + "order 1234567, 주민번호 900101-1234567, name Alice" + ) + + assert result.redacted_text == "order 1234567, 주민번호 900101-1234567, name Alice" + assert result.matches == () + assert result.warnings == ("unsupported_pii_classes_not_removed",) + + +def test_supports_e164_and_korean_landline_and_service_forms(): + result = redact_contact_data("+1 (415) 555-2671 / 02-1234-5678 / 1588-1234") + + assert result.redacted_text == ( + "[REDACTED_PHONE] / [REDACTED_PHONE] / [REDACTED_PHONE]" + ) + assert result.match_counts == {"phone": 3} + + +def test_rejects_non_string_and_oversized_input(): + with pytest.raises(TypeError): + redact_contact_data(123) # type: ignore[arg-type] + with pytest.raises(ValueError): + redact_contact_data("x", max_input_chars=0) + with pytest.raises(ValueError): + redact_contact_data("abcd", max_input_chars=3) diff --git a/backend/tests/test_url_evidence.py b/backend/tests/test_url_evidence.py new file mode 100644 index 000000000..dcc372139 --- /dev/null +++ b/backend/tests/test_url_evidence.py @@ -0,0 +1,51 @@ +from services.url_evidence import extract_url_evidence + + +def test_extracts_normalizes_and_preserves_unicode_offsets(): + result = extract_url_evidence("안내: HTTPS://Example.com/a?q=1#x.") + + assert len(result) == 1 + evidence = result[0] + assert evidence.raw_value == "HTTPS://Example.com/a?q=1#x" + assert evidence.normalized_value == "https://example.com/a?q=1#x" + assert evidence.source_start == 4 + assert evidence.source_end == 31 + assert evidence.source_locations == ((4, 31),) + assert evidence.validation_status == "valid" + + +def test_deduplicates_normalized_urls_without_losing_locations(): + result = extract_url_evidence("https://example.com https://EXAMPLE.com") + + assert len(result) == 1 + assert result[0].source_locations == ((0, 19), (20, 39)) + + +def test_marks_userinfo_without_treating_it_as_safe(): + result = extract_url_evidence("https://user:secret@example.com/path") + + assert result[0].normalized_value is None + assert result[0].contains_userinfo is True + assert result[0].validation_status == "invalid" + assert "userinfo_present" in result[0].warning_codes + + +def test_handles_parentheses_and_limits_input_and_matches(): + result = extract_url_evidence( + "(https://example.com/a), https://two.example/x https://three.example/y", + max_matches=2, + ) + + assert [item.normalized_value for item in result] == [ + "https://example.com/a", + "https://two.example/x", + ] + + +def test_malformed_authority_returns_invalid_evidence_instead_of_raising(): + result = extract_url_evidence("https://[not-an-ip https://[not-an-ip") + + assert result[0].validation_status == "invalid" + assert result[0].normalized_value is None + assert result[0].warning_codes == ("invalid_host",) + assert len(result[0].source_locations) == 2 From 97455ad475a4f5f37bbe78109915c6a227bbb212 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 11 Sep 2026 09:15:17 +0900 Subject: [PATCH 2/4] feat(tools): add secure content checksum component Add the pure checksum layer for issue #1247 with SHA-256, SHA3-256, and BLAKE2b-256 support, explicit UTF-8 text handling, chunked hashing equivalence, and fail-closed rejection of legacy MD5/SHA-1 algorithms. --- backend/services/content_checksum.py | 55 ++++++++++++++++++++++++++ backend/tests/test_content_checksum.py | 48 ++++++++++++++++++++++ 2 files changed, 103 insertions(+) create mode 100644 backend/services/content_checksum.py create mode 100644 backend/tests/test_content_checksum.py diff --git a/backend/services/content_checksum.py b/backend/services/content_checksum.py new file mode 100644 index 000000000..95eb1f492 --- /dev/null +++ b/backend/services/content_checksum.py @@ -0,0 +1,55 @@ +"""Secure content checksums for non-authentication integrity use. + +The normal surface permits only SHA-256, SHA3-256, and BLAKE2b-256. Legacy +digests are intentionally not exposed as compatibility options here. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Iterable + +SUPPORTED_ALGORITHMS = ("sha256", "sha3_256", "blake2b_256") +_ALGORITHM_FACTORIES = { + "sha256": hashlib.sha256, + "sha3_256": hashlib.sha3_256, + "blake2b_256": lambda: hashlib.blake2b(digest_size=32), +} + + +def _content_bytes(content: bytes | str) -> bytes: + if isinstance(content, bytes): + return content + if isinstance(content, str): + return content.encode("utf-8") + raise TypeError("content must be bytes or str") + + +def _new_hasher(algorithm: str): + try: + return _ALGORITHM_FACTORIES[algorithm]() + except KeyError as exc: + raise ValueError("unsupported checksum algorithm") from exc + + +def generate_content_checksum( + content: bytes | str, + algorithm: str = "sha256", +) -> str: + """Return a lowercase hexadecimal checksum for content.""" + hasher = _new_hasher(algorithm) + hasher.update(_content_bytes(content)) + return hasher.hexdigest() + + +def generate_chunked_checksum( + chunks: Iterable[bytes], + algorithm: str = "sha256", +) -> str: + """Return the same checksum as one-shot hashing over ordered byte chunks.""" + hasher = _new_hasher(algorithm) + for chunk in chunks: + if not isinstance(chunk, bytes): + raise TypeError("checksum chunks must be bytes") + hasher.update(chunk) + return hasher.hexdigest() diff --git a/backend/tests/test_content_checksum.py b/backend/tests/test_content_checksum.py new file mode 100644 index 000000000..aa9c93a2b --- /dev/null +++ b/backend/tests/test_content_checksum.py @@ -0,0 +1,48 @@ +import hashlib + +import pytest + +from services.content_checksum import ( + SUPPORTED_ALGORITHMS, + generate_chunked_checksum, + generate_content_checksum, +) + + +@pytest.mark.parametrize("algorithm", SUPPORTED_ALGORITHMS) +def test_supported_algorithms_match_hashlib(algorithm): + content = "Naruon checksum evidence" + expected = { + "sha256": hashlib.sha256(content.encode()).hexdigest(), + "sha3_256": hashlib.sha3_256(content.encode()).hexdigest(), + "blake2b_256": hashlib.blake2b(content.encode(), digest_size=32).hexdigest(), + }[algorithm] + + assert generate_content_checksum(content, algorithm) == expected + + +@pytest.mark.parametrize("algorithm", SUPPORTED_ALGORITHMS) +def test_chunked_checksum_matches_one_shot(algorithm): + chunks = [b"Naruon ", b"checksum ", b"evidence"] + + assert generate_chunked_checksum(chunks, algorithm) == generate_content_checksum( + b"".join(chunks), algorithm + ) + + +def test_text_encoding_is_explicit_utf8(): + assert generate_content_checksum("한글") == hashlib.sha256("한글".encode("utf-8")).hexdigest() + + +@pytest.mark.parametrize("algorithm", ["md5", "sha1", "unknown"]) +def test_unsupported_and_legacy_algorithms_fail_closed(algorithm): + with pytest.raises(ValueError, match="unsupported checksum algorithm"): + generate_content_checksum(b"content", algorithm) + + +def test_invalid_content_types_fail_closed_without_serialization(): + with pytest.raises(TypeError, match="content must be bytes or str"): + generate_content_checksum(123) # type: ignore[arg-type] + + with pytest.raises(TypeError, match="checksum chunks must be bytes"): + generate_chunked_checksum([b"ok", "not bytes"]) # type: ignore[list-item] From 4aaad40c0f482d775bb0b069c808e6c0d9f6d195 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 11 Sep 2026 09:39:43 +0900 Subject: [PATCH 3/4] test(tools): cover contact redaction spans and Unicode offsets Exercise source and replacement span round trips with Unicode prefixes, deterministic placeholders, and malformed overlapping phone/email-like candidates without broadening the supported PII contract. --- backend/tests/test_contact_data_redactor.py | 25 ++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_contact_data_redactor.py b/backend/tests/test_contact_data_redactor.py index 65e1e9df1..3f82244ae 100644 --- a/backend/tests/test_contact_data_redactor.py +++ b/backend/tests/test_contact_data_redactor.py @@ -4,16 +4,39 @@ def test_redacts_email_and_korean_phone_with_output_spans(): - result = redact_contact_data("문의: user@example.com, 010-1234-5678") + source = "문의: user@example.com, 010-1234-5678" + result = redact_contact_data(source) assert result.redacted_text == "문의: [REDACTED_EMAIL], [REDACTED_PHONE]" assert result.match_counts == {"email": 1, "phone": 1} assert [match.data_class for match in result.matches] == ["email", "phone"] assert all(match.detector_version == DETECTOR_VERSION for match in result.matches) for match in result.matches: + assert source[match.source_start : match.source_end] not in result.redacted_text assert result.redacted_text[match.replacement_start : match.replacement_end].startswith("[REDACTED_") +def test_spans_round_trip_with_unicode_prefix_and_multiple_placeholders(): + source = "안내 📬 first@example.com 및 +82 10 1234 5678" + result = redact_contact_data(source, placeholders=True) + + assert [ + source[match.source_start : match.source_end] for match in result.matches + ] == ["first@example.com", "+82 10 1234 5678"] + assert [ + result.redacted_text[match.replacement_start : match.replacement_end] + for match in result.matches + ] == ["[EMAIL_1]", "[PHONE_1]"] + assert result.redacted_text == "안내 📬 [EMAIL_1] 및 [PHONE_1]" + + +def test_overlapping_phone_candidate_does_not_consume_email_like_suffix(): + result = redact_contact_data("contact 01012345678@example.com") + + assert [match.data_class for match in result.matches] == ["phone"] + assert result.redacted_text == "contact [REDACTED_PHONE]@example.com" + + def test_placeholders_are_deterministic_and_do_not_expose_values(): result = redact_contact_data( "a@example.com a@example.com +82 10 1234 5678", placeholders=True From 0275f130731856c1e88b46f7cee28c7778b345f1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 11 Sep 2026 09:43:09 +0900 Subject: [PATCH 4/4] chore(tools): retire duplicate data-hygiene slice The generated component slice duplicates canonical owners #1418 (URL/contact hygiene) and #1361 (content checksum) while weakening their fail-closed bounds and evidence contracts. Preserve this branch history but return the effective tree to protected develop so no competing implementation can merge from this lane. --- backend/services/contact_data_redactor.py | 130 -------------- backend/services/content_checksum.py | 55 ------ backend/services/url_evidence.py | 185 -------------------- backend/tests/test_contact_data_redactor.py | 75 -------- backend/tests/test_content_checksum.py | 48 ----- backend/tests/test_url_evidence.py | 51 ------ 6 files changed, 544 deletions(-) delete mode 100644 backend/services/contact_data_redactor.py delete mode 100644 backend/services/content_checksum.py delete mode 100644 backend/services/url_evidence.py delete mode 100644 backend/tests/test_contact_data_redactor.py delete mode 100644 backend/tests/test_content_checksum.py delete mode 100644 backend/tests/test_url_evidence.py diff --git a/backend/services/contact_data_redactor.py b/backend/services/contact_data_redactor.py deleted file mode 100644 index d40b7fd38..000000000 --- a/backend/services/contact_data_redactor.py +++ /dev/null @@ -1,130 +0,0 @@ -"""Bounded redaction for the explicitly supported contact-data classes. - -This module is not a general PII anonymizer. It detects email addresses and -telephone numbers only; unsupported personal-data classes remain untouched. -""" - -from __future__ import annotations - -import re -from dataclasses import dataclass - -DETECTOR_VERSION = "contact-data-redactor.v1" -_DEFAULT_MAX_INPUT_CHARS = 1_048_576 -_EMAIL_PATTERN = re.compile( - r"(? bool: - if _UNSUPPORTED_ID_PATTERN.fullmatch(value.strip()): - return False - digits = re.sub(r"\D", "", value) - if not 8 <= len(digits) <= 15: - return False - # International candidates must carry an explicit country prefix. This - # avoids treating arbitrary long numbers in prose as contact data. - if value.lstrip().startswith("+"): - return len(digits) >= 8 and len(digits) <= 15 - - # Korean mobile, landline, and business/service numbers. Separators are - # permitted by the candidate regex, but the digit structure stays strict. - if digits.startswith("01"): - return len(digits) in {10, 11} - if digits.startswith("02"): - return len(digits) in {9, 10} - if digits.startswith("0") and len(digits) in {10, 11}: - return True - return digits[:2] in {"15", "16", "18"} and len(digits) == 8 - - -def _candidate_spans(text: str) -> list[tuple[int, int, str]]: - candidates = [(match.start(), match.end(), "email") for match in _EMAIL_PATTERN.finditer(text)] - candidates.extend( - (match.start(), match.end(), "phone") - for match in _PHONE_PATTERN.finditer(text) - if _phone_is_supported(match.group(0)) - ) - # Email wins when a malformed phone-like candidate overlaps it. The stable - # sort also makes output deterministic for equal boundaries. - candidates.sort(key=lambda item: (item[0], item[1], 0 if item[2] == "email" else 1)) - selected: list[tuple[int, int, str]] = [] - for candidate in candidates: - if selected and candidate[0] < selected[-1][1]: - continue - selected.append(candidate) - return selected - - -def redact_contact_data( - text: str, - *, - placeholders: bool = False, - max_input_chars: int = _DEFAULT_MAX_INPUT_CHARS, -) -> ContactRedactionResult: - """Redact supported email and telephone forms without retaining values.""" - if not isinstance(text, str): - raise TypeError("text must be a string") - if max_input_chars <= 0: - raise ValueError("max_input_chars must be positive") - if len(text) > max_input_chars: - raise ValueError("input exceeds the contact redaction limit") - - output: list[str] = [] - matches: list[ContactMatch] = [] - counts = {"email": 0, "phone": 0} - source_cursor = 0 - output_cursor = 0 - for start, end, data_class in _candidate_spans(text): - output.append(text[source_cursor:start]) - output_cursor += start - source_cursor - counts[data_class] += 1 - replacement = ( - f"[{data_class.upper()}_{counts[data_class]}]" - if placeholders - else f"[REDACTED_{data_class.upper()}]" - ) - replacement_start = output_cursor - output.append(replacement) - output_cursor += len(replacement) - matches.append( - ContactMatch(data_class, start, end, replacement_start, output_cursor) - ) - source_cursor = end - output.append(text[source_cursor:]) - return ContactRedactionResult( - redacted_text="".join(output), - matches=tuple(matches), - match_counts={key: value for key, value in counts.items() if value}, - warnings=("unsupported_pii_classes_not_removed",), - ) - - -__all__ = ["ContactMatch", "ContactRedactionResult", "DETECTOR_VERSION", "redact_contact_data"] diff --git a/backend/services/content_checksum.py b/backend/services/content_checksum.py deleted file mode 100644 index 95eb1f492..000000000 --- a/backend/services/content_checksum.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Secure content checksums for non-authentication integrity use. - -The normal surface permits only SHA-256, SHA3-256, and BLAKE2b-256. Legacy -digests are intentionally not exposed as compatibility options here. -""" - -from __future__ import annotations - -import hashlib -from collections.abc import Iterable - -SUPPORTED_ALGORITHMS = ("sha256", "sha3_256", "blake2b_256") -_ALGORITHM_FACTORIES = { - "sha256": hashlib.sha256, - "sha3_256": hashlib.sha3_256, - "blake2b_256": lambda: hashlib.blake2b(digest_size=32), -} - - -def _content_bytes(content: bytes | str) -> bytes: - if isinstance(content, bytes): - return content - if isinstance(content, str): - return content.encode("utf-8") - raise TypeError("content must be bytes or str") - - -def _new_hasher(algorithm: str): - try: - return _ALGORITHM_FACTORIES[algorithm]() - except KeyError as exc: - raise ValueError("unsupported checksum algorithm") from exc - - -def generate_content_checksum( - content: bytes | str, - algorithm: str = "sha256", -) -> str: - """Return a lowercase hexadecimal checksum for content.""" - hasher = _new_hasher(algorithm) - hasher.update(_content_bytes(content)) - return hasher.hexdigest() - - -def generate_chunked_checksum( - chunks: Iterable[bytes], - algorithm: str = "sha256", -) -> str: - """Return the same checksum as one-shot hashing over ordered byte chunks.""" - hasher = _new_hasher(algorithm) - for chunk in chunks: - if not isinstance(chunk, bytes): - raise TypeError("checksum chunks must be bytes") - hasher.update(chunk) - return hasher.hexdigest() diff --git a/backend/services/url_evidence.py b/backend/services/url_evidence.py deleted file mode 100644 index 70231173e..000000000 --- a/backend/services/url_evidence.py +++ /dev/null @@ -1,185 +0,0 @@ -"""Bounded, source-grounded extraction of absolute HTTP(S) URLs. - -The extractor is deliberately local-only: it identifies and validates syntax, -but never resolves hosts or performs network requests. ``source_locations`` -keeps every occurrence when normalized URLs are deduplicated. -""" - -from __future__ import annotations - -import re -from dataclasses import dataclass -from urllib.parse import SplitResult, urlsplit, urlunsplit - -_CANDIDATE = re.compile(r"(?i)https?://[^\s<>\"']+") -_TRAILING_PUNCTUATION = ".,;:!?" -_DEFAULT_MAX_INPUT_CHARS = 100_000 -_DEFAULT_MAX_MATCHES = 100 -_DEFAULT_MAX_MATCH_CHARS = 2_048 - - -@dataclass(frozen=True, slots=True) -class UrlEvidence: - """One normalized URL and every source location where it occurred.""" - - raw_value: str - normalized_value: str | None - source_start: int - source_end: int - source_locations: tuple[tuple[int, int], ...] - scheme_code: str - host_value: str | None - contains_userinfo: bool - validation_status: str - warning_codes: tuple[str, ...] - - -def _trim_candidate(candidate: str) -> str: - value = candidate.rstrip(_TRAILING_PUNCTUATION) - while value.endswith(")") and value.count(")") > value.count("("): - value = value[:-1] - return value - - -def _parsed_host(parsed: SplitResult) -> str | None: - try: - return parsed.hostname - except ValueError: - return None - - -def _normalize(parsed: SplitResult, host: str) -> str: - # Rebuild from parsed components so equivalent casing in the scheme/host - # does not create separate evidence records. Other source spelling stays. - username = parsed.username - password = parsed.password - userinfo = "" - if username is not None: - userinfo = username - if password is not None: - userinfo += f":{password}" - userinfo += "@" - authority_host = f"[{host}]" if ":" in host and not host.startswith("[") else host - port = f":{parsed.port}" if parsed.port is not None else "" - return urlunsplit( - (parsed.scheme.lower(), f"{userinfo}{authority_host}{port}", parsed.path, parsed.query, parsed.fragment) - ) - - -def extract_url_evidence( - text: str, - *, - max_input_chars: int = _DEFAULT_MAX_INPUT_CHARS, - max_matches: int = _DEFAULT_MAX_MATCHES, - max_match_chars: int = _DEFAULT_MAX_MATCH_CHARS, -) -> tuple[UrlEvidence, ...]: - """Extract bounded HTTP(S) evidence without fetching any URL. - - Repeated normalized URLs are represented once, with all source spans in - ``source_locations``. Offsets use Python's Unicode string indexing. - """ - if not isinstance(text, str): - raise TypeError("text must be a string") - if max_input_chars < 0 or max_matches < 0 or max_match_chars <= 0: - raise ValueError("extraction limits must be non-negative and finite") - if len(text) > max_input_chars: - raise ValueError("input exceeds the URL evidence limit") - - records: dict[str, UrlEvidence] = {} - for index, match in enumerate(_CANDIDATE.finditer(text)): - if index >= max_matches: - break - raw_value = _trim_candidate(match.group(0)) - start = match.start() - end = start + len(raw_value) - warnings: list[str] = [] - if len(raw_value) > max_match_chars: - raw_value = raw_value[:max_match_chars] - end = start + len(raw_value) - warnings.append("match_too_long") - - try: - parsed = urlsplit(raw_value) - except ValueError: - key = f"invalid:{raw_value}" - existing = records.get(key) - if existing is not None: - records[key] = UrlEvidence( - raw_value=existing.raw_value, - normalized_value=None, - source_start=existing.source_start, - source_end=existing.source_end, - source_locations=existing.source_locations + ((start, end),), - scheme_code=existing.scheme_code, - host_value=None, - contains_userinfo=existing.contains_userinfo, - validation_status="invalid", - warning_codes=existing.warning_codes, - ) - continue - records[key] = UrlEvidence( - raw_value=raw_value, - normalized_value=None, - source_start=start, - source_end=end, - source_locations=((start, end),), - scheme_code=raw_value.split(":", 1)[0].lower(), - host_value=None, - contains_userinfo="@" in raw_value, - validation_status="invalid", - warning_codes=("invalid_host",), - ) - continue - host = _parsed_host(parsed) - contains_userinfo = parsed.username is not None or parsed.password is not None - if contains_userinfo: - warnings.append("userinfo_present") - if parsed.scheme.lower() not in {"http", "https"}: - warnings.append("unsupported_scheme") - if host is None or not parsed.netloc: - warnings.append("missing_host") - try: - parsed.port - except ValueError: - warnings.append("invalid_port") - if "[" in parsed.netloc and "]" not in parsed.netloc: - warnings.append("invalid_host") - if not warnings: - normalized = _normalize(parsed, host) - status = "warning" if contains_userinfo else "valid" - else: - normalized = None - status = "invalid" - key = normalized or f"invalid:{raw_value}" - existing = records.get(key) - location = (start, end) - if existing is not None: - records[key] = UrlEvidence( - raw_value=existing.raw_value, - normalized_value=existing.normalized_value, - source_start=existing.source_start, - source_end=existing.source_end, - source_locations=existing.source_locations + (location,), - scheme_code=existing.scheme_code, - host_value=existing.host_value, - contains_userinfo=existing.contains_userinfo, - validation_status=existing.validation_status, - warning_codes=existing.warning_codes, - ) - continue - records[key] = UrlEvidence( - raw_value=raw_value, - normalized_value=normalized, - source_start=start, - source_end=end, - source_locations=(location,), - scheme_code=parsed.scheme.lower(), - host_value=host, - contains_userinfo=contains_userinfo, - validation_status=status, - warning_codes=tuple(dict.fromkeys(warnings)), - ) - return tuple(records.values()) - - -__all__ = ["UrlEvidence", "extract_url_evidence"] diff --git a/backend/tests/test_contact_data_redactor.py b/backend/tests/test_contact_data_redactor.py deleted file mode 100644 index 3f82244ae..000000000 --- a/backend/tests/test_contact_data_redactor.py +++ /dev/null @@ -1,75 +0,0 @@ -import pytest - -from services.contact_data_redactor import DETECTOR_VERSION, redact_contact_data - - -def test_redacts_email_and_korean_phone_with_output_spans(): - source = "문의: user@example.com, 010-1234-5678" - result = redact_contact_data(source) - - assert result.redacted_text == "문의: [REDACTED_EMAIL], [REDACTED_PHONE]" - assert result.match_counts == {"email": 1, "phone": 1} - assert [match.data_class for match in result.matches] == ["email", "phone"] - assert all(match.detector_version == DETECTOR_VERSION for match in result.matches) - for match in result.matches: - assert source[match.source_start : match.source_end] not in result.redacted_text - assert result.redacted_text[match.replacement_start : match.replacement_end].startswith("[REDACTED_") - - -def test_spans_round_trip_with_unicode_prefix_and_multiple_placeholders(): - source = "안내 📬 first@example.com 및 +82 10 1234 5678" - result = redact_contact_data(source, placeholders=True) - - assert [ - source[match.source_start : match.source_end] for match in result.matches - ] == ["first@example.com", "+82 10 1234 5678"] - assert [ - result.redacted_text[match.replacement_start : match.replacement_end] - for match in result.matches - ] == ["[EMAIL_1]", "[PHONE_1]"] - assert result.redacted_text == "안내 📬 [EMAIL_1] 및 [PHONE_1]" - - -def test_overlapping_phone_candidate_does_not_consume_email_like_suffix(): - result = redact_contact_data("contact 01012345678@example.com") - - assert [match.data_class for match in result.matches] == ["phone"] - assert result.redacted_text == "contact [REDACTED_PHONE]@example.com" - - -def test_placeholders_are_deterministic_and_do_not_expose_values(): - result = redact_contact_data( - "a@example.com a@example.com +82 10 1234 5678", placeholders=True - ) - - assert result.redacted_text == "[EMAIL_1] [EMAIL_2] [PHONE_1]" - assert "a@example.com" not in result.redacted_text - assert "+82" not in result.redacted_text - - -def test_avoids_phone_near_misses_and_warns_about_unsupported_classes(): - result = redact_contact_data( - "order 1234567, 주민번호 900101-1234567, name Alice" - ) - - assert result.redacted_text == "order 1234567, 주민번호 900101-1234567, name Alice" - assert result.matches == () - assert result.warnings == ("unsupported_pii_classes_not_removed",) - - -def test_supports_e164_and_korean_landline_and_service_forms(): - result = redact_contact_data("+1 (415) 555-2671 / 02-1234-5678 / 1588-1234") - - assert result.redacted_text == ( - "[REDACTED_PHONE] / [REDACTED_PHONE] / [REDACTED_PHONE]" - ) - assert result.match_counts == {"phone": 3} - - -def test_rejects_non_string_and_oversized_input(): - with pytest.raises(TypeError): - redact_contact_data(123) # type: ignore[arg-type] - with pytest.raises(ValueError): - redact_contact_data("x", max_input_chars=0) - with pytest.raises(ValueError): - redact_contact_data("abcd", max_input_chars=3) diff --git a/backend/tests/test_content_checksum.py b/backend/tests/test_content_checksum.py deleted file mode 100644 index aa9c93a2b..000000000 --- a/backend/tests/test_content_checksum.py +++ /dev/null @@ -1,48 +0,0 @@ -import hashlib - -import pytest - -from services.content_checksum import ( - SUPPORTED_ALGORITHMS, - generate_chunked_checksum, - generate_content_checksum, -) - - -@pytest.mark.parametrize("algorithm", SUPPORTED_ALGORITHMS) -def test_supported_algorithms_match_hashlib(algorithm): - content = "Naruon checksum evidence" - expected = { - "sha256": hashlib.sha256(content.encode()).hexdigest(), - "sha3_256": hashlib.sha3_256(content.encode()).hexdigest(), - "blake2b_256": hashlib.blake2b(content.encode(), digest_size=32).hexdigest(), - }[algorithm] - - assert generate_content_checksum(content, algorithm) == expected - - -@pytest.mark.parametrize("algorithm", SUPPORTED_ALGORITHMS) -def test_chunked_checksum_matches_one_shot(algorithm): - chunks = [b"Naruon ", b"checksum ", b"evidence"] - - assert generate_chunked_checksum(chunks, algorithm) == generate_content_checksum( - b"".join(chunks), algorithm - ) - - -def test_text_encoding_is_explicit_utf8(): - assert generate_content_checksum("한글") == hashlib.sha256("한글".encode("utf-8")).hexdigest() - - -@pytest.mark.parametrize("algorithm", ["md5", "sha1", "unknown"]) -def test_unsupported_and_legacy_algorithms_fail_closed(algorithm): - with pytest.raises(ValueError, match="unsupported checksum algorithm"): - generate_content_checksum(b"content", algorithm) - - -def test_invalid_content_types_fail_closed_without_serialization(): - with pytest.raises(TypeError, match="content must be bytes or str"): - generate_content_checksum(123) # type: ignore[arg-type] - - with pytest.raises(TypeError, match="checksum chunks must be bytes"): - generate_chunked_checksum([b"ok", "not bytes"]) # type: ignore[list-item] diff --git a/backend/tests/test_url_evidence.py b/backend/tests/test_url_evidence.py deleted file mode 100644 index dcc372139..000000000 --- a/backend/tests/test_url_evidence.py +++ /dev/null @@ -1,51 +0,0 @@ -from services.url_evidence import extract_url_evidence - - -def test_extracts_normalizes_and_preserves_unicode_offsets(): - result = extract_url_evidence("안내: HTTPS://Example.com/a?q=1#x.") - - assert len(result) == 1 - evidence = result[0] - assert evidence.raw_value == "HTTPS://Example.com/a?q=1#x" - assert evidence.normalized_value == "https://example.com/a?q=1#x" - assert evidence.source_start == 4 - assert evidence.source_end == 31 - assert evidence.source_locations == ((4, 31),) - assert evidence.validation_status == "valid" - - -def test_deduplicates_normalized_urls_without_losing_locations(): - result = extract_url_evidence("https://example.com https://EXAMPLE.com") - - assert len(result) == 1 - assert result[0].source_locations == ((0, 19), (20, 39)) - - -def test_marks_userinfo_without_treating_it_as_safe(): - result = extract_url_evidence("https://user:secret@example.com/path") - - assert result[0].normalized_value is None - assert result[0].contains_userinfo is True - assert result[0].validation_status == "invalid" - assert "userinfo_present" in result[0].warning_codes - - -def test_handles_parentheses_and_limits_input_and_matches(): - result = extract_url_evidence( - "(https://example.com/a), https://two.example/x https://three.example/y", - max_matches=2, - ) - - assert [item.normalized_value for item in result] == [ - "https://example.com/a", - "https://two.example/x", - ] - - -def test_malformed_authority_returns_invalid_evidence_instead_of_raising(): - result = extract_url_evidence("https://[not-an-ip https://[not-an-ip") - - assert result[0].validation_status == "invalid" - assert result[0].normalized_value is None - assert result[0].warning_codes == ("invalid_host",) - assert len(result[0].source_locations) == 2