Start implementing security tests

.github/workflows/ci.yml

@@ -35,7 +35,9 @@ jobs:
       - name: Run DB Migrations
         run:  docker-compose -f local.yml run --rm django python manage.py migrate --skip-checks
       - name: Run Django Tests
-        run:  docker-compose -f local.yml run django pytest
+        run:  docker-compose -f local.yml run --rm django pytest
+      - name: Run Django Tests
+        run:  docker-compose -f local.yml run --rm django coverage run manage.py test colander
       - name: Tear down the Stack
         run:  docker-compose -f local.yml down
 

.gitignore

@@ -333,3 +333,10 @@ colander/media/
 /colander/core/migrationszzz/
 
 trash/
+
+notes
+.coverage
+htmlcov
+coverage.lcov
+.mypy_cache
+.ruff_cache

.pylintrc

@@ -1,11 +1,13 @@
 [MASTER]
 load-plugins=pylint_django
 django-settings-module=config.settings.local
+ignore=migrations
+
 [FORMAT]
 max-line-length=120
 
 [MESSAGES CONTROL]
-disable=missing-docstring,invalid-name
+disable=missing-docstring,invalid-name,line-too-long,wrong-import-order,fixme
 
 [DESIGN]
 max-parents=13

colander/core/middlewares.py

@@ -1,22 +1,23 @@
 from django.conf import settings
+from django.http import HttpResponseForbidden
 
 from colander.core.forms import DocumentationForm
 from colander.core.models import Case
 
 
-class ActiveCaseMiddleware:
-    def __init__(self, get_response):
-        self.get_response = get_response
-        # One-time configuration and initialization.
-
-    def __call__(self, request):
-        response = self.get_response(request)
-        return response
-
-    # def process_view(self, request, view_func, view_args, view_kwargs):
-    #     print(view_func)
-    #     if not request.session.get('active_case'):
-    #         return redirect('collect_case_create_view')
+# class ActiveCaseMiddleware:
+#     def __init__(self, get_response):
+#         self.get_response = get_response
+#         # One-time configuration and initialization.
+#
+#     def __call__(self, request):
+#         response = self.get_response(request)
+#         return response
+#
+#     # def process_view(self, request, view_func, view_args, view_kwargs):
+#     #     print(view_func)
+#     #     if not request.session.get('active_case'):
+#     #         return redirect('collect_case_create_view')
 
 
 class ContextualCaseMiddleware:
@@ -29,22 +30,18 @@ def __call__(self, request):
 
     def process_view(self, request, view_func, view_args, view_kwargs):
         if request.user is None or not request.user.is_authenticated:
-            return
+            return HttpResponseForbidden()
 
         workspace_case_id = view_kwargs.pop('case_id', None)
         if workspace_case_id is None:
             return
 
         case = Case.objects.get(pk=workspace_case_id)
-        if case is None:
-            # print('process_view', 'contextual_case not found', workspace_case_id)
-            pass
+        if case and case.can_contribute(request.user):
+            request.contextual_case = case  # even if case is None
         else:
-            if case.can_contribute(request.user):
-                request.contextual_case = case  # even if case is None
-            else:
-                #print('process_view', 'user', request.user, 'cannot contribute to case', case)
-                pass
+            return HttpResponseForbidden()
+
 
 
 def contextual_case(request):
@@ -55,7 +52,10 @@ def contextual_case(request):
         request.user_cases = user_cases
         if hasattr(request, 'contextual_case'):
             ctx_case = request.contextual_case
-            request.documentation_form = DocumentationForm(initial={'documentation': ctx_case.documentation})
+            if ctx_case and ctx_case.can_contribute(request.user):
+                request.documentation_form = DocumentationForm(initial={'documentation': ctx_case.documentation})
+            else:
+                ctx_case = None
 
     return {
         'contextual_case': ctx_case,
@@ -64,22 +64,22 @@ def contextual_case(request):
     }
 
 
