[Cherry-pick to branch-1.3] [#12447] fix(mcp-server): Preserve the authorization scheme for static credentials (#12439)

docs/gravitino-mcp-server.md

@@ -109,7 +109,7 @@ You could config Gravitino MCP server by arguments, `uv run mcp_server -h` shows
 | `--gravitino-uri` | The URI of Gravitino server.                                                     | `http://127.0.0.1:8090`     | No       |
 | `--transport`     | Transport protocol: stdio (local), http / streamable-http (Streamable HTTP).     | `stdio`                     | No       |
 | `--mcp-url`       | The URL of MCP server if using HTTP transport.                                   | `http://127.0.0.1:8000/mcp` | No       |
-| `--token`         | OAuth2 Bearer token for Gravitino; or set `GRAVITINO_TOKEN`. See Authentication. | none (anonymous)            | No       |
+| `--token`         | Static credential for Gravitino; or set `GRAVITINO_TOKEN`. See Authentication.   | none (anonymous)            | No       |
 | `--tls-cert`      | PEM certificate to serve the endpoint over HTTPS. Requires `--tls-key`.          | none                        | No       |
 | `--tls-key`       | PEM private key to serve the endpoint over HTTPS. Requires `--tls-cert`.         | none                        | No       |
 
@@ -119,10 +119,26 @@ By default the MCP server talks to Gravitino anonymously. There are two ways to
 
 ### Static startup token (stdio and HTTP)
 
-Pass `--token` (or set the `GRAVITINO_TOKEN` environment variable) to authenticate the server with a static OAuth2 Bearer token. The value is treated as a Bearer token and sent as `Authorization: Bearer <token>`. The token is masked in the server's log output.
+Pass `--token` (or set the `GRAVITINO_TOKEN` environment variable) to authenticate the server with a static credential. The token is masked in the server's log output.
+
+A bare value is treated as an OAuth2 token and sent as `Authorization: Bearer <token>`. A value that already begins with an HTTP authentication scheme is forwarded with that scheme preserved, so the credential can match whatever `gravitino.authenticators` the server is configured with:
+
+| `--token` value             | `Authorization` header sent    |
+|-----------------------------|--------------------------------|
+| `abc`                       | `Bearer abc`                   |
+| `Bearer abc`                | `Bearer abc`                   |
+| `Basic dXNlcjpwYXNz`        | `Basic dXNlcjpwYXNz`           |
+| `Custom credentials`        | `Custom credentials`           |
+| empty or whitespace only    | none (anonymous)               |
+
+The built-in scheme names (`Basic`, `Bearer`, `Negotiate`) are recognized case-insensitively and normalized to the capitalization Gravitino's authenticators expect; a custom scheme name is forwarded unchanged.
+
+Because a bare token is only Bearer-prefixed when it carries no scheme, a static credential whose value contains a space and begins with a scheme-like word is interpreted as scheme plus credential. Quote such values with an explicit scheme (for example `--token "Bearer my secret"`) to keep them Bearer tokens.
 
 ```shell
 uv run mcp_server --metalake test --gravitino-uri http://127.0.0.1:8090 --token <your-token>
+# or, against a server configured with `gravitino.authenticators = basic`
+uv run mcp_server --metalake test --gravitino-uri http://127.0.0.1:8090 --token "Basic $(printf '%s' 'user:password' | base64)"
 # or
 export GRAVITINO_TOKEN=<your-token>
 uv run mcp_server --metalake test --gravitino-uri http://127.0.0.1:8090

mcp-server/mcp_server/core/audit.py

@@ -29,14 +29,22 @@ def _extract_principal(authorization: str) -> str:
 
     - "Basic <base64(user:secret)>" → "<user>"  (Gravitino simple auth)
     - "Bearer <token>"              → "bearer:<first-8-chars-of-token>"
+    - "<scheme> <credential>"       → "<scheme>:<first-8-chars-of-credential>"
     - empty / missing / unparsable  → "anonymous"
+
+    The credential may itself contain spaces (a custom scheme is free to use a
+    comma-separated parameter list), so only the scheme is split off. Falling
+    back to the scheme name keeps a static custom-scheme identity attributable
+    in the audit log instead of recording it as anonymous.
     """
     if not authorization:
         return "anonymous"
-    parts = authorization.split()
+    parts = authorization.split(None, 1)
     if len(parts) != 2:
         return "anonymous"
-    scheme, credential = parts[0].lower(), parts[1]
+    scheme, credential = parts[0].lower(), parts[1].strip()
+    if not credential:
+        return "anonymous"
     if scheme == "basic":
         try:
             decoded = base64.b64decode(credential, validate=True).decode(
@@ -46,9 +54,7 @@ def _extract_principal(authorization: str) -> str:
             return "anonymous"
         user = decoded.split(":", 1)[0]
         return user if user else "anonymous"
-    if scheme == "bearer":
-        return f"bearer:{credential[:8]}"
-    return "anonymous"
+    return f"{scheme}:{credential[:8]}"
 
 
 def emit(

mcp-server/mcp_server/core/context.py

@@ -17,6 +17,7 @@
 
 import asyncio
 import logging
+import re
 from collections import OrderedDict
 
 from mcp_server.client.factory import RESTClientFactory
@@ -31,6 +32,23 @@
 # (e.g. rotating tokens) come and go.
 _MAX_CACHED_CLIENTS = 128
 
+# An RFC 9110 auth-scheme uses the HTTP token syntax. Here it must be followed by
+# one or more spaces plus credentials. Requiring credentials preserves the legacy
+# behavior for a bare token whose value happens to be a scheme name (for example,
+# Bearer).
+_AUTHORIZATION_CREDENTIAL = re.compile(
+    r"^(?P<scheme>[!#$%&'*+\-.^_`|~0-9A-Za-z]+) +(?P<credential>\S.*)$"
+)
+
+# Gravitino currently matches its built-in schemes case-sensitively. HTTP scheme
+# names are case-insensitive, so normalize them before forwarding. Custom scheme
+# names remain unchanged for custom Gravitino authenticators.
+_CANONICAL_AUTH_SCHEMES = {
+    "basic": "Basic",
+    "bearer": "Bearer",
+    "negotiate": "Negotiate",
+}
+
 
 def _get_request_authorization() -> str:
     """Return the raw ``Authorization`` header of the current HTTP request.
@@ -55,11 +73,24 @@ def _get_request_authorization() -> str:
 def startup_authorization(setting: Setting) -> str:
     """The static --token rendered as an ``Authorization`` header value.
 
