Add Rust product security roadmap task updates

Author: ewilhelm1979-netizen

apps/product_security/forms.py

@@ -0,0 +1,12 @@
+from django import forms
+
+from .models import ProductSecurityRoadmapTask
+
+
+class ProductSecurityRoadmapTaskUpdateForm(forms.ModelForm):
+    class Meta:
+        model = ProductSecurityRoadmapTask
+        fields = ['status', 'priority', 'owner_role', 'due_in_days', 'dependency_text']
+        widgets = {
+            'dependency_text': forms.Textarea(attrs={'rows': 3}),
+        }

apps/product_security/services_rust.py

@@ -1,5 +1,6 @@
 import json
 from dataclasses import dataclass
+from urllib.error import HTTPError
 from urllib.request import Request, urlopen
 
 from django.conf import settings
@@ -386,6 +387,13 @@ class ProductSecurityRoadmapBridgeResult:
     snapshot: ProductSecuritySnapshotListItem | None
 
 
+@dataclass(frozen=True)
+class ProductSecurityRoadmapTaskUpdateBridgeResult:
+    product_id: int
+    roadmap_id: int
+    task: ProductSecurityRoadmapTaskItem
+
+
 class ProductSecurityRustClient:
     @staticmethod
     def _base_url() -> str:
@@ -496,6 +504,49 @@ def fetch_roadmap(request, tenant, product_id: int, timeout: int = 8) -> Product
             snapshot=ProductSecurityRustClient._snapshot_from_payload(snapshot_payload, tenant) if isinstance(snapshot_payload, dict) else None,
         )
 
+    @staticmethod
+    def update_roadmap_task(
+        request,
+        tenant,
+        task_id: int,
+        data: dict,
+        timeout: int = 8,
+    ) -> ProductSecurityRoadmapTaskUpdateBridgeResult | None:
+        base = ProductSecurityRustClient._base_url()
+        if not base:
+            raise RuntimeError('RUST_BACKEND_URL ist nicht gesetzt.')
+
+        rust_request = ProductSecurityRustClient._authenticated_json_request(
+            request,
+            tenant,
+            f'{base}/api/v1/product-security/roadmap-tasks/{int(task_id)}',
+            {
+                'status': data.get('status'),
+                'priority': data.get('priority') or '',
+                'owner_role': data.get('owner_role') or '',
+                'due_in_days': data.get('due_in_days'),
+                'dependency_text': data.get('dependency_text') or '',
+            },
+            method='PATCH',
+        )
+        try:
+            with urlopen(rust_request, timeout=timeout) as response:
+                payload = json.loads(response.read().decode('utf-8'))
+        except HTTPError as exc:
+            if exc.code == 404:
+                return None
+            raise
+
+        task_payload = payload.get('task') or {}
+        if not isinstance(task_payload, dict):
+            raise RuntimeError('Rust-Product-Security-Roadmaptask-Update hat kein task-Objekt geliefert.')
+
+        return ProductSecurityRoadmapTaskUpdateBridgeResult(
+            product_id=ProductSecurityRustClient._int_field(payload, 'product_id'),
+            roadmap_id=ProductSecurityRustClient._int_field(payload, 'roadmap_id'),
+            task=ProductSecurityRustClient._roadmap_task_from_payload(task_payload, tenant),
+        )
+
     @staticmethod
     def _product_from_payload(item: dict, tenant) -> ProductSecurityProductListItem:
         family_id = ProductSecurityRustClient._optional_int_field(item, 'family_id')
@@ -801,6 +852,14 @@ def _authenticated_request(request, tenant, url: str) -> Request:
             rust_request.add_header('X-ISCY-User-Email', user_email)
         return rust_request
 
+    @staticmethod
+    def _authenticated_json_request(request, tenant, url: str, payload: dict, method: str) -> Request:
+        rust_request = ProductSecurityRustClient._authenticated_request(request, tenant, url)
+        rust_request.data = json.dumps(payload).encode('utf-8')
+        rust_request.method = method
+        rust_request.add_header('Content-Type', 'application/json')
+        return rust_request
+
     @staticmethod
     def _int_field(payload: dict, key: str) -> int:
         return int(payload.get(key) or 0)
@@ -884,6 +943,27 @@ def fetch_roadmap(request, tenant, product_id: int):
                 raise RuntimeError('Rust product security backend ist aktiv, aber nicht erreichbar.') from exc
             return None
 
