Add ChannelType.check_credentials and use for Twilio claims and updates

temba/channels/models.py

@@ -140,6 +140,12 @@ def get_update_form(self):
             return UpdateChannelForm
         return self.update_form
 
+    def check_credentials(self, config: dict) -> bool:
+        """
+        Called to check the credentials passed are valid
+        """
+        return True
+
     def activate(self, channel):
         """
         Called when a channel of this type has been created. Can be used to setup things like callbacks required by the
@@ -714,6 +720,9 @@ def is_new(self):
         # is this channel newer than an hour
         return self.created_on > timezone.now() - timedelta(hours=1) or not self.last_sync
 
+    def check_credentials(self) -> bool:
+        return self.type.check_credentials(self.config)
+
     def release(self, user, *, trigger_sync: bool = True):
         """
         Releases this channel making it inactive

temba/channels/types/twilio/tests.py

@@ -265,15 +265,43 @@ def test_claim(self):
                 mock_numbers.call_args_list[-1][1], dict(voice_application_sid="", sms_application_sid="")
             )
 
+    @patch("temba.channels.types.twilio.views.TwilioClient", MockTwilioClient)
+    @patch("temba.channels.types.twilio.type.TwilioClient", MockTwilioClient)
+    @patch("twilio.request_validator.RequestValidator", MockRequestValidator)
     def test_update(self):
-        update_url = reverse("channels.channel_update", args=[self.channel.id])
+        self.org.connect_twilio("TEST_SID", "TEST_TOKEN", self.admin)
+        twilio_channel = self.org.channels.all().first()
+        twilio_channel.channel_type = "T"
+        twilio_channel.save()
+
+        update_url = reverse("channels.channel_update", args=[twilio_channel.id])
 
         self.login(self.admin)
         response = self.client.get(update_url)
         self.assertEqual(
-            ["name", "alert_email", "allow_international", "loc"], list(response.context["form"].fields.keys())
+            ["name", "allow_international", "account_sid", "auth_token", "loc"],
+            list(response.context["form"].fields.keys()),
         )
 
+        post_data = dict(name="Foo channel", allow_international=False, account_sid="ACC_SID", auth_token="ACC_Token")
+
+        response = self.client.post(update_url, post_data)
+
+        self.assertEqual(response.status_code, 302)
+
+        twilio_channel.refresh_from_db()
+        self.assertEqual(twilio_channel.name, "Foo channel")
+        # we used the primary credentials returned on the account fetch even though we submit the others
+        self.assertEqual(twilio_channel.config[Channel.CONFIG_ACCOUNT_SID], "AccountSid")
+        self.assertEqual(twilio_channel.config[Channel.CONFIG_AUTH_TOKEN], "AccountToken")
+        self.assertTrue(twilio_channel.check_credentials())
+
+        with patch("temba.channels.types.twilio.type.TwilioType.check_credentials") as mock_check_credentials:
+            mock_check_credentials.return_value = False
+
+            response = self.client.post(update_url, post_data)
+            self.assertFormError(response, "form", None, "Channel credentials don't appear to be valid.")
+
     @patch("temba.orgs.models.TwilioClient", MockTwilioClient)
     @patch("twilio.request_validator.RequestValidator", MockRequestValidator)
     def test_deactivate(self):
@@ -297,3 +325,14 @@ def test_deactivate(self):
 
     def test_get_error_ref_url(self):
         self.assertEqual("https://www.twilio.com/docs/api/errors/30006", TwilioType().get_error_ref_url(None, "30006"))
+
+    @patch("temba.channels.types.twilio.type.TwilioClient", MockTwilioClient)
+    @patch("twilio.request_validator.RequestValidator", MockRequestValidator)
+    def test_check_credentials(self):
+        self.assertTrue(TwilioType().check_credentials({"account_sid": "AccountSid", "auth_token": "AccountToken"}))
+
+        with patch("temba.tests.twilio.MockTwilioClient.MockAccount.fetch") as mock_fetch:
+            mock_fetch.side_effect = Exception("blah!")
+            self.assertFalse(
+                TwilioType().check_credentials({"account_sid": "AccountSid", "auth_token": "AccountToken"})
+            )

