refactor raising of scanprofile (#162)

account/mixins.py

@@ -1,9 +1,19 @@
+from datetime import datetime, timezone
+
 from django.contrib import messages
 from django.http import Http404
 from django.utils.translation import gettext_lazy as _
 from django.views import View
 
 from octopoes.connector.octopoes import OctopoesAPIConnector
+from octopoes.models import DeclaredScanProfile, ScanLevel, Reference
+
+from rocky.exceptions import (
+    IndemnificationNotPresentException,
+    AcknowledgedClearanceLevelTooLowException,
+    TrustedClearanceLevelTooLowException,
+    ClearanceLevelTooLowException,
+)
 from rocky.settings import OCTOPOES_API
 from tools.models import Organization, OrganizationMember, Indemnification
 
@@ -13,14 +23,20 @@ def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.organization = None
         self.octopoes_api_connector = None
-        self._may_update_scan_profile = False
         self.organization_member = None
+        self.indemnification_present = False
 
     def setup(self, request, *args, **kwargs):
         super().setup(request, *args, **kwargs)
+
+        # authentication/otp flow happens before setup
+        if not request.user.is_authenticated:
+            return
+
         organization_code = kwargs["organization_code"]
         try:
             self.organization = Organization.objects.get(code=organization_code)
+            self.indemnification_present = Indemnification.objects.filter(organization=self.organization).exists()
         except Organization.DoesNotExist:
             self.organization = None
         try:
@@ -44,29 +60,63 @@ def get_context_data(self, **kwargs):
         if self.organization:
             context["organization"] = self.organization
         context["organization_member"] = self.organization_member
-        context["may_update_scan_profile"] = self.may_update_scan_profile
+        context["may_update_clearance_level"] = self.may_update_clearance_level
         return context
 
     @property
-    def may_update_scan_profile(self):
-        if not Indemnification.objects.filter(organization=self.organization).exists():
+    def may_update_clearance_level(self) -> bool:
+        if not self.indemnification_present:
             return False
         if self.organization_member.acknowledged_clearance_level < 0:
             return False
         if self.organization_member.trusted_clearance_level < 0:
             return False
         return True
 
-    def verify_may_update_scan_profile(self) -> bool:
-        if not Indemnification.objects.filter(organization=self.organization).exists():
-            messages.add_message(self.request, messages.ERROR, _("No indemnification present for organization."))
-            return False
-
-        if self.organization_member.acknowledged_clearance_level < 0:
-            messages.add_message(self.request, messages.ERROR, _("Acknowledged clearance level too low."))
-            return False
+    def verify_raise_clearance_level(self, level: int) -> bool:
+        if not self.indemnification_present:
+            raise IndemnificationNotPresentException()
+        if self.organization_member.acknowledged_clearance_level < level:
+            raise AcknowledgedClearanceLevelTooLowException()
+        if self.organization_member.trusted_clearance_level < level:
+            raise TrustedClearanceLevelTooLowException()
+        return True
 
-        if self.organization_member.trusted_clearance_level < 0:
-            messages.add_message(self.request, messages.ERROR, _("Trusted clearance level too low."))
-            return False
+    def raise_clearance_level(self, ooi_reference: Reference, level: int) -> bool:
+        try:
+            self.verify_raise_clearance_level(level)
+            self.octopoes_api_connector.save_scan_profile(
+                DeclaredScanProfile(reference=ooi_reference, level=ScanLevel(level)),
+                datetime.now(timezone.utc),
+            )
+        except IndemnificationNotPresentException as exc:
+            messages.add_message(
+                self.request,
+                messages.ERROR,
+                _(
+                    "Could not raise clearance level of %s to L%s. \
+                    Indemnification not present at organization %s."
+                )
+                % (
+                    ooi_reference.human_readable,
+                    level,
+                    self.organization.name,
+                ),
+            )
+            raise exc
+        except ClearanceLevelTooLowException as exc:
+            messages.add_message(
+                self.request,
+                messages.ERROR,
+                _(
+                    "Could not raise clearance level of %s to L%s. \
+                    You acknowledged a clearance level of %s."
+                )
+                % (
+                    ooi_reference.human_readable,
+                    level,
+                    self.organization_member.acknowledged_clearance_level,
+                ),
+            )
+            raise exc
         return True

katalogus/views/change_clearance_level.py

@@ -37,15 +37,11 @@ def get(self, request, *args, **kwargs):
 
     def post(self, request, *args, **kwargs):
         """Start scanning oois at plugin detail page."""
-        if not self.verify_may_update_scan_profile():
+        if not self.indemnification_present:
             return self.get(request, *args, **kwargs)
 
         boefje = self.katalogus_client.get_boefje(self.plugin_id)
-        self.run_boefje_for_oois(
-            boefje=boefje,
-            oois=self.oois,
-            api_connector=self.octopoes_api_connector,
-        )
+        self.run_boefje_for_oois(boefje=boefje, oois=self.oois)
         messages.add_message(self.request, messages.SUCCESS, _("Scanning successfully scheduled."))
         del request.session["selected_oois"]  # delete session
         return redirect(reverse("task_list", kwargs={"organization_code": self.organization.code}))

katalogus/views/mixins.py

@@ -1,4 +1,3 @@
-from datetime import datetime, timezone
 from logging import getLogger
 from typing import List
 from uuid import uuid4
@@ -10,8 +9,8 @@
 
 from account.mixins import OrganizationView
 from katalogus.client import get_katalogus
-from octopoes.connector.octopoes import OctopoesAPIConnector
-from octopoes.models import OOI, DeclaredScanProfile
+from octopoes.models import OOI
+from rocky.exceptions import IndemnificationNotPresentException, ClearanceLevelTooLowException
 from rocky.scheduler import Boefje, BoefjeTask, QueuePrioritizedItem, client
 from rocky.views.mixins import OctopoesView
 
@@ -70,18 +69,14 @@ def run_boefje_for_oois(
         self,
         boefje: Boefje,
         oois: List[OOI],
-        api_connector: OctopoesAPIConnector,
     ) -> None:
 
         for ooi in oois:
             if ooi.scan_profile.level < boefje.scan_level:
-                api_connector.save_scan_profile(
-                    DeclaredScanProfile(
-                        reference=ooi.reference,
-                        level=boefje.scan_level,
-                    ),
-                    datetime.now(timezone.utc),
-                )
+                try:
+                    self.raise_clearance_level(ooi.reference, boefje.scan_level)
+                except (IndemnificationNotPresentException, ClearanceLevelTooLowException):
+                    continue
             self.run_boefje(boefje, ooi)
 
     def scan(self, view_args) -> None:
@@ -106,7 +101,7 @@ def scan(self, view_args) -> None:
         oois = [self.get_single_ooi(pk=ooi_id) for ooi_id in ooi_ids]
 
         try:
-            self.run_boefje_for_oois(boefje, oois, self.octopoes_api_connector)
+            self.run_boefje_for_oois(boefje, oois)
         except HTTPError:
             return
 

katalogus/views/plugin_detail_scan_oois.py

@@ -19,7 +19,7 @@ class PluginDetailScanOOI(BoefjeMixin, TemplateView):
 
     def post(self, request, *args, **kwargs):
         """Start scanning oois at plugin detail page."""
-        if not self.verify_may_update_scan_profile():
+        if not self.indemnification_present:
             return self.get(request, *args, **kwargs)
 
         selected_oois = request.POST.getlist("ooi")
@@ -31,7 +31,6 @@ def post(self, request, *args, **kwargs):
                 self.run_boefje_for_oois(
                     boefje=boefje,
                     oois=oois_with_clearance_level,
-                    api_connector=self.octopoes_api_connector,
                 )
             oois_without_clearance_level = self.get_oois_without_clearance_level(selected_oois)
             if oois_without_clearance_level:

onboarding/views.py

@@ -1,4 +1,3 @@
-from datetime import datetime, timezone
 from typing import Type, List, Dict, Any
 
 from django.contrib import messages
@@ -18,7 +17,6 @@
 from account.mixins import OrganizationView
 from katalogus.client import get_katalogus
 from octopoes.connector.octopoes import OctopoesAPIConnector
-from octopoes.models import DeclaredScanProfile
 from octopoes.models import OOI
 from octopoes.models.ooi.network import Network
 from octopoes.models.types import type_by_name
@@ -34,6 +32,7 @@
     KatIntroductionAdminStepsMixin,
     KatIntroductionRegistrationStepsMixin,
 )
