def test_fip_connection_for_address_scope(self):
        self.agent.conf.agent_mode = 'dvr_snat'
        (machine_same_scope, machine_diff_scope,
            router) = self._setup_address_scope('scope1', 'scope2', 'scope1')

        router.router[l3_constants.FLOATINGIP_KEY] = []
        fip_same_scope = '19.4.4.10'
        self._add_fip(router, fip_same_scope,
                      fixed_address=machine_same_scope.ip,
                      host=self.agent.conf.host,
                      fixed_ip_address_scope='scope1')
        fip_diff_scope = '19.4.4.11'
        self._add_fip(router, fip_diff_scope,
                      fixed_address=machine_diff_scope.ip,
                      host=self.agent.conf.host,
                      fixed_ip_address_scope='scope2')
        router.process(self.agent)

        br_ex = framework.get_ovs_bridge(
            self.agent.conf.external_network_bridge)
        src_machine = self.useFixture(
            machine_fixtures.FakeMachine(br_ex, '19.4.4.12/24'))
        # Floating ip should work no matter of address scope
        net_helpers.assert_ping(src_machine.namespace, fip_same_scope, 5)
        net_helpers.assert_ping(src_machine.namespace, fip_diff_scope, 5)

 def check_ping(self, from_site, to_site, instance=0, success=True):
     if success:
         net_helpers.assert_ping(from_site.vm[instance].namespace, to_site.vm[instance].port_ip, timeout=8, count=4)
     else:
         net_helpers.assert_no_ping(
             from_site.vm[instance].namespace, to_site.vm[instance].port_ip, timeout=8, count=4
         )

 def test_connection_from_same_address_scope(self):
     self.agent.conf.agent_mode = "dvr_snat"
     test_machine1, test_machine2, _ = self._setup_address_scope("scope1", "scope1")
     # Internal networks that are in the same address scope can connected
     # each other
     net_helpers.assert_ping(test_machine1.namespace, test_machine2.ip, 5)
     net_helpers.assert_ping(test_machine2.namespace, test_machine1.ip, 5)

    def _test_assert_pings_during_br_phys_setup_not_lost(self, provider_net):
        # Separate namespace is needed when pinging from one port to another,
        # otherwise Linux ping uses loopback instead for sending and receiving
        # ping, hence ignoring flow setup.
        ns_phys = self.useFixture(net_helpers.NamespaceFixture()).name

        ports = self.create_test_ports(amount=2)
        port_int = ports[0]
        port_phys = ports[1]
        ip_int = port_int['fixed_ips'][0]['ip_address']
        ip_phys = port_phys['fixed_ips'][0]['ip_address']

        self.setup_agent_and_ports(port_dicts=[port_int], create_tunnels=False,
                                   network=provider_net)

        self.plug_ports_to_phys_br(provider_net, [port_phys],
                                   namespace=ns_phys)

        # The OVS agent doesn't monitor the physical bridges, no notification
        # is sent when a port is up on a physical bridge, hence waiting only
        # for the ports connected to br-int
        self.wait_until_ports_state([port_int], up=True)
        # sanity pings before we start
        net_helpers.assert_ping(ns_phys, ip_int)
        net_helpers.assert_ping(self.namespace, ip_phys)

        with net_helpers.async_ping(ns_phys, [ip_int]) as done:
            while not done():
                self.agent.setup_physical_bridges(self.agent.bridge_mappings)
                time.sleep(0.25)

        with net_helpers.async_ping(self.namespace, [ip_phys]) as done:
            while not done():
                self.agent.setup_physical_bridges(self.agent.bridge_mappings)
                time.sleep(0.25)

 def _test_icmp_connectivity(self, direction, protocol, src_port, dst_port):
     src_namespace, ip_address = self._get_namespace_and_address(direction)
     try:
         net_helpers.assert_ping(src_namespace, ip_address)
     except RuntimeError:
         raise ConnectionTesterException(
             "ICMP packets can't get from %s namespace to %s address" % (
                 src_namespace, ip_address))

 def test_arp_spoof_disable_network_port(self):
     # block first and then disable port security to make sure old rules
     # are cleared
     self._setup_arp_spoof_for_port(self.dst_p.name, ["192.168.0.3"])
     self._setup_arp_spoof_for_port(self.dst_p.name, ["192.168.0.3"], device_owner=n_const.DEVICE_OWNER_ROUTER_GW)
     self.src_p.addr.add("%s/24" % self.src_addr)
     self.dst_p.addr.add("%s/24" % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr)

 def test_arp_spoof_disable_port_security(self):
     # block first and then disable port security to make sure old rules
     # are cleared
     self._setup_arp_spoof_for_port(self.dst_p.name, ["192.168.0.3"])
     self._setup_arp_spoof_for_port(self.dst_p.name, ["192.168.0.3"], psec=False)
     self.src_p.addr.add("%s/24" % self.src_addr)
     self.dst_p.addr.add("%s/24" % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)

 def test_arp_spoof_disable_network_port(self):
     # block first and then disable port security to make sure old rules
     # are cleared
     self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'])
     self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'],
                                    device_owner='network:router_gateway')
     self.src_p.addr.add('%s/24' % self.src_addr)
     self.dst_p.addr.add('%s/24' % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)

 def test_mac_spoof_blocks_wrong_mac(self):
     self._setup_arp_spoof_for_port(self.src_p.name, [self.src_addr])
     self._setup_arp_spoof_for_port(self.dst_p.name, [self.dst_addr])
     self.src_p.addr.add("%s/24" % self.src_addr)
     self.dst_p.addr.add("%s/24" % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr)
     # changing the allowed mac should stop the port from working
     self._setup_arp_spoof_for_port(self.src_p.name, [self.src_addr], mac="00:11:22:33:44:55")
     net_helpers.assert_no_ping(self.src_namespace, self.dst_addr)

 def test_arp_spoof_icmpv6_neigh_advt_allowed_address_pairs(self):
     self.src_addr = "2000::1"
     self.dst_addr = "2000::2"
     self._setup_arp_spoof_for_port(self.dst_p.name, ["2000::3", self.dst_addr])
     self.src_p.addr.add("%s/64" % self.src_addr)
     self.dst_p.addr.add("%s/64" % self.dst_addr)
     # make sure the IPv6 addresses are ready before pinging
     self.src_p.addr.wait_until_address_ready(self.src_addr)
     self.dst_p.addr.wait_until_address_ready(self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr)

 def _test_icmp_connectivity(self, direction, protocol, src_port, dst_port):
     src_namespace, ip_address = self._get_namespace_and_address(direction)
     ip_version = ip_lib.get_ip_version(ip_address)
     icmp_timeout = ICMP_VERSION_TIMEOUTS[ip_version]
     try:
         net_helpers.assert_ping(src_namespace, ip_address, timeout=icmp_timeout)
     except RuntimeError:
         raise ConnectionTesterException(
             "ICMP packets can't get from %s namespace to %s address" % (src_namespace, ip_address)
         )

 def test_arp_spoof_doesnt_block_ipv6(self):
     self.src_addr = "2000::1"
     self.dst_addr = "2000::2"
     self._setup_arp_spoof_for_port(self.src_p.name, [self.src_addr])
     self._setup_arp_spoof_for_port(self.dst_p.name, [self.dst_addr])
     self.src_p.addr.add("%s/64" % self.src_addr)
     self.dst_p.addr.add("%s/64" % self.dst_addr)
     # make sure the IPv6 addresses are ready before pinging
     self.src_p.addr.wait_until_address_ready(self.src_addr)
     self.dst_p.addr.wait_until_address_ready(self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)

    def test_connection_from_diff_address_scope_with_fip(self):
        (machine_same_scope, machine_diff_scope, router) = self._setup_address_scope("scope1", "scope2", "scope1")

        router.router[l3_constants.FLOATINGIP_KEY] = []
        fip = "19.4.4.11"
        self._add_fip(router, fip, fixed_address=machine_diff_scope.ip, fixed_ip_address_scope="scope2")
        router.process(self.agent)

        # For the internal networks that are in the same address scope as
        # external network, they should be able to reach the floating ip
        net_helpers.assert_ping(machine_same_scope.namespace, fip, 5)
        # For the port with fip, it should be able to reach the internal
        # networks that are in the same address scope as external network
        net_helpers.assert_ping(machine_diff_scope.namespace, machine_same_scope.ip, 5)

    def test_fip_connection_for_address_scope(self):
        (machine_same_scope, machine_diff_scope, router) = self._setup_address_scope("scope1", "scope2", "scope1")

        router.router[l3_constants.FLOATINGIP_KEY] = []
        fip_same_scope = "19.4.4.10"
        self._add_fip(router, fip_same_scope, fixed_address=machine_same_scope.ip, fixed_ip_address_scope="scope1")
        fip_diff_scope = "19.4.4.11"
        self._add_fip(router, fip_diff_scope, fixed_address=machine_diff_scope.ip, fixed_ip_address_scope="scope2")
        router.process(self.agent)

        br_ex = framework.get_ovs_bridge(self.agent.conf.external_network_bridge)
        src_machine = self.useFixture(machine_fixtures.FakeMachine(br_ex, "19.4.4.12/24"))
        # Floating ip should work no matter of address scope
        net_helpers.assert_ping(src_machine.namespace, fip_same_scope, 5)
        net_helpers.assert_ping(src_machine.namespace, fip_diff_scope, 5)

    def test_direct_route_for_address_scope(self):
        (machine_same_scope, machine_diff_scope, router) = self._setup_address_scope("scope1", "scope2", "scope1")

        gw_port = router.get_ex_gw_port()
        gw_ip = self._port_first_ip_cidr(gw_port).partition("/")[0]
        br_ex = framework.get_ovs_bridge(self.agent.conf.external_network_bridge)

        src_machine = self.useFixture(machine_fixtures.FakeMachine(br_ex, "19.4.4.12/24", gw_ip))
        # For the internal networks that are in the same address scope as
        # external network, they can directly route to external network
        net_helpers.assert_ping(src_machine.namespace, machine_same_scope.ip, 5)
        # For the internal networks that are not in the same address scope as
        # external networks. SNAT will be used. Direct route will not work
        # here.
        src_machine.assert_no_ping(machine_diff_scope.ip)

    def test_connection_from_diff_address_scope_with_fip(self):
        (machine_same_scope, machine_diff_scope,
            router) = self._setup_address_scope('scope1', 'scope2', 'scope1')
        fip = '19.4.4.11'
        self._add_fip(router, fip,
                      fixed_address=machine_diff_scope.ip,
                      host=self.agent.conf.host,
                      fixed_ip_address_scope='scope2')
        router.process(self.agent)

        # For the internal networks that are in the same address scope as
        # external network, they should be able to reach the floating ip
        net_helpers.assert_ping(machine_same_scope.namespace, fip, 5)
        # For the port with fip, it should be able to reach the internal
        # networks that are in the same address scope as external network
        net_helpers.assert_ping(machine_diff_scope.namespace,
                                machine_same_scope.ip, 5)

    def test_ipsec_site_connections(self):
        device = self.vpn_agent.device_drivers[0]
        # Mock the method below because it causes Exception:
        #   RuntimeError: Second simultaneous read on fileno 5 detected.
        #   Unless you really know what you're doing, make sure that only
        #   one greenthread can read any particular socket.  Consider using
        #   a pools.Pool. If you do know what you're doing and want to disable
        #   this error, call eventlet.debug.hub_prevent_multiple_readers(False)
        # Can reproduce the exception in the test only
        ip_lib.send_ip_addr_adv_notif = mock.Mock()
        # There are no vpn services yet. get_vpn_services_on_host returns
        # empty list
        device.agent_rpc.get_vpn_services_on_host = mock.Mock(
            return_value=[])
        # instantiate network resources "router", "private network"
        private_nets = list(PRIVATE_NET.subnet(24))
        site1 = self.site_setup(PUBLIC_NET[4], private_nets[1])
        site2 = self.site_setup(PUBLIC_NET[5], private_nets[2])
        # build vpn resources
        self.prepare_ipsec_conn_info(site1['vpn_service'],
                                     site2['vpn_service'])
        self.prepare_ipsec_conn_info(site2['vpn_service'],
                                     site1['vpn_service'])

        device.report_status = mock.Mock()
        device.agent_rpc.get_vpn_services_on_host = mock.Mock(
            return_value=[site1['vpn_service'],
                          site2['vpn_service']])

        net_helpers.assert_no_ping(site1['port_namespace'], site2['port_ip'],
                                   timeout=8, count=4)
        net_helpers.assert_no_ping(site2['port_namespace'], site1['port_ip'],
                                   timeout=8, count=4)

        device.sync(mock.Mock(), [{'id': site1['router'].router_id},
                                  {'id': site2['router'].router_id}])
        self.addCleanup(
            device._delete_vpn_processes,
            [site1['router'].router_id, site2['router'].router_id], [])

        net_helpers.assert_ping(site1['port_namespace'], site2['port_ip'],
                                timeout=8, count=4)
        net_helpers.assert_ping(site2['port_namespace'], site1['port_ip'],
                                timeout=8, count=4)

    def test_fip_connection_for_address_scope(self):
        (machine_same_scope, machine_diff_scope,
            router) = self._setup_address_scope('scope1', 'scope2', 'scope1')

        router.router[lib_constants.FLOATINGIP_KEY] = []
        fip_same_scope = '19.4.4.10'
        self._add_fip(router, fip_same_scope,
                      fixed_address=machine_same_scope.ip,
                      fixed_ip_address_scope='scope1')
        fip_diff_scope = '19.4.4.11'
        self._add_fip(router, fip_diff_scope,
                      fixed_address=machine_diff_scope.ip,
                      fixed_ip_address_scope='scope2')
        router.process()

        br_int = framework.get_ovs_bridge(
            self.agent.conf.ovs_integration_bridge)
        src_machine = self.useFixture(
            machine_fixtures.FakeMachine(br_int, '19.4.4.12/24'))
        # Floating ip should work no matter of address scope
        net_helpers.assert_ping(src_machine.namespace, fip_same_scope)
        net_helpers.assert_ping(src_machine.namespace, fip_diff_scope)

 def test_arp_spoof_allowed_address_pairs_0cidr(self):
     self._setup_arp_spoof_for_port(self.dst_p.name, ["9.9.9.9/0", "1.2.3.4"])
     self.src_p.addr.add("%s/24" % self.src_addr)
     self.dst_p.addr.add("%s/24" % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)

 def test_arp_spoof_allowed_address_pairs(self):
     self._setup_arp_spoof_for_port(self.dst_p.name, ["192.168.0.3", self.dst_addr])
     self.src_p.addr.add("%s/24" % self.src_addr)
     self.dst_p.addr.add("%s/24" % self.dst_addr)
     net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)