<s> from __future__ import with_statement <EOL> import os <EOL> import re <EOL> import warnings <EOL> from . context import StreamPositionRestore <EOL> RE_KEY = re . compile ( "<STR_LIT>" <EOL> "<STR_LIT>" , re . S ) <EOL> RE_CERTS = re . compile ( "<STR_LIT>" <EOL> "<STR_LIT>" , re . S ) <EOL> __all__ = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def parse_keyfile ( keyfile ) : <EOL> """<STR_LIT>""" <EOL> with StreamPositionRestore ( keyfile ) : <EOL> kf = keyfile . read ( ) <EOL> key = RE_KEY . search ( kf ) <EOL> certs = RE_CERTS . finditer ( kf ) <EOL> return ( str ( key . group ( <NUM_LIT:0> ) ) , <EOL> str ( certs . next ( ) . group ( <NUM_LIT:0> ) ) , <EOL> ( str ( c . group ( <NUM_LIT:0> ) ) for c in certs ) <EOL> ) <EOL> try : <EOL> import subprocess <EOL> from tempfile import NamedTemporaryFile <EOL> from functools import wraps <EOL> try : <EOL> check_output = subprocess . check_output <EOL> except AttributeError : <EOL> def check_output ( * args , ** kw ) : <EOL> kw [ "<STR_LIT>" ] = subprocess . PIPE <EOL> return subprocess . Popen ( * args , ** kw ) . communicate ( ) [ <NUM_LIT:0> ] <EOL> def find_executable ( name ) : <EOL> """<STR_LIT>""" <EOL> is_windows = os . name != "<STR_LIT>" <EOL> def check ( path ) : <EOL> return ( os . path . isfile ( path ) and <EOL> ( not is_windows or os . access ( path , os . X_OK ) ) ) <EOL> if not is_windows and not name . lower ( ) . endswith ( "<STR_LIT>" ) : <EOL> name += "<STR_LIT>" <EOL> if check ( name ) : <EOL> return name <EOL> for cand in os . environ [ "<STR_LIT>" ] . split ( os . pathsep ) : <EOL> cand = os . path . join ( cand , name ) <EOL> if check ( cand ) : <EOL> return cand <EOL> return None <EOL> openssl = find_executable ( "<STR_LIT>" ) <EOL> if not openssl : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> def sign_openssl ( keyfile , content ) : <EOL> """<STR_LIT>""" <EOL> key , cs , stack = parse_keyfile ( keyfile ) <EOL> with NamedTemporaryFile ( ) as signer : <EOL> print >> signer , key <EOL> print >> signer , cs <EOL> signer . flush ( ) <EOL> with NamedTemporaryFile ( ) as certfile : <EOL> for c in stack : <EOL> print >> certfile , c <EOL> certfile . flush ( ) <EOL> with NamedTemporaryFile ( ) as infile : <EOL> infile . write ( content ) <EOL> infile . flush ( ) <EOL> return check_output ( ( openssl , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , signer . name , <EOL> "<STR_LIT>" , certfile . name , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , infile . name ) , <EOL> bufsize = <NUM_LIT:0> ) <EOL> @ wraps ( sign_openssl ) <EOL> def sign_openssl_warn ( * args , ** kw ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( openssl , ) , <EOL> RuntimeWarning <EOL> ) <EOL> return sign_openssl ( * args , ** kw ) <EOL> sign_openssl_warn . generator = sign_openssl . generator = check_output ( ( openssl , "<STR_LIT:version>" ) ) . strip ( ) <EOL> except ImportError : <EOL> sign_openssl_warn = sign_openssl = None <EOL> try : <EOL> import M2Crypto as M2 <EOL> import M2Crypto . SMIME as M2S <EOL> import M2Crypto . X509 as M2X509 <EOL> from M2Crypto . BIO import MemoryBuffer as M2Buffer <EOL> from M2Crypto . EVP import EVPError as M2EVPError <EOL> def sign_m2 ( keyfile , content ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( keyfile , "<STR_LIT:name>" ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> with NamedTemporaryFile ( ) as kp : <EOL> with StreamPositionRestore ( kp ) : <EOL> with StreamPositionRestore ( keyfile ) : <EOL> kp . write ( keyfile . read ( ) ) <EOL> return sign_m2 ( kp , content ) <EOL> try : <EOL> stack = M2X509 . X509_Stack ( ) <EOL> _ , _ , certificates = parse_keyfile ( keyfile ) <EOL> for c in certificates : <EOL> cert = M2X509 . load_cert_string ( c ) <EOL> if ( cert . check_ca ( ) <EOL> and str ( cert . get_issuer ( ) ) == str ( cert . get_subject ( ) ) ) : <EOL> continue <EOL> stack . push ( cert ) <EOL> smime = M2S . SMIME ( ) <EOL> smime . load_key ( keyfile . name ) <EOL> smime . set_x509_stack ( stack ) <EOL> pkcs7 = M2Buffer ( ) <EOL> smime . sign ( M2Buffer ( content ) , <EOL> M2S . PKCS7_DETACHED | M2S . PKCS7_BINARY <EOL> ) . write_der ( pkcs7 ) <EOL> return pkcs7 . read ( ) <EOL> except M2EVPError , ex : <EOL> if re . search ( "<STR_LIT>" , ex . message ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> sign_m2 . generator = "<STR_LIT>" % M2 . version <EOL> except ImportError : <EOL> sign_m2 = None <EOL> sign = sign_m2 or sign_openssl_warn <EOL> if not sign : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) </s>
