Merge branch 'main' of https://github.com/weni-ai/flows into feature/…

…internal-endpoint-list-contact-fields

temba/api/v2/internals/contacts/serializers.py

@@ -1,5 +1,61 @@
 from rest_framework import serializers
 
+from django.conf import settings
+from django.contrib.auth import get_user_model
+
+from temba.contacts.models import ContactField, ContactURN
+from temba.orgs.models import Org
+
+User = get_user_model()
+
 
 class InternalContactSerializer(serializers.Serializer):
     contacts = serializers.ListSerializer(child=serializers.UUIDField(), min_length=1, max_length=100)
+
+
+class InternalContactFieldsValuesSerializer(serializers.Serializer):
+    project = serializers.UUIDField()
+    contact_fields = serializers.DictField(child=serializers.CharField(allow_null=True, allow_blank=True))
+    contact_urn = serializers.CharField(required=True)
+
+    def validate(self, data):
+        project_uuid = data.get("project")
+        contact_urn = data.get("contact_urn")
+
+        try:
+            org = Org.objects.get(proj_uuid=project_uuid)
+
+        except Org.DoesNotExist:
+            raise serializers.ValidationError({"project": "Project not found"})
+
+        contact = ContactURN.lookup(org, contact_urn)
+        if not contact:
+            raise serializers.ValidationError({"contact_urn": "Contact URN not found"})
+
+        return data
+
+    def validate_contact_fields(self, value):
+        if not value:
+            raise serializers.ValidationError("contact_fields must not be an empty dictionary")
+        return value
+
+    def update(self, instance, validated_data):
+        project_uuid = validated_data.get("project")
+        contact_urn = validated_data.get("contact_urn")
+        contact_fields = validated_data.get("contact_fields", {})
+        user = User.objects.get(email=settings.INTERNAL_USER_EMAIL)
+
+        org = Org.objects.get(proj_uuid=project_uuid)
+        urn = ContactURN.lookup(org, contact_urn)
+        contact = urn.contact
+
+        fields_to_update = {}
+        for key, value in contact_fields.items():
+            contact_field = ContactField.all_fields.filter(key=key, org=org).first()
+            if contact_field:
+                fields_to_update[contact_field] = value
+
+        mods = contact.update_fields(fields_to_update)
+        contact.modify(user, mods)
+
+        return instance

temba/api/v2/internals/contacts/tests.py

@@ -1,10 +1,18 @@
 from functools import wraps
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
+from rest_framework import status
+from rest_framework.response import Response
+
+from django.contrib.auth import get_user_model
 from django.test import override_settings
 
-from temba.tests import TembaTest
+from temba.api.v2.validators import LambdaURLValidator
 from temba.contacts.models import ContactField
+from temba.tests import TembaTest
+from temba.tests.mailroom import mock_mailroom
+
+User = get_user_model()
 
 
 CONTACT_FIELDS_ENDPOINT_PATH = "temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint"
