chrono/apps/export_import/api_views.py

@@ -29,6 +29,7 @@ from rest_framework.generics import GenericAPIView
 from rest_framework.response import Response
 
 from chrono.agendas.models import Agenda, Category, EventsType, Resource, UnavailabilityCalendar
+from chrono.api.utils import APIError
 from chrono.apps.export_import.models import Application, ApplicationElement
 from chrono.manager.utils import import_site
 
@@ -51,6 +52,13 @@ compare_urls = {
 }
 
 
+def get_klass_from_component_type(component_type):
+    try:
+        return klasses[component_type]
+    except KeyError:
+        raise Http404
+
+
 class Index(GenericAPIView):
     permission_classes = (permissions.IsAdminUser,)
 
@@ -149,7 +157,7 @@ class ListComponents(GenericAPIView):
     permission_classes = (permissions.IsAdminUser,)
 
     def get(self, request, *args, **kwargs):
-        klass = klasses[kwargs['component_type']]
+        klass = get_klass_from_component_type(kwargs['component_type'])
         order_by = 'slug'
         if klass == Group:
             order_by = 'name'
@@ -164,7 +172,7 @@ class ExportComponent(GenericAPIView):
     permission_classes = (permissions.IsAdminUser,)
 
     def get(self, request, slug, *args, **kwargs):
-        klass = klasses[kwargs['component_type']]
+        klass = get_klass_from_component_type(kwargs['component_type'])
         serialisation = get_object_or_404(klass, slug=slug).export_json()
         return Response({'data': serialisation})
 
@@ -176,7 +184,7 @@ class ComponentDependencies(GenericAPIView):
     permission_classes = (permissions.IsAdminUser,)
 
     def get(self, request, slug, *args, **kwargs):
-        klass = klasses[kwargs['component_type']]
+        klass = get_klass_from_component_type(kwargs['component_type'])
         component = get_object_or_404(klass, slug=slug)
 
         def dependency_dict(element):
@@ -190,7 +198,7 @@ component_dependencies = ComponentDependencies.as_view()
 
 
 def component_redirect(request, component_type, slug):
-    klass = klasses[component_type]
+    klass = get_klass_from_component_type(component_type)
     component = get_object_or_404(klass, slug=slug)
 
     if component_type not in klasses or component_type == 'roles':
@@ -231,96 +239,102 @@ class BundleCheck(GenericAPIView):
 
     def put(self, request, *args, **kwargs):
         tar_io = io.BytesIO(request.read())
-        with tarfile.open(fileobj=tar_io) as tar:
-            manifest = json.loads(tar.extractfile('manifest.json').read().decode())
-            application_slug = manifest.get('slug')
-            application_version = manifest.get('version_number')
-            if not application_slug or not application_version:
-                return Response({'data': {}})
-
-            differences = []
-            unknown_elements = []
-            no_history_elements = []
-            legacy_elements = []
-            content_types = ContentType.objects.get_for_models(
-                *[v for k, v in klasses.items() if k != 'roles']
-            )
-            for element in manifest.get('elements'):
-                component_type = element['type']
-                if component_type not in klasses or component_type == 'roles':
-                    continue
-                klass = klasses[component_type]
-                component_type = klasses_translation.get(component_type, component_type)
+        try:
+            with tarfile.open(fileobj=tar_io) as tar:
                 try:
