@@ -1,150 +1,118 @@
import os
import socket
from OpenSSL import crypto , SSL
# OpenVPN is fairly simple since it works on OpenSSL. The OpenVPN server contains
# a root certificate authority that can sign sub-certificates. The certificates
# have very little or no information on who they belong to besides a filename
# and any required information. Everything else is omitted or blank.
# The client certificate and private key are inserted into the .ovpn file
# which contains some settins as well and the entire thing is then ready for
# the user.
# EasyRSA generates a standard unsigned certificate, certificate request, and private key.
# It then signs the certificate against the CA then dumps the certificate request in the trash.
# The now signed certificate and private key are returned.
# Many features of pyOpenSSL package are depricated. So we are using the cryptography library instead.
# pip install cryptography
from cryptography import x509
from cryptography .x509 .oid import NameOID
from cryptography .hazmat .primitives import hashes
from cryptography .hazmat .primitives .asymmetric import rsa
from cryptography .hazmat .primitives .serialization import Encoding , PrivateFormat , NoEncryption
from cryptography .hazmat .primitives import serialization # Fix: Import serialization
from datetime import datetime , timedelta , UTC # Fix: Import datetime and timedelta
# Create a new keypair of specified algorithm and number of bits.
def make_keypair (algorithm = crypto .TYPE_RSA , numbits = 2048 ):
pkey = crypto .PKey ()
pkey .generate_key (algorithm , numbits )
return pkey
def make_keypair (numbits = 2048 ):
return rsa .generate_private_key (public_exponent = 65537 , key_size = numbits )
# Creates a certificate signing request (CSR) given the specified subject attributes.
def make_csr (pkey , CN , C = None , ST = None , L = None , O = None , OU = None , emailAddress = None , hashalgorithm = 'sha256WithRSAEncryption' ):
req = crypto .X509Req ()
req .get_subject ()
subj = req .get_subject ()
def make_csr (private_key , CN , C = None , ST = None , L = None , O = None , OU = None , emailAddress = None ):
subject = []
if C :
subj . C = C
subject . append ( x509 . NameAttribute ( NameOID . COUNTRY_NAME , C ))
if ST :
subj . ST = ST
subject . append ( x509 . NameAttribute ( NameOID . STATE_OR_PROVINCE_NAME , ST ))
if L :
subj . L = L
subject . append ( x509 . NameAttribute ( NameOID . LOCALITY_NAME , L ))
if O :
subj . O = O
subject . append ( x509 . NameAttribute ( NameOID . ORGANIZATION_NAME , O ))
if OU :
subj . OU = OU
subject . append ( x509 . NameAttribute ( NameOID . ORGANIZATIONAL_UNIT_NAME , OU ))
if CN :
subj . CN = CN
subject . append ( x509 . NameAttribute ( NameOID . COMMON_NAME , CN ))
if emailAddress :
subj . emailAddress = emailAddress
subject . append ( x509 . NameAttribute ( NameOID . EMAIL_ADDRESS , emailAddress ))
req .set_pubkey (pkey )
req .sign (pkey , hashalgorithm )
return req
csr = x509 .CertificateSigningRequestBuilder ().subject_name (x509 .Name (subject )).sign (
private_key , hashes .SHA256 ()
)
return csr
# Create a certificate authority (if we need one)
def create_ca (CN , C = "" , ST = "" , L = "" , O = "" , OU = "" , emailAddress = "" , hashalgorithm = 'sha256WithRSAEncryption' ):
def create_ca (CN , C = "" , ST = "" , L = "" , O = "" , OU = "" , emailAddress = "" ):
cakey = make_keypair ()
careq = make_csr (cakey , cn = CN )
cacert = crypto .X509 ()
cacert .set_serial_number (0 )
cacert .gmtime_adj_notBefore (0 )
cacert .gmtime_adj_notAfter (60 * 60 * 24 * 365 * 10 ) # 10 yrs - hard to beat this kind of cert!
cacert .set_issuer (careq .get_subject ())
cacert .set_subject (careq .get_subject ())
cacert .set_pubkey (careq .get_pubkey ())
cacert .set_version (2 )
# Set the extensions in two passes
cacert .add_extensions ([
crypto .X509Extension ('basicConstraints' , True ,'CA:TRUE' ),
crypto .X509Extension ('subjectKeyIdentifier' , True , 'hash' , subject = cacert )
])
# ... now we can set the authority key since it depends on the subject key
cacert .add_extensions ([
crypto .X509Extension ('authorityKeyIdentifier' , False , 'issuer:always, keyid:always' , issuer = cacert , subject = cacert )
subject = x509 .Name ([
x509 .NameAttribute (NameOID .COUNTRY_NAME , C ),
x509 .NameAttribute (NameOID .STATE_OR_PROVINCE_NAME , ST ),
x509 .NameAttribute (NameOID .LOCALITY_NAME , L ),
x509 .NameAttribute (NameOID .ORGANIZATION_NAME , O ),
x509 .NameAttribute (NameOID .ORGANIZATIONAL_UNIT_NAME , OU ),
x509 .NameAttribute (NameOID .COMMON_NAME , CN ),
])
cacert .sign (cakey , hashalgorithm )
return (cacert , cakey )
cacert = (
x509 .CertificateBuilder ()
.subject_name (subject )
.issuer_name (subject )
.public_key (cakey .public_key ())
.serial_number (x509 .random_serial_number ())
.not_valid_before (datetime .utcnow ()) # Fix: Use datetime.utcnow()
.not_valid_after (datetime .utcnow () + timedelta (days = 3650 )) # Fix: Use timedelta
.add_extension (x509 .BasicConstraints (ca = True , path_length = None ), critical = True )
.add_extension (x509 .SubjectKeyIdentifier .from_public_key (cakey .public_key ()), critical = False )
.sign (cakey , hashes .SHA256 ())
)
return cacert , cakey
# Create a new slave cert.
def create_slave_certificate (csr , cakey , cacert , serial ):
cert = crypto .X509 ()
cert .set_serial_number (serial )
cert .gmtime_adj_notBefore (0 )
cert .gmtime_adj_notAfter (60 * 60 * 24 * 365 * 10 ) # 10 yrs - hard to beat this kind of cert!
cert .set_issuer (cacert .get_subject ())
cert .set_subject (csr .get_subject ())
cert .set_pubkey (csr .get_pubkey ())
cert .set_version (2 )
extensions = []
extensions .append (crypto .X509Extension ('basicConstraints' , False ,'CA:FALSE' ))
extensions .append (crypto .X509Extension ('subjectKeyIdentifier' , False , 'hash' , subject = cert ))
extensions .append (crypto .X509Extension ('authorityKeyIdentifier' , False , 'keyid:always,issuer:always' , subject = cacert , issuer = cacert ))
cert .add_extensions (extensions )
cert .sign (cakey , 'sha256WithRSAEncryption' )
cert = (
x509 .CertificateBuilder ()
.subject_name (csr .subject )
.issuer_name (cacert .subject )
.public_key (csr .public_key ())
.serial_number (serial )
.not_valid_before ( datetime .now (UTC )) # Fix: Use datetime.utcnow()
.not_valid_after ( datetime .now (UTC ) + timedelta (days = 3650 )) # Fix: Use timedelta
.add_extension (x509 .BasicConstraints (ca = False , path_length = None ), critical = True )
.add_extension (x509 .SubjectKeyIdentifier .from_public_key (csr .public_key ()), critical = False )
.add_extension (
x509 .AuthorityKeyIdentifier .from_issuer_public_key (cakey .public_key ()), critical = False
)
.sign (cakey , hashes .SHA256 ())
)
return cert
# Dumps content to a string
def dump_file_in_mem (material , format = crypto .FILETYPE_PEM ):
dump_func = None
if isinstance (material , crypto .X509 ):
dump_func = crypto .dump_certificate
elif isinstance (material , crypto .PKey ):
dump_func = crypto .dump_privatekey
elif isinstance (material , crypto .X509Req ):
dump_func = crypto .dump_certificate_request
def dump_file_in_mem (material ):
if isinstance (material , x509 .Certificate ):
return material .public_bytes (Encoding .PEM )
elif isinstance (material , rsa .RSAPrivateKey ):
return material .private_bytes (
Encoding .PEM , PrivateFormat .TraditionalOpenSSL , NoEncryption ()
)
elif isinstance (material , x509 .CertificateSigningRequest ):
return material .public_bytes (Encoding .PEM )
else :
raise Exception ("Don't know how to dump content type to file: %s (%r)" % ( type (material ), material ) )
return dump_func ( format , material )
# Loads the file into the appropriate openssl object type.
def load_from_file ( materialfile , objtype , format = crypto . FILETYPE_PEM ):
if objtype is crypto . X509 :
load_func = crypto . load_certificate
elif objtype is crypto . X509Req :
load_func = crypto . load_certificate_request
elif objtype is crypto . PKey :
load_func = crypto . load_privatekey
raise Exception (f "Don't know how to dump content type: { type (material )} "
# Loads the file into the appropriate cryptography object type.
def load_from_file ( materialfile , objtype ):
with open ( materialfile , "rb" ) as f :
data = f . read ()
if objtype == x509 . Certificate :
return x509 . load_pem_x509_certificate ( data )
elif objtype == x509 . CertificateSigningRequest :
return x509 . load_pem_x509_csr ( data )
elif objtype == rsa . RSAPrivateKey :
return serialization . load_pem_private_key ( data , password = None )
else :
raise Exception ("Unsupported material type: %s" % (objtype ,))
with open (materialfile , 'r' ) as fp :
buf = fp .read ()
material = load_func (format , buf )
return material
def retrieve_key_from_file (keyfile ):
return load_from_file (keyfile , crypto .PKey )
def retrieve_csr_from_file (csrfile ):
return load_from_file (csrfile , crypto .X509Req )
def retrieve_cert_from_file (certfile ):
return load_from_file (certfile , crypto .X509 )
raise Exception (f"Unsupported material type: { objtype } )
def make_new_ovpn_file (ca_cert , ca_key , clientname , serial , commonoptspath , filepath ):
# Read our common options file first
f = open (commonoptspath , 'r' )
common = f .read ()
f .close ()
with open (commonoptspath , "r" ) as f :
common = f .read ()
cacert = retrieve_cert_from_file (ca_cert )
cakey = retrieve_key_from_file (ca_key )
cacert = load_from_file (ca_cert , x509 . Certificate )
cakey = load_from_file (ca_key , rsa . RSAPrivateKey )
# Generate a new private key pair for a new certificate.
key = make_keypair ()
@@ -155,17 +123,19 @@ def make_new_ovpn_file(ca_cert, ca_key, clientname, serial, commonoptspath, file
# Now we have a successfully signed certificate. We must now
# create a .ovpn file and then dump it somewhere.
clientkey = dump_file_in_mem (key )
clientkey = dump_file_in_mem (key )
clientcert = dump_file_in_mem (crt )
cacertdump = dump_file_in_mem (cacert )
ovpn = "%s<ca>\n %s</ca>\n <cert>\n %s</cert>\n <key>\n %s</key>\n " % (common , cacertdump , clientcert , clientkey )
# Write our file.
f = open (filepath , 'w' )
f .write (ovpn )
f .close ()
ovpn = f"{ common } \n { cacertdump .decode ('utf-8' )} \n <cert>\n { clientcert .decode ('utf-8' )} \n <key>\n { clientkey .decode ('utf-8' )} \n "
# Write our file.
with open (filepath , "w" ) as f :
f .write (ovpn )
if __name__ == "__main__" :
# "common.txt" is the common client configuration file
# justasictest is the new client name
# justasictest.ovpn file will be generated with common config and necessary certificates
make_new_ovpn_file ("ca.crt" , "ca.key" , "justasictest" , 0x0C , "common.txt" , "justastictest.ovpn" )
print ("Done" )
borun revised this gist
Justin Crawford created this gist
Justin Crawford
created
this gist