Introduce pytest verify_keys and check_keytimes

This commit introduces replacements for the 'check_keys' and
'check_keytimes' from the shell test library. 'check_keys' is renamed
to 'verify_keys' because it does not assert.

For that, we introduce more functions for the class Key. The
'match_properties' function is used in 'verify_keys' to see if a set of
KeyProperties match the Key. This speficially ignores timing metadata.
The function resembles what is in 'kasp.sh:check_key()'.

The 'match_timingmetadata' function is used in 'check_keytimes' to see
if the timing metadata of a set of KeyProperties match the Key. The
values are checked in all three key files (except if the private key is
not available (set with properties["private"]), or if it is a legacy key
(set with properties["legacy"]).

An additional check function is added, to check if the key relationships
are set correctly. It follows a similar pattern as 'check_keytimes'. If
"Predecessor" and/or "Successor" are expected to be set in the state
file, this function checks so, and also verifies that they are not set
if they should not be.

(cherry picked from commit 44ff63a50d660fd5e836b9bb0365d1ae6bdbf60a)

diff --git a/bin/tests/system/isctest/kasp.py b/bin/tests/system/isctest/kasp.py
index fe31c991c89ac3ee42cb7d0aa985f473f347d237..1cab813896143099b8774fca8b3582aaa5472ec9 100644 (file)
--- a/bin/tests/system/isctest/kasp.py
+++ b/bin/tests/system/isctest/kasp.py
@@ -398,6 +398,149 @@ class Key:
 
         return digest_fromfile == digest_fromwire
 
+    def is_metadata_consistent(self, key, metadata, checkval=True):
+        """
+        If 'key' exists in 'metadata' then it must also exist in the state
+        meta data. Otherwise, it must not exist in the state meta data.
+        If 'checkval' is True, the meta data values must also match.
+        """
+        if key in metadata:
+            if checkval:
+                value = self.get_metadata(key)
+                if value != f"{metadata[key]}":
+                    isctest.log.debug(
+                        f"{self.name} {key} METADATA MISMATCH: {value} - {metadata[key]}"
+                    )
+                return value == f"{metadata[key]}"
+
+            return self.get_metadata(key) != "undefined"
+
+        value = self.get_metadata(key, must_exist=False)
+        if value != "undefined":
+            isctest.log.debug(f"{self.name} {key} METADATA UNEXPECTED: {value}")
+        return value == "undefined"
+
+    def is_timing_consistent(self, key, timing, file, comment=False):
+        """
+        If 'key' exists in 'timing' then it must match the value in the state
+        timing data. Otherwise, it must also not exist in the state timing data.
+        """
+        if key in timing:
+            value = self.get_metadata(key, file=file, comment=comment)
+            if value != str(timing[key]):
+                isctest.log.debug(
+                    f"{self.name} {key} TIMING MISMATCH: {value} - {timing[key]}"
+                )
+            return value == str(timing[key])
+
+        value = self.get_metadata(key, file=file, comment=comment, must_exist=False)
+        if value != "undefined":
+            isctest.log.debug(f"{self.name} {key} TIMING UNEXPECTED: {value}")
+        return value == "undefined"
+
+    def match_properties(self, zone, properties):
+        """
+        Check the key with given properties.
+        """
+        if not properties.properties["expect"]:
+            return False
+
+        # Check file existence.
+        # Noop. If file is missing then the get_metadata calls will fail.
+
+        # Check the public key file.
+        role = properties.properties["role_full"]
+        comment = f"This is a {role} key, keyid {self.tag}, for {zone}."
+        if not isctest.util.file_contents_contain(self.keyfile, comment):
+            isctest.log.debug(f"{self.name} COMMENT MISMATCH: expected '{comment}'")
+            return False
+
+        ttl = properties.properties["dnskey_ttl"]
+        flags = properties.properties["flags"]
+        alg = properties.metadata["Algorithm"]
+        dnskey = f"{zone}. {ttl} IN DNSKEY {flags} 3 {alg}"
+        if not isctest.util.file_contents_contain(self.keyfile, dnskey):
+            isctest.log.debug(f"{self.name} DNSKEY MISMATCH: expected '{dnskey}'")
+            return False
+
+        # Now check the private key file.
+        if properties.properties["private"]:
+            # Retrieve creation date.
+            created = self.get_metadata("Generated")
+
+            pval = self.get_metadata("Created", file=self.privatefile)
+            if pval != created:
+                isctest.log.debug(
+                    f"{self.name} Created METADATA MISMATCH: {pval} - {created}"
+                )
+                return False
+            pval = self.get_metadata("Private-key-format", file=self.privatefile)
+            if pval != "v1.3":
+                isctest.log.debug(
+                    f"{self.name} Private-key-format METADATA MISMATCH: {pval} - v1.3"
+                )
+                return False
+            pval = self.get_metadata("Algorithm", file=self.privatefile)
+            if pval != f"{alg}":
+                isctest.log.debug(
+                    f"{self.name} Algorithm METADATA MISMATCH: {pval} - {alg}"
+                )
+                return False
+
+        # Now check the key state file.
+        if properties.properties["legacy"]:
+            return True
+
+        comment = f"This is the state of key {self.tag}, for {zone}."
+        if not isctest.util.file_contents_contain(self.statefile, comment):
+            isctest.log.debug(f"{self.name} COMMENT MISMATCH: expected '{comment}'")
+            return False
+
+        attributes = [
+            "Lifetime",
+            "Algorithm",
+            "Length",
+            "KSK",
+            "ZSK",
+            "GoalState",
+            "DNSKEYState",
+            "KRRSIGState",
+            "ZRRSIGState",
+            "DSState",
+        ]
+        for key in attributes:
+            if not self.is_metadata_consistent(key, properties.metadata):
+                return False
+
+        # A match is found.
+        return True
+
+    def match_timingmetadata(self, timings, file=None, comment=False):
+        if file is None:
+            file = self.statefile
+
+        attributes = [
+            "Generated",
+            "Created",
+            "Published",
+            "Publish",
+            "PublishCDS",
+            "SyncPublish",
+            "Active",
+            "Activate",
+            "Retired",
+            "Inactive",
+            "Revoked",
+            "Removed",
+            "Delete",
+        ]
+        for key in attributes:
+            if not self.is_timing_consistent(key, timings, file, comment=comment):
+                isctest.log.debug(f"{self.name} TIMING METADATA MISMATCH: {key}")
+                return False
+
+        return True
+
     def __lt__(self, other: "Key"):
         return self.name < other.name
 
@@ -459,6 +602,112 @@ def check_zone_is_signed(server, zone):
     assert signed
 
 
+def verify_keys(zone, keys, expected):
+    """
+    Checks keys for a configured zone. This verifies:
+    1. The expected number of keys exist in 'keys'.
+    2. The keys match the expected properties.
+    """
+
+    def _verify_keys():
+        # check number of keys matches expected.
+        if len(keys) != len(expected):
+            return False
+
+        if len(keys) == 0:
+            return True
+
+        for expect in expected:
+            expect.key = None
+
+        for key in keys:
+            found = False
+            i = 0
+            while not found and i < len(expected):
+                if expected[i].key is None:
+                    found = key.match_properties(zone, expected[i])
+                    if found:
+                        key.external = expected[i].properties["legacy"]
+                        expected[i].key = key
+                i += 1
+            if not found:
+                return False
+
+        return True
+
+    isctest.run.retry_with_timeout(_verify_keys, timeout=10)
+
+
+def check_keytimes(keys, expected):
+    """
+    Check the key timing metadata for all keys in 'keys'.
+    """
+    assert len(keys) == len(expected)
+
+    if len(keys) == 0:
+        return
+
+    for key in keys:
+        for expect in expected:
+            if expect.properties["legacy"]:
+                continue
+
+            if not key is expect.key:
+                continue
+
+            synonyms = {}
+            if "Generated" in expect.timing:
+                synonyms["Created"] = expect.timing["Generated"]
+            if "Published" in expect.timing:
+                synonyms["Publish"] = expect.timing["Published"]
+            if "PublishCDS" in expect.timing:
+                synonyms["SyncPublish"] = expect.timing["PublishCDS"]
+            if "Active" in expect.timing:
+                synonyms["Activate"] = expect.timing["Active"]
+            if "Retired" in expect.timing:
+                synonyms["Inactive"] = expect.timing["Retired"]
+            if "DeleteCDS" in expect.timing:
+                synonyms["SyncDelete"] = expect.timing["DeleteCDS"]
+            if "Revoked" in expect.timing:
+                synonyms["Revoked"] = expect.timing["Revoked"]
+            if "Removed" in expect.timing:
+                synonyms["Delete"] = expect.timing["Removed"]
+
+            assert key.match_timingmetadata(synonyms, file=key.keyfile, comment=True)
+            if expect.properties["private"]:
+                assert key.match_timingmetadata(synonyms, file=key.privatefile)
+            if not expect.properties["legacy"]:
+                assert key.match_timingmetadata(expect.timing)
+
+                state_changes = [
+                    "DNSKEYChange",
+                    "KRRSIGChange",
+                    "ZRRSIGChange",
+                    "DSChange",
+                ]
+                for change in state_changes:
+                    assert key.is_metadata_consistent(
+                        change, expect.timing, checkval=False
+                    )
+
+
+def check_keyrelationships(keys, expected):
+    """
+    Check the key relationships (Successor and Predecessor metadata).
+    """
+    for key in keys:
+        for expect in expected:
+            if expect.properties["legacy"]:
+                continue
+
+            if not key is expect.key:
+                continue
+
+            relationship_status = ["Predecessor", "Successor"]
+            for status in relationship_status:
+                assert key.is_metadata_consistent(status, expect.metadata)
+
+
 def check_dnssec_verify(server, zone):
     # Check if zone if DNSSEC valid with dnssec-verify.
     fqdn = f"{zone}."