Coverage for app / backend / src / couchers / notifications / quick_links.py: 75%
65 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-03 06:18 +0000
1"""
2Quick email actions, based on signed URLs that don't require login to submit, such as unsubscribe, quick decline
4It is called "unsubscribe" in some places for historical reasons (it was the first use case).
5"""
7import logging
9import grpc
10from google.protobuf.message import Message
11from sqlalchemy import select
12from sqlalchemy.orm import Session
14from couchers import urls
15from couchers.constants import DATETIME_INFINITY
16from couchers.context import CouchersContext, make_one_off_interactive_user_context
17from couchers.crypto import UNSUBSCRIBE_KEY_NAME, b64encode, generate_hash_signature, get_secret, verify_hash_signature
18from couchers.models import (
19 GroupChat,
20 GroupChatSubscription,
21 HostingStatus,
22 MeetupStatus,
23 Notification,
24 NotificationDeliveryType,
25 User,
26)
27from couchers.notifications import settings
28from couchers.notifications.utils import enum_from_topic_action
29from couchers.proto import auth_pb2, conversations_pb2, requests_pb2
30from couchers.proto.internal import unsubscribe_pb2
31from couchers.servicers.requests import Requests
32from couchers.sql import where_moderated_content_visible
33from couchers.utils import now
35logger = logging.getLogger(__name__)
38def _generate_quick_link(payload: Message) -> str:
39 payload.created.FromDatetime(now()) # type: ignore[attr-defined]
40 msg = payload.SerializeToString()
41 sig = generate_hash_signature(message=msg, key=get_secret(UNSUBSCRIBE_KEY_NAME))
42 return urls.quick_link(payload=b64encode(msg), sig=b64encode(sig))
45def generate_do_not_email(user: User) -> str:
46 return _generate_quick_link(
47 unsubscribe_pb2.UnsubscribePayload(
48 user_id=user.id,
49 do_not_email=unsubscribe_pb2.DoNotEmail(),
50 )
51 )
54def generate_unsub_topic_key(notification: Notification) -> str:
55 return _generate_quick_link(
56 unsubscribe_pb2.UnsubscribePayload(
57 user_id=notification.user_id,
58 topic_key=unsubscribe_pb2.UnsubscribeTopicKey(
59 topic=notification.topic,
60 key=notification.key,
61 ),
62 )
63 )
66def generate_unsub_topic_action(notification: Notification) -> str:
67 return _generate_quick_link(
68 unsubscribe_pb2.UnsubscribePayload(
69 user_id=notification.user_id,
70 topic_action=unsubscribe_pb2.UnsubscribeTopicAction(
71 topic=notification.topic,
72 action=notification.action,
73 ),
74 )
75 )
78def generate_quick_decline_link(host_request: requests_pb2.HostRequest) -> str:
79 return _generate_quick_link(
80 unsubscribe_pb2.UnsubscribePayload(
81 user_id=host_request.host_user_id,
82 host_request_quick_decline=unsubscribe_pb2.HostRequestQuickDecline(
83 host_request_id=host_request.host_request_id,
84 ),
85 )
86 )
89def respond_quick_link(request: auth_pb2.UnsubscribeReq, context: CouchersContext, session: Session) -> str:
90 """
91 Returns a response string or uses context.abort upon error
92 """
93 if not verify_hash_signature(message=request.payload, key=get_secret(UNSUBSCRIBE_KEY_NAME), sig=request.sig): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "wrong_signature")
95 payload = unsubscribe_pb2.UnsubscribePayload.FromString(request.payload)
96 user = session.execute(select(User).where(User.id == payload.user_id)).scalar_one()
97 if payload.HasField("do_not_email"):
98 logger.info(f"User {user.name} turning of emails")
99 user.do_not_email = True
100 user.hosting_status = HostingStatus.cant_host
101 user.meetup_status = MeetupStatus.does_not_want_to_meetup
102 return context.get_localized_string("quick_links.do_not_email")
103 if payload.HasField("topic_action"):
104 logger.info(f"User {user.name} unsubscribing from topic_action")
105 topic = payload.topic_action.topic
106 action = payload.topic_action.action
107 topic_action = enum_from_topic_action[topic, action]
108 # disable emails for this type
109 settings.set_preference(session, user.id, topic_action, NotificationDeliveryType.email, False)
110 return context.get_localized_string("quick_links.topic_action")
111 if payload.HasField("topic_key"): 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true
112 logger.info(f"User {user.name} unsubscribing from topic_key")
113 topic = payload.topic_key.topic
114 key = payload.topic_key.key
115 # a bunch of manual stuff
116 if topic == "chat":
117 group_chat_id = int(key)
118 subscription = session.execute(
119 where_moderated_content_visible(
120 select(GroupChatSubscription).join(
121 GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id
122 ),
123 context,
124 GroupChat,
125 is_list_operation=False,
126 )
127 .where(GroupChatSubscription.group_chat_id == group_chat_id)
128 .where(GroupChatSubscription.user_id == user.id)
129 .where(GroupChatSubscription.left == None)
130 ).scalar_one_or_none()
132 if subscription is None:
133 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
135 subscription.muted_until = DATETIME_INFINITY
136 return context.get_localized_string("quick_links.chat_unsub")
137 else:
138 context.abort_with_error_code(grpc.StatusCode.UNIMPLEMENTED, "cant_unsub_topic")
139 if payload.HasField("host_request_quick_decline"): 139 ↛ 149line 139 didn't jump to line 149 because the condition on line 139 was always true
140 Requests().RespondHostRequest(
141 request=requests_pb2.RespondHostRequestReq(
142 host_request_id=payload.host_request_quick_decline.host_request_id,
143 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED,
144 ),
145 context=make_one_off_interactive_user_context(couchers_context=context, user_id=payload.user_id),
146 session=session,
147 )
148 return context.get_localized_string("quick_links.host_request_quick_decline")
149 raise Exception("Unhandled quick link type")