Prototype descriptors and validation functions

tests/test_api.py

@@ -23,6 +23,7 @@
 import tuf.exceptions
 from tuf.api.metadata import (
     Metadata,
+    Root,
     Snapshot,
     Timestamp,
     Targets
@@ -38,6 +39,8 @@
     CanonicalJSONSerializer
 )
 
+import tuf.api.validators as validators
+
 from securesystemslib.interface import (
     import_ed25519_publickey_from_file,
     import_ed25519_privatekey_from_file
@@ -380,6 +383,172 @@ def test_support_for_unrecognized_fields(self):
             )
 
 
+class TestValidation(unittest.TestCase):
+
+    def test_validate_signed_attr(self):
+        # spec_version validation
+        for val in [None, True, 111, 1.1]:
+            with self.assertRaises(TypeError):
+                validators.validate_spec_version(val)
+        for val in ["", "1.11", "2", "0.1.1"]:
+            with self.assertRaises(ValueError):
+                validators.validate_spec_version(val)
+        validators.validate_spec_version("1.0.0")
+
+        # _type validation
+        for val in [None, True, 111, 1.1]:
+            with self.assertRaises(TypeError):
+                validators.validate_type(val)
+        for val in ["wrong", "", "ROOT", "timestamp1"]:
+            with self.assertRaises(ValueError):
+                validators.validate_type(val)
+        for val in ["root", "snapshot", "targets", "timestamp"]:
+            validators.validate_type(val)
+
+        # version validation
+        for val in [None, True, "1", "", 1.0]:
+            with self.assertRaises(TypeError):
+                validators.validate_version(val)
+        with self.assertRaises(ValueError):
+            validators.validate_version(-1)
+            validators.validate_version(0)
+        validators.validate_version(1)
+
+        # expiry validation
+        for val in [None, True, "1", "", 1.0]:
+            with self.assertRaises(TypeError):
+                validators.validate_expires(val)
+        past_time = datetime(1990, 1, 1)
+        with self.assertRaises(ValueError):
+            validators.validate_expires(past_time)
+        future_time = datetime(2050, 1, 1)
+        validators.validate_expires(future_time)
+
+
+    def test_root_specific_attr(self):
+        # consistent_snapshot Root attribute
+        for val in [None, "1", "", "False", 1]:
+            with self.assertRaises(TypeError):
+                validators.validate_consistent_snapshot(val)
+        validators.validate_consistent_snapshot(True)
+
+        # keyid Key attribute
+        for val in [None, False, 1]:
+            with self.assertRaises(TypeError):
+                validators.validate_keyid(val)
+        for val in ["", "12345"]:
+            with self.assertRaises(ValueError):
+                validators.validate_keyid(val)
+        validators.validate_keyid("4e777de0d275f9d28588dd9a1606cc748e548f9e22b6795b7cb3f63f98035fcb")
+
+        # keytype Key attribute
+        for val in [None, False, 1]:
+            with self.assertRaises(TypeError):
+                validators.validate_keytype(val)
+        validators.validate_keytype("rsa")
+
+        # scheme Key attribute
+        for val in [None, False, 1]:
+            with self.assertRaises(TypeError):
+                validators.validate_scheme(val)
+        validators.validate_scheme("rsassa-pss-sha256")
+
+        # keyval Key attribute
+        for val in [None, False, 1, ""]:
+            with self.assertRaises(TypeError):
+                validators.validate_keyval(val)
+        for val in [{}, {"a": 3}, {"public": "123456"}]:
+            with self.assertRaises(ValueError):
+                validators.validate_keyval(val)
+        validators.validate_keyval({
+            "public": "edcd0a32a07dce33f7c7873aaffbff36d20ea30787574ead335eefd337e4dacd"
+        })
+
+        # role Root attribute
+        for val in [None, False, 1]:
+            with self.assertRaises(TypeError):
+                validators.validate_role(val)
+        for val in ["", "123456"]:
+            with self.assertRaises(ValueError):
+                validators.validate_role(val)
+        validators.validate_role("timestamp")
+
+        # threshold Root attribute
+        for val in [None, False, "0", "1.0", 1.0]:
+            with self.assertRaises(TypeError):
+                validators.validate_threshold(val)
+        for val in [-1, 0]:
+            with self.assertRaises(ValueError):
+                validators.validate_threshold(val)
+        validators.validate_threshold(1)
+
+
+    def test_descriptors_example_usage(self):
+        data = {
+            '_type': 'snapshot',
+            'spec_version': '1.2.3',
+            'expires': datetime.now() + timedelta(hours=1),
+            'version': 3,
+            "consistent_snapshot": False,
+            "keys": {
+                "59a4df8af818e9ed7abe0764c0b47b4240952aa0d179b5b78346c470ac30278d": {
+                    "keytype": "ed25519",
+                    "keyval": {
+                    "public": "edcd0a32a07dce33f7c7873aaffbff36d20ea30787574ead335eefd337e4dacd"
+                    },
+                    "scheme": "ed25519"
+                },
+            },
+            "roles": {
+                "root": {
+                    "keyids": [
+                    "4e777de0d275f9d28588dd9a1606cc748e548f9e22b6795b7cb3f63f98035fcb"
+                    ],
+                    "threshold": 1
+                },
+            }
+        }
+        # Validation happens on initialization and during assigment
+        root = Root(**data)
+
+        # _type validation checks that the type is one of our expected roles
+        data['_type'] = 'wrong'
+        with self.assertRaises(ValueError):
+            root = Root(**data)
+        data['_type'] = 'snapshot'
+
+        # spec_version should be a specific string in sem format
+        # and be the same major version as the current tuf version
+        for val in ['', '1.11', '2', '0.1.1']:
+            with self.assertRaises(ValueError):
+                root.spec_version = val
+
+        # Version should be an int above 0.
+        with self.assertRaises(ValueError):
+            root.version = -1
+
+        # consistent_snapshot should be boolean
+        with self.assertRaises(TypeError):
+            root.consistent_snapshot = -1
+
+        # If we want to change root with a totally different role dict
+        # Without the neceserry format
+        new_roles = {
+            "roles" : 
+                {
+                    "DDS": {
+                        "keyids": [
+                        "4e777de0d275f9d28588dd9a1606cc748e548f9e22b6795b7cb3f63f98035fcb"
+                        ],
+                        "threshold": 1
+                    },
+                }
+        }
+        with self.assertRaises(ValueError):
+            root.roles = new_roles
+
+
+
 # Run unit test.
 if __name__ == '__main__':
     utils.configure_test_logging(sys.argv)

