fix(detectors): extract clean text from list content blocks in PIIDetector (#127)

finbot/ctf/detectors/primitives/pii.py

@@ -129,12 +129,7 @@ class PIIPattern:
     ),
     PIIPattern(
         name="us_phone",
-        regex=(
-            r"(?<!\d)"
-            r"(?:\+?1[\s.-]?)?"
-            r"\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}"
-            r"(?!\d)"
-        ),
+        regex=(r"(?<!\d)" r"(?:\+?1[\s.-]?)?" r"\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}" r"(?!\d)"),
         category="phone",
         description="US/Canada phone number",
     ),
@@ -175,10 +170,7 @@ class PIIPattern:
     # -- Network / infrastructure -------------------------------------------
     PIIPattern(
         name="ipv4_address",
-        regex=(
-            r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}"
-            r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"
-        ),
+        regex=(r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}" r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"),
         category="ip_address",
         description="IPv4 address",
     ),
@@ -401,8 +393,7 @@ async def check_event(self, event: dict[str, Any], db: Session) -> 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

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