temba/channels/types/twilio/type.py

@@ -1,4 +1,5 @@
 from twilio.base.exceptions import TwilioRestException
+from twilio.rest import Client as TwilioClient
 
 from django.urls import re_path
 from django.utils.translation import gettext_lazy as _
@@ -7,7 +8,7 @@
 from temba.utils.timezones import timezone_to_country_code
 
 from ...models import ChannelType
-from .views import SUPPORTED_COUNTRIES, ClaimView, SearchView
+from .views import SUPPORTED_COUNTRIES, ClaimView, SearchView, UpdateForm
 
 
 class TwilioType(ChannelType):
@@ -27,6 +28,7 @@ class TwilioType(ChannelType):
         "link": '<a target="_blank" href="https://www.twilio.com/">Twilio</a>'
     }
     claim_view = ClaimView
+    update_form = UpdateForm
 
     schemes = [URN.TEL_SCHEME]
     max_length = 1600
@@ -84,3 +86,15 @@ def get_urls(self):
 
     def get_error_ref_url(self, channel, code: str) -> str:
         return f"https://www.twilio.com/docs/api/errors/{code}"
+
+    def check_credentials(self, config: dict) -> bool:
+        account_sid = config.get("account_sid", None)
+        account_token = config.get("auth_token", None)
+
+        try:
+            client = TwilioClient(account_sid, account_token)
+            # get the actual primary auth tokens from twilio and use them
+            client.api.account.fetch()
+        except Exception:  # pragma: needs cover
+            return False
+        return True

temba/channels/types/twilio/views.py

@@ -1,23 +1,27 @@
+from typing import Any, Dict
+
 import phonenumbers
 from phonenumbers.phonenumberutil import region_code_for_number
 from smartmin.views import SmartFormView
 from twilio.base.exceptions import TwilioException, TwilioRestException
+from twilio.rest import Client as TwilioClient
 
 from django import forms
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.http import HttpResponseRedirect, JsonResponse
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 
 from temba.orgs.models import Org
 from temba.orgs.views import OrgPermsMixin
 from temba.utils import countries
-from temba.utils.fields import SelectWidget
+from temba.utils.fields import InputWidget, SelectWidget
 from temba.utils.timezones import timezone_to_country_code
 from temba.utils.uuid import uuid4
 
 from ...models import Channel
-from ...views import ALL_COUNTRIES, BaseClaimNumberMixin, ClaimViewMixin
+from ...views import ALL_COUNTRIES, BaseClaimNumberMixin, ClaimViewMixin, UpdateChannelForm
 
 SUPPORTED_COUNTRIES = {
     "AU",  # Australia
@@ -315,3 +319,54 @@ def form_valid(self, form, *args, **kwargs):
             return JsonResponse({"error": str(msg)})
 
         return JsonResponse(numbers, safe=False)
+
+
+class UpdateForm(UpdateChannelForm):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.add_config_field(
+            "account_sid",
+            forms.CharField(
+                max_length=128,
+                label=_("Twilio Account SID"),
+                required=True,
+                widget=InputWidget(attrs={"disabled": "disabled"}),
+            ),
+            default="",
+        )
+
+        self.add_config_field(
+            "auth_token",
+            forms.CharField(
+                max_length=128,
+                label=_("Twilio Account Auth Token"),
+                required=True,
+                widget=InputWidget(),
+            ),
+            default="",
+        )
+
+    def clean(self) -> Dict[str, Any]:
+        """
+        We override the clean method for Twilio we need to make sure we grab the primary auth tokens
+        """
+        account_sid = self.cleaned_data.get("account_sid", None)
+        account_token = self.cleaned_data.get("auth_token", None)
+
+        try:
+            client = TwilioClient(account_sid, account_token)
+
+            # get the actual primary auth tokens from twilio and use them
+            account = client.api.account.fetch()
+            self.cleaned_data["account_sid"] = account.sid
+            self.cleaned_data["auth_token"] = account.auth_token
+        except Exception:  # pragma: needs cover
+            raise ValidationError(
+                _("The Twilio account SID and Token seem invalid. Please check them again and retry.")
+            )
+
+        return super().clean()
+
+    class Meta(UpdateChannelForm.Meta):
+        fields = ("name",)

temba/channels/types/twilio_messaging_service/tests.py

@@ -4,6 +4,7 @@
 
 from django.urls import reverse
 
+from temba.channels.models import Channel
 from temba.orgs.models import Org
 from temba.tests import TembaTest
 from temba.tests.twilio import MockRequestValidator, MockTwilioClient
@@ -91,3 +92,42 @@ def test_get_error_ref_url(self):
             "https://www.twilio.com/docs/api/errors/30006",
             TwilioMessagingServiceType().get_error_ref_url(None, "30006"),
         )
