Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 44 additions & 15 deletions dojo/api_helpers/filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ def create_char_filters(
field_name: str,
help_text_header: str,
context: dict,
model_field_name: str | None = None,
) -> None:
"""
Create all the filters needed for a CharFilter.
Expand All @@ -62,49 +63,55 @@ def create_char_filters(
- Not Contains
- Starts with
- Ends with

``field_name`` is the public query-parameter prefix. ``model_field_name``
is the ORM field the filter resolves against; it defaults to ``field_name``
and only needs to be supplied when the two differ (e.g. a public
``created_at`` parameter backed by the model's ``created`` field).
"""
model_field = model_field_name or field_name
return StaticMethodFilters.set_class_variables(
context,
{
f"{field_name}_exact": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="iexact",
help_text=f"{help_text_header}: Exact Match",
),
f"{field_name}_not_exact": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="iexact",
help_text=f"{help_text_header}: Not Exact Match",
exclude=True,
),
f"{field_name}_contains": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="icontains",
help_text=f"{help_text_header}: Contains",
),
f"{field_name}_not_contains": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="icontains",
help_text=f"{help_text_header}: Not Contains",
exclude=True,
),
f"{field_name}_starts_with": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="istartswith",
help_text=f"{help_text_header}: Starts With",
),
f"{field_name}_ends_with": CharFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="iendswith",
help_text=f"{help_text_header}: Ends With",
),
f"{field_name}_includes": CharFieldInFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="in",
help_text=f"{help_text_header}: Included in List",
),
f"{field_name}_not_includes": CharFieldInFilter(
field_name=field_name,
field_name=model_field,
lookup_expr="in",
help_text=f"{help_text_header}: Not Included in List",
exclude=True,
Expand Down Expand Up @@ -190,13 +197,22 @@ def create_datetime_filters(
field_name: str,
help_text_header: str,
context: dict,
model_field_name: str | None = None,
) -> None:
"""Create a filter for setting datetime filters."""
"""
Create a filter for setting datetime filters.

``field_name`` is the public query-parameter name (it produces
``<field_name>_after`` / ``<field_name>_before``). ``model_field_name``
is the ORM field the range resolves against; it defaults to ``field_name``
and only needs to be supplied when the two differ (e.g. a public
``created_at`` parameter backed by the model's ``created`` field).
"""
return StaticMethodFilters.set_class_variables(
context,
{
field_name: DateTimeFromToRangeFilter(
field_name=field_name,
field_name=model_field_name or field_name,
help_text=f"{help_text_header}: DateTime Range Filter",
),
},
Expand All @@ -222,22 +238,35 @@ def create_boolean_filters(
@staticmethod
def create_ordering_filters(
context: dict,
field_names: Iterable[str],
field_names: Iterable[str | tuple[str, str]],
) -> None:
"""Create an ordering filter for all fields in the dict."""
"""
Create an ordering filter for all fields in the dict.

Each entry is either a string (the ORM field, exposed under the same
public name) or a ``(model_field, public_name)`` tuple for cases where
the query parameter differs from the ORM field (e.g. a public
``created_at`` ordering key backed by the model's ``created`` field).
"""
fields = [
field_name if isinstance(field_name, tuple) else (field_name, field_name)
for field_name in field_names
]
return StaticMethodFilters.set_class_variables(
context,
{"ordering": OrderingFilter(fields=[(field_name, field_name) for field_name in field_names])},
{"ordering": OrderingFilter(fields=fields)},
)


class CommonFilters(StaticMethodFilters):

"""Helpers for FilterSets to reduce copy/past code."""

# The public ``created_at``/``updated_at`` parameters resolve to the
# ``created``/``updated`` fields that BaseModel actually defines.
StaticMethodFilters.create_integer_filters("id", "ID", locals())
StaticMethodFilters.create_datetime_filters("created_at", "Created At", locals())
StaticMethodFilters.create_datetime_filters("updated_at", "Updated At", locals())
StaticMethodFilters.create_datetime_filters("created_at", "Created At", locals(), model_field_name="created")
StaticMethodFilters.create_datetime_filters("updated_at", "Updated At", locals(), model_field_name="updated")


def filter_timestamp(queryset, name, value):
Expand Down
18 changes: 10 additions & 8 deletions dojo/location/api/filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,10 @@
class AbstractedLocationFilter(StaticMethodFilters):
StaticMethodFilters.create_integer_filters("id", "ID", locals())
StaticMethodFilters.create_char_filters("location__tags__name", "Tags", locals())
StaticMethodFilters.create_char_filters("location__created_at", "Created At", locals())
StaticMethodFilters.create_char_filters("location__updated_at", "Updated At", locals())
StaticMethodFilters.create_char_filters(
"location__created_at", "Created At", locals(), model_field_name="location__created")
StaticMethodFilters.create_char_filters(
"location__updated_at", "Updated At", locals(), model_field_name="location__updated")
StaticMethodFilters.create_integer_filters("location__products__product", "Product ID", locals())
StaticMethodFilters.create_integer_filters("location__findings__finding", "Finding ID", locals())

Expand All @@ -38,8 +40,8 @@ class LocationFilter(CommonFilters):
"id",
"location_type",
"location_value",
"created_at",
"updated_at",
("created", "created_at"),
("updated", "updated_at"),
),
)

Expand All @@ -60,8 +62,8 @@ class LocationProductReferenceFilter(CommonFilters):
"product",
"product__name",
"status",
"created_at",
"updated_at",
("created", "created_at"),
("updated", "updated_at"),
),
)

