kiwi-vpn/api/kiwi_vpn_api/easyrsa.py
2022-03-31 16:59:14 +00:00

282 lines
7.2 KiB
Python

"""
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, KeyAlgorithm, Settings
from .db import Connection, 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) -> list[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:
"""
Represents an EasyRSA PKI.
"""
@property
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:
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:
"""
Call the `easyrsa` executable
"""
return subprocess.run(
[
"easyrsa", "--batch",
f"--pki-dir={self.output_directory}",
*easyrsa_args,
],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
check=True,
)
def __build_cert(
self,
cert_filename: Path,
expiry_days: int | None,
*easyrsa_args: str,
) -> crypto.X509:
"""
Create an X.509 certificate
"""
config = Config._
extra_args: list[str] = [
f"--passout=pass:{self.ca_password}",
f"--passin=pass:{self.ca_password}",
]
if expiry_days is not None:
extra_args += [f"--days={expiry_days}"]
if (algorithm := config.crypto.key_algorithm) is not None:
args_map = {
KeyAlgorithm.rsa2048: [
"--use-algo=rsa", "--keysize=2048"
],
KeyAlgorithm.rsa2048: [
"--use-algo=rsa", "--keysize=2048"
],
KeyAlgorithm.secp256r1: [
"--use-algo=ec", "--curve=secp256r1"
],
KeyAlgorithm.secp384r1: [
"--use-algo=ec", "--curve=secp384r1"
],
KeyAlgorithm.ed25519: [
"--use-algo=ed", "--curve=ed25519"
]
}
if algorithm not in args_map:
raise ValueError(f"Unexpected algorithm: {algorithm}")
extra_args += args_map[algorithm]
self.__easyrsa(
*extra_args,
*easyrsa_args
)
with open(
self.output_directory.joinpath(cert_filename), "rb"
) as cert_file:
return crypto.load_certificate(
crypto.FILETYPE_PEM, cert_file.read()
)
def init_pki(self) -> None:
"""
Clean the working directory
"""
self.__easyrsa("init-pki")
def build_ca(self) -> crypto.X509:
"""
Build the CA certificate
"""
cert = self.__build_cert(
Path("ca.crt"),
Config._.crypto.ca_expiry_days,
"--req-cn=kiwi-vpn-ca",
"build-ca",
)
# # this takes long!
# self.__easyrsa("gen-dh")
return cert
def issue(
self,
cert_type: str = "client",
dn: DistinguishedName = DistinguishedName.build(),
) -> crypto.X509:
"""
Issue a client or server certificate
"""
return self.__build_cert(
Path(f"issued/{dn.common_name}.crt"),
Config._.crypto.cert_expiry_days,
*dn.easyrsa_args,
f"build-{cert_type}-full",
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("server")
client = None
# check if configured
if (current_config := Config.load()) is not None:
# connect to database
Connection.connect(current_config.db.uri)
if (device := Device.get(1)) is not None:
with Connection.session as db:
db.add(device)
dn = DistinguishedName.build(device)
client = easy_rsa.issue("client", dn)
date_format, encoding = "%Y%m%d%H%M%SZ", "ascii"
for cert in (ca, server, client):
if cert is not None:
print(cert.get_subject().CN)
print(cert.get_signature_algorithm().decode(encoding))
assert (na := cert.get_notAfter()) is not None
print(datetime.strptime(na.decode(encoding), date_format))