Coverage for src/couchers/notifications/unsubscribe.py: 75%

1import logging 

2 

3import grpc 

4 

5from couchers import errors, urls 

6from couchers.constants import DATETIME_INFINITY 

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

8from couchers.db import session_scope 

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

10from couchers.notifications import settings 

11from couchers.notifications.utils import enum_from_topic_action 

12from couchers.sql import couchers_select as select 

13from proto.internal import unsubscribe_pb2 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18def _generate_unsubscribe_link(payload): 

19 msg = payload.SerializeToString() 

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

21 return urls.unsubscribe_link(payload=b64encode(msg), sig=b64encode(sig)) 

22 

23 

24def generate_do_not_email(user): 

25 return _generate_unsubscribe_link( 

26 unsubscribe_pb2.UnsubscribePayload( 

27 user_id=user.id, 

28 do_not_email=unsubscribe_pb2.DoNotEmail(), 

29 ) 

30 ) 

31 

32 

33def generate_unsub_topic_key(notification): 

34 return _generate_unsubscribe_link( 

35 unsubscribe_pb2.UnsubscribePayload( 

36 user_id=notification.user_id, 

37 topic_key=unsubscribe_pb2.UnsubscribeTopicKey( 

38 topic=notification.topic, 

39 key=notification.key, 

40 ), 

41 ) 

42 ) 

43 

44 

45def generate_unsub_topic_action(notification): 

46 return _generate_unsubscribe_link( 

47 unsubscribe_pb2.UnsubscribePayload( 

48 user_id=notification.user_id, 

49 topic_action=unsubscribe_pb2.UnsubscribeTopicAction( 

50 topic=notification.topic, 

51 action=notification.action, 

52 ), 

53 ) 

54 ) 

55 

56 

57def unsubscribe(request, context): 

58 """ 

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

60 """ 

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

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

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

64 with session_scope() as session: 

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

66 if payload.HasField("do_not_email"): 

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

68 user.do_not_email = True 

69 user.hosting_status = HostingStatus.cant_host 

70 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

71 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." 

72 if payload.HasField("topic_action"): 

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

74 topic = payload.topic_action.topic 

75 action = payload.topic_action.action 

76 topic_action = enum_from_topic_action[topic, action] 

77 # disable emails for this type 

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

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

80 if payload.HasField("topic_key"): 

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

82 topic = payload.topic_key.topic 

83 key = payload.topic_key.key 

84 # a bunch of manual stuff 

85 if topic == "chat": 

86 group_chat_id = int(key) 

87 subscription = session.execute( 

88 select(GroupChatSubscription) 

89 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

91 .where(GroupChatSubscription.left == None) 

92 ).scalar_one_or_none() 

93 

94 if not subscription: 

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

96 

97 subscription.muted_until = DATETIME_INFINITY 

98 return "That group chat has been muted." 

99 else: 

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