Coverage for src / qdrant_loader_mcp_server / transport / dependencies.py: 100%

1"""FastAPI dependencies for the MCP transport layer.""" 

2 

3import re 

4 

5from fastapi import HTTPException, Request 

6 

7_ALLOWED_ORIGIN_RE = re.compile(r"^https?://(localhost|127\.0\.0\.1)(:[0-9]+)?$") 

8 

9 

10async def get_mcp_handler(request: Request): 

11 """Retrieve the MCP handler from app state. 

12 

13 Raises HTTPException 503 if the handler has not been initialised yet 

14 (worker still starting up). 

15 """ 

16 handler = getattr(request.app.state, "mcp_handler", None) 

17 if handler is None: 

18 raise HTTPException(status_code=503, detail="Server is starting up") 

19 return handler 

20 

21 

22async def validate_origin(request: Request): 

23 """Validate the Origin header to prevent DNS rebinding attacks. 

24 

25 Allows requests without an Origin header (non-browser clients). 

26 Rejects origins that do not match localhost / 127.0.0.1. 

27 """ 

28 origin = request.headers.get("origin") 

29 if not origin: 

30 return 

31 if not _ALLOWED_ORIGIN_RE.match(origin): 

32 raise HTTPException(status_code=403, detail="Invalid origin")