Coverage for daklib/rpc_log.py: 97%

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# © 2026, Ansgar 🙀 <ansgar@debian.org> 

3 

4""" 

5Logging middleware for the DAK RPC server. 

6""" 

7 

8import contextvars 

9import logging 

10import time 

11import uuid 

12from collections.abc import Callable 

13from typing import Any, override 

14 

15import grpc 

16 

17from daklib.rpc_auth import current_auth 

18from daklib.rpc_peer import current_peer 

19 

20logger = logging.getLogger("dak.rpc") 

21 

22current_request_id: contextvars.ContextVar[str | None] = contextvars.ContextVar( 

23 "current_request_id", default=None 

24) 

25 

26 

27class RequestContextFilter(logging.Filter): 

28 """Logging filter that injects request_id, peer and auth_sub from context vars.""" 

29 

30 @override 

31 def filter(self, record: logging.LogRecord) -> bool: 

32 record.request_id = current_request_id.get() or "-" # type: ignore[attr-defined] 

33 peer = current_peer.get() 

34 record.peer = str(peer) if peer is not None else "-" # type: ignore[attr-defined] 

35 auth = current_auth.get() 

36 record.auth_sub = auth.sub if auth is not None else "-" # type: ignore[attr-defined] 

37 return True 

38 

39 

40class LoggingInterceptor(grpc.ServerInterceptor): 

41 """gRPC interceptor for request logging and error sanitization.""" 

42 

43 @override 

44 def intercept_service( 

45 self, 

46 continuation: "Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[Any, Any] | None]", 

47 handler_call_details: grpc.HandlerCallDetails, 

48 ) -> "grpc.RpcMethodHandler[Any, Any] | None": 

49 request_id = uuid.uuid4().hex 

50 current_request_id.set(request_id) 

51 

52 method = handler_call_details.method 

53 logger.info("request started: method=%s", method) 

54 

55 handler = continuation(handler_call_details) 

56 

57 if handler is None: 

58 logger.warning( 

59 "request finished: method=%s status=UNIMPLEMENTED (no handler)", method 

60 ) 

61 return None 

62 

63 if handler.unary_unary is not None: 63 ↛ 113line 63 didn't jump to line 113 because the condition on line 63 was always true

64 original_fn = handler.unary_unary 

65 

66 def logging_wrapper(request: Any, context: grpc.ServicerContext) -> Any: 

67 start = time.monotonic() 

68 try: 

69 response = original_fn(request, context) 

70 elapsed = time.monotonic() - start 

71 code = context.code() # type: ignore[attr-defined] 

72 status = code.name if code is not None else "OK" 

73 logger.info( 

74 "request finished: method=%s status=%s duration=%.3fs", 

75 method, 

76 status, 

77 elapsed, 

78 ) 

79 return response 

80 except Exception: 

81 elapsed = time.monotonic() - start 

82 code = context.code() # type: ignore[attr-defined] 

83 if code is not None: 

84 # Intentional abort (NOT_FOUND, PERMISSION_DENIED, etc.) 

85 # context.abort() sets state.code then raises bare Exception(). 

86 # Re-raise so the framework sends the intended error. 

87 logger.info( 

88 "request finished: method=%s status=%s duration=%.3fs", 

89 method, 

90 code.name, 

91 elapsed, 

92 ) 

93 raise 

94 else: 

95 # Unhandled exception (bug). Log full traceback server-side, 

96 # then replace with sanitized INTERNAL error. 

97 logger.exception( 

98 "unhandled exception: method=%s duration=%.3fs", 

99 method, 

100 elapsed, 

101 ) 

102 context.abort( 

103 grpc.StatusCode.INTERNAL, 

104 f"internal error (request {request_id})", 

105 ) 

106 

107 return grpc.unary_unary_rpc_method_handler( 

108 logging_wrapper, 

109 request_deserializer=handler.request_deserializer, 

110 response_serializer=handler.response_serializer, 

111 ) 

112 

113 return handler