Spotter-VM/basic/srv/vm/mgr/tools.py

# -*- coding: utf-8 -*-

import bcrypt
import dns.exception
import dns.resolver
import os
import requests
import shutil
import socket
import ssl
import subprocess

NULL_IP = '[100::1]'

def compile_url(domain, port, proto='https'):
    port = ':{}'.format(port) if (proto == 'https' and port != '443') or (proto == 'http' and port != '80') else ''
    host = '{}{}'.format(domain, port)
    return '{}://{}'.format(proto, host) if proto is not None else host

def get_local_ipv4():
    # Return first routable IPv4 address
    try:
        return subprocess.run(['/sbin/ip', 'route', 'get', '1'], check=True, stdout=subprocess.PIPE).stdout.decode().split()[-1]
    except:
        return None

def get_local_ipv6():
    # Return first routable IPv6 address
    try:
        return subprocess.run(['/sbin/ip', 'route', 'get', '2003::'], check=True, stdout=subprocess.PIPE).stdout.decode().split()[-3]
    except:
        return None

def get_external_ip(family):
    # Return external IP address of given family via 3rd party service
    allowed_gai_family = requests.packages.urllib3.util.connection.allowed_gai_family
    try:
        requests.packages.urllib3.util.connection.allowed_gai_family = lambda: family
        return requests.get('https://tools.dasm.cz/myip.php', timeout=5).text
    except:
        return None
    finally:
        requests.packages.urllib3.util.connection.allowed_gai_family = allowed_gai_family

def get_external_ipv4():
    # Return external IPv4 address
    return get_external_ip(socket.AF_INET)

def get_external_ipv6():
    # Return external IPv6 address
    return get_external_ip(socket.AF_INET6)

resolver = dns.resolver.Resolver()
resolver.timeout = 3
resolver.lifetime = 3
resolver.nameservers = ['8.8.8.8', '8.8.4.4', '2001:4860:4860::8888', '2001:4860:4860::8844']

def resolve_ip(domain, type):
    # Resolve domain name using Google Public DNS
    try:
        return resolver.query(domain, type)[0].address
    except dns.exception.Timeout:
        raise
    except:
        return None

def ping_url(url):
    try:
        return requests.post('https://tools.dasm.cz/vm-ping.php', data = {'url': url}, timeout=5).text == 'vm-pong'
    except requests.exceptions.Timeout:
        raise
    except:
        return False

def is_service_started(app):
    # Check OpenRC service status without calling any binary
    return os.path.exists(os.path.join('/run/openrc/started', app))

def is_service_autostarted(app):
    # Check OpenRC service enablement
    return os.path.exists(os.path.join('/etc/runlevels/default', app))

def start_service(service):
    subprocess.run(['/sbin/service', service, 'start'], check=True)

def stop_service(service):
    subprocess.run(['/sbin/service', service, 'stop'], check=True)

def restart_service(service):
    subprocess.run(['/sbin/service', service, 'restart'])

def reload_nginx():
    subprocess.run(['/sbin/service', 'nginx', 'reload'])

def restart_nginx():
    restart_service('nginx')

def get_cert_info(cert):
    data = ssl._ssl._test_decode_cert(cert)
    data['subject'] = dict(data['subject'][i][0] for i in range(len(data['subject'])))
    data['issuer'] = dict(data['issuer'][i][0] for i in range(len(data['issuer'])))
    return data

def adminpwd_hash(password):
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

def adminpwd_verify(password, hash):
    return bcrypt.checkpw(password.encode(), hash.encode())

def update_luks_password(oldpassword, newpassword):
    input = '{}\n{}'.format(oldpassword, newpassword).encode()
    subprocess.run(['cryptsetup', 'luksChangeKey', '/dev/sda2'], input=input, check=True)

def shutdown_vm():
    subprocess.run(['/sbin/poweroff'])

def reboot_vm():
    subprocess.run(['/sbin/reboot'])

def get_unused_ip():
    # This is a poor man's DHCP server which uses /etc/hosts as lease database
    # Leases the first unused IP from range 172.17.0.0/16
    leased = []
    with open('/etc/hosts', 'r') as fd:
        for line in fd.read().splitlines():
            if line.startswith('172.17'):
                ip = line.split()[0].split('.')
                leased.append(int(ip[2]) * 256 + int(ip[3]))
    for i in range(1, 65534):
        if i not in leased:
            break
    return '172.17.{}.{}'. format(i // 256, i % 256)

def update_hosts_lease(ip, app):
    hosts = []
    with open('/etc/hosts', 'r') as fd:
        for line in fd:
            if not line.strip().endswith(' {}'.format(app)):
                hosts.append(line)
    if ip:
        hosts.append('{} {}\n'.format(ip, app))
    with open('/etc/hosts', 'w') as fd:
        fd.writelines(hosts)

def get_container_ip(app):
    # Return an IP of a container. If the container doesn't exist, return address from IPv6 discard prefix instead
    with open('/etc/hosts', 'r') as fd:
        for line in fd:
            if line.strip().endswith(' {}'.format(app)):
                return line.split()[0]
    return NULL_IP

def set_container_ip(pid, ip):
    # Set IP in container based on PID given via lxc.hook.start-host hook
    cmd = 'ip addr add {}/16 broadcast 172.17.255.255 dev eth0 && ip route add default via 172.17.0.1'.format(ip)
    subprocess.run(['nsenter', '-a', '-t', pid, '--', '/bin/sh', '-c', cmd])

def remove_ephemeral_layer(app):
    layer = os.path.join('/var/lib/lxc', app, 'delta0')
    if os.path.exists(layer):
        for item in os.scandir(layer):
            shutil.rmtree(item.path) if item.is_dir() else os.unlink(item.path)