Coverage for app / backend / src / couchers / servicers / postal_verification.py: 92%
124 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-03 06:18 +0000
1import json
2import logging
4import grpc
5from google.protobuf import empty_pb2
6from sqlalchemy import exists, select
7from sqlalchemy.orm import Session
9from couchers.config import config
10from couchers.constants import (
11 POSTAL_VERIFICATION_CODE_LIFETIME,
12 POSTAL_VERIFICATION_MAX_ATTEMPTS,
13 POSTAL_VERIFICATION_RATE_LIMIT,
14)
15from couchers.context import CouchersContext
16from couchers.helpers.postal_verification import generate_postal_verification_code, has_postal_verification
17from couchers.jobs.enqueue import queue_job
18from couchers.jobs.handlers import send_postal_verification_postcard
19from couchers.models import User
20from couchers.models.notifications import NotificationTopicAction
21from couchers.models.postal_verification import PostalVerificationAttempt, PostalVerificationStatus
22from couchers.notifications.notify import notify
23from couchers.postal.address_validation import AddressValidationError, validate_address
24from couchers.proto import notification_data_pb2, postal_verification_pb2, postal_verification_pb2_grpc
25from couchers.proto.internal import jobs_pb2
26from couchers.utils import Timestamp_from_datetime, now
28logger = logging.getLogger(__name__)
30postalverificationstatus2pb = {
31 PostalVerificationStatus.pending_address_confirmation: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_PENDING_ADDRESS_CONFIRMATION,
32 PostalVerificationStatus.in_progress: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_IN_PROGRESS,
33 PostalVerificationStatus.awaiting_verification: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_AWAITING_VERIFICATION,
34 PostalVerificationStatus.succeeded: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_SUCCEEDED,
35 PostalVerificationStatus.failed: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_FAILED,
36 PostalVerificationStatus.cancelled: postal_verification_pb2.POSTAL_VERIFICATION_STATUS_CANCELLED,
37}
40def _attempt_to_address_pb(attempt: PostalVerificationAttempt) -> postal_verification_pb2.PostalAddress:
41 return postal_verification_pb2.PostalAddress(
42 address_line_1=attempt.address_line_1,
43 address_line_2=attempt.address_line_2 or "",
44 city=attempt.city,
45 state=attempt.state or "",
46 postal_code=attempt.postal_code or "",
47 country=attempt.country,
48 )
51class PostalVerification(postal_verification_pb2_grpc.PostalVerificationServicer):
52 def InitiatePostalVerification(
53 self,
54 request: postal_verification_pb2.InitiatePostalVerificationReq,
55 context: CouchersContext,
56 session: Session,
57 ) -> postal_verification_pb2.InitiatePostalVerificationRes:
58 """
59 Step 1: User submits address for validation.
60 """
61 if not config["ENABLE_POSTAL_VERIFICATION"]:
62 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "postal_verification_disabled")
64 # Check if there's an active attempt
65 has_active_attempt = session.execute(
66 select(
67 exists(
68 select(PostalVerificationAttempt)
69 .where(PostalVerificationAttempt.user_id == context.user_id)
70 .where(
71 PostalVerificationAttempt.status.in_(
72 [
73 PostalVerificationStatus.pending_address_confirmation,
74 PostalVerificationStatus.in_progress,
75 PostalVerificationStatus.awaiting_verification,
76 ]
77 )
78 )
79 )
80 )
81 ).scalar()
83 if has_active_attempt:
84 context.abort_with_error_code(
85 grpc.StatusCode.FAILED_PRECONDITION, "postal_verification_already_in_progress"
86 )
88 # Check rate limit: one initiation per 30 days
89 has_recent_attempt = session.execute(
90 select(
91 exists(
92 select(PostalVerificationAttempt)
93 .where(PostalVerificationAttempt.user_id == context.user_id)
94 .where(PostalVerificationAttempt.created > now() - POSTAL_VERIFICATION_RATE_LIMIT)
95 )
96 )
97 ).scalar()
99 if has_recent_attempt:
100 context.abort_with_error_code(grpc.StatusCode.RESOURCE_EXHAUSTED, "postal_verification_rate_limited")
102 # Validate required fields
103 if not request.address.address_line_1:
104 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "address_line_1_required")
105 if not request.address.city:
106 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "city_required")
107 if not request.address.country:
108 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "country_required")
110 # Validate address
111 try:
112 validated = validate_address(
113 address_line_1=request.address.address_line_1,
114 address_line_2=request.address.address_line_2 or None,
115 city=request.address.city,
116 state=request.address.state or None,
117 postal_code=request.address.postal_code or None,
118 country=request.address.country,
119 )
120 except AddressValidationError:
121 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "postal_address_invalid")
123 if not validated.is_deliverable: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "postal_address_undeliverable")
126 # Create attempt
127 attempt = PostalVerificationAttempt(
128 user_id=context.user_id,
129 status=PostalVerificationStatus.pending_address_confirmation,
130 address_line_1=validated.address_line_1,
131 address_line_2=validated.address_line_2,
132 city=validated.city,
133 state=validated.state,
134 postal_code=validated.postal_code,
135 country=validated.country,
136 original_address_json=json.dumps(
137 {
138 "address_line_1": request.address.address_line_1,
139 "address_line_2": request.address.address_line_2,
140 "city": request.address.city,
141 "state": request.address.state,
142 "postal_code": request.address.postal_code,
143 "country": request.address.country,
144 }
145 ),
146 )
147 session.add(attempt)
148 session.flush()
150 return postal_verification_pb2.InitiatePostalVerificationRes(
151 postal_verification_attempt_id=attempt.id,
152 corrected_address=postal_verification_pb2.PostalAddress(
153 address_line_1=validated.address_line_1,
154 address_line_2=validated.address_line_2 or "",
155 city=validated.city,
156 state=validated.state or "",
157 postal_code=validated.postal_code or "",
158 country=validated.country,
159 ),
160 address_was_corrected=validated.was_corrected,
161 )
163 def ConfirmPostalAddress(
164 self,
165 request: postal_verification_pb2.ConfirmPostalAddressReq,
166 context: CouchersContext,
167 session: Session,
168 ) -> postal_verification_pb2.ConfirmPostalAddressRes:
169 """
170 Step 2: User confirms address, we generate code and send postcard.
171 """
172 attempt = session.execute(
173 select(PostalVerificationAttempt)
174 .where(PostalVerificationAttempt.id == request.postal_verification_attempt_id)
175 .where(PostalVerificationAttempt.user_id == context.user_id)
176 ).scalar_one_or_none()
178 if not attempt:
179 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "postal_verification_attempt_not_found")
181 if attempt.status != PostalVerificationStatus.pending_address_confirmation: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "postal_verification_wrong_state")
184 attempt.verification_code = generate_postal_verification_code()
185 attempt.status = PostalVerificationStatus.in_progress
186 attempt.address_confirmed_at = now()
188 # Queue background job to send postcard
189 queue_job(
190 session,
191 job=send_postal_verification_postcard,
192 payload=jobs_pb2.SendPostalVerificationPostcardPayload(
193 postal_verification_attempt_id=attempt.id,
194 ),
195 )
197 return postal_verification_pb2.ConfirmPostalAddressRes()
199 def GetPostalVerificationStatus(
200 self,
201 request: postal_verification_pb2.GetPostalVerificationStatusReq,
202 context: CouchersContext,
203 session: Session,
204 ) -> postal_verification_pb2.GetPostalVerificationStatusRes:
205 """
206 Returns the user's postal verification status and current/latest attempt details.
207 """
208 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
210 has_verification = has_postal_verification(session, user)
212 # Always get the latest attempt for determining can_initiate and has_active_attempt
213 latest_attempt = session.execute(
214 select(PostalVerificationAttempt)
215 .where(PostalVerificationAttempt.user_id == user.id)
216 .order_by(PostalVerificationAttempt.created.desc())
217 .limit(1)
218 ).scalar_one_or_none()
220 # Check if user can initiate a new attempt (based on latest attempt)
221 can_initiate = True
222 next_attempt_allowed_at = None
223 has_active_attempt = False
225 if latest_attempt:
226 # Can't initiate if there's an active attempt
227 if latest_attempt.status in [
228 PostalVerificationStatus.pending_address_confirmation,
229 PostalVerificationStatus.in_progress,
230 PostalVerificationStatus.awaiting_verification,
231 ]:
232 can_initiate = False
233 has_active_attempt = True
234 else:
235 # Check rate limit
236 time_since_last = now() - latest_attempt.created
237 if time_since_last < POSTAL_VERIFICATION_RATE_LIMIT: 237 ↛ 241line 237 didn't jump to line 241 because the condition on line 237 was always true
238 can_initiate = False
239 next_attempt_allowed_at = latest_attempt.created + POSTAL_VERIFICATION_RATE_LIMIT
241 res = postal_verification_pb2.GetPostalVerificationStatusRes(
242 has_postal_verification=has_verification,
243 can_initiate_new_attempt=can_initiate,
244 has_active_attempt=has_active_attempt,
245 )
247 if next_attempt_allowed_at:
248 res.next_attempt_allowed_at.CopyFrom(Timestamp_from_datetime(next_attempt_allowed_at))
250 # Get specific attempt if requested, otherwise use latest
251 if request.postal_verification_attempt_id: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 attempt = session.execute(
253 select(PostalVerificationAttempt)
254 .where(PostalVerificationAttempt.id == request.postal_verification_attempt_id)
255 .where(PostalVerificationAttempt.user_id == context.user_id)
256 ).scalar_one_or_none()
257 else:
258 attempt = latest_attempt
260 if attempt:
261 res.postal_verification_attempt_id = attempt.id
262 res.status = postalverificationstatus2pb.get(
263 attempt.status, postal_verification_pb2.POSTAL_VERIFICATION_STATUS_UNKNOWN
264 )
265 res.address.CopyFrom(_attempt_to_address_pb(attempt))
266 res.created.CopyFrom(Timestamp_from_datetime(attempt.created))
267 if attempt.postcard_sent_at:
268 res.postcard_sent_at.CopyFrom(Timestamp_from_datetime(attempt.postcard_sent_at))
270 return res
272 def VerifyPostalCode(
273 self,
274 request: postal_verification_pb2.VerifyPostalCodeReq,
275 context: CouchersContext,
276 session: Session,
277 ) -> postal_verification_pb2.VerifyPostalCodeRes:
278 """
279 User submits the code from the postcard.
280 Looks up the user's active attempt (awaiting_verification status).
281 """
282 attempt = session.execute(
283 select(PostalVerificationAttempt)
284 .where(PostalVerificationAttempt.user_id == context.user_id)
285 .where(PostalVerificationAttempt.status == PostalVerificationStatus.awaiting_verification)
286 ).scalar_one_or_none()
288 if not attempt: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true
289 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "postal_verification_attempt_not_found")
291 # Check code expiry
292 if attempt.postcard_sent_at and (now() - attempt.postcard_sent_at) > POSTAL_VERIFICATION_CODE_LIFETIME:
293 attempt.status = PostalVerificationStatus.failed
294 notify(
295 session,
296 user_id=context.user_id,
297 topic_action=NotificationTopicAction.postal_verification__failed,
298 key="",
299 data=notification_data_pb2.PostalVerificationFailed(
300 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED
301 ),
302 )
303 session.commit()
304 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "postal_verification_code_expired")
306 # Normalize submitted code
307 submitted_code = request.code.strip().upper()
309 if submitted_code != attempt.verification_code:
310 attempt.code_attempts += 1
311 remaining = POSTAL_VERIFICATION_MAX_ATTEMPTS - attempt.code_attempts
313 if remaining <= 0:
314 attempt.status = PostalVerificationStatus.failed
315 notify(
316 session,
317 user_id=context.user_id,
318 topic_action=NotificationTopicAction.postal_verification__failed,
319 key="",
320 data=notification_data_pb2.PostalVerificationFailed(
321 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS
322 ),
323 )
324 return postal_verification_pb2.VerifyPostalCodeRes(
325 success=False,
326 remaining_attempts=0,
327 )
329 return postal_verification_pb2.VerifyPostalCodeRes(
330 success=False,
331 remaining_attempts=remaining,
332 )
334 # Success!
335 attempt.status = PostalVerificationStatus.succeeded
336 attempt.verified_at = now()
338 notify(
339 session,
340 user_id=context.user_id,
341 topic_action=NotificationTopicAction.postal_verification__success,
342 key="",
343 )
345 return postal_verification_pb2.VerifyPostalCodeRes(
346 success=True,
347 remaining_attempts=0,
348 )
350 def CancelPostalVerification(
351 self,
352 request: postal_verification_pb2.CancelPostalVerificationReq,
353 context: CouchersContext,
354 session: Session,
355 ) -> empty_pb2.Empty:
356 """
357 Cancels an active postal verification attempt.
358 """
359 attempt = session.execute(
360 select(PostalVerificationAttempt)
361 .where(PostalVerificationAttempt.id == request.postal_verification_attempt_id)
362 .where(PostalVerificationAttempt.user_id == context.user_id)
363 ).scalar_one_or_none()
365 if not attempt:
366 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "postal_verification_attempt_not_found")
368 # Can cancel any active attempt (not terminal states)
369 if attempt.status not in [ 369 ↛ 374line 369 didn't jump to line 374 because the condition on line 369 was never true
370 PostalVerificationStatus.pending_address_confirmation,
371 PostalVerificationStatus.in_progress,
372 PostalVerificationStatus.awaiting_verification,
373 ]:
374 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "postal_verification_cannot_cancel")
376 attempt.status = PostalVerificationStatus.cancelled
377 # Clear the verification code (required by db constraint and makes sense - code is no longer valid)
378 attempt.verification_code = None
380 return empty_pb2.Empty()
382 def ListPostalVerificationAttempts(
383 self,
384 request: postal_verification_pb2.ListPostalVerificationAttemptsReq,
385 context: CouchersContext,
386 session: Session,
387 ) -> postal_verification_pb2.ListPostalVerificationAttemptsRes:
388 """
389 Returns all postal verification attempts for the user.
390 """
391 attempts = session.execute(
392 select(PostalVerificationAttempt)
393 .where(PostalVerificationAttempt.user_id == context.user_id)
394 .order_by(PostalVerificationAttempt.created.desc())
395 ).scalars()
397 return postal_verification_pb2.ListPostalVerificationAttemptsRes(
398 attempts=[
399 postal_verification_pb2.PostalVerificationAttemptSummary(
400 postal_verification_attempt_id=attempt.id,
401 status=postalverificationstatus2pb.get(
402 attempt.status, postal_verification_pb2.POSTAL_VERIFICATION_STATUS_UNKNOWN
403 ),
404 address=_attempt_to_address_pb(attempt),
405 created=Timestamp_from_datetime(attempt.created),
406 verified_at=Timestamp_from_datetime(attempt.verified_at) if attempt.verified_at else None,
407 )
408 for attempt in attempts
409 ]
410 )