use ORG mode for client- and server certs

api/kiwi_vpn_api/config.py

@@ -194,7 +194,7 @@ class LockableCountry(LockableString):
     value: constr(max_length=2)
 
 
-class DNParts(BaseModel):
+class ServerDN(BaseModel):
     """
     This server's "distinguished name"
     """
@@ -204,6 +204,8 @@ class DNParts(BaseModel):
     city: LockableString
     organization: LockableString
     organizational_unit: LockableString
+    email: LockableString
+    common_name: str
 
 
 class CertificateAlgo(Enum):
@@ -227,9 +229,10 @@ class CryptoConfig(BaseModel):
     schemes: list[str] = ["bcrypt"]
 
     # pki settings
-    ca_password: str | None
     cert_algo: CertificateAlgo | None
-    expiry_days: int | None
+    ca_password: str | None
+    ca_expiry_days: int | None
+    cert_expiry_days: int | None
 
     @property
     def context(self) -> CryptContext:
@@ -244,15 +247,13 @@ class Config(BaseModel):
     Configuration for `kiwi-vpn-api`
     """
 
-    # common name for the server
-    server_name: str
     # may include client-to-client, cipher etc.
     openvpn_extra_options: dict[str, Any] | None
 
     db: DBConfig
     jwt: JWTConfig
     crypto: CryptoConfig
-    default_dn: DNParts
+    server_dn: ServerDN
 
     __singleton: Config | None = None
 

api/kiwi_vpn_api/easyrsa.py

@@ -2,14 +2,99 @@
 Python interface to EasyRSA CA.
 """
 
+from __future__ import annotations
+
 import subprocess
 from datetime import datetime
 from pathlib import Path
 
 from OpenSSL import crypto
 from passlib import pwd
+from pydantic import BaseModel
 
-from .config import Config, Settings
+from .config import CertificateAlgo, Config, Settings
+from .db import Device
+
+
+class DistinguishedName(BaseModel):
+    """
+    An `X.509 distinguished name` (DN) as specified in RFC 5280
+    """
+
+    country: str
+    state: str
+    city: str
+    organization: str
+    organizational_unit: str
+    email: str
+    common_name: str
+
+    @classmethod
+    def build(cls, device: Device | None = None) -> DistinguishedName:
+        """
+        Create a DN from the current config and an optional device
+        """
+
+        # extract server DN config
+        server_dn = Config._.server_dn
+        result = cls(
+            country=server_dn.country.value,
+            state=server_dn.state.value,
+            city=server_dn.city.value,
+            organization=server_dn.organization.value,
+            organizational_unit=server_dn.organizational_unit.value,
+            email=server_dn.email.value,
+            common_name=server_dn.common_name,
+        )
+
+        # no device specified -> done
+        if device is None:
+            return result
+
+        # don't override locked or empty fields
+        if not (server_dn.country.locked
+                or device.owner.country is None):
+            result.country = device.owner.country
+
+        if not (server_dn.state.locked
+                or device.owner.state is None):
+            result.state = device.owner.state
+
+        if not (server_dn.city.locked
+                or device.owner.city is None):
+            result.city = device.owner.city
+
+        if not (server_dn.organization.locked
+                or device.owner.organization is None):
+            result.organization = device.owner.organization
+
+        if not (server_dn.organizational_unit.locked
+                or device.owner.organizational_unit is None):
+            result.organizational_unit = device.owner.organizational_unit
+
+        # definitely use derived email and common_name
+        result.email = device.owner.email
+        result.common_name = f"{device.owner.name}_{device.name}"
+
+        return result
+
+    @property
+    def easyrsa_args(self) -> tuple[str]:
+        """
+        Pass this DN as arguments to easyrsa
+        """
+
+        return (
+            "--dn-mode=org",
+
+            f"--req-c={self.country}",
+            f"--req-st={self.state}",
+            f"--req-city={self.city}",
+            f"--req-org={self.organization}",
+            f"--req-ou={self.organizational_unit}",
+            f"--req-email={self.email}",
+            f"--req-cn={self.common_name}",
+        )
 
 
 class EasyRSA:
