neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall module

class neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall.FWGPortMap

Bases: object

create_port(port: OFPort, port_dict: dict[str, Any]) → None

delete_fwg(fwg_id: str) → None

get_fwg(fwg_id: str) → FirewallGroup | None

get_or_create_fwg(fwg_id: str) → FirewallGroup

remove_port(port: OFPort) → None

update_members(fwg_id: str, members: dict[str, list[str]]) → None

update_port(port: OFPort, port_dict: dict[str, Any]) → None

update_rules(fwg_id: str, ingress_rules: list[dict[str, Any]], egress_rules: list[dict[str, Any]]) → None

class neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall.FirewallGroup(id_: str)

Bases: object

get_ethertype_filtered_addresses(ethertype: str, exclude_addresses: list[str] | None = None) → list[str]

update_rules(ingress_rules: list[dict[str, Any]], egress_rules: list[dict[str, Any]]) → None

Update firewall group with ingress/egress rules.

If a rule has a protocol field, it is normalized to a number here in order to ease later processing.

class neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall.OFPort(port_dict: dict[str, Any], ovs_port: Any, vlan_tag: int)

Bases: object

property all_allowed_macs: set[str]

property ipv4_addresses: list[str]

property ipv6_addresses: list[str]

update(port_dict: dict[str, Any]) → None

class neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall.OVSFirewallDriver(agent_api: Any, sg_with_ovs: bool = False)

Bases: FirewallL2DriverBase

REQUIRED_PROTOCOLS = ['OpenFlow10', 'OpenFlow11', 'OpenFlow12', 'OpenFlow13', 'OpenFlow14']

add_flows_from_rules(port: OFPort) → None

create_firewall_group(ports_for_fwg: list[dict[str, Any]], firewall_group: dict[str, Any]) → None

Called when a firewall group is created.

create_rules_generator_for_port(port: OFPort) → Iterator[dict[str, Any]]

Returns a generator emitting rules valid for further processing

Injects necessary fields to feed one-by-one to rules module to transform into valid openflow rules.

delete_all_port_flows(port: OFPort) → None

Delete all flows for given port

delete_firewall_group(ports_for_fwg: list[dict[str, Any]], firewall_group: dict[str, Any]) → None

Called when a firewall group is deleted.

filter_defer_apply_off() → None

Turn off deferral of rules and apply the rules now.

filter_defer_apply_on() → None

Defer application of filtering rule.

get_ofport(port: dict[str, Any]) → OFPort | None

get_or_create_ofport(port: dict[str, Any]) → OFPort

Get ofport specified by port[‘device’], checking and reflecting ofport changes. If ofport is nonexistent, create and return one.

get_ovs_port(port_id: str) → Any

static initialize_bridge(int_br: Any) → Any

initialize_port_flows(port: OFPort) → None

Set base flows for port

Parameters:

port – OFPort instance

is_port_managed(port: dict[str, Any]) → bool

property ports: dict[str, dict[str, Any]]

Returns filtered ports.

prepare_port_filter(port: dict[str, Any]) → None

process_trusted_ports(ports: list[dict[str, Any]]) → None

Pass packets from these ports directly to ingress pipeline.

provides_arp_spoofing_protection = True

remove_port_filter(port: dict[str, Any]) → None

Remove port from firewall

All flows related to this port are removed from ovs. Port is also removed from ports managed by this firewall.

remove_trusted_ports(port_ids: list[str]) → None

update_firewall_group(ports_for_fwg: list[dict[str, Any]], firewall_group: dict[str, Any]) → None

Called when a firewall group is updated.

update_firewall_group_rules(fwg_id: str, ingress_rules: list[dict[str, Any]], egress_rules: list[dict[str, Any]]) → None

update_port_filter(port: dict[str, Any]) → None

Update rules for given port

Current existing filtering rules are removed and new ones are generated based on current loaded firewall group rules and members.

Note: port no security should be handled by security group in co-existence mode, otherwise fwg will handle it.

neutron_fwaas.services.firewall.service_drivers.agents.drivers.linux.l2.openvswitch_firewall.firewall.create_reg_numbers(flow_params: dict[str, Any]) → None

Replace reg_(port|net) values with defined register numbers