mirror of
https://github.com/kovidgoyal/calibre.git
synced 2025-07-07 10:14:46 -04:00
py3: port smtplib
This commit is contained in:
parent
79c21b14bc
commit
fd307c5eee
@ -12,6 +12,7 @@ This module implements a simple commandline SMTP client that supports:
|
|||||||
|
|
||||||
import sys, traceback, os, socket, encodings.idna as idna
|
import sys, traceback, os, socket, encodings.idna as idna
|
||||||
from calibre import isbytestring, force_unicode
|
from calibre import isbytestring, force_unicode
|
||||||
|
from calibre.constants import ispy3
|
||||||
|
|
||||||
|
|
||||||
def safe_localhost():
|
def safe_localhost():
|
||||||
@ -24,7 +25,7 @@ def safe_localhost():
|
|||||||
# https://bugs.launchpad.net/bugs/1256549
|
# https://bugs.launchpad.net/bugs/1256549
|
||||||
try:
|
try:
|
||||||
local_hostname = idna.ToASCII(force_unicode(fqdn))
|
local_hostname = idna.ToASCII(force_unicode(fqdn))
|
||||||
except:
|
except Exception:
|
||||||
local_hostname = 'localhost.localdomain'
|
local_hostname = 'localhost.localdomain'
|
||||||
else:
|
else:
|
||||||
# We can't find an fqdn hostname, so use a domain literal
|
# We can't find an fqdn hostname, so use a domain literal
|
||||||
@ -104,7 +105,7 @@ def get_mx(host, verbose=0):
|
|||||||
|
|
||||||
def sendmail_direct(from_, to, msg, timeout, localhost, verbose,
|
def sendmail_direct(from_, to, msg, timeout, localhost, verbose,
|
||||||
debug_output=None):
|
debug_output=None):
|
||||||
import calibre.utils.smtplib as smtplib
|
import polyglot.smtplib as smtplib
|
||||||
hosts = get_mx(to.split('@')[-1].strip(), verbose)
|
hosts = get_mx(to.split('@')[-1].strip(), verbose)
|
||||||
timeout=None # Non blocking sockets sometimes don't work
|
timeout=None # Non blocking sockets sometimes don't work
|
||||||
kwargs = dict(timeout=timeout, local_hostname=localhost or safe_localhost())
|
kwargs = dict(timeout=timeout, local_hostname=localhost or safe_localhost())
|
||||||
@ -133,7 +134,7 @@ def sendmail(msg, from_, to, localhost=None, verbose=0, timeout=None,
|
|||||||
if relay is None:
|
if relay is None:
|
||||||
for x in to:
|
for x in to:
|
||||||
return sendmail_direct(from_, x, msg, timeout, localhost, verbose)
|
return sendmail_direct(from_, x, msg, timeout, localhost, verbose)
|
||||||
import calibre.utils.smtplib as smtplib
|
import polyglot.smtplib as smtplib
|
||||||
cls = smtplib.SMTP_SSL if encryption == 'SSL' else smtplib.SMTP
|
cls = smtplib.SMTP_SSL if encryption == 'SSL' else smtplib.SMTP
|
||||||
timeout = None # Non-blocking sockets sometimes don't work
|
timeout = None # Non-blocking sockets sometimes don't work
|
||||||
port = int(port)
|
port = int(port)
|
||||||
@ -153,7 +154,7 @@ def sendmail(msg, from_, to, localhost=None, verbose=0, timeout=None,
|
|||||||
s.starttls(context=context)
|
s.starttls(context=context)
|
||||||
s.ehlo()
|
s.ehlo()
|
||||||
if username is not None and password is not None:
|
if username is not None and password is not None:
|
||||||
if encryption == 'SSL':
|
if encryption == 'SSL' and not ispy3:
|
||||||
s.sock = s.file.sslobj
|
s.sock = s.file.sslobj
|
||||||
s.login(username, password)
|
s.login(username, password)
|
||||||
ret = None
|
ret = None
|
||||||
|
@ -48,7 +48,7 @@ import re
|
|||||||
import email.utils
|
import email.utils
|
||||||
import base64
|
import base64
|
||||||
import hmac
|
import hmac
|
||||||
from email.base64mime import encode as encode_base64
|
from email.base64mime import body_encode as encode_base64
|
||||||
from sys import stderr
|
from sys import stderr
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
|
||||||
@ -282,7 +282,7 @@ class SMTP:
|
|||||||
# if that can't be calculated, that we should use a domain literal
|
# if that can't be calculated, that we should use a domain literal
|
||||||
# instead (essentially an encoded IP address like [A.B.C.D]).
|
# instead (essentially an encoded IP address like [A.B.C.D]).
|
||||||
fqdn = socket.getfqdn()
|
fqdn = socket.getfqdn()
|
||||||
if '.' in fqdn and fqdn != '.': # Changed by Kovid
|
if '.' in fqdn:
|
||||||
self.local_hostname = fqdn
|
self.local_hostname = fqdn
|
||||||
else:
|
else:
|
||||||
# We can't find an fqdn hostname, so use a domain literal
|
# We can't find an fqdn hostname, so use a domain literal
|
||||||
@ -345,11 +345,6 @@ class SMTP:
|
|||||||
"""Send `str' to the server."""
|
"""Send `str' to the server."""
|
||||||
if self.debuglevel > 0:
|
if self.debuglevel > 0:
|
||||||
raw = repr(str)
|
raw = repr(str)
|
||||||
if self.debuglevel < 2:
|
|
||||||
if len(raw) > 100:
|
|
||||||
raw = raw[:100] + '...'
|
|
||||||
if 'AUTH' in raw:
|
|
||||||
raw = 'AUTH <censored>'
|
|
||||||
self.debug('send:', raw)
|
self.debug('send:', raw)
|
||||||
if hasattr(self, 'sock') and self.sock:
|
if hasattr(self, 'sock') and self.sock:
|
||||||
try:
|
try:
|
||||||
|
73
src/polyglot/smtplib.py
Normal file
73
src/polyglot/smtplib.py
Normal file
@ -0,0 +1,73 @@
|
|||||||
|
#!/usr/bin/env python2
|
||||||
|
# vim:fileencoding=utf-8
|
||||||
|
# License: GPL v3 Copyright: 2019, Kovid Goyal <kovid at kovidgoyal.net>
|
||||||
|
|
||||||
|
from __future__ import absolute_import, division, print_function, unicode_literals
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from functools import partial
|
||||||
|
|
||||||
|
from polyglot.builtins import is_py3
|
||||||
|
|
||||||
|
|
||||||
|
class FilteredLog(object):
|
||||||
|
|
||||||
|
' Hide AUTH credentials from the log '
|
||||||
|
|
||||||
|
def __init__(self, debug_to=None):
|
||||||
|
self.debug_to = debug_to or partial(print, file=sys.stderr)
|
||||||
|
|
||||||
|
def __call__(self, *a):
|
||||||
|
if a and len(a) == 2 and a[0] == 'send:':
|
||||||
|
a = list(a)
|
||||||
|
raw = a[1]
|
||||||
|
if len(raw) > 100:
|
||||||
|
raw = raw[:100] + (b'...' if isinstance(raw, bytes) else '...')
|
||||||
|
q = b'AUTH ' if isinstance(raw, bytes) else 'AUTH '
|
||||||
|
if q in raw:
|
||||||
|
raw = 'AUTH <censored>'
|
||||||
|
a[1] = raw
|
||||||
|
self.debug_to(*a)
|
||||||
|
|
||||||
|
|
||||||
|
if is_py3:
|
||||||
|
import smtplib
|
||||||
|
|
||||||
|
class SMTP(smtplib.SMTP):
|
||||||
|
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
self.debug_to = FilteredLog(kw.pop('debug_to', None))
|
||||||
|
super().__init__(*a, **kw)
|
||||||
|
|
||||||
|
def _print_debug(self, *a):
|
||||||
|
if self.debug_to is not None:
|
||||||
|
self.debug_to(*a)
|
||||||
|
else:
|
||||||
|
super()._print_debug(*a)
|
||||||
|
|
||||||
|
class SMTP_SSL(smtplib.SMTP_SSL):
|
||||||
|
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
self.debug_to = FilteredLog(kw.pop('debug_to', None))
|
||||||
|
super().__init__(*a, **kw)
|
||||||
|
|
||||||
|
def _print_debug(self, *a):
|
||||||
|
if self.debug_to is not None:
|
||||||
|
self.debug_to(*a)
|
||||||
|
else:
|
||||||
|
super()._print_debug(*a)
|
||||||
|
|
||||||
|
else:
|
||||||
|
import calibre.utils.smtplib as smtplib
|
||||||
|
|
||||||
|
class SMTP(smtplib.SMTP):
|
||||||
|
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
kw['debug_to'] = FilteredLog(kw.get('debug_to'))
|
||||||
|
smtplib.SMTP.__init__(self, *a, **kw)
|
||||||
|
|
||||||
|
class SMTP_SSL(smtplib.SMTP_SSL):
|
||||||
|
|
||||||
|
def __init__(self, *a, **kw):
|
||||||
|
kw['debug_to'] = FilteredLog(kw.get('debug_to'))
|
||||||
|
smtplib.SMTP_SSL.__init__(self, *a, **kw)
|
Loading…
x
Reference in New Issue
Block a user