Support for private algorithms

dns/dnssec.py

@@ -20,6 +20,7 @@
 
 from typing import Any, cast, Callable, Dict, List, Optional, Set, Tuple, Union
 
+import abc
 import contextlib
 import functools
 import hashlib
@@ -192,6 +193,26 @@ def ok_to_validate_ds(self, algorithm: DSDigest) -> bool:
 default_policy = rfc_8624_policy
 
 
+class PrivateAlgorithmPublicKeyBase(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def public_bytes(self) -> bytes:
+        pass
+
+    @abc.abstractmethod
+    def verify(self, signature: bytes, data: bytes) -> None:
+        pass
+
+
+class PrivateAlgorithmPrivateKeyBase(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def public_key(self) -> PrivateAlgorithmPublicKeyBase:
+        pass
+
+    @abc.abstractmethod
+    def sign(self, data: bytes) -> bytes:
+        pass
+
+
 def make_ds(
     name: Union[dns.name.Name, str],
     key: dns.rdata.Rdata,
@@ -360,6 +381,13 @@ def _is_sha1(algorithm: int) -> bool:
     )
 
 
+def _is_private(algorithm: int) -> bool:
+    return algorithm in (
+        Algorithm.PRIVATEDNS,
+        Algorithm.PRIVATEOID,
+    )
+
+
 def _is_sha256(algorithm: int) -> bool:
     return algorithm in (Algorithm.RSASHA256, Algorithm.ECDSAP256SHA256)
 
@@ -397,6 +425,9 @@ def _ensure_algorithm_key_combination(algorithm: int, key: PublicKey) -> None:
         if algorithm == Algorithm.ED448:
             return
         raise AlgorithmKeyMismatch('algorithm "%s" not valid for ED448 key' % algorithm)
+    if isinstance(key, PrivateAlgorithmPublicKeyBase):
+        if _is_private(algorithm):
+            return
 
     raise TypeError("unsupported key type")
 
@@ -416,6 +447,8 @@ def _make_hash(algorithm: int) -> Any:
         return hashes.SHA512()
     if algorithm == Algorithm.ED448:
         return hashes.SHAKE256(114)
+    if _is_private(algorithm):
+        return None
 
     raise ValidationFailure("unknown hash for algorithm %u" % algorithm)
 
@@ -817,6 +850,12 @@ def _sign(
         signature = private_key.sign(data)
         if verify:
             private_key.public_key().verify(signature, data)
+    elif isinstance(private_key, PrivateAlgorithmPrivateKeyBase):
+        if not _is_private(dnskey.algorithm):
+            raise ValueError("Invalid DNSKEY algorithm for private algorithm key")
+        signature = private_key.sign(data)
+        if verify:
+            private_key.public_key().verify(signature, data)
     else:
         raise TypeError("Unsupported key algorithm")
 
@@ -966,6 +1005,8 @@ def encode_ecdsa_public_key(public_key: "ec.EllipticCurvePublicKey") -> bytes:
         key_bytes = public_key.public_bytes(
             encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw
         )
+    elif isinstance(public_key, PrivateAlgorithmPublicKeyBase):
+        key_bytes = public_key.public_bytes()
     else:
         raise TypeError("unsupported key algorithm")
 

tests/test_dnssec.py

@@ -1359,6 +1359,51 @@ def _test_signature(self, key, algorithm, rrset, signer=None, policy=None):
         rrsigset = dns.rrset.from_rdata(rrname, ttl, rrsig)
         dns.dnssec.validate(rrset=rrset, rrsigset=rrsigset, keys=keys, policy=policy)
 
+    def testSignaturePrivateAlgorithm(self):  # type: () -> None
+        SIGNED_FAKE_DATA = b"signed"
+        PRIVATE_ALGORITHM_NAME = dns.name.from_text("algorithm.example.com").to_wire()
+
+        class MyPublicKey(dns.dnssec.PrivateAlgorithmPublicKeyBase):
+            name = PRIVATE_ALGORITHM_NAME
+
+            def verify(self, signature: bytes, data: bytes):
+                return
+
+            def public_bytes(self) -> bytes:
+                return self.name + b"publickey"
+
+        class MyPrivateKey(dns.dnssec.PrivateAlgorithmPrivateKeyBase):
+            def public_key(self):
+                return MyPublicKey()
+
+            def sign(self, data: bytes) -> bytes:
+                return SIGNED_FAKE_DATA
+
+        key = MyPrivateKey()
+        algorithm = dns.dnssec.Algorithm.PRIVATEDNS
+        ttl = 60
+        lifetime = 3600
+        rrset = abs_soa
+        rrname = rrset.name
+        signer = rrname
+        dnskey = dns.dnssec.make_dnskey(
+            public_key=key.public_key(), algorithm=algorithm
+        )
+        dnskey_rrset = dns.rrset.from_rdata(signer, ttl, dnskey)
+        self.assertEqual(
+            dnskey.key[0 : len(PRIVATE_ALGORITHM_NAME)], PRIVATE_ALGORITHM_NAME
+        )
+        rrsig = dns.dnssec.sign(
+            rrset=rrset,
+            private_key=key,
+            dnskey=dnskey,
+            lifetime=lifetime,
+            signer=signer,
+            verify=True,
+            policy=None,
+        )
+        self.assertEqual(rrsig.signature, SIGNED_FAKE_DATA)
+
 
 if __name__ == "__main__":
     unittest.main()