Coverage for src/couchers/models/verification.py: 100%
71 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-25 10:58 +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
21from couchers.models.base import Base
22from couchers.utils import date_in_timezone, now
24if TYPE_CHECKING:
25 from couchers.models.users import User
28class StrongVerificationAttemptStatus(enum.Enum):
29 ## full data states
30 # completed, this now provides verification for a user
31 succeeded = enum.auto()
33 ## no data states
34 # in progress: waiting for the user to scan the Iris code or open the app
35 in_progress_waiting_on_user_to_open_app = enum.auto()
36 # in progress: waiting for the user to scan MRZ or NFC/chip
37 in_progress_waiting_on_user_in_app = enum.auto()
38 # in progress, waiting for backend to pull verification data
39 in_progress_waiting_on_backend = enum.auto()
40 # failed, no data
41 failed = enum.auto()
43 # duplicate, at our end, has data
44 duplicate = enum.auto()
46 ## minimal data states
47 # the data, except minimal deduplication data, was deleted
48 deleted = enum.auto()
51class PassportSex(enum.Enum):
52 """
53 We don't care about sex, we use gender on the platform. But passports apparently do.
54 """
56 male = enum.auto()
57 female = enum.auto()
58 unspecified = enum.auto()
61class StrongVerificationAttempt(Base):
62 """
63 An attempt to perform strong verification
64 """
66 __tablename__ = "strong_verification_attempts"
68 # our verification id
69 id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
71 # this is returned in the callback, and we look up the attempt via this
72 verification_attempt_token: Mapped[str] = mapped_column(String, unique=True)
74 user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
75 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
77 status: Mapped[StrongVerificationAttemptStatus] = mapped_column(
78 Enum(StrongVerificationAttemptStatus),
79 default=StrongVerificationAttemptStatus.in_progress_waiting_on_user_to_open_app,
80 )
82 ## full data
83 has_full_data: Mapped[bool] = mapped_column(Boolean, default=False)
84 # the data returned from iris, encrypted with a public key whose private key is kept offline
85 passport_encrypted_data: Mapped[bytes | None] = mapped_column(Binary, nullable=True)
86 passport_date_of_birth: Mapped[date | None] = mapped_column(Date, nullable=True)
87 passport_sex: Mapped[PassportSex | None] = mapped_column(Enum(PassportSex), nullable=True)
89 ## minimal data: this will not be deleted
90 has_minimal_data: Mapped[bool] = mapped_column(Boolean, default=False)
91 passport_expiry_date: Mapped[date | None] = mapped_column(Date, nullable=True)
92 passport_nationality: Mapped[str | None] = mapped_column(String, nullable=True)
93 # last three characters of the passport number
94 passport_last_three_document_chars: Mapped[str | None] = mapped_column(String, nullable=True)
96 iris_token: Mapped[str] = mapped_column(String, unique=True)
97 iris_session_id: Mapped[int] = mapped_column(BigInteger, unique=True)
99 passport_expiry_datetime = column_property(date_in_timezone(passport_expiry_date, "Etc/UTC")) # type: ignore[arg-type]
101 user: Mapped["User"] = relationship("User")
103 @hybrid_property
104 def is_valid(self) -> Any:
105 """
106 This only checks whether the attempt is a success and the passport is not expired, use `has_strong_verification` for full check
107 """
108 return (self.status == StrongVerificationAttemptStatus.succeeded) and (self.passport_expiry_datetime >= now())
110 @is_valid.expression
111 def is_valid(cls) -> Any: # noqa: ARG003,D102
112 return (cls.status == StrongVerificationAttemptStatus.succeeded) & (
113 func.coalesce(cls.passport_expiry_datetime >= func.now(), False)
114 )
116 @hybrid_property
117 def is_visible(self) -> bool:
118 return self.status != StrongVerificationAttemptStatus.deleted
120 @hybrid_method
121 def _raw_birthdate_match(self, user: "User | type[User]") -> Any:
122 """Does not check whether the SV attempt itself is not expired"""
123 return self.passport_date_of_birth == user.birthdate
125 @hybrid_method
126 def matches_birthdate(self, user: "User | type[User]") -> Any:
127 return self.is_valid & self._raw_birthdate_match(user)
129 @hybrid_method
130 def _raw_gender_match(self, user: "User | type[User]") -> Any:
131 """Does not check whether the SV attempt itself is not expired"""
132 return (
133 ((user.gender == "Woman") & (self.passport_sex == PassportSex.female)) # type: ignore[operator]
134 | ((user.gender == "Man") & (self.passport_sex == PassportSex.male))
135 | (self.passport_sex == PassportSex.unspecified)
136 | (user.has_passport_sex_gender_exception == True)
137 )
139 @hybrid_method
140 def matches_gender(self, user: "User | type[User]") -> Any:
141 return self.is_valid & self._raw_gender_match(user)
143 @hybrid_method
144 def has_strong_verification(self, user: "User | type[User]") -> Any:
145 return self.is_valid & self._raw_birthdate_match(user) & self._raw_gender_match(user)
147 __table_args__ = (
148 # used to look up verification status for a user
149 Index(
150 "ix_strong_verification_attempts_current",
151 user_id,
152 passport_expiry_date,
153 postgresql_where=status == StrongVerificationAttemptStatus.succeeded,
154 ),
155 # each passport can be verified only once
156 Index(
157 "ix_strong_verification_attempts_unique_succeeded",
158 passport_expiry_date,
159 passport_nationality,
160 passport_last_three_document_chars,
161 unique=True,
162 postgresql_where=(
163 (status == StrongVerificationAttemptStatus.succeeded)
164 | (status == StrongVerificationAttemptStatus.deleted)
165 ),
166 ),
167 # full data check
168 CheckConstraint(
169 "(has_full_data IS TRUE AND passport_encrypted_data IS NOT NULL AND passport_date_of_birth IS NOT NULL) OR \
170 (has_full_data IS FALSE AND passport_encrypted_data IS NULL AND passport_date_of_birth IS NULL)",
171 name="full_data_status",
172 ),
173 # minimal data check
174 CheckConstraint(
175 "(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 \
176 (has_minimal_data IS FALSE AND passport_expiry_date IS NULL AND passport_nationality IS NULL AND passport_last_three_document_chars IS NULL)",
177 name="minimal_data_status",
178 ),
179 # note on implications: p => q iff ~p OR q
180 # full data implies minimal data, has_minimal_data => has_full_data
181 CheckConstraint(
182 "(has_full_data IS FALSE) OR (has_minimal_data IS TRUE)",
183 name="full_data_implies_minimal_data",
184 ),
185 # succeeded implies full data
186 CheckConstraint(
187 "(NOT (status = 'succeeded')) OR (has_full_data IS TRUE)",
188 name="succeeded_implies_full_data",
189 ),
190 # in_progress/failed implies no_data
191 CheckConstraint(
192 "(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)",
193 name="in_progress_failed_iris_implies_no_data",
194 ),
195 # deleted or duplicate implies minimal data
196 CheckConstraint(
197 "(NOT ((status = 'deleted') OR (status = 'duplicate'))) OR (has_minimal_data IS TRUE)",
198 name="deleted_duplicate_implies_minimal_data",
199 ),
200 )
203class StrongVerificationCallbackEvent(Base):
204 __tablename__ = "strong_verification_callback_events"
206 id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
207 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
209 verification_attempt_id: Mapped[int] = mapped_column(ForeignKey("strong_verification_attempts.id"), index=True)
211 iris_status: Mapped[str] = mapped_column(String)