Scan level inheritance chain

octopoes/octopoes/api/router.py

@@ -1,8 +1,8 @@
 import uuid
 from datetime import datetime, timezone
+from http import HTTPStatus
 from logging import getLogger
 from typing import List, Optional, Set, Type
-from http import HTTPStatus
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Path, status
 from requests import RequestException, HTTPError
@@ -23,6 +23,7 @@
 )
 from octopoes.models.datetime import TimezoneAwareDatetime
 from octopoes.models.exception import ObjectNotFoundException
+from octopoes.models.explanation import InheritanceSection
 from octopoes.models.origin import Origin, OriginType, OriginParameter
 from octopoes.models.pagination import Paginated
 from octopoes.models.tree import ReferenceTree
@@ -258,6 +259,21 @@ def recalculate_scan_profiles(
     xtdb_session_.commit()
 
 
+@router.get("/scan_profiles/inheritance")
+def get_scan_profile_inheritance(
+    octopoes: OctopoesService = Depends(octopoes_service),
+    valid_time: datetime = Depends(extract_valid_time),
+    reference: Reference = Depends(extract_reference),
+) -> List[InheritanceSection]:
+    ooi = octopoes.get_ooi(reference, valid_time)
+    start = InheritanceSection(
+        reference=ooi.reference, level=ooi.scan_profile.level, scan_profile_type=ooi.scan_profile.scan_profile_type
+    )
+    if ooi.scan_profile.scan_profile_type == ScanProfileType.DECLARED:
+        return [start]
+    return octopoes.get_scan_profile_inheritance(reference, valid_time, [start])
+
+
 @router.post("/node")
 def create_node(
     client: str = Depends(extract_client),

octopoes/octopoes/connector/octopoes.py

@@ -18,6 +18,7 @@
     DEFAULT_SCAN_PROFILE_TYPE_FILTER,
 )
 from octopoes.models.exception import ObjectNotFoundException
+from octopoes.models.explanation import InheritanceSection
 from octopoes.models.origin import Origin, OriginParameter
 from octopoes.models.pagination import Paginated
 from octopoes.models.tree import ReferenceTree
@@ -154,3 +155,10 @@ def create_node(self):
 
     def delete_node(self):
         self.session.delete(f"/{self.client}/node")
+
+    def get_scan_profile_inheritance(
+        self, reference: Reference, valid_time: Optional[datetime] = None
+    ) -> List[InheritanceSection]:
+        params = {"reference": str(reference), "valid_time": valid_time}
+        res = self.session.get(f"/{self.client}/scan_profiles/inheritance", params=params)
+        return parse_obj_as(List[InheritanceSection], res.json())

octopoes/octopoes/core/service.py

@@ -25,9 +25,14 @@
     ScanProfileType,
 )
 from octopoes.models.exception import ObjectNotFoundException
+from octopoes.models.explanation import InheritanceSection
 from octopoes.models.origin import Origin, OriginType, OriginParameter
 from octopoes.models.pagination import Paginated
-from octopoes.models.path import get_max_scan_level_issuance, get_paths_to_neighours
+from octopoes.models.path import (
+    get_max_scan_level_issuance,
+    get_paths_to_neighours,
+    get_max_scan_level_inheritance,
+)
 from octopoes.models.tree import ReferenceTree
 from octopoes.repositories.ooi_repository import OOIRepository
 from octopoes.repositories.origin_parameter_repository import OriginParameterRepository
@@ -419,3 +424,68 @@ def list_random_ooi(self, amount: int, valid_time: datetime) -> List[OOI]:
         oois = self.ooi_repository.list_random(amount, valid_time)
         self._populate_scan_profiles(oois, valid_time)
         return oois
