|
| 1 | +import re |
| 2 | + |
| 3 | +class NetworkParserError(BaseException): |
| 4 | + pass |
| 5 | + |
| 6 | + |
| 7 | +class CloudscaleNetworkInterface(): |
| 8 | + def __init__(self): |
| 9 | + self._uuid = None |
| 10 | + # List of addresses in the form: |
| 11 | + # [{subnet: UUID1, address: IPv4}, {subnet: UUID2, address: IPv4)] |
| 12 | + # start with empty list |
| 13 | + self._addresses = [] |
| 14 | + self.assign_address = True |
| 15 | + |
| 16 | + def set_uuid(self, uuid): |
| 17 | + self._uuid = uuid |
| 18 | + |
| 19 | + def add_subnet(self, uuid, address): |
| 20 | + if uuid and address: |
| 21 | + self._addresses.append({"subnet": uuid, "address": address}) |
| 22 | + elif uuid: |
| 23 | + self._addresses.append({"subnet": uuid}) |
| 24 | + else: |
| 25 | + self._addresses.append({"address": address}) |
| 26 | + |
| 27 | + def as_json(self): |
| 28 | + if self._uuid and not self._addresses and self.assign_address: |
| 29 | + return {'network': self._uuid} |
| 30 | + elif self._uuid and self._addresses and self.assign_address: |
| 31 | + return {'network': self._uuid, 'addresses': self._addresses} |
| 32 | + elif not self._uuid and self.assign_address: |
| 33 | + return {'addresses': self._addresses} |
| 34 | + elif self._uuid and not self.assign_address: |
| 35 | + return {'network': self._uuid, 'addresses': []} |
| 36 | + raise NetworkParserError("Cannot create json string out of your interface definiton") |
| 37 | + |
| 38 | + |
| 39 | +class CloudscaleInterfaceParameterParser(): |
| 40 | + def __init__(self, string): |
| 41 | + self._string = string |
| 42 | + self._cursor = 0 |
| 43 | + |
| 44 | + def peek_strings(self, *args): |
| 45 | + for arg in args: |
| 46 | + if self._string[self._cursor:].startswith(arg): |
| 47 | + return arg |
| 48 | + |
| 49 | + def expect_string(self, arg): |
| 50 | + if arg and self._string[self._cursor:].startswith(arg): |
| 51 | + self._cursor += len(arg) |
| 52 | + return arg |
| 53 | + else: |
| 54 | + raise NetworkParserError(f"Expected '{arg}', but found '{self._string[self._cursor:]}' at position {self._cursor}") |
| 55 | + |
| 56 | + def expect_uuid(self): |
| 57 | + m = re.match(r'(^[\w]*-[\w]*-[\w]*-[\w]*-[\w]*)', self._string[self._cursor:]) |
| 58 | + try: |
| 59 | + uuid = str(m.group(0)) |
| 60 | + self._cursor += len(uuid) |
| 61 | + return uuid |
| 62 | + except Exception: |
| 63 | + raise NetworkParserError(f"Expected valid UUID, but found {self._string[self._cursor:]}") |
| 64 | + |
| 65 | + def expect_address(self): |
| 66 | + m = re.match(r'(^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})', self._string[self._cursor:]) |
| 67 | + try: |
| 68 | + address = str(m.group(0)) |
| 69 | + self._cursor += len(address) |
| 70 | + return address |
| 71 | + except Exception: |
| 72 | + raise NetworkParserError(f"Expected valid IPv4 address, but found '{self._string[self._cursor:]}'") |
| 73 | + |
| 74 | + def expect_string_or_uuid(self, arg=None): |
| 75 | + try: |
| 76 | + return self.expect_string(arg) |
| 77 | + except NetworkParserError: |
| 78 | + try: |
| 79 | + return self.expect_uuid() |
| 80 | + except NetworkParserError: |
| 81 | + raise NetworkParserError(f"Expected UUID or '{arg}', but found '{self._string[self._cursor:]}' at position {self._cursor}") |
| 82 | + |
| 83 | + def expect_end(self): |
| 84 | + if self._cursor != len(self._string): |
| 85 | + raise NetworkParserError(f"Expected the end, but found '{self._string[self._cursor:]}'") |
| 86 | + |
| 87 | + def has_empty_address(self): |
| 88 | + if self._string[self._cursor:] == 'address=': |
| 89 | + return True |
| 90 | + else: |
| 91 | + return False |
| 92 | + |
| 93 | + def parse_address(self): |
| 94 | + self.expect_string('address=') |
| 95 | + return self.expect_address() |
| 96 | + |
| 97 | + def parse_subnet(self): |
| 98 | + self.expect_string('subnet=') |
| 99 | + uuid = self.expect_string_or_uuid() |
| 100 | + if self.peek_strings(')', ',', '') == ',': |
| 101 | + self.expect_string(',') |
| 102 | + return uuid, self.parse_address() |
| 103 | + return uuid, None |
| 104 | + |
| 105 | + def parse_multiple_subnets(self, interface): |
| 106 | + while self.peek_strings('(') == '(': |
| 107 | + self.expect_string('(') |
| 108 | + subnet, address = self.parse_subnet() |
| 109 | + interface.add_subnet(subnet, address) |
| 110 | + self.expect_string(')') |
| 111 | + if self.peek_strings(',') == ',': |
| 112 | + self.expect_string(',') |
| 113 | + |
| 114 | +def parse_interface(string): |
| 115 | + parser = CloudscaleInterfaceParameterParser(string) |
| 116 | + interface = CloudscaleNetworkInterface() |
| 117 | + type_ = parser.peek_strings('network', 'subnet','(') |
| 118 | + if type_ == 'network': |
| 119 | + parser.expect_string('network') |
| 120 | + parser.expect_string('=') |
| 121 | + interface.set_uuid(parser.expect_string_or_uuid('public')) |
| 122 | + if parser.peek_strings(',') == ',': |
| 123 | + parser.expect_string(',') |
| 124 | + if parser.peek_strings('(') != '(': |
| 125 | + type_ = parser.peek_strings('subnet', 'address') |
| 126 | + if type_ == 'subnet': |
| 127 | + subnet, address = parser.parse_subnet() |
| 128 | + interface.add_subnet(subnet, address) |
| 129 | + elif type_ == 'address': |
| 130 | + if parser.has_empty_address(): |
| 131 | + parser.expect_string('address=') |
| 132 | + interface.assign_address = False |
| 133 | + else: |
| 134 | + raise NetworkParserError(f"Cannot set a fixed IP address without a subnet") |
| 135 | + else: |
| 136 | + parser.parse_multiple_subnets(interface) |
| 137 | + elif type_ == 'subnet': |
| 138 | + subnet, address = parser.parse_subnet() |
| 139 | + interface.add_subnet(subnet, address) |
| 140 | + elif type_ == '(': |
| 141 | + if not parser.peek_strings('(subnet') == '(subnet': |
| 142 | + raise NetworkParserError(f"Cannot add an interface without network or subnet definition") |
| 143 | + parser.parse_multiple_subnets(interface) |
| 144 | + else: |
| 145 | + raise NetworkParserError(f"Cannot add an interface without network or subnet definition") |
| 146 | + |
| 147 | + parser.expect_end() |
| 148 | + return interface |
0 commit comments