(PC-32146)[API] feat: add ubble v2 identification start

api/.env.development

@@ -123,4 +123,4 @@ VIRUSTOTAL_BACKEND=pcapi.connectors.virustotal.LoggerBackend
 WEBAPP_V2_REDIRECT_URL=http://localhost:3000
 WEBAPP_V2_URL=http://localhost:3000
 ZENDESK_SELL_API_URL=https://api.getbase.com
-ZENDESK_SELL_BACKEND=pcapi.core.external.zendesk_sell_backends.logger.LoggerBackend
+ZENDESK_SELL_BACKEND=pcapi.core.external.zendesk_sell_backends.logger.LoggerBackend

api/.env.testauto

@@ -54,8 +54,9 @@ SUPPORT_EMAIL_ADDRESS=support@example.com
 SUPPORT_PRO_EMAIL_ADDRESS=support-pro@example.com
 TITELIVE_EPAGINE_API_AUTH_URL=https://login.epagine.example.com
 TITELIVE_EPAGINE_API_URL=https://epagine.example.com
+UBBLE_API_URL=https://api.ubble.example.com
 UBBLE_MOCK_API_URL=
 UBBLE_WEBHOOK_SECRET=fake_ubblewebhook_secret # ggignore
 WEBAPP_V2_REDIRECT_URL=https://webapp-v2.example.com
 WEBAPP_V2_URL=https://webapp-v2.example.com
-ZENDESK_SELL_BACKEND=pcapi.core.external.zendesk_sell_backends.testing.TestingBackend
+ZENDESK_SELL_BACKEND=pcapi.core.external.zendesk_sell_backends.testing.TestingBackend

api/src/pcapi/connectors/beneficiaries/ubble.py

@@ -1,15 +1,19 @@
 from contextlib import suppress
+import functools
 import logging
 import typing
 
 from pydantic.v1 import networks as pydantic_networks
+from pydantic.v1 import parse_obj_as
 from urllib3 import exceptions as urllib3_exceptions
 
 from pcapi import settings
+from pcapi.connectors.serialization import ubble_serializers
 from pcapi.core import logging as core_logging
 from pcapi.core.fraud import models as fraud_models
 from pcapi.core.fraud.ubble import models as ubble_fraud_models
 from pcapi.core.users import models as users_models
+from pcapi.models.feature import FeatureToggle
 from pcapi.utils import requests
 
 
@@ -80,6 +84,109 @@ def get_content(self, identification_id: str) -> fraud_models.UbbleContent:
         raise NotImplementedError()
 
 
