ED25519 support for mTLS helpers

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.2.3
+current_version = 2.3.0
 commit = False
 tag = False
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "libpvarki"
-version = "2.2.3"
+version = "2.3.0"
 description = "Common helpers like standard logging init"
 authors = ["Eero af Heurlin <eero.afheurlin@iki.fi>"]
 homepage = "https://github.com/pvarki/python-libpvarki/"

src/libpvarki/__init__.py

@@ -1,3 +1,3 @@
 """Common helpers like standard logging init"""
 
-__version__ = "2.2.3"  # NOTE Use `bump2version --config-file patch` to bump versions correctly
+__version__ = "2.3.0"  # NOTE Use `bump2version --config-file patch` to bump versions correctly

src/libpvarki/mtlshelp/csr.py

@@ -1,6 +1,6 @@
 """Create keys and CSRs"""
 
-from typing import Mapping, Sequence, Tuple, Iterable
+from typing import Mapping, Sequence, Tuple, Iterable, Optional
 from pathlib import Path
 import logging
 import stat
@@ -9,12 +9,12 @@
 
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes, serialization
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import rsa, ed25519
 from cryptography.x509.oid import NameOID
 from cryptography.x509.name import _NAME_TO_NAMEOID
 
 LOGGER = logging.getLogger(__name__)
-KPTYPE = rsa.RSAPrivateKey  # TODO: should this be more than a type alias?
+KPTYPE = rsa.RSAPrivateKey | ed25519.Ed25519PrivateKey
 PUBDIR_MODE = stat.S_IRWXU | stat.S_IRGRP | stat.S_IROTH | stat.S_IXGRP | stat.S_IXOTH
 PRIVDIR_MODE = stat.S_IRWXU
 
@@ -44,9 +44,12 @@ def resolve_filepaths(basedir: Path, nameprefix: str) -> Tuple[Path, Path, Path]
     return privkeypath, pubkeypath, csrpath
 
 
-def create_keypair(privkeypath: Path, pubkeypath: Path, ktype: str = "RSA", ksize: int = 4096) -> rsa.RSAPrivateKey:
+def create_keypair(privkeypath: Path, pubkeypath: Path, ktype: str = "RSA", ksize: int = 4096) -> KPTYPE:
     """Generate a keypair, saves files to given paths (directory must exist) and returns the
-    keypair object"""
+    keypair object
+
+    keysize only used for RSA
+    """
     for check_path in (privkeypath, pubkeypath):
         if not check_path.parent.exists():
             LOGGER.error("Path {} does not exist".format(check_path.parent))
@@ -56,9 +59,13 @@ def create_keypair(privkeypath: Path, pubkeypath: Path, ktype: str = "RSA", ksiz
             raise ValueError("Invalid path {}".format(check_path))
         if check_path.exists():
             LOGGER.warning("{} already exists, it will be overwritten".format(check_path))
-    LOGGER.info("Generating {} keypair of size {}, this will take a moment".format(ktype, ksize))
+    ckp: Optional[KPTYPE] = None
     if ktype == "RSA":
+        LOGGER.info("Generating {} keypair of size {}, this will take a moment".format(ktype, ksize))
         ckp = rsa.generate_private_key(public_exponent=65537, key_size=ksize)
+    elif ktype == "ED25519":
+        LOGGER.info("Generating {} keypair".format(ktype))
+        ckp = ed25519.Ed25519PrivateKey.generate()
     else:
         raise NotImplementedError(f"Key type {ktype} not supported")
     LOGGER.info("Keygen done")
@@ -86,19 +93,26 @@ async def async_create_keypair(
     pubkeypath: Path,
     ktype: str = "RSA",
     ksize: int = 4096,
-) -> rsa.RSAPrivateKey:
+) -> KPTYPE:
     """Async wrapper for create_keypair see it for details"""
     return await asyncio.get_event_loop().run_in_executor(None, create_keypair, privkeypath, pubkeypath, ktype, ksize)
 
 
 def sign_and_write_csrfile(
     builder: x509.CertificateSigningRequestBuilder,
-    keypair: rsa.RSAPrivateKey,
+    keypair: KPTYPE,
     csrpath: Path,
     digest: str,
 ) -> str:
-    """internal helper to be more DRY, returns the PEM"""
-    csr = builder.sign(keypair, HASHER_MAP[digest.lower()]())
+    """internal helper to be more DRY, returns the PEM
+
+    digest only used for RSA; also must be one of HASHER_MAP keys
+    """
+    if isinstance(keypair, ed25519.Ed25519PrivateKey):
+        # Algorithm must be None when signing via ed25519 or ed448
+        csr = builder.sign(keypair, None)
+    else:
+        csr = builder.sign(keypair, HASHER_MAP[digest.lower()]())
     csr_pem = csr.public_bytes(serialization.Encoding.PEM)
     csrpath.write_bytes(csr_pem)
     csrpath.chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)  # everyone can read
@@ -144,7 +158,7 @@ def _build_csr_builder(
 
 
 def create_client_csr(
-    keypair: rsa.RSAPrivateKey,
+    keypair: KPTYPE,
     csrpath: Path,
     reqdn: Mapping[str, str],
     digest: str = "sha256",
@@ -164,7 +178,7 @@ def create_client_csr(
 
 
 async def async_create_client_csr(
-    keypair: rsa.RSAPrivateKey,
+    keypair: KPTYPE,
     csrpath: Path,
     reqdn: Mapping[str, str],
     digest: str = "sha256",
@@ -173,7 +187,7 @@ async def async_create_client_csr(
     return await asyncio.get_event_loop().run_in_executor(None, create_client_csr, keypair, csrpath, reqdn, digest)
 
 
-def create_server_csr(keypair: rsa.RSAPrivateKey, csrpath: Path, names: Sequence[str], digest: str = "sha256") -> str:
+def create_server_csr(keypair: KPTYPE, csrpath: Path, names: Sequence[str], digest: str = "sha256") -> str:
     """Generate CSR file with serverAuth extended usage, returns the PEM encoded contents
     First name will go to CN, all names will go to subjectAltNames
 
