362 lines
11 KiB
Python
362 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
# vi:si:et:sw=4:sts=4:ts=4
|
|
|
|
|
|
import os
|
|
import sys
|
|
from PIL import Image
|
|
from io import StringIO, BytesIO
|
|
import re
|
|
import stdnum.isbn
|
|
import socket
|
|
import io
|
|
import gzip
|
|
import time
|
|
import hashlib
|
|
from datetime import datetime
|
|
import subprocess
|
|
import base64
|
|
|
|
import ox
|
|
import ed25519
|
|
from OpenSSL.crypto import (
|
|
load_privatekey, load_certificate,
|
|
dump_privatekey, dump_certificate,
|
|
FILETYPE_ASN1, FILETYPE_PEM, PKey, TYPE_RSA,
|
|
X509, X509Extension
|
|
)
|
|
from Crypto.PublicKey import RSA
|
|
from Crypto.Util.asn1 import DerSequence
|
|
|
|
|
|
from meta.utils import normalize_isbn, find_isbns
|
|
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENCODING='base64'
|
|
|
|
def cleanup_id(key, value):
|
|
if key == 'isbn':
|
|
value = normalize_isbn(value)
|
|
if key in ('lccn', 'olid', 'oclc'):
|
|
value = ''.join([v for v in value if v!='-'])
|
|
return value
|
|
|
|
def valid_olid(id):
|
|
return id.startswith('OL') and id.endswith('M')
|
|
|
|
def get_positions(ids, pos):
|
|
'''
|
|
>>> get_positions([1,2,3,4], [2,4])
|
|
{2: 1, 4: 3}
|
|
'''
|
|
positions = {}
|
|
for i in pos:
|
|
try:
|
|
positions[i] = ids.index(i)
|
|
except:
|
|
pass
|
|
return positions
|
|
|
|
def get_by_key(objects, key, value):
|
|
obj = [o for o in objects if o.get(key) == value]
|
|
return obj and obj[0] or None
|
|
|
|
def get_by_id(objects, id):
|
|
return get_by_key(objects, 'id', id)
|
|
|
|
def resize_image(data, width=None, size=None):
|
|
if isinstance(data, bytes):
|
|
data = BytesIO(data)
|
|
else:
|
|
data = StringIO(data)
|
|
source = Image.open(data)
|
|
if source.mode not in ('1', 'CMYK', 'L', 'RGB', 'RGBA', 'RGBX', 'YCbCr'):
|
|
source = source.convert('RGB')
|
|
source_width = source.size[0]
|
|
source_height = source.size[1]
|
|
if size:
|
|
if source_width > source_height:
|
|
width = size
|
|
height = int(width / (float(source_width) / source_height))
|
|
height = height - height % 2
|
|
else:
|
|
height = size
|
|
width = int(height * (float(source_width) / source_height))
|
|
width = width - width % 2
|
|
|
|
else:
|
|
height = int(width / (float(source_width) / source_height))
|
|
height = height - height % 2
|
|
|
|
width = max(width, 1)
|
|
height = max(height, 1)
|
|
|
|
if width < source_width:
|
|
resize_method = Image.ANTIALIAS
|
|
else:
|
|
resize_method = Image.BICUBIC
|
|
output = source.resize((width, height), resize_method)
|
|
o = BytesIO()
|
|
output.save(o, format='jpeg')
|
|
data = o.getvalue()
|
|
o.close()
|
|
return data
|
|
|
|
def sort_title(title):
|
|
|
|
title = title.replace('Æ', 'Ae')
|
|
if isinstance(title, str):
|
|
title = str(title)
|
|
title = ox.sort_string(title)
|
|
|
|
#title
|
|
title = re.sub('[\'!¿¡,\.;\-"\:\*\[\]]', '', title)
|
|
return title.strip()
|
|
|
|
def get_position_by_id(list, key):
|
|
for i in range(0, len(list)):
|
|
if list[i]['id'] == key:
|
|
return i
|
|
return -1
|
|
|
|
def valid(key, value, sig):
|
|
'''
|
|
validate that value was signed by key
|
|
'''
|
|
if isinstance(sig, str):
|
|
sig = sig.encode()
|
|
if isinstance(value, str):
|
|
value = value.encode()
|
|
if isinstance(key, str):
|
|
key = key.encode()
|
|
vk = ed25519.VerifyingKey(key, encoding=ENCODING)
|
|
try:
|
|
vk.verify(sig, value, encoding=ENCODING)
|
|
#except ed25519.BadSignatureError:
|
|
except:
|
|
return False
|
|
return True
|
|
|
|
def get_user_id(private_key, cert_path):
|
|
if os.path.exists(private_key):
|
|
with open(private_key) as fd:
|
|
key = load_privatekey(FILETYPE_PEM, fd.read())
|
|
if key.bits() != 1024:
|
|
os.unlink(private_key)
|
|
else:
|
|
user_id = get_service_id(private_key)
|
|
if not os.path.exists(private_key):
|
|
if os.path.exists(cert_path):
|
|
os.unlink(cert_path)
|
|
folder = os.path.dirname(private_key)
|
|
if not os.path.exists(folder):
|
|
os.makedirs(folder)
|
|
os.chmod(folder, 0o700)
|
|
key = PKey()
|
|
key.generate_key(TYPE_RSA, 1024)
|
|
with open(private_key, 'wb') as fd:
|
|
os.chmod(private_key, 0o600)
|
|
fd.write(dump_privatekey(FILETYPE_PEM, key))
|
|
os.chmod(private_key, 0o400)
|
|
user_id = get_service_id(private_key)
|
|
if not os.path.exists(cert_path):
|
|
ca = X509()
|
|
ca.set_version(2)
|
|
ca.set_serial_number(1)
|
|
ca.get_subject().CN = user_id
|
|
ca.gmtime_adj_notBefore(0)
|
|
ca.gmtime_adj_notAfter(24 * 60 * 60)
|
|
ca.set_issuer(ca.get_subject())
|
|
ca.set_pubkey(key)
|
|
ca.add_extensions([
|
|
X509Extension(b"basicConstraints", True, b"CA:TRUE, pathlen:0"),
|
|
X509Extension(b"nsCertType", True, b"sslCA"),
|
|
X509Extension(b"extendedKeyUsage", True,
|
|
b"serverAuth,clientAuth,emailProtection,timeStamping,msCodeInd,msCodeCom,msCTLSign,msSGC,msEFS,nsSGC"),
|
|
X509Extension(b"keyUsage", False, b"keyCertSign, cRLSign"),
|
|
X509Extension(b"subjectKeyIdentifier", False, b"hash", subject=ca),
|
|
])
|
|
ca.sign(key, "sha256")
|
|
with open(cert_path, 'wb') as fd:
|
|
fd.write(dump_certificate(FILETYPE_PEM, ca))
|
|
return user_id
|
|
|
|
def get_service_id(private_key_file=None, cert=None):
|
|
'''
|
|
service_id is the first half of the sha1 of the rsa public key encoded in base32
|
|
'''
|
|
if private_key_file:
|
|
with open(private_key_file, 'rb') as fd:
|
|
private_key = fd.read()
|
|
public_key = RSA.importKey(private_key).publickey().exportKey('DER')[22:]
|
|
# compute sha1 of public key and encode first half in base32
|
|
service_id = base64.b32encode(hashlib.sha1(public_key).digest()[:10]).lower().decode()
|
|
'''
|
|
# compute public key from priate key and export in DER format
|
|
# ignoring the SPKI header(22 bytes)
|
|
key = load_privatekey(FILETYPE_PEM, private_key)
|
|
cert = X509()
|
|
cert.set_pubkey(key)
|
|
public_key = dump_privatekey(FILETYPE_ASN1, cert.get_pubkey())[22:]
|
|
# compute sha1 of public key and encode first half in base32
|
|
service_id = base64.b32encode(hashlib.sha1(public_key).digest()[:10]).lower().decode()
|
|
'''
|
|
elif cert:
|
|
# compute sha1 of public key and encode first half in base32
|
|
key = load_certificate(FILETYPE_ASN1, cert).get_pubkey()
|
|
pub_der = DerSequence()
|
|
pub_der.decode(dump_privatekey(FILETYPE_ASN1, key))
|
|
public_key = RSA.construct((pub_der._seq[1], pub_der._seq[2])).exportKey('DER')[22:]
|
|
service_id = base64.b32encode(hashlib.sha1(public_key).digest()[:10]).lower().decode()
|
|
return service_id
|
|
|
|
def get_public_ipv6():
|
|
try:
|
|
host = ('2a01:4f8:120:3201::3', 25519)
|
|
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
|
|
s.settimeout(1)
|
|
s.connect(host)
|
|
ip = s.getsockname()[0]
|
|
s.close()
|
|
except:
|
|
ip = None
|
|
return ip
|
|
|
|
def get_interface():
|
|
interface = ''
|
|
if sys.platform == 'darwin' or sys.platform.startswith('freebsd'):
|
|
#cmd = ['/usr/sbin/netstat', '-rn']
|
|
cmd = ['/sbin/route', '-n', 'get', 'default']
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
|
|
stdout, stderr = p.communicate()
|
|
stdout = stdout.decode('utf-8')
|
|
interface = [[p.strip() for p in s.split(':', 1)]
|
|
for s in stdout.strip().split('\n') if 'interface' in s]
|
|
if interface:
|
|
interface = '%%%s' % interface[0][1]
|
|
else:
|
|
interface = ''
|
|
return interface
|
|
|
|
def get_local_ipv4():
|
|
ip = None
|
|
if sys.platform == 'darwin' or sys.platform.startswith('freebsd'):
|
|
cmd = ['/sbin/route', '-n', 'get', 'default']
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
|
|
stdout, stderr = p.communicate()
|
|
stdout = stdout.decode('utf-8')
|
|
interface = [[p.strip() for p in s.split(':', 1)]
|
|
for s in stdout.strip().split('\n') if 'interface' in s]
|
|
if interface:
|
|
interface = interface[0][1]
|
|
cmd = ['ifconfig', interface]
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
|
|
stdout, stderr = p.communicate()
|
|
stdout = stdout.decode('utf-8')
|
|
ips = [l for l in stdout.split('\n') if 'inet ' in l]
|
|
if ips:
|
|
ip = ips[0].strip().split(' ')[1]
|
|
else:
|
|
cmd = ['ip', 'route', 'show']
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, close_fds=True)
|
|
stdout, stderr = p.communicate()
|
|
stdout = stdout.decode('utf-8')
|
|
local = [l for l in stdout.split('\n') if 'default' in l]
|
|
if local:
|
|
dev = local[0].split(' ')[4]
|
|
local_ip = [l for l in stdout.split('\n')
|
|
if dev in l and not 'default' in l and 'src' in l]
|
|
ip = [p for p in local_ip[0].split(' ')[1:] if '.' in p][0]
|
|
return ip
|
|
|
|
def update_dict(root, data):
|
|
for key in data:
|
|
keys = [part.replace('\0', '\\.') for part in key.replace('\\.', '\0').split('.')]
|
|
value = data[key]
|
|
p = root
|
|
while len(keys)>1:
|
|
key = keys.pop(0)
|
|
if isinstance(p, list):
|
|
p = p[get_position_by_id(p, key)]
|
|
else:
|
|
if key not in p:
|
|
p[key] = {}
|
|
p = p[key]
|
|
if value == None and keys[0] in p:
|
|
del p[keys[0]]
|
|
else:
|
|
p[keys[0]] = value
|
|
|
|
def remove_empty_folders(prefix):
|
|
empty = []
|
|
for root, folders, files in os.walk(prefix):
|
|
if not folders and not files:
|
|
empty.append(root)
|
|
for folder in empty:
|
|
remove_empty_tree(folder)
|
|
|
|
def remove_empty_tree(leaf):
|
|
while leaf:
|
|
if not os.path.exists(leaf):
|
|
leaf = os.path.dirname(leaf)
|
|
elif os.path.isdir(leaf) and not os.listdir(leaf):
|
|
logger.debug('rmdir %s', leaf)
|
|
os.rmdir(leaf)
|
|
else:
|
|
break
|
|
|
|
utc_0 = int(time.mktime(datetime(1970, 1, 1).timetuple()))
|
|
|
|
def datetime2ts(dt):
|
|
return int(time.mktime(dt.utctimetuple())) - utc_0
|
|
|
|
def ts2datetime(ts):
|
|
return datetime.utcfromtimestamp(float(ts))
|
|
|
|
def run(*cmd):
|
|
p = subprocess.Popen(cmd, close_fds=True)
|
|
p.wait()
|
|
return p.returncode
|
|
|
|
def get(*cmd):
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
|
|
stdout, error = p.communicate()
|
|
return stdout.decode()
|
|
|
|
def makefolder(path):
|
|
dirname = os.path.dirname(path)
|
|
if not os.path.exists(dirname):
|
|
os.makedirs(dirname)
|
|
|
|
|
|
def open_folder(folder=None, path=None):
|
|
cmd = []
|
|
if path and not folder:
|
|
folder = os.path.dirname(path)
|
|
if folder and not path:
|
|
path = folder
|
|
if sys.platform == 'darwin':
|
|
if folder and not path:
|
|
path = folder
|
|
cmd += ['open', '-R', path]
|
|
elif sys.platform.startswith('linux'):
|
|
cmd += ['xdg-open', folder]
|
|
else:
|
|
logger.debug('unsupported platform %s', sys.platform)
|
|
subprocess.Popen(cmd, close_fds=True)
|
|
|
|
def can_connect_dns(host="8.8.8.8", port=53):
|
|
"""
|
|
host: 8.8.8.8 (google-public-dns-a.google.com)
|
|
port: 53/tcp
|
|
"""
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(1)
|
|
s.connect((host, port))
|
|
return True
|
|
except:
|
|
pass
|
|
return False
|