+
+    @patch("temba.channels.types.twilio.views.TwilioClient", MockTwilioClient)
+    @patch("temba.channels.types.twilio.type.TwilioClient", MockTwilioClient)
+    @patch("twilio.request_validator.RequestValidator", MockRequestValidator)
+    def test_update(self):
+        self.org.connect_twilio("TEST_SID", "TEST_TOKEN", self.admin)
+        tms_channel = self.org.channels.all().first()
+        tms_channel.channel_type = "TMS"
+        tms_channel.save()
+
+        update_url = reverse("channels.channel_update", args=[tms_channel.id])
+
+        self.login(self.admin)
+        response = self.client.get(update_url)
+        self.assertEqual(
+            ["name", "allow_international", "account_sid", "auth_token", "loc"],
+            list(response.context["form"].fields.keys()),
+        )
+
+        post_data = dict(name="Foo channel", allow_international=False, account_sid="ACC_SID", auth_token="ACC_Token")
+
+        response = self.client.post(update_url, post_data)
+
+        self.assertEqual(response.status_code, 302)
+
+        tms_channel.refresh_from_db()
+        self.assertEqual(tms_channel.name, "Foo channel")
+        # we used the primary credentials returned on the account fetch even though we submit the others
+        self.assertEqual(tms_channel.config[Channel.CONFIG_ACCOUNT_SID], "AccountSid")
+        self.assertEqual(tms_channel.config[Channel.CONFIG_AUTH_TOKEN], "AccountToken")
+        self.assertTrue(tms_channel.check_credentials())
+
+        with patch(
+            "temba.channels.types.twilio_messaging_service.type.TwilioMessagingServiceType.check_credentials"
+        ) as mock_check_credentials:
+            mock_check_credentials.return_value = False
+
+            response = self.client.post(update_url, post_data)
+            self.assertFormError(response, "form", None, "Channel credentials don't appear to be valid.")

temba/channels/types/twilio_messaging_service/type.py

@@ -1,6 +1,7 @@
 from django.utils.translation import gettext_lazy as _
 
-from temba.channels.types.twilio.views import SUPPORTED_COUNTRIES
+from temba.channels.types.twilio.type import TwilioType
+from temba.channels.types.twilio.views import SUPPORTED_COUNTRIES, UpdateForm
 from temba.contacts.models import URN
 from temba.utils.timezones import timezone_to_country_code
 
@@ -23,6 +24,8 @@ class TwilioMessagingServiceType(ChannelType):
     icon = "icon-channel-twilio"
 
     claim_view = ClaimView
+    update_form = UpdateForm
+
     claim_blurb = _(
         "You can connect a messaging service from your Twilio account to benefit from %(link)s features."
     ) % {"link": '<a target="_blank" href="https://www.twilio.com/copilot">Twilio Copilot</a>'}
@@ -50,3 +53,6 @@ def is_recommended_to(self, org, user):
 
     def get_error_ref_url(self, channel, code: str) -> str:
         return f"https://www.twilio.com/docs/api/errors/{code}"
+
+    def check_credentials(self, config: dict) -> bool:
+        return TwilioType().check_credentials(config)

temba/channels/types/twilio_whatsapp/tests.py

