Coverage for src/couchers/notifications/quick_links.py: 78%

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.orm import Session 

12 

13from couchers import urls 

14from couchers.constants import DATETIME_INFINITY 

15from couchers.context import CouchersContext, make_one_off_interactive_user_context 

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

17from couchers.models import ( 

18 GroupChatSubscription, 

19 HostingStatus, 

20 MeetupStatus, 

21 Notification, 

22 NotificationDeliveryType, 

23 User, 

24) 

25from couchers.notifications import settings 

26from couchers.notifications.utils import enum_from_topic_action 

27from couchers.proto import auth_pb2, conversations_pb2, requests_pb2 

28from couchers.proto.internal import unsubscribe_pb2 

29from couchers.servicers.requests import Requests 

30from couchers.sql import couchers_select as select 

31from couchers.utils import now 

32 

33logger = logging.getLogger(__name__) 

34 

35 

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

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

38 msg = payload.SerializeToString() 

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

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

41 

42 

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

44 return _generate_quick_link( 

45 unsubscribe_pb2.UnsubscribePayload( 

46 user_id=user.id, 

47 do_not_email=unsubscribe_pb2.DoNotEmail(), 

48 ) 

49 ) 

50 

51 

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

53 return _generate_quick_link( 

54 unsubscribe_pb2.UnsubscribePayload( 

55 user_id=notification.user_id, 

56 topic_key=unsubscribe_pb2.UnsubscribeTopicKey( 

57 topic=notification.topic, 

58 key=notification.key, 

59 ), 

60 ) 

61 ) 

62 

63 

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

65 return _generate_quick_link( 

66 unsubscribe_pb2.UnsubscribePayload( 

67 user_id=notification.user_id, 

68 topic_action=unsubscribe_pb2.UnsubscribeTopicAction( 

69 topic=notification.topic, 

70 action=notification.action, 

71 ), 

72 ) 

73 ) 

74 

75 

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

77 return _generate_quick_link( 

78 unsubscribe_pb2.UnsubscribePayload( 

79 user_id=host_request.host_user_id, 

80 host_request_quick_decline=unsubscribe_pb2.HostRequestQuickDecline( 

81 host_request_id=host_request.host_request_id, 

82 ), 

83 ) 

84 ) 

85 

86 

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

88 """ 

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

90 """ 

91 if not verify_hash_signature(message=request.payload, key=get_secret(UNSUBSCRIBE_KEY_NAME), sig=request.sig): 

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

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

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

95 if payload.HasField("do_not_email"): 

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

97 user.do_not_email = True 

98 user.hosting_status = HostingStatus.cant_host 

99 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

100 return context.get_localized_string("quick_links", "do_not_email") 

101 if payload.HasField("topic_action"): 

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

103 topic = payload.topic_action.topic 

104 action = payload.topic_action.action 

105 topic_action = enum_from_topic_action[topic, action] 

106 # disable emails for this type 

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

108 return context.get_localized_string("quick_links", "topic_action") 

109 if payload.HasField("topic_key"): 

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

111 topic = payload.topic_key.topic 

112 key = payload.topic_key.key 

113 # a bunch of manual stuff 

114 if topic == "chat": 

115 group_chat_id = int(key) 

116 subscription = session.execute( 

117 select(GroupChatSubscription) 

118 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

120 .where(GroupChatSubscription.left == None) 

121 ).scalar_one_or_none() 

122 

123 if subscription is None: 

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

125 

126 assert subscription is not None 

127 subscription.muted_until = DATETIME_INFINITY 

128 return context.get_localized_string("quick_links", "chat_unsub") 

129 else: 

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

131 if payload.HasField("host_request_quick_decline"): 

132 Requests().RespondHostRequest( # type: ignore[no-untyped-call] 

133 request=requests_pb2.RespondHostRequestReq( 

134 host_request_id=payload.host_request_quick_decline.host_request_id, 

135 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

136 ), 

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

138 session=session, 

139 ) 

140 return context.get_localized_string("quick_links", "host_request_quick_decline") 

141 raise Exception("Unhandled quick link type")