<commit_before># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for :py:mod:`flocker.route._iptables`.
"""

from __future__ import print_function

from os import getuid
from socket import socket
from unittest import skipUnless

from netifaces import AF_INET, interfaces, ifaddresses
from ipaddr import IPAddress

from twisted.trial.unittest import TestCase

from .. import create


def connect_nonblocking(ip, port):
    """
    Attempt a TCP connection to the given address without blocking.
    """
    client = socket()
    client.setblocking(False)
    client.connect_ex((ip.exploded, port))
    return client



def is_environment_configured():
    """
    Determine whether it is possible to exercise the proxy setup functionality
    in the current execution environment.

    :return: :obj:`True` if the proxy setup functionality could work given the
        underlying system and the privileges of this process, :obj:`False`
        otherwise.
    """
    # TODO: A nicer approach would be to create a new network namespace,
    # configure a couple interfaces with a couple addresses in it, and run the
    # test in the context of that network namespace.
    #
    # Something like:
    #
    #    ip netns create flocker-testing
    #    ip link add veth0 type veth peer name veth1
    #    ip link set veth1 netns flocker-testing
    #    ip netns exec flocker-testing ip link set dev veth1 up
    #    ip netns exec flocker-testing ip address add 10.0.0.1/24 dev veth1
    #    ip netns exec flocker-testing ip link set dev lo up
    #
    # Or, require such to be configured already.  That setup requires
    # privileged capabilities (probably CAP_SYS_ADMIN?) though.
    #
    # The functionality under test probably also requires privileged
    # capabilities (at least CAP_NET_ADMIN I think?) though.
    #
    # So for now just require root and crap on the host system. :/
    #
    # You might want to run these tests in a container! ;)
    #
    # -exarkun
    return getuid() == 0


ADDRESSES = [
    IPAddress(address['addr'])
    for name in interfaces()
    for address in ifaddresses(name).get(AF_INET, [])
]


class CreateTests(TestCase):
    """
    Tests for the creation of new external routing rules.
    """
    @skipUnless(
        is_environment_configured(),
        "Cannot test port forwarding without suitable test environment.")
    @skipUnless(
        len(ADDRESSES) >= 2,
        "Cannot test proxying without at least two addresses.")
    def setUp(self):
        """
        Select some addresses between which to proxy and set up a server to act
        as the target of the proxying.
        """
        self.serverAddress = ADDRESSES[0]
        self.proxyAddress = ADDRESSES[1]


        # This is the target of the proxy which will be created.
        self.server = socket()
        self.server.bind((self.serverAddress.exploded, 0))
        self.server.listen(1)

        # This is used to accept connections over the local network stack.
        # They should be nearly instantaneous.  If they are not then something
        # is *probably* wrong (and hopefully it isn't just an instance of the
        # machine being so loaded the local network stack can't complete a TCP
        # handshake in under one second...).
        self.server.settimeout(1)

        self.port = self.server.getsockname()[1]


    def test_setup(self):
        """
        A connection attempt to the server created in ``setUp`` is successful.
        """
        client = connect_nonblocking(self.serverAddress, self.port)
        accepted, client_address = self.server.accept()
        self.assertEqual(client.getsockname(), client_address)


    def test_connection(self):
        """
        A connection attempt is forwarded to the specified destination address.
        """
        # Note - we're leaking iptables rules into the system here.
        # https://github.com/hybridlogic/flocker/issues/22
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()
        self.assertEqual(client.getsockname(), client_address)


    def test_client_to_server(self):
        """
        A proxied connection will deliver bytes from the client side to the
        server side.
        """
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()

        client.send(b"x")
        self.assertEqual(b"x", accepted.recv(1))


    def test_server_to_client(self):
        """
        A proxied connection will deliver bytes from the server side to the
        client side.
        """
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()

        accepted.send(b"x")
        self.assertEqual(b"x", client.recv(1))<commit_msg>Add a test that connections to non-local addresses are not proxied.<commit_after># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for :py:mod:`flocker.route._iptables`.
"""

from __future__ import print_function

from errno import ECONNREFUSED
from os import getuid, getpid
from socket import error, socket
from unittest import skipUnless
from subprocess import check_output

from netifaces import AF_INET, interfaces, ifaddresses
from ipaddr import IPAddress, IPNetwork

from twisted.trial.unittest import SkipTest, TestCase

from .. import create


def connect_nonblocking(ip, port):
    """
    Attempt a TCP connection to the given address without blocking.
    """
    client = socket()
    client.setblocking(False)
    client.connect_ex((ip.exploded, port))
    return client



