[change] Create reusable class for importing CSV

openwisp_ipam/api/utils.py

@@ -1,5 +1,5 @@
 from django.core.exceptions import ValidationError
-from rest_framework.exceptions import NotFound
+from rest_framework.exceptions import NotFound, PermissionDenied
 
 
 class FilterByOrganization:
@@ -49,3 +49,37 @@ def get_organization_queryset(self):
 class FilterByParentManaged(FilterByParent):
     def get_organization_queryset(self, qs):
         return qs.filter(organization__in=self.request.user.organizations_managed)
+
+
+class AuthorizeCSVImport:
+    def post(self, request):
+        self.assert_organization_permissions(request)
+
+    def assert_organization_permissions(self, request):
+        if request.user.is_superuser:
+            return
+        organization = self.get_csv_organization()
+        if not organization:
+            # if organization in CSV doesn't exist, then check if
+            # user can create new organizations
+            create_org_permission = request.user.permissions.filter(
+                codename='add_organization'
+            )
+            if create_org_permission.exists():
+                return
+        else:
+            # else check if user has access to organization in CSV
+            if str(organization.pk) in self.get_user_organizations():
+                return
+        raise PermissionDenied()
+
+    def get_csv_organization(self):
+        raise NotImplementedError()
+
+    def get_user_organizations(self):
+        raise NotImplementedError()
+
+
+class AuthorizeCSVOrgManaged(AuthorizeCSVImport):
+    def get_user_organizations(self):
+        return self.request.user.organizations_managed

openwisp_ipam/api/views.py

@@ -1,5 +1,6 @@
 import csv
 from collections import OrderedDict
+from copy import deepcopy
 
 import swapper
 from django.http import HttpResponse
@@ -29,7 +30,11 @@
     IpRequestSerializer,
     SubnetSerializer,
 )
-from .utils import FilterByOrganizationManaged, FilterByParentManaged
+from .utils import (
+    AuthorizeCSVOrgManaged,
+    FilterByOrganizationManaged,
+    FilterByParentManaged,
+)
 
 IpAddress = swapper.load_model('openwisp_ipam', 'IpAddress')
 Subnet = swapper.load_model('openwisp_ipam', 'Subnet')
@@ -42,6 +47,15 @@ def get_parent_queryset(self):
         return qs
 
 
+class ImportSubnetCSVMixin(AuthorizeCSVOrgManaged):
+    def get_csv_organization(self):
+        data = self.subnet_model._get_csv_reader(
+            self, deepcopy(self.request.FILES['csvfile'])
+        )
+        org = Organization.objects.get(name=list(data)[2][0].strip())
+        return org
+
+
 class ListViewPagination(pagination.PageNumberPagination):
     page_size = 10
     page_size_query_param = 'page_size'
@@ -220,19 +234,20 @@ def post(self, request, *args, **kwargs):
         return Response(None)
 
 
-class ImportSubnetView(CreateAPIView):
+class ImportSubnetView(ImportSubnetCSVMixin, CreateAPIView):
     subnet_model = Subnet
     queryset = Subnet.objects.none()
     serializer_class = ImportSubnetSerializer
     authentication_classes = (BearerAuthentication, SessionAuthentication)
     permission_classes = (DjangoModelPermissions,)
 
     def post(self, request, *args, **kwargs):
+        super().post(request)
         file = request.FILES['csvfile']
         if not file.name.endswith(('.csv', '.xls', '.xlsx')):
             return Response({'error': _('File type not supported.')}, status=400)
         try:
-            self.subnet_model().import_csv(file, self.request.user)
+            self.subnet_model().import_csv(file)
         except CsvImportException as e:
             return Response({'error': _(str(e))}, status=400)
         return Response({'detail': _('Data imported successfully.')})

openwisp_ipam/base/models.py

@@ -3,7 +3,7 @@
 from ipaddress import ip_address, ip_network
 
 import xlrd
-from django.core.exceptions import PermissionDenied, ValidationError
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils.translation import gettext_lazy as _
 from openwisp_users.mixins import ShareableOrgMixin
@@ -136,11 +136,11 @@ def request_ip(self, options=None):
         ip_address.save()
         return ip_address
 
-    def _read_subnet_data(self, reader, *user):
+    def _read_subnet_data(self, reader):
         subnet_model = load_model('openwisp_ipam', 'Subnet')
         subnet_name = next(reader)[0].strip()
         subnet_value = next(reader)[0].strip()
-        subnet_org = self._get_or_create_org(next(reader)[0].strip(), *user)
+        subnet_org = self._get_or_create_org(next(reader)[0].strip())
         try:
             subnet = subnet_model.objects.get(
                 subnet=subnet_value, organization=subnet_org
@@ -178,7 +178,7 @@ def _read_ipaddress_data(self, reader, subnet):
         for ip in ipaddress_list:
             ip.save()
 
-    def import_csv(self, file, *user):
+    def _get_csv_reader(self, file):
         if file.name.endswith(('.xls', '.xlsx')):
             book = xlrd.open_workbook(file_contents=file.read())
             sheet = book.sheet_by_index(0)
@@ -188,17 +188,15 @@ def import_csv(self, file, *user):
             reader = iter(row)
         else:
             reader = csv.reader(StringIO(file.read().decode('utf-8')), delimiter=',')
-        subnet = self._read_subnet_data(reader, *user)
+        return reader
+
+    def import_csv(self, file):
+        reader = self._get_csv_reader(file)
+        subnet = self._read_subnet_data(reader)
         next(reader)
         next(reader)
         self._read_ipaddress_data(reader, subnet)
 
-    def validate_permission(self, org, user):
-        if not (user.is_superuser or user.is_manager(org)):
-            raise PermissionDenied(
-                "User does not have access to the specified organization"
-            )
-
     def export_csv(self, subnet_id, writer):
         ipaddress_model = load_model('openwisp_ipam', 'IpAddress')
         subnet = load_model('openwisp_ipam', 'Subnet').objects.get(pk=subnet_id)
@@ -216,11 +214,9 @@ def export_csv(self, subnet_id, writer):
                 row.append(str(getattr(obj, field.name)))
             writer.writerow(row)
 
-    def _get_or_create_org(self, org_name, *user):
+    def _get_or_create_org(self, org_name):
         try:
             instance = Organization.objects.get(name=org_name)
-            if user:
-                self.validate_permission(instance, *user)
         except ValidationError as e:
             raise CsvImportException(str(e))
         except Organization.DoesNotExist:

openwisp_ipam/tests/test_multitenant.py

@@ -225,7 +225,6 @@ def test_request_ip(self):
     def test_import_subnet(self):
         csv_data = """Monachers - Matera,
         10.27.1.0/24,
-        Monachers,
         org_a,
         ip address,description
         10.27.1.1,Monachers