Harden URL redaction in inventory/review failure output (Issue #171)

gitea_audit.py

@@ -19,6 +19,8 @@ Design constraints:
 import os
 import json
 import datetime
+import re
+import urllib.parse
 
 # Result states for an audited action.
 ALLOWED = "allowed"
@@ -33,9 +35,87 @@ _SECRET_KEY_HINTS = ("token", "password", "secret", "authorization", "auth")
 # A string value starting with one of these has the following run redacted.
 _SECRET_VALUE_PREFIXES = ("token ", "Basic ", "Bearer ")
 
+# Known synthetic test-only domains/hostnames to preserve
+_SYNTHETIC_HOSTS = {
+    "example.com",
+    "example.test",
+    "example.invalid",
+    "internal.example",
+    "localhost",
+    "gitea.example.com",
+    "x"
+}
+
+# Known real service hostnames to redact even if not part of a full URL
+_REAL_HOSTS = {"gitea.prgs.cc", "gitea.dadeschools.net"}
+
+
+def redact_urls(text: str) -> str:
+    """Redact raw URLs, query-string secrets, and URL credentials.
+
+    Synthetic example/test-only domains are preserved to keep test fixtures functional
+    except for any embedded credentials or query parameters containing secrets.
+    """
+    if not isinstance(text, str) or not text:
+        return text
+
+    url_pattern = re.compile(r'(https?://[^\s)>\]}]+)', re.IGNORECASE)
+
+    def replace_url(match):
+        url_str = match.group(1)
+        try:
+            parsed = urllib.parse.urlsplit(url_str)
+            host = (parsed.hostname or "").lower()
+
+            is_synthetic = False
+            for sh in _SYNTHETIC_HOSTS:
+                if host == sh or host.endswith("." + sh):
+                    is_synthetic = True
+                    break
+
+            if is_synthetic:
+                # Rebuild synthetic URL to redact any credentials or query secrets
+                new_netloc = parsed.netloc
+                if parsed.username or parsed.password:
+                    netloc_clean = parsed.hostname
+                    if parsed.port:
+                        netloc_clean = f"{netloc_clean}:{parsed.port}"
+                    new_netloc = f"[REDACTED_USER]:[REDACTED_PASS]@{netloc_clean}"
+
+                new_query = parsed.query
+                if parsed.query:
+                    params = urllib.parse.parse_qsl(parsed.query)
+                    clean_params = []
+                    for k, v in params:
+                        if any(hint in k.lower() for hint in ("token", "password", "secret", "auth", "key")):
+                            clean_params.append((k, "[REDACTED]"))
+                        else:
+                            clean_params.append((k, v))
+                    new_query = urllib.parse.urlencode(clean_params)
+
+                return urllib.parse.urlunsplit((
+                    parsed.scheme,
+                    new_netloc,
+                    parsed.path,
+                    new_query,
+                    parsed.fragment
+                ))
+            else:
+                return "[REDACTED_URL]"
+        except Exception:
+            return "[REDACTED_URL]"
+
+    out = url_pattern.sub(replace_url, text)
+
+    # Redact raw occurrences of real service hostnames afterward
+    for host in _REAL_HOSTS:
+        out = out.replace(host, "[REDACTED_HOST]")
+
+    return out
+
 
 def _redact_str(text):
-    """Redact anything that looks like an Authorization credential in *text*."""
+    """Redact anything that looks like an Authorization credential or raw URL in *text*."""
     if not isinstance(text, str) or not text:
         return text
     out = text
@@ -50,7 +130,7 @@ def _redact_str(text):
                 j += 1
             out = out[:i] + prefix + REDACTED + out[j:]
             idx = i + len(prefix) + len(REDACTED)
-    return out
+    return redact_urls(out)
 
 
 def redact(value):

mcp_server.py

@@ -761,7 +761,7 @@ _SECRET_PREFIXES = ("token ", "Basic ")
 
 
 def _redact(text: str) -> str:
-    """Strip anything that looks like an Authorization credential from *text*.
+    """Strip anything that looks like an Authorization credential or raw URL from *text*.
 
     Errors raised by ``api_request`` echo the server response body, not the
     request headers, so a token should never appear — this is defence in depth
@@ -782,7 +782,9 @@ def _redact(text: str) -> str:
                 j += 1
             out = out[:i] + prefix + "[REDACTED]" + out[j:]
             idx = i + len(prefix) + len("[REDACTED]")
-    return out
+    # Redact raw URLs, query secrets, hostnames, etc.
+    import gitea_audit
+    return gitea_audit.redact_urls(out)
 
 
 # Review states that carry a submitted verdict. Gitea also emits PENDING

tests/test_audit.py

@@ -59,6 +59,33 @@ class TestRedaction(unittest.TestCase):
         self.assertEqual(gitea_audit.redact(42), 42)
         self.assertIsNone(gitea_audit.redact(None))
 
+    def test_redacts_urls(self):
+        # 1. Real service URLs are completely redacted
+        self.assertEqual(
+            gitea_audit._redact_str("failed for http://gitea.prgs.cc/api/v1/repos/x"),
+            "failed for [REDACTED_URL]"
+        )
+        # 2. Hostname is redacted even without full URL prefix
+        self.assertEqual(
+            gitea_audit._redact_str("error contacting gitea.prgs.cc directly"),
+            "error contacting [REDACTED_HOST] directly"
+        )
+        # 3. Synthetic test-only URLs are preserved
+        self.assertEqual(
+            gitea_audit._redact_str("mock URL: https://gitea.example.com/api/v1/repos/x"),
+            "mock URL: https://gitea.example.com/api/v1/repos/x"
+        )
+        # 4. Embedded credentials in synthetic URLs are redacted
+        self.assertEqual(
+            gitea_audit._redact_str("mock creds: https://user:[email protected]/path"),
+            "mock creds: https://[REDACTED_USER]:[REDACTED_PASS]@example.com/path"
+        )
+        # 5. Query secrets in synthetic URLs are redacted
+        self.assertEqual(
+            gitea_audit._redact_str("mock query: https://localhost:3003/api?token=abc-123&other=val"),
+            "mock query: https://localhost:3003/api?token=%5BREDACTED%5D&other=val"
+        )
+
 
 class TestBuildEvent(unittest.TestCase):
 

tests/test_pr_queue_inventory.py

@@ -209,6 +209,28 @@ class TestPRQueueInventory(unittest.TestCase):
         # uses project's redaction pattern (token prefix -> [REDACTED])
         # even if url leaks, the exception is redacted per pattern
 
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.get_profile")
+    def test_inventory_failure_output_redacts_real_service_urls(self, mock_get_profile, _auth, mock_api, mock_get_all):
+        """Failure paths must fully redact real service hostnames and URLs from inventory output."""
+        mock_get_profile.return_value = {
+            "profile_name": "gitea-author",
+            "allowed_operations": ["read"],
+            "forbidden_operations": [],
+            "base_url": None,
+        }
+        mock_get_all.side_effect = Exception("failed to connect to https://gitea.prgs.cc/api/v1/repos/x")
+        mock_api.return_value = {"login": "jcwalker3"}
+
+        result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs")
+        self.assertFalse(result["success"])
+        msg = result.get("message", "")
+        self.assertIn("=== PR Queue Inventory ===", msg)
+        self.assertIn("[REDACTED_URL]", msg)
+        self.assertNotIn("gitea.prgs.cc", msg)
+
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)