@@ -127,3 +135,195 @@ def test_get_contact_fields_with_multiple_fields(self):
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(data.get("results"), expected_result)
+
+
+class UpdateContactFieldsViewTest(TembaTest):
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_request_without_body(self, mock_protected_resource):
+        url = "/api/v2/internals/update_contacts_fields"
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+        response = self.client.patch(url)
+
+        self.assertEqual(response.status_code, 400)
+
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_request_no_project(self, mock_protected_resource):
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+
+        url = "/api/v2/internals/update_contacts_fields"
+
+        body = {
+            "contact_urn": "Nick Name",
+            "contact_fields": {"cpf": "12345678912"},
+        }
+
+        response = self.client.patch(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json(), {"project": ["This field is required."]})
+
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_request_incorrect_project(self, mock_protected_resource):
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+
+        url = "/api/v2/internals/update_contacts_fields"
+
+        body = {
+            "project": self.org.uuid,
+            "contact_urn": "Nick Name",
+            "contact_fields": {"cpf": "12345678912"},
+        }
+
+        response = self.client.patch(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json(), {"project": ["Project not found"]})
+
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_request_invalid_contact_urn(self, mock_protected_resource):
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+
+        url = "/api/v2/internals/update_contacts_fields"
+
+        body = {
+            "project": self.org.proj_uuid,
+            "contact_urn": "ext:hello@hello.ign",
+            "contact_fields": {"cpf": "12345678912"},
+        }
+
+        response = self.client.patch(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json(), {"contact_urn": ["Contact URN not found"]})
+
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_request_no_contact_fields(self, mock_protected_resource):
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+
+        url = "/api/v2/internals/update_contacts_fields"
+
+        body = {
+            "project": self.org.proj_uuid,
+            "contact_urn": "ext:hello@hello.ign",
+            "contact_fields": {},
+        }
+
+        response = self.client.patch(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json(), {"contact_fields": ["contact_fields must not be an empty dictionary"]})
+
+    @mock_mailroom
+    @override_settings(INTERNAL_USER_EMAIL="super@user.com")
+    @patch.object(LambdaURLValidator, "protected_resource")
+    def test_success(self, mr_mocks, mock_protected_resource):
+        self.create_contact("Rigbt", urns=["twitterid:0000000"])
+        self.create_field("last_name", "Last name")
+
+        mock_protected_resource.return_value = Response({"message": "Access granted!"}, status=status.HTTP_200_OK)
+
+        url = "/api/v2/internals/update_contacts_fields"
+
+        body = {
+            "project": self.org.proj_uuid,
+            "contact_urn": "twitterid:0000000",
+            "contact_fields": {"last_name": "Cube"},
+        }
+
+        response = self.client.patch(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), {"message": "Contact fields updated successfully"})
+
+
+class InternalContactFieldsEndpointTest(TembaTest):
+    def setUp(self):
+        super().setUp()
+        User.objects.create(username="Mordecai", email="mordecai@msn.com")
+
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.authentication_classes", [])
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.permission_classes", [])
+    def test_request_without_body(self):
+        url = "/api/v2/internals/contacts_fields"
+        response = self.client.post(url)
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response.json(), {"error": "Project not provided"})
+
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.authentication_classes", [])
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.permission_classes", [])
+    def test_project_not_found(self):
+        url = "/api/v2/internals/contacts_fields"
+        body = {
+            "project": self.org.uuid,
+            "label": "Nick Name",
+            "value_type": "text",
+        }
+        response = self.client.post(url, data=body, content_type="application/json")
+
+        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.json(), {"error": "Project not found"})
+
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.authentication_classes", [])
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.permission_classes", [])
+    def test_user_not_found(self):
+        mock_user = MagicMock(spec=User)
+        mock_user.is_authenticated = False
+        mock_user.email = "mockuser@example.com"
+
+        with patch("rest_framework.request.Request.user", mock_user):
+
+            url = "/api/v2/internals/contacts_fields"
+            body = {
+                "project": self.org.proj_uuid,
+                "label": "Nick Name",
+                "value_type": "text",
+            }
+            response = self.client.post(url, data=body, content_type="application/json")
+
+            self.assertEqual(response.status_code, 404)
+            self.assertEqual(response.json(), {"error": "User not found"})
+
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.authentication_classes", [])
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.permission_classes", [])
+    def test_serializer_error(self):
+        mock_user = MagicMock(spec=User)
+        mock_user.is_authenticated = True
+        mock_user.email = "mordecai@msn.com"
+
+        with patch("rest_framework.request.Request.user", mock_user):
+
+            url = "/api/v2/internals/contacts_fields"
+            body = {
+                "project": self.org.proj_uuid,
+                "label": "Nick Name",
+                "value_type": "T",
+            }
+            response = self.client.post(url, data=body, content_type="application/json")
+
+            self.assertEqual(response.status_code, 400)
+
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.authentication_classes", [])
+    @patch("temba.api.v2.internals.contacts.views.InternalContactFieldsEndpoint.permission_classes", [])
+    def test_success(self):
+        mock_user = MagicMock(spec=User)
+        mock_user.is_authenticated = True
+        mock_user.email = "mordecai@msn.com"
+
+        with patch("rest_framework.request.Request.user", mock_user):
+
+            url = "/api/v2/internals/contacts_fields"
+            body = {
+                "project": self.org.proj_uuid,
+                "label": "Nick Name",
+                "value_type": "text",
+            }
+            response = self.client.post(url, data=body, content_type="application/json")
+
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json(), {"label": "Nick Name", "value_type": "T"})

