<s> """<STR_LIT>""" <EOL> from __future__ import with_statement <EOL> __version__ = '<STR_LIT>' <EOL> import re <EOL> import blinker <EOL> import smtplib <EOL> import sys <EOL> import time <EOL> import unicodedata <EOL> from email import charset <EOL> from email . encoders import encode_base64 <EOL> from email . mime . base import MIMEBase <EOL> from email . mime . multipart import MIMEMultipart <EOL> from email . mime . text import MIMEText <EOL> from email . header import Header <EOL> from email . utils import formatdate , formataddr , make_msgid , parseaddr <EOL> from contextlib import contextmanager <EOL> from flask import current_app <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> <EOL> PY34 = PY3 and sys . version_info [ <NUM_LIT:1> ] >= <NUM_LIT:4> <EOL> if PY3 : <EOL> string_types = str , <EOL> text_type = str <EOL> from email import policy <EOL> message_policy = policy . SMTP <EOL> else : <EOL> string_types = basestring , <EOL> text_type = unicode <EOL> message_policy = None <EOL> charset . add_charset ( '<STR_LIT:utf-8>' , charset . SHORTEST , None , '<STR_LIT:utf-8>' ) <EOL> class FlaskMailUnicodeDecodeError ( UnicodeDecodeError ) : <EOL> def __init__ ( self , obj , * args ) : <EOL> self . obj = obj <EOL> UnicodeDecodeError . __init__ ( self , * args ) <EOL> def __str__ ( self ) : <EOL> original = UnicodeDecodeError . __str__ ( self ) <EOL> return '<STR_LIT>' % ( original , self . obj , type ( self . obj ) ) <EOL> def force_text ( s , encoding = '<STR_LIT:utf-8>' , errors = '<STR_LIT:strict>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( s , text_type ) : <EOL> return s <EOL> try : <EOL> if not isinstance ( s , string_types ) : <EOL> if PY3 : <EOL> if isinstance ( s , bytes ) : <EOL> s = text_type ( s , encoding , errors ) <EOL> else : <EOL> s = text_type ( s ) <EOL> elif hasattr ( s , '<STR_LIT>' ) : <EOL> s = s . __unicode__ ( ) <EOL> else : <EOL> s = text_type ( bytes ( s ) , encoding , errors ) <EOL> else : <EOL> s = s . decode ( encoding , errors ) <EOL> except UnicodeDecodeError as e : <EOL> if not isinstance ( s , Exception ) : <EOL> raise FlaskMailUnicodeDecodeError ( s , * e . args ) <EOL> else : <EOL> s = '<STR_LIT:U+0020>' . join ( [ force_text ( arg , encoding , strings_only , <EOL> errors ) for arg in s ] ) <EOL> return s <EOL> def sanitize_subject ( subject , encoding = '<STR_LIT:utf-8>' ) : <EOL> try : <EOL> subject . encode ( '<STR_LIT:ascii>' ) <EOL> except UnicodeEncodeError : <EOL> try : <EOL> subject = Header ( subject , encoding ) . encode ( ) <EOL> except UnicodeEncodeError : <EOL> subject = Header ( subject , '<STR_LIT:utf-8>' ) . encode ( ) <EOL> return subject <EOL> def sanitize_address ( addr , encoding = '<STR_LIT:utf-8>' ) : <EOL> if isinstance ( addr , string_types ) : <EOL> addr = parseaddr ( force_text ( addr ) ) <EOL> nm , addr = addr <EOL> try : <EOL> nm = Header ( nm , encoding ) . encode ( ) <EOL> except UnicodeEncodeError : <EOL> nm = Header ( nm , '<STR_LIT:utf-8>' ) . encode ( ) <EOL> try : <EOL> addr . encode ( '<STR_LIT:ascii>' ) <EOL> except UnicodeEncodeError : <EOL> if '<STR_LIT:@>' in addr : <EOL> localpart , domain = addr . split ( '<STR_LIT:@>' , <NUM_LIT:1> ) <EOL> localpart = str ( Header ( localpart , encoding ) ) <EOL> domain = domain . encode ( '<STR_LIT>' ) . decode ( '<STR_LIT:ascii>' ) <EOL> addr = '<STR_LIT:@>' . join ( [ localpart , domain ] ) <EOL> else : <EOL> addr = Header ( addr , encoding ) . encode ( ) <EOL> return formataddr ( ( nm , addr ) ) <EOL> def sanitize_addresses ( addresses , encoding = '<STR_LIT:utf-8>' ) : <EOL> return map ( lambda e : sanitize_address ( e , encoding ) , addresses ) <EOL> def _has_newline ( line ) : <EOL> """<STR_LIT>""" <EOL> if line and ( '<STR_LIT:\r>' in line or '<STR_LIT:\n>' in line ) : <EOL> return True <EOL> return False <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mail ) : <EOL> self . mail = mail <EOL> def __enter__ ( self ) : <EOL> if self . mail . suppress : <EOL> self . host = None <EOL> else : <EOL> self . host = self . configure_host ( ) <EOL> self . num_emails = <NUM_LIT:0> <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_value , tb ) : <EOL> if self . host : <EOL> self . host . quit ( ) <EOL> def configure_host ( self ) : <EOL> if self . mail . use_ssl : <EOL> host = smtplib . SMTP_SSL ( self . mail . server , self . mail . port ) <EOL> else : <EOL> host = smtplib . SMTP ( self . mail . server , self . mail . port ) <EOL> host . set_debuglevel ( int ( self . mail . debug ) ) <EOL> if self . mail . use_tls : <EOL> host . starttls ( ) <EOL> if self . mail . username and self . mail . password : <EOL> host . login ( self . mail . username , self . mail . password ) <EOL> return host <EOL> def send ( self , message , envelope_from = None ) : <EOL> """<STR_LIT>""" <EOL> assert message . send_to , "<STR_LIT>" <EOL> assert message . sender , ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if message . has_bad_headers ( ) : <EOL> raise BadHeaderError <EOL> if message . date is None : <EOL> message . date = time . time ( ) <EOL> if self . host : <EOL> self . host . sendmail ( sanitize_address ( envelope_from or message . sender ) , <EOL> list ( sanitize_addresses ( message . send_to ) ) , <EOL> message . as_bytes ( ) if PY3 else message . as_string ( ) , <EOL> message . mail_options , <EOL> message . rcpt_options ) <EOL> email_dispatched . send ( message , app = current_app . _get_current_object ( ) ) <EOL> self . num_emails += <NUM_LIT:1> <EOL> if self . num_emails == self . mail . max_emails : <EOL> self . num_emails = <NUM_LIT:0> <EOL> if self . host : <EOL> self . host . quit ( ) <EOL> self . host = self . configure_host ( ) <EOL> def send_message ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . send ( Message ( * args , ** kwargs ) ) <EOL> class BadHeaderError ( Exception ) : <EOL> pass <EOL> class Attachment ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename = None , content_type = None , data = None , <EOL> disposition = None , headers = None ) : <EOL> self . filename = filename <EOL> self . content_type = content_type <EOL> self . data = data <EOL> self . disposition = disposition or '<STR_LIT>' <EOL> self . headers = headers or { } <EOL> class Message ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , subject = '<STR_LIT>' , <EOL> recipients = None , <EOL> body = None , <EOL> html = None , <EOL> alts = None , <EOL> sender = None , <EOL> cc = None , <EOL> bcc = None , <EOL> attachments = None , <EOL> reply_to = None , <EOL> date = None , <EOL> charset = None , <EOL> extra_headers = None , <EOL> mail_options = None , <EOL> rcpt_options = None ) : <EOL> sender = sender or current_app . extensions [ '<STR_LIT>' ] . default_sender <EOL> if isinstance ( sender , tuple ) : <EOL> sender = "<STR_LIT>" % sender <EOL> self . recipients = recipients or [ ] <EOL> self . subject = subject <EOL> self . sender = sender <EOL> self . reply_to = reply_to <EOL> self . cc = cc or [ ] <EOL> self . bcc = bcc or [ ] <EOL> self . body = body <EOL> self . alts = dict ( alts or { } ) <EOL> self . html = html <EOL> self . date = date <EOL> self . msgId = make_msgid ( ) <EOL> self . charset = charset <EOL> self . extra_headers = extra_headers <EOL> self . mail_options = mail_options or [ ] <EOL> self . rcpt_options = rcpt_options or [ ] <EOL> self . attachments = attachments or [ ] <EOL> @ property <EOL> def send_to ( self ) : <EOL> return set ( self . recipients ) | set ( self . bcc or ( ) ) | set ( self . cc or ( ) ) <EOL> @ property <EOL> def html ( self ) : <EOL> return self . alts . get ( '<STR_LIT:html>' ) <EOL> @ html . setter <EOL> def html ( self , value ) : <EOL> if value is None : <EOL> self . alts . pop ( '<STR_LIT:html>' , None ) <EOL> else : <EOL> self . alts [ '<STR_LIT:html>' ] = value <EOL> def _mimetext ( self , text , subtype = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> charset = self . charset or '<STR_LIT:utf-8>' <EOL> return MIMEText ( text , _subtype = subtype , _charset = charset ) <EOL> def _message ( self ) : <EOL> """<STR_LIT>""" <EOL> ascii_attachments = current_app . extensions [ '<STR_LIT>' ] . ascii_attachments <EOL> encoding = self . charset or '<STR_LIT:utf-8>' <EOL> attachments = self . attachments or [ ] <EOL> if len ( attachments ) == <NUM_LIT:0> and not self . alts : <EOL> msg = self . _mimetext ( self . body ) <EOL> elif len ( attachments ) > <NUM_LIT:0> and not self . alts : <EOL> msg = MIMEMultipart ( ) <EOL> msg . attach ( self . _mimetext ( self . body ) ) <EOL> else : <EOL> msg = MIMEMultipart ( ) <EOL> alternative = MIMEMultipart ( '<STR_LIT>' ) <EOL> alternative . attach ( self . _mimetext ( self . body , '<STR_LIT>' ) ) <EOL> for mimetype , content in self . alts . items ( ) : <EOL> alternative . attach ( self . _mimetext ( content , mimetype ) ) <EOL> msg . attach ( alternative ) <EOL> if self . subject : <EOL> msg [ '<STR_LIT>' ] = sanitize_subject ( force_text ( self . subject ) , encoding ) <EOL> msg [ '<STR_LIT>' ] = sanitize_address ( self . sender , encoding ) <EOL> msg [ '<STR_LIT>' ] = '<STR_LIT:U+002CU+0020>' . join ( list ( set ( sanitize_addresses ( self . recipients , encoding ) ) ) ) <EOL> msg [ '<STR_LIT>' ] = formatdate ( self . date , localtime = True ) <EOL> msg [ '<STR_LIT>' ] = self . msgId <EOL> if self . cc : <EOL> msg [ '<STR_LIT>' ] = '<STR_LIT:U+002CU+0020>' . join ( list ( set ( sanitize_addresses ( self . cc , encoding ) ) ) ) <EOL> if self . reply_to : <EOL> msg [ '<STR_LIT>' ] = sanitize_address ( self . reply_to , encoding ) <EOL> if self . extra_headers : <EOL> for k , v in self . extra_headers . items ( ) : <EOL> msg [ k ] = v <EOL> SPACES = re . compile ( r'<STR_LIT>' , re . UNICODE ) <EOL> for attachment in attachments : <EOL> f = MIMEBase ( * attachment . content_type . split ( '<STR_LIT:/>' ) ) <EOL> f . set_payload ( attachment . data ) <EOL> encode_base64 ( f ) <EOL> filename = attachment . filename <EOL> if filename and ascii_attachments : <EOL> filename = unicodedata . normalize ( '<STR_LIT>' , filename ) <EOL> filename = filename . encode ( '<STR_LIT:ascii>' , '<STR_LIT:ignore>' ) . decode ( '<STR_LIT:ascii>' ) <EOL> filename = SPACES . sub ( u'<STR_LIT:U+0020>' , filename ) . strip ( ) <EOL> try : <EOL> filename and filename . encode ( '<STR_LIT:ascii>' ) <EOL> except UnicodeEncodeError : <EOL> if not PY3 : <EOL> filename = filename . encode ( '<STR_LIT:utf8>' ) <EOL> filename = ( '<STR_LIT>' , '<STR_LIT>' , filename ) <EOL> f . add_header ( '<STR_LIT>' , <EOL> attachment . disposition , <EOL> filename = filename ) <EOL> for key , value in attachment . headers . items ( ) : <EOL> f . add_header ( key , value ) <EOL> msg . attach ( f ) <EOL> if message_policy : <EOL> msg . policy = message_policy <EOL> return msg <EOL> def as_string ( self ) : <EOL> return self . _message ( ) . as_string ( ) <EOL> def as_bytes ( self ) : <EOL> if PY34 : <EOL> return self . _message ( ) . as_bytes ( ) <EOL> else : <EOL> return self . _message ( ) . as_string ( ) . encode ( self . charset or '<STR_LIT:utf-8>' ) <EOL> def __str__ ( self ) : <EOL> return self . as_string ( ) <EOL> def __bytes__ ( self ) : <EOL> return self . as_bytes ( ) <EOL> def has_bad_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = [ self . sender , self . reply_to ] + self . recipients <EOL> for header in headers : <EOL> if _has_newline ( header ) : <EOL> return True <EOL> if self . subject : <EOL> if _has_newline ( self . subject ) : <EOL> for linenum , line in enumerate ( self . subject . split ( '<STR_LIT:\r\n>' ) ) : <EOL> if not line : <EOL> return True <EOL> if linenum > <NUM_LIT:0> and line [ <NUM_LIT:0> ] not in '<STR_LIT>' : <EOL> return True <EOL> if _has_newline ( line ) : <EOL> return True <EOL> if len ( line . strip ( ) ) == <NUM_LIT:0> : <EOL> return True <EOL> return False <EOL> def is_bad_headers ( self ) : <EOL> from warnings import warn <EOL> msg = '<STR_LIT>' <EOL> warn ( DeprecationWarning ( msg ) , stacklevel = <NUM_LIT:1> ) <EOL> return self . has_bad_headers ( ) <EOL> def send ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> connection . send ( self ) <EOL> def add_recipient ( self , recipient ) : <EOL> """<STR_LIT>""" <EOL> self . recipients . append ( recipient ) <EOL> def attach ( self , <EOL> filename = None , <EOL> content_type = None , <EOL> data = None , <EOL> disposition = None , <EOL> headers = None ) : <EOL> """<STR_LIT>""" <EOL> self . attachments . append ( <EOL> Attachment ( filename , content_type , data , disposition , headers ) ) <EOL> class _MailMixin ( object ) : <EOL> @ contextmanager <EOL> def record_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> if not email_dispatched : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> outbox = [ ] <EOL> def _record ( message , app ) : <EOL> outbox . append ( message ) <EOL> email_dispatched . connect ( _record ) <EOL> try : <EOL> yield outbox <EOL> finally : <EOL> email_dispatched . disconnect ( _record ) <EOL> def send ( self , message ) : <EOL> """<STR_LIT>""" <EOL> with self . connect ( ) as connection : <EOL> message . send ( connection ) <EOL> def send_message ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . send ( Message ( * args , ** kwargs ) ) <EOL> def connect ( self ) : <EOL> """<STR_LIT>""" <EOL> app = getattr ( self , "<STR_LIT>" , None ) or current_app <EOL> try : <EOL> return Connection ( app . extensions [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> class _Mail ( _MailMixin ) : <EOL> def __init__ ( self , server , username , password , port , use_tls , use_ssl , <EOL> default_sender , debug , max_emails , suppress , <EOL> ascii_attachments = False ) : <EOL> self . server = server <EOL> self . username = username <EOL> self . password = password <EOL> self . port = port <EOL> self . use_tls = use_tls <EOL> self . use_ssl = use_ssl <EOL> self . default_sender = default_sender <EOL> self . debug = debug <EOL> self . max_emails = max_emails <EOL> self . suppress = suppress <EOL> self . ascii_attachments = ascii_attachments <EOL> class Mail ( _MailMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app = None ) : <EOL> self . app = app <EOL> if app is not None : <EOL> self . state = self . init_app ( app ) <EOL> else : <EOL> self . state = None <EOL> def init_mail ( self , config , debug = False , testing = False ) : <EOL> return _Mail ( <EOL> config . get ( '<STR_LIT>' , '<STR_LIT:127.0.0.1>' ) , <EOL> config . get ( '<STR_LIT>' ) , <EOL> config . get ( '<STR_LIT>' ) , <EOL> config . get ( '<STR_LIT>' , <NUM_LIT> ) , <EOL> config . get ( '<STR_LIT>' , False ) , <EOL> config . get ( '<STR_LIT>' , False ) , <EOL> config . get ( '<STR_LIT>' ) , <EOL> int ( config . get ( '<STR_LIT>' , debug ) ) , <EOL> config . get ( '<STR_LIT>' ) , <EOL> config . get ( '<STR_LIT>' , testing ) , <EOL> config . get ( '<STR_LIT>' , False ) <EOL> ) <EOL> def init_app ( self , app ) : <EOL> """<STR_LIT>""" <EOL> state = self . init_mail ( app . config , app . debug , app . testing ) <EOL> app . extensions = getattr ( app , '<STR_LIT>' , { } ) <EOL> app . extensions [ '<STR_LIT>' ] = state <EOL> return state <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . state , name , None ) <EOL> signals = blinker . Namespace ( ) <EOL> email_dispatched = signals . signal ( "<STR_LIT>" , doc = """<STR_LIT>""" ) </s>
