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)