Coverage for app/backend/src/couchers/middleware/proto_annotations.py: 95%

1""" 

2Reading our custom proto annotations (see proto/annotations.proto) off the descriptor pool. 

3 

4Everything that knows how descriptors, service/method options and extensions fit together lives here, so 

5callers work in terms of "the auth level for this method" rather than in terms of protobuf machinery. 

6 

7Lookups are cached on (pool, method), which is bounded because the pool is a process-wide singleton and the 

8descriptors fix the set of methods. functools.cache doesn't store exceptions, so a request naming a method 

9that doesn't exist raises every time rather than accumulating cache entries. 

10""" 

11 

12from functools import cache 

13from typing import Any, cast 

14 

15import grpc 

16from google.protobuf.descriptor import MethodDescriptor, ServiceDescriptor 

17from google.protobuf.descriptor_pool import DescriptorPool 

18from google.protobuf.message import Message 

19 

20from couchers.constants import ( 

21 MISSING_AUTH_LEVEL_ERROR_MESSAGE, 

22 NONEXISTENT_API_CALL_ERROR_MESSAGE, 

23) 

24from couchers.middleware.errors import CallRejectedError 

25from couchers.proto import annotations_pb2 

26from couchers.proto.annotations_pb2 import AuthLevel 

27 

28 

29def split_method(method: str) -> tuple[str, str]: 

30 """Split a gRPC method path, e.g. "/org.couchers.api.core.API/GetUser", into service and method names.""" 

31 _, service_name, method_name = method.split("/") 

32 return service_name, method_name 

33 

34 

35def find_service(pool: DescriptorPool, service_name: str) -> ServiceDescriptor: 

36 try: 

37 return cast(ServiceDescriptor, pool.FindServiceByName(service_name)) # type: ignore[no-untyped-call] 

38 except KeyError: 

39 raise CallRejectedError(NONEXISTENT_API_CALL_ERROR_MESSAGE, grpc.StatusCode.UNIMPLEMENTED) from None 

40 

41 

42def find_method(pool: DescriptorPool, method: str) -> MethodDescriptor: 

43 service_name, method_name = split_method(method) 

44 return cast(MethodDescriptor, find_service(pool, service_name).FindMethodByName(method_name)) # type: ignore[no-untyped-call] 

45 

46 

47def service_extension(pool: DescriptorPool, service_name: str, extension: Any) -> Message: 

48 """The value of a service-level extension; protobuf returns the default instance when it isn't set.""" 

49 return cast(Message, find_service(pool, service_name).GetOptions().Extensions[extension]) 

50 

51 

52def method_extension(pool: DescriptorPool, method: str, extension: Any) -> Message: 

53 """The value of a method-level extension; protobuf returns the default instance when it isn't set.""" 

54 return cast(Message, find_method(pool, method).GetOptions().Extensions[extension]) 

55 

56 

57def optional_field(message: Message, field: str) -> int | None: 

58 """Read an optional scalar field, honouring proto field presence, so an explicit 0 differs from unset.""" 

59 return getattr(message, field) if message.HasField(field) else None 

60 

61 

62@cache 

63def find_auth_level(pool: DescriptorPool, method: str) -> AuthLevel.ValueType: 

64 service_name, _ = split_method(method) 

65 level = find_service(pool, service_name).GetOptions().Extensions[annotations_pb2.auth_level] 

66 validate_auth_level(level) 

67 return level 

68 

69 

70def validate_auth_level(auth_level: AuthLevel.ValueType) -> None: 

71 # if unknown auth level, then it wasn't set and something's wrong 

72 if auth_level == annotations_pb2.AUTH_LEVEL_UNKNOWN: 

73 raise CallRejectedError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL) 

74 

75 if auth_level not in { 75 ↛ 82line 75 didn't jump to line 82 because the condition on line 75 was never true

76 annotations_pb2.AUTH_LEVEL_OPEN, 

77 annotations_pb2.AUTH_LEVEL_JAILED, 

78 annotations_pb2.AUTH_LEVEL_SECURE, 

79 annotations_pb2.AUTH_LEVEL_EDITOR, 

80 annotations_pb2.AUTH_LEVEL_ADMIN, 

81 }: 

82 raise CallRejectedError(MISSING_AUTH_LEVEL_ERROR_MESSAGE, grpc.StatusCode.INTERNAL)