-def active_case(request):
-    active_case = None
-    user_cases = []
-    if request.user and request.user.is_authenticated:
-        user_cases = Case.get_user_cases(request.user)
-        request.user_cases = user_cases
-        if 'active_case' in request.session:
-            try:
-                active_case = Case.objects.get(id=request.session['active_case'])
-                request.active_case = active_case
-                request.documentation_form = DocumentationForm(initial={'documentation': active_case.documentation})
-            except Exception:
-                pass
-
-    return {
-        'active_case': active_case,
-        'user_cases': user_cases,
-        'cyberchef_base_url': settings.CYBERCHEF_BASE_URL,
-    }
+# def active_case(request):
+#     active_case = None
+#     user_cases = []
+#     if request.user and request.user.is_authenticated:
+#         user_cases = Case.get_user_cases(request.user)
+#         request.user_cases = user_cases
+#         if 'active_case' in request.session:
+#             try:
+#                 active_case = Case.objects.get(id=request.session['active_case'])
+#                 request.active_case = active_case
+#                 request.documentation_form = DocumentationForm(initial={'documentation': active_case.documentation})
+#             except Exception:
+#                 pass
+#
+#     return {
+#         'active_case': active_case,
+#         'user_cases': user_cases,
+#         'cyberchef_base_url': settings.CYBERCHEF_BASE_URL,
+#     }

colander/core/tests/security/test_collect.py

@@ -0,0 +1,70 @@
+from django.test import TestCase, Client
+from django.urls import reverse
+
+from colander.core.models import Case, ColanderTeam, ActorType, Actor
+from colander.users.models import User
+
+
+class TestCaseTeam(TestCase):
+    password = '8F7JbzWGES8hH4zWM6R1MPPCI5'
+
+    @classmethod
+    def setUpTestData(cls):
+        # User 1
+        cls.user_1 = User.objects.create_user(username='u1', password=cls.password)
+        cls.case_1 = Case.objects.create(name='case 1', owner=cls.user_1, description='xx')
+        cls.team_1 = ColanderTeam.objects.create(name='t1', owner=cls.user_1)
+        # User 2
+        cls.user_2 = User.objects.create_user(username='u2', password=cls.password)
+        cls.case_2 = Case.objects.create(name='case 2', owner=cls.user_2, description='xx')
+        cls.team_2 = ColanderTeam.objects.create(name='t2', owner=cls.user_2)
+
+    def test_collect_actors(self):
+        c1 = Client()
+        login1 = c1.login(username=self.user_1.username, password=self.password)
+        c2 = Client()
+        login2 = c2.login(username=self.user_2.username, password=self.password)
+        actor_type = ActorType.objects.create(short_name='APT', name='apt')
+
+        # User can't add an actor to a case they can't contribute to
+        response = c1.post(reverse('collect_actor_create_view',
+                                   kwargs={'case_id': str(self.case_2.id)}),
+                           {
+                               'name': 'foo',
+                               'type': str(actor_type.id)
+                           })
+        self.assertEqual(response.status_code, 403, "User can't add an actor to a case they can't contribute to")
+
+        # User can add an actor to a case they own
+        response = c1.post(reverse('collect_actor_create_view',
+                                   kwargs={'case_id': str(self.case_1.id)}),
+                           {
+                               'name': 'foo',
+                               'type': str(actor_type.id),
+                               'tlp': 'WHITE',
+                               'pap': 'WHITE',
+                               'save_actor': '',
+                           })
+        # print(response.content)
+        self.assertEqual(response.status_code, 302, "User can add an actor to a case they own")
+        actor_1 = Actor.objects.get(name='foo')
+        self.assertIsNotNone(actor_1)
+
+        # User can't edit an actor to a case they can't contribute to
+        response = c2.post(reverse('collect_actor_update_view',
+                                   kwargs={'case_id': str(self.case_1.id), 'pk': str(actor_1.id)}),
+                           {
+                               'name': 'bar',
+                               'type': str(actor_type.id)
+                           })
+        self.assertEqual(response.status_code, 403, "User can't edit an actor to a case they can't contribute to")
+        actor_1.refresh_from_db()
+        self.assertEqual(actor_1.name, 'foo', "User can't edit an actor to a case they can't contribute to")
+
+        # User can't delete an actor to a case they can't contribute to
+        response = c2.get(reverse('collect_actor_delete_view',
+                                   kwargs={'case_id': str(self.case_1.id), 'pk': str(actor_1.id)}))
+        self.assertEqual(response.status_code, 403, "User can't delete an actor to a case they can't contribute to")
+        actor_1.refresh_from_db()
+        self.assertEqual(actor_1.name, 'foo', "User can't delete an actor to a case they can't contribute to")
+

colander/core/tests/security/test_teams.py