@@ -194,7 +208,7 @@ def create_server_csr(keypair: rsa.RSAPrivateKey, csrpath: Path, names: Sequence
 
 
 async def async_create_server_csr(
-    keypair: rsa.RSAPrivateKey,
+    keypair: KPTYPE,
     csrpath: Path,
     names: Sequence[str],
     digest: str = "sha256",

tests/mtls/test_helpers.py

@@ -63,33 +63,35 @@ def create_subdirs(datadir: Path) -> Tuple[Path, Path]:
     return privpath, pubpath
 
 
-def test_keypair_create(tmp_path: Path) -> None:
+@pytest.mark.parametrize("ktype", ("RSA", "ED25519"))
+def test_keypair_create(tmp_path: Path, ktype: str) -> None:
     """Test normal create case"""
     privpath, pubpath = create_subdirs(tmp_path)
-    ckp = create_keypair(privpath, pubpath, ksize=1024)  # small key to save time
+    ckp = create_keypair(privpath, pubpath, ktype=ktype, ksize=1024)  # small key to save time
     check_keypair(ckp, privpath, pubpath)
 
 
 @pytest.mark.asyncio
-async def test_keypair_create_async(tmp_path: Path) -> None:
+@pytest.mark.parametrize("ktype", ("RSA", "ED25519"))
+async def test_keypair_create_async(tmp_path: Path, ktype: str) -> None:
     """Test the async wrapper"""
     privpath, pubpath = create_subdirs(tmp_path)
-    ckp = await async_create_keypair(privpath, pubpath, ksize=1024)  # small key to save time
+    ckp = await async_create_keypair(privpath, pubpath, ktype=ktype, ksize=1024)  # small key to save time
     check_keypair(ckp, privpath, pubpath)
 
 
-@pytest_asyncio.fixture
-async def keypair(tmp_path: Path) -> AsyncGenerator[Tuple[KPTYPE, Path, Path], None]:
+@pytest_asyncio.fixture(params=["RSA", "ED25519"])
+async def keypair(tmp_path: Path, request: pytest.FixtureRequest) -> AsyncGenerator[Tuple[KPTYPE, Path, Path], None]:
     """Fixture to create keypair"""
     privpath, pubpath = create_subdirs(tmp_path)
-    ckp = await async_create_keypair(privpath, pubpath, ksize=1024)  # small key to save time
+    ckp = await async_create_keypair(privpath, pubpath, ktype=request.param, ksize=1024)  # small key to save time
     check_keypair(ckp, privpath, pubpath)
     yield ckp, privpath, pubpath
 
 
 def check_csr(pemdata: str, expect_cn: str) -> None:
     """Check the CSR"""
-    assert pemdata.startswith("-----BEGIN CERTIFICATE REQUEST-----\nMII")
+    assert pemdata.startswith("-----BEGIN CERTIFICATE REQUEST-----\n")
     parsed = cryptography.x509.load_pem_x509_csr(pemdata.encode("utf-8"))
     dname = parsed.subject.rfc4514_string()
     LOGGER.debug("dname: {}".format(dname))

tests/test_libpvarki.py

@@ -5,4 +5,4 @@
 
 def test_version() -> None:
     """Make sure version matches expected"""
-    assert __version__ == "2.2.3"
+    assert __version__ == "2.3.0"