temba/api/v2/internals/contacts/urls.py

@@ -1,8 +1,9 @@
 from django.urls import path
 
-from .views import InternalContactFieldsEndpoint, InternalContactView
+from .views import InternalContactFieldsEndpoint, InternalContactView, UpdateContactFieldsView
 
 urlpatterns = [
     path("contacts", InternalContactView.as_view(), name="internal_contacts"),
     path("contacts_fields", InternalContactFieldsEndpoint.as_view(), name="internal_contacts_fields"),
+    path("update_contacts_fields", UpdateContactFieldsView.as_view(), name="internal_update_contacts_fields"),
 ]

temba/api/v2/internals/contacts/views.py

@@ -1,22 +1,24 @@
+from rest_framework import status
 from rest_framework.exceptions import AuthenticationFailed, NotAuthenticated
 from rest_framework.permissions import IsAuthenticated
+from rest_framework.renderers import JSONRenderer
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
-from rest_framework import status
 from weni.internal.authenticators import InternalOIDCAuthentication
 from weni.internal.permissions import CanCommunicateInternally
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core import exceptions as django_exceptions
 
-from temba.api.v2.internals.contacts.serializers import InternalContactSerializer
-from temba.api.v2.internals.views import APIViewMixin
-from temba.api.v2.serializers import (
-    ContactFieldReadSerializer,
-    ContactFieldWriteSerializer,
+from temba.api.v2.internals.contacts.serializers import (
+    InternalContactFieldsValuesSerializer,
+    InternalContactSerializer,
 )
+from temba.api.v2.internals.views import APIViewMixin
+from temba.api.v2.serializers import ContactFieldReadSerializer, ContactFieldWriteSerializer
+from temba.api.v2.validators import LambdaURLValidator
 from temba.contacts.models import Contact, ContactField
 from temba.orgs.models import Org
 
@@ -108,3 +110,17 @@ def post(self, request, *args, **kwargs):
             return Response(serializer.validated_data)
 
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class UpdateContactFieldsView(APIViewMixin, APIView, LambdaURLValidator):
+    renderer_classes = [JSONRenderer]
+
+    def patch(self, request, *args, **kwargs):
+        validation_response = self.protected_resource(request)  # pragma: no cover
+        if validation_response.status_code != 200:  # pragma: no cover
+            return validation_response
+        serializer = InternalContactFieldsValuesSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.update(instance=None, validated_data=serializer.validated_data)
+            return Response({"message": "Contact fields updated successfully"}, status=status.HTTP_200_OK)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

temba/api/v2/validators.py

@@ -1,5 +1,10 @@
+import requests
+from rest_framework import status
+from rest_framework.response import Response
 from rest_framework.validators import UniqueValidator, qs_filter
 
+from django.conf import settings
+
 
 class UniqueForOrgValidator(UniqueValidator):
     """
@@ -22,3 +27,24 @@ def __call__(self, value, serializer_field):
         self.org = serializer_field.context["org"]
 
         super().__call__(value, serializer_field)
+
+
+class LambdaURLValidator:  # pragma: no cover
+    def is_valid_url(self, sts_url):
+        return sts_url.startswith("https://sts.amazonaws.com/?Action=GetCallerIdentity&") and (".." not in sts_url)
+
+    def protected_resource(self, request):
+        try:
+            sts_url = request.headers.get("Authorization").split("Bearer ", 2)[1]
+            if not self.is_valid_url(sts_url):
+                return Response({"message": "Invalid sts"}, status=status.HTTP_400_BAD_REQUEST)
+
+            response = requests.request(method="GET", url=sts_url, headers={"Accept": "application/json"}, timeout=30)
+
+            identity_arn = response.json()["GetCallerIdentityResponse"]["GetCallerIdentityResult"]["Arn"]
+            if identity_arn in settings.LAMBDA_ALLOWED_ROLES:
+                return Response({"message": "Access granted!", "role": identity_arn})
+            else:
+                return Response({"message": "Invalid arn"}, status=status.HTTP_401_UNAUTHORIZED)
+        except Exception as e:
+            return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)