Coverage for src / mysingle / auth / deps / kong.py: 0%

1from typing import Optional 

2 

3from fastapi import Request 

4 

5from mysingle.constants import ( 

6 HEADER_CORRELATION_ID, 

7 HEADER_KONG_REQUEST_ID, 

8 HEADER_KONG_USER_ID, 

9 HEADER_USER_ID, 

10) 

11 

12 

13def _get_header(request: Request, key: str) -> Optional[str]: 

14 """Case-insensitive, trimmed header getter. 

15 

16 Starlette's Headers mapping is case-insensitive, so we can use a single 

17 canonical lowercase key. Returns None for missing or empty values. 

18 """ 

19 val = request.headers.get(key) 

20 if val is None: 

21 return None 

22 val = val.strip() 

23 return val or None 

24 

25 

26def get_kong_user_id(request: Request) -> Optional[str]: 

27 """ 

28 애플리케이션 최종 사용자 ID. 

29 

30 Kong Gateway에서 JWT 플러그인을 통해 주입하는 헤더: 

31 - X-Consumer-Custom-ID: JWT의 sub 클레임 값 (원본) 

32 - X-User-Id: 다운스트림 서비스로 전파되는 표준 헤더 

33 

34 우선순위: 

35 1. X-User-Id (서비스 간 전파 표준) 

36 2. X-Consumer-Custom-ID (Kong JWT 플러그인 원본) 

37 """ 

38 # 우선순위 1: 서비스 간 전파 표준 헤더 

39 user_id = _get_header(request, HEADER_USER_ID.lower()) 

40 if user_id: 

41 return user_id 

42 

43 # 우선순위 2: Kong JWT 플러그인 원본 헤더 

44 return _get_header(request, HEADER_KONG_USER_ID.lower()) 

45 

46 

47def is_kong_authenticated(request: Request) -> bool: 

48 """Kong 헤더가 존재하면 인증된 것으로 판단""" 

49 return bool(get_kong_user_id(request)) 

50 

51 

52def get_kong_correlation_id(request: Request) -> Optional[str]: 

53 """ 

54 Correlation ID 추출 

55 

56 X-Correlation-Id 헤더를 우선 사용하며, 없으면 대체 헤더 확인 

57 """ 

58 return _get_header(request, HEADER_CORRELATION_ID.lower()) or _get_header( 

59 request, "correlation-id" 

60 ) 

61 

62 

63def get_kong_request_id(request: Request) -> Optional[str]: 

64 """ 

65 Kong Request ID 추출 

66 

67 X-Kong-Request-Id 헤더를 우선 사용하며, 없으면 대체 헤더 확인 

68 """ 

69 return _get_header(request, HEADER_KONG_REQUEST_ID.lower()) or _get_header( 

70 request, "x-request-id" 

71 ) 

72 

73 

74def get_kong_upstream_latency(request: Request) -> Optional[str]: 

75 """업스트림 지연시간(ms)""" 

76 return _get_header(request, "x-kong-upstream-latency") 

77 

78 

79def get_kong_proxy_latency(request: Request) -> Optional[str]: 

80 """프록시 지연시간(ms)""" 

81 return _get_header(request, "x-kong-proxy-latency")