@@ -18,11 +103,19 @@ class EasyRSA:
     """
 
     @property
-    def pki_directory(self) -> Path:
+    def output_directory(self) -> Path:
+        """
+        Where certificates are stored
+        """
+
         return Settings._.data_dir.joinpath("pki")
 
     @property
     def ca_password(self) -> str:
+        """
+        Get CA password from config, or generate a new one
+        """
+
         config = Config._
 
         if (ca_password := config.crypto.ca_password) is None:
@@ -40,10 +133,14 @@ class EasyRSA:
         self,
         *easyrsa_args: str,
     ) -> subprocess.CompletedProcess:
+        """
+        Call the `easyrsa` executable
+        """
+
         return subprocess.run(
             [
                 "easyrsa", "--batch",
-                f"--pki-dir={self.pki_directory}",
+                f"--pki-dir={self.output_directory}",
                 *easyrsa_args,
             ],
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
@@ -53,81 +150,113 @@ class EasyRSA:
     def __build_cert(
         self,
         cert_filename: Path,
+        expiry_days: int | None,
         *easyrsa_args: str,
     ) -> crypto.X509:
-        self.__easyrsa(*easyrsa_args)
+        """
+        Create an X.509 certificate
+        """
+
+        config = Config._
+
+        extra_args: tuple[str] = (
+            f"--passout=pass:{self.ca_password}",
+            f"--passin=pass:{self.ca_password}",
+        )
+
+        if expiry_days is not None:
+            extra_args += tuple([f"--days={expiry_days}"])
+
+        if (algo := config.crypto.cert_algo) is not None:
+            if algo is CertificateAlgo.rsa2048:
+                extra_args += ("--use-algo=rsa", "--keysize=2048")
+
+            elif algo is CertificateAlgo.rsa4096:
+                extra_args += ("--use-algo=rsa", "--keysize=4096")
+
+            elif algo is CertificateAlgo.secp256r1:
+                extra_args += ("--use-algo=ec", "--curve=secp256r1")
+
+            elif algo is CertificateAlgo.secp384r1:
+                extra_args += ("--use-algo=ec", "--curve=secp384r1")
+
+            elif algo is CertificateAlgo.ed25519:
+                extra_args += ("--use-algo=ed", "--curve=ed25519")
+
+            else:
+                raise ValueError(f"Unexpected algorithm: {algo}")
+
+        self.__easyrsa(
+            *extra_args,
+            *easyrsa_args
+        )
 
         with open(
-            self.pki_directory.joinpath(cert_filename), "r"
+            self.output_directory.joinpath(cert_filename), "r"
         ) as cert_file:
             return crypto.load_certificate(
                 crypto.FILETYPE_PEM, cert_file.read()
             )
 
     def init_pki(self) -> bool:
+        """
+        Clean the working directory
+        """
+
         self.__easyrsa("init-pki")
 
-    def build_ca(
+    def build_ca(self) -> crypto.X509:
-        self,
+        """
-    ) -> crypto.X509:
+        Build the CA certificate
-        config = Config._
+        """
 
         cert = self.__build_cert(
             Path("ca.crt"),
+            Config._.crypto.ca_expiry_days,
 
-            f"--passout=pass:{self.ca_password}",
+            "--req-cn=kiwi-vpn-ca",
-            f"--passin=pass:{self.ca_password}",
-
-            # "--dn-mode=org",
-            # "--req-c=EX",
-            # "--req-st=EXAMPLE",
-            # "--req-city=EXAMPLE",
-            # "--req-org=EXAMPLE",
-            # "--req-ou=EXAMPLE",
-            # "--req-email=EXAMPLE",
-
-            f"--req-cn={config.server_name}",
-            f"--days={config.crypto.expiry_days}",
-
-            # "--use-algo=ed",
-            # "--curve=ed25519",
 
             "build-ca",
         )
 
+        # # this takes long!
         # self.__easyrsa("gen-dh")
         return cert
 
     def issue(
         self,
-        cn: str = "kiwi-vpn-client",
+        cert_type: str = "client",
-        cert_type: str = "client"
+        dn: DistinguishedName = DistinguishedName.build(),
     ) -> crypto.X509:
-        config = Config._
+        """
+        Issue a client or server certificate
+        """
 
         return self.__build_cert(
-            Path(f"issued/{cn}.crt"),
+            Path(f"issued/{dn.common_name}.crt"),
+            Config._.crypto.cert_expiry_days,
 
-            f"--passin=pass:{self.ca_password}",
+            *dn.easyrsa_args,
-            f"--days={config.crypto.expiry_days}",
 
             f"build-{cert_type}-full",
-            cn,
+            dn.common_name,
             "nopass",
         )
 
 
+# some basic test
+
 if __name__ == "__main__":
     easy_rsa = EasyRSA()
     easy_rsa.init_pki()
 
     ca = easy_rsa.build_ca()
-    server = easy_rsa.issue(cert_type="server", cn="kiwi-vpn-server")
+    server = easy_rsa.issue("server")
-    client = easy_rsa.issue(cert_type="client", cn="kiwi-vpn-client")
+    # client = easy_rsa.issue(cert_type="client", cn="kiwi-vpn-client")
 
     date_format, encoding = "%Y%m%d%H%M%SZ", "ascii"
 
-    for cert in [ca, server, client]:
+    for cert in (ca, server):
         print(cert.get_subject().CN)
         print(cert.get_signature_algorithm().decode(encoding))
         print(datetime.strptime(