Enable configurable enrollment key kty/size/alg/crv (#65)

nodeman/nodes.py

@@ -158,7 +158,7 @@ async def create_node(
 
     domain = request.app.settings.nodes.domain
 
-    node_enrollment_key = JWK.generate(kty="oct", size=256, alg="HS256")
+    node_enrollment_key = request.app.generate_enrollment_key()
 
     if name is None:
         node = TapirNode.create_next_node(domain=domain)

nodeman/server.py

@@ -75,6 +75,13 @@ def __init__(self, settings: Settings):
         else:
             self.ca_client = None
 
+        self.generate_enrollment_key_kwargs = self.settings.enrollment.generate_kwargs()
+        logger.debug("Enrollment key kwargs: %s", self.generate_enrollment_key_kwargs)
+
+    def generate_enrollment_key(self) -> JWK:
+        """Generate enrollment key"""
+        return JWK.generate(**self.generate_enrollment_key_kwargs)
+
     @staticmethod
     def get_internal_ca_client(settings: InternalCaSettings) -> InternalCertificateAuthority:
         res = InternalCertificateAuthority.load(

nodeman/settings.py

@@ -2,6 +2,7 @@
 from typing import Annotated, Self
 
 from argon2 import PasswordHasher
+from jwcrypto.jwk import JWK
 from pydantic import (
     AnyHttpUrl,
     BaseModel,
@@ -62,6 +63,41 @@ class NodesSettings(BaseModel):
     )
 
 
+class EnrollmentSettings(BaseModel):
+    kty: str = Field(default="oct")
+    alg: str = Field(default="HS256")
+    crv: str | None = Field(default=None)
+    size: int | None = Field(default=None)
+
+    @model_validator(mode="after")
+    def validate_jwk_parameters(self) -> Self:
+        kwargs = self.generate_kwargs()
+        try:
+            JWK.generate(**kwargs)
+        except Exception as exc:
+            raise ValueError("Invalid enrollment key parameters") from exc
+        return self
+
+    def generate_kwargs(self) -> dict[str, str | int]:
+        match self.kty:
+            case "oct":
+                if self.crv:
+                    raise ValueError(f"Cannot specify curve for {self.kty}")
+                return {"kty": self.kty, "alg": self.alg, "size": self.size or 256}
+            case "RSA":
+                if self.crv:
+                    raise ValueError(f"Cannot specify curve for {self.kty}")
+                return {"kty": self.kty, "alg": self.alg, "size": self.size or 2048}
+            case "EC" | "OKP":
+                if self.crv is None:
+                    raise ValueError("Unknown curve")
+                if self.size:
+                    raise ValueError(f"size not supported for {self.kty}")
+                return {"kty": self.kty, "alg": self.alg, "crv": self.crv}
+            case _:
+                raise ValueError("Unsupported key type")
+
+
 class User(BaseModel):
     username: Annotated[str, StringConstraints(min_length=2, max_length=32, pattern=r"^[a-zA-Z0-9_-]+$")]
     password_hash: str
@@ -87,6 +123,7 @@ class Settings(BaseSettings):
     internal_ca: InternalCaSettings | None = None
 
     nodes: NodesSettings = Field(default=NodesSettings())
+    enrollment: EnrollmentSettings = Field(default=EnrollmentSettings())
 
     legacy_nodes_directory: DirectoryPath | None = None
 

tests/test.toml

@@ -8,6 +8,16 @@ domain = "test.dnstapir.se"
 trusted_jwks = "tests/trusted_jwks.json"
 mqtt_broker = "mqtts://localhost"
 
+[enrollment]
+kty = "oct"
+alg = "HS256"
+size = 256
+
+#[enrollment]
+#kty = "OKP"
+#crv = "Ed25519"
+#alg = "EdDSA"
+
 [[users]]
 username = "username"
 password_hash = "$argon2id$v=19$m=65536,t=3,p=4$2UbbvL5YpSjGyeha++HE5g$o8iGuvAgrl0azPFDK79mCYQT10nqIGyU1XLipGwL4rc"