mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-17 05:41:25 +02:00
Merge pull request #1195 from modoboa/refactor/python3_fixes
Refactor/python3 fixes
This commit is contained in:
+22
-21
@@ -9,7 +9,8 @@ import string
|
||||
|
||||
from django.utils.encoding import force_bytes, force_text
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
from modoboa.parameters import tools as param_tools
|
||||
|
||||
@@ -29,27 +30,27 @@ def random_key(l=16):
|
||||
|
||||
|
||||
def encrypt(clear):
|
||||
key = force_bytes(
|
||||
param_tools.get_global_parameter("secret_key", app="core"))
|
||||
backend = default_backend()
|
||||
cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=backend)
|
||||
encryptor = cipher.encryptor()
|
||||
block_size = algorithms.AES.block_size
|
||||
if len(clear) % block_size:
|
||||
clear += " " * (block_size - len(clear) % block_size)
|
||||
ct = encryptor.update(force_bytes(clear)) + encryptor.finalize()
|
||||
return force_text(base64.b64encode(ct))
|
||||
|
||||
|
||||
def decrypt(ct):
|
||||
backend = default_backend()
|
||||
key = param_tools.get_global_parameter("secret_key", app="core")
|
||||
obj = AES.new(key, AES.MODE_ECB)
|
||||
if type(clear) is str:
|
||||
clear = clear.encode("utf-8")
|
||||
if len(clear) % AES.block_size:
|
||||
clear += (
|
||||
force_bytes(" ") * (AES.block_size - len(clear) % AES.block_size)
|
||||
)
|
||||
ciph = obj.encrypt(clear)
|
||||
ciph = base64.b64encode(ciph)
|
||||
return force_text(ciph)
|
||||
|
||||
|
||||
def decrypt(ciph):
|
||||
obj = AES.new(
|
||||
param_tools.get_global_parameter(
|
||||
"secret_key", app="core"), AES.MODE_ECB
|
||||
)
|
||||
ciph = base64.b64decode(ciph)
|
||||
clear = obj.decrypt(ciph)
|
||||
return clear.rstrip(' ')
|
||||
cipher = Cipher(
|
||||
algorithms.AES(force_bytes(key)), modes.ECB(), backend=backend)
|
||||
ct = base64.b64decode(force_bytes(ct))
|
||||
decryptor = cipher.decryptor()
|
||||
clear = decryptor.update(ct) + decryptor.finalize()
|
||||
return force_text(clear.rstrip(b" "))
|
||||
|
||||
|
||||
def get_password(request):
|
||||
|
||||
@@ -78,7 +78,7 @@ class Email(object):
|
||||
"""
|
||||
fname = msg.get_filename()
|
||||
if fname is not None:
|
||||
if type(fname) is six.text_type:
|
||||
if isinstance(fname, six.text_type):
|
||||
fname = fname.encode("utf-8")
|
||||
decoded = decode_header(fname)
|
||||
value = (
|
||||
@@ -218,7 +218,9 @@ class Email(object):
|
||||
body = lxml.html.tostring(html)
|
||||
else:
|
||||
body = lxml.html.tostring(body)
|
||||
body = re.sub("<(/?)body", lambda m: "<%sdiv" % m.group(1), body)
|
||||
body = re.sub(
|
||||
"<(/?)body", lambda m: "<%sdiv" % m.group(1),
|
||||
body.decode())
|
||||
return body
|
||||
|
||||
|
||||
@@ -284,7 +286,7 @@ def prepare_addresses(addresses, usage="header"):
|
||||
continue
|
||||
name, addr = parseaddr(address)
|
||||
if name and usage == "header":
|
||||
result.append("%s <%s>" % (Header(name, 'utf8').encode(), addr))
|
||||
result.append("%s <%s>" % (Header(name, 'utf8'), addr))
|
||||
else:
|
||||
result.append(addr)
|
||||
if usage == "header":
|
||||
|
||||
+26
-27
@@ -53,22 +53,21 @@ http://svn.plone.org/svn/collective/mxmImapClient/trunk/imapUTF7.py
|
||||
"""
|
||||
from __future__ import print_function, unicode_literals
|
||||
|
||||
from django.utils.encoding import force_text
|
||||
|
||||
import binascii
|
||||
import codecs
|
||||
|
||||
# encoding
|
||||
|
||||
PRINTABLE = set(range(0x20, 0x26)) | set(range(0x27, 0x7f))
|
||||
|
||||
def modified_base64(s):
|
||||
s = s.encode('utf-16be')
|
||||
return binascii.b2a_base64(s).rstrip('\n=').replace('/', ',')
|
||||
|
||||
def modified_utf7(s):
|
||||
s_utf7 = s.encode('utf-7')
|
||||
return s_utf7[1:-1].replace(b'/', b',')
|
||||
|
||||
|
||||
def doB64(_in, r):
|
||||
if _in:
|
||||
r.append('&%s-' % modified_base64(''.join(_in)))
|
||||
r.extend([b'&', modified_utf7(''.join(_in)), b'-'])
|
||||
del _in[:]
|
||||
|
||||
|
||||
@@ -76,44 +75,44 @@ def encoder(s):
|
||||
r = []
|
||||
_in = []
|
||||
for c in s:
|
||||
ordC = ord(c)
|
||||
if 0x20 <= ordC <= 0x25 or 0x27 <= ordC <= 0x7e:
|
||||
if ord(c) in PRINTABLE:
|
||||
doB64(_in, r)
|
||||
r.append(c)
|
||||
r.append(c.encode())
|
||||
elif c == '&':
|
||||
doB64(_in, r)
|
||||
r.append('&-')
|
||||
r.append(b'&-')
|
||||
else:
|
||||
_in.append(c)
|
||||
doB64(_in, r)
|
||||
return (str(''.join(r)), len(s))
|
||||
return (b''.join(r), len(s))
|
||||
|
||||
|
||||
# decoding
|
||||
|
||||
def modified_unbase64(s):
|
||||
b = binascii.a2b_base64(s.replace(',', '/') + '===')
|
||||
return force_text(b, encoding='utf-16be')
|
||||
|
||||
def modified_unutf7(s):
|
||||
s_utf7 = b'+' + s.replace(b',', b'/') + b'-'
|
||||
return s_utf7.decode('utf-7')
|
||||
|
||||
|
||||
def decoder(s):
|
||||
r = []
|
||||
decode = []
|
||||
decoded = bytearray()
|
||||
for c in s:
|
||||
if c == '&' and not decode:
|
||||
decode.append('&')
|
||||
elif c == '-' and decode:
|
||||
if len(decode) == 1:
|
||||
if c == ord('&') and not decoded:
|
||||
decoded.append(ord('&'))
|
||||
elif c == ord('-') and decoded:
|
||||
if len(decoded) == 1:
|
||||
r.append('&')
|
||||
else:
|
||||
r.append(modified_unbase64(''.join(decode[1:])))
|
||||
decode = []
|
||||
elif decode:
|
||||
decode.append(c)
|
||||
r.append(modified_unutf7(decoded[1:]))
|
||||
decoded = bytearray()
|
||||
elif decoded:
|
||||
decoded.append(c)
|
||||
else:
|
||||
r.append(c)
|
||||
if decode:
|
||||
r.append(modified_unbase64(''.join(decode[1:])))
|
||||
r.append(chr(c))
|
||||
if decoded:
|
||||
r.append(modified_unutf7(decoded[1:]))
|
||||
bin_str = ''.join(r)
|
||||
return (bin_str, len(s))
|
||||
|
||||
|
||||
+1
-1
@@ -22,7 +22,7 @@ jsonfield==1.0.3
|
||||
passlib==1.6.5
|
||||
progressbar33==2.4
|
||||
py-dateutil==2.2
|
||||
pycrypto
|
||||
cryptography
|
||||
pytz
|
||||
requests
|
||||
rfc6266
|
||||
|
||||
Reference in New Issue
Block a user