+P = typing.ParamSpec("P")
+
+
+def log_and_handle_response_status(
+    request_type: str,
+) -> typing.Callable[[typing.Callable[P, fraud_models.UbbleContent]], typing.Callable[P, fraud_models.UbbleContent]]:
+    def log_response_status_and_reraise_if_needed(
+        ubble_content_function: typing.Callable[P, fraud_models.UbbleContent]
+    ) -> typing.Callable[P, fraud_models.UbbleContent]:
+        @functools.wraps(ubble_content_function)
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> fraud_models.UbbleContent:
+            try:
+                ubble_content = ubble_content_function(*args, **kwargs)
+
+                logger.info(
+                    "Valid response from Ubble",
+                    extra={"identification_id": str(ubble_content.identification_id), "request_type": request_type},
+                )
+
+                return ubble_content
+            except requests.exceptions.HTTPError as e:
+                response = e.response
+                if response.status_code == 429 or response.status_code >= 500:
+                    logger.error(
+                        f"Ubble {request_type}: External error: %s",
+                        response.status_code,
+                        extra={
+                            "alert": "Ubble error",
+                            "error_type": "http",
+                            "status_code": response.status_code,
+                            "request_type": request_type,
+                            "response_text": response.text,
+                            "url": response.url,
+                        },
+                    )
+                    raise requests.ExternalAPIException(is_retryable=True) from e
+
+                logger.error(
+                    f"Ubble {request_type}: Unexpected error: %s",
+                    response.status_code,
+                    extra={
+                        "alert": "Ubble error",
+                        "error_type": "http",
+                        "status_code": response.status_code,
+                        "request_type": request_type,
+                        "response_text": response.text,
+                        "url": response.url,
+                    },
+                )
+                raise requests.ExternalAPIException(is_retryable=True) from e
+            except (urllib3_exceptions.HTTPError, requests.exceptions.RequestException) as e:
+                logger.error(
+                    "Ubble %s: Network error",
+                    request_type,
+                    extra={
+                        "exception": e,
+                        "alert": "Ubble error",
+                        "error_type": "network",
+                        "request_type": request_type,
+                    },
+                )
+                raise requests.ExternalAPIException(is_retryable=True) from e
+
+        return wrapper
+
+    return log_response_status_and_reraise_if_needed
+
+
+class UbbleV2Backend(UbbleBackend):
+    @log_and_handle_response_status("create-and-start-idv")
+    def start_identification(  # pylint: disable=too-many-positional-arguments
+        self,
+        user_id: int,
+        first_name: str,
+        last_name: str,
+        webhook_url: str,
+        redirect_url: str,
+    ) -> fraud_models.UbbleContent:
+        response = requests.post(
+            build_url("/v2/create-and-start-idv", user_id),
+            json={
+                "declared_data": {"name": f"{first_name} {last_name}"},
+                "webhook_url": webhook_url,
+                "redirect_url": redirect_url,
+            },
+            cert=(settings.UBBLE_CLIENT_CERTIFICATE_PATH, settings.UBBLE_CLIENT_KEY_PATH),
+        )
+        response.raise_for_status()
+
+        ubble_identification = parse_obj_as(ubble_serializers.UbbleIdentificationResponse, response.json())
+        ubble_content = ubble_serializers.convert_identification_to_ubble_content(ubble_identification)
+
+        logger.info(
+            "Ubble identification started",
+            extra={"identification_id": str(ubble_content.identification_id), "status": str(ubble_content.status)},
+        )
+
+        return ubble_content
+
+    def get_content(self, identification_id: str) -> fraud_models.UbbleContent:
+        raise NotImplementedError()
+
+
 class UbbleV1Backend(UbbleBackend):
     def start_identification(  # pylint: disable=too-many-positional-arguments
         self,
@@ -229,6 +336,8 @@ def get_content(self, identification_id: str) -> fraud_models.UbbleContent:
 
 
 def _get_ubble_backend() -> UbbleBackend:
+    if FeatureToggle.WIP_UBBLE_V2.is_active():
+        return UbbleV2Backend()
     return UbbleV1Backend()
 
 

api/src/pcapi/connectors/serialization/ubble_serializers.py

@@ -0,0 +1,112 @@
+import contextlib
+import datetime
+
+import pydantic.v1 as pydantic_v1
+
+from pcapi.core.fraud import models as fraud_models
+from pcapi.core.fraud.ubble import models as ubble_models
+from pcapi.core.users import models as users_models
+
+
+class UbbleDeclaredData(pydantic_v1.BaseModel):
+    name: str
+    birth_date: datetime.date | None
+
+
+class UbbleLink(pydantic_v1.BaseModel):
+    href: pydantic_v1.HttpUrl
+
+
+class UbbleLinks(pydantic_v1.BaseModel):
+    self: UbbleLink
+    applicant: UbbleLink
+    verification_url: UbbleLink
+
+
+class UbbleDocument(pydantic_v1.BaseModel):
+    full_name: str
+    birth_date: datetime.date | None
+    document_type: str
+    document_number: str | None
+    gender: users_models.GenderEnum | None
+    front_image_signed_url: str
+    back_image_signed_url: str | None
+
+    @pydantic_v1.validator("gender", pre=True)
+    def parse_gender(cls, gender: str | None) -> users_models.GenderEnum | None:
+        if not gender:
+            return None
+        with contextlib.suppress(KeyError):
+            return users_models.GenderEnum[gender]
+        return None
+
+
+class UbbleResponseCode(pydantic_v1.BaseModel):
+    response_code: int
+
+
+class UbbleIdentificationResponse(pydantic_v1.BaseModel):
+    # https://docs.ubble.ai/#tag/Identity-verifications/operation/create_and_start_identity_verification
+    id: str
+    applicant_id: str
+    user_journey_id: str
+    # TODO clean up imports
+    status: ubble_models.UbbleIdentificationStatus
+    declared_data: UbbleDeclaredData
+    links: UbbleLinks = pydantic_v1.Field(alias="_links")
+    documents: list[UbbleDocument]
+    response_codes: list[UbbleResponseCode]
+    webhook_url: str
+    redirect_url: str
+    created_on: datetime.datetime
+    modified_on: datetime.datetime
+
+    @property
+    def document(self) -> UbbleDocument | None:
+        return self.documents[0] if self.documents else None
+
+    @property
+    def fraud_reason_codes(self) -> list[fraud_models.FraudReasonCode]:
+        return [
+            fraud_models.UBBLE_REASON_CODE_MAPPING.get(
+                response_code.response_code, fraud_models.FraudReasonCode.ID_CHECK_BLOCKED_OTHER
+            )
+            for response_code in self.response_codes
+        ]
+
+    class Config:
+        use_enum_values = True
+
+
+def convert_identification_to_ubble_content(identification: UbbleIdentificationResponse) -> fraud_models.UbbleContent:
+    document = identification.document
+    if not document:
+        first_name, last_name = None, None
+    else:
+        first_name, last_name = document.full_name.split(" ", maxsplit=1)
+
+    content = fraud_models.UbbleContent(
+        birth_date=getattr(document, "birth_date", None),
+        document_type=getattr(document, "document_type", None),
+        first_name=first_name,
+        gender=getattr(document, "gender", None),
+        id_document_number=getattr(document, "document_number", None),
+        identification_id=identification.id,
+        identification_url=identification.links.verification_url.href,
+        last_name=last_name,
+        reason_codes=identification.fraud_reason_codes,
+        registration_datetime=identification.created_on,
+        signed_image_back_url=getattr(document, "back_image_signed_url", None),
+        signed_image_front_url=getattr(document, "front_image_signed_url", None),
+        status=identification.status,
+        comment=None,
+        expiry_date_score=None,
+        married_name=None,
+        ove_score=None,
+        reference_data_check_score=None,
+        processed_datetime=None,
+        score=None,
+        status_updated_at=None,
+        supported=None,
+    )
+    return content

api/src/pcapi/core/fraud/models.py

@@ -355,7 +355,7 @@ class UbbleContent(common_models.IdentityCheckContent):
     first_name: str | None
     gender: users_models.GenderEnum | None
     id_document_number: str | None
-    identification_id: pydantic_v1.UUID4 | None
+    identification_id: str | None
     identification_url: pydantic_v1.HttpUrl | None
     last_name: str | None
     married_name: str | None

api/src/pcapi/core/fraud/ubble/models.py

@@ -5,6 +5,15 @@
 
 
 class UbbleIdentificationStatus(enum.Enum):
+    # ubble v2
+    PENDING = "pending"
+    CAPTURE_IN_PROGRESS = "capture_in_progress"
+    CHECKS_IN_PROGRESS = "checks_in_progress"
+    APPROVED = "approved"
+    DECLINED = "declined"
+    RETRY_REQUIRED = "retry_required"
+    REFUSED = "refused"
+    # ubble v1
     UNINITIATED = "uninitiated"  # Identification has only been created (user has not started the verification flow)
     INITIATED = "initiated"  # User has started the verification flow
     PROCESSING = "processing"  # User has ended the verification flow, identification-url is not usable anymore

api/src/pcapi/settings.py

@@ -255,6 +255,8 @@
 UBBLE_CLIENT_ID = secrets_utils.get("UBBLE_CLIENT_ID", "")
 UBBLE_CLIENT_SECRET = secrets_utils.get("UBBLE_CLIENT_SECRET", "")
 UBBLE_WEBHOOK_SECRET = secrets_utils.get("UBBLE_WEBHOOK_SECRET")
+UBBLE_CLIENT_CERTIFICATE_PATH = os.environ.get("UBBLE_CLIENT_CERTIFICATE_PATH", "ubble.crt")
+UBBLE_CLIENT_KEY_PATH = os.environ.get("UBBLE_CLIENT_KEY_PATH", "ubble.key")
 UBBLE_SUBSCRIPTION_LIMITATION_DAYS = os.environ.get("UBBLE_SUBSCRIPTION_LIMITATION_DAYS", 90)
 UBBLE_MOCK_API_URL = os.environ.get("UBBLE_MOCK_API_URL", None)
 ENABLE_UBBLE_TEST_EMAIL = bool(int(os.environ.get("ENABLE_UBBLE_TEST_EMAIL", 1)))