Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 139 additions & 0 deletions agent-governance-python/agent-os/docs/mcp-auth-tls.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
# MCP Auth Enforcement — TLS Gate

`McpAuthPolicy` enforces per-server authentication method allowlists for MCP
connections. When a server entry has `require_tls: true` (the default), the
policy gate validates that the connection URL uses a TLS-secured transport
(`https` or `wss`) before allowing the connection.

## URL resolution order

The TLS gate resolves the URL to check using the following priority:

1. **Caller-supplied URL** — the `url` argument passed to `check()`.
2. **Configured entry URL** — the `url` field on the `McpServerEntry`.
3. **No URL available** — both sources are empty.

When the caller does not supply a URL (or passes `""`), the gate falls back to
the configured `entry.url`. This prevents a caller from bypassing the TLS
requirement by simply omitting the URL argument.

When neither source provides a URL, the gate permits the connection. This
preserves spec S10.12 compatibility: `add_server()` with default `url=""` must
remain allowed for registrations where the URL is determined at connection time.

## Configuration

```python
from agent_os.mcp_auth_enforcement import McpAuthPolicy, McpServerEntry

policy = McpAuthPolicy(
servers=[
McpServerEntry(
name="finance-tools",
url="https://mcp.internal/finance",
allowed_auth_methods=["mtls"],
require_tls=True, # default
),
],
)

# Caller supplies url → checked directly
result = policy.check("finance-tools", auth_method="mtls",
url="https://mcp.internal/finance")
assert result.allowed

# Caller omits url → entry.url is checked
result = policy.check("finance-tools", auth_method="mtls")
assert result.allowed # entry.url is https

# Non-TLS entry.url is caught even without caller url
policy_http = McpAuthPolicy(
servers=[
McpServerEntry(
name="insecure",
url="http://mcp.internal/api",
allowed_auth_methods=["oauth2"],
require_tls=True,
),
],
)
result = policy_http.check("insecure", auth_method="oauth2")
assert not result.allowed # http scheme denied
```

## YAML configuration

```yaml
mcp_auth_policy:
deny_none: true
default_allowed_methods: [oauth2, mtls, bearer]
servers:
- name: finance-tools
url: https://mcp.internal/finance
allowed_auth_methods: [mtls]
require_tls: true
- name: dev-local
url: "" # S10.12: empty URL allowed
allowed_auth_methods: [oauth2]
require_tls: true
```

## Allowed TLS schemes

Only `https` and `wss` are treated as TLS-secured transports. All other
schemes — `http`, `ws`, `ftp`, `gopher`, bare hostnames without a scheme —
are rejected when `require_tls` is `true`.

## Security considerations

Prior to the fix for
[#3785](https://github.com/microsoft/agent-governance-toolkit/issues/3785),
the TLS gate only checked the caller-supplied `url` and never consulted
`entry.url`. When `url` was omitted (or passed as `""`), the TLS check was
skipped entirely, even when the server entry had `require_tls: true` and a
configured non-TLS URL.

The fix ensures the configured URL is always evaluated when no caller URL is
provided, closing the bypass while preserving backward compatibility for
entries that legitimately have no URL configured.

Note that the `url` parameter to `check()` is trusted input — it must be the
actual destination URL as resolved by the host, not a value supplied by an
untrusted caller. A caller that controls the `url` argument can claim any
scheme. The TLS gate validates the *stated* destination; verifying that the
connection actually reaches that destination is the transport layer's
responsibility.

## URL resolution truth table

| Caller URL | Entry URL | `require_tls` | Effective URL | Result |
|------------|-----------|---------------|---------------|--------|
| `https://…` | `http://…` | `true` | caller (`https`) | ✅ allowed |
| `http://…` | `https://…` | `true` | caller (`http`) | ❌ denied |
| _(empty)_ | `https://…` | `true` | entry (`https`) | ✅ allowed |
| _(empty)_ | `http://…` | `true` | entry (`http`) | ❌ denied |
| _(empty)_ | _(empty)_ | `true` | _(none)_ | ✅ allowed (S10.12) |
| _(any)_ | _(any)_ | `false` | _(skipped)_ | ✅ allowed |

The caller URL always takes precedence when supplied. When both are empty,
there is nothing to evaluate, so the gate permits the connection — this is
fail-open on a missing URL, not fail-open on TLS.

## Troubleshooting

**"Server 'X' requires TLS but URL scheme '' is not in the TLS allowlist"**

The entry has `require_tls: true` and a `url` with no scheme (e.g., a bare
hostname like `mcp.internal:8443`). Add the scheme explicitly:
`url: https://mcp.internal:8443`.

**"Server 'X' requires TLS but URL scheme 'http' is not in the TLS allowlist"**

The entry's configured URL or the caller-supplied URL uses `http`. Either
update the URL to `https`, or set `require_tls: false` if TLS is genuinely
not required for this server.

**Connection allowed but entry.url is http — why?**

If `require_tls` is `false`, the TLS gate is skipped entirely regardless of
the URL scheme. Check that `require_tls: true` is set on the server entry.
Original file line number Diff line number Diff line change
Expand Up @@ -125,8 +125,12 @@ def check(self, server_name: str, auth_method: str, url: str = "") -> AuthCheckR
Args:
server_name: Name of the MCP server.
auth_method: Authentication method being used.
url: Server URL used for TLS enforcement. When omitted, the
configured server entry URL is used.
url: The actual destination URL for this connection. When
``require_tls`` is ``true``, the scheme of this URL is
checked against the TLS allowlist (``https``, ``wss``).
If empty, the configured ``entry.url`` is used as
fallback. Pass the trusted, live destination — not an
unverified value from the caller.

Returns:
AuthCheckResult indicating whether the connection is allowed.
Expand Down
209 changes: 209 additions & 0 deletions agent-governance-python/agent-os/tests/test_mcp_auth_enforcement.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,3 +223,212 @@ def test_yaml_default_require_tls_false_disables_the_floor(self):
""")
result = policy.check("legacy-server", auth_method="oauth2", url="http://mcp.internal/tools")
assert result.allowed


# ---------------------------------------------------------------
# Parametrized TLS gate matrix
# Sweep all combinations of caller-URL scheme × entry-URL scheme ×
# require_tls. This is the "200-case matrix" carloshvp ran manually;
# pinning it in CI prevents regressions.
# ---------------------------------------------------------------

_TLS_SCHEMES = ("https", "wss")
_NON_TLS_SCHEMES = ("http", "ws", "ftp")


def _url(scheme: str) -> str:
"""Build a minimal valid URL for a scheme."""
if not scheme:
return ""
return f"{scheme}://mcp.internal/api"


class TestTlsGateMatrix:
"""Full caller-URL × entry-URL × require_tls sweep."""

@pytest.mark.parametrize("entry_scheme", _TLS_SCHEMES)
def test_tls_entry_no_caller_url_allowed(self, entry_scheme):
"""TLS entry.url + no caller url → allowed."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url=_url(entry_scheme),
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2")
assert result.allowed, f"entry={entry_scheme} should be allowed"

@pytest.mark.parametrize("entry_scheme", _NON_TLS_SCHEMES)
def test_non_tls_entry_no_caller_url_denied(self, entry_scheme):
"""Non-TLS entry.url + no caller url → denied."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url=_url(entry_scheme),
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2")
assert not result.allowed, f"entry={entry_scheme} should be denied"
assert "TLS" in result.reason

@pytest.mark.parametrize("caller_scheme", _TLS_SCHEMES)
@pytest.mark.parametrize("entry_scheme", _NON_TLS_SCHEMES)
def test_tls_caller_overrides_non_tls_entry(self, caller_scheme, entry_scheme):
"""TLS caller url wins over non-TLS entry.url → allowed."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url=_url(entry_scheme),
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2", url=_url(caller_scheme))
assert result.allowed

@pytest.mark.parametrize("caller_scheme", _NON_TLS_SCHEMES)
@pytest.mark.parametrize("entry_scheme", _TLS_SCHEMES)
def test_non_tls_caller_overrides_tls_entry(self, caller_scheme, entry_scheme):
"""Non-TLS caller url wins over TLS entry.url → denied."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url=_url(entry_scheme),
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2", url=_url(caller_scheme))
assert not result.allowed

@pytest.mark.parametrize("entry_scheme", list(_TLS_SCHEMES) + list(_NON_TLS_SCHEMES) + [""])
def test_require_tls_false_always_allows(self, entry_scheme):
"""require_tls=False + any scheme → allowed."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url=_url(entry_scheme) if entry_scheme else "",
allowed_auth_methods=["oauth2"], require_tls=False),
])
result = policy.check("s", "oauth2")
assert result.allowed

