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

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 

11from sqlalchemy import select 

12from sqlalchemy.orm import Session 

13 

14from couchers import urls 

15from couchers.constants import DATETIME_INFINITY 

16from couchers.context import CouchersContext, make_one_off_interactive_user_context 

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

18from couchers.models import ( 

19 GroupChat, 

20 GroupChatSubscription, 

21 HostingStatus, 

22 MeetupStatus, 

23 Notification, 

24 NotificationDeliveryType, 

25 NotificationTopicAction, 

26 User, 

27) 

28from couchers.notifications import settings 

29from couchers.notifications.utils import enum_from_topic_action 

30from couchers.proto import auth_pb2, conversations_pb2, requests_pb2 

31from couchers.proto.internal import unsubscribe_pb2 

32from couchers.servicers.requests import Requests 

33from couchers.sql import where_moderated_content_visible 

34from couchers.utils import now 

35 

36logger = logging.getLogger(__name__) 

37 

38 

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

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

41 msg = payload.SerializeToString() 

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

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

44 

45 

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

47 return _generate_quick_link( 

48 unsubscribe_pb2.UnsubscribePayload( 

49 user_id=user.id, 

50 do_not_email=unsubscribe_pb2.DoNotEmail(), 

51 ) 

52 ) 

53 

54 

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

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

57 raise ValueError( 

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

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

60 ) 

61 return _generate_quick_link( 

62 unsubscribe_pb2.UnsubscribePayload( 

63 user_id=notification.user_id, 

64 topic_key=unsubscribe_pb2.UnsubscribeTopicKey( 

65 topic=notification.topic, 

66 key=notification.key, 

67 ), 

68 ) 

69 ) 

70 

71 

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

73 return _generate_quick_link( 

74 unsubscribe_pb2.UnsubscribePayload( 

75 user_id=notification.user_id, 

76 topic_action=unsubscribe_pb2.UnsubscribeTopicAction( 

77 topic=notification.topic, 

78 action=notification.action, 

79 ), 

80 ) 

81 ) 

82 

83 

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

85 return _generate_quick_link( 

86 unsubscribe_pb2.UnsubscribePayload( 

87 user_id=host_request.host_user_id, 

88 host_request_quick_decline=unsubscribe_pb2.HostRequestQuickDecline( 

89 host_request_id=host_request.host_request_id, 

90 ), 

91 ) 

92 ) 

93 

94 

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

96 """ 

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

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

99 """ 

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

101 return topic_action == NotificationTopicAction.chat__message 

102 

103 

104def respond_quick_link(request: auth_pb2.UnsubscribeReq, context: CouchersContext, session: Session) -> str: 

105 """ 

106 Returns a response string or uses context.abort upon error 

107 """ 

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

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

110 payload = unsubscribe_pb2.UnsubscribePayload.FromString(request.payload) 

111 user = session.execute(select(User).where(User.id == payload.user_id)).scalar_one() 

112 if payload.HasField("do_not_email"): 

113 logger.info(f"User {user.name} turning of emails") 

114 user.do_not_email = True 

115 user.hosting_status = HostingStatus.cant_host 

116 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

117 return context.localization.localize_string("quick_links.do_not_email") 

118 if payload.HasField("topic_action"): 

119 logger.info(f"User {user.name} unsubscribing from topic_action") 

120 topic = payload.topic_action.topic 

121 action = payload.topic_action.action 

122 topic_action = enum_from_topic_action[topic, action] 

123 # disable emails for this type 

124 settings.set_preference(session, user.id, topic_action, NotificationDeliveryType.email, False) 

125 return context.localization.localize_string("quick_links.topic_action") 

126 if payload.HasField("topic_key"): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 logger.info(f"User {user.name} unsubscribing from topic_key") 

128 topic = payload.topic_key.topic 

129 key = payload.topic_key.key 

130 # a bunch of manual stuff 

131 if topic == "chat": 

132 group_chat_id = int(key) 

133 subscription = session.execute( 

134 where_moderated_content_visible( 

135 select(GroupChatSubscription).join( 

136 GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id 

137 ), 

138 context, 

139 GroupChat, 

140 is_list_operation=False, 

141 ) 

142 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

143 .where(GroupChatSubscription.user_id == user.id) 

144 .where(GroupChatSubscription.left == None) 

145 ).scalar_one_or_none() 

146 

147 if subscription is None: 

148 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

149 

150 subscription.muted_until = DATETIME_INFINITY 

151 return context.localization.localize_string("quick_links.chat_unsub") 

152 else: 

153 context.abort_with_error_code(grpc.StatusCode.UNIMPLEMENTED, "cant_unsub_topic") 

154 if payload.HasField("host_request_quick_decline"): 154 ↛ 164line 154 didn't jump to line 164 because the condition on line 154 was always true

155 Requests().RespondHostRequest( 

156 request=requests_pb2.RespondHostRequestReq( 

157 host_request_id=payload.host_request_quick_decline.host_request_id, 

158 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

159 ), 

160 context=make_one_off_interactive_user_context(couchers_context=context, user_id=payload.user_id), 

161 session=session, 

162 ) 

163 return context.localization.localize_string("quick_links.host_request_quick_decline") 

164 raise Exception("Unhandled quick link type")