fix(#6844): support wildcard hostnames in SSRF hook egress allowlist

docs/contributing/runtime-implementation.md

@@ -225,7 +225,7 @@ Also: empty/whitespace-only stdin is treated as "no tool call" and allowed by ev
 | Variable | Read by | Behaviour |
 |----------|---------|-----------|
 | `TIRITH_FAIL_ON`, `TIRITH_REQUIRED` | `tirith_check.py` | written by `appendHookEnv`; `TIRITH_REQUIRED=1` turns the fail-open into fail-closed |
-| `FULLSEND_EGRESS_ALLOWLIST` | `ssrf_pretool.py` | comma-separated `host:port` entries, exact hostnames only — wildcards are skipped with a warning on stderr; on DNS failure the hook defers to the L7 proxy for allowlisted hosts instead of failing closed; if DNS succeeds but resolves to a blocked IP, the allowlist is not consulted |
+| `FULLSEND_EGRESS_ALLOWLIST` | `ssrf_pretool.py` | comma-separated `host:port` entries; supports exact hostnames and leading-wildcard patterns (e.g. `*.example.com:443`) matched by domain-anchored suffix comparison — bare `*` and mid-string globs are rejected with a warning on stderr; on DNS failure the hook defers to the L7 proxy for allowlisted hosts instead of failing closed; if DNS succeeds but resolves to a blocked IP, the allowlist is not consulted |
 | `FULLSEND_TOOL_ALLOWLIST` | `tool_allowlist_pretool.py` | fail-closed when unset |
 | `FULLSEND_CANARY_TOKEN` | both canary hooks | no-ops when empty; supply it via harness `env.sandbox`/`host_files` |
 | `FULLSEND_TRACE_ID` | all scripts | correlates findings with the run |

internal/security/hooks/ssrf_pretool.py

@@ -136,11 +136,36 @@ def _parse_egress_allowlist() -> set[tuple[str, int]]:
         if not entry:
             continue
         if "*" in entry:
-            print(
-                f"WARNING: wildcard entry '{entry}' in FULLSEND_EGRESS_ALLOWLIST "
-                "is not supported and will be ignored — use exact hostnames",
-                file=sys.stderr,
-            )
+            # Accept leading wildcard entries like *.domain.com:port.
+            # Reject bare '*' and mid-string globs (e.g. 'atl*an.net').
+            if ":" in entry:
+                wc_host, _, wc_port_str = entry.rpartition(":")
+                try:
+                    wc_port = int(wc_port_str)
+                except ValueError:
+                    print(
+                        f"WARNING: malformed port in '{entry}' in FULLSEND_EGRESS_ALLOWLIST "
+                        "— entry ignored",
+                        file=sys.stderr,
+                    )
+                    continue
+            else:
+                wc_host = entry
+                wc_port = 0
+            wc_host_clean = wc_host.lower().rstrip(".")
+            if (
+                wc_host_clean.startswith("*.")
+                and len(wc_host_clean) > 2
+                and "*" not in wc_host_clean[2:]
+            ):
+                entries.add((wc_host_clean, wc_port))
+            else:
+                print(
+                    f"WARNING: wildcard entry '{entry}' in FULLSEND_EGRESS_ALLOWLIST "
+                    "is not supported and will be ignored — use exact hostnames "
+                    "or leading wildcard patterns like *.domain.com",
+                    file=sys.stderr,
+                )
             continue
         if ":" in entry:
             host, _, port_str = entry.rpartition(":")
@@ -173,7 +198,20 @@ def _is_host_allowlisted(hostname: str, port: int | None) -> bool:
     if port is not None and (hostname, port) in allowlist:
         return True
     # Check host-only match (port 0 sentinel means any port).
-    return (hostname, 0) in allowlist
+    if (hostname, 0) in allowlist:
+        return True
+    # Check wildcard entries: *.domain matches any subdomain of domain
+    # but not the bare domain itself (require at least one subdomain label).
+    for entry_host, entry_port in allowlist:
+        if not entry_host.startswith("*."):
+            continue
+        # *.atlassian.net → suffix ".atlassian.net"
+        suffix = entry_host[1:]
+        if hostname.endswith(suffix) and (
+            entry_port == 0 or (port is not None and entry_port == port)
+        ):
+            return True
+    return False
 
 
 def log_finding(scanner: str, name: str, severity: str, detail: str, action: str):

internal/security/hooks/ssrf_pretool_test.py

@@ -1047,16 +1047,57 @@ def test_ipv6_full_address_brackets_stripped(self, hook):
             result = hook._parse_egress_allowlist()
             assert ("2001:db8::1", 8443) in result
 
-    def test_wildcard_entries_ignored(self, hook, capsys):
+    def test_leading_wildcard_entries_accepted(self, hook):
         with mock.patch.dict(
             os.environ,
             {"FULLSEND_EGRESS_ALLOWLIST": "*.internal:443,exact.host:8443"},
+        ):
+            result = hook._parse_egress_allowlist()
+            assert len(result) == 2
+            assert ("*.internal", 443) in result
+            assert ("exact.host", 8443) in result
+
+    def test_bare_wildcard_rejected(self, hook, capsys):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*:443,exact.host:8443"},
         ):
             result = hook._parse_egress_allowlist()
             assert len(result) == 1
             assert ("exact.host", 8443) in result
             captured = capsys.readouterr()
