fix(authentication): Fix PAT authentication

Author: timothedelion

packages/gg_api_core/src/gg_api_core/client.py

@@ -135,11 +135,13 @@ def _init_personal_access_token(self, personal_access_token: str | None = None):
                     "HTTP/SSE mode requires per-request authentication via Authorization headers. "
                     "For local OAuth authentication, use stdio transport (unset MCP_PORT)."
                 )
+            else:
+                # HTTP mode and no personal access token provided
+                # Token will be extracted from Authorization header per-request via get_client()
+                logger.info("HTTP/SSE mode: token will be provided via Authorization header per-request")
+                self._oauth_token = None
         else:
-            if personal_access_token:
-                logger.info("Using provided PAT")
-                self._oauth_token = personal_access_token
-            elif personal_access_token := os.environ.get("GITGUARDIAN_PERSONAL_ACCESS_TOKEN"):
+            if personal_access_token := os.environ.get("GITGUARDIAN_PERSONAL_ACCESS_TOKEN"):
                 logger.info("Using PAT from environment variable")
                 self._oauth_token = personal_access_token
             else:
@@ -260,7 +262,7 @@ async def _ensure_api_token(self):
         and in test environments.
         """
 
-        if self._oauth_token is not None:
+        if getattr(self, "_oauth_token", None) is not None:
             return
 
         if not is_oauth_enabled():
@@ -269,7 +271,7 @@ async def _ensure_api_token(self):
         # Use a global lock to prevent parallel OAuth flows across all client instances
         async with _oauth_lock:
             # Double-check pattern: another thread might have completed OAuth while we waited for the lock
-            if self._oauth_token is not None:
+            if getattr(self, "_oauth_token", None) is not None:
                 logger.debug("OAuth token already available after waiting for lock")
                 return
 

packages/gg_api_core/src/gg_api_core/utils.py

@@ -1,7 +1,11 @@
 import logging
+import os
 import re
 from urllib.parse import urljoin as urllib_urljoin
 
+from fastmcp.server.dependencies import get_http_headers
+from mcp.server.fastmcp.exceptions import ValidationError
+
 from .client import GitGuardianClient
 
 # Setup logger
@@ -27,25 +31,100 @@ def get_client(personal_access_token: str | None = None) -> GitGuardianClient:
     with that token (not cached). This is useful for per-request authentication
     via HTTP Authorization headers.
 
+    In HTTP/SSE mode (when MCP_PORT is set), this function automatically extracts
+    the token from the Authorization header of the current request.
+
     Args:
         personal_access_token: Optional Personal Access Token to use for authentication.
             If provided, a new client instance is created with this token.
 
     Returns:
         GitGuardianClient: The cached client instance or a new instance with the provided PAT
     """
-    # If a PAT is provided, create a new client instance (don't use singleton)
+    # Check if we're in HTTP/SSE mode (MCP_PORT is set)
+    mcp_port = os.environ.get("MCP_PORT")
+
+    logger.debug(
+        f"get_client() called: mcp_port={mcp_port}, personal_access_token={'provided' if personal_access_token else 'None'}"
+    )
+
+    if mcp_port and not personal_access_token:
+        # In HTTP mode, get token from Authorization header or raise
+        logger.debug("HTTP mode detected, extracting token from request headers")
+        try:
+            personal_access_token = get_personal_access_token_from_request()
+            logger.info("Successfully extracted token from HTTP request headers")
+        except ValidationError as e:
+            logger.error(f"Failed to extract token from HTTP headers: {e}")
+            raise
+
+    # If a PAT is provided (or extracted from headers), create a new client instance (don't use singleton)
     if personal_access_token:
         logger.debug("Creating new GitGuardian client with provided Personal Access Token")
         return get_gitguardian_client(personal_access_token=personal_access_token)
 
     # Otherwise, use the singleton pattern
+    logger.debug("Using singleton client (no PAT provided)")
     global _client_singleton
     if _client_singleton is None:
+        logger.info("Creating singleton client instance")
         _client_singleton = get_gitguardian_client()
     return _client_singleton
 
 
+def get_personal_access_token_from_request():
+    """Extract personal access token from HTTP request headers.
+
+    Raises:
+        ValidationError: If headers are missing or invalid
+    """
+    try:
+        headers = get_http_headers()
+        logger.debug(f"Retrieved HTTP headers: {list(headers.keys()) if headers else 'None'}")
+    except Exception as e:
+        logger.error(f"Failed to get HTTP headers: {e}")
+        raise ValidationError(f"Failed to retrieve HTTP headers: {e}")
+
+    if not headers:
+        logger.error("No HTTP headers available in current context")
+        raise ValidationError("No HTTP headers available - Authorization header required in HTTP mode")
+
+    auth_header = headers.get("authorization") or headers.get("Authorization")
+    if not auth_header:
+        logger.error(f"Missing Authorization header. Available headers: {list(headers.keys())}")
+        raise ValidationError("Missing Authorization header - required in HTTP mode")
+
+    token = _extract_token_from_auth_header(auth_header)
+    if not token:
+        logger.error("Failed to extract token from Authorization header")
+        raise ValidationError("Invalid Authorization header format")
+
+    logger.debug("Successfully extracted token from Authorization header")
+    return token
+
+
+def _extract_token_from_auth_header(auth_header: str) -> str | None:
+    """Extract token from Authorization header.
+
+    Supports formats:
+    - Bearer <token>
+    - Token <token>
+    - <token> (raw)
+    """
+    auth_header = auth_header.strip()
+
+    if auth_header.lower().startswith("bearer "):
+        return auth_header[7:].strip()
+
+    if auth_header.lower().startswith("token "):
+        return auth_header[6:].strip()
+
+    if auth_header:
+        return auth_header
+
+    return None
+
+
 def parse_repo_url(remote_url: str) -> str | None:
     """Parse repository name from git remote URL.
 

tests/test_oauth_config_validation.py

@@ -22,9 +22,10 @@ def test_raises_error_when_both_mcp_port_and_oauth_enabled(self):
     def test_allows_mcp_port_without_oauth(self):
         """Test that MCP_PORT can be set without ENABLE_LOCAL_OAUTH."""
         with patch.dict(os.environ, {"MCP_PORT": "8080", "ENABLE_LOCAL_OAUTH": "false"}, clear=False):
-            # Should not raise
+            # Should not raise - HTTP mode allows client creation (token provided per-request)
             client = GitGuardianClient()
             assert client is not None
+            assert client._oauth_token is None  # Token will be provided per-request
 
     def test_allows_oauth_without_mcp_port(self):
         """Test that ENABLE_LOCAL_OAUTH can be set without MCP_PORT."""
@@ -66,6 +67,7 @@ def test_empty_string_is_not_true(self):
             # Should not raise - empty string is treated as false
             client = GitGuardianClient()
             assert client is not None
+            assert client._oauth_token is None  # Token will be provided per-request
 
     def test_unset_defaults_to_true_and_conflicts_with_mcp_port(self):
         """Test that unset ENABLE_LOCAL_OAUTH defaults to true, which conflicts with MCP_PORT."""