Merge pull request #1951 from Kinto/1942-validate-record-id

Validate record ID (fixes #1942)

CHANGELOG.rst

@@ -21,6 +21,7 @@ This document describes changes between each past release.
 **Bug Fixes**
 
 - Like query now returns 400 when a non string value is used. (#1899)
+- Record ID is validated if explicitly mentioned in the collection schema (#1942)
 
 **Internal changes**
 

kinto/schema_validation.py

@@ -68,7 +68,11 @@ def validate(data, schema):
     return _schema_cache[cache_key].validate(data)
 
 
-def validate_schema(data, schema, ignore_fields=[]):
+def validate_schema(data, schema, id_field, ignore_fields=[]):
+    # Only ignore the `id` field if the schema does not explicitly mention it.
+    if id_field not in schema.get("properties", {}):
+        ignore_fields += (id_field,)
+
     required_fields = [f for f in schema.get("required", []) if f not in ignore_fields]
     # jsonschema doesn't accept 'required': [] yet.
     # See https://github.com/Julian/jsonschema/issues/337.
@@ -100,7 +104,7 @@ def validate_schema(data, schema, ignore_fields=[]):
         raise e
 
 
-def validate_from_bucket_schema_or_400(data, resource_name, request, ignore_fields=[]):
+def validate_from_bucket_schema_or_400(data, resource_name, request, id_field, ignore_fields=[]):
     """Lookup in the parent objects if a schema was defined for this resource.
 
     If the schema validation feature is enabled, if a schema is/are defined, and if the
@@ -131,7 +135,7 @@ def validate_from_bucket_schema_or_400(data, resource_name, request, ignore_fiel
     # Validate or fail with 400.
     schema = bucket[metadata_field]
     try:
-        validate_schema(data, schema, ignore_fields=ignore_fields)
+        validate_schema(data, schema, ignore_fields=ignore_fields, id_field=id_field)
     except ValidationError as e:
         raise_invalid(request, name=e.field, description=e.message)
     except RefResolutionError as e:

kinto/views/collections.py

@@ -30,13 +30,13 @@ def process_object(self, new, old=None):
         new = super().process_object(new, old)
 
         # Remove internal and auto-assigned fields.
-        internal_fields = (
-            self.model.id_field,
-            self.model.modified_field,
-            self.model.permissions_field,
-        )
+        internal_fields = (self.model.modified_field, self.model.permissions_field)
         validate_from_bucket_schema_or_400(
-            new, resource_name="collection", request=self.request, ignore_fields=internal_fields
+            new,
+            resource_name="collection",
+            request=self.request,
+            ignore_fields=internal_fields,
+            id_field=self.model.id_field,
         )
         return new
 

kinto/views/groups.py

@@ -37,13 +37,13 @@ def process_object(self, new, old=None):
         new = super().process_object(new, old)
 
         # Remove internal and auto-assigned fields.
-        internal_fields = (
-            self.model.id_field,
-            self.model.modified_field,
-            self.model.permissions_field,
-        )
+        internal_fields = (self.model.modified_field, self.model.permissions_field)
         validate_from_bucket_schema_or_400(
-            new, resource_name="group", request=self.request, ignore_fields=internal_fields
+            new,
+            resource_name="group",
+            request=self.request,
+            ignore_fields=internal_fields,
+            id_field=self.model.id_field,
         )
 
         return new

kinto/views/records.py

@@ -61,8 +61,7 @@ def process_object(self, new, old=None):
             return new
 
         # Remove internal and auto-assigned fields from schemas and record.
-        internal_fields = (
-            self.model.id_field,
+        ignored_fields = (
             self.model.modified_field,
             self.schema_field,
             self.model.permissions_field,
@@ -71,9 +70,10 @@ def process_object(self, new, old=None):
         # The schema defined on the collection will be validated first.
         if "schema" in self._collection:
             schema = self._collection["schema"]
-
             try:
-                validate_schema(new, schema, ignore_fields=internal_fields)
+                validate_schema(
+                    new, schema, ignore_fields=ignored_fields, id_field=self.model.id_field
+                )
             except ValidationError as e:
                 raise_invalid(self.request, name=e.field, description=e.message)
             except RefResolutionError as e:
@@ -85,7 +85,11 @@ def process_object(self, new, old=None):
 
         # Validate also from the record:schema field defined on the bucket.
         validate_from_bucket_schema_or_400(
-            new, resource_name="record", request=self.request, ignore_fields=internal_fields
+            new,
+            resource_name="record",
+            request=self.request,
+            ignore_fields=ignored_fields,
+            id_field=self.model.id_field,
         )
 
         return new

tests/test_views_schema_group.py

@@ -4,7 +4,8 @@
 
 
 BUCKET_URL = "/buckets/blog"
-GROUP_URL = "/buckets/blog/groups/tarnac-nine"
+GROUPS_URL = "/buckets/blog/groups"
+GROUP_URL = GROUPS_URL + "/tarnac-nine"
 
 
 SCHEMA = {
@@ -42,7 +43,7 @@ def get_app_settings(cls, extras=None):
         return settings
 
 
-class CollectionValidationTest(BaseWebTestWithSchema, unittest.TestCase):
+class GroupValidationTest(BaseWebTestWithSchema, unittest.TestCase):
     def setUp(self):
         super().setUp()
         resp = self.app.put_json(
@@ -61,6 +62,19 @@ def test_groups_are_invalid_if_do_not_match_schema(self):
         self.app.put_json(GROUP_URL, {"data": {"phone": 42}}, headers=self.headers, status=400)
 
 
+class ValidateIDField(BaseWebTestWithSchema, unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        schema = {"type": "object", "properties": {"id": {"type": "string", "pattern": "^[0-7]$"}}}
+        self.app.put_json(BUCKET_URL, {"data": {"group:schema": schema}}, headers=self.headers)
+
+    def test_group_id_is_accepted_if_valid(self):
+        self.app.post_json(GROUPS_URL, {"data": {"id": "1"}}, headers=self.headers)
+
+    def test_group_id_is_rejected_if_does_not_match(self):
+        self.app.post_json(GROUPS_URL, {"data": {"id": "a"}}, headers=self.headers, status=400)
+
+
 SCHEMA_UNRESOLVABLE = {"properties": {"phone": {"$ref": "#/definitions/phone"}}}
 
 

tests/test_views_schema_record.py

@@ -272,6 +272,20 @@ def test_record_validation_can_reject_records(self):
         )
 
 
+class ValidateIDField(BaseWebTestWithSchema, unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        # See bug Kinto/kinto#1942
+        schema = {"type": "object", "properties": {"id": {"type": "string", "pattern": "^[0-7]$"}}}
+        self.app.put_json(COLLECTION_URL, {"data": {"schema": schema}}, headers=self.headers)
+
+    def test_record_id_is_accepted_if_valid(self):
+        self.app.post_json(RECORDS_URL, {"data": {"id": "1"}}, headers=self.headers)
+
+    def test_record_id_is_rejected_if_does_not_match(self):
+        self.app.post_json(RECORDS_URL, {"data": {"id": "a"}}, headers=self.headers, status=400)
+
+
 class BucketRecordSchema(BaseWebTestWithSchema, unittest.TestCase):
     def setUp(self):
         super().setUp()