kiwi-vpn/api/kiwi_vpn_api/easyrsa.py

319 lines
8.2 KiB
Python
Raw Normal View History

2022-03-30 21:18:54 +00:00
"""
Python interface to EasyRSA CA.
"""
from __future__ import annotations
2022-03-22 00:34:06 +00:00
import subprocess
from datetime import datetime
2022-04-01 15:39:48 +00:00
from enum import Enum, auto
2022-03-22 00:34:06 +00:00
from pathlib import Path
from OpenSSL import crypto
from passlib import pwd
from pydantic import BaseModel
2022-03-22 00:34:06 +00:00
2022-03-31 16:59:14 +00:00
from .config import Config, KeyAlgorithm, Settings
2022-03-30 23:59:25 +00:00
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
2022-03-31 16:34:36 +00:00
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
2022-04-02 00:08:27 +00:00
def easyrsa_env(self) -> dict[str, str]:
"""
Pass this DN as arguments to easyrsa
"""
2022-04-02 00:08:27 +00:00
return {
"EASYRSA_DN": "org",
2022-04-02 00:08:27 +00:00
"EASYRSA_REQ_COUNTRY": self.country,
"EASYRSA_REQ_PROVINCE": self.state,
"EASYRSA_REQ_CITY": self.city,
"EASYRSA_REQ_ORG": self.organization,
"EASYRSA_REQ_OU": self.organizational_unit,
"EASYRSA_REQ_EMAIL": self.email,
"EASYRSA_REQ_CN": self.common_name,
}
2022-03-22 00:34:06 +00:00
2022-04-01 15:39:48 +00:00
class CertificateType(Enum):
"""
Possible types of certificates
"""
ca = auto()
client = auto()
server = auto()
def __str__(self) -> str:
return self._name_
2022-03-30 21:18:54 +00:00
class EasyRSA:
"""
Represents an EasyRSA PKI.
"""
2022-04-02 00:08:27 +00:00
__mapKeyAlgorithm = {
KeyAlgorithm.rsa2048: {
"EASYRSA_ALGO": "rsa",
"EASYRSA_KEY_SIZE": "2048",
},
KeyAlgorithm.rsa4096: {
"EASYRSA_ALGO": "rsa",
"EASYRSA_KEY_SIZE": "2048",
},
KeyAlgorithm.secp256r1: {
"EASYRSA_ALGO": "ec",
"EASYRSA_CURVE": "secp256r1",
},
KeyAlgorithm.secp384r1: {
"EASYRSA_ALGO": "ec",
"EASYRSA_CURVE": "secp384r1",
},
KeyAlgorithm.ed25519: {
"EASYRSA_ALGO": "ed",
"EASYRSA_CURVE": "ed25519",
},
None: {},
}
2022-03-30 21:18:54 +00:00
@property
def output_directory(self) -> Path:
"""
Where certificates are stored
"""
2022-03-30 21:18:54 +00:00
return Settings._.data_dir.joinpath("pki")
@property
def ca_password(self) -> str:
"""
Get CA password from config, or generate a new one
"""
2022-03-30 21:18:54 +00:00
config = Config._
if (ca_password := config.crypto.ca_password) is None:
2022-03-31 23:15:49 +00:00
# generate and save new CA password
2022-03-30 21:18:54 +00:00
ca_password = pwd.genword(
length=32,
charset="ascii_62",
)
2022-03-22 00:34:06 +00:00
2022-03-30 21:18:54 +00:00
config.crypto.ca_password = ca_password
config.save()
2022-03-22 00:34:06 +00:00
2022-03-30 21:18:54 +00:00
return config.crypto.ca_password
2022-03-22 00:34:06 +00:00
def __easyrsa(
self,
2022-04-02 00:08:27 +00:00
*easyrsa_cmd: str,
**easyrsa_env: str,
2022-03-22 00:34:06 +00:00
) -> subprocess.CompletedProcess:
"""
Call the `easyrsa` executable
"""
2022-03-22 00:34:06 +00:00
return subprocess.run(
[
2022-04-02 00:08:27 +00:00
"/usr/local/bin/easyrsa",
*easyrsa_cmd,
2022-03-22 00:34:06 +00:00
],
2022-04-02 00:08:27 +00:00
env={
"EASYRSA_BATCH": "1",
"EASYRSA_PKI": str(self.output_directory),
**easyrsa_env,
},
2022-03-22 00:34:06 +00:00
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
check=True,
)
def __build_cert(
self,
cert_filename: Path,
2022-04-02 00:08:27 +00:00
*easyrsa_cmd: str,
**easyrsa_env: str,
2022-03-22 00:34:06 +00:00
) -> crypto.X509:
"""
Create an X.509 certificate
"""
2022-03-30 22:27:17 +00:00
config = Config._
2022-04-02 00:08:27 +00:00
if ((algorithm := config.crypto.key_algorithm)
not in EasyRSA.__mapKeyAlgorithm):
raise ValueError(f"Unexpected algorithm: {algorithm}")
# include expiry options
if (ca_expiry_days := config.crypto.ca_expiry_days) is not None:
easyrsa_env["EASYRSA_CA_EXPIRE"] = str(ca_expiry_days)
if (cert_expiry_days := config.crypto.cert_expiry_days) is not None:
easyrsa_env["EASYRSA_CERT_EXPIRE"] = str(cert_expiry_days)
2022-03-30 22:27:17 +00:00
2022-03-31 23:15:49 +00:00
# call easyrsa
2022-03-30 22:27:17 +00:00
self.__easyrsa(
2022-04-02 00:08:27 +00:00
*easyrsa_cmd,
# include CA password
EASYRSA_PASSOUT=f"pass:{self.ca_password}",
EASYRSA_PASSIN=f"pass:{self.ca_password}",
# include algorithm options
**EasyRSA.__mapKeyAlgorithm[algorithm],
**easyrsa_env,
2022-03-30 22:27:17 +00:00
)
2022-03-22 00:34:06 +00:00
2022-03-31 23:15:49 +00:00
# parse the new certificate
2022-03-22 00:34:06 +00:00
with open(
2022-03-31 16:32:07 +00:00
self.output_directory.joinpath(cert_filename), "rb"
2022-03-22 00:34:06 +00:00
) as cert_file:
return crypto.load_certificate(
crypto.FILETYPE_PEM, cert_file.read()
)
2022-03-31 16:32:07 +00:00
def init_pki(self) -> None:
"""
2022-03-31 23:15:49 +00:00
Clean working directory
"""
2022-03-22 00:34:06 +00:00
self.__easyrsa("init-pki")
2022-03-30 22:27:17 +00:00
def build_ca(self) -> crypto.X509:
"""
Build the CA certificate
"""
2022-03-30 21:18:54 +00:00
2022-03-24 23:27:35 +00:00
cert = self.__build_cert(
2022-03-22 00:34:06 +00:00
Path("ca.crt"),
"build-ca",
2022-04-02 00:08:27 +00:00
EASYRSA_DN="cn_only",
EASYRSA_REQ_CN="kiwi-vpn-ca",
2022-03-22 00:34:06 +00:00
)
# # this takes long!
2022-03-30 21:18:54 +00:00
# self.__easyrsa("gen-dh")
2022-03-24 23:27:35 +00:00
return cert
2022-03-22 00:34:06 +00:00
def issue(
self,
2022-04-01 15:39:48 +00:00
cert_type: CertificateType = CertificateType.client,
dn: DistinguishedName | None = None,
2022-04-01 15:39:48 +00:00
) -> crypto.X509 | None:
"""
Issue a client or server certificate
"""
2022-03-30 21:18:54 +00:00
if dn is None:
dn = DistinguishedName.build()
2022-04-01 15:39:48 +00:00
if not (cert_type is CertificateType.client
or cert_type is CertificateType.server):
return None
2022-03-22 00:34:06 +00:00
return self.__build_cert(
Path(f"issued/{dn.common_name}.crt"),
2022-03-22 00:34:06 +00:00
f"build-{cert_type}-full",
dn.common_name,
2022-03-22 00:34:06 +00:00
"nopass",
2022-04-02 00:08:27 +00:00
**dn.easyrsa_env,
2022-03-22 00:34:06 +00:00
)
# some basic test
2022-03-22 00:34:06 +00:00
if __name__ == "__main__":
2022-03-30 21:18:54 +00:00
easy_rsa = EasyRSA()
2022-03-22 00:57:09 +00:00
easy_rsa.init_pki()
2022-03-22 00:34:06 +00:00
2022-03-30 21:18:54 +00:00
ca = easy_rsa.build_ca()
2022-04-01 15:39:48 +00:00
server = easy_rsa.issue(CertificateType.server)
2022-03-30 23:59:25 +00:00
client = None
# check if configured
2022-03-31 16:32:07 +00:00
if (current_config := Config.load()) is not None:
2022-03-30 23:59:25 +00:00
# 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)
2022-04-01 15:39:48 +00:00
client = easy_rsa.issue(dn=dn)
2022-03-22 00:34:06 +00:00
date_format, encoding = "%Y%m%d%H%M%SZ", "ascii"
2022-03-22 00:57:09 +00:00
2022-03-30 23:59:25 +00:00
for cert in (ca, server, client):
if cert is not None:
print(cert.get_subject().CN)
print(cert.get_signature_algorithm().decode(encoding))
2022-03-31 16:32:07 +00:00
assert (na := cert.get_notAfter()) is not None
print(datetime.strptime(na.decode(encoding), date_format))