(rules)
| 1061 | |
| 1062 | |
| 1063 | def parse_network_rules(rules): |
| 1064 | ret = [] |
| 1065 | |
| 1066 | if rules is None or len(rules) == 0: |
| 1067 | return ret |
| 1068 | |
| 1069 | lines = rules.split('NEXT;')[:-1] |
| 1070 | for line in lines: |
| 1071 | tokens = line.split(';', 3) |
| 1072 | if len(tokens) != 4: |
| 1073 | continue |
| 1074 | |
| 1075 | ruletype, protocol = tokens[0].split(':') |
| 1076 | start = int(tokens[1]) |
| 1077 | end = int(tokens[2]) |
| 1078 | cidrs = tokens.pop() |
| 1079 | |
| 1080 | ipv4 = [] |
| 1081 | ipv6 = [] |
| 1082 | for ip in cidrs.split(","): |
| 1083 | try: |
| 1084 | network = ipaddress.ip_network(ip, False) |
| 1085 | if network.version == 4: |
| 1086 | ipv4.append(ip) |
| 1087 | else: |
| 1088 | ipv6.append(ip) |
| 1089 | except: |
| 1090 | pass |
| 1091 | |
| 1092 | ret.append({'ipv4': ipv4, 'ipv6': ipv6, 'ruletype': ruletype, |
| 1093 | 'start': start, 'end': end, 'protocol': protocol}) |
| 1094 | |
| 1095 | return ret |
| 1096 | |
| 1097 | |
| 1098 | def add_network_rules(vm_name, vm_id, vm_ip, vm_ip6, signature, seqno, vmMac, rules, vif, brname, sec_ips): |
no test coverage detected