-                    component = klass.objects.get(slug=element['slug'])
-                except klass.DoesNotExist:
-                    unknown_elements.append(
-                        {
-                            'type': component_type,
-                            'slug': element['slug'],
-                        }
-                    )
-                    continue
-                elements_qs = ApplicationElement.objects.filter(
-                    application__slug=application_slug,
-                    content_type=content_types[klass],
-                    object_id=component.pk,
+                    manifest = json.loads(tar.extractfile('manifest.json').read().decode())
+                except KeyError:
+                    raise APIError(_('Invalid tar file, missing manifest'))
+                application_slug = manifest.get('slug')
+                application_version = manifest.get('version_number')
+                if not application_slug or not application_version:
+                    return Response({'data': {}})
+
+                differences = []
+                unknown_elements = []
+                no_history_elements = []
+                legacy_elements = []
+                content_types = ContentType.objects.get_for_models(
+                    *[v for k, v in klasses.items() if k != 'roles']
                 )
-                if not elements_qs.exists():
-                    # object exists, but not linked to the application
-                    legacy_elements.append(
-                        {
-                            'type': component.application_component_type,
-                            'slug': str(component.slug),
-                            # information needed here, Relation objects may not exist yet in hobo
-                            'text': component.label,
-                            'url': reverse(
-                                'api-export-import-component-redirect',
-                                kwargs={
-                                    'slug': str(component.slug),
-                                    'component_type': component.application_component_type,
-                                },
-                            ),
-                        }
+                for element in manifest.get('elements'):
+                    component_type = element['type']
+                    if component_type not in klasses or component_type == 'roles':
+                        continue
+                    klass = klasses[component_type]
+                    component_type = klasses_translation.get(component_type, component_type)
+                    try:
+                        component = klass.objects.get(slug=element['slug'])
+                    except klass.DoesNotExist:
+                        unknown_elements.append(
+                            {
+                                'type': component_type,
+                                'slug': element['slug'],
+                            }
+                        )
+                        continue
+                    elements_qs = ApplicationElement.objects.filter(
+                        application__slug=application_slug,
+                        content_type=content_types[klass],
+                        object_id=component.pk,
                     )
-                    continue
-                snapshot_for_app = (
-                    klass.get_snapshot_model()
-                    .objects.filter(
-                        instance=component,
-                        application_slug=application_slug,
-                        application_version=application_version,
-                    )
-                    .order_by('timestamp')
-                    .last()
-                )
-                if not snapshot_for_app:
-                    # no snapshot for this bundle
-                    no_history_elements.append(
-                        {
-                            'type': element['type'],
-                            'slug': element['slug'],
-                        }
-                    )
-                    continue
-                last_snapshot = (
-                    klass.get_snapshot_model().objects.filter(instance=component).latest('timestamp')
-                )
-                if snapshot_for_app.pk != last_snapshot.pk:
-                    differences.append(
-                        {
-                            'type': element['type'],
-                            'slug': element['slug'],
-                            'url': '%s?version1=%s&version2=%s'
-                            % (
-                                request.build_absolute_uri(
-                                    reverse(compare_urls[component_type], args=[component.pk])
+                    if not elements_qs.exists():
+                        # object exists, but not linked to the application
+                        legacy_elements.append(
+                            {
+                                'type': component.application_component_type,
+                                'slug': str(component.slug),
+                                # information needed here, Relation objects may not exist yet in hobo
+                                'text': component.label,
+                                'url': reverse(
+                                    'api-export-import-component-redirect',
+                                    kwargs={
+                                        'slug': str(component.slug),
+                                        'component_type': component.application_component_type,
+                                    },
                                 ),
-                                snapshot_for_app.pk,
-                                last_snapshot.pk,
-                            ),
-                        }
+                            }
+                        )
+                        continue
+                    snapshot_for_app = (
+                        klass.get_snapshot_model()
+                        .objects.filter(
+                            instance=component,
+                            application_slug=application_slug,
+                            application_version=application_version,
+                        )
+                        .order_by('timestamp')
+                        .last()
                     )
+                    if not snapshot_for_app:
+                        # no snapshot for this bundle
+                        no_history_elements.append(
+                            {
+                                'type': element['type'],
+                                'slug': element['slug'],
+                            }
+                        )
+                        continue
+                    last_snapshot = (
+                        klass.get_snapshot_model().objects.filter(instance=component).latest('timestamp')
+                    )
+                    if snapshot_for_app.pk != last_snapshot.pk:
+                        differences.append(
+                            {
+                                'type': element['type'],
+                                'slug': element['slug'],
+                                'url': '%s?version1=%s&version2=%s'
+                                % (
+                                    request.build_absolute_uri(
+                                        reverse(compare_urls[component_type], args=[component.pk])
+                                    ),
+                                    snapshot_for_app.pk,
+                                    last_snapshot.pk,
+                                ),
+                            }
+                        )
+        except tarfile.TarError:
+            raise APIError(_('Invalid tar file'))
 
         return Response(
             {
@@ -344,25 +358,40 @@ class BundleImport(GenericAPIView):
     def put(self, request, *args, **kwargs):
         tar_io = io.BytesIO(request.read())
         components = {}
-        with tarfile.open(fileobj=tar_io) as tar:
-            manifest = json.loads(tar.extractfile('manifest.json').read().decode())
-            self.application = Application.update_or_create_from_manifest(
-                manifest,
-                tar,
-                editable=not self.install,
-            )
-
-            for element in manifest.get('elements'):
-                component_type = element['type']
-                if component_type not in klasses or component_type == 'roles':
-                    continue
-                component_type = klasses_translation.get(component_type, component_type)
-                if component_type not in components:
-                    components[component_type] = []
-                component_content = (
-                    tar.extractfile('%s/%s' % (element['type'], element['slug'])).read().decode()
+        try:
+            with tarfile.open(fileobj=tar_io) as tar:
+                try:
+                    manifest = json.loads(tar.extractfile('manifest.json').read().decode())
+                except KeyError:
+                    raise APIError(_('Invalid tar file, missing manifest'))
+                self.application = Application.update_or_create_from_manifest(
+                    manifest,
+                    tar,
+                    editable=not self.install,
                 )
-                components[component_type].append(json.loads(component_content).get('data'))
+
+                for element in manifest.get('elements'):
+                    component_type = element['type']
+                    if component_type not in klasses or element['type'] == 'roles':
+                        continue
+                    component_type = klasses_translation.get(component_type, component_type)
+                    if component_type not in components:
+                        components[component_type] = []
+                    try:
+                        component_content = (
+                            tar.extractfile('%s/%s' % (element['type'], element['slug'])).read().decode()
+                        )
+                    except KeyError:
+                        raise APIError(
+                            _(
+                                'Invalid tar file, missing component %s/%s'
+                                % (element['type'], element['slug'])
+                            )
+                        )
+                    components[component_type].append(json.loads(component_content).get('data'))
+        except tarfile.TarError:
+            raise APIError(_('Invalid tar file'))
+
         # init cache of application elements, from manifest
         self.application_elements = set()
         # import agendas

chrono/apps/export_import/models.py

@@ -55,10 +55,10 @@ class Application(models.Model):
             slug=manifest.get('slug'), defaults={'editable': editable}
         )
         application.name = manifest.get('application')
-        application.description = manifest.get('description')
-        application.documentation_url = manifest.get('documentation_url')
+        application.description = manifest.get('description') or ''
+        application.documentation_url = manifest.get('documentation_url') or ''
         application.version_number = manifest.get('version_number') or 'unknown'
-        application.version_notes = manifest.get('version_notes')
+        application.version_notes = manifest.get('version_notes') or ''
         if not editable:
             application.editable = editable
         application.visible = manifest.get('visible', True)

tests/api/test_applification_api.py

@@ -166,6 +166,9 @@ def test_list(app, admin_user):
         'data': [{'id': group.pk, 'text': 'group1', 'type': 'roles', 'urls': {}, 'uuid': None}]
     }
 
+    # unknown component type
+    app.get('/api/export-import/unknown/', status=404)
+
 
 def test_export_agenda(app, admin_user):
     app.authorization = ('Basic', ('admin', 'admin'))
@@ -196,6 +199,9 @@ def test_export_minor_components(app, admin_user):
     # unknown component
     app.get('/api/export-import/agendas/foo/', status=404)
 
+    # unknown component type
+    app.get('/api/export-import/unknown/foo/', status=404)
+
 
 def test_agenda_dependencies_category(app, admin_user):
     app.authorization = ('Basic', ('admin', 'admin'))
@@ -364,6 +370,11 @@ def test_unknown_compoment_dependencies(app, admin_user):
     app.get('/api/export-import/agendas/foo/dependencies/', status=404)
 
 
+def test_unknown_compoment_type_dependencies(app, admin_user):
+    app.authorization = ('Basic', ('admin', 'admin'))
+    app.get('/api/export-import/unknown/foo/dependencies/', status=404)
+
+
 def test_redirect(app, user):
     app.authorization = ('Basic', ('john', 'doe'))
     agenda = Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
@@ -427,6 +438,9 @@ def test_redirect(app, user):
         == f'/manage/unavailability-calendar/{unavailability_calendar.pk}/history/compare/?version1=bar&version2=bar&application=foo'
     )
 
+    # unknown component type
+    app.get('/api/export-import/unknown/foo/redirect/', status=404)
+
 
 def create_bundle(app, admin_user, visible=True, version_number='42.0'):
     app.authorization = ('Basic', ('admin', 'admin'))
@@ -584,6 +598,38 @@ def test_bundle_import(app, admin_user):
             assert last_snapshot.application_slug == 'test'
             assert last_snapshot.application_version == '42.1'
 
+    # bad file format
+    resp = app.put('/api/export-import/bundle-import/', b'garbage')
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file'
+
+    # missing manifest
+    tar_io = io.BytesIO()
+    with tarfile.open(mode='w', fileobj=tar_io) as tar:
+        foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
+        tarinfo = tarfile.TarInfo('foo.json')
+        tarinfo.size = len(foo_fd.getvalue())
+        tar.addfile(tarinfo, fileobj=foo_fd)
+    resp = app.put('/api/export-import/bundle-import/', tar_io.getvalue())
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'
+
+    # missing component
+    tar_io = io.BytesIO()
+    with tarfile.open(mode='w', fileobj=tar_io) as tar:
+        manifest_json = {
+            'application': 'Test',
+            'slug': 'test',
+            'elements': [{'type': 'agendas', 'slug': 'foo', 'name': 'foo'}],
+        }
+        manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
+        tarinfo = tarfile.TarInfo('manifest.json')
+        tarinfo.size = len(manifest_fd.getvalue())
+        tar.addfile(tarinfo, fileobj=manifest_fd)
+    resp = app.put('/api/export-import/bundle-import/', tar_io.getvalue())
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file, missing component agendas/foo'
+
 
 def test_bundle_declare(app, admin_user):
     app.authorization = ('Basic', ('admin', 'admin'))
@@ -613,7 +659,7 @@ def test_bundle_declare(app, admin_user):
         content_type=ContentType.objects.get_for_model(Agenda),
         object_id=last_page.pk + 1,
     )
-    # and remove agendas to have unkown references in manifest
+    # and remove agendas to have unknown references in manifest
     Agenda.objects.all().delete()
 
     resp = app.put('/api/export-import/bundle-declare/', bundle)
@@ -622,6 +668,38 @@ def test_bundle_declare(app, admin_user):
     assert application.visible is True
     assert ApplicationElement.objects.count() == 4  # category, events_type, unavailability_calendar, resource
 
+    # bad file format
+    resp = app.put('/api/export-import/bundle-declare/', b'garbage')
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file'
+
+    # missing manifest
+    tar_io = io.BytesIO()
+    with tarfile.open(mode='w', fileobj=tar_io) as tar:
+        foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
+        tarinfo = tarfile.TarInfo('foo.json')
+        tarinfo.size = len(foo_fd.getvalue())
+        tar.addfile(tarinfo, fileobj=foo_fd)
+    resp = app.put('/api/export-import/bundle-declare/', tar_io.getvalue())
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'
+
+    # missing component
+    tar_io = io.BytesIO()
+    with tarfile.open(mode='w', fileobj=tar_io) as tar:
+        manifest_json = {
+            'application': 'Test',
+            'slug': 'test',
+            'elements': [{'type': 'agendas', 'slug': 'foo', 'name': 'foo'}],
+        }
+        manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
+        tarinfo = tarfile.TarInfo('manifest.json')
+        tarinfo.size = len(manifest_fd.getvalue())
+        tar.addfile(tarinfo, fileobj=manifest_fd)
+    resp = app.put('/api/export-import/bundle-declare/', tar_io.getvalue())
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file, missing component agendas/foo'
+
 
 def test_bundle_unlink(app, admin_user, bundle):
     app.authorization = ('Basic', ('admin', 'admin'))
@@ -944,3 +1022,19 @@ def test_bundle_check(app, admin_user):
             'legacy_elements': [],
         }
     }
+
+    # bad file format
+    resp = app.put('/api/export-import/bundle-check/', b'garbage')
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file'
+
+    # missing manifest
+    tar_io = io.BytesIO()
+    with tarfile.open(mode='w', fileobj=tar_io) as tar:
+        foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
+        tarinfo = tarfile.TarInfo('foo.json')
+        tarinfo.size = len(foo_fd.getvalue())
+        tar.addfile(tarinfo, fileobj=foo_fd)
+    resp = app.put('/api/export-import/bundle-check/', tar_io.getvalue())
+    assert resp.json['err']
+    assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'