Merge pull request #28493 from dimagi/cz/role-ui

add ability to create domain-scoped API keys

Author: kaapstorm

corehq/apps/api/tests/test_auth.py

@@ -22,6 +22,9 @@ def setUpClass(cls):
         cls.password = '***'
         cls.user = WebUser.create(cls.domain, cls.username, cls.password, None, None)
         cls.api_key, _ = HQApiKey.objects.get_or_create(user=WebUser.get_django_user(cls.user))
+        cls.domain_api_key, _ = HQApiKey.objects.get_or_create(user=WebUser.get_django_user(cls.user),
+                                                               name='domain-scoped',
+                                                               domain=cls.domain)
 
     @classmethod
     def tearDownClass(cls):
@@ -85,6 +88,19 @@ def test_auth_type_basic(self):
 
 class LoginAndDomainAuthenticationTest(AuthenticationTestBase):
 
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.domain2 = 'api-test-other'
+        cls.project2 = Domain.get_or_create_with_name(cls.domain2, is_active=True)
+        cls.user.add_domain_membership(cls.domain2, is_admin=True)
+        cls.user.save()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.project2.delete()
+        super().tearDownClass()
+
     def test_login_no_auth_no_domain(self):
         self.assertAuthenticationFail(LoginAndDomainAuthentication(), self._get_request())
 
@@ -94,6 +110,32 @@ def test_login_no_auth_with_domain(self):
     def test_login_with_domain(self):
         self.assertAuthenticationSuccess(LoginAndDomainAuthentication(),
                                          self._get_request_with_api_key(domain=self.domain))
+        self.assertAuthenticationSuccess(LoginAndDomainAuthentication(),
+                                         self._get_request_with_api_key(domain=self.domain2))
+
+    def test_login_with_domain_key(self):
+        self.assertAuthenticationSuccess(
+            LoginAndDomainAuthentication(),
+            self._get_request(
+                self.domain,
+                HTTP_AUTHORIZATION=self._construct_api_auth_header(
+                    self.username,
+                    self.domain_api_key
+                )
+            )
+        )
+
+    def test_login_with_domain_key_wrong(self):
+        self.assertAuthenticationFail(
+            LoginAndDomainAuthentication(),
+            self._get_request(
+                self.domain2,
+                HTTP_AUTHORIZATION=self._construct_api_auth_header(
+                    self.username,
+                    self.domain_api_key
+                )
+            )
+        )
 
     def test_login_with_wrong_domain(self):
         project = Domain.get_or_create_with_name('api-test-fail', is_active=True)

corehq/apps/settings/forms.py

@@ -257,21 +257,31 @@ def __init__(self, **kwargs):
 
 
 class HQApiKeyForm(forms.Form):
+    ALL_DOMAINS = ''
     name = forms.CharField()
     ip_allowlist = SimpleArrayField(
         forms.GenericIPAddressField(),
         label=ugettext_lazy("Allowed IP Addresses (comma separated)"),
         required=False
     )
+    domain = forms.ChoiceField(
+        required=False,
+        help_text=ugettext_lazy("Limit the key's access to a single project space")
+    )
 
     def __init__(self, *args, **kwargs):
+        self.couch_user = kwargs.pop('couch_user')
         super().__init__(*args, **kwargs)
 
+        user_domains = self.couch_user.get_domains()
+        all_domains = (self.ALL_DOMAINS, _('All Projects'))
+        self.fields['domain'].choices = [all_domains] + [(d, d) for d in user_domains]
         self.helper = HQFormHelper()
         self.helper.layout = Layout(
             crispy.Fieldset(
                 ugettext_lazy("Add New API Key"),
                 crispy.Field('name'),
+                crispy.Field('domain'),
                 crispy.Field('ip_allowlist'),
             ),
             hqcrispy.FormActions(
@@ -292,6 +302,7 @@ def create_key(self, user):
                 name=self.cleaned_data['name'],
                 ip_allowlist=self.cleaned_data['ip_allowlist'],
                 user=user,
+                domain=self.cleaned_data['domain'] or '',
             )
             return new_key
 

corehq/apps/settings/templates/settings/user_api_keys.html

@@ -8,6 +8,7 @@
   <script type="text/html" id="base-user-api-key-template">
     <td data-bind="text: name"></td>
     <td data-bind="text: key"></td>
+    <td data-bind="text: domain"></td>
     <td data-bind="text: ip_allowlist"></td>
     <td data-bind="text: created"></td>
     <td>
@@ -59,7 +60,9 @@ <h3>
   <script type="text/html" id="new-user-api-key-template">
     <td data-bind="text: name"></td>
     <td data-bind="text: key"></td>
+    <td data-bind="text: domain"></td>
     <td data-bind="text: ip_allowlist"></td>
     <td data-bind="text: created"></td>
+    <td></td>
   </script>
 {% endblock %}

corehq/apps/settings/views.py

@@ -528,6 +528,7 @@ def column_names(self):
         return [
             _("Name"),
             _("API Key"),
+            _("Project"),
             _("IP Allowlist"),
             _("Created"),
             _("Delete"),
@@ -546,6 +547,7 @@ def paginated_list(self):
                     "id": api_key.id,
                     "name": api_key.name,
                     "key": redacted_key,
+                    "domain": api_key.domain or _('All Projects'),
                     "ip_allowlist": (
                         ", ".join(api_key.ip_allowlist)
                         if api_key.ip_allowlist else _("All IP Addresses")
@@ -562,8 +564,8 @@ def post(self, *args, **kwargs):
 
     def get_create_form(self, is_blank=False):
         if self.request.method == 'POST' and not is_blank:
-            return HQApiKeyForm(self.request.POST)
-        return HQApiKeyForm()
+            return HQApiKeyForm(self.request.POST, couch_user=self.request.couch_user)
+        return HQApiKeyForm(couch_user=self.request.couch_user)
 
     def get_create_item_data(self, create_form):
         try:
@@ -576,6 +578,7 @@ def get_create_item_data(self, create_form):
                 'id': new_api_key.id,
                 'name': new_api_key.name,
                 'key': f"{new_api_key.key} ({copy_key_message})",
+                "domain": new_api_key.domain or _('All Projects'),
                 'ip_allowlist': new_api_key.ip_allowlist,
                 'created': new_api_key.created.isoformat()
             },

corehq/apps/users/admin.py

@@ -45,3 +45,11 @@ class DomainPermissionsMirrorAdmin(admin.ModelAdmin):
 
 
 admin.site.register(DomainPermissionsMirror, DomainPermissionsMirrorAdmin)
+
+
+class HQApiKeyAdmin(admin.ModelAdmin):
+    list_display = ['user', 'name', 'created', 'domain']
+    list_filter = ['created', 'domain']
+
+
+admin.site.register(HQApiKey, HQApiKeyAdmin)

corehq/apps/users/decorators.py

@@ -77,17 +77,30 @@ def require_api_permission(permission, data=None, login_decorator=login_and_doma
     permissions_to_check = {permission, api_access_permission}
 
     def permission_check(request, domain):
-        if getattr(request, 'api_key', None) and request.api_key.role:
-            role = request.api_key.role
+        # first check user permissions and return immediately if not valid
+        user_has_permission = all(
+            request.couch_user.has_permission(domain, p, data=data)
+            for p in permissions_to_check
+        )
+        if not user_has_permission:
+            return False
+
+        # then check domain and role scopes, if present
+        api_key = getattr(request, 'api_key', None)
+
+        if not api_key:
+            return True  # only api keys support additional checks
+        elif api_key.role:
             return (
-                role.domain == domain
-                and all(request.couch_user.has_permission(domain, p, data=data)
-                        for p in permissions_to_check)
-                and all(role.permissions.has(p, data) for p in permissions_to_check)
+                api_key.role.domain == domain
+                and all(api_key.role.permissions.has(p, data) for p in permissions_to_check)
             )
+        elif api_key.domain:
+            # we've already checked for user and role permissions so all that's left is domain matching
+            return domain == api_key.domain
         else:
-            return all(request.couch_user.has_permission(domain, p, data=data)
-                       for p in permissions_to_check)
+            # unscoped API key defaults to user permissions
+            return True
 
     return require_permission_raw(
         None, login_decorator,

corehq/apps/users/models.py

@@ -3081,4 +3081,6 @@ def role(self):
                 return UserRole.get(self.role_id)
             except ResourceNotFound:
                 logging.exception('no role with id %s found in domain %s' % (self.role_id, self.domain))
+        elif self.domain:
+            return CouchUser.from_django_user(self.user).get_domain_membership(self.domain).role
         return None