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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

51 statements  

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, 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_mute_all(user_id): 

25 return _generate_unsubscribe_link( 

26 unsubscribe_pb2.UnsubscribePayload( 

27 user_id=user_id, 

28 all=unsubscribe_pb2.MuteAll(), 

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("all"): 

67 logger.info(f"User {user.name} unsubscribing from all") 

68 # todo: some other system when out of preview 

69 user.new_notifications_enabled = False 

70 return "You've been unsubscribed from all non-security notifications" 

71 if payload.HasField("topic_action"): 

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

73 topic = payload.topic_action.topic 

74 action = payload.topic_action.action 

75 topic_action = enum_from_topic_action[topic, action] 

76 # disable emails for this type 

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

78 return "You've been unsubscribed from all email notifications of that type" 

79 if payload.HasField("topic_key"): 

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

81 topic = payload.topic_key.topic 

82 key = payload.topic_key.key 

83 # a bunch of manual stuff 

84 if topic == "chat": 

85 group_chat_id = int(key) 

86 subscription = session.execute( 

87 select(GroupChatSubscription) 

88 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

90 .where(GroupChatSubscription.left == None) 

91 ).scalar_one_or_none() 

92 

93 if not subscription: 

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

95 

96 subscription.muted_until = DATETIME_INFINITY 

97 return "That group chat has been muted." 

98 else: 

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