+
+    def get_scan_profile_inheritance(
+        self, reference: Reference, valid_time: datetime, inheritance_chain: List[InheritanceSection]
+    ) -> List[InheritanceSection]:
+        neighbour_cache = self.ooi_repository.get_neighbours(reference, valid_time)
+
+        last_inheritance_level = inheritance_chain[-1].level
+        visited = {inheritance.reference for inheritance in inheritance_chain}
+
+        # load scan profiles for all neighbours
+        neighbours_: List[OOI] = [
+            neighbour
+            for neighbours in neighbour_cache.values()
+            for neighbour in neighbours
+            if neighbour.reference not in visited
+        ]
+        self._populate_scan_profiles(neighbours_, valid_time)
+
+        # collect all inheritances
+        inheritances = []
+        for path, neighbours in neighbour_cache.items():
+            segment = path.segments[0]
+            for neighbour in neighbours:
+                segment_inheritance = get_max_scan_level_inheritance(segment)
+                if (
+                    segment_inheritance is None
+                    or neighbour.reference in visited
+                    or neighbour.scan_profile.level < last_inheritance_level
+                ):
+                    continue
+
+                inherited_level = min(get_max_scan_level_inheritance(segment), neighbour.scan_profile.level)
+                inheritances.append(
+                    InheritanceSection(
+                        segment=str(segment),
+                        reference=neighbour.reference,
+                        level=inherited_level,
+                        scan_profile_type=neighbour.scan_profile.scan_profile_type,
+                    )
+                )
+
+        # sort per ooi, per level, ascending
+        inheritances.sort(key=lambda x: x.level)
+
+        # if any declared, return highest straight away
+        declared_inheritances = [
+            inheritance for inheritance in inheritances if inheritance.scan_profile_type == ScanProfileType.DECLARED
+        ]
+        if declared_inheritances:
+            return inheritance_chain + [declared_inheritances[-1]]
+
+        # group by ooi, as the list is already sorted, it will contain the highest inheritance
+        highest_inheritance_per_neighbour = {
+            inheritance.reference: inheritance for inheritance in reversed(inheritances)
+        }
+
+        # traverse depth-first, highest levels first
+        for inheritance in sorted(highest_inheritance_per_neighbour.values(), key=lambda x: x.level, reverse=True):
+            expl = self.get_scan_profile_inheritance(
+                inheritance.reference, valid_time, inheritance_chain + [inheritance]
+            )
+            if expl[-1].scan_profile_type == ScanProfileType.DECLARED:
+                return expl
+
+        return inheritance_chain

octopoes/octopoes/models/explanation.py

@@ -0,0 +1,15 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from octopoes.models import Reference, ScanProfileType
+
+
+class InheritanceSection(BaseModel):
+    reference: Reference
+    level: int
+    segment: Optional[str]
+    scan_profile_type: ScanProfileType
+
+    class Config:
+        arbitrary_types_allowed = True

octopoes/tests/robot/01_scan_profiles.robot

@@ -59,7 +59,6 @@ Empty Scan Profiles
     Verify Scan Profile Mutation Queue    ${REF_IPADDR}    ${{[0]}}
     Verify Scan Profile Mutation Queue    ${REF_RESOLVEDHOSTNAME}    ${{[0]}}
 
-
 *** Keywords ***
 Setup Test
     Start Monitoring    ${QUEUE_URI}

octopoes/tests/robot/06_scan_profile_inheritance.robot

@@ -0,0 +1,31 @@
+*** Settings ***
+Resource            robot.resource
+
+Test Setup          Setup Test
+Test Teardown       Teardown Test
+
+
+*** Test Cases ***
+Simple Scan Profile Inheritance
+    Declare Scan Profile    ${REF_HOSTNAME}    ${4}
+    ${response_data}    Get Scan Profile Inheritance    ${REF_IPADDR}
+    Length Should Be    ${response_data}    3
+    Should Be Equal As Strings    ${response_data[1]["reference"]}    ${REF_RESOLVEDHOSTNAME}
+
+*** Keywords ***
+Setup Test
+    Start Monitoring    ${QUEUE_URI}
+    Insert Normalizer Output
+    Await Sync
+
+Teardown Test
+    Cleanup
+    Await Sync
+    Stop Monitoring
+
+Get Scan Profile Inheritance
+    [Arguments]    ${reference}
+    ${response}    Get    ${OCTOPOES_URI}/scan_profiles/inheritance   params=reference=${reference}
+    Should Be Equal As Integers    ${response.status_code}    200
+    ${response_data}    Set Variable    ${response.json()}
+    [Return]    ${response_data}

octopoes/tests/robot/robot.resource

@@ -63,14 +63,6 @@ Insert Regular Declarations
 
     Await Sync
 
-Get Url
-    [Arguments]    ${url}
-    ${response}    Get    ${url}
-    Should Be Equal As Integers    ${response.status_code}    200
-    Should Be Equal    ${response.headers["content-type"]}    application/json
-    ${response_data}    Set Variable    ${response.json()}
-    RETURN    ${response_data}
-
 Get Messages From Queue
     [Arguments]    ${queue}    ${ackmode}
     &{data}    Create dictionary    count=10000    ackmode=${ackmode}    encoding=auto    truncate=50000
@@ -116,7 +108,10 @@ Await Sync
     Wait For XTDB Synced
 
 Get Objects
-    ${response_data}    Get Url    ${OCTOPOES_URI}/objects
+    ${response}    Get   ${OCTOPOES_URI}/objects
+    Should Be Equal As Integers    ${response.status_code}    200
+    Should Be Equal    ${response.headers["content-type"]}    application/json
+    ${response_data}    Set Variable    ${response.json()}
     RETURN    ${response_data}
 
 Get Valid Time Params

