<s> '''<STR_LIT>''' <EOL> '''<STR_LIT>''' <EOL> from flask import g <EOL> from . . database import db_session , NoResultFound <EOL> from . . models import Certificate as DBCertificate , PrivateKey as DBPrivateKey , certificate_private_key_assoc <EOL> from m2certs import RSAPrivateKey , Certificate , CertificateRequest , Identity , NID_userId <EOL> MDM_CA_CN = '<STR_LIT>' <EOL> MDM_DEVICE_CN = '<STR_LIT>' <EOL> def get_ca ( ) : <EOL> ca = getattr ( g , '<STR_LIT>' , None ) <EOL> if ca is None : <EOL> ca = g . _mdm_ca = CA ( ) <EOL> return ca <EOL> class CA ( object ) : <EOL> def __init__ ( self ) : <EOL> try : <EOL> q = db_session . query ( DBCertificate , DBPrivateKey ) . join ( DBCertificate , DBPrivateKey . certificates ) . filter ( DBCertificate . cert_type == '<STR_LIT>' ) <EOL> cert , pk = q . one ( ) <EOL> self . ca_privkey = RSAPrivateKey . load ( str ( pk . pem_key ) ) <EOL> self . ca_cert = Certificate . load ( str ( cert . pem_certificate ) ) <EOL> except NoResultFound : <EOL> self . ca_privkey = RSAPrivateKey ( ) <EOL> ca_csr = mdm_cacert_csr = CertificateRequest ( self . ca_privkey , CN = MDM_CA_CN ) <EOL> self . ca_cert = Certificate . cacert_from_req ( ca_csr ) <EOL> db_ca_privkey = DBPrivateKey ( ) <EOL> db_ca_privkey . pem_key = self . ca_privkey . get_pem ( ) <EOL> db_session . add ( db_ca_privkey ) <EOL> db_ca_cert = DBCertificate ( ) <EOL> db_ca_cert . cert_type = '<STR_LIT>' <EOL> db_ca_cert . pem_certificate = self . ca_cert . get_pem ( ) <EOL> db_session . add ( db_ca_cert ) <EOL> db_ca_privkey . certificates . append ( db_ca_cert ) <EOL> db_session . commit ( ) <EOL> def get_cacert ( self ) : <EOL> return self . ca_cert <EOL> def gen_new_device_identity ( self ) : <EOL> '''<STR_LIT>''' <EOL> dev_key = RSAPrivateKey ( ) <EOL> dev_csr = CertificateRequest ( dev_key , CN = MDM_DEVICE_CN ) <EOL> dev_crt = Certificate . cert_from_req_signed_by_cacert ( dev_csr , self . ca_cert , self . ca_privkey ) <EOL> db_dev_crt = DBCertificate ( ) <EOL> db_dev_crt . cert_type = '<STR_LIT>' <EOL> db_dev_crt . pem_certificate = dev_crt . get_pem ( ) <EOL> db_session . add ( db_dev_crt ) <EOL> db_session . commit ( ) <EOL> return ( Identity ( dev_key , dev_crt ) , db_dev_crt ) <EOL> class PushCertificate ( Certificate ) : <EOL> def get_topic ( self ) : <EOL> x509_uid_name_entries = self . get_m2_cert ( ) . get_subject ( ) . get_entries_by_nid ( NID_userId ) <EOL> if len ( x509_uid_name_entries ) != <NUM_LIT:1> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> topic = x509_uid_name_entries [ <NUM_LIT:0> ] . get_data ( ) . as_text ( ) <EOL> return topic <EOL> def get_or_generate_web_certificate ( cn ) : <EOL> try : <EOL> q = db_session . query ( DBCertificate , DBPrivateKey ) . join ( DBCertificate , DBPrivateKey . certificates ) . filter ( DBCertificate . cert_type == '<STR_LIT>' ) <EOL> res = q . first ( ) <EOL> if not res : <EOL> res = q . one ( ) <EOL> db_web_crt , db_web_pk = res <EOL> return ( db_web_crt . pem_certificate , db_web_pk . pem_key ) <EOL> except NoResultFound : <EOL> web_pk = RSAPrivateKey ( ) <EOL> web_csr = CertificateRequest ( web_pk , CN = cn ) <EOL> web_crt = Certificate . cacert_from_req ( web_csr ) <EOL> db_web_pk = DBPrivateKey ( ) <EOL> db_web_pk . pem_key = web_pk . get_pem ( ) <EOL> db_session . add ( db_web_pk ) <EOL> db_web_crt = DBCertificate ( ) <EOL> db_web_crt . cert_type = '<STR_LIT>' <EOL> db_web_crt . pem_certificate = web_crt . get_pem ( ) <EOL> db_session . add ( db_web_crt ) <EOL> db_web_pk . certificates . append ( db_web_crt ) <EOL> db_session . commit ( ) <EOL> return ( web_crt . get_pem ( ) , web_pk . get_pem ( ) ) </s>