+    @staticmethod
+    def update_roadmap_task(request, tenant, task_id: int, data: dict):
+        if tenant is None:
+            return None
+
+        backend = str(getattr(settings, 'PRODUCT_SECURITY_BACKEND', 'rust_service') or '').strip().lower()
+        if backend != 'rust_service':
+            return None
+
+        if not ProductSecurityRustClient._base_url():
+            if ProductSecurityBridge._strict_rust_mode():
+                raise RuntimeError('Rust product security backend ist aktiv, aber RUST_BACKEND_URL ist nicht gesetzt.')
+            return None
+
+        try:
+            return ProductSecurityRustClient.update_roadmap_task(request, tenant, task_id, data)
+        except Exception as exc:
+            if ProductSecurityBridge._strict_rust_mode():
+                raise RuntimeError('Rust product security backend ist aktiv, aber nicht erreichbar.') from exc
+            return None
+
     @staticmethod
     def _strict_rust_mode() -> bool:
         return bool(getattr(settings, 'RUST_STRICT_MODE', False))

apps/product_security/tests.py

@@ -7,6 +7,7 @@
 
 from apps.organizations.models import Tenant
 from apps.product_security.models import Product, ProductFamily, ProductSecuritySnapshot
+from apps.product_security.services import ProductSecurityService
 
 
 User = get_user_model()
@@ -197,6 +198,103 @@ def test_product_roadmap_can_use_rust_bridge(self, mock_urlopen):
         self.assertContains(response, "Rust Product Roadmap")
         self.assertContains(response, "Rust Roadmap Task")
 
+    def test_product_roadmap_task_update_uses_local_backend(self):
+        roadmap = ProductSecurityService.generate_product_roadmap(self.product_a)
+        task = roadmap.tasks.first()
+        self.client.force_login(self.user_a)
+
+        response = self.client.post(
+            reverse("product_security:roadmap-task-edit", args=[task.pk]),
+            {
+                "status": "DONE",
+                "priority": "MEDIUM",
+                "owner_role": "Product Security Office",
+                "due_in_days": "9",
+                "dependency_text": "Reviewed locally",
+            },
+        )
+
+        self.assertRedirects(
+            response,
+            reverse("product_security:roadmap", args=[self.product_a.pk]),
+            fetch_redirect_response=False,
+        )
+        task.refresh_from_db()
+        self.assertEqual(task.status, "DONE")
+        self.assertEqual(task.priority, "MEDIUM")
+        self.assertEqual(task.owner_role, "Product Security Office")
+        self.assertEqual(task.due_in_days, 9)
+        self.assertEqual(task.dependency_text, "Reviewed locally")
+
+    @patch("apps.product_security.services_rust.urlopen")
+    def test_product_roadmap_task_update_can_use_rust_bridge(self, mock_urlopen):
+        roadmap = ProductSecurityService.generate_product_roadmap(self.product_a)
+        task = roadmap.tasks.first()
+        task_payload = self._rust_roadmap_task_payload()
+        task_payload.update({
+            "id": task.id,
+            "roadmap_id": roadmap.id,
+            "status": "DONE",
+            "status_label": "Erledigt",
+            "priority": "MEDIUM",
+            "owner_role": "Product Security Office",
+            "due_in_days": 9,
+            "dependency_text": "Reviewed in Rust",
+        })
+        self._mock_rust_response(mock_urlopen, {
+            "api_version": "v1",
+            "product_id": self.product_a.id,
+            "roadmap_id": roadmap.id,
+            "task": task_payload,
+        })
+        self.client.force_login(self.user_a)
+
+        with self.settings(PRODUCT_SECURITY_BACKEND="rust_service", RUST_BACKEND_URL="http://rust-backend:9000"):
+            response = self.client.post(
+                reverse("product_security:roadmap-task-edit", args=[task.pk]),
+                {
+                    "status": "DONE",
+                    "priority": "MEDIUM",
+                    "owner_role": "Product Security Office",
+                    "due_in_days": "9",
+                    "dependency_text": "Reviewed in Rust",
+                },
+            )
+
+        self.assertRedirects(
+            response,
+            reverse("product_security:roadmap", args=[self.product_a.pk]),
+            fetch_redirect_response=False,
+        )
+        rust_request = mock_urlopen.call_args.args[0]
+        body = json.loads(rust_request.data.decode("utf-8"))
+        self.assertEqual(
+            rust_request.full_url,
+            f"http://rust-backend:9000/api/v1/product-security/roadmap-tasks/{task.id}",
+        )
+        self.assertEqual(rust_request.get_method(), "PATCH")
+        self.assertEqual(body["status"], "DONE")
+        self.assertEqual(body["owner_role"], "Product Security Office")
+        self.assertEqual(body["due_in_days"], 9)
+
+    def test_product_roadmap_task_update_blocks_foreign_tenant_access(self):
+        roadmap = ProductSecurityService.generate_product_roadmap(self.product_b)
+        task = roadmap.tasks.first()
+        self.client.force_login(self.user_a)
+
+        response = self.client.post(
+            reverse("product_security:roadmap-task-edit", args=[task.pk]),
+            {
+                "status": "DONE",
+                "priority": "MEDIUM",
+                "owner_role": "Product Security Office",
+                "due_in_days": "9",
+                "dependency_text": "Nope",
+            },
+        )
+
+        self.assertEqual(response.status_code, 404)
+
     def _rust_payload(self):
         return {
             "api_version": "v1",

apps/product_security/urls.py

@@ -1,10 +1,20 @@
 from django.urls import path
-from .views import ProductDetailView, ProductListView, ProductRoadmapView
+from .views import (
+    ProductDetailView,
+    ProductListView,
+    ProductRoadmapView,
+    ProductSecurityRoadmapTaskUpdateView,
+)
 
 app_name = 'product_security'
 
 urlpatterns = [
     path('', ProductListView.as_view(), name='list'),
     path('<int:pk>/', ProductDetailView.as_view(), name='detail'),
     path('<int:pk>/roadmap/', ProductRoadmapView.as_view(), name='roadmap'),
+    path(
+        'roadmap/tasks/<int:pk>/edit/',
+        ProductSecurityRoadmapTaskUpdateView.as_view(),
+        name='roadmap-task-edit',
+    ),
 ]

apps/product_security/views.py

@@ -1,8 +1,11 @@
 from django.contrib.auth.mixins import LoginRequiredMixin
+from django.http import HttpResponseRedirect
 from django.shortcuts import get_object_or_404
-from django.views.generic import DetailView, ListView, TemplateView
+from django.urls import reverse
+from django.views.generic import DetailView, ListView, TemplateView, UpdateView
 
-from .models import Product, ProductSecuritySnapshot
+from .forms import ProductSecurityRoadmapTaskUpdateForm
+from .models import Product, ProductSecurityRoadmapTask, ProductSecuritySnapshot
 from .services_rust import ProductSecurityBridge
 from .services import ProductSecurityService
 
@@ -130,3 +133,35 @@ def get_context_data(self, **kwargs):
             'product_security_source': 'django',
         })
         return context
