--- /dev/null
+import re
+
+
+class Net:
+ def __init__(self, name):
+ self.name = name
+ self.id = None
+ self.subnet_name = None
+ self.subnet_id = None
+ self.subnet_creation_time = None
+ self.subnet_update_time = None
+ self.gateway_ip = None
+ self.segmentation_id = None # not set
+ self._cidr = None
+ self.start_end_dict = None
+ self._issued_ip_addresses = dict()
+
+ def get_short_id(self):
+ """
+ Returns a shortened UUID, with only the first 6 characters.
+
+ :return: First 6 characters of the UUID
+ :rtype: ``str``
+ """
+ return str(self.id)[:6]
+
+ def get_new_ip_address(self, port_name):
+ """
+ Calculates the next unused IP Address which belongs to the subnet.
+
+ :param port_name: Specifies the port.
+ :type port_name: ``str``
+ :return: Returns a unused IP Address or none if all are in use.
+ :rtype: ``str``
+ """
+ if self.start_end_dict is None:
+ return None
+
+ int_start_ip = Net.ip_2_int(self.start_end_dict['start']) + 2 # First address as network address not usable
+ # Second one is for gateways only
+ int_end_ip = Net.ip_2_int(self.start_end_dict['end']) - 1 # Last address for broadcasts
+ while int_start_ip in self._issued_ip_addresses and int_start_ip <= int_end_ip:
+ int_start_ip += 1
+
+ if int_start_ip > int_end_ip:
+ return None
+
+ self._issued_ip_addresses[int_start_ip] = port_name
+ return Net.int_2_ip(int_start_ip) + '/' + self._cidr.rsplit('/', 1)[1]
+
+ def assign_ip_address(self, cidr, port_name):
+ """
+ Assigns the IP address to the port if it is currently NOT used.
+
+ :param cidr: The cidr used by the port - e.g. 10.0.0.1/24
+ :type cidr: ``str``
+ :param port_name: The port name
+ :type port_name: ``str``
+ :return: * *False*: If the IP address is already issued or if it is not within this subnet mask.
+ * *True*: Else
+ """
+ int_ip = Net.cidr_2_int(cidr)
+ if int_ip in self._issued_ip_addresses:
+ return False
+
+ int_start_ip = Net.ip_2_int(self.start_end_dict['start']) + 1 # First address as network address not usable
+ int_end_ip = Net.ip_2_int(self.start_end_dict['end']) - 1 # Last address for broadcasts
+ if int_ip < int_start_ip or int_ip > int_end_ip:
+ return False
+
+ self._issued_ip_addresses[int_ip] = port_name
+ return True
+
+ def is_my_ip(self, cidr, port_name):
+ """
+ Checks if the IP is registered for this port name.
+
+ :param cidr: The cidr used by the port - e.g. 10.0.0.1/24
+ :type cidr: ``str``
+ :param port_name: The port name
+ :type port_name: ``str``
+ :return: Returns true if the IP address belongs to the port name. Else it returns false.
+ """
+ int_ip = Net.cidr_2_int(cidr)
+
+ if not int_ip in self._issued_ip_addresses:
+ return False
+
+ if self._issued_ip_addresses[int_ip] == port_name:
+ return True
+ return False
+
+ def withdraw_ip_address(self, ip_address):
+ """
+ Removes the IP address from the list of issued addresses, thus other ports can use it.
+
+ :param ip_address: The issued IP address.
+ :type ip_address: ``str``
+ """
+ if ip_address is None:
+ return
+
+ if "/" in ip_address:
+ address, suffix = ip_address.rsplit('/', 1)
+ else:
+ address = ip_address
+ int_ip_address = Net.ip_2_int(address)
+ if int_ip_address in self._issued_ip_addresses.keys():
+ del self._issued_ip_addresses[int_ip_address]
+
+ def reset_issued_ip_addresses(self):
+ """
+ Resets all issued IP addresses.
+ """
+ self._issued_ip_addresses = dict()
+
+ def update_port_name_for_ip_address(self, ip_address, port_name):
+ """
+ Updates the port name of the issued IP address.
+
+ :param ip_address: The already issued IP address.
+ :type ip_address: ``str``
+ :param port_name: The new port name
+ :type port_name: ``str``
+ """
+ address, suffix = ip_address.rsplit('/', 1)
+ int_ip_address = Net.ip_2_int(address)
+ self._issued_ip_addresses[int_ip_address] = port_name
+
+ def set_cidr(self, cidr):
+ """
+ Sets the CIDR for the subnet. It previously checks for the correct CIDR format.
+
+ :param cidr: The new CIDR for the subnet.
+ :type cidr: ``str``
+ :return: * *True*: When the new CIDR was set successfully.
+ * *False*: If the CIDR format was wrong.
+ :rtype: ``bool``
+ """
+ if cidr is None:
+ if self._cidr is not None:
+ import emuvim.api.heat.ip_handler as IP
+ IP.free_cidr(self._cidr, self.subnet_id)
+ self._cidr = None
+ self.reset_issued_ip_addresses()
+ self.start_end_dict = dict()
+ return True
+ if not Net.check_cidr_format(cidr):
+ return False
+
+ self.reset_issued_ip_addresses()
+ self.start_end_dict = Net.calculate_start_and_end_dict(cidr)
+ self._cidr = cidr
+ return True
+
+ def get_cidr(self):
+ """
+ Gets the CIDR.
+
+ :return: The CIDR
+ :rtype: ``str``
+ """
+ return self._cidr
+
+ def clear_cidr(self):
+ self._cidr = None
+ self.start_end_dict = dict()
+ self.reset_issued_ip_addresses()
+
+ def delete_subnet(self):
+ self.subnet_id = None
+ self.subnet_name = None
+ self.subnet_creation_time = None
+ self.subnet_update_time = None
+ self.set_cidr(None)
+
+ @staticmethod
+ def calculate_start_and_end_dict(cidr):
+ """
+ Calculates the start and end IP address for the subnet.
+
+ :param cidr: The CIDR for the subnet.
+ :type cidr: ``str``
+ :return: Dict with start and end ip address
+ :rtype: ``dict``
+ """
+ address, suffix = cidr.rsplit('/', 1)
+ int_suffix = int(suffix)
+ int_address = Net.ip_2_int(address)
+ address_space = 2 ** 32 - 1
+
+ for x in range(0, 31 - int_suffix):
+ address_space = ~(~address_space | (1 << x))
+
+ start = int_address & address_space
+ end = start + (2 ** (32 - int_suffix) - 1)
+
+ return {'start': Net.int_2_ip(start), 'end': Net.int_2_ip(end)}
+
+ @staticmethod
+ def cidr_2_int(cidr):
+ if cidr is None:
+ return None
+ ip = cidr.rsplit('/', 1)[0]
+ return Net.ip_2_int(ip)
+
+ @staticmethod
+ def ip_2_int(ip):
+ """
+ Converts a IP address to int.
+
+ :param ip: IP address
+ :type ip: ``str``
+ :return: IP address as int.
+ :rtype: ``int``
+ """
+ o = map(int, ip.split('.'))
+ res = (16777216 * o[0]) + (65536 * o[1]) + (256 * o[2]) + o[3]
+ return res
+
+ @staticmethod
+ def int_2_ip(int_ip):
+ """
+ Converts a int IP address to string.
+
+ :param int_ip: Int IP address.
+ :type int_ip: ``int``
+ :return: IP address
+ :rtype: ``str``
+ """
+ o1 = int(int_ip / 16777216) % 256
+ o2 = int(int_ip / 65536) % 256
+ o3 = int(int_ip / 256) % 256
+ o4 = int(int_ip) % 256
+ return '%(o1)s.%(o2)s.%(o3)s.%(o4)s' % locals()
+
+ @staticmethod
+ def check_cidr_format(cidr):
+ """
+ Checks the CIDR format. An valid example is: 192.168.0.0/29
+
+ :param cidr: CIDR to be checked.
+ :type cidr: ``str``
+ :return: * *True*: If the Format is correct.
+ * *False*: If it is not correct.
+ :rtype: ``bool``
+ """
+ r = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/\d{2}')
+ if r.match(cidr):
+ return True
+ return False
+
+ def create_network_dict(self):
+ """
+ Creates the network description dictionary.
+
+ :return: Network description.
+ :rtype: ``dict``
+ """
+ network_dict = dict()
+ network_dict["status"] = "ACTIVE" # TODO do we support inactive networks?
+ if self.subnet_id == None:
+ network_dict["subnets"] = []
+ else:
+ network_dict["subnets"] = [self.subnet_id]
+ network_dict["name"] = self.name
+ network_dict["admin_state_up"] = True # TODO is it always true?
+ network_dict["tenant_id"] = "abcdefghijklmnopqrstuvwxyz123456" # TODO what should go in here
+ network_dict["id"] = self.id
+ network_dict["shared"] = False # TODO is it always false?
+ return network_dict
+
+ def create_subnet_dict(self):
+ """
+ Creates the subnet description dictionary.
+
+ :return: Subnet description.
+ :rtype: ``dict``
+ """
+ subnet_dict = dict()
+ subnet_dict["name"] = self.subnet_name
+ subnet_dict["network_id"] = self.id
+ subnet_dict["tenant_id"] = "abcdefghijklmnopqrstuvwxyz123456" # TODO what should go in here?
+ subnet_dict["created_at"] = self.subnet_creation_time
+ subnet_dict["dns_nameservers"] = []
+ subnet_dict["allocation_pools"] = [self.start_end_dict]
+ subnet_dict["host_routers"] = []
+ subnet_dict["gateway_ip"] = self.gateway_ip
+ subnet_dict["ip_version"] = "4"
+ subnet_dict["cidr"] = self.get_cidr()
+ subnet_dict["updated_at"] = self.subnet_update_time
+ subnet_dict["id"] = self.subnet_id
+ subnet_dict["enable_dhcp"] = False # TODO do we support DHCP?
+ return subnet_dict
+
+ def __eq__(self, other):
+ if self.name == other.name and self.subnet_name == other.subnet_name and \
+ self.gateway_ip == other.gateway_ip and \
+ self.segmentation_id == other.segmentation_id and \
+ self._cidr == other._cidr and \
+ self.start_end_dict == other.start_end_dict:
+ return True
+ return False
+
+ def __hash__(self):
+ return hash((self.name,
+ self.subnet_name,
+ self.gateway_ip,
+ self.segmentation_id,
+ self._cidr,
+ self.start_end_dict))
projects / osm / vim-emu.git / blobdiff