Timestamp validator

ckan/logic/schema.py

@@ -644,7 +644,8 @@ def default_activity_list_schema(
         natural_number_validator: Validator,
         limit_to_configured_maximum: ValidatorFactory,
         ignore_missing: Validator, boolean_validator: Validator,
-        ignore_not_sysadmin: Validator, list_of_strings: Validator):
+        ignore_not_sysadmin: Validator, list_of_strings: Validator,
+        datetime_from_timestamp_validator: Validator):
 
     schema = default_pagination_schema()
     schema['id'] = [not_missing, unicode_safe]
@@ -656,8 +657,8 @@ def default_activity_list_schema(
         ignore_missing, ignore_not_sysadmin, boolean_validator]
     schema['activity_types'] = [ignore_missing, list_of_strings]
     schema['exclude_activity_types'] = [ignore_missing, list_of_strings]
-    schema['before'] = [ignore_missing]
-    schema['after'] = [ignore_missing]
+    schema['before'] = [ignore_missing, datetime_from_timestamp_validator]
+    schema['after'] = [ignore_missing, datetime_from_timestamp_validator]
     return schema
 
 

ckan/logic/validators.py

@@ -134,6 +134,15 @@ def is_positive_integer(value: Any, context: Context) -> Any:
         raise Invalid(_('Must be a positive integer'))
     return value
 
+def datetime_from_timestamp_validator(value: Any, context: Context) -> Any:
+    if value is missing or value is None:
+        return None
+    try:
+        value = datetime.datetime.fromtimestamp(float(value))
+    except (TypeError, ValueError):
+        raise Invalid(_('Must be a float timestamp'))
+    return value
+
 def boolean_validator(value: Any, context: Context) -> Any:
     '''
     Return a boolean for value.

ckan/tests/controllers/test_package.py

@@ -2052,6 +2052,20 @@ def test_create_dataset(self, app):
         assert dataset["id"] in href.select_one("a")["href"].split("/", 2)[-1]
         assert dataset["title"] in href.text.strip()
 
+    @pytest.mark.ckan_config("ckan.activity_list_limit", "3")
+    def test_invalid_get_params(self, app):
+
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+
+        url = url_for("dataset.activity", id=dataset["id"])
+        response = app.get(
+            url,
+            query_string={'before': 'XXX'},
+            status=400
+        )
+        assert 'Invalid parameters' in response.body
+
     @pytest.mark.ckan_config("ckan.activity_list_limit", "3")
     def test_next_page_button(self, app):
 

ckan/views/dataset.py

@@ -1109,11 +1109,6 @@ def activity(
     before = h.get_request_param('before')
     activity_type = h.get_request_param('activity_type')
 
-    if after and before:
-        raise ValidationError(
-            {'after': ['Cannot be used together with `before']}
-        )
-
     context = cast(Context, {
         u'model': model,
         u'session': model.Session,
@@ -1126,10 +1121,6 @@ def activity(
     max_limit = config.get_value('ckan.activity_list_limit_max')
     limit = min(base_limit, max_limit)
     activity_types = [activity_type] if activity_type else None
-    if after is not None:
-        after = datetime.fromtimestamp(float(after))
-    if before is not None:
-        before = datetime.fromtimestamp(float(before))
     is_first_page = after is None and before is None
 
     try:
@@ -1150,9 +1141,11 @@ def activity(
         )
         dataset_type = pkg_dict[u'type'] or u'dataset'
     except NotFound:
-        return base.abort(404, _(u'Dataset not found'))
+        return base.abort(404, _('Dataset not found'))
     except NotAuthorized:
-        return base.abort(403, _(u'Unauthorized to read dataset %s') % id)
+        return base.abort(403, _('Unauthorized to read dataset %s') % id)
+    except ValidationError:
+        return base.abort(400, _('Invalid parameters'))
 
     prev_page = None
     next_page = None