Merge pull request #34461 from dimagi/nh/redirect_url

Allow migrated domains to redirect form submissions to new host

corehq/apps/domain/decorators.py

@@ -1,11 +1,12 @@
 import logging
 from functools import wraps
+from urllib.parse import urljoin
 
 from django.conf import settings
 from django.contrib import messages
-from django.http import (
+from django.http import HttpRequest
+from django.http.response import (
     Http404,
-    HttpRequest,
     HttpResponse,
     HttpResponseForbidden,
     HttpResponseRedirect,
@@ -17,6 +18,7 @@
 from django.utils.translation import gettext as _
 from django.views import View
 
+from django_digest.decorators import httpdigest
 from django_otp import match_token
 from django_prbac.utils import has_privilege
 from oauth2_provider.oauth2_backends import get_oauthlib_core
@@ -56,7 +58,6 @@
     TWO_FACTOR_SUPERUSER_ROLLOUT,
 )
 from corehq.util.soft_assert import soft_assert
-from django_digest.decorators import httpdigest
 
 auth_logger = logging.getLogger("commcare_auth")
 
@@ -85,7 +86,10 @@ def login_and_domain_required(view_func):
     def _inner(req, domain, *args, **kwargs):
         user = req.user
         domain_name, domain_obj = load_domain(req, domain)
-        def call_view(): return view_func(req, domain_name, *args, **kwargs)
+
+        def call_view():
+            return view_func(req, domain_name, *args, **kwargs)
+
         if not domain_obj:
             msg = _('The domain "{domain}" was not found.').format(domain=domain_name)
             raise Http404(msg)
@@ -454,10 +458,10 @@ def _inner(request, domain, *args, **kwargs):
             domain_obj = Domain.get_by_name(domain)
             _ensure_request_couch_user(request)
             if (
-                not api_key and
-                not getattr(request, 'skip_two_factor_check', False) and
-                domain_obj and
-                _two_factor_required(view_func, domain_obj, request)
+                not api_key
+                and not getattr(request, 'skip_two_factor_check', False)
+                and domain_obj
+                and _two_factor_required(view_func, domain_obj, request)
             ):
                 token = request.META.get('HTTP_X_COMMCAREHQ_OTP')
                 if not token and 'otp' in request.GET:
@@ -667,6 +671,13 @@ def _inner(request, *args, **kwargs):
 
 def check_domain_migration(view_func):
     def wrapped_view(request, domain, *args, **kwargs):
+        domain_obj = Domain.get_by_name(domain)
+        if domain_obj.redirect_url:
+            # IMPORTANT!
+            #     We assume that the domain name is the same on both
+            #     environments.
+            url = urljoin(domain_obj.redirect_url, request.path)
+            return HttpResponseRedirect(url)
         if DATA_MIGRATION.enabled(domain):
             auth_logger.info(
                 "Request rejected domain=%s reason=%s request=%s",

corehq/apps/domain/management/commands/redirect_url.py

@@ -0,0 +1,67 @@
+from django.core.exceptions import ValidationError
+from django.core.management.base import BaseCommand, CommandError
+from django.core.validators import URLValidator
+
+from corehq.apps.domain.models import Domain
+from corehq.toggles import DATA_MIGRATION
+
+
+class Command(BaseCommand):
+    help = (
+        'Sets the redirect URL for a "308 Permanent Redirect" response to '
+        'form submissions and syncs. Only valid for domains that have been '
+        'migrated to new environments. Set the schema and hostname only '
+        '(e.g. "https://example.com/"). The rest of the path will be appended '
+        'for redirecting different endpoints. THIS FEATURE ASSUMES THE DOMAIN '
+        'NAME IS THE SAME ON BOTH ENVIRONMENTS.'
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument('domain')
+        parser.add_argument(
+            '--set',
+            help="The URL to redirect to",
+        )
+        parser.add_argument(
+            '--unset',
+            help="Remove the current redirect",
+            action='store_true',
+        )
+
+    def handle(self, domain, **options):
+        domain_obj = Domain.get_by_name(domain)
+
+        if options['set']:
+            _assert_data_migration(domain)
+            url = options['set']
+            _assert_valid_url(url)
+            domain_obj.redirect_url = url
+            domain_obj.save()
+
+        elif options['unset']:
+            domain_obj.redirect_url = ''
+            domain_obj.save()
+
+        if domain_obj.redirect_url:
+            self.stdout.write(
+                'Form submissions and syncs are redirected to '
+                f'{domain_obj.redirect_url}'
+            )
+        else:
+            self.stdout.write('Redirect URL not set')
+
+
+def _assert_data_migration(domain):
+    if not DATA_MIGRATION.enabled(domain):
+        raise CommandError(f'Domain {domain} is not migrated.')
+
+
+def _assert_valid_url(url):
+    if not url.startswith('https'):
+        raise CommandError(f'{url} is not a secure URL.')
+
+    validate = URLValidator()
+    try:
+        validate(url)
+    except ValidationError:
+        raise CommandError(f'{url} is not a valid URL.')

corehq/apps/domain/models.py

@@ -449,6 +449,9 @@ class Domain(QuickCachedDocumentMixin, BlobMixin, Document, SnapshotMixin):
     ga_opt_out = BooleanProperty(default=False)
     orphan_case_alerts_warning = BooleanProperty(default=False)
 
+    # For domains that have been migrated to a different environment
+    redirect_url = StringProperty()
+
     @classmethod
     def wrap(cls, data):
         # for domains that still use original_doc

corehq/apps/domain/tests/test_redirect_url.py

@@ -0,0 +1,185 @@
+from contextlib import contextmanager
+from io import StringIO
+
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.test import TestCase
+from django.test.client import Client
+from django.urls import reverse
+
+from corehq.apps.domain.models import Domain
+from corehq.apps.domain.shortcuts import create_domain
+from corehq.apps.users.models import WebUser
+from corehq.const import OPENROSA_VERSION_2
+from corehq.middleware import OPENROSA_VERSION_HEADER
+from corehq.util.test_utils import flag_enabled
+
+DOMAIN = 'test-redirect-url'
+
+
+class TestRedirectUrlCommand(TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain_obj = create_domain(DOMAIN)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.domain_obj.delete()
+        super().tearDownClass()
+
+    def test_data_migration_not_enabled(self):
+        with self.assertRaisesRegex(
+            CommandError,
+            r'^Domain test\-redirect\-url is not migrated\.$'
+        ):
+            self._call_redirect_url('--set', 'https://example.com/')
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_no_https(self):
+        with self.assertRaisesRegex(
+            CommandError,
+            r'^http://example.com/ is not a secure URL\.$'
+        ):
+            self._call_redirect_url('--set', 'http://example.com/')
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_bad_url(self):
+        with self.assertRaisesRegex(
+            CommandError,
+            r'^https://example/ is not a valid URL\.$'
+        ):
+            self._call_redirect_url('--set', 'https://example/')
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_set_url(self):
+        stdout = self._call_redirect_url('--set', 'https://example.com/')
+        self.assertEqual(
+            stdout,
+            'Form submissions and syncs are redirected to '
+            'https://example.com/\n'
+        )
+
+    def test_unset_url_data_migration_not_enabled(self):
+        with _set_redirect_url():
+            stdout = self._call_redirect_url('--unset')
+            self.assertEqual(stdout, 'Redirect URL not set\n')
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_unset_url_migration_enabled(self):
+        with _set_redirect_url():
+            stdout = self._call_redirect_url('--unset')
+            self.assertEqual(stdout, 'Redirect URL not set\n')
+
+    def test_return_set_url_data_migration_not_enabled(self):
+        with _set_redirect_url():
+            stdout = self._call_redirect_url()
+            self.assertEqual(
+                stdout,
+                'Form submissions and syncs are redirected to '
+                'https://example.com/\n'
+            )
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_return_set_url_migration_enabled(self):
+        with _set_redirect_url():
+            stdout = self._call_redirect_url()
+            self.assertEqual(
+                stdout,
+                'Form submissions and syncs are redirected to '
+                'https://example.com/\n'
+            )
+
+    def test_return_unset_url_data_migration_not_enabled(self):
+        stdout = self._call_redirect_url()
+        self.assertEqual(stdout, 'Redirect URL not set\n')
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_return_unset_url_migration_enabled(self):
+        stdout = self._call_redirect_url()
+        self.assertEqual(stdout, 'Redirect URL not set\n')
+
+    @staticmethod
+    def _call_redirect_url(*args, **kwargs):
+        stdout = StringIO()
+        call_command(
+            'redirect_url', DOMAIN, *args,
+            stdout=stdout, **kwargs,
+        )
+        return stdout.getvalue()
+
+
+class TestCheckDomainMigration(TestCase):
+    """
+    Tests the ``receiver_post`` view, which is wrapped with the
+    ``corehq.apps.domain.decorators.check_domain_migration`` decorator.
+
+    All relevant views are protected by that decorator during and after
+    data migration. These tests verify that the decorator returns a 302
+    redirect response when the domain's ``redirect_url`` is set, and a
+    503 service unavailable response when it is not set.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain_obj = create_domain(DOMAIN)
+        cls.user = WebUser.create(
+            None, 'admin', 'Passw0rd!',
+            None, None,
+        )
+        cls.user.add_domain_membership(DOMAIN, is_admin=True)
+        cls.user.save()
+        cls.client = Client()
+        cls.client.login(username='admin', password='Passw0rd!')
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.user.delete(DOMAIN, deleted_by=None)
+        cls.domain_obj.delete()
+        super().tearDownClass()
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_redirect_response(self):
+        with _set_redirect_url():
+            response = self._submit_form()
+            self.assertEqual(response.status_code, 302)
+            self.assertEqual(
+                response.url,
+                'https://example.com/a/test-redirect-url/receiver/'
+            )
+
+    @flag_enabled('DATA_MIGRATION')
+    def test_service_unavailable_response(self):
+        response = self._submit_form()
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(
+            response.content.decode('utf-8'),
+            'Service Temporarily Unavailable',
+        )
+
+    def _submit_form(self):
+        form = """<?xml version='1.0' ?>
+        <form>Not a real form</form>
+        """
+        with StringIO(form) as f:
+            response = self.client.post(
+                reverse("receiver_post", args=[DOMAIN]),
+                {"xml_submission_file": f},
+                **{OPENROSA_VERSION_HEADER: OPENROSA_VERSION_2}
+            )
+        return response
+
+
+@contextmanager
+def _set_redirect_url():
+    domain_obj = Domain.get_by_name(DOMAIN)
+    domain_obj.redirect_url = 'https://example.com/'
+    domain_obj.save()
+    try:
+        yield
+    finally:
+        domain_obj = Domain.get_by_name(DOMAIN)
+        domain_obj.redirect_url = ''
+        domain_obj.save()