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