108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
# vi:si:et:sw=4:sts=4:ts=4
|
|
|
|
|
|
from functools import wraps
|
|
from threading import Thread
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
from urllib.parse import urlparse
|
|
|
|
import ed25519
|
|
|
|
from . import settings
|
|
|
|
import logging
|
|
logger = logging.getLogger('utils')
|
|
|
|
|
|
def get_interface():
|
|
interface = ''
|
|
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()
|
|
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()
|
|
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()
|
|
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()
|
|
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 get_public_ipv6():
|
|
n = urlparse(settings.server['directory_service'])
|
|
host = (n.hostname, n.port)
|
|
try:
|
|
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 run_async(func):
|
|
@wraps(func)
|
|
def async_func(*args, **kwargs):
|
|
func_hl = Thread(target = func, args = args, kwargs = kwargs)
|
|
func_hl.start()
|
|
return func_hl
|
|
|
|
return async_func
|
|
|
|
# ed25519 utils
|
|
def valid(key, value, sig):
|
|
'''
|
|
validate that value was signed by key
|
|
'''
|
|
if isinstance(sig, str):
|
|
sig = sig.encode('utf-8')
|
|
if isinstance(value, str):
|
|
value = value.encode('utf-8')
|
|
if isinstance(key, str):
|
|
key = key.encode('utf-8')
|
|
vk = ed25519.VerifyingKey(key, encoding=settings.ENCODING)
|
|
try:
|
|
vk.verify(sig, value, encoding=settings.ENCODING)
|
|
#except ed25519.BadSignatureError:
|
|
except:
|
|
return False
|
|
return True
|
|
|