Integrate Docker API functionality by adding endpoints for listing running containers and retrieving code paths. Update Docker Compose configuration to allow socket access. Include new dependencies in pyproject.toml for Docker integration. Remove obsolete migration files to clean up the project structure.

Author: fr4iser90

backend/core/docker_service.py

@@ -0,0 +1,87 @@
+import docker
+from docker.errors import DockerException, NotFound
+
+def list_running_containers():
+    """
+    Connects to the Docker daemon and lists running containers.
+    Returns a list of dictionaries, each containing container info,
+    or an error message string.
+    """
+    try:
+        client = docker.from_env()
+        containers = client.containers.list()
+        container_list = []
+        for container in containers:
+            container_list.append({
+                "id": container.short_id,
+                "name": container.name,
+                "image": container.attrs['Config']['Image'],
+                "status": container.status
+            })
+        if not container_list:
+            return "No running containers found."
+        return container_list
+    except DockerException as e:
+        return f"Docker API error: {str(e)}"
+    except Exception as e:
+        return f"An unexpected error occurred: {str(e)}"
+
+def get_container_code_paths(container_id: str):
+    """
+    Retrieves the host paths for volume mounts of a given container.
+
+    Args:
+        container_id: The ID or name of the Docker container.
+
+    Returns:
+        A list of host source paths for the container's mounts,
+        or an error message string if an issue occurs.
+    """
+    try:
+        client = docker.from_env()
+        container = client.containers.get(container_id)
+        mounts = container.attrs.get('Mounts', [])
+        host_paths = []
+        for mount in mounts:
+            if mount.get('Type') == 'volume' or mount.get('Type') == 'bind': # Consider both volumes and binds
+                # For named volumes, 'Source' might be the volume name, not a host path.
+                # For binds, 'Source' is the host path.
+                # We are primarily interested in host paths that could contain code.
+                if mount.get('Source') and mount.get('Source').startswith('/'): # Heuristic: host paths are absolute
+                    host_paths.append(mount['Source'])
+        
+        if not host_paths and mounts:
+             return f"Container '{container_id}' has mounts, but no direct host source paths found: {mounts}"
+        elif not host_paths:
+            return f"No volume mounts found for container '{container_id}' that appear to be host paths."
+            
+        return host_paths
+    except NotFound:
+        return f"Container '{container_id}' not found."
+    except DockerException as e:
+        return f"Docker API error while inspecting container '{container_id}': {str(e)}"
+    except Exception as e:
+        return f"An unexpected error occurred while inspecting container '{container_id}': {str(e)}"
+
+if __name__ == '__main__':
+    print("Attempting to list running containers...")
+    running_containers = list_running_containers()
+    if isinstance(running_containers, str): # Error message
+        print(running_containers)
+    elif running_containers:
+        print("Running containers:")
+        for cont in running_containers:
+            print(f"  ID: {cont['id']}, Name: {cont['name']}, Image: {cont['image']}, Status: {cont['status']}")
+            print(f"    Attempting to get code paths for {cont['id']} ({cont['name']})...")
+            paths = get_container_code_paths(cont['id'])
+            if isinstance(paths, str): # Error message
+                print(f"    Error/Info: {paths}")
+            elif paths:
+                print("    Found potential host paths:")
+                for path in paths:
+                    print(f"      - {path}")
+            else: # Empty list but no error string
+                print(f"    No specific host paths identified for {cont['id']}.")
+
+    else: # Empty list from list_running_containers, but not an error string.
+        print("No running containers were found to inspect.") 

backend/core/migrations/0007_project_default_tool_settings_json_and_more.py

@@ -0,0 +1 @@
+# Diese Datei macht das 'tests'-Verzeichnis zu einem Python-Paket 

backend/core/migrations/0008_project_project_main_targets_json_and_more.py

