163 lines
4.2 KiB
Python
163 lines
4.2 KiB
Python
""" Various utils """
|
|
|
|
from . import settings
|
|
import uuid
|
|
|
|
import subprocess
|
|
import sys
|
|
import asyncio
|
|
import functools
|
|
|
|
|
|
class NumberedClass:
|
|
""" A class that counts its current instance number """
|
|
|
|
next_id = 1 # Count from 1: IP addresses and id 0 are not a good mix.
|
|
|
|
@classmethod
|
|
def get_id(cls):
|
|
out = cls.next_id
|
|
cls.next_id += 1
|
|
return out
|
|
|
|
def __init__(self):
|
|
self.id = self.get_id()
|
|
|
|
|
|
class LibvirtObject(NumberedClass):
|
|
""" A class that has the basic attributes of a libvirt objects:
|
|
* id
|
|
* uuid
|
|
and the applicable restrictions
|
|
"""
|
|
|
|
class TooMany(Exception):
|
|
def __init__(self, count):
|
|
self.count = count
|
|
|
|
def __str__(self):
|
|
return (
|
|
"Limit number reached. The current instance #{} does not fit in IPv4"
|
|
).format(self.count)
|
|
|
|
def __init__(self, conn):
|
|
super().__init__()
|
|
|
|
self.conn = conn
|
|
|
|
if self.id > 250:
|
|
raise self.TooMany(self.id)
|
|
|
|
self.uuid = uuid.uuid4()
|
|
|
|
|
|
class MACAddress:
|
|
""" A MAC address for a NIC or bridge """
|
|
|
|
def __init__(self, link_id, dev_id=None):
|
|
""" link_id: id of the current link
|
|
dev_id: id of the current NIC. If the device is a bridge, use None
|
|
"""
|
|
if dev_id is None:
|
|
dev_id = 0xFE
|
|
|
|
self.link_id = link_id
|
|
self.dev_id = dev_id
|
|
|
|
def __str__(self):
|
|
return "52:54:00:{nwid:02x}:{link_id:02x}:{dev_id:02x}".format(
|
|
nwid=settings.NETWORK_ID, link_id=self.link_id, dev_id=self.dev_id
|
|
)
|
|
|
|
def __repr__(self):
|
|
return str(self)
|
|
|
|
|
|
class Addrv4:
|
|
""" An address in IPv4 """
|
|
|
|
def __init__(self, link_id, dev_id=None, host_address=False):
|
|
""" link_id: id of the current link
|
|
dev_id: id of the current NIC. Use None to get the base address
|
|
host_address: if True and dev_id is None, returns the host machine address
|
|
on this network instead of the base address
|
|
"""
|
|
|
|
self.host_address = host_address
|
|
|
|
if dev_id is None:
|
|
if host_address:
|
|
dev_id = 0xFE
|
|
else:
|
|
dev_id = 0
|
|
|
|
self.link_id = link_id
|
|
self.dev_id = dev_id
|
|
self.netmask = "255.255.255.0"
|
|
self.prefix = 24
|
|
|
|
def __str__(self):
|
|
return "{base_range}.{link_id}.{dev_id}".format(
|
|
base_range=settings.IPV4_RANGE, link_id=self.link_id, dev_id=self.dev_id,
|
|
)
|
|
|
|
def __repr__(self):
|
|
return str(self)
|
|
|
|
|
|
class Addrv6:
|
|
""" An address in IPv6 """
|
|
|
|
def __init__(self, link_id, dev_id=None, host_address=False):
|
|
""" link_id: id of the current link
|
|
dev_id: id of the current NIC. Use None to get the base address
|
|
host_address: if True and dev_id is None, returns the host machine address
|
|
on this network instead of the base address
|
|
"""
|
|
self.host_address = host_address
|
|
|
|
if dev_id is None:
|
|
if host_address:
|
|
dev_id = 0xFE
|
|
else:
|
|
dev_id = 0
|
|
|
|
self.link_id = link_id
|
|
self.dev_id = dev_id
|
|
self.prefix = 64
|
|
|
|
def __str__(self):
|
|
return "{base_range}:{link_id:04x}::{dev_id:04x}".format(
|
|
base_range=settings.IPV6_RANGE, link_id=self.link_id, dev_id=self.dev_id,
|
|
)
|
|
|
|
|
|
def run_cmd_retry(command, *args, **kwargs):
|
|
rc = subprocess.run(command, *args, **kwargs)
|
|
while rc.returncode != 0:
|
|
print("Command failed. Try again:", file=sys.stderr)
|
|
rc = subprocess.run(command, *args, **kwargs)
|
|
return rc
|
|
|
|
|
|
def run_in_executor(f):
|
|
""" Decorator: transforms a blocking function into an awaitable one """
|
|
|
|
@functools.wraps(f)
|
|
def inner(*args, **kwargs):
|
|
loop = asyncio.get_running_loop()
|
|
return loop.run_in_executor(None, functools.partial(f, *args, **kwargs))
|
|
|
|
return inner
|
|
|
|
|
|
def ensure_sudo_rights():
|
|
""" Updates sudo credentials to ensure that the user is logged in afterwards,
|
|
without typing in a password """
|
|
run_cmd_retry(["sudo", "-v"])
|
|
|
|
|
|
def drop_sudo_rights():
|
|
""" Drop sudo credentials """
|
|
run_cmd_retry(["sudo", "-k"])
|