Coverage for src/qdrant_loader/webhooks/auth.py: 69%

1"""Webhook authentication and authorization (WS-6).""" 

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7from functools import lru_cache 

8from typing import Any 

9 

10from fastapi import Header, HTTPException, Query, Request, status 

11 

12from qdrant_loader.utils.logging import LoggingConfig 

13 

14logger = LoggingConfig.get_logger(__name__) 

15 

16WEBHOOK_SECRET_ENV_VAR = "WEBHOOK_SECRET" 

17WEBHOOK_QUERY_PARAM = "secret" 

18 

19WEBHOOK_USE_SECRETS_MANAGER = os.getenv( 

20 "WEBHOOK_USE_SECRETS_MANAGER", "false" 

21).lower() in ("true", "1", "yes") 

22 

23WEBHOOK_TRUSTED_PROXY = os.getenv("WEBHOOK_TRUSTED_PROXY", None) 

24 

25COGNITO_REGION = os.getenv("COGNITO_REGION", "") 

26COGNITO_USER_POOL_ID = os.getenv("COGNITO_USER_POOL_ID", "") 

27COGNITO_APP_CLIENT_ID = os.getenv("COGNITO_APP_CLIENT_ID", "") 

28 

29 

30@lru_cache(maxsize=128) 

31def _get_webhook_secret_from_env() -> str: 

32 return os.getenv(WEBHOOK_SECRET_ENV_VAR, "") 

33 

34 

35def _load_project_secrets() -> dict[str, str]: 

36 raw = os.getenv("WEBHOOK_SECRETS", "") 

37 if not raw: 

38 return {} 

39 try: 

40 parsed = json.loads(raw) 

41 if isinstance(parsed, dict): 

42 return {str(k): str(v) for k, v in parsed.items()} 

43 except json.JSONDecodeError: 

44 logger.warning("WEBHOOK_SECRETS is not valid JSON; ignoring") 

45 return {} 

46 

47 

48def _cognito_jwt_enabled() -> bool: 

49 # Read fresh rather than trusting the import-time WEBHOOK_ENABLE_COGNITO_JWT 

50 # constant: callers may run before .env has been loaded into os.environ. 

51 return os.getenv("WEBHOOK_ENABLE_COGNITO_JWT", "false").lower() in ( 

52 "true", 

53 "1", 

54 "yes", 

55 ) 

56 

57 

58def webhook_auth_configured() -> bool: 

59 """Return True if at least one webhook authentication method is configured. 

60 

61 Checked at server startup so misconfiguration fails closed instead of 

62 silently accepting unauthenticated requests. 

63 """ 

64 has_global_secret = bool(os.getenv(WEBHOOK_SECRET_ENV_VAR)) 

65 has_json_project_secret = any(_load_project_secrets().values()) 

66 has_project_secret = any( 

67 key.startswith("WEBHOOK_SECRET_") and bool(value) 

68 for key, value in os.environ.items() 

69 ) 

70 return ( 

71 has_global_secret 

72 or has_json_project_secret 

73 or has_project_secret 

74 or _cognito_jwt_enabled() 

75 ) 

76 

77 

78WEBHOOK_AUTH_NOT_CONFIGURED_MESSAGE = ( 

79 "Webhook authentication is not configured. Set WEBHOOK_SECRET, " 

80 "WEBHOOK_SECRETS, WEBHOOK_SECRET_<PROJECT_ID>, or enable Cognito JWT " 

81 "(WEBHOOK_ENABLE_COGNITO_JWT=true)." 

82) 

83 

84 

85async def get_webhook_secret( 

86 project_id: str | None = None, 

87 workspace_id: str | None = None, 

88) -> str: 

89 """Resolve webhook secret for a workspace/project.""" 

90 if WEBHOOK_USE_SECRETS_MANAGER: 

91 logger.warning( 

92 "Secrets Manager requested but not yet implemented (WS-6)", 

93 feature_flag="WEBHOOK_USE_SECRETS_MANAGER", 

94 ) 

95 

96 if project_id: 

97 project_secrets = _load_project_secrets() 

98 if project_id in project_secrets: 

99 return project_secrets[project_id] 

100 env_key = f"WEBHOOK_SECRET_{project_id.upper().replace('-', '_')}" 

101 project_env_secret = os.getenv(env_key) 

102 if project_env_secret: 

103 return project_env_secret 

104 

105 _ = workspace_id 

106 return _get_webhook_secret_from_env() 

107 

108 

109def get_client_ip(request: Request) -> str: 

110 """Extract client IP, honoring X-Forwarded-For only from a trusted proxy.""" 

111 forwarded_for = request.headers.get("X-Forwarded-For") 

112 if forwarded_for and WEBHOOK_TRUSTED_PROXY: 

113 if request.client and request.client.host == WEBHOOK_TRUSTED_PROXY: 

114 return forwarded_for.split(",", 1)[0].strip() 

115 logger.warning( 

116 "X-Forwarded-For header from untrusted source; ignoring", 

117 client_host=request.client.host if request.client else None, 

118 trusted_proxy=WEBHOOK_TRUSTED_PROXY, 

119 ) 

120 elif forwarded_for and not WEBHOOK_TRUSTED_PROXY: 

121 logger.debug( 

122 "X-Forwarded-For present but WEBHOOK_TRUSTED_PROXY not set; using request.client", 

123 ) 

124 

125 if request.client: 

126 return request.client.host 

127 return "unknown" 

128 

129 

130def _extract_bearer_token(authorization: str | None) -> str | None: 

131 if not authorization: 

132 return None 

133 auth = authorization.strip() 

134 if auth.lower().startswith("bearer "): 

135 return auth.split(None, 1)[1].strip() 

136 return auth 

137 

138 

139def _looks_like_jwt(token: str) -> bool: 

140 return token.count(".") == 2 

141 

142 

143class CognitoJWTValidator: 

144 """Validate Cognito JWT tokens for application routes (WS-6).""" 

145 

146 @staticmethod 

147 def _issuer() -> str: 

148 if not COGNITO_REGION or not COGNITO_USER_POOL_ID: 

149 raise HTTPException( 

150 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

151 detail="Cognito is not configured.", 

152 ) 

153 return ( 

154 f"https://cognito-idp.{COGNITO_REGION}.amazonaws.com/" 

155 f"{COGNITO_USER_POOL_ID}" 

156 ) 

157 

158 @classmethod 

159 async def validate_token(cls, token: str) -> dict[str, Any]: 

160 if not _cognito_jwt_enabled(): 

161 return {"sub": "local-dev"} 

162 

163 try: 

164 import jwt 

165 from jwt import PyJWKClient 

166 except ImportError as exc: 

167 raise HTTPException( 

168 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

169 detail="PyJWT is required for Cognito validation. " 

170 "Install qdrant-loader[server].", 

171 ) from exc 

172 

173 issuer = cls._issuer() 

174 jwks_url = f"{issuer}/.well-known/jwks.json" 

175 

176 try: 

177 jwk_client = PyJWKClient(jwks_url) 

178 signing_key = jwk_client.get_signing_key_from_jwt(token) 

179 decode_kwargs: dict[str, Any] = { 

180 "algorithms": ["RS256"], 

181 "issuer": issuer, 

182 "options": {"verify_aud": bool(COGNITO_APP_CLIENT_ID)}, 

183 } 

184 if COGNITO_APP_CLIENT_ID: 

185 decode_kwargs["audience"] = COGNITO_APP_CLIENT_ID 

186 return jwt.decode(token, signing_key.key, **decode_kwargs) 

187 except Exception as exc: 

188 logger.warning("Cognito JWT validation failed", error=str(exc)) 

189 raise HTTPException( 

190 status_code=status.HTTP_401_UNAUTHORIZED, 

191 detail="Invalid or expired Cognito token.", 

192 ) from exc 

193 

194 @staticmethod 

195 def extract_workspace_id(claims: dict[str, Any]) -> str | None: 

196 return claims.get("custom:workspace_id") or claims.get("workspace") 

197 

198 

199async def verify_webhook_token( 

200 project_id: str | None = None, 

201 webhook_token: str | None = Query(None, alias=WEBHOOK_QUERY_PARAM), 

202 authorization: str | None = Header(None, convert_underscores=False), 

203) -> None: 

204 """Verify webhook access for Jira-compatible endpoints. 

205 

206 Jira Cloud only supports shared-secret query tokens, so webhook routes accept 

207 the project-scoped WEBHOOK_SECRET via Bearer header or ?secret= query param. 

208 Cognito JWT is validated when enabled and the bearer token is a JWT. 

209 """ 

210 secret = await get_webhook_secret(project_id=project_id) 

211 token_value = _extract_bearer_token(authorization) or webhook_token 

212 

213 if _cognito_jwt_enabled() and token_value and _looks_like_jwt(token_value): 

214 await CognitoJWTValidator.validate_token(token_value) 

215 return 

216 

217 if not secret: 

218 logger.error( 

219 "Webhook secret is not configured", 

220 env_var=WEBHOOK_SECRET_ENV_VAR, 

221 project_id=project_id, 

222 ) 

223 raise HTTPException( 

224 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, 

225 detail="Webhook authentication is not configured.", 

226 ) 

227 

228 if webhook_token and not authorization: 

229 logger.warning( 

230 "Using webhook token via URL query param is insecure; prefer Authorization: Bearer header", 

231 param=WEBHOOK_QUERY_PARAM, 

232 ) 

233 

234 if not token_value or token_value != secret: 

235 logger.warning( 

236 "Unauthorized webhook request", 

237 project_id=project_id, 

238 received=bool(token_value), 

239 ) 

240 raise HTTPException( 

241 status_code=status.HTTP_401_UNAUTHORIZED, 

242 detail="Invalid or missing webhook token.", 

243 ) 

244 

245 

246async def verify_cognito_token( 

247 authorization: str | None = Header(None, convert_underscores=False), 

248) -> dict[str, Any]: 

249 """Dependency for non-webhook routes that require Cognito JWT (WS-6).""" 

250 token_value = _extract_bearer_token(authorization) 

251 if not token_value: 

252 raise HTTPException( 

253 status_code=status.HTTP_401_UNAUTHORIZED, 

254 detail="Authorization header required.", 

255 ) 

256 return await CognitoJWTValidator.validate_token(token_value) 

257 

258 

259async def verify_ingest_auth( 

260 project_id: str | None = Query(None), 

261 webhook_token: str | None = Query(None, alias=WEBHOOK_QUERY_PARAM), 

262 authorization: str | None = Header(None, convert_underscores=False), 

263) -> None: 

264 """Authenticate POST /ingest (API clients). 

265 

266 Prefers Authorization: Bearer with Cognito JWT when enabled, otherwise the 

267 same project-scoped webhook secret used for connector webhooks. 

268 """ 

269 await verify_webhook_token( 

270 project_id=project_id, 

271 webhook_token=webhook_token, 

272 authorization=authorization, 

273 )