diff --git a/dojo/url/filters.py b/dojo/url/filters.py index ca7132aa92..b9ea7ce8a0 100644 --- a/dojo/url/filters.py +++ b/dojo/url/filters.py @@ -1,22 +1,34 @@ import logging +from django.core.validators import EMPTY_VALUES +from django.db.models import Exists, OuterRef from django.forms import HiddenInput from django_filters import ( NumberFilter, ) from dojo.api_helpers.filters import StaticMethodFilters +from dojo.authorization.roles_permissions import Permissions # from tagulous.forms import TagWidget # import tagulous +from dojo.location.models import LocationFindingReference, LocationProductReference from dojo.location.queries import get_authorized_locations from dojo.location.status import FindingLocationStatus, ProductLocationStatus +from dojo.product.queries import get_authorized_products logger = logging.getLogger(__name__) BOOLEAN_CHOICES = (("false", "No"), ("true", "Yes")) EARLIEST_FINDING = None +# Relations that leave the Location and reach another product's rows, mapped to the +# reference model that carries them and the path from that model to its product. +OUTWARD_RELATIONS = { + "products": (LocationProductReference, "product__in"), + "findings": (LocationFindingReference, "finding__test__engagement__product__in"), +} + class URLFilter(StaticMethodFilters): StaticMethodFilters.create_char_filters("url__protocol", "Protocol", locals()) @@ -56,6 +68,39 @@ def __init__(self, *args, **kwargs): self.user = kwargs.pop("user") super().__init__(*args, **kwargs) + def filter_queryset(self, queryset): + """ + Match each predicate against the caller's own references only. + + A Location is shared by every product that references it, so a predicate that + joins outward can be satisfied by a reference the caller cannot see. Narrowing + the result afterwards does not help: that is a second, independent join, and the + row still qualifies through its own product. + """ + authorized_products = get_authorized_products(Permissions.Product_View, self.user) + for name, value in self.form.cleaned_data.items(): + declared = self.filters[name] + relation, _, remainder = (declared.field_name or "").partition("__") + outward = OUTWARD_RELATIONS.get(relation) + if outward is None or value in EMPTY_VALUES: + queryset = declared.filter(queryset, value) + continue + reference_model, product_path = outward + lookup = "in" if isinstance(value, list | tuple) else declared.lookup_expr + matching_references = reference_model.objects.filter( + **{ + "location": OuterRef("pk"), + f"{remainder}__{lookup}" if remainder else lookup: value, + product_path: authorized_products, + }, + ) + queryset = ( + queryset.exclude(Exists(matching_references)) + if declared.exclude + else queryset.filter(Exists(matching_references)) + ) + return queryset + @property def qs(self): parent = super().qs diff --git a/unittests/test_location_filter_join_scoping.py b/unittests/test_location_filter_join_scoping.py new file mode 100644 index 0000000000..0d24f2ab91 --- /dev/null +++ b/unittests/test_location_filter_join_scoping.py @@ -0,0 +1,184 @@ +from crum import impersonate +from django.utils.timezone import now + +from dojo.authorization.roles_permissions import Roles +from dojo.location.models import Location, LocationProductReference +from dojo.location.queries import annotate_location_counts_and_status, get_authorized_locations +from dojo.location.status import ProductLocationStatus +from dojo.models import ( + Dojo_User, + Engagement, + Finding, + Product, + Product_Member, + Product_Type, + Role, + Test, + Test_Type, + User, +) +from dojo.url.filters import URLFilter +from dojo.url.models import URL +from unittests.dojo_test_case import DojoTestCase, skip_unless_v3, versioned_fixtures + +SHARED_HOST = "joinscope-shared.example.com" +OWN_HOST = "joinscope-own.example.com" + +VICTIM_PRODUCT_NAME = "JoinScope Product B" +VICTIM_PRODUCT_TAG = "joinscope-b-product-tag" +VICTIM_FINDING_TAG = "joinscope-b-finding-tag" +OWN_PRODUCT_NAME = "JoinScope Product A" +OWN_PRODUCT_TAG = "joinscope-a-product-tag" +OWN_FINDING_TAG = "joinscope-a-finding-tag" + + +@skip_unless_v3 +@versioned_fixtures +class TestLocationFilterJoinScoping(DojoTestCase): + + """ + Two products sharing one deduplicated Location. The endpoint list filters join + outward from the Location into products and findings, so a predicate must only + ever be matched against references the caller is authorized for. Otherwise the + shared row acts as an oracle for the other product's data. + """ + + fixtures = ["dojo_testdata.json"] + + @classmethod + def setUpTestData(cls): + prod_type, _ = Product_Type.objects.get_or_create(name="JoinScope PT") + test_type, _ = Test_Type.objects.get_or_create(name="JoinScope Scan") + + def build(name, product_tag, finding_title, finding_tag): + product = Product.objects.create(name=name, description=name, prod_type=prod_type) + product.tags = [product_tag] + product.save() + engagement = Engagement.objects.create( + product=product, name=f"{name} 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=finding_title, severity="High", + numerical_severity="S1", active=True, verified=True, + ) + finding.tags = [finding_tag] + finding.save() + return product, finding + + cls.product_a, finding_a = build( + OWN_PRODUCT_NAME, OWN_PRODUCT_TAG, "JoinScope A Finding", OWN_FINDING_TAG) + cls.product_b, finding_b = build( + VICTIM_PRODUCT_NAME, VICTIM_PRODUCT_TAG, "JoinScope B Finding", VICTIM_FINDING_TAG) + + cls.alice = User.objects.create_user( + username="joinscope_alice", + password="not-a-real-secret", # noqa: S106 - test fixture user + ) + # Legacy authorization reads authorized_users, the role model reads Product_Member. + cls.product_a.authorized_users.add(Dojo_User.objects.get(pk=cls.alice.pk)) + Product_Member.objects.create( + product=cls.product_a, user=cls.alice, role=Role.objects.get(id=Roles.Reader)) + + # One URL referenced by both products dedupes to a single Location row. Only + # product B has a finding on it, so any match through a finding is B's. + cls.shared = URL.get_or_create_from_values( + protocol="https", host=SHARED_HOST, path="app").location + cls.shared.associate_with_product(cls.product_a) + cls.shared.associate_with_finding(finding_b, audit_time=now()) + + # Control: a Location only product A references. + cls.own = URL.get_or_create_from_values( + protocol="https", host=OWN_HOST, path="x").location + cls.own.associate_with_product(cls.product_a) + cls.own.associate_with_finding(finding_a, audit_time=now()) + + # Distinct statuses per product, so a status predicate that matches can only + # have matched one of them. + LocationProductReference.objects.filter(product=cls.product_a).update( + status=ProductLocationStatus.Active) + LocationProductReference.objects.filter(location=cls.shared, product=cls.product_b).update( + status=ProductLocationStatus.Mitigated) + + cls.admin = User.objects.filter(is_superuser=True).first() + + def _matches(self, user, **params): + """Location ids the endpoint list returns for these filter parameters.""" + with impersonate(user): + base = annotate_location_counts_and_status( + get_authorized_locations( + "view", + Location.objects.filter(id__in=[self.shared.id, self.own.id]), + user, + ), + user=user, + ) + return set(URLFilter(params, queryset=base, user=user).qs.values_list("id", flat=True)) + + def test_shared_location_is_a_single_row(self): + """The premise: both products reference one deduplicated Location.""" + self.assertEqual(Location.objects.filter(url__host=SHARED_HOST).count(), 1) + related = {p.name for p in self.shared.all_related_products()} + self.assertEqual(related, {OWN_PRODUCT_NAME, VICTIM_PRODUCT_NAME}) + self.assertEqual(self._matches(self.alice), {self.shared.id, self.own.id}) + + def test_other_products_data_never_matches(self): + for label, params in ( + ("finding tag, exact", {"findings__finding__tags__name_exact": VICTIM_FINDING_TAG}), + ("finding tag, prefix", {"findings__finding__tags__name_starts_with": "joinscope-b"}), + ("finding tag, substring", {"findings__finding__tags__name_contains": "b-finding"}), + ("finding tag, list", {"findings__finding__tags__name_includes": VICTIM_FINDING_TAG}), + ("product name, exact", {"products__product__name_exact": VICTIM_PRODUCT_NAME}), + ("product name, prefix", {"products__product__name_starts_with": VICTIM_PRODUCT_NAME}), + ("product name, substring", {"products__product__name_contains": "Product B"}), + ("product tag, exact", {"products__product__tags__name_exact": VICTIM_PRODUCT_TAG}), + ("product tag, prefix", {"products__product__tags__name_starts_with": "joinscope-b"}), + ("product id", {"products__product__id_equals": self.product_b.id}), + ("product reference status", {"products__status_equals": [ProductLocationStatus.Mitigated]}), + ("finding reference status", {"findings__status_equals": ["Active"]}), + ): + with self.subTest(label): + self.assertNotIn( + self.shared.id, self._matches(self.alice, **params), + f"{label} matched through another product's reference", + ) + + def test_own_data_still_matches(self): + both = {self.shared.id, self.own.id} + for label, params, expected in ( + ("own product name", {"products__product__name_exact": OWN_PRODUCT_NAME}, both), + ("own product name prefix", {"products__product__name_starts_with": OWN_PRODUCT_NAME}, both), + ("own product tag", {"products__product__tags__name_exact": OWN_PRODUCT_TAG}, both), + ("own product id", {"products__product__id_equals": self.product_a.id}, both), + ("own finding tag", {"findings__finding__tags__name_exact": OWN_FINDING_TAG}, {self.own.id}), + ("own reference status", {"products__status_equals": [ProductLocationStatus.Active]}, both), + ): + with self.subTest(label): + self.assertEqual(self._matches(self.alice, **params), expected) + + def test_excluding_a_hidden_product_does_not_hide_the_callers_row(self): + """A negated predicate must not fire on a reference the caller cannot see.""" + self.assertEqual( + self._matches(self.alice, products__product__id_not_equals=self.product_b.id), + {self.shared.id, self.own.id}, + ) + + def test_superuser_still_sees_everything(self): + self.assertEqual( + self._matches(self.admin, products__product__name_exact=VICTIM_PRODUCT_NAME), + {self.shared.id}, + ) + self.assertEqual( + self._matches(self.admin, findings__finding__tags__name_exact=VICTIM_FINDING_TAG), + {self.shared.id}, + ) + + def test_url_filters_are_unaffected(self): + self.assertEqual( + self._matches(self.alice, url__host_exact=SHARED_HOST), {self.shared.id}) + self.assertEqual( + self._matches(self.alice, url__host_contains="joinscope-own"), {self.own.id})