+from rocky.exceptions import IndemnificationNotPresentException, ClearanceLevelTooLowException
 from rocky.views.indemnification_add import IndemnificationAddView
 from rocky.views.ooi_report import Report, DNSReport, build_findings_list_from_store
 from rocky.views.ooi_view import BaseOOIFormView, SingleOOITreeMixin, BaseOOIDetailView
@@ -258,20 +257,17 @@ def get(self, request, *args, **kwargs):
         return super().get(request, *args, **kwargs)
 
     def post(self, request, *args, **kwargs):
-        if not self.verify_may_update_scan_profile():
+
+        ooi = self.get_ooi()
+        level = int(self.request.session["clearance_level"])
+        try:
+            self.raise_clearance_level(ooi.reference, level)
+        except (IndemnificationNotPresentException, ClearanceLevelTooLowException):
             return self.get(request, *args, **kwargs)
 
-        self.set_clearance_level()
         self.enable_selected_boefjes()
         return redirect(get_ooi_url("step_report", self.get_ooi_id(), self.organization.code))
 
-    def set_clearance_level(self):
-        ooi = self.get_ooi()
-        self.octopoes_api_connector.save_scan_profile(
-            DeclaredScanProfile(reference=ooi.reference, level=self.request.session["clearance_level"]),
-            valid_time=datetime.now(timezone.utc),
-        )
-
     def enable_selected_boefjes(self) -> None:
         if not self.request.session.get("selected_boefjes"):
             return