Expand All @@ -82,7 +84,7 @@ class LocationFindingReferenceFilter(CommonFilters):
"finding",
"finding__severity",
"status",
"created_at",
"updated_at",
("created", "created_at"),
("updated", "updated_at"),
),
)
159 changes: 159 additions & 0 deletions unittests/test_location_filter_timestamp.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
from django.utils.timezone import now, timedelta

from dojo.location.api.filters import (
LocationFilter,
LocationFindingReferenceFilter,
LocationProductReferenceFilter,
)
from dojo.location.models import (
Location,
LocationFindingReference,
LocationProductReference,
)
from dojo.models import (
Engagement,
Finding,
Product,
Product_Type,
Test,
Test_Type,
)
from dojo.url.api.filters import URLFilter
from dojo.url.models import URL
from unittests.dojo_test_case import DojoTestCase, skip_unless_v3, versioned_fixtures

HOST = "timestamp-filter.example.com"


@skip_unless_v3
@versioned_fixtures
class TestLocationFilterTimestamps(DojoTestCase):

"""
The Location filtersets expose ``created_at``/``updated_at`` query parameters
(range filters and ordering keys), but the underlying models store those
timestamps in the ``created``/``updated`` fields inherited from BaseModel.

A request such as ``GET /api/v2/location/?created_at_after=...`` therefore
raised ``FieldError: Cannot resolve keyword 'created_at' into field`` and
returned HTTP 500 to the customer. These tests reproduce that path against
every affected filterset and assert the parameters resolve to the real
model fields.
"""

fixtures = ["dojo_testdata.json"]

@classmethod
def setUpTestData(cls):
prod_type = Product_Type.objects.create(name="Timestamp PT")
cls.product = Product.objects.create(
name="Timestamp Product", description="ts", prod_type=prod_type,
)
engagement = Engagement.objects.create(
product=cls.product, name="ts eng",
target_start=now().date(), target_end=now().date(),
)
test = Test.objects.create(
engagement=engagement,
test_type=Test_Type.objects.create(name="Timestamp Scan"),
target_start=now(), target_end=now(),
)
cls.finding = Finding.objects.create(
test=test, title="Timestamp Finding", severity="High",
numerical_severity="S1", active=True, verified=True,
)
cls.location = URL.get_or_create_from_values(
protocol="https", host=HOST, path="app").location
cls.location.associate_with_product(cls.product)
cls.location.associate_with_finding(cls.finding, audit_time=now())

