fix: stop normalising resource and issuer identifiers

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+- Resource and issuer identifiers are never rewritten (RFC 8414 §3.3 / RFC 9728 §3.3 require the advertised value to be *identical* to the configured one). `build_prm_url` and `build_metadata_url` are formed by pure insertion, preserving the identifier's path exactly — including any trailing slash — and AS-metadata issuer comparison is now an exact string match. Identifiers are validated at construction (absolute http(s) URL with an authority and no fragment) and raise `ValueError` otherwise; trailing slashes, host case, and explicit ports are legal and preserved. **Migration**: if your configured issuer or resource differs from your authorization server's actual identifier by a trailing slash, correct the config — the SDK no longer silently reconciles them.
+
+### Fixed
+- A configured issuer whose identifier legitimately ends in `/` no longer has every token rejected. The trailing slash was silently stripped at client creation and the stripped value compared against the token's `iss`, which RFC 9068 requires to carry the slash; discovery now also resolves the RFC 8414 well-known URL for such issuers correctly.
+
 ## [0.3.0] - 2026-07-21
 
 ### Added

README.md

@@ -20,6 +20,7 @@ from authplane_fastmcp import authplane_auth
 from fastmcp import FastMCP
 from fastmcp.server.auth import require_scopes
 
+
 async def main() -> None:
     result = await authplane_auth(
         issuer="https://auth.company.com",
@@ -37,6 +38,7 @@ async def main() -> None:
     finally:
         await result.aclose()
 
+
 asyncio.run(main())
 ```
 

authplane-fastmcp/README.md

@@ -32,9 +32,7 @@ async def main():
     )
 
     @mcp.tool(auth=require_scopes("tools/query"))
-    async def query_database(
-        query: str, token: AccessToken = CurrentAccessToken()
-    ) -> str:
+    async def query_database(query: str, token: AccessToken = CurrentAccessToken()) -> str:
         user_id = token.claims.get("sub")
         return f"Query: {query}, User: {user_id}"
 

authplane-fastmcp/docs/user-guide.md

@@ -37,6 +37,7 @@ import asyncio
 from fastmcp import FastMCP
 from authplane_fastmcp import authplane_auth
 
+
 async def main() -> None:
     result = await authplane_auth(
         issuer="https://auth.company.com",
@@ -55,6 +56,7 @@ async def main() -> None:
     finally:
         await result.aclose()
 
+
 asyncio.run(main())
 ```
 
@@ -98,11 +100,13 @@ Use FastMCP's built-in `require_scopes` decorator to enforce per-tool scope requ
 ```python
 from fastmcp.server.auth import require_scopes
 
+
 @mcp.tool(auth=require_scopes("tools/query"))
 def query(sql: str) -> str:
     """Requires the tools/query scope."""
     return f"Ran: {sql}"  # replace with your real handler
 
+
 @mcp.tool(auth=require_scopes("tools/admin", "tools/delete"))
 def delete_all() -> str:
     """Requires BOTH tools/admin AND tools/delete scopes."""
@@ -119,21 +123,22 @@ FastMCP enforces scopes **before** the handler runs by **filtering tools the cal
 from fastmcp.dependencies import CurrentAccessToken
 from fastmcp.server.auth import AccessToken
 
+
 @mcp.tool()
 async def my_tool(data: str, token: AccessToken = CurrentAccessToken()) -> str:
     # Standard JWT claims
-    sub = token.claims.get("sub")         # Subject (user ID)
-    jti = token.claims.get("jti")         # JWT ID
-    iss = token.claims.get("iss")         # Issuer
-    aud = token.claims.get("aud")         # Audience
-    exp = token.claims.get("exp")         # Expiration (Unix timestamp)
-    nbf = token.claims.get("nbf")         # Not before
-    iat = token.claims.get("iat")         # Issued at
+    sub = token.claims.get("sub")  # Subject (user ID)
+    jti = token.claims.get("jti")  # JWT ID
+    iss = token.claims.get("iss")  # Issuer
+    aud = token.claims.get("aud")  # Audience
+    exp = token.claims.get("exp")  # Expiration (Unix timestamp)
+    nbf = token.claims.get("nbf")  # Not before
+    iat = token.claims.get("iat")  # Issued at
 
     # OAuth claims
-    client_id = token.client_id           # Client ID
-    scopes = token.scopes                 # List of granted scopes
-    expires_at = token.expires_at         # Expiration (Unix timestamp)
+    client_id = token.client_id  # Client ID
+    scopes = token.scopes  # List of granted scopes
+    expires_at = token.expires_at  # Expiration (Unix timestamp)
 
     # Custom claims
     tenant = token.claims.get("tenant_id")
@@ -149,6 +154,7 @@ The `claims` dict contains the **full JWT payload** including all standard and c
 ```python
 from fastmcp.server.dependencies import get_access_token
 
+
 @mcp.tool()
 async def my_tool(data: str) -> str:
     token = get_access_token()  # Returns None if unauthenticated
@@ -220,10 +226,12 @@ Implement your own revocation logic with an async callable:
 ```python
 from authplane import VerifiedClaims
 
+
 async def check_blocklist(claims: VerifiedClaims, raw_token: str) -> bool:
     """Return True to reject the token (it is revoked)."""
     return await redis_client.sismember("revoked_tokens", claims.jti)
 
+
 await authplane_auth(
     issuer="https://auth.company.com",
     base_url="https://mcp.company.com",
@@ -253,8 +261,8 @@ result = await authplane_auth(
 downstream = await result.client.exchange(
     TokenExchangeOptions(
         subject_token=inbound_token,
-        scope="tools/add",                           # narrow to the minimum
-        resources=("https://downstream.example",),   # RFC 8707 audience binding
+        scope="tools/add",  # narrow to the minimum
+        resources=("https://downstream.example",),  # RFC 8707 audience binding
     )
 )
 
@@ -287,6 +295,7 @@ from authplane import ConsentRequiredError
 from authplane.oauth import TokenExchangeOptions
 from mcp.shared.exceptions import UrlElicitationRequiredError
 
+
 @mcp.tool(auth=require_scopes("tools/call_downstream"))
 async def call_downstream(payload: str) -> str:
     try:
@@ -379,6 +388,7 @@ When `fetch_settings` is provided, `dev_mode` is ignored for both metadata and J
 ```python
 import asyncio
 
+
 async def main() -> None:
     result = await authplane_auth(...)
     try:
@@ -387,6 +397,7 @@ async def main() -> None:
     finally:
         await result.aclose()
 
+
 asyncio.run(main())
 ```
 

authplane-mcp/docs/user-guide.md

@@ -101,12 +101,14 @@ Use the `require_scope()` helper at the top of tool handlers to enforce per-tool
 ```python
 from authplane_mcp import require_scope
 
+
 @mcp.tool()
 async def query(sql: str) -> str:
     """Requires the tools/query scope."""
     require_scope("tools/query")
     return f"Ran: {sql}"  # replace with your real handler
 
+
 @mcp.tool()
 async def delete_all() -> str:
     """Requires the tools/admin scope."""
@@ -127,14 +129,15 @@ Use the MCP SDK's `get_access_token()` to access the validated token in tool han
 ```python
 from mcp.server.auth.middleware.auth_context import get_access_token
 
+
 @mcp.tool()
 async def my_tool(data: str) -> str:
     token = get_access_token()
     if token:
-        client_id = token.client_id       # Client ID
-        scopes = token.scopes             # List of granted scopes
-        expires_at = token.expires_at     # Expiration (Unix timestamp)
-        resource = token.resource         # Resource (audience) URL
+        client_id = token.client_id  # Client ID
+        scopes = token.scopes  # List of granted scopes
+        expires_at = token.expires_at  # Expiration (Unix timestamp)
+        resource = token.resource  # Resource (audience) URL
     return f"Processing {data}"
 ```
 
@@ -211,10 +214,12 @@ Implement your own revocation logic with an async callable:
 ```python
 from authplane import VerifiedClaims
 
+
 async def check_blocklist(claims: VerifiedClaims, raw_token: str) -> bool:
     """Return True to reject the token (it is revoked)."""
     return await redis_client.sismember("revoked_tokens", claims.jti)
 
+
 await authplane_mcp_auth(
     issuer="https://auth.company.com",
     resource="https://mcp.company.com",
@@ -244,8 +249,8 @@ result = await authplane_mcp_auth(
 downstream = await result.client.exchange(
     TokenExchangeOptions(
         subject_token=inbound_token,
-        scope="tools/add",                           # narrow to the minimum
-        resources=("https://downstream.example",),   # RFC 8707 audience binding
+        scope="tools/add",  # narrow to the minimum
+        resources=("https://downstream.example",),  # RFC 8707 audience binding
     )
 )
 
@@ -278,6 +283,7 @@ The adapter handles this for you. The `client` returned by `authplane_mcp_auth(.
 ```python
 from authplane.oauth import TokenExchangeOptions
 
+
 @mcp.tool()
 async def call_downstream(user_token: str, payload: str) -> str:
     downstream = await result.client.exchange(
@@ -379,6 +385,7 @@ When `fetch_settings` is provided, `dev_mode` is ignored for both metadata and J
 ```python
 import asyncio
 
+
 async def main() -> None:
     auth_result = await authplane_mcp_auth(...)
     try:
@@ -387,6 +394,7 @@ async def main() -> None:
     finally:
         await auth_result.aclose()
 
+
 asyncio.run(main())
 ```
 

authplane/client.py

@@ -17,6 +17,7 @@
     JWKSCache,
     MetadataCache,
     build_metadata_url,
+    validate_identifier,
 )
 from .net import FetchSettings
 from .net.ssrf import SSRFError
@@ -134,7 +135,10 @@ async def create(
                 circuit trips. Default 30s.
         """
         client = cls()
-        client._issuer = issuer.rstrip("/")
+        # RFC 8414 Section 3.3 — the issuer is an opaque identifier compared
+        # with simple string equality (against metadata and token iss); it is
+        # validated but never rewritten.
+        client._issuer = validate_identifier(issuer, "issuer")
 
         # Dev mode
         resolved_dev_mode = (

authplane/docs/user-guide.md

@@ -137,11 +137,11 @@ res = client.resource(
     resource="https://api.example.com",
     scopes=["read"],
     inbound_dpop=InboundDPoPOptions(
-        replay_store=InMemoryDPoPReplayStore(),     # process-scoped by default
+        replay_store=InMemoryDPoPReplayStore(),  # process-scoped by default
         max_proof_age_seconds=300,
         clock_skew_seconds=30,
         allowed_proof_algorithms=("RS256", "ES256"),
-        required=True,                              # reject bearer-only tokens
+        required=True,  # reject bearer-only tokens
     ),
 )
 ```
@@ -159,13 +159,16 @@ For each incoming request that may carry a DPoP-bound token, build a
 ```python
 from dataclasses import dataclass
 
+
 @dataclass
 class IncomingRequest:
     """Implements DPoPRequestContext."""
+
     method: str
     url: str
     proof: str | None
 
+
 claims = await res.verify(
     token,
     dpop_request=IncomingRequest(
@@ -261,6 +264,7 @@ res = client.resource(
 ```python
 from authplane import VerifiedClaims
 
+
 async def my_revocation_checker(claims: VerifiedClaims, raw_token: str) -> bool:
     return claims.jti in revoked_jtis
 ```
@@ -415,12 +419,12 @@ For multi-instance or shared-state deployments, provide your own `DPoPNonceStore
 ```python
 from authplane import DPoPKeyMaterial, DPoPNonceStore, DPoPProvider
 
+
 class MyNonceStore:
-    def get(self, key: str) -> str:
-        ...
+    def get(self, key: str) -> str: ...
+
+    def put(self, key: str, nonce: str) -> None: ...
 
-    def put(self, key: str, nonce: str) -> None:
-        ...
 
 provider = DPoPProvider(
     DPoPKeyMaterial.from_pem(private_key_pem),
@@ -597,8 +601,8 @@ except AuthplaneError as e:
 Generate an RFC 9728 protected resource metadata document with:
 
 ```python
-prm = res.prm_response()      # the document body (a dict)
-url = res.prm_url()           # the well-known URL where clients can fetch it
+prm = res.prm_response()  # the document body (a dict)
+url = res.prm_url()  # the well-known URL where clients can fetch it
 ```
 
 Example output:

authplane/internal/__init__.py

@@ -9,6 +9,7 @@
 )
 from .document_fetcher import DocumentFetcher
 from .fetch_result import FetchResult
+from .identifiers import validate_identifier
 from .metadata import MetadataCache
 from .urls import build_metadata_url, build_prm_url
 
@@ -23,4 +24,5 @@
     "build_metadata_url",
     "build_prm_url",
     "parse_expires_at",
+    "validate_identifier",
 ]

authplane/internal/identifiers.py

@@ -0,0 +1,33 @@
+"""Validation for resource and issuer identifiers.
+
+RFC 8414 Section 3.3 and RFC 9728 Section 3.3 require the advertised
+issuer/resource to be identical to the configured value — a simple string
+comparison, not RFC 3986 equivalence. The SDK therefore never rewrites an
+identifier: well-known URLs are formed by inserting the well-known path
+segment between the authority and the identifier's path (RFC 8414 Section 3 /
+RFC 9728 Section 3), and validation rejects structurally unusable identifiers
+instead of repairing them.
+"""
+
+from urllib.parse import urlparse
+
+
+def validate_identifier(value: str, label: str) -> str:
+    """Validate that *value* is an absolute http(s) URL identifier.
+
+    The identifier must have an http or https scheme, an authority, and no
+    fragment (RFC 8707 Section 2 forbids fragments in resource identifiers).
+    Returns *value* unchanged — trailing slashes, host case, and explicit
+    ports are all legal identifier variations and are preserved verbatim.
+
+    Raises:
+        ValueError: When the identifier is structurally invalid.
+    """
+    parsed = urlparse(value)
+    if parsed.scheme not in ("https", "http"):
+        raise ValueError(f"{label} must be an absolute http or https URL: {value!r}")
+    if not parsed.netloc:
+        raise ValueError(f"{label} must include an authority: {value!r}")
+    if "#" in value:
+        raise ValueError(f"{label} must not contain a fragment: {value!r}")
+    return value

authplane/internal/metadata.py

@@ -30,7 +30,7 @@ def __init__(
             on_change=on_change,
             error_factory=lambda msg: MetadataFetchError(msg),
         )
-        self._expected_issuer = expected_issuer.rstrip("/")
+        self._expected_issuer = expected_issuer
         self._allow_http = allow_http
 
     def _validate_endpoint_url(self, field: str, value: str) -> None:
@@ -50,9 +50,11 @@ def _validate_endpoint_url(self, field: str, value: str) -> None:
             )
 
     def _validate_metadata(self, metadata: dict[str, Any]) -> dict[str, Any]:
-        issuer = str(metadata.get("issuer", "")).rstrip("/")
+        issuer = str(metadata.get("issuer", ""))
         if not issuer:
             raise MetadataFetchError("AS metadata missing required 'issuer' field")
+        # RFC 8414 Section 3.3 — the returned issuer MUST be identical to the
+        # configured one; simple string comparison, no normalisation.
         if self._expected_issuer and issuer != self._expected_issuer:
             raise MetadataFetchError(
                 f"AS metadata issuer mismatch: expected {self._expected_issuer!r}, got {issuer!r}"