Task/callback UI tests

Makefile

@@ -58,6 +58,7 @@ run-dev:
 .PHONY: format
 format:
 	ruff check --select I --fix .
+	ruff check
 	ruff format .
 	mypy ./
 	npx prettier --write app/assets/javascripts app/assets/stylesheets tests_cypress/cypress/e2e

app/main/forms.py

@@ -23,6 +23,7 @@
     SelectField,
     SelectMultipleField,
     StringField,
+    SubmitField,
     TextAreaField,
     ValidationError,
     validators,
@@ -49,6 +50,7 @@
     LettersNumbersAndFullStopsOnly,
     NoCommasInPlaceHolders,
     OnlySMSCharacters,
+    ValidCallbackUrl,
     ValidEmail,
     ValidGovEmail,
     validate_email_from,
@@ -342,7 +344,8 @@ def bind_field(self, form, unbound_field, options):
             filters = [strip_whitespace] if not issubclass(unbound_field.field_class, no_filter_fields) else []
             filters += unbound_field.kwargs.get("filters", [])
             bound = unbound_field.bind(form=form, filters=filters, **options)
-            bound.get_form = weakref.ref(form)  # GC won't collect the form if we don't use a weakref
+            # GC won't collect the form if we don't use a weakref
+            bound.get_form = weakref.ref(form)
             return bound
 
 
@@ -1407,15 +1410,16 @@ def __init__(self, *args, **kwargs):
 
 class CallbackForm(StripWhitespaceForm):
     def validate(self, extra_validators=None):
-        return super().validate(extra_validators) or self.url.data == ""
+        return super().validate(extra_validators)
 
 
 class ServiceReceiveMessagesCallbackForm(CallbackForm):
     url = StringField(
         "URL",
         validators=[
             DataRequired(message=_l("This cannot be empty")),
-            Regexp(regex="^https.*", message=_l("Must be a valid https URL")),
+            Regexp(regex="^https.*", message=_l("Enter a URL that starts with https://")),
+            ValidCallbackUrl(),
         ],
     )
     bearer_token = PasswordFieldShowHasContent(
@@ -1432,7 +1436,8 @@ class ServiceDeliveryStatusCallbackForm(CallbackForm):
         "URL",
         validators=[
             DataRequired(message=_l("This cannot be empty")),
-            Regexp(regex="^https.*", message=_l("Must be a valid https URL")),
+            Regexp(regex="^https.*", message=_l("Enter a URL that starts with https://")),
+            ValidCallbackUrl(),
         ],
     )
     bearer_token = PasswordFieldShowHasContent(
@@ -1442,6 +1447,7 @@ class ServiceDeliveryStatusCallbackForm(CallbackForm):
             Length(min=10, message=_l("Must be at least 10 characters")),
         ],
     )
+    test_response_time = SubmitField()
 
 
 class InternationalSMSForm(StripWhitespaceForm):
@@ -1885,7 +1891,8 @@ class BrandingPoolForm(StripWhitespaceForm):
 
     pool_branding = RadioField(
         _l("Select alternate logo"),
-        choices=[],  # Choices by default, override to get more refined options.
+        # Choices by default, override to get more refined options.
+        choices=[],
         validators=[DataRequired(message=_l("You must select an option to continue"))],
     )
 

app/main/validators.py

@@ -2,7 +2,9 @@
 import time
 
 import pwnedpasswords
-from flask import current_app
+import requests
+import validators
+from flask import current_app, g
 from flask_babel import _
 from flask_babel import lazy_gettext as _l
 from notifications_utils.field import Field
@@ -11,7 +13,7 @@
 from wtforms import ValidationError
 from wtforms.validators import Email
 
-from app import formatted_list, service_api_client
+from app import current_service, formatted_list, service_api_client
 from app.main._blocked_passwords import blocked_passwords
 from app.utils import Spreadsheet, email_safe, email_safe_name, is_gov_user
 
@@ -25,7 +27,8 @@ def __init__(self, message=None):
     def __call__(self, form, field):
         if current_app.config.get("HIPB_ENABLED", None):
             hibp_bad_password_found = False
-            for i in range(0, 3):  # Try 3 times. If the HIPB API is down then fall back to the old banlist.
+            # Try 3 times. If the HIPB API is down then fall back to the old banlist.
+            for i in range(0, 3):
                 try:
                     response = pwnedpasswords.check(field.data)
                     if response > 0:
@@ -141,6 +144,57 @@ def __call__(self, form, field):
             raise ValidationError(self.message)
 
 
+class ValidCallbackUrl:
+    def __init__(self, message="Enter a URL that starts with https://"):
+        self.message = message
+
+    def __call__(self, form, field):
+        if field.data:
+            validate_callback_url(field.data, form.bearer_token.data)
+
+
+def validate_callback_url(service_callback_url, bearer_token):
+    """Validates a callback URL, checking that it is https and by sending a POST request to the URL with a health_check parameter.
+    4xx responses are considered invalid. 5xx responses are considered valid as it indicates there is at least a service running
+    at the URL, and we are sending a payload that the service will not understand.
+
+    Args:
+        service_callback_url (str): The url to validate.
+        bearer_token (str): The bearer token to use in the request, specified by the user requesting callbacks.
+
+    Raises:
+        ValidationError: If the URL is not HTTPS or the http response is 4xx.
+    """
+    if not validators.url(service_callback_url):
+        current_app.logger.warning(
+            f"Unable to create callback for service: {current_service.id}. Error: Invalid callback URL format: URL: {service_callback_url}"
+        )
+        raise ValidationError(_l("Enter a URL that starts with https://"))
+
+    try:
+        response = requests.post(
+            url=service_callback_url,
+            allow_redirects=True,
+            data={"health_check": "true"},
+            headers={"Content-Type": "application/json", "Authorization": f"Bearer {bearer_token}"},
+            timeout=2,
+        )
+
+        g.callback_response_time = response.elapsed.total_seconds()
+
+        if response.status_code < 500 and response.status_code >= 400:
+            current_app.logger.warning(
+                f"Unable to create callback for service: {current_service.id} Error: Callback URL not reachable URL: {service_callback_url}"
+            )
+            raise ValidationError(_l("Check your service is running and not using a proxy we cannot access"))
+
+    except requests.RequestException as e:
+        current_app.logger.warning(
+            f"Unable to create callback for service: {current_service.id} Error: Callback URL not reachable URL: {service_callback_url} Exception: {e}"
+        )
+        raise ValidationError(_l("Check your service is running and not using a proxy we cannot access"))
+
+
 def validate_email_from(form, field):
     if email_safe(field.data) != field.data.lower():
         # fix their data instead of only warning them