Coverage for app/backend/src/couchers/notifications/quick_links.py: 89%

1""" 

2Quick email actions, based on signed URLs that don't require login to submit, such as unsubscribe, quick decline 

3 

4It is called "unsubscribe" in some places for historical reasons (it was the first use case). 

5""" 

6 

7import logging 

8 

9import grpc 

10from google.protobuf.message import Message 

11 

12from couchers import urls 

13from couchers.context import CouchersContext 

14from couchers.crypto import UNSUBSCRIBE_KEY_NAME, b64encode, generate_hash_signature, get_secret, verify_hash_signature 

15from couchers.models import ( 

16 Notification, 

17 NotificationTopicAction, 

18 User, 

19) 

20from couchers.proto import requests_pb2 

21from couchers.proto.internal import unsubscribe_pb2 

22from couchers.utils import now 

23 

24logger = logging.getLogger(__name__) 

25 

26 

27def _generate_quick_link(payload: Message) -> str: 

28 payload.created.FromDatetime(now()) # type: ignore[attr-defined] 

29 msg = payload.SerializeToString() 

30 sig = generate_hash_signature(message=msg, key=get_secret(UNSUBSCRIBE_KEY_NAME)) 

31 return urls.quick_link(payload=b64encode(msg), sig=b64encode(sig)) 

32 

33 

34def decode_quick_link(payload: bytes, sig: bytes, context: CouchersContext) -> unsubscribe_pb2.UnsubscribePayload: 

35 if not verify_hash_signature(message=payload, key=get_secret(UNSUBSCRIBE_KEY_NAME), sig=sig): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "wrong_signature") 

37 

38 return unsubscribe_pb2.UnsubscribePayload.FromString(payload) 

39 

40 

41def generate_do_not_email(user: User) -> str: 

42 return _generate_quick_link( 

43 unsubscribe_pb2.UnsubscribePayload( 

44 user_id=user.id, 

45 do_not_email=unsubscribe_pb2.DoNotEmail(), 

46 ) 

47 ) 

48 

49 

50def generate_unsub_topic_key(notification: Notification) -> str: 

51 if not notification.key: 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

52 raise ValueError( 

53 f"Cannot generate topic_key unsubscribe link for notification with empty key " 

54 f"(topic_action={notification.topic_action})" 

55 ) 

56 return _generate_quick_link( 

57 unsubscribe_pb2.UnsubscribePayload( 

58 user_id=notification.user_id, 

59 topic_key=unsubscribe_pb2.UnsubscribeTopicKey( 

60 topic=notification.topic, 

61 key=notification.key, 

62 ), 

63 ) 

64 ) 

65 

66 

67def generate_unsub_topic_action(notification: Notification) -> str: 

68 return _generate_quick_link( 

69 unsubscribe_pb2.UnsubscribePayload( 

70 user_id=notification.user_id, 

71 topic_action=unsubscribe_pb2.UnsubscribeTopicAction( 

72 topic=notification.topic, 

73 action=notification.action, 

74 ), 

75 ) 

76 ) 

77 

78 

79def generate_quick_decline_link(host_request: requests_pb2.HostRequest) -> str: 

80 return _generate_quick_link( 

81 unsubscribe_pb2.UnsubscribePayload( 

82 user_id=host_request.host_user_id, 

83 host_request_quick_decline=unsubscribe_pb2.HostRequestQuickDecline( 

84 host_request_id=host_request.host_request_id, 

85 ), 

86 ) 

87 ) 

88 

89 

90def can_unsubscribe_topic_key(topic_action: NotificationTopicAction) -> bool: 

91 """ 

92 Determines whether a user can unsubscribe from a specific topic key 

93 (e.g. muting a specific chat). 

94 """ 

95 # Only chat__message has a meaningful key (the chat ID); chat__missed_messages is a summary with no specific chat 

96 return topic_action == NotificationTopicAction.chat__message