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 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: str | NotificationTopicAction) -> bool: 

91 """ 

92 Determines whether a user can unsubscribe from all notification actions 

93 concerning a given topic. 

94 """ 

95 if isinstance(topic, NotificationTopicAction): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 topic = topic.topic 

97 # We currently only support unsubscribing from chat topics 

98 return topic == NotificationTopicAction.chat__message.topic 

99 

100 

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

102 """ 

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

104 """ 

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

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

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

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

109 if payload.HasField("do_not_email"): 

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

111 user.do_not_email = True 

112 user.hosting_status = HostingStatus.cant_host 

113 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

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

115 if payload.HasField("topic_action"): 

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

117 topic = payload.topic_action.topic 

118 action = payload.topic_action.action 

119 topic_action = enum_from_topic_action[topic, action] 

120 # disable emails for this type 

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

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

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

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

125 topic = payload.topic_key.topic 

126 key = payload.topic_key.key 

127 # a bunch of manual stuff 

128 if topic == "chat": 

129 group_chat_id = int(key) 

130 subscription = session.execute( 

131 where_moderated_content_visible( 

132 select(GroupChatSubscription).join( 

133 GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id 

134 ), 

135 context, 

136 GroupChat, 

137 is_list_operation=False, 

138 ) 

139 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

141 .where(GroupChatSubscription.left == None) 

142 ).scalar_one_or_none() 

143 

144 if subscription is None: 

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

146 

147 subscription.muted_until = DATETIME_INFINITY 

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

149 else: 

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

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

152 Requests().RespondHostRequest( 

153 request=requests_pb2.RespondHostRequestReq( 

154 host_request_id=payload.host_request_quick_decline.host_request_id, 

155 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

156 ), 

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

158 session=session, 

159 ) 

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

161 raise Exception("Unhandled quick link type")