feat(api): allow CSK import on domain creation

Author: nils-wisiol

api/desecapi/dnssec.py

@@ -0,0 +1,55 @@
+import base64
+import re
+
+import dns.dnssec
+from cryptography.hazmat.primitives.asymmetric import ec
+
+
+def parse_csk_private_key(private_key: str) -> dict:
+    if not private_key or not private_key.strip():
+        raise ValueError("Missing private key material")
+
+    algorithm = None
+    private_b64 = None
+    for line in private_key.strip().splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        if line.lower().startswith("algorithm:"):
+            match = re.search(r"\b(\d+)\b", line)
+            if match:
+                algorithm = int(match.group(1))
+        elif line.lower().startswith("privatekey:"):
+            private_b64 = line.split(":", 1)[1].strip()
+
+    if algorithm is None:
+        raise ValueError("Missing algorithm in private key")
+    if private_b64 is None:
+        raise ValueError("Missing PrivateKey in private key")
+
+    if algorithm != 13:
+        raise ValueError("Unsupported algorithm")
+
+    try:
+        private_bytes = base64.b64decode(private_b64, validate=True)
+    except Exception as exc:
+        raise ValueError("Invalid base64 private key") from exc
+
+    if len(private_bytes) > 32:
+        raise ValueError("Invalid private key length")
+    if len(private_bytes) < 32:
+        private_bytes = private_bytes.rjust(32, b"\x00")
+
+    private_value = int.from_bytes(private_bytes, "big")
+    if private_value == 0:
+        raise ValueError("Invalid private key value")
+
+    private_key_obj = ec.derive_private_key(private_value, ec.SECP256R1())
+    dnskey = dns.dnssec.make_dnskey(
+        private_key_obj.public_key(), algorithm=13, flags=257, protocol=3
+    ).to_text()
+
+    return {
+        "algorithm": algorithm,
+        "dnskey": dnskey,
+    }

api/desecapi/knot.py

@@ -10,6 +10,7 @@
 import dns.query
 import dns.rcode
 import dns.rdtypes.ANY.DNSKEY
+import dns.rdata
 import dns.rdatatype
 import dns.tsig
 import dns.tsigkeyring
@@ -183,6 +184,26 @@ def ensure_default_ns(name):
     _send_update_with_retry(update)
 
 
+def import_csk_key(name, *, dnskey, private_key=None):
+    if not wait_for_zone(name, attempts=60, interval_seconds=0.5):
+        raise KnotException(f"Knot zone {name} not ready for updates")
+    update = _new_update(name)
+    apex = name.rstrip(".") + "."
+    update.add(apex, settings.DEFAULT_NS_TTL, "DNSKEY", dnskey)
+    try:
+        key_rdata = dns.rdata.from_text(
+            dns.rdataclass.IN, dns.rdatatype.DNSKEY, dnskey
+        )
+        cds_records = [
+            dns.dnssec.make_ds(dns.name.from_text(name), key_rdata, algo).to_text()
+            for algo in (2, 4)
+        ]
+        update.replace(apex, 0, "CDS", *cds_records)
+    except Exception:
+        pass
+    _send_update_with_retry(update)
+
+
 def wait_for_zone(name, *, attempts=20, interval_seconds=0.5) -> bool:
     query = dns.message.make_query(name, dns.rdatatype.SOA)
     query_timeout = min(settings.NSLORD_KNOT_TIMEOUT, 1.0)

api/desecapi/pdns.py

@@ -166,6 +166,24 @@ def get_keys(domain):
     ]
 
 
