Coverage for polar/kit/http.py: 39%

1from typing import Annotated 2 ctx1ab

2from urllib.parse import parse_qs, urlencode, urlparse, urlunparse 2 ctx1ab

3 

4from fastapi import Depends, Query 2 ctx1ab

5from safe_redirect_url import url_has_allowed_host_and_scheme 2 ctx1ab

6 

7from polar.config import settings 2 ctx1ab

8 

9 

10def get_safe_return_url(return_to: str | None) -> str: 2 ctx1ab

11 # Unsafe URL -> fallback to default 

12 if return_to is None or not url_has_allowed_host_and_scheme( 

13 return_to, settings.ALLOWED_HOSTS 

14 ): 

15 return settings.generate_frontend_url(settings.FRONTEND_DEFAULT_RETURN_PATH) 

16 

17 # For paths, ensure we have an absolute URL on the frontend 

18 url_info = urlparse(return_to) 

19 if not url_info.netloc: 

20 return settings.generate_frontend_url(return_to) 

21 

22 return return_to 

23 

24 

25async def _get_safe_return_url_dependency(return_to: str | None = Query(None)) -> str: 2 ctx1ab

26 return get_safe_return_url(return_to) 

27 

28 

29ReturnTo = Annotated[str, Depends(_get_safe_return_url_dependency)] 2 ctx1ab

30 

31 

32def add_query_parameters(url: str, **kwargs: str | list[str]) -> str: 2 ctx1ab

33 scheme, netloc, path, params, query, fragment = urlparse(url) 

34 parsed_query = parse_qs(query) 

35 return urlunparse( 

36 ( 

37 scheme, 

38 netloc, 

39 path, 

40 params, 

41 urlencode({**parsed_query, **kwargs}, doseq=True), 

42 fragment, 

43 ) 

44 )