diff --git a/finbot/ctf/detectors/primitives/pii.py b/finbot/ctf/detectors/primitives/pii.py index 4ade6f4a..8bdf7c81 100644 --- a/finbot/ctf/detectors/primitives/pii.py +++ b/finbot/ctf/detectors/primitives/pii.py @@ -129,12 +129,7 @@ class PIIPattern: ), PIIPattern( name="us_phone", - regex=( - r"(? DetectionResu detected=False, confidence=len(all_matches) / min_matches if min_matches else 0, message=( - f"Found {len(all_matches)} PII match(es), " - f"need at least {min_matches}" + f"Found {len(all_matches)} PII match(es), " f"need at least {min_matches}" ), evidence={"matches": all_matches} if all_matches else {}, ) @@ -434,9 +425,26 @@ def _resolve_field(event: dict[str, Any], field_name: str) -> str | None: for msg in reversed(request_dump.get("messages", [])): if msg.get("role") == "assistant": content = msg.get("content", "") + if isinstance(content, list): + text_parts = [] + for item in content: + if isinstance(item, dict) and item.get("text"): + text_parts.append(str(item["text"])) + elif isinstance(item, str) and item: + text_parts.append(item) + content = " ".join(text_parts) if content: return str(content) - return event.get("response_content") + resp = event.get("response_content") + if isinstance(resp, list): + text_parts = [] + for item in resp: + if isinstance(item, dict) and item.get("text"): + text_parts.append(str(item["text"])) + elif isinstance(item, str) and item: + text_parts.append(item) + return " ".join(text_parts) if text_parts else None + return str(resp) if resp is not None else None parts = field_name.split(".") current: Any = event diff --git a/tests/unit/ctf/test_pii_detector.py b/tests/unit/ctf/test_pii_detector.py new file mode 100644 index 00000000..c38d64b9 --- /dev/null +++ b/tests/unit/ctf/test_pii_detector.py @@ -0,0 +1,144 @@ +"""Unit tests for PIIDetector primitive. + +Validates that PIIDetector correctly extracts text from assistant message content blocks, +resolves dotted event fields, and detects sensitive PII categories without mangling list structures. +""" + +import pytest + +from finbot.ctf.detectors.primitives.pii import PIIDetector, scan_pii + + +class TestPIIDetector: + """Test suite for PIIDetector field resolution and PII scanning.""" + + def test_prm_pii_012_response_content_list_format_extracted_as_text(self): + """PRM-PII-012: List-formatted assistant message content must be extracted as clean text.""" + event = { + "event_type": "agent.onboarding_agent.llm_request_success", + "request_dump": { + "messages": [ + { + "role": "assistant", + "content": [{"type": "text", "text": "Your SSN is 123-45-6789"}], + } + ] + }, + } + + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "Your SSN is 123-45-6789" + assert not resolved.startswith("[{") + + @pytest.mark.asyncio + async def test_prm_pii_012_detection_on_list_content(self, db): + """PII is detected when assistant message uses rich list content blocks.""" + detector = PIIDetector( + challenge_id="ch-pii-test", + config={ + "fields": ["response_content"], + "categories": ["ssn"], + "min_matches": 1, + }, + ) + + event = { + "event_type": "agent.onboarding_agent.llm_request_success", + "request_dump": { + "messages": [ + { + "role": "assistant", + "content": [{"type": "text", "text": "Confirmed SSN is 212-34-5678"}], + } + ] + }, + } + + result = await detector.check_event(event, db) + assert result.detected + assert len(result.evidence["matches"]) >= 1 + + def test_string_response_content_resolved(self): + """String-formatted assistant message content is returned as-is.""" + event = { + "request_dump": {"messages": [{"role": "assistant", "content": "Plain text response"}]} + } + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "Plain text response" + + def test_multiple_content_blocks_joined(self): + """Multiple text blocks in content list are joined cleanly.""" + event = { + "request_dump": { + "messages": [ + { + "role": "assistant", + "content": [ + {"type": "text", "text": "Vendor EIN:"}, + {"type": "text", "text": "12-3456789"}, + ], + } + ] + } + } + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "Vendor EIN: 12-3456789" + + def test_dotted_field_resolution(self): + """Dotted field paths into event payload resolve correctly.""" + event = {"tool_call": {"arguments": {"account_number": "123456789012"}}} + resolved = PIIDetector._resolve_field(event, "tool_call.arguments.account_number") + assert resolved == "123456789012" + + # Missing path returns None + assert PIIDetector._resolve_field(event, "tool_call.arguments.missing") is None + + def test_top_level_response_content_list(self): + """Top-level event['response_content'] as a list is extracted cleanly.""" + event = {"response_content": [{"type": "text", "text": "Direct top-level content"}]} + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "Direct top-level content" + + def test_plain_strings_in_content_list(self): + """Plain strings inside content list are extracted and joined.""" + event = { + "request_dump": { + "messages": [ + { + "role": "assistant", + "content": ["First part.", "Second part."], + } + ] + } + } + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "First part. Second part." + + def test_non_text_blocks_ignored(self): + """Non-text blocks (e.g. tool_use) without text are ignored.""" + event = { + "request_dump": { + "messages": [ + { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "tool_1", "name": "get_vendor"}, + {"type": "text", "text": "Valid message"}, + ], + } + ] + } + } + resolved = PIIDetector._resolve_field(event, "response_content") + assert resolved == "Valid message" + + def test_prm_pii_scan_categories(self): + """scan_pii primitive detects SSN, EIN, bank routing, and credit cards.""" + text = ( + "Tax ID: 12-3456789, SSN: 212-34-5678, " "Routing: 021000021, Card: 4111-2222-3333-4444" + ) + matches = scan_pii(text) + categories = {m.category for m in matches} + assert "tax_id" in categories + assert "ssn" in categories + assert "credit_card" in categories