+
+
+class ProductSecurityRoadmapTaskUpdateView(LoginRequiredMixin, UpdateView):
+    model = ProductSecurityRoadmapTask
+    form_class = ProductSecurityRoadmapTaskUpdateForm
+    template_name = 'product_security/roadmap_task_form.html'
+    context_object_name = 'task'
+
+    def get_queryset(self):
+        tenant = getattr(self.request, 'tenant', None)
+        return ProductSecurityRoadmapTask.objects.for_tenant(tenant).select_related('roadmap__product')
+
+    def get_success_url(self):
+        return reverse('product_security:roadmap', kwargs={'pk': self.object.roadmap.product_id})
+
+    def form_valid(self, form):
+        rust_result = ProductSecurityBridge.update_roadmap_task(
+            self.request,
+            getattr(self.request, 'tenant', None),
+            self.object.pk,
+            form.cleaned_data,
+        )
+        if rust_result is not None:
+            self.object.status = rust_result.task.status
+            self.object.priority = rust_result.task.priority
+            self.object.owner_role = rust_result.task.owner_role
+            self.object.due_in_days = rust_result.task.due_in_days
+            self.object.dependency_text = rust_result.task.dependency_text
+            return HttpResponseRedirect(
+                reverse('product_security:roadmap', kwargs={'pk': rust_result.product_id})
+            )
+        return super().form_valid(form)

docs/RUST_REWRITE_ROADMAP.md

@@ -100,7 +100,8 @@ ISCY schrittweise von Django/Python auf Rust ueberfuehren, ohne Fachfunktionalit
   - `GET /api/v1/product-security/overview` liefert Product-Security-Matrix, Posture-Kennzahlen, Produktliste und letzte Snapshots fuer den aktuellen Tenant
   - `GET /api/v1/product-security/products/{product_id}` liefert den Detailbaum mit Releases, Komponenten, Threat Models, TARAs, Schwachstellen, AI-Systemen, PSIRT, Advisories, Snapshot und Roadmap-Aufgaben
   - `GET /api/v1/product-security/products/{product_id}/roadmap` liefert Roadmap, Snapshot und Aufgaben fuer die Produkt-Roadmap
-  - Django kann Product-Security-Liste, Produktdetail und Roadmap ueber `PRODUCT_SECURITY_BACKEND=rust_service` aus Rust lesen und im Nicht-Strict-Modus auf lokale ORM-Daten zurueckfallen
+  - `PATCH /api/v1/product-security/roadmap-tasks/{task_id}` aktualisiert Status, Prioritaet, Owner-Rolle, Ziel-Tage und Abhaengigkeitstext tenantgeschuetzt
+  - Django kann Product-Security-Liste, Produktdetail, Roadmap und Roadmap-Task-Updates ueber `PRODUCT_SECURITY_BACKEND=rust_service` aus Rust bedienen und im Nicht-Strict-Modus auf lokale ORM-Daten zurueckfallen
 
 ## App-Migrationsreihenfolge