kiwi-vpn/api/kiwi_vpn_api/easyrsa.py

import subprocess
from datetime import datetime
from pathlib import Path

from OpenSSL import crypto
from passlib import pwd


class EasyRSA:
    __directory: Path | None
    __ca_password: str | None

    def __init__(self, directory: Path) -> None:
        self.__directory = directory

    def set_ca_password(self, password: str | None = None) -> None:
        if password is None:
            password = pwd.genword(length=32, charset="ascii_62")

        self.__ca_password = password
        print(self.__ca_password)

    def __easyrsa(
        self,
        *easyrsa_args: str,
    ) -> subprocess.CompletedProcess:
        return subprocess.run(
            [
                "easyrsa", "--batch",
                f"--pki-dir={self.__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.__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,
        days: int = 365 * 50,
        cn: str = "kiwi-vpn-ca"
    ) -> crypto.X509:
        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={cn}",
            f"--days={days}",

            # "--use-algo=ed",
            # "--curve=ed25519",

            "build-ca",
        )

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

    def issue(
        self,
        days: int = 365 * 50,
        cn: str = "kiwi-vpn-client",
        cert_type: str = "client"
    ) -> crypto.X509:
        return self.__build_cert(
            Path(f"issued/{cn}.crt"),

            f"--passin=pass:{self.__ca_password}",
            f"--days={days}",

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


if __name__ == "__main__":
    easy_rsa = EasyRSA(Path("tmp/easyrsa"))
    easy_rsa.init_pki()
    easy_rsa.set_ca_password()

    ca = easy_rsa.build_ca(cn="kiwi-vpn-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))