def test_location_created_at_range_resolves_to_created_field(self):
"""`/api/v2/location/?created_at_after=...` must not raise FieldError."""
past = (now() - timedelta(days=1)).isoformat()
future = (now() + timedelta(days=1)).isoformat()
matched = set(
LocationFilter(
{"created_at_after": past, "created_at_before": future},
queryset=Location.objects.filter(id=self.location.id),
).qs.values_list("id", flat=True),
)
self.assertEqual(matched, {self.location.id})

excluded = set(
LocationFilter(
{"created_at_before": past},
queryset=Location.objects.filter(id=self.location.id),
).qs.values_list("id", flat=True),
)
self.assertEqual(excluded, set())

def test_location_updated_at_range_resolves_to_updated_field(self):
future = (now() + timedelta(days=1)).isoformat()
past = (now() - timedelta(days=1)).isoformat()
matched = set(
LocationFilter(
{"updated_at_after": past, "updated_at_before": future},
queryset=Location.objects.filter(id=self.location.id),
).qs.values_list("id", flat=True),
)
self.assertEqual(matched, {self.location.id})

def test_location_ordering_by_timestamps_does_not_raise(self):
for ordering in ("created_at", "-created_at", "updated_at", "-updated_at"):
ids = list(
LocationFilter(
{"ordering": ordering},
queryset=Location.objects.filter(id=self.location.id),
).qs.values_list("id", flat=True),
)
self.assertEqual(ids, [self.location.id])

def test_location_findings_created_at_range_resolves(self):
"""`/api/v2/location_findings/?created_at_after=...` must not raise."""
past = (now() - timedelta(days=1)).isoformat()
future = (now() + timedelta(days=1)).isoformat()
qs = LocationFindingReference.objects.filter(location=self.location)
matched = set(
LocationFindingReferenceFilter(
{"created_at_after": past, "created_at_before": future},
queryset=qs,
).qs.values_list("id", flat=True),
)
self.assertEqual(matched, set(qs.values_list("id", flat=True)))
self.assertTrue(matched)

def test_location_findings_ordering_by_timestamps_does_not_raise(self):
qs = LocationFindingReference.objects.filter(location=self.location)
for ordering in ("created_at", "-updated_at"):
evaluated = list(
LocationFindingReferenceFilter({"ordering": ordering}, queryset=qs).qs,
)
self.assertEqual(len(evaluated), qs.count())

def test_location_products_created_at_range_resolves(self):
past = (now() - timedelta(days=1)).isoformat()
future = (now() + timedelta(days=1)).isoformat()
qs = LocationProductReference.objects.filter(location=self.location)
matched = set(
LocationProductReferenceFilter(
{"created_at_after": past, "created_at_before": future},
queryset=qs,
).qs.values_list("id", flat=True),
)
self.assertEqual(matched, set(qs.values_list("id", flat=True)))
self.assertTrue(matched)

def test_location_products_ordering_by_timestamps_does_not_raise(self):
qs = LocationProductReference.objects.filter(location=self.location)
for ordering in ("updated_at", "-created_at"):
evaluated = list(
LocationProductReferenceFilter({"ordering": ordering}, queryset=qs).qs,
)
self.assertEqual(len(evaluated), qs.count())

def test_url_location_timestamp_char_filters_do_not_raise(self):
"""`/api/v2/url/` inherits location__created_at/location__updated_at char filters."""
qs = URL.objects.filter(host=HOST)
for param in ("location__created_at_contains", "location__updated_at_contains"):
evaluated = list(URLFilter({param: str(now().year)}, queryset=qs).qs)
self.assertEqual(len(evaluated), 1)
Loading