-            assert "wildcard entry '*.internal:443'" in captured.err
+            assert "wildcard entry '*:443'" in captured.err
+
+    def test_mid_string_glob_rejected(self, hook, capsys):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "atl*an.net:443,exact.host:8443"},
+        ):
+            result = hook._parse_egress_allowlist()
+            assert len(result) == 1
+            assert ("exact.host", 8443) in result
+            captured = capsys.readouterr()
+            assert "wildcard entry 'atl*an.net:443'" in captured.err
+
+    def test_leading_wildcard_no_port(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net"},
+        ):
+            result = hook._parse_egress_allowlist()
+            assert ("*.atlassian.net", 0) in result
+
+    def test_trailing_dot_wildcard_collapses_rejected(self, hook, capsys):
+        """*..:443 must not silently become ('*', 443) after rstrip('.')."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*..:443,exact.host:8443"},
+        ):
+            result = hook._parse_egress_allowlist()
+            assert len(result) == 1
+            assert ("exact.host", 8443) in result
+            captured = capsys.readouterr()
+            assert "wildcard entry '*..:443'" in captured.err
 
     def test_malformed_port_warns(self, hook, capsys):
         with mock.patch.dict(
@@ -1071,6 +1112,89 @@ def test_malformed_port_warns(self, hook, capsys):
             assert "host.internal:notaport" in captured.err
 
 
+class TestWildcardAllowlistMatching:
+    """Verify wildcard suffix matching in _is_host_allowlisted."""
+
+    def test_wildcard_matches_subdomain(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+        ):
+            assert hook._is_host_allowlisted("redhat.atlassian.net", 443) is True
+
+    def test_wildcard_matches_multi_level_subdomain(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+        ):
+            assert hook._is_host_allowlisted("sub.redhat.atlassian.net", 443) is True
+
+    def test_wildcard_does_not_match_base_domain(self, hook):
+        """*.atlassian.net must not match atlassian.net (no subdomain)."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+        ):
+            assert hook._is_host_allowlisted("atlassian.net", 443) is False
+
+    def test_wildcard_does_not_match_suffix_spoof(self, hook):
+        """*.atlassian.net must not match atlassian.net.evil.com."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+        ):
+            assert hook._is_host_allowlisted("atlassian.net.evil.com", 443) is False
+
+    def test_wildcard_port_must_match(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+        ):
+            assert hook._is_host_allowlisted("redhat.atlassian.net", 8080) is False
+
+    def test_wildcard_port_zero_matches_any(self, hook):
+        """Wildcard entry without port matches any port."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net"},
+        ):
+            assert hook._is_host_allowlisted("redhat.atlassian.net", 443) is True
+            assert hook._is_host_allowlisted("redhat.atlassian.net", 8080) is True
+
+    def test_bare_wildcard_does_not_match(self, hook, capsys):
+        """Bare * must not match any hostname."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*:443"},
+        ):
+            assert hook._is_host_allowlisted("anything.com", 443) is False
+            # Consume warning output
+            capsys.readouterr()
+
+    def test_mid_string_glob_does_not_match(self, hook, capsys):
+        """Mid-string globs like atl*an.net must not match."""
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "atl*an.net:443"},
+        ):
+            assert hook._is_host_allowlisted("atlassian.net", 443) is False
+            capsys.readouterr()
+
+    def test_wildcard_case_insensitive(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.Atlassian.NET:443"},
+        ):
+            assert hook._is_host_allowlisted("Redhat.ATLASSIAN.net", 443) is True
+
+    def test_wildcard_trailing_dot_stripped(self, hook):
+        with mock.patch.dict(
+            os.environ,
+            {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net.:443"},
+        ):
+            assert hook._is_host_allowlisted("redhat.atlassian.net", 443) is True
+
+
 class TestEgressAllowlistValidateUrl:
     """Verify validate_url respects the egress allowlist on DNS failure."""
 
@@ -1195,3 +1319,44 @@ def test_ipv6_bracket_allowlist_entry_matches(self, hook):
         ):
             result = hook.validate_url(url)
             assert result is None  # allowed
+
+    def test_wildcard_allowlist_allows_subdomain_on_dns_failure(self, hook):
+        """Wildcard *.atlassian.net:443 allows redhat.atlassian.net when DNS fails."""
+        url = "https://redhat.atlassian.net/rest/api/2/issue/PROJ-1"
+        with (
+            mock.patch("socket.getaddrinfo", side_effect=socket.gaierror("no DNS")),
+            mock.patch.dict(
+                os.environ,
+                {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+            ),
+        ):
+            result = hook.validate_url(url)
+            assert result is None  # allowed
+
+    def test_wildcard_allowlist_blocks_base_domain_on_dns_failure(self, hook):
+        """Wildcard *.atlassian.net must not match the bare domain atlassian.net."""
+        url = "https://atlassian.net/something"
+        with (
+            mock.patch("socket.getaddrinfo", side_effect=socket.gaierror("no DNS")),
+            mock.patch.dict(
+                os.environ,
+                {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+            ),
+        ):
+            result = hook.validate_url(url)
+            assert result is not None
+            assert "fail-closed" in result
+
+    def test_wildcard_allowlist_blocks_wrong_port_on_dns_failure(self, hook):
+        """Wildcard on port 443 must not match port 8080."""
+        url = "https://redhat.atlassian.net:8080/api"
+        with (
+            mock.patch("socket.getaddrinfo", side_effect=socket.gaierror("no DNS")),
+            mock.patch.dict(
+                os.environ,
+                {"FULLSEND_EGRESS_ALLOWLIST": "*.atlassian.net:443"},
+            ),
+        ):
+            result = hook.validate_url(url)
+            assert result is not None
+            assert "fail-closed" in result