+def list_cryptokeys(domain_name):
+    return _pdns_get(NSLORD, "/zones/%s/cryptokeys" % pdns_id(domain_name)).json()
+
+
+def set_cryptokey_active(domain_name, key_id, active):
+    _pdns_put(
+        NSLORD,
+        "/zones/%s/cryptokeys/%s" % (pdns_id(domain_name), key_id),
+        data={"active": bool(active)},
+    )
+
+
+def delete_cryptokey(domain_name, key_id):
+    _pdns_delete(
+        NSLORD, "/zones/%s/cryptokeys/%s" % (pdns_id(domain_name), key_id)
+    )
+
+
 def get_zone(domain):
     """
     Retrieves a dict representation of the zone from pdns
@@ -262,6 +280,37 @@ def create_zone_master(name, master_host="nslord"):
     )
 
 
+def import_csk_key(name, *, dnskey, private_key):
+    response = _pdns_post(
+        NSLORD,
+        "/zones/%s/cryptokeys" % pdns_id(name),
+        {
+            "keytype": "csk",
+            "active": True,
+            "published": True,
+            "content": private_key,
+        },
+    )
+    cryptokey = response.json()
+    imported_id = cryptokey.get("id")
+    keys = list_cryptokeys(name)
+    if imported_id is None:
+        for key in keys:
+            if key.get("dnskey") == dnskey:
+                imported_id = key.get("id")
+                break
+    if imported_id is None:
+        return cryptokey
+    for key in keys:
+        if key.get("id") == imported_id:
+            if not key.get("active", False):
+                set_cryptokey_active(name, imported_id, True)
+            continue
+        delete_cryptokey(name, key["id"])
+    rectify_zone(name)
+    return cryptokey
+
+
 def delete_zone(name, server):
     _pdns_delete(server, "/zones/" + pdns_id(name))
 
@@ -282,6 +331,10 @@ def axfr_to_master(zone):
     _pdns_put(NSMASTER, "/zones/%s/axfr-retrieve" % pdns_id(zone))
 
 
+def rectify_zone(name):
+    _pdns_put(NSLORD, "/zones/%s/rectify" % pdns_id(name))
+
+
 def construct_catalog_rrset(
     zone=None, delete=False, subname=None, qtype="PTR", rdata=None
 ):

api/desecapi/pdns_change_tracker.py

@@ -74,6 +74,7 @@ def pch_do(self):
     def __init__(self):
         self._domain_additions = set()
         self._domain_deletions = set()
+        self._domain_create_payload = {}
         self._rr_set_additions = {}
         self._rr_set_modifications = {}
         self._rr_set_deletions = {}
@@ -119,6 +120,7 @@ def __enter__(self):
         )
         self._domain_additions = set()
         self._domain_deletions = set()
+        self._domain_create_payload = {}
         self._rr_set_additions = {}
         self._rr_set_modifications = {}
         self._rr_set_deletions = {}
@@ -227,7 +229,7 @@ def _nslord_for_domain(self, domain_name):
         return nslord or Domain.NSLord.PDNS
 
     @abstractmethod
-    def _create_domain_change(self, domain_name, nslord):
+    def _create_domain_change(self, domain_name, nslord, create_payload):
         raise NotImplementedError()
 
     @abstractmethod
@@ -258,7 +260,10 @@ def _compute_changes(self):
         for domain_name in self._rr_set_additions.keys() | self._domain_additions:
             nslord = self._nslord_for_domain(domain_name)
             if domain_name in self._domain_additions:
-                changes.append(self._create_domain_change(domain_name, nslord))
+                create_payload = self._domain_create_payload.get(domain_name)
+                changes.append(
+                    self._create_domain_change(domain_name, nslord, create_payload)
+                )
 
             additions = self._rr_set_additions.get(domain_name, set())
             modifications = self._rr_set_modifications.get(domain_name, set())
@@ -356,6 +361,9 @@ def _domain_updated(self, domain: Domain, created=False, deleted=False):
                 deletions.remove(name)
             else:
                 additions.add(name)
+                create_payload = getattr(domain, "_csk_private_key_data", None)
+                if create_payload:
+                    self._domain_create_payload[name] = create_payload
         elif deleted:
             if name in additions:
                 additions.remove(name)
@@ -430,12 +438,22 @@ def pdns_do(self):
             self.nslord_do()
 
     class CreateDomain(PDNSChange):
+        def __init__(self, domain_name, create_payload=None):
+            super().__init__(domain_name)
+            self._create_payload = create_payload or {}
+
         @property
         def axfr_required(self):
             return True
 
         def nslord_do(self):
             pdns.create_zone_lord(self.domain_name)
+            if self._create_payload:
+                pdns.import_csk_key(
+                    self.domain_name,
+                    dnskey=self._create_payload["dnskey"],
+                    private_key=self._create_payload["private_key"],
+                )
             pdns.create_zone_master(self.domain_name)
             pdns.update_catalog(self.domain_name)
 
@@ -534,8 +552,8 @@ def __str__(self):
                 )
             )
 
-    def _create_domain_change(self, domain_name, nslord):
-        return PDNSChangeTracker.CreateDomain(domain_name)
+    def _create_domain_change(self, domain_name, nslord, create_payload):
+        return PDNSChangeTracker.CreateDomain(domain_name, create_payload)
 
     def _delete_domain_change(self, domain_name, nslord):
         return PDNSChangeTracker.DeleteDomain(domain_name)
@@ -553,13 +571,23 @@ class KnotChange(BaseChangeTracker.Change):
         pass
 
     class CreateDomain(KnotChange):
+        def __init__(self, domain_name, create_payload=None):
+            super().__init__(domain_name)
+            self._create_payload = create_payload or {}
+
         @property
         def axfr_required(self):
             return True
 
         def nslord_do(self):
             knot.create_zone(self.domain_name)
             knot.ensure_default_ns(self.domain_name)
+            if self._create_payload:
+                knot.import_csk_key(
+                    self.domain_name,
+                    dnskey=self._create_payload["dnskey"],
+                    private_key=self._create_payload["private_key"],
+                )
             pdns.create_zone_master(
                 self.domain_name, master_host=settings.NSLORD_KNOT_HOST
             )
@@ -626,8 +654,8 @@ def __str__(self):
                 )
             )
 
-    def _create_domain_change(self, domain_name, nslord):
-        return KnotChangeTracker.CreateDomain(domain_name)
+    def _create_domain_change(self, domain_name, nslord, create_payload):
+        return KnotChangeTracker.CreateDomain(domain_name, create_payload)
 
     def _delete_domain_change(self, domain_name, nslord):
         return KnotChangeTracker.DeleteDomain(domain_name)
@@ -646,8 +674,8 @@ def _backend(self, nslord):
             return KnotChangeTracker
         return PDNSChangeTracker
 
-    def _create_domain_change(self, domain_name, nslord):
-        return self._backend(nslord).CreateDomain(domain_name)
+    def _create_domain_change(self, domain_name, nslord, create_payload):
+        return self._backend(nslord).CreateDomain(domain_name, create_payload)
 
     def _delete_domain_change(self, domain_name, nslord):
         return self._backend(nslord).DeleteDomain(domain_name)

api/desecapi/serializers/domains.py

@@ -3,6 +3,7 @@
 from django.conf import settings
 from rest_framework import serializers
 
+from desecapi import dnssec
 from desecapi.models import Domain, RR_SET_TYPES_AUTOMATIC
 from desecapi.validators import ReadOnlyOnUpdateValidator
 
@@ -15,6 +16,7 @@ class DomainSerializer(serializers.ModelSerializer):
         "name_unavailable": "This domain name conflicts with an existing domain, or is disallowed by policy.",
     }
     zonefile = serializers.CharField(write_only=True, required=False, allow_blank=True)
+    csk_private_key = serializers.CharField(write_only=True, required=False)
     nslord = serializers.ChoiceField(
         choices=Domain.NSLord.choices, required=False, write_only=True
     )
@@ -29,6 +31,7 @@ class Meta:
             "minimum_ttl",
             "touched",
             "zonefile",
+            "csk_private_key",
             "nslord",
         )
         read_only_fields = (
@@ -42,6 +45,7 @@ class Meta:
     def __init__(self, *args, include_keys=False, **kwargs):
         self.include_keys = include_keys
         self.import_zone = None
+        self._csk_private_key_data = None
         super().__init__(*args, **kwargs)
 
     def get_fields(self):
@@ -100,11 +104,32 @@ def parse_zonefile(self, domain_name: str, zonefile: str):
     def validate(self, attrs):
         if attrs.get("zonefile") is not None:
             self.parse_zonefile(attrs.get("name"), attrs.pop("zonefile"))
+        if attrs.get("csk_private_key") is not None:
+            private_key = attrs.get("csk_private_key")
+            if not private_key.strip():
+                raise serializers.ValidationError(
+                    {"csk_private_key": ["Missing private key material."]}
+                )
+            try:
+                parsed = dnssec.parse_csk_private_key(private_key)
+            except ValueError as exc:
+                raise serializers.ValidationError(
+                    {"csk_private_key": [str(exc)]}
+                ) from exc
+            self._csk_private_key_data = {
+                "private_key": private_key,
+                "dnskey": parsed["dnskey"],
+                "algorithm": parsed["algorithm"],
+            }
         return super().validate(attrs)
 
     def create(self, validated_data):
+        validated_data.pop("csk_private_key", None)
         # save domain
-        domain: Domain = super().create(validated_data)
+        domain = Domain(**validated_data)
+        if self._csk_private_key_data is not None:
+            domain._csk_private_key_data = self._csk_private_key_data
+        domain.save()
 
         # save RRsets if zonefile was given
         nodes = getattr(self.import_zone, "nodes", None)

test/e2e2/conftest.py

@@ -241,14 +241,18 @@ def login(self, email: str, password: str) -> requests.Response:
     def domain_list(self) -> requests.Response:
         return self.get("/domains/").json()
 
-    def domain_create(self, name, zonefile=None, nslord=None) -> requests.Response:
+    def domain_create(
+        self, name, zonefile=None, nslord=None, csk_private_key=None
+    ) -> requests.Response:
         if name in self.domains:
             raise ValueError
         data = {"name": name}
         if zonefile is not None:
             data['zonefile'] = zonefile
         if nslord is not None:
             data['nslord'] = nslord
+        if csk_private_key is not None:
+            data['csk_private_key'] = csk_private_key
         response = self.post("/domains/", data=data)
         self.domains[name] = response.json()
         if nslord is not None: