Merge pull request #36044 from dimagi/es/ip-blocking

Restrict project access by IP address

Author: minhaminha

corehq/apps/domain/deletion.py

@@ -409,6 +409,7 @@ def _delete_demo_user_restores(domain_name):
     ModelDeletion('integration', 'SimprintsIntegration', 'domain'),
     ModelDeletion('integration', 'KycConfig', 'domain'),
     ModelDeletion('integration', 'MoMoConfig', 'domain'),
+    ModelDeletion('ip_access', 'IPAccessConfig', 'domain'),
     ModelDeletion('linked_domain', 'DomainLink', 'linked_domain', ['DomainLinkHistory']),
     CustomDeletion('scheduling', _delete_sms_content_events_schedules, [
         'SMSContent', 'EmailContent', 'SMSSurveyContent',

corehq/apps/domain/forms.py

@@ -1,5 +1,6 @@
 import datetime
 import io
+import ipaddress
 import json
 import logging
 import uuid
@@ -222,6 +223,92 @@ def save(self, user, domain):
         return True
 
 
+class IPAccessConfigForm(forms.Form):
+    """
+    Form for updating a project's IP Access Configuration
+    """
+    country_allowlist = forms.MultipleChoiceField(
+        label=_("Allowed Countries"),
+        choices=sorted(list(COUNTRIES.items()), key=lambda x: x[1]),
+        required=False,
+    )
+
+    ip_allowlist = forms.CharField(
+        label=_("Allowed IPs"),
+        required=False,
+        help_text='IPs that will be allowed access to your project, regardless of country of origin. '
+                  'Please configure your list to be comma and space separated, '
+                  'e.g. 192.168.0.1, 192.168.1.1, 192.168.2.1',
+    )
+
+    ip_denylist = forms.CharField(
+        label=_("Denied IPs"),
+        required=False,
+        help_text='IPs that will be denied access to your project, regardless of country of origin.',
+    )
+
+    comment = forms.CharField(
+        label=_("Additional Notes"),
+        widget=forms.Textarea(attrs={"class": "vertical-resize"}),
+        required=False
+    )
+
+    def __init__(self, *args, **kwargs):
+        self.current_ip = kwargs.pop('current_ip', None)
+        self.current_country = kwargs.pop('current_country', None)
+        super(IPAccessConfigForm, self).__init__(*args, **kwargs)
+        self.helper = hqcrispy.HQFormHelper(self)
+        self.helper.form_id = 'ip-access-config-form'
+        self.helper.layout = crispy.Layout(
+            crispy.Fieldset(
+                _("Edit IP Access Config"),
+                "country_allowlist",
+                "ip_allowlist",
+                "ip_denylist",
+                "comment"
+            ),
+            hqcrispy.FormActions(
+                StrictButton(
+                    _("Update IP Access Config"),
+                    type="submit",
+                    css_class='btn-primary',
+                )
+            )
+        )
+
+    def clean(self):
+        allow_list = self.cleaned_data['ip_allowlist'].split(", ") if self.cleaned_data['ip_allowlist'] else []
+        deny_list = self.cleaned_data['ip_denylist'].split(", ") if self.cleaned_data['ip_denylist'] else []
+
+        # Ensure an IP isn't in both lists
+        if (allow_list and deny_list) and set(allow_list).intersection(set(deny_list)):
+            self.add_error('ip_allowlist', _("There are IP addresses in both the Allowed and Denied lists. "
+                                             "Please ensure an IP address is only in one list at a time."))
+
+        # Ensure inputs are valid IPs, checks both IPv4 and IPv6
+        for ip in allow_list + deny_list:
+            try:
+                ipaddress.ip_address(ip)
+            except ValueError as e:
+                raise ValidationError(e)
+
+        self.cleaned_data['ip_allowlist'] = allow_list
+        self.cleaned_data['ip_denylist'] = deny_list
+
+        # Additional validation
+        if self.cleaned_data['country_allowlist']:
+            if not settings.MAXMIND_LICENSE_KEY:
+                self.add_error('country_allowlist', _("The Allowed Countries field cannot be saved because "
+                                                      "MaxMind is not configured for your environment"))
+            elif (self.current_country and self.current_country not in self.cleaned_data['country_allowlist']
+                  and self.current_ip not in self.cleaned_data['ip_allowlist']):
+                self.add_error('country_allowlist', _("Please add your own country or IP to the Allowed IPs field "
+                                                      "to avoid being locked out."))
+        if self.current_ip in self.cleaned_data['ip_denylist']:
+            self.add_error('ip_denylist', _("You cannot put your current IP address in the Denied IPs field"))
+        return self.cleaned_data
+
+
 class TransferDomainFormErrors(object):
     USER_DNE = gettext_lazy('The user being transferred to does not exist')
     DOMAIN_MISMATCH = gettext_lazy('Mismatch in domains when confirming')

corehq/apps/domain/static/domain/js/ip_access_config.js

@@ -0,0 +1,7 @@
+import "hqwebapp/js/htmx_and_alpine";
+import $ from 'jquery';
+import 'select2/dist/js/select2.full.min';
+
+$(function () {
+    $("#id_country_allowlist").select2();
+});

corehq/apps/domain/templates/domain/admin/ip_access_config.html

@@ -0,0 +1,10 @@
+{% extends "hqwebapp/bootstrap5/base_section.html" %}
+{% load hq_shared_tags %}
+{% load crispy_forms_tags %}
+{% load i18n %}
+
+{% js_entry "domain/js/ip_access_config" %}
+
+{% block page_content %}
+  {% crispy ip_access_config_form %}
+{% endblock %}

corehq/apps/domain/urls.py

@@ -67,6 +67,7 @@
     DefaultProjectSettingsView,
     EditBasicProjectInfoView,
     EditDomainAlertView,
+    EditIPAccessConfigView,
     EditMyProjectSettingsView,
     EditPrivacySecurityView,
     FeaturePreviewsView,
@@ -142,6 +143,7 @@
     url(r'^call_center_owner_options/', CallCenterOwnerOptionsView.as_view(),
         name=CallCenterOwnerOptionsView.url_name),
     url(r'^privacy/$', EditPrivacySecurityView.as_view(), name=EditPrivacySecurityView.urlname),
+    url(r'^ip_access/$', EditIPAccessConfigView.as_view(), name=EditIPAccessConfigView.urlname),
     url(r'^subscription/change/$', SelectPlanView.as_view(), name=SelectPlanView.urlname),
     url(r'^subscription/change/confirm/$', ConfirmSelectedPlanView.as_view(),
         name=ConfirmSelectedPlanView.urlname),

corehq/apps/domain/views/__init__.py

@@ -56,6 +56,7 @@
     EditBasicProjectInfoView,
     EditMyProjectSettingsView,
     EditPrivacySecurityView,
+    EditIPAccessConfigView,
     FeaturePreviewsView,
     CustomPasswordResetView,
     RecoveryMeasuresHistory,

corehq/apps/domain/views/settings.py

@@ -24,7 +24,7 @@
 
 from corehq.apps.accounting.decorators import always_allow_project_access
 from corehq.apps.enterprise.mixins import ManageMobileWorkersMixin
-from dimagi.utils.web import json_response
+from dimagi.utils.web import json_response, get_ip
 
 from corehq import feature_previews, privileges, toggles
 from corehq.apps.app_manager.dbaccessors import get_apps_in_domain
@@ -50,13 +50,15 @@
     DomainMetadataForm,
     PrivacySecurityForm,
     ProjectSettingsForm,
+    IPAccessConfigForm,
     clean_password
 )
 from corehq.apps.domain.models import Domain
 from corehq.apps.domain.views.base import BaseDomainView
 from corehq.apps.hqwebapp.decorators import use_bootstrap5
 from corehq.apps.hqwebapp.models import Alert
 from corehq.apps.hqwebapp.signals import clear_login_attempts
+from corehq.apps.ip_access.models import IPAccessConfig, get_ip_country
 from corehq.apps.locations.permissions import location_safe
 from corehq.apps.ota.models import MobileRecoveryMeasure
 from corehq.apps.users.decorators import require_can_manage_domain_alerts
@@ -271,6 +273,77 @@ def post(self, request, *args, **kwargs):
         return self.get(request, *args, **kwargs)
 
 
+@method_decorator([
+    domain_admin_required,
+    use_bootstrap5,
+    toggles.IP_ACCESS_CONTROLS.required_decorator(),
+], name='dispatch')
+class EditIPAccessConfigView(BaseProjectSettingsView):
+    template_name = 'domain/admin/ip_access_config.html'
+    urlname = 'ip_access_config'
+    page_title = gettext_lazy("IP Access")
+
+    @property
+    @memoized
+    def form(self):
+        initial = {}
+        domain_config = self.ip_access_config
+        if domain_config:
+            display_spacer = ", "
+            initial.update({
+                'country_allowlist': domain_config.country_allowlist,
+                'ip_allowlist': display_spacer.join(domain_config.ip_allowlist),
+                'ip_denylist': display_spacer.join(domain_config.ip_denylist),
+                'comment': domain_config.comment
+            })
+        if self.request.method == 'POST':
+            return IPAccessConfigForm(self.request.POST, initial=initial,
+                                      current_ip=self.current_ip, current_country=self.ip_country)
+        return IPAccessConfigForm(initial=initial)
+
+    @cached_property
+    def ip_access_config(self):
+        try:
+            return IPAccessConfig.objects.get(domain=self.domain)
+        except IPAccessConfig.DoesNotExist:
+            return None
+
+    @cached_property
+    def ip_country(self):
+        if settings.MAXMIND_LICENSE_KEY:
+            return get_ip_country(self.current_ip)
+        return None
+
+    @property
+    def current_ip(self):
+        return get_ip(self.request)
+
+    @property
+    def page_context(self):
+        return {
+            'ip_access_config_form': self.form,
+        }
+
+    def post(self, request, *args, **kwargs):
+        if self.form.is_valid():
+            domain_config = self.ip_access_config
+            should_save = True
+            if not domain_config:
+                should_save = False
+                domain_config = IPAccessConfig()
+                domain_config.domain = self.domain
+            for attr, value in self.form.cleaned_data.items():
+                current_value = getattr(domain_config, attr)
+                if value != current_value:
+                    should_save = True
+                    setattr(domain_config, attr, value)
+            if should_save:
+                domain_config.save()
+                messages.success(request, _("Your IP Access settings have been saved!"))
+            return HttpResponseRedirect(reverse(self.urlname, args=[self.domain]))
+        return self.get(request, *args, **kwargs)
+
+
 @location_safe
 def logo(request, domain):
     logo = Domain.get_by_name(domain).get_custom_logo()

corehq/apps/dump_reload/sql/dump.py

@@ -157,6 +157,7 @@
     FilteredModelIteratorBuilder('integration.SimprintsIntegration', SimpleFilter('domain')),
     FilteredModelIteratorBuilder('integration.KycConfig', SimpleFilter('domain')),
     FilteredModelIteratorBuilder('integration.MoMoConfig', SimpleFilter('domain')),
+    FilteredModelIteratorBuilder('ip_access.IPAccessConfig', SimpleFilter('domain')),
     FilteredModelIteratorBuilder('phonelog.DeviceReportEntry', SimpleFilter('domain')),
     FilteredModelIteratorBuilder('phonelog.ForceCloseEntry', SimpleFilter('domain')),
     FilteredModelIteratorBuilder('phonelog.UserErrorEntry', SimpleFilter('domain')),

corehq/apps/hqwebapp/views.py

@@ -100,7 +100,12 @@
 from corehq.toggles import CLOUDCARE_LATEST_BUILD
 from corehq.util.context_processors import commcare_hq_names
 from corehq.util.email_event_utils import handle_email_sns_event
-from corehq.util.metrics import create_metrics_event, metrics_counter, metrics_gauge
+from corehq.util.metrics import (
+    create_metrics_event,
+    limit_domains,
+    metrics_counter,
+    metrics_gauge,
+)
 from corehq.util.metrics.const import TAG_UNKNOWN, MPM_MAX
 from corehq.util.metrics.utils import sanitize_url
 from corehq.util.public_only_requests.public_only_requests import get_public_only_session
@@ -114,7 +119,6 @@
 from dimagi.utils.django.request import mutable_querydict
 from dimagi.utils.logging import notify_exception, notify_error
 from dimagi.utils.web import get_url_base
-from no_exceptions.exceptions import Http403
 from soil import DownloadBase
 from soil import views as soil_views
 
@@ -332,8 +336,8 @@ def server_up(req):
 
 
 @use_bootstrap5
-def _no_permissions_message(request, template_name="403.html", message=None):
-    t = loader.get_template(template_name)
+def _no_permissions_message(request, message=None):
+    t = loader.get_template("403.html")
     return t.render(
         context={
             'MEDIA_URL': settings.MEDIA_URL,
@@ -345,16 +349,18 @@ def _no_permissions_message(request, template_name="403.html", message=None):
 
 
 @use_bootstrap5
-def no_permissions(request, redirect_to=None, template_name="403.html", message=None, exception=None):
-    """
-    403 error handler.
-    """
-    return HttpResponseForbidden(_no_permissions_message(request, template_name, message))
+def no_permissions(request, redirect_to=None, message=None, exception=None):
+    """403 error handler. Called automatically by Django on PermissionDenied
 
-
-@use_bootstrap5
-def no_permissions_exception(request, template_name="403.html", message=None):
-    return Http403(_no_permissions_message(request, template_name, message))
+    :param exception: Instance of PermissionDenied or subclass. Class name will
+    be reported to datadog.
+    """
+    metrics_counter('commcare.no_permissions.count', tags={
+        'domain': limit_domains(getattr(request, 'domain', '__other__')),
+        'exception_type': exception.__class__.__name__ if exception else 'Other',
+    })
+    message = message or getattr(exception, 'user_facing_message', None)
+    return HttpResponseForbidden(_no_permissions_message(request, message))
 
 
 @use_bootstrap5

corehq/apps/ip_access/__init__.py

@@ -0,0 +1,5 @@
+from django.contrib import admin
+
+from .models import IPAccessConfig
+
+admin.site.register(IPAccessConfig)

corehq/apps/ip_access/admin.py

@@ -0,0 +1,52 @@
+from django.core.exceptions import PermissionDenied
+from django.utils.deprecation import MiddlewareMixin
+from django.utils.translation import gettext_lazy as _
+
+from dimagi.utils.web import get_ip
+
+from corehq.toggles import IP_ACCESS_CONTROLS
+
+from .models import IPAccessConfig
+
+
+class IPPermissionDenied(PermissionDenied):
+    user_facing_message = _(
+        "You cannot access this page from your current IP address"
+    )
+
+
+class IPAccessMiddleware(MiddlewareMixin):
+
+    def process_view(self, request, view_func, view_args, view_kwargs):
+        if not request.user.is_authenticated or 'domain' not in view_kwargs:
+            return
+
+        domain = view_kwargs['domain']
+        if IP_ACCESS_CONTROLS.enabled(domain) and not is_valid_ip(request, domain):
+            raise IPPermissionDenied()
+
+
+def is_valid_ip(request, domain):
+    ip = get_ip(request)
+    block_key = f"hq_session_blocked_ips-{domain}"
+    allow_key = f"hq_session_ips-{domain}"
+    if block_key not in request.session:
+        request.session[block_key] = []
+    if allow_key not in request.session:
+        request.session[allow_key] = []
+
+    if ip in request.session[block_key]:
+        return False
+    if ip in request.session[allow_key]:
+        return True
+
+    try:
+        config = IPAccessConfig.objects.get(domain=domain)
+    except IPAccessConfig.DoesNotExist:
+        config = None
+    if not config or config.is_allowed(ip):
+        request.session[allow_key].append(ip)
+        return True
+    else:
+        request.session[block_key].append(ip)
+        return False