def test_both_empty_urls_allowed_s10_12(self):
"""S10.12: both caller and entry url empty → allowed (no URL to check)."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url="",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2", url="")
assert result.allowed


class TestTlsGateEdgeCases:
"""Edge cases not covered by the scheme matrix."""

def test_entry_url_bare_hostname_no_scheme_denied(self):
"""A bare hostname (no scheme) is denied under require_tls."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url="mcp.internal:8443",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2")
assert not result.allowed

def test_entry_url_uppercase_scheme_normalized(self):
"""Scheme comparison is case-insensitive."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url="HTTPS://mcp.internal/api",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2")
assert result.allowed

def test_caller_url_whitespace_treated_as_present(self):
"""A caller URL containing only whitespace still has an empty scheme → denied."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url="https://mcp.internal",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2", url=" ")
assert not result.allowed

def test_reason_message_includes_scheme(self):
"""Denial reason must include the offending scheme for debugging."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="s", url="http://mcp.internal",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("s", "oauth2")
assert "http" in result.reason.lower()
assert "TLS" in result.reason

def test_reason_message_includes_server_name(self):
"""Denial reason must include the server name."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="finance-db", url="http://mcp.internal",
allowed_auth_methods=["oauth2"], require_tls=True),
])
result = policy.check("finance-db", "oauth2")
assert "finance-db" in result.reason

def test_add_then_check_entry_url_fallback(self):
"""Dynamic add_server also benefits from entry.url fallback."""
policy = McpAuthPolicy()
policy.add_server(McpServerEntry(
name="dynamic", url="http://mcp.internal",
allowed_auth_methods=["oauth2"], require_tls=True,
))
assert not policy.check("dynamic", "oauth2").allowed

def test_remove_server_clears_entry(self):
"""After remove_server, server falls back to default policy."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="tmp", url="http://bad",
allowed_auth_methods=["oauth2"], require_tls=True),
])
assert not policy.check("tmp", "oauth2").allowed
policy.remove_server("tmp")
# Falls back to default (oauth2 is in default allowed)
assert policy.check("tmp", "oauth2").allowed

def test_multiple_servers_independent(self):
"""TLS decision for one server does not affect another."""
policy = McpAuthPolicy(servers=[
McpServerEntry(name="secure", url="https://a.internal",
allowed_auth_methods=["oauth2"], require_tls=True),
McpServerEntry(name="insecure", url="http://b.internal",
allowed_auth_methods=["oauth2"], require_tls=True),
])
assert policy.check("secure", "oauth2").allowed
assert not policy.check("insecure", "oauth2").allowed


class TestTlsGateYamlIntegration:
"""YAML round-trip tests for the TLS gate."""

def test_yaml_mixed_tls_servers(self):
policy = McpAuthPolicy.from_yaml("""
mcp_auth_policy:
servers:
- name: prod-api
url: https://api.prod.internal
allowed_auth_methods: [mtls]
require_tls: true
- name: staging-api
url: http://api.staging.internal
allowed_auth_methods: [oauth2]
require_tls: true
- name: dev-local
url: ""
allowed_auth_methods: [oauth2]
require_tls: true
- name: monitoring
url: http://metrics.internal
allowed_auth_methods: [api_key]
require_tls: false
""")
assert policy.check("prod-api", "mtls").allowed
assert not policy.check("staging-api", "oauth2").allowed
assert policy.check("dev-local", "oauth2").allowed # S10.12
assert policy.check("monitoring", "api_key").allowed # require_tls=false

def test_yaml_caller_url_overrides_configured(self):
policy = McpAuthPolicy.from_yaml("""
mcp_auth_policy:
servers:
- name: flexible
url: http://mcp.internal
allowed_auth_methods: [oauth2]
require_tls: true
""")
# Entry is http → denied without caller url
assert not policy.check("flexible", "oauth2").allowed
# Caller overrides with https → allowed
assert policy.check("flexible", "oauth2", url="https://mcp.secure.com").allowed
Loading