Scope classic search and endpoint report reads of shared location tags

dojo/endpoint/models.py

@@ -239,6 +239,11 @@ def __str__(self):
     def get_absolute_url(self):
         return reverse("view_endpoint", args=[str(self.id)])
 
+    @property
+    def readable_tags(self):
+        """Mirror of ``Location.readable_tags``. An Endpoint belongs to one product, so all of them."""
+        return list(self.tags.all())
+
     @classmethod
     @contextlib.contextmanager
     def allow_endpoint_init(cls):

dojo/search/views.py

@@ -17,7 +17,8 @@
 from dojo.finding.ui.filters import FindingFilter, FindingFilterWithoutObjectLookups
 from dojo.forms import FindingBulkUpdateForm, SimpleSearchForm
 from dojo.location.feature import locations_enabled
-from dojo.location.queries import get_authorized_locations, prefetch_for_locations
+from dojo.location.models import Location
+from dojo.location.queries import get_authorized_locations, prefetch_for_locations, readable_tag_match
 from dojo.models import Engagement, Finding, Finding_Template, Product, Test
 from dojo.product.queries import get_authorized_app_analysis, get_authorized_languages, get_authorized_products
 from dojo.test.queries import get_authorized_tests
@@ -237,7 +238,11 @@ def simple_search(request):
                 tagged_tests = authorized_tests.filter(Q1 | Q2).exclude(Q3 | Q4).distinct()[:max_results].prefetch_related("tags")
                 tagged_engagements = authorized_engagements.filter(Q1 | Q2).exclude(Q3 | Q4).distinct()[:max_results].prefetch_related("tags")
                 tagged_products = authorized_products.filter(Q1 | Q2).exclude(Q3 | Q4).distinct()[:max_results].prefetch_related("tags")
-                tagged_endpoints = authorized_endpoints.filter(Q1 | Q2).exclude(Q3 | Q4).distinct()[:max_results].prefetch_related("tags")
+                if locations_enabled():
+                    L1, L2, L3, L4 = location_tag_queries(tag, tags, not_tag, not_tags)
+                    tagged_endpoints = authorized_endpoints.filter(L1 | L2).exclude(L3 | L4).distinct()[:max_results].prefetch_related("tags")
+                else:
+                    tagged_endpoints = authorized_endpoints.filter(Q1 | Q2).exclude(Q3 | Q4).distinct()[:max_results].prefetch_related("tags")
             else:
                 tagged_findings = None
                 tagged_finding_templates = None
@@ -477,7 +482,42 @@ def vulnerability_id_fix(keyword):
     return keyword
 
 
+def location_tag_queries(tag, tags, not_tag, not_tags):
+    """The four tag predicates of ``simple_search``, matched over readable tag sets only."""
+    def match(**lookups):
+        return readable_tag_match("pk", **lookups)
+    return (
+        match(tags__name__contains=tag) if tag else Q(),
+        match(tags__name__in=tags) if tags else Q(),
+        match(tags__name__contains=not_tag) if not_tag else Q(),
+        match(tags__name__in=not_tags) if not_tags else Q(),
+    )
+
+
+def apply_location_tag_filters(qs, operators):
+    """
+    The Location branch of :func:`apply_tag_filters`.
+
+    A Location row is shared by every product referencing it and so is its tag set, so a
+    filter joining the tag relation directly matches through other products' tags and turns
+    a substring lookup into an oracle over a set the page withholds. These run the same
+    predicate the templates render, so a filter never matches on a withheld value.
+    """
+    if "tag" in operators:
+        qs = qs.filter(readable_tag_match("pk", tags__name__contains=",".join(operators["tag"])))
+    if "tags" in operators:
+        qs = qs.filter(readable_tag_match("pk", tags__name__in=operators["tags"]))
+    if "not-tag" in operators:
+        qs = qs.exclude(readable_tag_match("pk", tags__name__contains=",".join(operators["not-tag"])))
+    if "not-tags" in operators:
+        qs = qs.exclude(readable_tag_match("pk", tags__name__in=operators["not-tags"]))
+    return qs
+
+
 def apply_tag_filters(qs, operators, *, skip_relations=False):
+    if qs.model is Location:
+        return apply_location_tag_filters(qs, operators)
+
     tag_filters = {"tag": ""}
 
     if qs.model == Finding:

dojo/templates/dojo/request_endpoint_report.html

@@ -63,7 +63,7 @@ <h3 class="has-filters">
                                 {% if V3_FEATURE_LOCATIONS %}
                                     <td>
                                         <a href="{% url 'view_endpoint' e.id %}">{{ e|truncatechars_html:70 }}</a>
-                                        {% include "dojo/snippets/tags.html" with tags=e.tags.all %}
+                                        {% include "dojo/snippets/tags.html" with tags=e.readable_tags %}
                                     </td>
                                     <td><a href="{% url 'all_findings' %}?endpoints={{ e.id }}&location_status=Active">{{ e.active_findings }}</a></td>
                                     <td><a href="{% url 'product' %}?endpoints={{ e.id }}&location_status=Active">{{ e.active_products }}</a></td>

dojo/templates/dojo/simple_search.html

@@ -271,7 +271,7 @@ <h2> {{ name }} <i class="fa-solid fa-circle-question has-popover"
                             <tr>
                                 <td>
                                     <a href="{% url 'view_endpoint' e.id %}">{{ e }}{% if e.is_broken %} <span data-toggle="tooltip" title="{% trans "Endpoint is broken. Check documentation to look for fix process" %}" >&#128681;</span>{% endif %}</a>
-                                    {% include "dojo/snippets/tags.html" with tags=e.tags.all %}
+                                    {% include "dojo/snippets/tags.html" with tags=e.readable_tags %}
                                 </td>
                                 {% if e.product %}
                                     <td>
@@ -421,7 +421,7 @@ <h2> {{ name }} <i class="fa-solid fa-circle-question has-popover"
                                 <td>{% trans "Endpoint" %}</td>
                                 <td>
                                     <a class="search-finding" href="{% url 'view_endpoint' endpoint.id %}">{{ endpoint }}{% if endpoint.is_broken %} <span data-toggle="tooltip" title="{% trans 'Endpoint is broken. Check documentation to look for fix process' %}" >&#128681;</span>{% endif %}</a>
-                                    {% include "dojo/snippets/tags.html" with tags=endpoint.tags.all %}
+                                    {% include "dojo/snippets/tags.html" with tags=endpoint.readable_tags %}
                                 </td>
                             </tr>
                         {% endfor %}

unittests/test_location_tag_ui_scoping.py

@@ -0,0 +1,122 @@
+"""
+Location tag scoping on the two pages ``test_location_tag_scoping.py`` does not cover.
+
+That suite asserts the REST bodies and filters only, so the classic search page and the
+Product Endpoint Report kept rendering and matching the raw shared tag relation. Both read
+a Location the caller is authorized for, whose tag set belongs to every product on the row.
+"""
+from django.test import override_settings
+from django.urls import reverse
+from django.utils.timezone import now
+
+from dojo.authorization.roles_permissions import Roles
+from dojo.models import (
+    Dojo_User,
+    Engagement,
+    Finding,
+    Product,
+    Product_Member,
+    Product_Type,
+    Role,
+    Test,
+    Test_Type,
+    User,
+)
+from dojo.url.models import URL
+from unittests.dojo_test_case import DojoTestCase, skip_unless_v3
+
+SHARED_HOST = "uitagscope-shared.example.test"
+OWN_HOST = "uitagscope-own.example.test"
+
+FOREIGN_TAG = "bsecretr7k2q9"
+OWN_TAG = "aownlabel"
+
+
+@skip_unless_v3
+@override_settings(WATSON_SEARCH_ENABLED=True)
+class LocationTagUIScopingTest(DojoTestCase):
+    @classmethod
+    def setUpTestData(cls):
+        prod_type, _ = Product_Type.objects.get_or_create(name="UITagScope PT")
+        test_type, _ = Test_Type.objects.get_or_create(name="UITagScope Scan")
+
+        def product(name):
+            return Product.objects.create(name=name, description=name, prod_type=prod_type)
+
+        cls.mine = product("UITagScope Mine")
+        cls.theirs = product("UITagScope Theirs")
+
+        cls.alice = User.objects.create_user(
+            username="uitagscope_alice",
+            password="not-a-real-secret",  # noqa: S106 - test fixture user
+        )
+        Product_Member.objects.create(
+            user=cls.alice, product=cls.mine, role=Role.objects.get(id=Roles.Reader),
+        )
+        cls.mine.authorized_users.add(Dojo_User.objects.get(pk=cls.alice.pk))
+
+        engagement = Engagement.objects.create(
+            product=cls.mine, name="UITagScope eng",
+            target_start=now().date(), target_end=now().date(),
+        )
+        test = Test.objects.create(
+            engagement=engagement, test_type=test_type, target_start=now(), target_end=now(),
+        )
+        finding = Finding.objects.create(
+            test=test, title="UITagScope finding", severity="High", numerical_severity="S1",
+            active=True, verified=True, description="body", reporter=cls.alice,
+        )
+
+        def location(host, tag, *products):
+            loc = URL.get_or_create_from_values(protocol="https", host=host, path="x").location
+            for prod in products:
+                loc.associate_with_product(prod)
+            loc.associate_with_finding(finding, audit_time=now())
+            loc.tags.set([tag])
+            return loc
+
+        # Alice's product references it, so the row is hers to see. The tag came from theirs.
+        cls.shared = location(SHARED_HOST, FOREIGN_TAG, cls.mine, cls.theirs)
+        cls.own = location(OWN_HOST, OWN_TAG, cls.mine)
+
+    def _search(self, query):
+        self.client.force_login(self.alice)
+        response = self.client.get(reverse("simple_search"), {"query": query})
+        self.assertEqual(response.status_code, 200)
+        return response.content.decode()
+
+    def _report(self):
+        """The report options page, which lists the locations the report will cover."""
+        self.client.force_login(self.alice)
+        response = self.client.get(reverse("product_endpoint_report", args=(self.mine.id,)))
+        self.assertEqual(response.status_code, 200)
+        return response.content.decode()
+
+    def test_search_withholds_a_foreign_products_tag(self):
+        body = self._search("uitagscope")
+        self.assertIn(SHARED_HOST, body)
+        self.assertNotIn(FOREIGN_TAG, body)
+
+    def test_search_still_renders_my_own_tag(self):
+        self.assertIn(OWN_TAG, self._search("uitagscope"))
+
+    def test_search_tag_operator_does_not_match_a_foreign_tag(self):
+        for query in (f"tag:{FOREIGN_TAG}", f"tag:{FOREIGN_TAG[:6]}", f"tags:{FOREIGN_TAG}"):
+            self.assertNotIn(SHARED_HOST, self._search(query), query)
+
+    def test_search_tag_operator_still_matches_my_own_tag(self):
+        for query in (f"tag:{OWN_TAG}", f"tag:{OWN_TAG[:6]}", f"tags:{OWN_TAG}"):
+            body = self._search(query)
+            self.assertIn(OWN_HOST, body, query)
+            self.assertIn(OWN_TAG, body, query)
+
+    def test_search_negated_tag_operator_treats_a_foreign_tag_set_as_empty(self):
+        self.assertIn(SHARED_HOST, self._search(f"not-tag:{FOREIGN_TAG}"))
+
+    def test_product_endpoint_report_withholds_a_foreign_products_tag(self):
+        body = self._report()
+        self.assertIn(SHARED_HOST, body)
+        self.assertNotIn(FOREIGN_TAG, body)
+
+    def test_product_endpoint_report_still_renders_my_own_tag(self):
+        self.assertIn(OWN_TAG, self._report())