@@ -0,0 +1,126 @@
+from django.test import TestCase, Client
+from django.urls import reverse
+
+from colander.core.models import Case, ColanderTeam
+from colander.users.models import User
+
+
+class TestCaseTeam(TestCase):
+    password = '8F7JbzWGES8hH4zWM6R1MPPCI5'
+
+    @classmethod
+    def setUpTestData(cls):
+        # User 1
+        cls.user_1 = User.objects.create_user(username='u1', password=cls.password)
+        cls.case_1 = Case.objects.create(name='case 1', owner=cls.user_1, description='xx')
+        cls.team_1 = ColanderTeam.objects.create(name='t1', owner=cls.user_1)
+        # User 2
+        cls.user_2 = User.objects.create_user(username='u2', password=cls.password)
+        cls.case_2 = Case.objects.create(name='case 2', owner=cls.user_2, description='xx')
+        cls.team_2 = ColanderTeam.objects.create(name='t2', owner=cls.user_2)
+
+    def test_case_management(self):
+        c1 = Client()
+        login1 = c1.login(username=self.user_1.username, password=self.password)
+        self.assertTrue(login1, "User can login")
+
+        # User can't add a case they don't own to their team
+        response = c1.post(reverse('case_update_view', kwargs={'pk': str(self.case_2.id)}),
+                           {
+                               'name': self.case_2.name,
+                               'description': 'yy',
+                               'owner': self.user_2,
+                               'tlp': self.case_2.tlp,
+                               'pap': self.case_2.pap,
+                               'teams': [str(self.team_1.id)]})
+        self.assertEqual(response.status_code, 403, "User can't add a case they don't own to their team")
+
+        # User can't add a case they don't own to a team they don't own
+        response = c1.post(reverse('case_update_view', kwargs={'pk': str(self.case_2.id)}),
+                           {
+                               'name': self.case_2.name,
+                               'description': 'zz',
+                               'owner': self.user_2,
+                               'tlp': self.case_2.tlp,
+                               'pap': self.case_2.pap,
+                               'teams': [str(self.team_2.id)]})
+        self.assertEqual(response.status_code, 403, "User can't add a case they don't own to a team they don't own")
+
+        # User can't add their case to a team they don't own
+        response = c1.post(reverse('case_update_view', kwargs={'pk': str(self.case_1.id)}),
+                           {
+                               'name': self.case_1.name,
+                               'description': 'yy',
+                               'owner': self.user_1,
+                               'tlp': self.case_1.tlp,
+                               'pap': self.case_1.pap,
+                               'teams': [str(self.team_2.id)]})
+        self.assertEqual(response.status_code, 200, "User can't add their case to a team they don't own")
+        case = Case.objects.get(id=self.case_1.id)
+        self.assertFalse(self.team_2 in case.teams.all(), "User can't add their case to a team they don't own")
+
+        # User can add their case to a team they own
+        response = c1.post(reverse('case_update_view', kwargs={'pk': str(self.case_1.id)}),
+                           data={
+                               'name': self.case_1.name,
+                               'description': 'zz',
+                               'owner': self.user_1,
+                               'tlp': self.case_1.tlp,
+                               'pap': self.case_1.pap,
+                               'teams': str(self.team_1.id)})
+        self.assertEqual(response.status_code, 302, "User can add their case to a team they own")
+        case = Case.objects.get(id=self.case_1.id)
+        self.assertTrue(self.team_1 in case.teams.all(), "User can add their case to a team they own")
+
+    def test_team_edition(self):
+        c1 = Client()
+        login1 = c1.login(username=self.user_1.username, password=self.password)
+        self.assertTrue(login1, "User can login")
+
+        # User can't add themselves to a team they don't own
+        response = c1.post(reverse('collaborate_team_update_view', kwargs={'pk': str(self.team_2.id)}),
+                           {'contributors': str(self.user_1.id)})
+        self.assertEqual(response.status_code, 403, "User can't add themselves to a team they don't own")
+
+        # User can't add a contributor to a team they don't own
+        response = c1.post(reverse('collaborate_team_update_view', kwargs={'pk': str(self.team_2.id)}),
+                           {'contributors': str(self.user_2.id)})
+        self.assertEqual(response.status_code, 403, "User can't add a contributor to a team they don't own")
+
+        # User can't add themselves to a team they don't own
+        response = c1.post(reverse('collaborate_team_add_remove_contributor', kwargs={'pk': str(self.team_2.id)}),
+                           {'add_contributor': True, 'contributor_id': str(self.user_1.contributor_id)})
+        self.assertEqual(response.status_code, 403)
+        self.team_2.refresh_from_db()
+        self.assertFalse(self.user_1 in self.team_2.contributors.all(), "User can't add themselves to a team they don't own")
+
+        # User can't add a contributor to a team they don't own
+        response = c1.post(reverse('collaborate_team_add_remove_contributor', kwargs={'pk': str(self.team_2.id)}),
+                           {'add_contributor': True, 'contributor_id': str(self.user_2.contributor_id)})
+        self.assertEqual(response.status_code, 403)
+        self.team_2.refresh_from_db()
+        self.assertFalse(self.user_2 in self.team_2.contributors.all(), "User can't add a contributor to a team they don't own")
+
+        # User can't add themselves as contributor to their team
+        response = c1.post(reverse('collaborate_team_add_remove_contributor', kwargs={'pk': str(self.team_1.id)}),
+                           {'add_contributor': True, 'contributor_id': str(self.user_1.contributor_id)})
+        self.assertEqual(response.status_code, 302)
+        self.team_1.refresh_from_db()
+        self.assertFalse(self.user_1 in self.team_1.contributors.all(), "User can't add themselves as contributor to their team")
+
+        # User can add a contributor to their team
+        response = c1.post(reverse('collaborate_team_add_remove_contributor', kwargs={'pk': str(self.team_1.id)}),
+                           {'add_contributor': True, 'contributor_id': str(self.user_2.contributor_id)})
+        self.assertEqual(response.status_code, 302)
+        self.team_1.refresh_from_db()
+        self.assertTrue(self.user_2 in self.team_1.contributors.all(), "User can add a contributor to their team")
+
+    def test_team_deletion(self):
+        c1 = Client()
+        login1 = c1.login(username=self.user_1.username, password=self.password)
+        self.assertTrue(login1, "User can login")
+
+        # User can't delete a team they don't own
+        response = c1.get(reverse('collaborate_team_delete_view', kwargs={'pk': str(self.team_2.id)}))
+        self.assertEqual(response.status_code, 403)
+        self.assertTrue(ColanderTeam.objects.filter(id=str(self.team_2.id)).exists())

