Coverage for /Users/antonigmitruk/golf/src/golf/auth/helpers.py: 0%

72 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-08-16 18:46 +0200

1"""Helper functions for working with authentication in MCP context.""" 

2 

3from contextvars import ContextVar 

4from typing import Any 

5 

6from starlette.requests import Request 

7 

8# Context variable to store the current request's API key 

9_current_api_key: ContextVar[str | None] = ContextVar("current_api_key", default=None) 

10 

11 

12def get_provider_token() -> str | None: 

13 """ 

14 Get a provider token (legacy function - no longer supported in Golf 0.2.x). 

15 

16 In Golf 0.2.x, use FastMCP's built-in auth providers for OAuth flows. 

17 This function returns None and is kept for backwards compatibility. 

18 """ 

19 # Legacy OAuth provider support removed in Golf 0.2.x 

20 # Use FastMCP 2.11+ auth providers instead 

21 return None 

22 

23 

24def extract_token_from_header(auth_header: str) -> str | None: 

25 """Extract bearer token from Authorization header. 

26 

27 Args: 

28 auth_header: Authorization header value 

29 

30 Returns: 

31 Bearer token or None if not present/valid 

32 """ 

33 if not auth_header: 

34 return None 

35 

36 parts = auth_header.split() 

37 if len(parts) != 2 or parts[0].lower() != "bearer": 

38 return None 

39 

40 return parts[1] 

41 

42 

43def set_api_key(api_key: str | None) -> None: 

44 """Set the API key for the current request context. 

45 

46 This is an internal function used by the middleware. 

47 

48 Args: 

49 api_key: The API key to store in the context 

50 """ 

51 _current_api_key.set(api_key) 

52 

53 

54def get_api_key() -> str | None: 

55 """Get the API key from the current request context. 

56 

57 This function should be used in tools to retrieve the API key 

58 that was sent in the request headers. 

59 

60 Returns: 

61 The API key if available, None otherwise 

62 

63 Example: 

64 # In a tool file 

65 from golf.auth import get_api_key 

66 

67 async def call_api(): 

68 api_key = get_api_key() 

69 if not api_key: 

70 return {"error": "No API key provided"} 

71 

72 # Use the API key in your request 

73 headers = {"Authorization": f"Bearer {api_key}"} 

74 ... 

75 """ 

76 # Try to get directly from HTTP request if available (FastMCP pattern) 

77 try: 

78 # This follows the FastMCP pattern for accessing HTTP requests 

79 from fastmcp.server.dependencies import get_http_request 

80 

81 request = get_http_request() 

82 

83 if request and hasattr(request, "state") and hasattr(request.state, "api_key"): 

84 api_key = request.state.api_key 

85 return api_key 

86 

87 # Get the API key configuration 

88 from golf.auth.api_key import get_api_key_config 

89 

90 api_key_config = get_api_key_config() 

91 

92 if api_key_config and request: 

93 # Extract API key from headers 

94 header_name = api_key_config.header_name 

95 header_prefix = api_key_config.header_prefix 

96 

97 # Case-insensitive header lookup 

98 api_key = None 

99 for k, v in request.headers.items(): 

100 if k.lower() == header_name.lower(): 

101 api_key = v 

102 break 

103 

104 # Strip prefix if configured 

105 if api_key and header_prefix and api_key.startswith(header_prefix): 

106 api_key = api_key[len(header_prefix) :] 

107 

108 if api_key: 

109 return api_key 

110 except (ImportError, RuntimeError): 

111 # FastMCP not available or not in HTTP context 

112 pass 

113 except Exception: 

114 pass 

115 

116 # Final fallback: environment variable (for development/testing) 

117 import os 

118 

119 env_api_key = os.environ.get("API_KEY") 

120 if env_api_key: 

121 return env_api_key 

122 

123 return None 

124 

125 

126def get_api_key_from_request(request: Request) -> str | None: 

127 """Get the API key from a specific request object. 

128 

129 This is useful when you have direct access to the request object. 

130 

131 Args: 

132 request: The Starlette Request object 

133 

134 Returns: 

135 The API key if available, None otherwise 

136 """ 

137 # Check request state first (set by our middleware) 

138 if hasattr(request, "state") and hasattr(request.state, "api_key"): 

139 return request.state.api_key 

140 

141 # Fall back to context variable 

142 return _current_api_key.get() 

143 

144 

145def debug_api_key_context() -> dict[str, Any]: 

146 """Debug function to inspect API key context. 

147 

148 Returns a dictionary with debugging information about the current 

149 API key context. Useful for troubleshooting authentication issues. 

150 

151 Returns: 

152 Dictionary with debug information 

153 """ 

154 import asyncio 

155 import os 

156 import sys 

157 

158 debug_info = { 

159 "context_var_value": _current_api_key.get(), 

160 "has_async_task": False, 

161 "task_id": None, 

162 "main_module_has_storage": False, 

163 "main_module_has_context": False, 

164 "request_id_from_context": None, 

165 "env_vars": { 

166 "API_KEY": bool(os.environ.get("API_KEY")), 

167 "GOLF_API_KEY_DEBUG": os.environ.get("GOLF_API_KEY_DEBUG", "false"), 

168 }, 

169 } 

170 

171 try: 

172 task = asyncio.current_task() 

173 if task: 

174 debug_info["has_async_task"] = True 

175 debug_info["task_id"] = id(task) 

176 except Exception: 

177 pass 

178 

179 try: 

180 main_module = sys.modules.get("__main__") 

181 if main_module: 

182 debug_info["main_module_has_storage"] = hasattr(main_module, "api_key_storage") 

183 debug_info["main_module_has_context"] = hasattr(main_module, "request_id_context") 

184 

185 if hasattr(main_module, "request_id_context"): 

186 request_id_context = main_module.request_id_context 

187 debug_info["request_id_from_context"] = request_id_context.get() 

188 except Exception: 

189 pass 

190 

191 return debug_info