@@ -0,0 +1,146 @@
+from django.urls import reverse
+from django.contrib.auth import get_user_model
+from rest_framework import status
+from rest_framework.test import APITestCase
+from unittest.mock import patch
+
+User = get_user_model()
+
+class DockerAPITests(APITestCase):
+    def setUp(self):
+        # Create an admin user
+        self.admin_user = User.objects.create_superuser(
+            username='admin_test',
+            email='admin_test@example.com',
+            password='admin_password123'
+        )
+        # Create a regular user
+        self.regular_user = User.objects.create_user(
+            username='user_test',
+            email='user_test@example.com',
+            password='user_password123'
+        )
+        
+        self.list_containers_url = reverse('docker-list-containers')
+        # URL for GetDockerContainerPathsView (needs a placeholder container_id)
+        self.container_paths_url_template = reverse('docker-container-paths', kwargs={'container_id': 'PLACEHOLDER_ID'})
+
+    def test_list_docker_containers_permissions_and_basic_success(self):
+        """
+        Test permissions for the list_docker_containers endpoint and a basic success scenario.
+        """
+        # 1. Test unauthenticated access
+        response = self.client.get(self.list_containers_url)
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+        # 2. Test access with regular user (should be forbidden)
+        self.client.force_authenticate(user=self.regular_user)
+        response = self.client.get(self.list_containers_url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.client.force_authenticate(user=None) # Clear authentication
+
+        # 3. Test access with admin user (should be successful)
+        # We mock the actual service call to avoid real Docker interaction
+        mock_container_data = [
+            {"id": "abc123xyz", "name": "test_container_1", "image": "test_image:latest", "status": "running"},
+            {"id": "def456uvw", "name": "test_container_2", "image": "another_image:1.0", "status": "running"}
+        ]
+        
+        with patch('core.views.list_running_containers') as mock_list_containers:
+            mock_list_containers.return_value = mock_container_data
+            
+            self.client.force_authenticate(user=self.admin_user)
+            response = self.client.get(self.list_containers_url)
+            
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            self.assertEqual(len(response.data), 2)
+            self.assertEqual(response.data[0]['name'], 'test_container_1')
+            mock_list_containers.assert_called_once() # Ensure our service function was called
+
+    def test_list_docker_containers_service_error(self):
+        """
+        Test the list_docker_containers endpoint when the service returns an error.
+        """
+        error_message = "Docker API error: Connection refused"
+        with patch('core.views.list_running_containers') as mock_list_containers:
+            mock_list_containers.return_value = error_message
+            
+            self.client.force_authenticate(user=self.admin_user)
+            response = self.client.get(self.list_containers_url)
+            
+            self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+            self.assertIn('error', response.data)
+            self.assertEqual(response.data['error'], error_message)
+            mock_list_containers.assert_called_once()
+
+    def test_get_container_paths_permissions_and_success(self):
+        """
+        Test permissions and basic success scenario for GetDockerContainerPathsView.
+        """
+        test_container_id = "test_container_123"
+        specific_container_paths_url = self.container_paths_url_template.replace('PLACEHOLDER_ID', test_container_id)
+
+        # 1. Test unauthenticated access
+        response = self.client.get(specific_container_paths_url)
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+        # 2. Test access with regular user (should be forbidden)
+        self.client.force_authenticate(user=self.regular_user)
+        response = self.client.get(specific_container_paths_url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.client.force_authenticate(user=None)  # Clear authentication
+
+        # 3. Test access with admin user (should be successful)
+        mock_path_data = ['/mnt/code/project_a', '/var/www/html']
+        with patch('core.views.get_container_code_paths') as mock_get_paths:
+            mock_get_paths.return_value = mock_path_data
+            
+            self.client.force_authenticate(user=self.admin_user)
+            response = self.client.get(specific_container_paths_url)
+            
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            self.assertEqual(len(response.data), 2)
+            self.assertEqual(response.data[0], '/mnt/code/project_a')
+            # Ensure the service function was called with the correct container_id
+            mock_get_paths.assert_called_once_with(test_container_id)
+
+    def test_get_container_paths_not_found(self):
+        """
+        Test GetDockerContainerPathsView when the container is not found.
+        """
+        test_container_id = "non_existent_container"
+        specific_container_paths_url = self.container_paths_url_template.replace('PLACEHOLDER_ID', test_container_id)
+        error_message = f"Container '{test_container_id}' not found."
+
+        with patch('core.views.get_container_code_paths') as mock_get_paths:
+            mock_get_paths.return_value = error_message
+            
+            self.client.force_authenticate(user=self.admin_user)
+            response = self.client.get(specific_container_paths_url)
+            
+            self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+            self.assertIn('error', response.data)
+            self.assertEqual(response.data['error'], error_message)
+            mock_get_paths.assert_called_once_with(test_container_id)
+
+    def test_get_container_paths_service_error(self):
+        """
+        Test GetDockerContainerPathsView when the service returns a generic error.
+        """
+        test_container_id = "another_container_id"
+        specific_container_paths_url = self.container_paths_url_template.replace('PLACEHOLDER_ID', test_container_id)
+        error_message = "Docker API error: Unexpected issue"
+
+        with patch('core.views.get_container_code_paths') as mock_get_paths:
+            mock_get_paths.return_value = error_message
+            
+            self.client.force_authenticate(user=self.admin_user)
+            response = self.client.get(specific_container_paths_url)
+            
+            self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+            self.assertIn('error', response.data)
+            self.assertEqual(response.data['error'], error_message)
+            mock_get_paths.assert_called_once_with(test_container_id)
+
+    # Additional test methods for GetDockerContainerPathsView will be added here
+    # and for error cases of ListDockerContainersView. 

backend/core/migrations/0009_scanjob_target_info_scanjob_tool_settings.py

@@ -4,7 +4,8 @@
     ProjectViewSet, ScanTargetViewSet, TargetGroupViewSet, 
     SecurityToolViewSet, ScanConfigurationViewSet, UserProfileViewSet,
     ApiKeyViewSet, ScanTriggerViewSet, ScanJobViewSet, 
-    ProjectMembershipViewSet, UserViewSet, CIScanTriggerViewSet
+    ProjectMembershipViewSet, UserViewSet, CIScanTriggerViewSet,
+    ListDockerContainersView, GetDockerContainerPathsView
 )
 
 router = DefaultRouter()
@@ -23,4 +24,6 @@
 
 urlpatterns = [
     path('', include(router.urls)),
+    path('docker/containers/', ListDockerContainersView.as_view(), name='docker-list-containers'),
+    path('docker/containers/<str:container_id>/paths/', GetDockerContainerPathsView.as_view(), name='docker-container-paths'),
 ] 

backend/core/tests/__init__.py

@@ -26,6 +26,10 @@
 from .tasks import execute_scan_job, simulate_bandit_scan # Ensure both are imported
 from django.contrib.auth import get_user_model
 from .authentication import ApiKeyAuthentication # Import the custom authentication class
+# Docker API Integration Imports
+from rest_framework.views import APIView
+from .docker_service import list_running_containers, get_container_code_paths # Assuming docker_service.py is in the same app directory
+# End Docker API Integration Imports
 User = get_user_model()
 
 # Create your views here.
@@ -584,3 +588,37 @@ def create(self, request, *args, **kwargs):
             response_serializer = ScanJobSerializer(scan_job)
             return Response(response_serializer.data, status=status.HTTP_201_CREATED)
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+# Docker API Views
+class ListDockerContainersView(APIView):
+    """
+    Lists all currently running Docker containers.
+    Requires admin privileges.
+    """
+    permission_classes = [IsAdminUser]
+
+    def get(self, request, *args, **kwargs):
+        containers_or_error = list_running_containers()
+        if isinstance(containers_or_error, str): # Error message returned
+            if "No running containers found" in containers_or_error:
+                 return Response({"message": containers_or_error}, status=status.HTTP_200_OK) # Or 404 if preferred
+            return Response({"error": containers_or_error}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+        return Response(containers_or_error, status=status.HTTP_200_OK)
+
+class GetDockerContainerPathsView(APIView):
+    """
+    Retrieves potential host code paths for a given Docker container.
+    Requires admin privileges.
+    """
+    permission_classes = [IsAdminUser]
+
+    def get(self, request, container_id, *args, **kwargs):
+        paths_or_error = get_container_code_paths(container_id)
+        if isinstance(paths_or_error, str): # Error message returned
+            if "not found" in paths_or_error.lower():
+                return Response({"error": paths_or_error}, status=status.HTTP_404_NOT_FOUND)
+            # For other errors from the service, consider them server-side issues or specific Docker issues
+            return Response({"error": paths_or_error}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+        if not paths_or_error: # Empty list, but valid response
+            return Response([], status=status.HTTP_200_OK)
+        return Response(paths_or_error, status=status.HTTP_200_OK)

backend/core/tests/test_docker_api.py

@@ -23,6 +23,7 @@ dj-rest-auth = ">=6.0.0,<7.0.0"
 django-allauth = ">=0.63.0,<0.64.0"
 requests = ">=2.30.0,<3.0.0"
 django-celery-beat = ">=2.6.0,<3.0.0"
+docker = "^7.0.0"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

backend/core/urls.py

@@ -47,6 +47,7 @@ services:
       - ./backend:/app
       - backend_static_collected:/app/staticfiles_collected
       - backend_media_data:/app/media
+      - /var/run/docker.sock:/var/run/docker.sock
     ports:
       - "8000:8000"
     env_file: