From bc37a74b6f78c6a0e5283e6e70183248b97c67f0 Mon Sep 17 00:00:00 2001 From: svader0 Date: Mon, 24 Aug 2026 12:06:36 -0500 Subject: [PATCH] Scope Location tag reads to the caller's own products A Location row is deduplicated globally and its tag set is one field on that row, shared by every product that references it. Reads of that set, and the filters that match on it, now require the caller to be authorized for every product on the row. The same predicate backs both, so a filter never matches a value the response body withholds. --- dojo/location/api/endpoint_compat.py | 30 ++- dojo/location/api/filters.py | 3 +- dojo/location/api/serializers.py | 12 +- dojo/location/api/tag_filters.py | 99 ++++++++++ dojo/location/models.py | 17 ++ dojo/location/queries.py | 40 ++++ dojo/templates/dojo/snippets/endpoints.html | 4 +- dojo/templates/dojo/url/list.html | 2 +- dojo/templates/dojo/url/view.html | 4 +- dojo/url/ui/forms.py | 2 +- unittests/test_location_tag_scoping.py | 196 ++++++++++++++++++++ 11 files changed, 390 insertions(+), 19 deletions(-) create mode 100644 dojo/location/api/tag_filters.py create mode 100644 unittests/test_location_tag_scoping.py diff --git a/dojo/location/api/endpoint_compat.py b/dojo/location/api/endpoint_compat.py index ccfc156a9bc..caed9961c05 100644 --- a/dojo/location/api/endpoint_compat.py +++ b/dojo/location/api/endpoint_compat.py @@ -26,10 +26,15 @@ from dojo.api_v2 import serializers from dojo.api_v2.prefetch import PrefetchListMixin, PrefetchRetrieveMixin -from dojo.api_v2.serializers import TagListSerializerField from dojo.api_v2.views import report_generate from dojo.authorization.api_permissions import check_object_permission -from dojo.filters import CharFieldFilterANDExpression, CharFieldInFilter, OrderingFilter +from dojo.filters import OrderingFilter +from dojo.location.api.tag_filters import ( + ReadableHasTagsFilter, + ReadableTagANDFilter, + ReadableTagFilter, + ReadableTagInFilter, +) from dojo.location.models import LocationFindingReference, LocationProductReference from dojo.location.queries import ( authorized_finding_references, @@ -77,6 +82,10 @@ def has_object_permission(self, request, view, obj): # Endpoint compatibility ########## +# Rows here are LocationProductReference, so tag lookups resolve against location_id. +_LOCATION = "location_id" + + class V3EndpointCompatibleFilterSet(FilterSet): """Endpoint-compatible FilterSet.""" @@ -95,12 +104,12 @@ class V3EndpointCompatibleFilterSet(FilterSet): location_id = NumberFilter(field_name="location__id", lookup_expr="exact") - tag = CharFilter(field_name="location__tags__name", lookup_expr="icontains", help_text="Tag name contains") - tags = CharFieldInFilter(field_name="location__tags__name", lookup_expr="in", help_text="Comma separated list of exact tags (uses OR for multiple values)") - tags__and = CharFieldFilterANDExpression(field_name="location__tags__name", help_text="Comma separated list of exact tags to match with an AND expression") - not_tag = CharFilter(field_name="location__tags__name", lookup_expr="icontains", help_text="Not Tag name contains", exclude=True) - not_tags = CharFieldInFilter(field_name="location__tags__name", lookup_expr="in", help_text="Comma separated list of exact tags not present on model", exclude=True) - has_tags = BooleanFilter(field_name="location__tags", lookup_expr="isnull", exclude=True, label="Has tags") + tag = ReadableTagFilter(lookup_expr="icontains", location_field=_LOCATION, help_text="Tag name contains") + tags = ReadableTagInFilter(location_field=_LOCATION, help_text="Comma separated list of exact tags (uses OR for multiple values)") + tags__and = ReadableTagANDFilter(location_field=_LOCATION, help_text="Comma separated list of exact tags to match with an AND expression") + not_tag = ReadableTagFilter(lookup_expr="icontains", exclude=True, location_field=_LOCATION, help_text="Not Tag name contains") + not_tags = ReadableTagInFilter(exclude=True, location_field=_LOCATION, help_text="Comma separated list of exact tags not present on model") + has_tags = ReadableHasTagsFilter(location_field=_LOCATION, label="Has tags") o = OrderingFilter( fields=( @@ -126,7 +135,7 @@ class V3EndpointCompatibleSerializer(ModelSerializer): path = CharField(source="location.url.path") query = CharField(source="location.url.query") fragment = CharField(source="location.url.fragment") - tags = TagListSerializerField(source="location.tags") + tags = SerializerMethodField() location_id = IntegerField(source="location.id") active_finding_count = IntegerField(read_only=True) @@ -134,6 +143,9 @@ class Meta: model = LocationProductReference exclude = ("location",) + def get_tags(self, obj: LocationProductReference) -> list[str]: + return sorted(tag.name for tag in obj.location.readable_tags) + class V3EndpointCompatibleViewSet(PrefetchListMixin, PrefetchRetrieveMixin, viewsets.ReadOnlyModelViewSet): diff --git a/dojo/location/api/filters.py b/dojo/location/api/filters.py index 9cab6c21bfd..9021c92098e 100644 --- a/dojo/location/api/filters.py +++ b/dojo/location/api/filters.py @@ -3,6 +3,7 @@ from django_filters import NumberFilter from dojo.api_helpers.filters import CommonFilters, StaticMethodFilters +from dojo.location.api.tag_filters import create_readable_tag_filters from dojo.location.status import FindingLocationStatus, ProductLocationStatus @@ -28,7 +29,7 @@ class LocationFilter(CommonFilters): # ordering (the order of the fields is enforced) CommonFilters.create_char_filters("location_type", "Location Type", locals()) CommonFilters.create_char_filters("location_value", "Location Value", locals()) - CommonFilters.create_char_filters("tags__name", "Tags", locals()) + create_readable_tag_filters("Tags", locals()) CommonFilters.create_integer_filters("products__product", "Product ID", locals()) CommonFilters.create_integer_filters("findings__finding", "Finding ID", locals()) CommonFilters.create_ordering_filters( diff --git a/dojo/location/api/serializers.py b/dojo/location/api/serializers.py index 2695736390e..155fa86c69d 100644 --- a/dojo/location/api/serializers.py +++ b/dojo/location/api/serializers.py @@ -1,7 +1,7 @@ from __future__ import annotations from rest_framework.relations import PrimaryKeyRelatedField -from rest_framework.serializers import CharField +from rest_framework.serializers import CharField, SerializerMethodField from dojo.api_helpers.serializers import BaseModelSerializer from dojo.api_v2.serializers import TagListSerializerField @@ -38,8 +38,14 @@ class LocationSerializer(BaseModelSerializer): """Serializer for the Location model with serializers for the related objects.""" - tags = TagListSerializerField(required=False) - inherited_tags = TagListSerializerField(required=False) + tags = SerializerMethodField() + inherited_tags = SerializerMethodField() + + def get_tags(self, obj: Location) -> list[str]: + return sorted(tag.name for tag in obj.readable_tags) + + def get_inherited_tags(self, obj: Location) -> list[str]: + return sorted(tag.name for tag in obj.readable_inherited_tags) class Meta: diff --git a/dojo/location/api/tag_filters.py b/dojo/location/api/tag_filters.py new file mode 100644 index 00000000000..9225eb1af30 --- /dev/null +++ b/dojo/location/api/tag_filters.py @@ -0,0 +1,99 @@ +""" +Tag filters that match only the tag sets the caller may read. + +A Location row is shared by every product referencing it and so is its tag set, so a filter +that joins the tag relation directly matches through other products' tags. That turns any +substring lookup into a character-by-character oracle over a tag set the caller cannot read +in a response body. Every filter here runs its lookup inside +``dojo.location.queries.readable_tag_match`` instead, which is the same predicate the +serializers use, so a filter never matches on a value the body would withhold. +""" +from django_filters import BooleanFilter, CharFilter +from django_filters.constants import EMPTY_VALUES + +from dojo.api_helpers.filters import CharFieldInFilter, StaticMethodFilters +from dojo.location.queries import readable_tag_match + + +class _ReadableTagFilterMixin: + location_field = "pk" + + def __init__(self, *args, location_field="pk", **kwargs): + self.location_field = location_field + super().__init__(*args, **kwargs) + + def _match(self, **lookups): + return readable_tag_match(self.location_field, **lookups) + + def _apply(self, qs, match): + return qs.exclude(match) if self.exclude else qs.filter(match) + + +class ReadableTagFilter(_ReadableTagFilterMixin, CharFilter): + def filter(self, qs, value): + if value in EMPTY_VALUES: + return qs + return self._apply(qs, self._match(**{f"tags__name__{self.lookup_expr}": value})) + + +class ReadableTagInFilter(_ReadableTagFilterMixin, CharFieldInFilter): + def filter(self, qs, value): + names = _names(value) + if not names: + return qs + return self._apply(qs, self._match(tags__name__in=names)) + + +class ReadableTagANDFilter(ReadableTagInFilter): + def filter(self, qs, value): + for name in _names(value): + qs = qs.filter(self._match(tags__name=name)) + return qs + + +class ReadableHasTagsFilter(_ReadableTagFilterMixin, BooleanFilter): + def filter(self, qs, value): + if value in EMPTY_VALUES: + return qs + match = self._match(tags__isnull=False) + return qs.filter(match) if value else qs.exclude(match) + + +def _names(value): + if not value: + return [] + if isinstance(value, str): + value = value.split(",") + return [name.strip() for name in value if name and name.strip()] + + +def create_readable_tag_filters(help_text_header, context, *, location_field="pk"): + """Drop-in replacement for ``create_char_filters`` on a Location tag relation.""" + def char(lookup, label, *, exclude=False): + return ReadableTagFilter( + lookup_expr=lookup, + exclude=exclude, + location_field=location_field, + help_text=f"{help_text_header}: {label}", + ) + + def in_list(label, *, exclude=False): + return ReadableTagInFilter( + exclude=exclude, + location_field=location_field, + help_text=f"{help_text_header}: {label}", + ) + + return StaticMethodFilters.set_class_variables( + context, + { + "tags__name_exact": char("iexact", "Exact Match"), + "tags__name_not_exact": char("iexact", "Not Exact Match", exclude=True), + "tags__name_contains": char("icontains", "Contains"), + "tags__name_not_contains": char("icontains", "Not Contains", exclude=True), + "tags__name_starts_with": char("istartswith", "Starts With"), + "tags__name_ends_with": char("iendswith", "Ends With"), + "tags__name_includes": in_list("Included in List"), + "tags__name_not_includes": in_list("Not Included in List", exclude=True), + }, + ) diff --git a/dojo/location/models.py b/dojo/location/models.py index c20b6eaa34a..99fc308886f 100644 --- a/dojo/location/models.py +++ b/dojo/location/models.py @@ -86,6 +86,23 @@ class Location(BaseModel): def __str__(self): return self.location_value + @property + def readable_tags(self): + """Tags on this row, or none of them when another product also labels the row.""" + # ponytail: one EXISTS per rendered row; annotate the list querysets if it shows up. + from dojo.location.queries import location_tags_readable # noqa: PLC0415 + if not location_tags_readable(self): + return [] + return list(self.tags.all()) + + @property + def readable_inherited_tags(self): + """The inherited subset of :attr:`readable_tags`.""" + from dojo.location.queries import location_tags_readable # noqa: PLC0415 + if not location_tags_readable(self): + return [] + return list(self.inherited_tags.all()) + def status_from_finding(self, finding: Finding) -> str: """Determine the status the reference should carry based on the status of the finding""" # Set the default status to Active to be on the safe side diff --git a/dojo/location/queries.py b/dojo/location/queries.py index b6c30660255..4a36f38d817 100644 --- a/dojo/location/queries.py +++ b/dojo/location/queries.py @@ -112,6 +112,46 @@ def locations_shared_outside(locations, products): return locations.filter(Exists(foreign_products) | Exists(foreign_findings)) +def readable_tag_locations(user=None): + """ + Locations whose tag set is entirely the caller's to read. + + A Location row is deduplicated globally and its tag set is one field shared by every + product referencing it, with no record of which product contributed which tag. So the + set is only the caller's to read when they are authorized for every product on the row. + """ + products = get_authorized_products(Permissions.Product_View, user=user) + return Location.objects.exclude( + Exists( + LocationProductReference.objects.filter( + location=OuterRef("pk"), + ).exclude(product__in=products), + ) + | Exists( + LocationFindingReference.objects.filter( + location=OuterRef("pk"), + ).exclude(finding__test__engagement__product__in=products), + ), + ) + + +def location_tags_readable(location, user=None): + """Whether the caller may read the shared tag set on ``location``.""" + return readable_tag_locations(user).filter(pk=location.pk).exists() + + +def readable_tag_match(location_field, user=None, **lookups): + """ + ``Exists`` over readable tag sets, for filtering without joining the tag relation. + + Use this rather than a joined ``filter()``: the host view runs ``distinct("url__host")``, + which a bare ``.distinct()`` added to deduplicate a join would clear. + """ + return Exists( + readable_tag_locations(user).filter(pk=OuterRef(location_field), **lookups), + ) + + def annotate_location_counts_and_status(locations, user=None): # Annotate the queryset with counts of findings # This aggregates the total and active findings by joining LocationFindingReference. diff --git a/dojo/templates/dojo/snippets/endpoints.html b/dojo/templates/dojo/snippets/endpoints.html index 88045dc812a..8778fe024a5 100644 --- a/dojo/templates/dojo/snippets/endpoints.html +++ b/dojo/templates/dojo/snippets/endpoints.html @@ -182,7 +182,7 @@

Vulnerable Endpoints / Systems ({{ finding.active_endpoint_count }}) {% if V3_FEATURE_LOCATIONS %} {{ endpoint.location|url_shortener }}{% if endpoint.is_broken %} 🚩{% endif %} - {% include "dojo/snippets/tags.html" with tags=endpoint.location.tags.all %} + {% include "dojo/snippets/tags.html" with tags=endpoint.location.readable_tags %} {{ endpoint.status }} {{ endpoint.created|date }} @@ -253,7 +253,7 @@

Mitigated Endpoints / Systems ({{ finding.mitigated_endpoint_count }}) {% if V3_FEATURE_LOCATIONS %} {{ endpoint.location|url_shortener }}{% if endpoint.is_broken %} 🚩{% endif %} - {% include "dojo/snippets/tags.html" with tags=endpoint.location.tags.all %} + {% include "dojo/snippets/tags.html" with tags=endpoint.location.readable_tags %} {{ endpoint.get_status_display }} {{ endpoint.auditor }} diff --git a/dojo/templates/dojo/url/list.html b/dojo/templates/dojo/url/list.html index 8e3651a39b5..80273272499 100644 --- a/dojo/templates/dojo/url/list.html +++ b/dojo/templates/dojo/url/list.html @@ -126,7 +126,7 @@

title="Endpoint is broken. Check documentation to look for fix process">🚩 {% endif %} - {% include "dojo/snippets/tags.html" with tags=location.tags.all %} + {% include "dojo/snippets/tags.html" with tags=location.readable_tags %} {% endif %} {% if not product_tab %} diff --git a/dojo/templates/dojo/url/view.html b/dojo/templates/dojo/url/view.html index 03fb6ed3d33..f7f533a4219 100644 --- a/dojo/templates/dojo/url/view.html +++ b/dojo/templates/dojo/url/view.html @@ -213,12 +213,12 @@

Host

{% endif %} - {% if not host_view and location.tags.exists %} + {% if not host_view and location.readable_tags %}

Tags

-
{% include "dojo/snippets/tags.html" with tags=location.tags.all %}
+
{% include "dojo/snippets/tags.html" with tags=location.readable_tags %}
{% endif %} {% if not host_view and metadata %} diff --git a/dojo/url/ui/forms.py b/dojo/url/ui/forms.py index 753c3d2da40..fcf14b220c8 100644 --- a/dojo/url/ui/forms.py +++ b/dojo/url/ui/forms.py @@ -25,7 +25,7 @@ class Meta: def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) if self.instance is not None and hasattr(self.instance, "location"): - self.fields["tags"].initial = self.instance.location.tags.all() + self.fields["tags"].initial = self.instance.location.readable_tags def clean_tags(self): tag_validator(self.cleaned_data.get("tags")) diff --git a/unittests/test_location_tag_scoping.py b/unittests/test_location_tag_scoping.py new file mode 100644 index 00000000000..f2edcf6297a --- /dev/null +++ b/unittests/test_location_tag_scoping.py @@ -0,0 +1,196 @@ +from rest_framework.authtoken.models import Token +from rest_framework.test import APIClient + +from dojo.authorization.roles_permissions import Roles +from dojo.location.status import ProductLocationStatus +from dojo.models import ( + Dojo_User, + Product, + Product_Member, + Product_Type, + Role, + System_Settings, + User, +) +from dojo.tags.inheritance import apply_inherited_tags_for_locations +from dojo.url.models import URL +from unittests.dojo_test_case import DojoTestCase, skip_unless_v3 + +FOREIGN = "tagscope-foreign.example.test" +OWN = "tagscope-own.example.test" +BOTH_MINE = "tagscope-both-mine.example.test" +INHERIT = "tagscope-inherit.example.test" + +FOREIGN_TAG = "foreignsecret" +OWN_TAG = "myowntag" +INHERITED_TAG = "inheritedsecret" + + +@skip_unless_v3 +class LocationTagScopingTest(DojoTestCase): + + """ + A Location row is deduplicated globally and carries one tag set for every product that + references it. The tag set must only be served to a caller authorized for every product + on the row, and the tag filters must not match on a set the body withholds. + """ + + @classmethod + def setUpTestData(cls): + prod_type, _ = Product_Type.objects.get_or_create(name="TagScope PT") + reader = Role.objects.get(id=Roles.Reader) + + def product(name, **kwargs): + return Product.objects.create(name=name, description=name, prod_type=prod_type, **kwargs) + + cls.mine = product("TagScope Mine") + cls.also_mine = product("TagScope Also Mine") + cls.theirs = product("TagScope Theirs") + cls.theirs_inheriting = product( + "TagScope Theirs Inheriting", enable_product_tag_inheritance=True, + ) + cls.theirs_inheriting.tags.set([INHERITED_TAG]) + + cls.alice = User.objects.create_user( + username="tagscope_alice", + password="not-a-real-secret", # noqa: S106 - test fixture user + ) + alice = Dojo_User.objects.get(pk=cls.alice.pk) + for owned in (cls.mine, cls.also_mine): + Product_Member.objects.create(user=cls.alice, product=owned, role=reader) + owned.authorized_users.add(alice) + + def location(host, *products): + loc = URL.get_or_create_from_values(protocol="https", host=host, path="x").location + for prod in products: + loc.associate_with_product(prod, status=ProductLocationStatus.Active) + return loc + + # Shared with a product Alice cannot see: the tag set is not hers to read. + cls.foreign = location(FOREIGN, cls.mine, cls.theirs) + cls.foreign.tags.set([FOREIGN_TAG]) + + # Only Alice's product references it. + cls.own = location(OWN, cls.mine) + cls.own.tags.set([OWN_TAG]) + + # Shared, but only between two products Alice holds: still hers to read. + cls.both_mine = location(BOTH_MINE, cls.mine, cls.also_mine) + cls.both_mine.tags.set([OWN_TAG]) + + # Product tag inheritance writes the union of contributing products' tags here. + cls.inherit = location(INHERIT, cls.mine, cls.theirs_inheriting) + apply_inherited_tags_for_locations([cls.inherit], product=cls.theirs_inheriting) + + cls.admin = User.objects.create_superuser( + username="tagscope_admin", + email="tagscope_admin@example.test", + password="not-a-real-secret", # noqa: S106 - test fixture user + ) + cls.alice_token = Token.objects.get_or_create(user=cls.alice)[0].key + cls.admin_token = Token.objects.get_or_create(user=cls.admin)[0].key + + def setUp(self): + super().setUp() + settings = System_Settings.objects.get(no_cache=True) + settings.enable_product_tag_inheritance = False + settings.save() + + def _client(self, token): + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Token {token}") + return client + + def _get(self, url, token=None): + response = self._client(token or self.alice_token).get(url, format="json") + self.assertEqual(response.status_code, 200, response.content[:300]) + return response.json() + + def _endpoint_tags(self, token=None): + return { + row["host"]: row["tags"] + for row in self._get("/api/v2/endpoints/", token)["results"] + } + + def _location_row(self, host, token=None): + return next( + row for row in self._get("/api/v2/location/", token)["results"] + if host in row["location_value"] + ) + + def _hosts(self, query, token=None): + return {row["host"] for row in self._get(f"/api/v2/endpoints/?{query}", token)["results"]} + + def test_premise_each_url_is_one_shared_row(self): + self.assertEqual({p.name for p in self.foreign.all_related_products()}, + {self.mine.name, self.theirs.name}) + names = {p["name"] for p in self._get("/api/v2/products/")["results"]} + self.assertNotIn(self.theirs.name, names) + + def test_endpoint_body_withholds_a_foreign_products_tags(self): + self.assertEqual(self._endpoint_tags()[FOREIGN], []) + + def test_endpoint_body_withholds_inherited_foreign_product_tags(self): + self.assertEqual(self._endpoint_tags()[INHERIT], []) + + def test_endpoint_body_still_serves_an_unshared_rows_tags(self): + self.assertEqual(self._endpoint_tags()[OWN], [OWN_TAG]) + + def test_endpoint_body_still_serves_a_row_shared_only_within_my_products(self): + self.assertEqual(self._endpoint_tags()[BOTH_MINE], [OWN_TAG]) + + def test_endpoint_body_still_complete_for_a_superuser(self): + self.assertEqual(self._endpoint_tags(self.admin_token)[FOREIGN], [FOREIGN_TAG]) + + def test_endpoint_filters_do_not_match_a_foreign_tag(self): + for query in ( + f"tag={FOREIGN_TAG[:12]}", + f"tags={FOREIGN_TAG}", + f"tags__and={FOREIGN_TAG}", + f"tag={INHERITED_TAG[:12]}", + ): + self.assertEqual(self._hosts(query), set(), query) + + def test_endpoint_negated_filters_treat_a_foreign_tag_set_as_empty(self): + self.assertIn(FOREIGN, self._hosts(f"not_tag={FOREIGN_TAG[:12]}")) + self.assertNotIn(FOREIGN, self._hosts("has_tags=true")) + + def test_endpoint_filters_still_match_my_own_tags(self): + self.assertEqual(self._hosts(f"tag={OWN_TAG}"), {OWN, BOTH_MINE}) + self.assertEqual(self._hosts(f"tags={OWN_TAG}"), {OWN, BOTH_MINE}) + self.assertEqual(self._hosts(f"tags__and={OWN_TAG}"), {OWN, BOTH_MINE}) + self.assertEqual(self._hosts("has_tags=true"), {OWN, BOTH_MINE}) + + def test_location_body_withholds_tags_and_inherited_tags(self): + foreign = self._location_row(FOREIGN) + self.assertEqual(foreign["tags"], []) + inherit = self._location_row(INHERIT) + self.assertEqual(inherit["tags"], []) + self.assertEqual(inherit["inherited_tags"], []) + + def test_location_body_still_serves_my_own_tags(self): + self.assertEqual(self._location_row(OWN)["tags"], [OWN_TAG]) + + def test_location_body_still_complete_for_a_superuser(self): + self.assertEqual( + self._location_row(INHERIT, self.admin_token)["inherited_tags"], [INHERITED_TAG], + ) + + def test_location_filters_do_not_match_a_foreign_tag(self): + for query in ( + f"tags__name_exact={FOREIGN_TAG}", + f"tags__name_contains={FOREIGN_TAG[:12]}", + f"tags__name_starts_with={FOREIGN_TAG[:9]}", + f"tags__name_ends_with={FOREIGN_TAG[-4:]}", + f"tags__name_includes={FOREIGN_TAG}", + f"tags__name_exact={INHERITED_TAG}", + ): + self.assertEqual(self._get(f"/api/v2/location/?{query}")["count"], 0, query) + + def test_location_filters_still_match_my_own_tags(self): + for query in ( + f"tags__name_exact={OWN_TAG}", + f"tags__name_contains={OWN_TAG[:12]}", + f"tags__name_includes={OWN_TAG}", + ): + self.assertEqual(self._get(f"/api/v2/location/?{query}")["count"], 2, query)