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

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 

10 

11from couchers import errors, urls 

12from couchers.constants import DATETIME_INFINITY 

13from couchers.context import make_one_off_interactive_user_context 

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

15from couchers.models import GroupChatSubscription, HostingStatus, MeetupStatus, NotificationDeliveryType, User 

16from couchers.notifications import settings 

17from couchers.notifications.utils import enum_from_topic_action 

18from couchers.servicers.requests import Requests 

19from couchers.sql import couchers_select as select 

20from couchers.utils import now 

21from proto import conversations_pb2, requests_pb2 

22from proto.internal import unsubscribe_pb2 

23 

24logger = logging.getLogger(__name__) 

25 

26 

27def _generate_quick_link(payload): 

28 payload.created.FromDatetime(now()) 

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 generate_do_not_email(user): 

35 return _generate_quick_link( 

36 unsubscribe_pb2.UnsubscribePayload( 

37 user_id=user.id, 

38 do_not_email=unsubscribe_pb2.DoNotEmail(), 

39 ) 

40 ) 

41 

42 

43def generate_unsub_topic_key(notification): 

44 return _generate_quick_link( 

45 unsubscribe_pb2.UnsubscribePayload( 

46 user_id=notification.user_id, 

47 topic_key=unsubscribe_pb2.UnsubscribeTopicKey( 

48 topic=notification.topic, 

49 key=notification.key, 

50 ), 

51 ) 

52 ) 

53 

54 

55def generate_unsub_topic_action(notification): 

56 return _generate_quick_link( 

57 unsubscribe_pb2.UnsubscribePayload( 

58 user_id=notification.user_id, 

59 topic_action=unsubscribe_pb2.UnsubscribeTopicAction( 

60 topic=notification.topic, 

61 action=notification.action, 

62 ), 

63 ) 

64 ) 

65 

66 

67def generate_quick_decline_link(host_request): 

68 return _generate_quick_link( 

69 unsubscribe_pb2.UnsubscribePayload( 

70 user_id=host_request.host_user_id, 

71 host_request_quick_decline=unsubscribe_pb2.HostRequestQuickDecline( 

72 host_request_id=host_request.host_request_id, 

73 ), 

74 ) 

75 ) 

76 

77 

78def respond_quick_link(request, context, session): 

79 """ 

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

81 """ 

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

83 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.WRONG_SIGNATURE) 

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

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

86 if payload.HasField("do_not_email"): 

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

88 user.do_not_email = True 

89 user.hosting_status = HostingStatus.cant_host 

90 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

91 return "You will not receive any non-security emails, and your hosting status has been turned off. You may still receive the newsletter, and need to unsubscribe from it separately." 

92 if payload.HasField("topic_action"): 

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

94 topic = payload.topic_action.topic 

95 action = payload.topic_action.action 

96 topic_action = enum_from_topic_action[topic, action] 

97 # disable emails for this type 

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

99 return "You've been unsubscribed from email notifications of that type." 

100 if payload.HasField("topic_key"): 

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

102 topic = payload.topic_key.topic 

103 key = payload.topic_key.key 

104 # a bunch of manual stuff 

105 if topic == "chat": 

106 group_chat_id = int(key) 

107 subscription = session.execute( 

108 select(GroupChatSubscription) 

109 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

111 .where(GroupChatSubscription.left == None) 

112 ).scalar_one_or_none() 

113 

114 if not subscription: 

115 context.abort(grpc.StatusCode.NOT_FOUND, errors.CHAT_NOT_FOUND) 

116 

117 subscription.muted_until = DATETIME_INFINITY 

118 return "That group chat has been muted." 

119 else: 

120 context.abort(grpc.StatusCode.UNIMPLEMENTED, errors.CANT_UNSUB_TOPIC) 

121 if payload.HasField("host_request_quick_decline"): 

122 Requests().RespondHostRequest( 

123 request=requests_pb2.RespondHostRequestReq( 

124 host_request_id=payload.host_request_quick_decline.host_request_id, 

125 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

126 ), 

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

128 session=session, 

129 ) 

130 return "Thank you for responding to the host request!"