tuf/api/metadata.py

@@ -30,6 +30,18 @@
     MetadataSerializer,
     SignedSerializer,
 )
+from tuf.api.validators import (
+    METADATA_TYPES,
+    Bool,
+    DateTime,
+    Dictionary,
+    Integer,
+    OneOf,
+    String,
+    _check_dict_elements_uniqueness,
+    _check_semantic_versioning,
+    _check_str_one_of_metadata_types,
+)
 
 
 class Metadata:
@@ -324,6 +336,13 @@ class Signed:
 
     """
 
+    _type = OneOf(METADATA_TYPES)
+    version = Integer(minvalue=1)
+    # The minimum length is 5 because "X.Y.Z" has 5 characters.
+    spec_version = String(minsize=5, predicate=_check_semantic_versioning)
+    expires = DateTime()
+    unrecognized_fields = Dictionary(keys_type=str)
+
     # NOTE: Signed is a stupid name, because this might not be signed yet, but
     # we keep it to match spec terminology (I often refer to this as "payload",
     # or "inner metadata")
@@ -440,6 +459,11 @@ class Root(Signed):
 
     """
 
+    consistent_snapshot = Bool()
+    keys = Dictionary(keys_type=str)
+    roles = Dictionary(
+        keys_type=str, predicate_keys=_check_str_one_of_metadata_types
+    )
     # TODO: determine an appropriate value for max-args and fix places where
     # we violate that. This __init__ function takes 7 arguments, whereas the
     # default max-args value for pylint is 5