<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import base64 <EOL> import binascii <EOL> import os <EOL> from Crypto . PublicKey import RSA <EOL> from cryptography import exceptions <EOL> from cryptography . hazmat import backends <EOL> from cryptography . hazmat . primitives . asymmetric import padding <EOL> from cryptography . hazmat . primitives import hashes <EOL> from cryptography . hazmat . primitives import serialization <EOL> from cryptography import x509 <EOL> from oslo_concurrency import processutils <EOL> from oslo_log import log as logging <EOL> from oslo_utils import excutils <EOL> from oslo_utils import fileutils <EOL> import paramiko <EOL> import six <EOL> import nova . conf <EOL> from nova import context <EOL> from nova import db <EOL> from nova import exception <EOL> from nova . i18n import _ , _LE <EOL> from nova import utils <EOL> LOG = logging . getLogger ( __name__ ) <EOL> CONF = nova . conf . CONF <EOL> def ca_folder ( project_id = None ) : <EOL> if CONF . crypto . use_project_ca and project_id : <EOL> return os . path . join ( CONF . crypto . ca_path , '<STR_LIT>' , project_id ) <EOL> return CONF . crypto . ca_path <EOL> def ca_path ( project_id = None ) : <EOL> return os . path . join ( ca_folder ( project_id ) , CONF . crypto . ca_file ) <EOL> def key_path ( project_id = None ) : <EOL> return os . path . join ( ca_folder ( project_id ) , CONF . crypto . key_file ) <EOL> def crl_path ( project_id = None ) : <EOL> return os . path . join ( ca_folder ( project_id ) , CONF . crypto . crl_file ) <EOL> def fetch_ca ( project_id = None ) : <EOL> if not CONF . crypto . use_project_ca : <EOL> project_id = None <EOL> ca_file_path = ca_path ( project_id ) <EOL> if not os . path . exists ( ca_file_path ) : <EOL> raise exception . CryptoCAFileNotFound ( project = project_id ) <EOL> with open ( ca_file_path , '<STR_LIT:r>' ) as cafile : <EOL> return cafile . read ( ) <EOL> def ensure_ca_filesystem ( ) : <EOL> """<STR_LIT>""" <EOL> ca_dir = ca_folder ( ) <EOL> if not os . path . exists ( ca_path ( ) ) : <EOL> genrootca_sh_path = os . path . abspath ( <EOL> os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> fileutils . ensure_tree ( ca_dir ) <EOL> utils . execute ( "<STR_LIT>" , genrootca_sh_path , cwd = ca_dir ) <EOL> def generate_fingerprint ( public_key ) : <EOL> try : <EOL> pub_bytes = public_key . encode ( '<STR_LIT:utf-8>' ) <EOL> serialization . load_ssh_public_key ( <EOL> pub_bytes , backends . default_backend ( ) ) <EOL> pub_data = base64 . b64decode ( public_key . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:1> ] ) <EOL> digest = hashes . Hash ( hashes . MD5 ( ) , backends . default_backend ( ) ) <EOL> digest . update ( pub_data ) <EOL> md5hash = digest . finalize ( ) <EOL> raw_fp = binascii . hexlify ( md5hash ) <EOL> if six . PY3 : <EOL> raw_fp = raw_fp . decode ( '<STR_LIT:ascii>' ) <EOL> return '<STR_LIT::>' . join ( a + b for a , b in zip ( raw_fp [ : : <NUM_LIT:2> ] , raw_fp [ <NUM_LIT:1> : : <NUM_LIT:2> ] ) ) <EOL> except Exception : <EOL> raise exception . InvalidKeypair ( <EOL> reason = _ ( '<STR_LIT>' ) ) <EOL> def generate_x509_fingerprint ( pem_key ) : <EOL> try : <EOL> if isinstance ( pem_key , six . text_type ) : <EOL> pem_key = pem_key . encode ( '<STR_LIT:utf-8>' ) <EOL> cert = x509 . load_pem_x509_certificate ( <EOL> pem_key , backends . default_backend ( ) ) <EOL> raw_fp = binascii . hexlify ( cert . fingerprint ( hashes . SHA1 ( ) ) ) <EOL> if six . PY3 : <EOL> raw_fp = raw_fp . decode ( '<STR_LIT:ascii>' ) <EOL> return '<STR_LIT::>' . join ( a + b for a , b in zip ( raw_fp [ : : <NUM_LIT:2> ] , raw_fp [ <NUM_LIT:1> : : <NUM_LIT:2> ] ) ) <EOL> except ( ValueError , TypeError , binascii . Error ) as ex : <EOL> raise exception . InvalidKeypair ( <EOL> reason = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % ex ) <EOL> def generate_key ( bits ) : <EOL> """<STR_LIT>""" <EOL> rsa = RSA . generate ( bits ) <EOL> key = paramiko . RSAKey ( vals = ( rsa . e , rsa . n ) ) <EOL> key . d = rsa . d <EOL> key . p = rsa . p <EOL> key . q = rsa . q <EOL> return key <EOL> def generate_key_pair ( bits = <NUM_LIT> ) : <EOL> key = generate_key ( bits ) <EOL> keyout = six . StringIO ( ) <EOL> key . write_private_key ( keyout ) <EOL> private_key = keyout . getvalue ( ) <EOL> public_key = '<STR_LIT>' % ( key . get_name ( ) , key . get_base64 ( ) ) <EOL> fingerprint = generate_fingerprint ( public_key ) <EOL> return ( private_key , public_key , fingerprint ) <EOL> def fetch_crl ( project_id ) : <EOL> """<STR_LIT>""" <EOL> if not CONF . crypto . use_project_ca : <EOL> project_id = None <EOL> crl_file_path = crl_path ( project_id ) <EOL> if not os . path . exists ( crl_file_path ) : <EOL> raise exception . CryptoCRLFileNotFound ( project = project_id ) <EOL> with open ( crl_file_path , '<STR_LIT:r>' ) as crlfile : <EOL> return crlfile . read ( ) <EOL> def decrypt_text ( project_id , text ) : <EOL> private_key_file = key_path ( project_id ) <EOL> if not os . path . exists ( private_key_file ) : <EOL> raise exception . ProjectNotFound ( project_id = project_id ) <EOL> with open ( private_key_file , '<STR_LIT:rb>' ) as f : <EOL> data = f . read ( ) <EOL> try : <EOL> priv_key = serialization . load_pem_private_key ( <EOL> data , None , backends . default_backend ( ) ) <EOL> return priv_key . decrypt ( text , padding . PKCS1v15 ( ) ) <EOL> except ( ValueError , TypeError , exceptions . UnsupportedAlgorithm ) as exc : <EOL> raise exception . DecryptionFailure ( reason = six . text_type ( exc ) ) <EOL> def ssh_encrypt_text ( ssh_public_key , text ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( text , six . text_type ) : <EOL> text = text . encode ( '<STR_LIT:utf-8>' ) <EOL> try : <EOL> pub_bytes = ssh_public_key . encode ( '<STR_LIT:utf-8>' ) <EOL> pub_key = serialization . load_ssh_public_key ( <EOL> pub_bytes , backends . default_backend ( ) ) <EOL> return pub_key . encrypt ( text , padding . PKCS1v15 ( ) ) <EOL> except Exception as exc : <EOL> raise exception . EncryptionFailure ( reason = six . text_type ( exc ) ) <EOL> def revoke_cert ( project_id , file_name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> file_name , cwd = ca_folder ( project_id ) ) <EOL> utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , CONF . crypto . crl_file , cwd = ca_folder ( project_id ) ) <EOL> except OSError : <EOL> raise exception . ProjectNotFound ( project_id = project_id ) <EOL> except processutils . ProcessExecutionError : <EOL> raise exception . RevokeCertFailure ( project_id = project_id ) <EOL> def revoke_certs_by_user ( user_id ) : <EOL> """<STR_LIT>""" <EOL> admin = context . get_admin_context ( ) <EOL> for cert in db . certificate_get_all_by_user ( admin , user_id ) : <EOL> revoke_cert ( cert [ '<STR_LIT>' ] , cert [ '<STR_LIT>' ] ) <EOL> def revoke_certs_by_project ( project_id ) : <EOL> """<STR_LIT>""" <EOL> admin = context . get_admin_context ( ) <EOL> for cert in db . certificate_get_all_by_project ( admin , project_id ) : <EOL> revoke_cert ( cert [ '<STR_LIT>' ] , cert [ '<STR_LIT>' ] ) <EOL> def revoke_certs_by_user_and_project ( user_id , project_id ) : <EOL> """<STR_LIT>""" <EOL> admin = context . get_admin_context ( ) <EOL> for cert in db . certificate_get_all_by_user_and_project ( admin , <EOL> user_id , project_id ) : <EOL> revoke_cert ( cert [ '<STR_LIT>' ] , cert [ '<STR_LIT>' ] ) <EOL> def _project_cert_subject ( project_id ) : <EOL> """<STR_LIT>""" <EOL> return CONF . crypto . project_cert_subject % ( project_id , utils . isotime ( ) ) <EOL> def _user_cert_subject ( user_id , project_id ) : <EOL> """<STR_LIT>""" <EOL> return CONF . crypto . user_cert_subject % ( project_id , user_id , <EOL> utils . isotime ( ) ) <EOL> def generate_x509_cert ( user_id , project_id , bits = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> subject = _user_cert_subject ( user_id , project_id ) <EOL> with utils . tempdir ( ) as tmpdir : <EOL> keyfile = os . path . abspath ( os . path . join ( tmpdir , '<STR_LIT>' ) ) <EOL> csrfile = os . path . abspath ( os . path . join ( tmpdir , '<STR_LIT>' ) ) <EOL> utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , keyfile , str ( bits ) ) <EOL> utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , keyfile , '<STR_LIT>' , <EOL> csrfile , '<STR_LIT>' , '<STR_LIT>' , subject ) <EOL> with open ( keyfile ) as f : <EOL> private_key = f . read ( ) <EOL> with open ( csrfile ) as f : <EOL> csr = f . read ( ) <EOL> ( serial , signed_csr ) = sign_csr ( csr , project_id ) <EOL> fname = os . path . join ( ca_folder ( project_id ) , '<STR_LIT>' % serial ) <EOL> cert = { '<STR_LIT>' : user_id , <EOL> '<STR_LIT>' : project_id , <EOL> '<STR_LIT>' : fname } <EOL> db . certificate_create ( context . get_admin_context ( ) , cert ) <EOL> return ( private_key , signed_csr ) <EOL> def generate_winrm_x509_cert ( user_id , bits = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> subject = '<STR_LIT>' % user_id <EOL> upn = '<STR_LIT>' % user_id <EOL> with utils . tempdir ( ) as tmpdir : <EOL> keyfile = os . path . abspath ( os . path . join ( tmpdir , '<STR_LIT>' ) ) <EOL> conffile = os . path . abspath ( os . path . join ( tmpdir , '<STR_LIT>' ) ) <EOL> _create_x509_openssl_config ( conffile , upn ) <EOL> ( certificate , _err ) = utils . execute ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , conffile , '<STR_LIT>' , '<STR_LIT>' % bits , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , keyfile , '<STR_LIT>' , subject , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> binary = True ) <EOL> ( out , _err ) = utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , keyfile , '<STR_LIT>' , '<STR_LIT>' , <EOL> process_input = certificate , <EOL> binary = True ) <EOL> private_key = base64 . b64encode ( out ) <EOL> fingerprint = generate_x509_fingerprint ( certificate ) <EOL> if six . PY3 : <EOL> private_key = private_key . decode ( '<STR_LIT:ascii>' ) <EOL> certificate = certificate . decode ( '<STR_LIT:utf-8>' ) <EOL> return ( private_key , certificate , fingerprint ) <EOL> def _create_x509_openssl_config ( conffile , upn ) : <EOL> content = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" "<STR_LIT>" ) <EOL> with open ( conffile , '<STR_LIT:w>' ) as file : <EOL> file . write ( content % upn ) <EOL> def _ensure_project_folder ( project_id ) : <EOL> if not os . path . exists ( ca_path ( project_id ) ) : <EOL> geninter_sh_path = os . path . abspath ( <EOL> os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> utils . execute ( '<STR_LIT>' , geninter_sh_path , project_id , <EOL> _project_cert_subject ( project_id ) , cwd = ca_folder ( ) ) <EOL> def generate_vpn_files ( project_id ) : <EOL> project_folder = ca_folder ( project_id ) <EOL> key_fn = os . path . join ( project_folder , '<STR_LIT>' ) <EOL> crt_fn = os . path . join ( project_folder , '<STR_LIT>' ) <EOL> if os . path . exists ( crt_fn ) : <EOL> return <EOL> key , csr = generate_x509_cert ( '<STR_LIT>' , project_id , <NUM_LIT> ) <EOL> with open ( key_fn , '<STR_LIT:w>' ) as keyfile : <EOL> keyfile . write ( key ) <EOL> with open ( crt_fn , '<STR_LIT:w>' ) as crtfile : <EOL> crtfile . write ( csr ) <EOL> def sign_csr ( csr_text , project_id = None ) : <EOL> if not CONF . crypto . use_project_ca : <EOL> project_id = None <EOL> if not project_id : <EOL> return _sign_csr ( csr_text , ca_folder ( ) ) <EOL> _ensure_project_folder ( project_id ) <EOL> return _sign_csr ( csr_text , ca_folder ( project_id ) ) <EOL> def _sign_csr ( csr_text , ca_folder ) : <EOL> with utils . tempdir ( ) as tmpdir : <EOL> inbound = os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> outbound = os . path . join ( tmpdir , '<STR_LIT>' ) <EOL> try : <EOL> with open ( inbound , '<STR_LIT:w>' ) as csrfile : <EOL> csrfile . write ( csr_text ) <EOL> except IOError : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . exception ( _LE ( '<STR_LIT>' ) ) <EOL> LOG . debug ( '<STR_LIT>' , ca_folder ) <EOL> fileutils . ensure_tree ( ca_folder ) <EOL> utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , outbound , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , inbound , cwd = ca_folder ) <EOL> out , _err = utils . execute ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , outbound , <EOL> '<STR_LIT>' , '<STR_LIT>' , cwd = ca_folder ) <EOL> serial = out . rpartition ( '<STR_LIT:=>' ) [ <NUM_LIT:2> ] . strip ( ) <EOL> with open ( outbound , '<STR_LIT:r>' ) as crtfile : <EOL> return ( serial , crtfile . read ( ) ) </s>
