(PC-32147)[API] feat: add ubble v2 webhook

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

@@ -1,9 +1,10 @@
 from contextlib import suppress
-import re
 import functools
 import logging
+import re
 import typing
 
+import flask
 from pydantic.v1 import networks as pydantic_networks
 from pydantic.v1 import parse_obj_as
 from urllib3 import exceptions as urllib3_exceptions
@@ -30,10 +31,12 @@
 
 
 def start_identification(
-    user_id: int, first_name: str, last_name: str, webhook_url: str, redirect_url: str
+    user_id: int, first_name: str, last_name: str, redirect_url: str, webhook_url: str | None = None
 ) -> fraud_models.UbbleContent:
     ubble_backend = _get_ubble_backend()
-    return ubble_backend.start_identification(user_id, first_name, last_name, webhook_url, redirect_url)
+    return ubble_backend.start_identification(
+        user_id=user_id, first_name=first_name, last_name=last_name, redirect_url=redirect_url, webhook_url=webhook_url
+    )
 
 
 def get_content(identification_id: str) -> fraud_models.UbbleContent:
@@ -80,13 +83,14 @@ def download_ubble_picture(http_url: pydantic_networks.HttpUrl) -> tuple[str | N
 
 
 class UbbleBackend:
-    def start_identification(  # pylint: disable=too-many-positional-arguments
+    def start_identification(
         self,
+        *,
         user_id: int,
         first_name: str,
         last_name: str,
-        webhook_url: str,
         redirect_url: str,
+        webhook_url: str | None = None,
     ) -> fraud_models.UbbleContent:
         raise NotImplementedError()
 
@@ -164,14 +168,17 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> fraud_models.UbbleContent:
 
 class UbbleV2Backend(UbbleBackend):
     @log_and_handle_ubble_response("create-and-start-idv")
-    def start_identification(  # pylint: disable=too-many-positional-arguments
+    def start_identification(
         self,
+        *,
         user_id: int,
         first_name: str,
         last_name: str,
-        webhook_url: str,
         redirect_url: str,
+        webhook_url: str | None = None,
     ) -> fraud_models.UbbleContent:
+        if webhook_url is None:
+            webhook_url = flask.url_for("Public API.ubble_v2_webhook_update_application_status", _external=True)
         response = requests.post(
             build_url("/v2/create-and-start-idv", user_id),
             json={
@@ -206,14 +213,17 @@ def get_content(self, identification_id: str) -> fraud_models.UbbleContent:
 
 
 class UbbleV1Backend(UbbleBackend):
-    def start_identification(  # pylint: disable=too-many-positional-arguments
+    def start_identification(
         self,
+        *,
         user_id: int,
         first_name: str,
         last_name: str,
-        webhook_url: str,
         redirect_url: str,
+        webhook_url: str | None = None,
     ) -> fraud_models.UbbleContent:
+        if webhook_url is None:
+            webhook_url = flask.url_for("Public API.ubble_webhook_update_application_status", _external=True)
         session = configure_session()
 
         data = {

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

@@ -2,6 +2,7 @@
 import datetime
 import enum
 import logging
+import re
 
 import pydantic.v1 as pydantic_v1
 
@@ -150,6 +151,26 @@ def _get_first_and_last_name(document: UbbleDocument) -> tuple[str, str]:
     return first_name, last_name
 
 
+class WebhookBodyV2(pydantic_v1.BaseModel):
+    # https://docs.ubble.ai/#section/Webhooks/Body
+    identity_verification_id: str
+    status: UbbleIdentificationStatus
+
+    class Config:
+        use_enum_values = True
+
+
+# Ubble only consider HTTP status 200 and 201 as success
+# but we are not able to respond with empty body unless we return a 204 HTTP status
+# so we need a dummy reponse_model to be used for the webhook response
+class WebhookDummyReponse(pydantic_v1.BaseModel):
+    status: str = "ok"
+
+
+class WebhookStoreIdPicturesRequest(pydantic_v1.BaseModel):
+    identification_id: str
+
+
 # DEPRECATED Ubble V1
 
 
@@ -282,3 +303,24 @@ class UbbleIdentificationIncludedDocFaceMatches(UbbleIdentificationIncluded):
 class UbbleIdentificationResponse(pydantic_v1.BaseModel):
     data: UbbleIdentificationData
     included: list[UbbleIdentificationIncluded]
+
+
+class Configuration(pydantic_v1.BaseModel):
+    id: int
+    name: str
+
+
+class WebhookRequest(pydantic_v1.BaseModel):
+    identification_id: str
+    status: UbbleIdentificationStatus
+    configuration: Configuration
+
+
+UBBLE_SIGNATURE_RE = re.compile(r"^ts=(?P<ts>\d+),v1=(?P<v1>\S{64})$")
+
+
+class WebhookRequestHeaders(pydantic_v1.BaseModel):
+    ubble_signature: str = pydantic_v1.Field(..., regex=UBBLE_SIGNATURE_RE.pattern, alias="Ubble-Signature")
+
+    class Config:
+        extra = "allow"

api/src/pcapi/core/subscription/ubble/api.py

@@ -5,7 +5,6 @@
 import shutil
 import tempfile
 
-import flask
 from pydantic.v1.networks import HttpUrl
 
 from pcapi import settings
@@ -105,7 +104,6 @@ def start_ubble_workflow(user: users_models.User, first_name: str, last_name: st
         user_id=user.id,
         first_name=first_name,
         last_name=last_name,
-        webhook_url=flask.url_for("Public API.ubble_webhook_update_application_status", _external=True),
         redirect_url=redirect_url,
     )
     fraud_check = subscription_api.initialize_identity_fraud_check(

api/src/pcapi/routes/external/users_subscription.py

@@ -1,6 +1,7 @@
 import logging
 
 from pcapi.connectors.dms import api as dms_connector_api
+from pcapi.connectors.serialization import ubble_serializers
 from pcapi.core.fraud.exceptions import IncompatibleFraudCheckStatus
 from pcapi.core.fraud.models import FraudCheckStatus
 from pcapi.core.fraud.ubble import api as ubble_fraud_api
@@ -27,27 +28,53 @@ def dms_webhook_update_application_status(form: dms_validation.DMSWebhookRequest
     dms_subscription_api.handle_dms_application(dms_application)
 
 
+@public_api.route("/webhooks/ubble/v2/application_status", methods=["POST"])
+@spectree_serialize(
+    on_success_status=200,
+    response_model=ubble_serializers.WebhookDummyReponse,  # type: ignore[arg-type]
+)
+def ubble_v2_webhook_update_application_status(
+    body: ubble_serializers.WebhookBodyV2,
+) -> ubble_serializers.WebhookDummyReponse:
+    if body.status in (
+        ubble_serializers.UbbleIdentificationStatus.CHECKS_IN_PROGRESS,
+        ubble_serializers.UbbleIdentificationStatus.APPROVED,
+        ubble_serializers.UbbleIdentificationStatus.DECLINED,
+        ubble_serializers.UbbleIdentificationStatus.RETRY_REQUIRED,
+        ubble_serializers.UbbleIdentificationStatus.INCONCLUSIVE,
+        ubble_serializers.UbbleIdentificationStatus.REFUSED,
+    ):
+        return _update_ubble_workflow(body.identity_verification_id, body.status)
+    return ubble_serializers.WebhookDummyReponse()
+
+
 @public_api.route("/webhooks/ubble/application_status", methods=["POST"])
 @ubble_validation.require_ubble_signature
 @spectree_serialize(
-    headers=ubble_validation.WebhookRequestHeaders,  # type: ignore[arg-type]
+    headers=ubble_serializers.WebhookRequestHeaders,  # type: ignore[arg-type]
     on_success_status=200,
-    response_model=ubble_validation.WebhookDummyReponse,  # type: ignore[arg-type]
+    response_model=ubble_serializers.WebhookDummyReponse,  # type: ignore[arg-type]
 )
 def ubble_webhook_update_application_status(
-    body: ubble_validation.WebhookRequest,
-) -> ubble_validation.WebhookDummyReponse:
-    log_extra_data = {"identification_id": body.identification_id, "status": str(body.status)}
+    body: ubble_serializers.WebhookRequest,
+) -> ubble_serializers.WebhookDummyReponse:
+    return _update_ubble_workflow(body.identification_id, body.status)
+
+
+def _update_ubble_workflow(
+    identification_id: str, status: ubble_serializers.UbbleIdentificationStatus
+) -> ubble_serializers.WebhookDummyReponse:
+    log_extra_data = {"identification_id": identification_id, "status": str(status)}
     logger.info("Ubble webhook called", extra=log_extra_data)
 
-    fraud_check = ubble_fraud_api.get_ubble_fraud_check(body.identification_id)
+    fraud_check = ubble_fraud_api.get_ubble_fraud_check(identification_id)
     if not fraud_check:
-        raise ValueError(f"no Ubble fraud check found with identification_id {body.identification_id}")
+        raise ValueError(f"no Ubble fraud check found with identification_id {identification_id}")
 
     finished_status = [FraudCheckStatus.OK, FraudCheckStatus.KO, FraudCheckStatus.CANCELED, FraudCheckStatus.SUSPICIOUS]
     if fraud_check.status in finished_status:
         logger.warning("Ubble fraud check already has finished status", extra=log_extra_data)
-        return ubble_validation.WebhookDummyReponse()
+        return ubble_serializers.WebhookDummyReponse()
 
     try:
         ubble_subscription_api.update_ubble_workflow(fraud_check)
@@ -60,17 +87,17 @@ def ubble_webhook_update_application_status(
         logger.exception("Could not update Ubble workflow", extra=log_extra_data)
         raise ApiErrors({"msg": "an error occured during workflow update"}, status_code=500)
 
-    return ubble_validation.WebhookDummyReponse()
+    return ubble_serializers.WebhookDummyReponse()
 
 
 @public_api.route("/webhooks/ubble/store_id_pictures", methods=["POST"])
 @spectree_serialize(
     on_success_status=200,
-    response_model=ubble_validation.WebhookDummyReponse,  # type: ignore[arg-type]
+    response_model=ubble_serializers.WebhookDummyReponse,  # type: ignore[arg-type]
 )
 def ubble_webhook_store_id_pictures(
-    body: ubble_validation.WebhookStoreIdPicturesRequest,
-) -> ubble_validation.WebhookDummyReponse:
+    body: ubble_serializers.WebhookStoreIdPicturesRequest,
+) -> ubble_serializers.WebhookDummyReponse:
     logger.info("Webhook store id pictures called ", extra={"identification_id": body.identification_id})
     try:
         ubble_subscription_api.archive_ubble_user_id_pictures(body.identification_id)
@@ -83,4 +110,4 @@ def ubble_webhook_store_id_pictures(
     except IncompatibleFraudCheckStatus as err:
         raise ApiErrors({"err": str(err)}, status_code=422)
 
-    return ubble_validation.WebhookDummyReponse()
+    return ubble_serializers.WebhookDummyReponse()

api/src/pcapi/validation/routes/ubble.py

@@ -1,57 +1,22 @@
 import functools
 import hashlib
 import hmac
-import re
 from typing import Any
 from typing import Callable
 
 import flask
-import pydantic.v1 as pydantic_v1
 
 from pcapi import settings
 from pcapi.connectors.serialization import ubble_serializers
 from pcapi.models.api_errors import ForbiddenError
 
 
-UBBLE_SIGNATURE_RE = re.compile(r"^ts=(?P<ts>\d+),v1=(?P<v1>\S{64})$")
-
-
-# TODO move this file to serialization
-class Configuration(pydantic_v1.BaseModel):
-    id: int
-    name: str
-
-
-class WebhookRequest(pydantic_v1.BaseModel):
-    identification_id: str
-    status: ubble_serializers.UbbleIdentificationStatus
-    configuration: Configuration
-
-
-class WebhookRequestHeaders(pydantic_v1.BaseModel):
-    ubble_signature: str = pydantic_v1.Field(..., regex=UBBLE_SIGNATURE_RE.pattern, alias="Ubble-Signature")
-
-    class Config:
-        extra = "allow"
-
-
-# Ubble only consider HTTP status 200 and 201 as success
-# but we are not able to respond with empty body unless we return a 204 HTTP status
-# so we need a dummy reponse_model to be used for the webhook response
-class WebhookDummyReponse(pydantic_v1.BaseModel):
-    status: str = "ok"
-
-
-class WebhookStoreIdPicturesRequest(pydantic_v1.BaseModel):
-    identification_id: str
-
-
 def require_ubble_signature(route_function: Callable[..., Any]) -> Callable:
     @functools.wraps(route_function)
     def validate_ubble_signature(*args: Any, **kwargs: Any) -> flask.Response:
         error = ForbiddenError(errors={"signature": ["Invalid signature"]})
         signature = getattr(
-            UBBLE_SIGNATURE_RE.match(flask.request.headers.get("Ubble-Signature", "")),
+            ubble_serializers.UBBLE_SIGNATURE_RE.match(flask.request.headers.get("Ubble-Signature", "")),
             "groupdict",
             lambda: None,
         )()

api/tests/core/subscription/ubble/test_api.py

@@ -71,7 +71,7 @@ def test_start_ubble_workflow(self, requests_mock):
         assert fraud_check.status == fraud_models.FraudCheckStatus.STARTED
 
         ubble_request = requests_mock.last_request.json()
-        assert ubble_request["webhook_url"] == "http://localhost/webhooks/ubble/application_status"
+        assert ubble_request["webhook_url"] == "http://localhost/webhooks/ubble/v2/application_status"
 
         assert push_testing.requests[0] == {
             "can_be_asynchronously_retried": True,

api/tests/routes/external/user_subscription_test.py

@@ -961,13 +961,33 @@ def test_dms_field_error_then_corrected(self, execute_query, client):
         assert fraud_check.status == fraud_models.FraudCheckStatus.PENDING
 
 
+class UbbleWebhookV2Test:
+    @pytest.mark.parametrize(
+        "status",
+        [
+            ubble_serializers.UbbleIdentificationStatus.PENDING,
+            ubble_serializers.UbbleIdentificationStatus.CAPTURE_IN_PROGRESS,
+        ],
+    )
+    def test_ignore_events_before_identification_conclusion(self, client, requests_mock, status):
+        with mock.patch(
+            "pcapi.core.subscription.ubble.api.update_ubble_workflow",
+        ) as mocked_update:
+            client.post(
+                "/webhooks/ubble/application_status",
+                json={"identity_verification_id": "idv_qwerty123", "status": status.value},
+            )
+
+            mocked_update.assert_not_called()
+
+
 @pytest.mark.usefixtures("db_session")
 class UbbleWebhookTest:
-    def _get_request_body(self, fraud_check, status) -> ubble_routes.WebhookRequest:
-        return ubble_routes.WebhookRequest(
+    def _get_request_body(self, fraud_check, status) -> ubble_serializers.WebhookRequest:
+        return ubble_serializers.WebhookRequest(
             identification_id=fraud_check.thirdPartyId,
             status=status,
-            configuration=ubble_routes.Configuration(
+            configuration=ubble_serializers.Configuration(
                 id=fraud_check.user.id,
                 name="Pass Culture",
             ),
@@ -1477,7 +1497,7 @@ def test_ubble_test_emails_not_actives_on_production(self, client, ubble_mocker,
 
     def _init_decision_test(
         self,
-    ) -> tuple[users_models.User, fraud_models.BeneficiaryFraudCheck, ubble_routes.WebhookRequest]:
+    ) -> tuple[users_models.User, fraud_models.BeneficiaryFraudCheck, ubble_serializers.WebhookRequest]:
         birth_date = datetime.datetime.utcnow().date() - relativedelta.relativedelta(years=18, months=6)
         user = users_factories.UserFactory(dateOfBirth=datetime.datetime.combine(birth_date, datetime.time(0, 0)))
         identification_id = str(uuid.uuid4())