-    The CLI token is treated as an OAuth2 Bearer token. Empty string when no
-    token is configured (anonymous). This is the identity used in stdio mode and
-    the fallback for HTTP requests that carry no ``Authorization`` header.
+    A value containing a valid HTTP authentication scheme and credentials is
+    forwarded as an Authorization credential. Built-in Gravitino scheme names
+    are normalized to the capitalization its authenticators expect, while a
+    custom scheme name is preserved. A bare token is treated as OAuth2 and
+    prefixed with ``Bearer``. Empty string when no token is configured
+    (anonymous). This is the identity used in stdio mode and the fallback for
+    HTTP requests that carry no ``Authorization`` header.
     """
-    return f"Bearer {setting.token}" if setting.token else ""
+    token = setting.token.strip()
+    if not token:
+        return ""
+    match = _AUTHORIZATION_CREDENTIAL.fullmatch(token)
+    if match:
+        scheme = match.group("scheme")
+        credential = match.group("credential")
+        canonical_scheme = _CANONICAL_AUTH_SCHEMES.get(scheme.lower(), scheme)
+        return f"{canonical_scheme} {credential}"
+    return f"Bearer {token}"
 
 
 class GravitinoContext:

mcp-server/mcp_server/core/setting.py

@@ -33,9 +33,11 @@ class Setting:  # pylint: disable=too-many-instance-attributes
     tags: Set[str] = field(default_factory=set)
     transport: str = DefaultSetting.default_transport
     mcp_url: str = DefaultSetting.default_mcp_url
-    # Static OAuth2 Bearer token. Sent on every request in stdio mode; in HTTP
-    # mode it is only the fallback used when an incoming request carries no
-    # Authorization header (per-request identity takes priority).
+    # Static authorization credential. A bare value is treated as an OAuth2 Bearer
+    # token; a value containing a valid scheme and credential (``Basic ...``) is
+    # forwarded as an Authorization credential. Sent on every request in stdio
+    # mode; in HTTP mode it is only the fallback used when an incoming request
+    # carries no Authorization header (per-request identity takes priority).
     # Empty string means anonymous (no Authorization header sent).
     # repr=False keeps the raw value out of the dataclass-generated __repr__.
     token: str = field(default="", repr=False)
@@ -45,7 +47,9 @@ class Setting:  # pylint: disable=too-many-instance-attributes
     tls_key: str = ""
 
     def __str__(self) -> str:
-        token_display = "***" if self.token else ""
+        # Mirror startup_authorization: a whitespace-only token is anonymous on
+        # the wire, so it must not be logged as a configured identity.
+        token_display = "***" if self.token.strip() else ""
         return (
             f"Setting(metalake={self.metalake}, gravitino_uri={self.gravitino_uri}, "
             f"tags={self.tags}, transport={self.transport}, mcp_url={self.mcp_url}, "

mcp-server/mcp_server/main.py

@@ -112,9 +112,12 @@ def _parse_args():
         "--token",
         type=str,
         default=os.environ.get("GRAVITINO_TOKEN", ""),
-        help="Static OAuth2 Bearer token used to authenticate to Gravitino. "
-        "In stdio mode it is sent on every request; in HTTP mode it is only the "
-        "fallback when an incoming request carries no Authorization header "
+        help="Static credential used as the Authorization header when "
+        "authenticating to Gravitino. A bare token is treated as an OAuth2 Bearer "
+        "token; a value containing a valid scheme and credential, such as "
+        "'Basic <base64>', is sent as an Authorization credential. In stdio mode "
+        "it is sent on every request; in HTTP mode it is only the fallback when an "
+        "incoming request carries no Authorization header "
         "(per-request identity takes priority). "
         "Can also be set via the GRAVITINO_TOKEN environment variable. "
         "When omitted, requests are sent without authentication.",

mcp-server/tests/unit/test_audit.py

@@ -123,11 +123,30 @@ def test_basic_auth_invalid_base64_returns_anonymous(self):
             audit._extract_principal("Basic not-valid-base64!!"), "anonymous"
         )
 
-    def test_unknown_scheme_returns_anonymous(self):
+    def test_other_scheme_falls_back_to_scheme_prefix(self):
+        """A non-Basic scheme stays attributable via '<scheme>:<first-8>'."""
         self.assertEqual(
-            audit._extract_principal("Negotiate abc123"), "anonymous"
+            audit._extract_principal("Negotiate abc123"), "negotiate:abc123"
         )
 
+    def test_custom_scheme_with_multi_word_credential(self):
+        """A credential containing spaces is not treated as unparsable."""
+        self.assertEqual(
+            audit._extract_principal("Custom-Scheme key=abcdefghij, sig=xy"),
+            "custom-scheme:key=abcd",
+        )
+
+    def test_extra_space_between_scheme_and_credential(self):
+        """Repeated separators do not break Basic decoding."""
+        self.assertEqual(
+            audit._extract_principal("Basic   YWxpY2U6ZHVtbXk="), "alice"
+        )
+
+    def test_scheme_without_credential_returns_anonymous(self):
+        """A scheme with no credential has no identity to report."""
+        self.assertEqual(audit._extract_principal("Bearer"), "anonymous")
+        self.assertEqual(audit._extract_principal("Bearer   "), "anonymous")
+
 
 class TestAuditMiddlewareIntegration(unittest.TestCase):
     """Integration tests: AuditMiddleware emits records via the full MCP tool path."""

mcp-server/tests/unit/test_auth_flow.py

@@ -24,7 +24,7 @@
 from mcp_server.client.plain.plain_rest_client_operation import (
     PlainRESTClientOperation,
 )
-from mcp_server.core.context import GravitinoContext
+from mcp_server.core.context import GravitinoContext, startup_authorization
 from mcp_server.core.setting import Setting
 from mcp_server.main import _parse_args
 
@@ -126,6 +126,13 @@ def test_empty_token_shows_empty_in_str(self):
         setting = Setting(metalake="ml", token="")
         self.assertNotIn("***", str(setting))
 
+    def test_whitespace_only_token_shows_empty_in_str(self):
+        """A whitespace-only token is anonymous on the wire, so do not mask it
+        as a configured identity."""
+        setting = Setting(metalake="ml", token="   ")
+        self.assertEqual(startup_authorization(setting), "")
+        self.assertNotIn("***", str(setting))
+
 
 class TestTokenArgParsing(unittest.TestCase):
     """Verify --token CLI argument and GRAVITINO_TOKEN env var precedence."""
@@ -157,6 +164,77 @@ def test_no_token_anywhere_defaults_to_empty(self):
         self.assertEqual(args.token, "")
 
 
+class TestStartupAuthorization(unittest.TestCase):
+    """Verify startup_authorization renders the static --token correctly."""
+
+    def test_bare_token_is_prefixed_with_bearer(self):
+        """A bare token is treated as OAuth2 and prefixed with Bearer."""
+        setting = Setting(metalake="ml", token="abc")
+        self.assertEqual(startup_authorization(setting), "Bearer abc")
+
+    def test_bearer_token_is_not_double_wrapped(self):
+        """A value already carrying the Bearer scheme is used verbatim."""
+        setting = Setting(metalake="ml", token="Bearer abc")
+        self.assertEqual(startup_authorization(setting), "Bearer abc")
+
+    def test_basic_token_passes_through(self):
+        """A value carrying the Basic scheme is used verbatim."""
+        setting = Setting(metalake="ml", token="Basic dXNlcjpwYXNz")
+        self.assertEqual(startup_authorization(setting), "Basic dXNlcjpwYXNz")
+
+    def test_builtin_scheme_is_canonicalized_case_insensitively(self):
+        """A built-in scheme is matched case-insensitively and canonicalized."""
+        test_cases = (
+            ("basic credentials", "Basic credentials"),
+            ("bearer credentials", "Bearer credentials"),
+            ("negotiate credentials", "Negotiate credentials"),
+        )
+        for token, expected in test_cases:
+            with self.subTest(token=token):
+                setting = Setting(metalake="ml", token=token)
+                self.assertEqual(startup_authorization(setting), expected)
+
+    def test_scheme_separator_is_normalized(self):
+        """Extra spaces between a scheme and credential are collapsed."""
+        setting = Setting(metalake="ml", token="Basic   credentials")
+        self.assertEqual(startup_authorization(setting), "Basic credentials")
+
+    def test_custom_scheme_passes_through(self):
+        """A syntactically valid custom scheme is not wrapped in Bearer."""
+        setting = Setting(metalake="ml", token="Custom-Scheme credentials")
+        self.assertEqual(
+            startup_authorization(setting), "Custom-Scheme credentials"
+        )
+
+    def test_empty_token_stays_empty(self):
+        """No token configured yields an empty Authorization value."""
+        setting = Setting(metalake="ml", token="")
+        self.assertEqual(startup_authorization(setting), "")
+
+    def test_whitespace_only_token_stays_empty(self):
+        """A whitespace-only token does not produce an Authorization value."""
+        setting = Setting(metalake="ml", token="   ")
+        self.assertEqual(startup_authorization(setting), "")
+
+    def test_bare_token_is_stripped_then_prefixed(self):
+        """Surrounding whitespace is stripped before prefixing a bare token."""
+        setting = Setting(metalake="ml", token="  abc  ")
+        self.assertEqual(startup_authorization(setting), "Bearer abc")
+
+    def test_scheme_word_with_no_credential_is_bare_token(self):
+        """A scheme-like word with nothing after it is treated as a bare token."""
+        setting = Setting(metalake="ml", token="Bearer")
+        self.assertEqual(startup_authorization(setting), "Bearer Bearer")
+
+    def test_invalid_scheme_syntax_is_treated_as_bare_token(self):
+        """A value without a valid HTTP scheme is treated as a bare token."""
+        setting = Setting(metalake="ml", token="Not/A/Scheme credentials")
+        self.assertEqual(
+            startup_authorization(setting),
+            "Bearer Not/A/Scheme credentials",
+        )
+
+
 class TestGravitinoContextTokenPropagation(_RealFactoryTestCase):
     """Verify GravitinoContext passes token from Setting to the REST client."""
 
@@ -177,6 +255,23 @@ def test_context_propagates_token(self):
         finally:
             _close(rest_client)
 
+    def test_context_propagates_basic_credential(self):
+        """A static Basic credential reaches the client with canonical casing."""
+        setting = Setting(
+            metalake="ml",
+            gravitino_uri="http://localhost:8090",
+            token="basic YWxpY2U6cGFzc3dvcmQ=",
+        )
+        ctx = GravitinoContext(setting)
+        rest_client = ctx.rest_client()
+        try:
+            self.assertEqual(
+                _headers_of(rest_client).get("Authorization"),
+                "Basic YWxpY2U6cGFzc3dvcmQ=",
+            )
+        finally:
+            _close(rest_client)
+
     def test_context_anonymous_when_no_token(self):
         """Empty token in Setting → no Authorization header in REST calls."""
         setting = Setting(