def is_environment_configured():
    """
    Determine whether it is possible to exercise the proxy setup functionality
    in the current execution environment.

    :return: :obj:`True` if the proxy setup functionality could work given the
        underlying system and the privileges of this process, :obj:`False`
        otherwise.
    """
    # TODO: A nicer approach would be to create a new network namespace,
    # configure a couple interfaces with a couple addresses in it, and run the
    # test in the context of that network namespace.
    #
    # Something like:
    #
    #    ip netns create flocker-testing
    #    ip link add veth0 type veth peer name veth1
    #    ip link set veth1 netns flocker-testing
    #    ip netns exec flocker-testing ip link set dev veth1 up
    #    ip netns exec flocker-testing ip address add 10.0.0.1/24 dev veth1
    #    ip netns exec flocker-testing ip link set dev lo up
    #
    # Or, require such to be configured already.  That setup requires
    # privileged capabilities (probably CAP_SYS_ADMIN?) though.
    #
    # The functionality under test probably also requires privileged
    # capabilities (at least CAP_NET_ADMIN I think?) though.
    #
    # So for now just require root and crap on the host system. :/
    #
    # You might want to run these tests in a container! ;)
    #
    # -exarkun
    return getuid() == 0


ADDRESSES = [
    IPAddress(address['addr'])
    for name in interfaces()
    for address in ifaddresses(name).get(AF_INET, [])
]


class CreateTests(TestCase):
    """
    Tests for the creation of new external routing rules.
    """
    @skipUnless(
        is_environment_configured(),
        "Cannot test port forwarding without suitable test environment.")
    @skipUnless(
        len(ADDRESSES) >= 2,
        "Cannot test proxying without at least two addresses.")
    def setUp(self):
        """
        Select some addresses between which to proxy and set up a server to act
        as the target of the proxying.
        """
        self.serverAddress = ADDRESSES[0]
        self.proxyAddress = ADDRESSES[1]


        # This is the target of the proxy which will be created.
        self.server = socket()
        self.server.bind((self.serverAddress.exploded, 0))
        self.server.listen(1)

        # This is used to accept connections over the local network stack.
        # They should be nearly instantaneous.  If they are not then something
        # is *probably* wrong (and hopefully it isn't just an instance of the
        # machine being so loaded the local network stack can't complete a TCP
        # handshake in under one second...).
        self.server.settimeout(1)

        self.port = self.server.getsockname()[1]


    def test_setup(self):
        """
        A connection attempt to the server created in ``setUp`` is successful.
        """
        client = connect_nonblocking(self.serverAddress, self.port)
        accepted, client_address = self.server.accept()
        self.assertEqual(client.getsockname(), client_address)


    def test_connection(self):
        """
        A connection attempt is forwarded to the specified destination address.
        """
        # Note - we're leaking iptables rules into the system here.
        # https://github.com/hybridlogic/flocker/issues/22
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()
        self.assertEqual(client.getsockname(), client_address)


    def test_client_to_server(self):
        """
        A proxied connection will deliver bytes from the client side to the
        server side.
        """
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()

        client.send(b"x")
        self.assertEqual(b"x", accepted.recv(1))


    def test_server_to_client(self):
        """
        A proxied connection will deliver bytes from the server side to the
        client side.
        """
        create(self.serverAddress, self.port)

        client = connect_nonblocking(self.proxyAddress, self.port)
        accepted, client_address = self.server.accept()

        accepted.send(b"x")
        self.assertEqual(b"x", client.recv(1))


    def test_remote_connections_unaffected(self):
        """
        A connection attempt to an IP not assigned to this host on the proxied
        port is not proxied.
        """
        networks = {
            IPNetwork("10.0.0.0/8"),
            IPNetwork("172.16.0.0/12"),
            IPNetwork("192.168.0.0/16")}

        for address in ADDRESSES:
            for network in networks:
                if address in network:
                    networks.remove(network)
                    break
        if not networks:
            raise SkipTest("No private networks available")

        network = next(iter(networks))
        gateway = network[1]
        address = network[2]

        # Create a remote "host" that the test can reliably fail a connection
        # attempt to.
        pid = getpid()
        veth0 = b"veth_" + hex(pid)
        veth1 = b"veth1"
        network_namespace = b"%s.%s" % (self.id(), getpid())

        def run(cmd):
            check_output(cmd.split())

        self.addCleanup(run, b"ip netns delete " + network_namespace)
        self.addCleanup(run, b"ip link delete " + veth0)

        ops = [
            b"ip netns add %(netns)s",
            b"ip link add %(veth0)s type veth peer name %(veth1)s",
            b"ip address add %(gateway)s dev %(veth0)s",
            b"ip link set dev %(veth0)s up",
            b"ip link set %(veth1)s netns %(netns)s",
            b"ip netns exec %(netns)s ip link set dev %(veth1)s up",
            b"ip netns exec %(netns)s ip address add %(address)s dev %(veth1)s",
            b"ip route add %(network)s dev %(veth0)s scope link",
            b"ip netns exec %(netns)s ip route add default dev %(veth1)s",
        ]

        params = dict(
            netns=network_namespace, veth0=veth0, veth1=veth1,
            address=address, gateway=gateway, network=network,
            )
        for op in ops:
            run(op % params)

        create(self.serverAddress, self.port)

        client = socket()
        client.settimeout(1)
        exception = self.assertRaises(
            error, client.connect, (str(address), self.port))
        self.assertEqual(ECONNREFUSED, exception.errno)
{'fraction_non_alphanumeric': 0.08619713754346243, 'fraction_numerical': 0.007520012937656667, 'mean_word_length': 2.4997170345217885, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 5, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 2, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2644045', 'n_tokens_mistral': 3537, 'n_tokens_neox': 3368, 'n_words': 1192}