rocky/rocky/templates/oois/ooi_detail.html

@@ -97,6 +97,11 @@ <h2>{% translate "Scan" %} {{ ooi.human_readable }} {% translate "using boefjes"
                         {% endif %}
                     </div>
                 {% endif %}
+
+              {% if ooi.scan_profile.scan_profile_type == 'inherited' %}
+                {% include "partials/explanations.html" %}
+              {% endif %}
+
             </div>
         </section>
     </main>

rocky/rocky/templates/partials/explanations.html

@@ -0,0 +1,42 @@
+{% load i18n %}
+{% load ooi_extra %}
+
+{% spaceless %}
+        <div class="horizontal-scroll">
+            <h2>{% translate "Clearance level inheritance" %}</h2>
+            {% if clearance_level_inheritance %}
+            <table>
+                <thead>
+                    <tr>
+                        <th>{% translate "Type" %}</th>
+                        <th>{% translate "Object Type" %}</th>
+                        <th>{% translate "OOI" %}</th>
+                        <th>{% translate "Clearance level" %}</th>
+                    </tr>
+                </thead>
+                <tbody>
+                    {% for section in clearance_level_inheritance %}
+                        <tr>
+
+                            {% if forloop.first %}
+                                <td>{% translate "OOI" %}</td>
+                            {% elif forloop.last %}
+                                <td>{% translate "Origin" %}</td>
+                            {% else %}
+                                <td>Object</td>
+                            {% endif %}
+                            <td>
+                              <a href="{% ooi_url 'ooi_detail' section.primary_key organization.code %}">{{ section.human_readable }}</a>
+                            </td>
+                            <td>{{ section.object_type }}</td>
+                            <td>L{{ section.level }}</td>
+                        </tr>
+
+                    {% endfor %}
+                </tbody>
+            </table>
+              {% else %}
+                <a href="{% ooi_url 'ooi_detail' ooi_id=ooi organization_code=organization.code show_clearance_level_inheritance=True %}">{% translate "Show clearance level inheritance" %}</a>
+              {% endif %}
+        </div>
+{% endspaceless %}

rocky/rocky/views/mixins.py

@@ -14,6 +14,7 @@
 from octopoes.connector import ObjectNotFoundException
 from octopoes.connector.octopoes import OctopoesAPIConnector
 from octopoes.models import OOI, Reference, ScanLevel, ScanProfileType
+from octopoes.models.explanation import InheritanceSection
 from octopoes.models.ooi.findings import Finding
 from octopoes.models.origin import Origin, OriginType
 from octopoes.models.tree import ReferenceTree
@@ -119,6 +120,9 @@ def get_depth(self, default_depth=DEPTH_DEFAULT) -> int:
         except ValueError:
             return default_depth
 
+    def get_scan_profile_inheritance(self, ooi: OOI) -> List[InheritanceSection]:
+        return self.octopoes_api_connector.get_scan_profile_inheritance(ooi.reference)
+
 
 class OOIList:
     HARD_LIMIT = 99_999_999

rocky/rocky/views/ooi_detail.py

@@ -12,7 +12,8 @@
 from katalogus.client import get_katalogus
 from katalogus.utils import get_enabled_boefjes_for_ooi_class
 from katalogus.views.mixins import BoefjeMixin
-from octopoes.models import OOI
+from octopoes.models import OOI, Reference
+
 from rocky import scheduler
 from rocky.views.mixins import OOIBreadcrumbsMixin
 from rocky.views.ooi_detail_related_object import OOIRelatedObjectAddView
@@ -49,7 +50,8 @@ def post(self, request, *args, **kwargs):
 
         self.ooi = self.get_ooi()
 
-        if not self.handle_page_action(request.POST.get("action")):
+        action = self.request.POST.get("action")
+        if not self.handle_page_action(action):
             return self.get(request, status_code=500, *args, **kwargs)
 
         success_message = (
@@ -192,4 +194,16 @@ def get_context_data(self, **kwargs):
             "scan_history_search",
             "scan_history_page",
         ]
+        if self.request.GET.get("show_clearance_level_inheritance"):
+            clearance_level_inheritance = self.get_scan_profile_inheritance(self.ooi)
+            formatted_inheritance = [
+                {
+                    "object_type": Reference.from_str(section.reference).class_,
+                    "primary_key": section.reference,
+                    "human_readable": Reference.from_str(section.reference).human_readable,
+                    "level": section.level,
+                }
+                for section in clearance_level_inheritance
+            ]
+            context["clearance_level_inheritance"] = formatted_inheritance
         return context