kiwi-vpn/api/kiwi_vpn_api/easyrsa.py

"""
Python interface to EasyRSA CA.
"""

import subprocess
from datetime import datetime
from pathlib import Path

from OpenSSL import crypto
from passlib import pwd

from .config import Config, Settings


class EasyRSA:
    """
    Represents an EasyRSA PKI.
    """

    @property
    def pki_directory(self) -> Path:
        return Settings._.data_dir.joinpath("pki")

    @property
    def ca_password(self) -> str:
        config = Config._

        if (ca_password := config.crypto.ca_password) is None:
            ca_password = pwd.genword(
                length=32,
                charset="ascii_62",
            )

            config.crypto.ca_password = ca_password
            config.save()

        return config.crypto.ca_password

    def __easyrsa(
        self,
        *easyrsa_args: str,
    ) -> subprocess.CompletedProcess:
        return subprocess.run(
            [
                "easyrsa", "--batch",
                f"--pki-dir={self.pki_directory}",
                *easyrsa_args,
            ],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            check=True,
        )

    def __build_cert(
        self,
        cert_filename: Path,
        *easyrsa_args: str,
    ) -> crypto.X509:
        self.__easyrsa(*easyrsa_args)

        with open(
            self.pki_directory.joinpath(cert_filename), "r"
        ) as cert_file:
            return crypto.load_certificate(
                crypto.FILETYPE_PEM, cert_file.read()
            )

    def init_pki(self) -> bool:
        self.__easyrsa("init-pki")

    def build_ca(
        self,
    ) -> crypto.X509:
        config = Config._

        cert = self.__build_cert(
            Path("ca.crt"),

            f"--passout=pass:{self.ca_password}",
            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",
        )

        # self.__easyrsa("gen-dh")
        return cert

    def issue(
        self,
        cn: str = "kiwi-vpn-client",
        cert_type: str = "client"
    ) -> crypto.X509:
        config = Config._

        return self.__build_cert(
            Path(f"issued/{cn}.crt"),

            f"--passin=pass:{self.ca_password}",
            f"--days={config.crypto.expiry_days}",

            f"build-{cert_type}-full",
            cn,
            "nopass",
        )


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")
    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]:
        print(cert.get_subject().CN)
        print(cert.get_signature_algorithm().decode(encoding))
        print(datetime.strptime(
            cert.get_notAfter().decode(encoding), date_format))