@@ -305,7 +301,7 @@ def get_success_url(self, **kwargs):
         return get_ooi_url("step_setup_scan_select_plugins", self.request.GET.get("ooi_id"), self.organization.code)
 
     def form_valid(self, form):
-        self.request.session["clearance_level"] = form.data["level"]
+        self.request.session["clearance_level"] = form.cleaned_data["level"]
         self.add_success_notification()
         return super().form_valid(form)
 

rocky/exceptions.py

@@ -1,2 +1,18 @@
 class RockyError(Exception):
     pass
+
+
+class IndemnificationNotPresentException(Exception):
+    pass
+
+
+class ClearanceLevelTooLowException(Exception):
+    pass
+
+
+class AcknowledgedClearanceLevelTooLowException(ClearanceLevelTooLowException):
+    pass
+
+
+class TrustedClearanceLevelTooLowException(ClearanceLevelTooLowException):
+    pass

rocky/templates/oois/ooi_list.html

@@ -33,7 +33,7 @@ <h1>{% translate 'Object list' %}</h1>
           {% if ooi_list %}
             <fieldset>
               <div class="horizontal-view">
-                {% if may_update_scan_profile %}
+                {% if may_update_clearance_level %}
                   <select id="scan-profiles" name="scan-profile">
                     {% for level in scan_levels %}
                       <option value="{{ level }}">
@@ -47,10 +47,8 @@ <h1>{% translate 'Object list' %}</h1>
                           value="update-scan-profile">
                     {% translate 'Set clearance level' %}
                   </button>
-                  <button type="submit" class="dropdown-button" name="action" value="delete">{% translate 'Delete object(s)' %}</button>
-                {% else %}
-                  <button type="submit" class="dropdown-button" name="action" value="delete">{% translate 'Delete object(s)' %}</button>
                 {% endif %}
+                <button type="submit" class="dropdown-button" name="action" value="delete">{% translate 'Delete object(s)' %}</button>
               </div>
             </fieldset>
           {% endif %}

rocky/views/mixins.py

@@ -19,8 +19,8 @@
 from octopoes.models.tree import ReferenceTree
 from octopoes.models.types import get_relations, get_collapsed_types, type_by_name
 from rocky.bytes_client import get_bytes_client
-from tools.forms.settings import DEPTH_MAX, DEPTH_DEFAULT
 from tools.forms.base import ObservedAtForm
+from tools.forms.settings import DEPTH_MAX, DEPTH_DEFAULT
 from tools.models import Organization
 from tools.ooi_helpers import (
     get_knowledge_base_data_for_ooi_store,

rocky/views/ooi_detail.py

@@ -34,7 +34,7 @@ class OOIDetailView(
     scan_history_limit = 10
 
     def post(self, request, *args, **kwargs):
-        if not self.verify_may_update_scan_profile():
+        if not self.indemnification_present:
             return self.get(request, *args, **kwargs)
 
         if "action" not in self.request.POST:
@@ -61,7 +61,7 @@ def handle_page_action(self, action: str) -> bool:
 
                 boefje = get_katalogus(self.organization.code).get_boefje(boefje_id)
                 ooi = self.get_single_ooi(pk=ooi_id)
-                self.run_boefje_for_oois(boefje, [ooi], self.api_connector)
+                self.run_boefje_for_oois(boefje, [ooi])
                 return True
 
         except RequestException as exception: