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
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
1import logging
3import grpc
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
15logger = logging.getLogger(__name__)
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))
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 )
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 )
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 )
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()
93 if not subscription:
94 context.abort(grpc.StatusCode.NOT_FOUND, errors.CHAT_NOT_FOUND)
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)