Coverage for app/backend/src/couchers/models/verification.py: 100%
76 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1import enum
2from datetime import date, datetime
3from typing import TYPE_CHECKING, Any
5from sqlalchemy import (
6 BigInteger,
7 Boolean,
8 CheckConstraint,
9 Date,
10 DateTime,
11 Enum,
12 ForeignKey,
13 Index,
14 String,
15 func,
16)
17from sqlalchemy import LargeBinary as Binary
18from sqlalchemy.ext.hybrid import hybrid_method, hybrid_property
19from sqlalchemy.orm import Mapped, column_property, mapped_column, relationship
20from sqlalchemy.sql.elements import ColumnElement
22from couchers.models.base import Base
23from couchers.utils import date_in_timezone, now
25if TYPE_CHECKING:
26 from couchers.models.users import User
29class StrongVerificationAttemptStatus(enum.Enum):
30 ## full data states
31 # completed, this now provides verification for a user
32 succeeded = enum.auto()
34 ## no data states
35 # in progress: waiting for the user to scan the Iris code or open the app
36 in_progress_waiting_on_user_to_open_app = enum.auto()
37 # in progress: waiting for the user to scan MRZ or NFC/chip
38 in_progress_waiting_on_user_in_app = enum.auto()
39 # in progress, waiting for backend to pull verification data
40 in_progress_waiting_on_backend = enum.auto()
41 # failed, no data
42 failed = enum.auto()
44 # duplicate, at our end, has data
45 duplicate = enum.auto()
47 ## minimal data states
48 # the data, except minimal deduplication data, was deleted
49 deleted = enum.auto()
52class PassportSex(enum.Enum):
53 """
54 We don't care about sex, we use gender on the platform. But passports apparently do.
55 """
57 male = enum.auto()
58 female = enum.auto()
59 unspecified = enum.auto()
62class StrongVerificationAttempt(Base, kw_only=True):
63 """
64 An attempt to perform strong verification
65 """
67 __tablename__ = "strong_verification_attempts"
69 # our verification id
70 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False)
72 # this is returned in the callback, and we look up the attempt via this
73 verification_attempt_token: Mapped[str] = mapped_column(String, unique=True)
75 user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
76 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False)
78 status: Mapped[StrongVerificationAttemptStatus] = mapped_column(
79 Enum(StrongVerificationAttemptStatus),
80 default=StrongVerificationAttemptStatus.in_progress_waiting_on_user_to_open_app,
81 )
83 ## full data
84 has_full_data: Mapped[bool] = mapped_column(Boolean, default=False)
85 # the data returned from iris, encrypted with a public key whose private key is kept offline
86 passport_encrypted_data: Mapped[bytes | None] = mapped_column(Binary, default=None)
87 passport_date_of_birth: Mapped[date | None] = mapped_column(Date, default=None)
88 passport_sex: Mapped[PassportSex | None] = mapped_column(Enum(PassportSex), default=None)
90 ## minimal data: this will not be deleted
91 has_minimal_data: Mapped[bool] = mapped_column(Boolean, default=False)
92 passport_expiry_date: Mapped[date | None] = mapped_column(Date, default=None)
93 passport_nationality: Mapped[str | None] = mapped_column(String, default=None)
94 # last three characters of the passport number
95 passport_last_three_document_chars: Mapped[str | None] = mapped_column(String, default=None)
97 iris_token: Mapped[str] = mapped_column(String, unique=True)
98 iris_session_id: Mapped[int] = mapped_column(BigInteger, unique=True)
100 passport_expiry_datetime = column_property(date_in_timezone(passport_expiry_date, "Etc/UTC"))
102 user: Mapped[User] = relationship(init=False)
104 @hybrid_property
105 def is_valid(self) -> bool:
106 """
107 This only checks whether the attempt is a success and the passport is not expired, use `has_strong_verification` for full check
108 """
109 return (self.status == StrongVerificationAttemptStatus.succeeded) and (self.passport_expiry_datetime >= now())
111 @is_valid.inplace.expression
112 @classmethod
113 def _is_valid_expression(cls) -> ColumnElement[bool]:
114 return (cls.status == StrongVerificationAttemptStatus.succeeded) & (
115 func.coalesce(cls.passport_expiry_datetime >= func.now(), False)
116 )
118 @hybrid_property
119 def is_visible(self) -> bool:
120 return self.status != StrongVerificationAttemptStatus.deleted
122 @hybrid_method
123 def _is_own_user(self, user: User | type[User]) -> Any:
124 # an attempt only verifies its own user: as a SQL expression there's no control flow to carry that, so the
125 # identity check has to be part of the predicate or a caller can pair any attempt with any user
126 return self.user_id == user.id
128 @hybrid_method
129 def _raw_birthdate_match(self, user: User | type[User]) -> Any:
130 """Does not check whether the SV attempt itself is not expired"""
131 return self.passport_date_of_birth == user.birthdate
133 @hybrid_method
134 def matches_birthdate(self, user: User | type[User]) -> Any:
135 return self._is_own_user(user) & self.is_valid & self._raw_birthdate_match(user)
137 @hybrid_method
138 def _raw_gender_match(self, user: User | type[User]) -> Any:
139 """Does not check whether the SV attempt itself is not expired"""
140 return (
141 ((user.gender == "Woman") & (self.passport_sex == PassportSex.female)) # type: ignore[operator]
142 | ((user.gender == "Man") & (self.passport_sex == PassportSex.male))
143 | (self.passport_sex == PassportSex.unspecified)
144 | (user.has_passport_sex_gender_exception == True)
145 )
147 @hybrid_method
148 def matches_gender(self, user: User | type[User]) -> Any:
149 return self._is_own_user(user) & self.is_valid & self._raw_gender_match(user)
151 @hybrid_method
152 def has_strong_verification(self, user: User | type[User]) -> Any:
153 return self._is_own_user(user) & self.is_valid & self._raw_birthdate_match(user) & self._raw_gender_match(user)
155 __table_args__ = (
156 # used to look up verification status for a user
157 Index(
158 "ix_strong_verification_attempts_current",
159 user_id,
160 passport_expiry_date,
161 postgresql_where=status == StrongVerificationAttemptStatus.succeeded,
162 ),
163 # each passport can be verified only once
164 Index(
165 "ix_strong_verification_attempts_unique_succeeded",
166 passport_expiry_date,
167 passport_nationality,
168 passport_last_three_document_chars,
169 unique=True,
170 postgresql_where=(
171 (status == StrongVerificationAttemptStatus.succeeded)
172 | (status == StrongVerificationAttemptStatus.deleted)
173 ),
174 ),
175 # full data check
176 CheckConstraint(
177 "(has_full_data IS TRUE AND passport_encrypted_data IS NOT NULL AND passport_date_of_birth IS NOT NULL) OR \
178 (has_full_data IS FALSE AND passport_encrypted_data IS NULL AND passport_date_of_birth IS NULL)",
179 name="full_data_status",
180 ),
181 # minimal data check
182 CheckConstraint(
183 "(has_minimal_data IS TRUE AND passport_expiry_date IS NOT NULL AND passport_nationality IS NOT NULL AND passport_last_three_document_chars IS NOT NULL) OR \
184 (has_minimal_data IS FALSE AND passport_expiry_date IS NULL AND passport_nationality IS NULL AND passport_last_three_document_chars IS NULL)",
185 name="minimal_data_status",
186 ),
187 # note on implications: p => q iff ~p OR q
188 # full data implies minimal data, has_minimal_data => has_full_data
189 CheckConstraint(
190 "(has_full_data IS FALSE) OR (has_minimal_data IS TRUE)",
191 name="full_data_implies_minimal_data",
192 ),
193 # succeeded implies full data
194 CheckConstraint(
195 "(NOT (status = 'succeeded')) OR (has_full_data IS TRUE)",
196 name="succeeded_implies_full_data",
197 ),
198 # in_progress/failed implies no_data
199 CheckConstraint(
200 "(NOT ((status = 'in_progress_waiting_on_user_to_open_app') OR (status = 'in_progress_waiting_on_user_in_app') OR (status = 'in_progress_waiting_on_backend') OR (status = 'failed'))) OR (has_minimal_data IS FALSE)",
201 name="in_progress_failed_iris_implies_no_data",
202 ),
203 # deleted or duplicate implies minimal data
204 CheckConstraint(
205 "(NOT ((status = 'deleted') OR (status = 'duplicate'))) OR (has_minimal_data IS TRUE)",
206 name="deleted_duplicate_implies_minimal_data",
207 ),
208 )
211class StrongVerificationCallbackEvent(Base, kw_only=True):
212 __tablename__ = "strong_verification_callback_events"
214 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False)
215 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False)
217 verification_attempt_id: Mapped[int] = mapped_column(ForeignKey("strong_verification_attempts.id"), index=True)
219 iris_status: Mapped[str] = mapped_column(String)