colander/core/views/collaborate_views.py

@@ -1,5 +1,7 @@
 from django.contrib.auth.decorators import login_required
 from django.contrib.auth.mixins import LoginRequiredMixin
+from django.core.exceptions import PermissionDenied
+from django.http import HttpResponseForbidden
 from django.shortcuts import redirect
 from django.urls import reverse_lazy
 from django.contrib import messages
@@ -64,6 +66,8 @@ def get_context_data(self, **kwargs):
 @login_required
 def add_remove_team_contributor(request, pk):
     team = ColanderTeam.objects.get(id=pk)
+    if team.owner != request.user:  # Todo: support team administrators
+        return HttpResponseForbidden()
     if request.method == 'POST' and team.owner == request.user:
         form = AddRemoveTeamContributorForm(request.POST)
         if form.is_valid():
@@ -92,6 +96,9 @@ def add_remove_team_contributor(request, pk):
 
 @login_required
 def delete_team_view(request, pk):
-    obj = ColanderTeam.objects.get(id=pk)
-    obj.delete()
+    team = ColanderTeam.objects.get(id=pk)
+    if team.owner == request.user:
+        team.delete()
+    else:
+        return HttpResponseForbidden()
     return redirect("collaborate_team_create_view")

colander/core/views/views.py

@@ -59,11 +59,12 @@ class CaseContextMixin(AccessMixin):
     active_case = None
 
     def dispatch(self, request, *args, **kwargs):
-        #print("CaseContextMixin", "dispatch", request, hasattr(request, 'contextual_case') )
-        #self.active_case = get_active_case(request, kwargs['case_id'])
-        if hasattr(request, 'contextual_case'):
+        # print("CaseContextMixin", "dispatch", request, hasattr(request, 'contextual_case') )
+        if hasattr(request, 'contextual_case') and request.contextual_case.can_contribute(request.user):
             self.active_case = request.contextual_case
-        return super().dispatch(request, *args, **kwargs)
+            return super().dispatch(request, *args, **kwargs)
+        else:
+            return self.handle_no_permission()
 
     def get_success_url(self):
         #print("get_success_url", self.active_case)