fix(agent-os): apply the require_tls gate to the configured server URL

agent-governance-python/agent-os/src/agent_os/mcp_auth_enforcement.py

@@ -43,6 +43,9 @@
 # Supported auth methods (aligns with 2026 MCP roadmap)
 VALID_AUTH_METHODS = {"oauth2", "mtls", "api_key", "bearer", "none"}
 
+# Only these URL schemes represent an actual TLS-secured transport.
+TLS_SCHEMES = frozenset({"https", "wss"})
+
 
 @dataclass
 class McpServerEntry:
@@ -112,13 +115,48 @@ def remove_server(self, name: str) -> bool:
         """Remove a server from the allowlist."""
         return self._servers.pop(name, None) is not None
 
+    @staticmethod
+    def _tls_violation(url: str) -> str | None:
+        """Return why *url* is not TLS-secured, or ``None`` if it is.
+
+        Scheme-based rather than prefix-based: ``ftp://``, ``ws://`` and a URL
+        with no scheme at all are all non-TLS, and only ``https``/``wss`` are
+        actual TLS transports.
+
+        A non-empty URL with **no scheme** is a violation rather than a pass. It
+        is the shape a bare host or a glob pattern takes (``mcp.internal``,
+        ``*.internal/finance``), and there is no way to tell from it whether the
+        connection will be encrypted -- so a TLS requirement cannot be
+        considered satisfied. Callers decide whether to invoke this at all for
+        an empty URL.
+
+        The returned string never embeds the URL itself. It reaches
+        ``AuthCheckResult.reason``, which audit backends promote to attributes
+        (``otel_audit_backend`` maps it to ``agt.audit.reason``), and a server
+        URL can carry a token in its userinfo or query string. The scheme is
+        the only part needed to explain the decision, and it is not secret.
+        """
+        try:
+            scheme = urlparse(url).scheme.lower()
+        except (ValueError, AttributeError):
+            scheme = ""
+        if scheme in TLS_SCHEMES:
+            return None
+        allowlist = ", ".join(sorted(TLS_SCHEMES))
+        if not scheme:
+            return f"the URL has no scheme, so TLS cannot be verified (need {allowlist})"
+        return f"URL scheme {scheme!r} is not in the TLS allowlist ({allowlist})"
+
     def check(self, server_name: str, auth_method: str, url: str = "") -> AuthCheckResult:
         """Check if an auth method is allowed for the given MCP server.
 
         Args:
             server_name: Name of the MCP server.
             auth_method: Authentication method being used.
-            url: Server URL (for logging/audit).
+            url: Server URL for the connection being checked. When omitted, the
+                URL registered on the matching :class:`McpServerEntry` is used
+                instead, so a server configured with a plaintext URL is blocked
+                whether or not the caller repeats that URL here.
 
         Returns:
             AuthCheckResult indicating whether the connection is allowed.
@@ -146,26 +184,31 @@ def check(self, server_name: str, auth_method: str, url: str = "") -> AuthCheckR
         entry = self._servers.get(server_name)
         if entry:
             if auth_method in entry.allowed_auth_methods:
-                # TLS check: parse the URL's scheme rather than testing
-                # for the literal "http://" prefix. Previously, a URL
-                # using `ftp://`, `ws://`, `gopher://`, or no scheme at
-                # all bypassed the require_tls gate because none of
-                # them start with "http://" — only `https://` and
-                # `wss://` represent actual TLS-secured transports.
-                if entry.require_tls and url:
-                    try:
-                        scheme = urlparse(url).scheme.lower()
-                    except (ValueError, AttributeError):
-                        scheme = ""
-                    if scheme not in {"https", "wss"}:
+                # TLS check: the scheme is parsed and matched against the
+                # allowlist below, so only `https://` and `wss://` count as
+                # TLS-secured transports.
+                #
+                # The URL to check falls back to the one registered on the
+                # entry. `url` defaults to "" and the gate used to be skipped
+                # entirely when it was falsy, so a server configured with a
+                # plaintext URL and require_tls=True was allowed by any caller
+                # that did not repeat the URL -- including every caller on the
+                # from_yaml path, where the URL lives in config and the call
+                # site has no reason to pass it again.
+                #
+                # When neither the caller nor the entry supplies a URL there is
+                # nothing to evaluate, and the gate stays skipped as before.
+                # Denying that case would be a separate policy decision about
+                # every server registered without a URL, not this fix.
+                effective_url = url or entry.url
+                if entry.require_tls and effective_url:
+                    violation = self._tls_violation(effective_url)
+                    if violation is not None:
                         return AuthCheckResult(
                             allowed=False,
                             server_name=server_name,
                             auth_method=auth_method,
-                            reason=(
-                                f"Server '{server_name}' requires TLS but URL scheme "
-                                f"{scheme!r} is not in the TLS allowlist (https, wss)"
-                            ),
+                            reason=f"Server '{server_name}' requires TLS but {violation}",
                         )
                 return AuthCheckResult(
                     allowed=True,

agent-governance-python/agent-os/tests/test_mcp_auth_enforcement.py

@@ -107,6 +107,105 @@ def test_result_fields(self):
         assert len(result.reason) > 0
 
 
+class TestTlsGateUsesTheConfiguredUrl:
+    """``require_tls`` must hold for the URL the server is registered with.
+
+    The gate only ran when the *caller* passed a ``url``, which defaults to
+    ``""``. So a server configured with a plaintext URL and ``require_tls=True``
+    was allowed by any caller that did not repeat that URL -- including every
+    caller on the ``from_yaml`` path, where the URL lives in config and the call
+    site has no reason to pass it again.
+    """
+
+    @staticmethod
+    def _policy(url: str, **kwargs: object) -> McpAuthPolicy:
+        return McpAuthPolicy(
+            servers=[
+                McpServerEntry(name="finance", url=url, allowed_auth_methods=["mtls"], **kwargs),
+            ]
+        )
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://mcp.internal/finance",
+            "ws://mcp.internal/finance",
+            "ftp://mcp.internal/finance",
+            "gopher://mcp.internal",
+            # No scheme: the shape a bare host or a glob pattern takes. Nothing
+            # here says the transport is encrypted, so TLS is not satisfied.
+            "mcp.internal/finance",
+            "*.internal/finance",
+        ],
+    )
+    def test_plaintext_configured_url_is_denied_without_a_caller_url(self, url: str) -> None:
+        result = self._policy(url).check("finance", auth_method="mtls")
+        assert not result.allowed
+        assert "requires TLS" in result.reason
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://user:s3cret@mcp.internal/finance",
+            "http://mcp.internal/finance?access_token=s3cret",
+            # No scheme, so the whole string would have been echoed verbatim.
+            "mcp.internal/finance?access_token=s3cret",
+        ],
+    )
+    def test_denial_reason_does_not_echo_the_url(self, url: str) -> None:
+        # `reason` is promoted to audit attributes (otel_audit_backend maps it
+        # to agt.audit.reason), and a server URL can carry a token in its
+        # userinfo or query string. The scheme alone explains the decision.
+        result = self._policy(url).check("finance", auth_method="mtls")
+        assert not result.allowed
+        assert "s3cret" not in result.reason
+        assert url not in result.reason
+
+    @pytest.mark.parametrize("url", ["https://mcp.internal/finance", "wss://mcp.internal/ws"])
+    def test_tls_configured_url_is_allowed(self, url: str) -> None:
+        assert self._policy(url).check("finance", auth_method="mtls").allowed
+
+    def test_scheme_case_is_ignored(self) -> None:
+        assert self._policy("HTTPS://MCP.INTERNAL").check("finance", auth_method="mtls").allowed
+
+    def test_caller_url_still_wins_over_the_configured_one(self) -> None:
+        # The caller knows the URL actually being dialed; a plaintext dial
+        # against an https-registered server must still be blocked.
+        policy = self._policy("https://mcp.internal/finance")
+        assert policy.check("finance", auth_method="mtls").allowed
+        assert not policy.check(
+            "finance", auth_method="mtls", url="http://mcp.internal/finance"
+        ).allowed
+
+    def test_no_url_anywhere_leaves_the_gate_skipped(self) -> None:
+        # Unchanged behaviour: with no URL from either source there is nothing
+        # to evaluate. Denying this would be a separate policy decision about
+        # every server registered without a URL.
+        assert self._policy("").check("finance", auth_method="mtls").allowed
+
+    def test_require_tls_false_allows_plaintext(self) -> None:
+        policy = self._policy("http://mcp.internal/finance", require_tls=False)
+        assert policy.check("finance", auth_method="mtls").allowed
+
+    def test_auth_method_denial_is_not_masked_by_the_tls_gate(self) -> None:
+        # A method outside the allowlist must still be reported as such, not as
+        # a TLS problem — the allowlist check comes first.
+        result = self._policy("http://mcp.internal/finance").check("finance", auth_method="oauth2")
+        assert not result.allowed
+        assert "not allowed for server" in result.reason
+
+    def test_yaml_configured_plaintext_server_is_denied(self) -> None:
+        policy = McpAuthPolicy.from_yaml("""
+mcp_auth_policy:
+  servers:
+    - name: finance-tools
+      url: http://mcp.internal/finance
+      allowed_auth_methods: [mtls]
+      require_tls: true
+""")
+        assert not policy.check("finance-tools", auth_method="mtls").allowed
+
+
 class TestFromYaml:
     def test_parse_yaml(self):
         policy = McpAuthPolicy.from_yaml("""