(self)
| 278 | logging.debug("EGRESS rule configured for protocol ==> %s, action ==> %s", rule['protocol'], rule['action']) |
| 279 | |
| 280 | def add_routing_rules(self): |
| 281 | fw = self.config.get_nft_ipv4_fw() |
| 282 | logging.info("Processing routing firewall rules %s: %s" % (self.dbag, fw)) |
| 283 | chains_added = False |
| 284 | egress_policy = None |
| 285 | for item in self.dbag: |
| 286 | if item == "id": |
| 287 | continue |
| 288 | rule = self.dbag[item] |
| 289 | |
| 290 | network = ipaddress.ip_network(self.config.cmdline().get_eth0_ip() + "/" + self.config.cmdline().get_cidr_size(), False) |
| 291 | guest_cidr = network.with_prefixlen |
| 292 | if chains_added is False: |
| 293 | parent_chain = "FORWARD" |
| 294 | chain = "fw_chain_egress" |
| 295 | parent_chain_rule = "ip saddr %s jump %s" % (guest_cidr, chain) |
| 296 | fw.append({'type': "chain", 'chain': chain}) |
| 297 | fw.append({'type': "", 'chain': parent_chain, 'rule': parent_chain_rule}) |
| 298 | chain = "fw_chain_ingress" |
| 299 | parent_chain_rule = "ip daddr %s jump %s" % (guest_cidr, chain) |
| 300 | fw.append({'type': "chain", 'chain': chain}) |
| 301 | fw.append({'type': "", 'chain': parent_chain, 'rule': parent_chain_rule}) |
| 302 | if rule['default_egress_policy']: |
| 303 | egress_policy = "accept" |
| 304 | else: |
| 305 | egress_policy = "drop" |
| 306 | chains_added = True |
| 307 | |
| 308 | rstr = "" |
| 309 | |
| 310 | chain = "fw_chain_ingress" |
| 311 | if 'traffic_type' in rule and rule['traffic_type'].lower() == "egress": |
| 312 | chain = "fw_chain_egress" |
| 313 | |
| 314 | saddr = "" |
| 315 | if 'source_cidr_list' in rule and len(rule['source_cidr_list']) > 0: |
| 316 | source_cidrs = rule['source_cidr_list'] |
| 317 | if len(source_cidrs) == 1: |
| 318 | source_cidrs = source_cidrs[0] |
| 319 | else: |
| 320 | source_cidrs = "{" + (",".join(source_cidrs)) + "}" |
| 321 | saddr = "ip saddr " + source_cidrs |
| 322 | daddr = "" |
| 323 | if 'dest_cidr_list' in rule and len(rule['dest_cidr_list']) > 0: |
| 324 | dest_cidrs = rule['dest_cidr_list'] |
| 325 | if len(dest_cidrs) == 1: |
| 326 | dest_cidrs = dest_cidrs[0] |
| 327 | else: |
| 328 | dest_cidrs = "{" + (",".join(dest_cidrs)) + "}" |
| 329 | daddr = "ip daddr " + dest_cidrs |
| 330 | |
| 331 | proto = "" |
| 332 | protocol = rule['protocol'] |
| 333 | if protocol != "all": |
| 334 | icmp_type = "" |
| 335 | proto = protocol |
| 336 | if proto == "icmp": |
| 337 | proto = proto_str = "icmp" |
no test coverage detected