@@ -140,3 +140,42 @@ def test_get_error_ref_url(self):
         self.assertEqual(
             "https://www.twilio.com/docs/api/errors/30006", TwilioWhatsappType().get_error_ref_url(None, "30006")
         )
+
+    @patch("temba.channels.types.twilio.views.TwilioClient", MockTwilioClient)
+    @patch("temba.channels.types.twilio.type.TwilioClient", MockTwilioClient)
+    @patch("twilio.request_validator.RequestValidator", MockRequestValidator)
+    def test_update(self):
+        self.org.connect_twilio("TEST_SID", "TEST_TOKEN", self.admin)
+        twilio_whatsapp = self.org.channels.all().first()
+        twilio_whatsapp.channel_type = "TWA"
+        twilio_whatsapp.save()
+
+        update_url = reverse("channels.channel_update", args=[twilio_whatsapp.id])
+
+        self.login(self.admin)
+        response = self.client.get(update_url)
+        self.assertEqual(
+            ["name", "allow_international", "account_sid", "auth_token", "loc"],
+            list(response.context["form"].fields.keys()),
+        )
+
+        post_data = dict(name="Foo channel", allow_international=False, account_sid="ACC_SID", auth_token="ACC_Token")
+
+        response = self.client.post(update_url, post_data)
+
+        self.assertEqual(response.status_code, 302)
+
+        twilio_whatsapp.refresh_from_db()
+        self.assertEqual(twilio_whatsapp.name, "Foo channel")
+        # we used the primary credentials returned on the account fetch even though we submit the others
+        self.assertEqual(twilio_whatsapp.config[Channel.CONFIG_ACCOUNT_SID], "AccountSid")
+        self.assertEqual(twilio_whatsapp.config[Channel.CONFIG_AUTH_TOKEN], "AccountToken")
+        self.assertTrue(twilio_whatsapp.check_credentials())
+
+        with patch(
+            "temba.channels.types.twilio_whatsapp.type.TwilioWhatsappType.check_credentials"
+        ) as mock_check_credentials:
+            mock_check_credentials.return_value = False
+
+            response = self.client.post(update_url, post_data)
+            self.assertFormError(response, "form", None, "Channel credentials don't appear to be valid.")

temba/channels/types/twilio_whatsapp/type.py

@@ -1,5 +1,7 @@
 from django.utils.translation import gettext_lazy as _
 
+from temba.channels.types.twilio.type import TwilioType
+from temba.channels.types.twilio.views import UpdateForm
 from temba.contacts.models import URN
 
 from ...models import ChannelType
@@ -24,6 +26,7 @@ class TwilioWhatsappType(ChannelType):
     ) % {"link": '<a target="_blank" href="https://www.twilio.com/whatsapp/">Twilio WhatsApp</a>'}
 
     claim_view = ClaimView
+    update_form = UpdateForm
 
     schemes = [URN.WHATSAPP_SCHEME]
     max_length = 1600
@@ -58,3 +61,6 @@ class TwilioWhatsappType(ChannelType):
 
     def get_error_ref_url(self, channel, code: str) -> str:
         return f"https://www.twilio.com/docs/api/errors/{code}"
+
+    def check_credentials(self, config: dict) -> bool:
+        return TwilioType().check_credentials(config)

temba/channels/views.py

@@ -1,6 +1,7 @@
 import logging
 from collections import defaultdict
 from datetime import timedelta
+from typing import Any
 
 import nexmo
 import phonenumbers
@@ -424,6 +425,14 @@ def add_config_field(self, config_key: str, field, *, default):
         self.fields[config_key] = field
         self.config_fields.append(config_key)
 
+    def clean(self) -> dict[str, Any]:
+        cleaned_data = super().clean()
+        updated_config = self.object.config | cleaned_data
+
+        if not Channel.get_type_from_code(self.object.channel_type).check_credentials(updated_config):
+            raise ValidationError(_("Channel credentials don't appear to be valid."))
+        return cleaned_data
+
     class Meta:
         model = Channel
         fields = ("name", "alert_email")