path: root/net/test/multinetwork_test.py

#!/usr/bin/python3
#
# Copyright 2014 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import cstruct
import ctypes
import errno
import os
import random
from socket import *  # pylint: disable=wildcard-import
import struct
import time           # pylint: disable=unused-import
import unittest

from scapy import all as scapy

import csocket
import iproute
import multinetwork_base
import net_test
import netlink
import packets

# For brevity.
UDP_PAYLOAD = net_test.UDP_PAYLOAD

IPV6_FLOWINFO = 11

SYNCOOKIES_SYSCTL = "/proc/sys/net/ipv4/tcp_syncookies"
TCP_MARK_ACCEPT_SYSCTL = "/proc/sys/net/ipv4/tcp_fwmark_accept"


class OutgoingTest(multinetwork_base.MultiNetworkBaseTest):

  # How many times to run outgoing packet tests.
  ITERATIONS = 5

  def CheckPingPacket(self, version, netid, routing_mode, packet):
    s = self.BuildSocket(version, net_test.PingSocket, netid, routing_mode)

    myaddr = self.MyAddress(version, netid)
    mysockaddr = self.MySocketAddress(version, netid)
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.bind((mysockaddr, packets.PING_IDENT))
    net_test.SetSocketTos(s, packets.PING_TOS)

    dstaddr = self.GetRemoteAddress(version)
    dstsockaddr = self.GetRemoteSocketAddress(version)
    desc, expected = packets.ICMPEcho(version, myaddr, dstaddr)
    msg = "IPv%d ping: expected %s on %s" % (
        version, desc, self.GetInterfaceName(netid))

    s.sendto(packet + packets.PING_PAYLOAD, (dstsockaddr, 19321))

    self.ExpectPacketOn(netid, msg, expected)

  def CheckTCPSYNPacket(self, version, netid, routing_mode):
    s = self.BuildSocket(version, net_test.TCPSocket, netid, routing_mode)

    myaddr = self.MyAddress(version, netid)
    dstaddr = self.GetRemoteAddress(version)
    dstsockaddr = self.GetRemoteSocketAddress(version)
    desc, expected = packets.SYN(53, version, myaddr, dstaddr,
                                 sport=None, seq=None)


    # Non-blocking TCP connects always return EINPROGRESS.
    self.assertRaisesErrno(errno.EINPROGRESS, s.connect, (dstsockaddr, 53))
    msg = "IPv%s TCP connect: expected %s on %s" % (
        version, desc, self.GetInterfaceName(netid))
    self.ExpectPacketOn(netid, msg, expected)
    s.close()

  def CheckUDPPacket(self, version, netid, routing_mode):
    s = self.BuildSocket(version, net_test.UDPSocket, netid, routing_mode)

    myaddr = self.MyAddress(version, netid)
    dstaddr = self.GetRemoteAddress(version)
    dstsockaddr = self.GetRemoteSocketAddress(version)

    desc, expected = packets.UDP(version, myaddr, dstaddr, sport=None)
    msg = "IPv%s UDP %%s: expected %s on %s" % (
        version, desc, self.GetInterfaceName(netid))

    s.sendto(UDP_PAYLOAD, (dstsockaddr, 53))
    self.ExpectPacketOn(netid, msg % "sendto", expected)

    # IP_UNICAST_IF doesn't seem to work on connected sockets, so no TCP.
    if routing_mode != "ucast_oif":
      s.connect((dstsockaddr, 53))
      s.send(UDP_PAYLOAD)
      self.ExpectPacketOn(netid, msg % "connect/send", expected)
      s.close()

  def CheckRawGrePacket(self, version, netid, routing_mode):
    s = self.BuildSocket(version, net_test.RawGRESocket, netid, routing_mode)

    inner_version = {4: 6, 6: 4}[version]
    inner_src = self.MyAddress(inner_version, netid)
    inner_dst = self.GetRemoteAddress(inner_version)
    inner = bytes(packets.UDP(inner_version, inner_src, inner_dst, sport=None)[1])

    ethertype = {4: net_test.ETH_P_IP, 6: net_test.ETH_P_IPV6}[inner_version]
    # A GRE header can be as simple as two zero bytes and the ethertype.
    packet = struct.pack("!i", ethertype) + inner
    myaddr = self.MyAddress(version, netid)
    dstaddr = self.GetRemoteAddress(version)

    s.sendto(packet, (dstaddr, IPPROTO_GRE))
    desc, expected = packets.GRE(version, myaddr, dstaddr, ethertype, inner)
    msg = "Raw IPv%d GRE with inner IPv%d UDP: expected %s on %s" % (
        version, inner_version, desc, self.GetInterfaceName(netid))
    self.ExpectPacketOn(netid, msg, expected)

  def CheckOutgoingPackets(self, routing_mode):
    for _ in range(self.ITERATIONS):
      for netid in self.tuns:

        self.CheckPingPacket(4, netid, routing_mode, self.IPV4_PING)
        # Kernel bug.
        if routing_mode != "oif":
          self.CheckPingPacket(6, netid, routing_mode, self.IPV6_PING)

        # IP_UNICAST_IF doesn't seem to work on connected sockets, so no TCP.
        if routing_mode != "ucast_oif":
          self.CheckTCPSYNPacket(4, netid, routing_mode)
          self.CheckTCPSYNPacket(6, netid, routing_mode)
          self.CheckTCPSYNPacket(5, netid, routing_mode)

        self.CheckUDPPacket(4, netid, routing_mode)
        self.CheckUDPPacket(6, netid, routing_mode)
        self.CheckUDPPacket(5, netid, routing_mode)

        # Creating raw sockets on non-root UIDs requires properly setting
        # capabilities, which is hard to do from Python.
        # IP_UNICAST_IF is not supported on raw sockets.
        if routing_mode not in ["uid", "ucast_oif"]:
          self.CheckRawGrePacket(4, netid, routing_mode)
          self.CheckRawGrePacket(6, netid, routing_mode)

  def testMarkRouting(self):
    """Checks that socket marking selects the right outgoing interface."""
    self.CheckOutgoingPackets("mark")

  def testUidRouting(self):
    """Checks that UID routing selects the right outgoing interface."""
    self.CheckOutgoingPackets("uid")

  def testOifRouting(self):
    """Checks that oif routing selects the right outgoing interface."""
    self.CheckOutgoingPackets("oif")

  def testUcastOifRouting(self):
    """Checks that ucast oif routing selects the right outgoing interface."""
    self.CheckOutgoingPackets("ucast_oif")

  def CheckRemarking(self, version, use_connect):
    modes = ["mark", "oif", "uid"]
    # Setting UNICAST_IF on connected sockets does not work.
    if not use_connect:
      modes += ["ucast_oif"]

    for mode in modes:
      s = net_test.UDPSocket(self.GetProtocolFamily(version))

      # Figure out what packets to expect.
      sport = net_test.BindRandomPort(version, s)
      dstaddr = {4: self.IPV4_ADDR, 6: self.IPV6_ADDR}[version]
      unspec = {4: "0.0.0.0", 6: "::"}[version]  # Placeholder.
      desc, expected = packets.UDP(version, unspec, dstaddr, sport)

      # If we're testing connected sockets, connect the socket on the first
      # netid now.
      if use_connect:
        netid = list(self.tuns.keys())[0]
        self.SelectInterface(s, netid, mode)
        s.connect((dstaddr, 53))
        expected.src = self.MyAddress(version, netid)

      # For each netid, select that network without closing the socket, and
      # check that the packets sent on that socket go out on the right network.
      #
      # For connected sockets, routing is cached in the socket's destination
      # cache entry. In this case, we check that selecting the network a second
      # time on the same socket (except via SO_BINDTODEVICE, or SO_MARK on 5.0+
      # kernels) does not change routing, but that subsequently invalidating the
      # destination cache entry does. This is a bug in the kernel because
      # re-selecting the netid should cause routing to change, and future
      # kernels may fix this bug for per-UID routing and ucast_oif routing like
      # they already have for mark-based routing. But until they do, this
      # behaviour provides a convenient way to check that InvalidateDstCache
      # actually works.
      prevnetid = None
      for netid in self.tuns:
        self.SelectInterface(s, netid, mode)
        if not use_connect:
          expected.src = self.MyAddress(version, netid)

        def ExpectSendUsesNetid(netid):
          connected_str = "Connected" if use_connect else "Unconnected"
          msg = "%s UDPv%d socket remarked using %s: expecting %s on %s" % (
              connected_str, version, mode, desc, self.GetInterfaceName(netid))
          if use_connect:
            s.send(UDP_PAYLOAD)
          else:
            s.sendto(UDP_PAYLOAD, (dstaddr, 53))
          self.ExpectPacketOn(netid, msg, expected)

        # Does this socket have a stale dst cache entry that we need to clear?
        def SocketHasStaleDstCacheEntry():
          if not prevnetid:
            # This is the first time we're marking the socket.
            return False
          if not use_connect:
            # Non-connected sockets never have dst cache entries.
            return False
          if mode in ["uid", "ucast_oif"]:
            # No kernel invalidates the dst cache entry if the UID or the
            # UCAST_OIF socket option changes.
            return True
          if mode == "oif":
            # Changing SO_BINDTODEVICE always invalidates the dst cache entry.
            return False
          if mode == "mark":
            # Changing the mark invalidates the dst cache entry in 5.0+.
            return net_test.LINUX_VERSION < (5, 0, 0)
          raise AssertionError("%s must be one of %s" % (mode, modes))

        if SocketHasStaleDstCacheEntry():
            ExpectSendUsesNetid(prevnetid)
            # ... until we invalidate it.
            self.InvalidateDstCache(version, prevnetid)

        # In any case, future sends must be correct.
        ExpectSendUsesNetid(netid)

        self.SelectInterface(s, None, mode)
        prevnetid = netid

  def testIPv4Remarking(self):
    """Checks that updating the mark on an IPv4 socket changes routing."""
    self.CheckRemarking(4, False)
    self.CheckRemarking(4, True)

  def testIPv6Remarking(self):
    """Checks that updating the mark on an IPv6 socket changes routing."""
    self.CheckRemarking(6, False)
    self.CheckRemarking(6, True)

  def testIPv6StickyPktinfo(self):
    for _ in range(self.ITERATIONS):
      for netid in self.tuns:
        s = net_test.UDPSocket(AF_INET6)

        # Set a flowlabel.
        net_test.SetFlowLabel(s, net_test.IPV6_ADDR, 0xdead)
        s.setsockopt(net_test.SOL_IPV6, net_test.IPV6_FLOWINFO_SEND, 1)

        # Set some destination options.
        nonce = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c"
        dstopts = b"".join([
            b"\x11\x02",               # Next header=UDP, 24 bytes of options.
            b"\x01\x06", b"\x00" * 6,  # PadN, 6 bytes of padding.
            b"\x8b\x0c",               # ILNP nonce, 12 bytes.
            nonce
        ])
        s.setsockopt(net_test.SOL_IPV6, IPV6_DSTOPTS, dstopts)
        s.setsockopt(net_test.SOL_IPV6, IPV6_UNICAST_HOPS, 255)

        pktinfo = multinetwork_base.MakePktInfo(6, None, self.ifindices[netid])

        # Set the sticky pktinfo option.
        s.setsockopt(net_test.SOL_IPV6, IPV6_PKTINFO, pktinfo)

        # Specify the flowlabel in the destination address.
        s.sendto(UDP_PAYLOAD, (net_test.IPV6_ADDR, 53, 0xdead, 0))

        sport = s.getsockname()[1]
        srcaddr = self.MyAddress(6, netid)
        expected = (scapy.IPv6(src=srcaddr, dst=net_test.IPV6_ADDR,
                               fl=0xdead, hlim=255) /
                    scapy.IPv6ExtHdrDestOpt(
                        options=[scapy.PadN(optdata="\x00\x00\x00\x00\x00\x00"),
                                 scapy.HBHOptUnknown(otype=0x8b,
                                                     optdata=nonce)]) /
                    scapy.UDP(sport=sport, dport=53) /
                    UDP_PAYLOAD)
        msg = "IPv6 UDP using sticky pktinfo: expected UDP packet on %s" % (
            self.GetInterfaceName(netid))
        self.ExpectPacketOn(netid, msg, expected)

  def CheckPktinfoRouting(self, version):
    for _ in range(self.ITERATIONS):
      for netid in self.tuns:
        family = self.GetProtocolFamily(version)
        s = net_test.UDPSocket(family)

        if version == 6:
          # Create a flowlabel so we can use it.
          net_test.SetFlowLabel(s, net_test.IPV6_ADDR, 0xbeef)

          # Specify some arbitrary options.
          # We declare the flowlabel as ctypes.c_uint32 because on a 32-bit
          # Python interpreter an integer greater than 0x7fffffff (such as our
          # chosen flowlabel after being passed through htonl) is converted to
          # long, and _MakeMsgControl doesn't know what to do with longs.
          cmsgs = [
              (net_test.SOL_IPV6, IPV6_HOPLIMIT, 39),
              (net_test.SOL_IPV6, IPV6_TCLASS, 0x83),
              (net_test.SOL_IPV6, IPV6_FLOWINFO, ctypes.c_uint(htonl(0xbeef))),
          ]
        else:
          # Support for setting IPv4 TOS and TTL via cmsg only appeared in 3.13.
          cmsgs = []
          s.setsockopt(net_test.SOL_IP, IP_TTL, 39)
          s.setsockopt(net_test.SOL_IP, IP_TOS, 0x83)

        dstaddr = self.GetRemoteAddress(version)
        self.SendOnNetid(version, s, dstaddr, 53, netid, UDP_PAYLOAD, cmsgs)

        sport = s.getsockname()[1]
        srcaddr = self.MyAddress(version, netid)

        desc, expected = packets.UDPWithOptions(version, srcaddr, dstaddr,
                                                sport=sport)

        msg = "IPv%d UDP using pktinfo routing: expected %s on %s" % (
            version, desc, self.GetInterfaceName(netid))
        self.ExpectPacketOn(netid, msg, expected)

  def testIPv4PktinfoRouting(self):
    self.CheckPktinfoRouting(4)

  def testIPv6PktinfoRouting(self):
    self.CheckPktinfoRouting(6)


class MarkTest(multinetwork_base.InboundMarkingTest):

  def CheckReflection(self, version, gen_packet, gen_reply):
    """Checks that replies go out on the same interface as the original.

    For each combination:
     - Calls gen_packet to generate a packet to that IP address.
     - Writes the packet generated by gen_packet on the given tun
       interface, causing the kernel to receive it.
     - Checks that the kernel's reply matches the packet generated by
       gen_reply.

    Args:
      version: An integer, 4 or 6.
      gen_packet: A function taking an IP version (an integer), a source
        address and a destination address (strings), and returning a scapy
        packet.
      gen_reply: A function taking the same arguments as gen_packet,
        plus a scapy packet, and returning a scapy packet.
    """
    for netid, iif, ip_if, myaddr, remoteaddr in self.Combinations(version):
      # Generate a test packet.
      desc, packet = gen_packet(version, remoteaddr, myaddr)

      # Test with mark reflection enabled and disabled.
      for reflect in [0, 1]:
        self.SetMarkReflectSysctls(reflect)
        # HACK: IPv6 ping replies always do a routing lookup with the
        # interface the ping came in on. So even if mark reflection is not
        # working, IPv6 ping replies will be properly reflected. Don't
        # fail when that happens.
        if reflect or desc == "ICMPv6 echo":
          reply_desc, reply = gen_reply(version, myaddr, remoteaddr, packet)
        else:
          reply_desc, reply = None, None

        msg = self._FormatMessage(iif, ip_if, "reflect=%d" % reflect,
                                  desc, reply_desc)
        self._ReceiveAndExpectResponse(netid, packet, reply, msg)

  def SYNToClosedPort(self, *args):
    return packets.SYN(999, *args)

  def testIPv4ICMPErrorsReflectMark(self):
    self.CheckReflection(4, packets.UDP, packets.ICMPPortUnreachable)

  def testIPv6ICMPErrorsReflectMark(self):
    self.CheckReflection(6, packets.UDP, packets.ICMPPortUnreachable)

  def testIPv4PingRepliesReflectMarkAndTos(self):
    self.CheckReflection(4, packets.ICMPEcho, packets.ICMPReply)

  def testIPv6PingRepliesReflectMarkAndTos(self):
    self.CheckReflection(6, packets.ICMPEcho, packets.ICMPReply)

  def testIPv4RSTsReflectMark(self):
    self.CheckReflection(4, self.SYNToClosedPort, packets.RST)

  def testIPv6RSTsReflectMark(self):
    self.CheckReflection(6, self.SYNToClosedPort, packets.RST)


class TCPAcceptTest(multinetwork_base.InboundMarkingTest):

  MODE_BINDTODEVICE = "SO_BINDTODEVICE"
  MODE_INCOMING_MARK = "incoming mark"
  MODE_EXPLICIT_MARK = "explicit mark"
  MODE_UID = "uid"

  @classmethod
  def setUpClass(cls):
    super(TCPAcceptTest, cls).setUpClass()

    # Open a port so we can observe SYN+ACKs. Since it's a dual-stack socket it
    # will accept both IPv4 and IPv6 connections. We do this here instead of in
    # each test so we can use the same socket every time. That way, if a kernel
    # bug causes incoming packets to mark the listening socket instead of the
    # accepted socket, the test will fail as soon as the next address/interface
    # combination is tried.
    cls.listensocket = net_test.IPv6TCPSocket()
    cls.listenport = net_test.BindRandomPort(6, cls.listensocket)

  def _SetTCPMarkAcceptSysctl(self, value):
    self.SetSysctl(TCP_MARK_ACCEPT_SYSCTL, value)

  def CheckTCPConnection(self, mode, listensocket, netid, version,
                         myaddr, remoteaddr, packet, reply, msg):
    establishing_ack = packets.ACK(version, remoteaddr, myaddr, reply)[1]

    # Attempt to confuse the kernel.
    self.InvalidateDstCache(version, netid)

    self.ReceivePacketOn(netid, establishing_ack)

    # If we're using UID routing, the accept() call has to be run as a UID that
    # is routed to the specified netid, because the UID of the socket returned
    # by accept() is the effective UID of the process that calls it. It doesn't
    # need to be the same UID; any UID that selects the same interface will do.
    with net_test.RunAsUid(self.UidForNetid(netid)):
      s, _ = listensocket.accept()

    try:
      # Check that data sent on the connection goes out on the right interface.
      desc, data = packets.ACK(version, myaddr, remoteaddr, establishing_ack,
                               payload=UDP_PAYLOAD)
      s.send(UDP_PAYLOAD)
      self.ExpectPacketOn(netid, msg + ": expecting %s" % desc, data)
      self.InvalidateDstCache(version, netid)

      # Keep up our end of the conversation.
      ack = packets.ACK(version, remoteaddr, myaddr, data)[1]
      self.InvalidateDstCache(version, netid)
      self.ReceivePacketOn(netid, ack)

      mark = self.GetSocketMark(s)
    finally:
      self.InvalidateDstCache(version, netid)
      s.close()
      self.InvalidateDstCache(version, netid)

    if mode == self.MODE_INCOMING_MARK:
      self.assertEqual(netid, mark & self.NETID_FWMASK,
                        msg + ": Accepted socket: Expected mark %d, got %d" % (
                            netid, mark))
    elif mode != self.MODE_EXPLICIT_MARK:
      self.assertEqual(0, self.GetSocketMark(listensocket))

    # Check the FIN was sent on the right interface, and ack it. We don't expect
    # this to fail because by the time the connection is established things are
    # likely working, but a) extra tests are always good and b) extra packets
    # like the FIN (and retransmitted FINs) could cause later tests that expect
    # no packets to fail.
    desc, fin = packets.FIN(version, myaddr, remoteaddr, ack)
    self.ExpectPacketOn(netid, msg + ": expecting %s after close" % desc, fin)

    desc, finack = packets.FIN(version, remoteaddr, myaddr, fin)
    self.ReceivePacketOn(netid, finack)

    # Since we called close() earlier, the userspace socket object is gone, so
    # the socket has no UID. If we're doing UID routing, the ack might be routed
    # incorrectly. Not much we can do here.
    desc, finackack = packets.ACK(version, myaddr, remoteaddr, finack)
    self.ExpectPacketOn(netid, msg + ": expecting final ack", finackack)

  def CheckTCP(self, version, modes):
    """Checks that incoming TCP connections work.

    Args:
      version: An integer, 4 or 6.
      modes: A list of modes to excercise.
    """
    for syncookies in [0, 2]:
      for mode in modes:
        for netid, iif, ip_if, myaddr, remoteaddr in self.Combinations(version):
          listensocket = self.listensocket
          listenport = listensocket.getsockname()[1]

          accept_sysctl = 1 if mode == self.MODE_INCOMING_MARK else 0
          self._SetTCPMarkAcceptSysctl(accept_sysctl)
          self.SetMarkReflectSysctls(accept_sysctl)

          bound_dev = iif if mode == self.MODE_BINDTODEVICE else None
          self.BindToDevice(listensocket, bound_dev)

          mark = netid if mode == self.MODE_EXPLICIT_MARK else 0
          self.SetSocketMark(listensocket, mark)

          uid = self.UidForNetid(netid) if mode == self.MODE_UID else 0
          os.fchown(listensocket.fileno(), uid, -1)

          # Generate the packet here instead of in the outer loop, so
          # subsequent TCP connections use different source ports and
          # retransmissions from old connections don't confuse subsequent
          # tests.
          desc, packet = packets.SYN(listenport, version, remoteaddr, myaddr)

          if mode:
            reply_desc, reply = packets.SYNACK(version, myaddr, remoteaddr,
                                               packet)
          else:
            reply_desc, reply = None, None

          extra = "mode=%s, syncookies=%d" % (mode, syncookies)
          msg = self._FormatMessage(iif, ip_if, extra, desc, reply_desc)
          reply = self._ReceiveAndExpectResponse(netid, packet, reply, msg)
          if reply:
            self.CheckTCPConnection(mode, listensocket, netid, version, myaddr,
                                    remoteaddr, packet, reply, msg)

  def testBasicTCP(self):
    self.CheckTCP(4, [None, self.MODE_BINDTODEVICE, self.MODE_EXPLICIT_MARK])
    self.CheckTCP(6, [None, self.MODE_BINDTODEVICE, self.MODE_EXPLICIT_MARK])

  def testIPv4MarkAccept(self):
    self.CheckTCP(4, [self.MODE_INCOMING_MARK])

  def testIPv6MarkAccept(self):
    self.CheckTCP(6, [self.MODE_INCOMING_MARK])

  def testIPv4UidAccept(self):
    self.CheckTCP(4, [self.MODE_UID])

  def testIPv6UidAccept(self):
    self.CheckTCP(6, [self.MODE_UID])

  def testIPv6ExplicitMark(self):
    self.CheckTCP(6, [self.MODE_EXPLICIT_MARK])

@unittest.skipUnless(multinetwork_base.HAVE_AUTOCONF_TABLE,
                     "need support for per-table autoconf")
class RIOTest(multinetwork_base.MultiNetworkBaseTest):
  """Test for IPv6 RFC 4191 route information option

  Relevant kernel commits:
    upstream:
      f104a567e673 ipv6: use rt6_get_dflt_router to get default router in rt6_route_rcv
      bbea124bc99d net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs.

    android-4.9:
      d860b2e8a7f1 FROMLIST: net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs

    android-4.4:
      e953f89b8563 net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs.

    android-4.1:
      84f2f47716cd net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs.

    android-3.18:
      65f8936934fa net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs.

    android-3.10:
      161e88ebebc7 net: ipv6: Add sysctl for minimum prefix len acceptable in RIOs.

  """

  def setUp(self):
    super(RIOTest, self).setUp()
    self.NETID = random.choice(self.NETIDS)
    self.IFACE = self.GetInterfaceName(self.NETID)
    # return min/max plen to default values before each test case
    self.SetAcceptRaRtInfoMinPlen(0)
    self.SetAcceptRaRtInfoMaxPlen(0)

  def GetRoutingTable(self):
    return self._TableForNetid(self.NETID)

  def SetAcceptRaRtInfoMinPlen(self, plen):
    self.SetSysctl(
        "/proc/sys/net/ipv6/conf/%s/accept_ra_rt_info_min_plen"
        % self.IFACE, plen)

  def GetAcceptRaRtInfoMinPlen(self):
    return int(self.GetSysctl(
        "/proc/sys/net/ipv6/conf/%s/accept_ra_rt_info_min_plen" % self.IFACE))

  def SetAcceptRaRtInfoMaxPlen(self, plen):
    self.SetSysctl(
        "/proc/sys/net/ipv6/conf/%s/accept_ra_rt_info_max_plen"
        % self.IFACE, plen)

  def GetAcceptRaRtInfoMaxPlen(self):
    return int(self.GetSysctl(
        "/proc/sys/net/ipv6/conf/%s/accept_ra_rt_info_max_plen" % self.IFACE))

  def SendRIO(self, rtlifetime, plen, prefix, prf):
    options = scapy.ICMPv6NDOptRouteInfo(rtlifetime=rtlifetime, plen=plen,
                                         prefix=prefix, prf=prf)
    self.SendRA(self.NETID, options=(options,))

  def FindRoutesWithDestination(self, destination):
    canonical = net_test.CanonicalizeIPv6Address(destination)
    return [r for _, r in self.iproute.DumpRoutes(6, self.GetRoutingTable())
            if ('RTA_DST' in r and r['RTA_DST'] == canonical)]

  def FindRoutesWithGateway(self):
    return [r for _, r in self.iproute.DumpRoutes(6, self.GetRoutingTable())
            if 'RTA_GATEWAY' in r]

  def CountRoutes(self):
    return len(self.iproute.DumpRoutes(6, self.GetRoutingTable()))

  def GetRouteExpiration(self, route):
    return float(route['RTA_CACHEINFO'].expires) / 100.0

  def AssertExpirationInRange(self, routes, lifetime, epsilon):
    self.assertTrue(routes)
    found = False
    # Assert that at least one route in routes has the expected lifetime
    for route in routes:
      expiration = self.GetRouteExpiration(route)
      if expiration < lifetime - epsilon:
        continue
      if expiration > lifetime + epsilon:
        continue
      found = True
    self.assertTrue(found)

  def DelRA6(self, prefix, plen):
    version = 6
    netid = self.NETID
    table = self._TableForNetid(netid)
    router = self._RouterAddress(netid, version)
    ifindex = self.ifindices[netid]
    self.iproute._Route(version, iproute.RTPROT_RA, iproute.RTM_DELROUTE,
                        table, prefix, plen, router, ifindex, None, None)

  def testSetAcceptRaRtInfoMinPlen(self):
    for plen in range(-1, 130):
      self.SetAcceptRaRtInfoMinPlen(plen)
      self.assertEqual(plen, self.GetAcceptRaRtInfoMinPlen())

  def testSetAcceptRaRtInfoMaxPlen(self):
    for plen in range(-1, 130):
      self.SetAcceptRaRtInfoMaxPlen(plen)
      self.assertEqual(plen, self.GetAcceptRaRtInfoMaxPlen())

  def testZeroRtLifetime(self):
    PREFIX = "2001:db8:8901:2300::"
    RTLIFETIME = 73500
    PLEN = 56
    PRF = 0
    self.SetAcceptRaRtInfoMaxPlen(PLEN)
    self.SendRIO(RTLIFETIME, PLEN, PREFIX, PRF)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    self.assertTrue(self.FindRoutesWithDestination(PREFIX))
    # RIO with rtlifetime = 0 should remove from routing table
    self.SendRIO(0, PLEN, PREFIX, PRF)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    self.assertFalse(self.FindRoutesWithDestination(PREFIX))

  def testMinPrefixLenRejection(self):
    PREFIX = "2001:db8:8902:2345::"
    RTLIFETIME = 70372
    PRF = 0
    # sweep from high to low to avoid spurious failures from late arrivals.
    for plen in range(130, 1, -1):
      self.SetAcceptRaRtInfoMinPlen(plen)
      # RIO with plen < min_plen should be ignored
      self.SendRIO(RTLIFETIME, plen - 1, PREFIX, PRF)
    # Give the kernel time to notice our RAs
    time.sleep(0.1)
    # Expect no routes
    routes = self.FindRoutesWithDestination(PREFIX)
    self.assertFalse(routes)

  def testMaxPrefixLenRejection(self):
    PREFIX = "2001:db8:8903:2345::"
    RTLIFETIME = 73078
    PRF = 0
    # sweep from low to high to avoid spurious failures from late arrivals.
    for plen in range(-1, 128, 1):
      self.SetAcceptRaRtInfoMaxPlen(plen)
      # RIO with plen > max_plen should be ignored
      self.SendRIO(RTLIFETIME, plen + 1, PREFIX, PRF)
    # Give the kernel time to notice our RAs
    time.sleep(0.1)
    # Expect no routes
    routes = self.FindRoutesWithDestination(PREFIX)
    self.assertFalse(routes)

  def testSimpleAccept(self):
    PREFIX = "2001:db8:8904:2345::"
    RTLIFETIME = 9993
    PRF = 0
    PLEN = 56
    self.SetAcceptRaRtInfoMinPlen(48)
    self.SetAcceptRaRtInfoMaxPlen(64)
    self.SendRIO(RTLIFETIME, PLEN, PREFIX, PRF)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    routes = self.FindRoutesWithGateway()
    self.AssertExpirationInRange(routes, RTLIFETIME, 1)
    self.DelRA6(PREFIX, PLEN)

  def testEqualMinMaxAccept(self):
    PREFIX = "2001:db8:8905:2345::"
    RTLIFETIME = 6326
    PLEN = 21
    PRF = 0
    self.SetAcceptRaRtInfoMinPlen(PLEN)
    self.SetAcceptRaRtInfoMaxPlen(PLEN)
    self.SendRIO(RTLIFETIME, PLEN, PREFIX, PRF)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    routes = self.FindRoutesWithGateway()
    self.AssertExpirationInRange(routes, RTLIFETIME, 1)
    self.DelRA6(PREFIX, PLEN)

  def testZeroLengthPrefix(self):
    PREFIX = "2001:db8:8906:2345::"
    RTLIFETIME = self.RA_VALIDITY * 2
    PLEN = 0
    PRF = 0
    # Max plen = 0 still allows default RIOs!
    self.SetAcceptRaRtInfoMaxPlen(PLEN)
    self.SendRA(self.NETID)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    default = self.FindRoutesWithGateway()
    self.AssertExpirationInRange(default, self.RA_VALIDITY, 1)
    # RIO with prefix length = 0, should overwrite default route lifetime
    # note that the RIO lifetime overwrites the RA lifetime.
    self.SendRIO(RTLIFETIME, PLEN, PREFIX, PRF)
    # Give the kernel time to notice our RA
    time.sleep(0.01)
    default = self.FindRoutesWithGateway()
    self.AssertExpirationInRange(default, RTLIFETIME, 1)
    self.DelRA6(PREFIX, PLEN)

  def testManyRIOs(self):
    RTLIFETIME = 68012
    PLEN = 56
    PRF = 0
    COUNT = 1000
    baseline = self.CountRoutes()
    self.SetAcceptRaRtInfoMaxPlen(56)
    # Send many RIOs compared to the expected number on a healthy system.
    for i in range(0, COUNT):
      prefix = "2001:db8:%x:1100::" % i
      self.SendRIO(RTLIFETIME, PLEN, prefix, PRF)
    time.sleep(0.1)
    self.assertEqual(COUNT + baseline, self.CountRoutes())
    for i in range(0, COUNT):
      prefix = "2001:db8:%x:1100::" % i
      self.DelRA6(prefix, PLEN)
    # Expect that we can return to baseline config without lingering routes.
    self.assertEqual(baseline, self.CountRoutes())

class RATest(multinetwork_base.MultiNetworkBaseTest):

  ND_ROUTER_ADVERT = 134
  ND_OPT_PREF64 = 38
  Pref64Option = cstruct.Struct("pref64_option", "!BBH12s",
                                "type length lft_plc prefix")

  def testDoesNotHaveObsoleteSysctl(self):
    self.assertFalse(os.path.isfile(
        "/proc/sys/net/ipv6/route/autoconf_table_offset"))

  @unittest.skipUnless(multinetwork_base.HAVE_AUTOCONF_TABLE,
                       "no support for per-table autoconf")
  def testPurgeDefaultRouters(self):

    def CheckIPv6Connectivity(expect_connectivity):
      for netid in self.NETIDS:
        s = net_test.UDPSocket(AF_INET6)
        self.SetSocketMark(s, netid)
        if expect_connectivity:
          self.assertTrue(s.sendto(UDP_PAYLOAD, (net_test.IPV6_ADDR, 1234)))
        else:
          self.assertRaisesErrno(errno.ENETUNREACH, s.sendto, UDP_PAYLOAD,
                                 (net_test.IPV6_ADDR, 1234))

    try:
      CheckIPv6Connectivity(True)
      self.SetIPv6SysctlOnAllIfaces("accept_ra", 1)
      self.SetSysctl("/proc/sys/net/ipv6/conf/all/forwarding", 1)
      CheckIPv6Connectivity(False)
    finally:
      self.SetSysctl("/proc/sys/net/ipv6/conf/all/forwarding", 0)
      for netid in self.NETIDS:
        self.SendRA(netid)
      CheckIPv6Connectivity(True)

  def testOnlinkCommunication(self):
    """Checks that on-link communication goes direct and not through routers."""
    for netid in self.tuns:
      # Send a UDP packet to a random on-link destination.
      s = net_test.UDPSocket(AF_INET6)
      iface = self.GetInterfaceName(netid)
      self.BindToDevice(s, iface)
      # dstaddr can never be our address because GetRandomDestination only fills
      # in the lower 32 bits, but our address has 0xff in the byte before that
      # (since it's constructed from the EUI-64 and so has ff:fe in the middle).
      dstaddr = self.GetRandomDestination(self.OnlinkPrefix(6, netid))
      s.sendto(UDP_PAYLOAD, (dstaddr, 53))

      # Expect an NS for that destination on the interface.
      myaddr = self.MyAddress(6, netid)
      mymac = self.MyMacAddress(netid)
      desc, expected = packets.NS(myaddr, dstaddr, mymac)
      msg = "Sending UDP packet to on-link destination: expecting %s" % desc
      time.sleep(0.0001)  # Required to make the test work on kernel 3.1(!)
      self.ExpectPacketOn(netid, msg, expected)

      # Send an NA.
      tgtmac = "02:00:00:00:%02x:99" % netid
      _, reply = packets.NA(dstaddr, myaddr, tgtmac)
      # Don't use ReceivePacketOn, since that uses the router's MAC address as
      # the source. Instead, construct our own Ethernet header with source
      # MAC of tgtmac.
      reply = scapy.Ether(src=tgtmac, dst=mymac) / reply
      self.ReceiveEtherPacketOn(netid, reply)

      # Expect the kernel to send the original UDP packet now that the ND cache
      # entry has been populated.
      sport = s.getsockname()[1]
      desc, expected = packets.UDP(6, myaddr, dstaddr, sport=sport)
      msg = "After NA response, expecting %s" % desc
      self.ExpectPacketOn(netid, msg, expected)

  # This test documents a known issue: routing tables are never deleted.
  @unittest.skipUnless(multinetwork_base.HAVE_AUTOCONF_TABLE,
                       "no support for per-table autoconf")
  def testLeftoverRoutes(self):
    def GetNumRoutes():
      with open("/proc/net/ipv6_route") as ipv6_route:
        return len(ipv6_route.readlines())

    num_routes = GetNumRoutes()
    for i in range(10, 20):
      try:
        self.tuns[i] = self.CreateTunInterface(i)
        self.SendRA(i)
        self.tuns[i].close()
      finally:
        del self.tuns[i]
    self.assertLess(num_routes, GetNumRoutes())

  def SendNdUseropt(self, option):
    options = scapy.ICMPv6NDOptRouteInfo(rtlifetime=rtlifetime, plen=plen,
                                         prefix=prefix, prf=prf)
    self.SendRA(self.NETID, options=(options,))

  def MakePref64Option(self, prefix, lifetime):
    prefix = inet_pton(AF_INET6, prefix)[:12]
    lft_plc = (lifetime & 0xfff8) | 0  # 96-bit prefix length
    return self.Pref64Option((self.ND_OPT_PREF64, 2, lft_plc, prefix))

  def testPref64UserOption(self):
    # Open a netlink socket to receive RTM_NEWNDUSEROPT messages.
    s = netlink.NetlinkSocket(netlink.NETLINK_ROUTE, iproute.RTMGRP_ND_USEROPT)

    # Send an RA with the PREF64 option.
    netid = random.choice(self.NETIDS)
    opt = self.MakePref64Option("64:ff9b::", 300)
    self.SendRA(netid, options=(opt.Pack(),))

    # Check that we get an an RTM_NEWNDUSEROPT message on the socket with the
    # expected option.
    csocket.SetSocketTimeout(s.sock, 100)
    try:
      data = s._Recv()
    except IOError as e:
      self.fail("Should have received an RTM_NEWNDUSEROPT message. "
                "Please ensure the kernel supports receiving the "
                "PREF64 RA option. Error: %s" % e)

    # Check that the message is received correctly.
    nlmsghdr, data = cstruct.Read(data, netlink.NLMsgHdr)
    self.assertEqual(iproute.RTM_NEWNDUSEROPT, nlmsghdr.type)

    # Check the option contents.
    ndopthdr, data = cstruct.Read(data, iproute.NdUseroptMsg)
    self.assertEqual(AF_INET6, ndopthdr.family)
    self.assertEqual(self.ND_ROUTER_ADVERT, ndopthdr.icmp_type)
    self.assertEqual(len(opt), ndopthdr.opts_len)

    actual_opt = self.Pref64Option(data)
    self.assertEqual(opt, actual_opt)



class PMTUTest(multinetwork_base.InboundMarkingTest):

  PAYLOAD_SIZE = 1400
  dstaddrs = set()

  def GetSocketMTU(self, version, s):
    if version == 6:
      ip6_mtuinfo = s.getsockopt(net_test.SOL_IPV6, csocket.IPV6_PATHMTU, 32)
      unused_sockaddr, mtu = struct.unpack("=28sI", ip6_mtuinfo)
      return mtu
    else:
      return s.getsockopt(net_test.SOL_IP, csocket.IP_MTU)

  def DisableFragmentationAndReportErrors(self, version, s):
    if version == 4:
      s.setsockopt(net_test.SOL_IP, csocket.IP_MTU_DISCOVER,
                   csocket.IP_PMTUDISC_DO)
      s.setsockopt(net_test.SOL_IP, net_test.IP_RECVERR, 1)
    else:
      s.setsockopt(net_test.SOL_IPV6, csocket.IPV6_DONTFRAG, 1)
      s.setsockopt(net_test.SOL_IPV6, net_test.IPV6_RECVERR, 1)

  def CheckPMTU(self, version, use_connect, modes):

    def SendBigPacket(version, s, dstaddr, netid, payload):
      if use_connect:
        s.send(payload)
      else:
        self.SendOnNetid(version, s, dstaddr, 1234, netid, payload, [])

    for netid in self.tuns:
      for mode in modes:
        s = self.BuildSocket(version, net_test.UDPSocket, netid, mode)
        self.DisableFragmentationAndReportErrors(version, s)

        srcaddr = self.MyAddress(version, netid)
        dst_prefix, intermediate = {
            4: ("172.19.", "172.16.9.12"),
            6: ("2001:db8::", "2001:db8::1")
        }[version]

        # Run this test often enough (e.g., in presubmits), and eventually
        # we'll be unlucky enough to pick the same address twice, in which
        # case the test will fail because the kernel will already have seen
        # the lower MTU. Don't do this.
        dstaddr = self.GetRandomDestination(dst_prefix)
        while dstaddr in self.dstaddrs:
          dstaddr = self.GetRandomDestination(dst_prefix)
        self.dstaddrs.add(dstaddr)

        if use_connect:
          s.connect((dstaddr, 1234))

        payload = self.PAYLOAD_SIZE * b"a"

        # Send a packet and receive a packet too big.
        SendBigPacket(version, s, dstaddr, netid, payload)
        received = self.ReadAllPacketsOn(netid)
        self.assertEqual(1, len(received),
                          "unexpected packets: %s" % received[1:])
        _, toobig = packets.ICMPPacketTooBig(version, intermediate, srcaddr,
                                             received[0])
        self.ReceivePacketOn(netid, toobig)

        # Check that another send on the same socket returns EMSGSIZE.
        self.assertRaisesErrno(
            errno.EMSGSIZE,
            SendBigPacket, version, s, dstaddr, netid, payload)

        # If this is a connected socket, make sure the socket MTU was set.
        # Note that in IPv4 this only started working in Linux 3.6!
        if use_connect:
          self.assertEqual(packets.PTB_MTU, self.GetSocketMTU(version, s))

        s.close()

        # Check that other sockets pick up the PMTU we have been told about by
        # connecting another socket to the same destination and getting its MTU.
        # This new socket can use any method to select its outgoing interface;
        # here we use a mark for simplicity.
        s2 = self.BuildSocket(version, net_test.UDPSocket, netid, "mark")
        s2.connect((dstaddr, 1234))
        self.assertEqual(packets.PTB_MTU, self.GetSocketMTU(version, s2))

        # Also check the MTU reported by ip route get, this time using the oif.
        routes = self.iproute.GetRoutes(dstaddr, self.ifindices[netid], 0, None)
        self.assertTrue(routes)
        route = routes[0]
        rtmsg, attributes = route
        self.assertEqual(iproute.RTN_UNICAST, rtmsg.type)
        metrics = attributes["RTA_METRICS"]
        self.assertEqual(packets.PTB_MTU, metrics["RTAX_MTU"])

  def testIPv4BasicPMTU(self):
    """Tests IPv4 path MTU discovery.

    Relevant kernel commits:
      upstream net-next:
        6a66271 ipv4, fib: pass LOOPBACK_IFINDEX instead of 0 to flowi4_iif

      android-3.10:
        4bc64dd ipv4, fib: pass LOOPBACK_IFINDEX instead of 0 to flowi4_iif
    """

    self.CheckPMTU(4, True, ["mark", "oif"])
    self.CheckPMTU(4, False, ["mark", "oif"])

  def testIPv6BasicPMTU(self):
    self.CheckPMTU(6, True, ["mark", "oif"])
    self.CheckPMTU(6, False, ["mark", "oif"])

  def testIPv4UIDPMTU(self):
    self.CheckPMTU(4, True, ["uid"])
    self.CheckPMTU(4, False, ["uid"])

  def testIPv6UIDPMTU(self):
    self.CheckPMTU(6, True, ["uid"])
    self.CheckPMTU(6, False, ["uid"])

  # Making Path MTU Discovery work on unmarked  sockets requires that mark
  # reflection be enabled. Otherwise the kernel has no way to know what routing
  # table the original packet used, and thus it won't be able to clone the
  # correct route.

  def testIPv4UnmarkedSocketPMTU(self):
    self.SetMarkReflectSysctls(1)
    try:
      self.CheckPMTU(4, False, [None])
    finally:
      self.SetMarkReflectSysctls(0)

  def testIPv6UnmarkedSocketPMTU(self):
    self.SetMarkReflectSysctls(1)
    try:
      self.CheckPMTU(6, False, [None])
    finally:
      self.SetMarkReflectSysctls(0)


class UidRoutingTest(multinetwork_base.MultiNetworkBaseTest):
  """Tests that per-UID routing works properly.

  Relevant kernel commits:
    upstream net-next:
      7d99569460 net: ipv4: Don't crash if passing a null sk to ip_do_redirect.
      d109e61bfe net: ipv4: Don't crash if passing a null sk to ip_rt_update_pmtu.
      35b80733b3 net: core: add missing check for uid_range in rule_exists.
      e2d118a1cb net: inet: Support UID-based routing in IP protocols.
      622ec2c9d5 net: core: add UID to flows, rules, and routes
      86741ec254 net: core: Add a UID field to struct sock.

    android-3.18:
      b004e79504 net: ipv4: Don't crash if passing a null sk to ip_rt_update_pmtu.
      04c0eace81 net: inet: Support UID-based routing in IP protocols.
      18c36d7b71 net: core: add UID to flows, rules, and routes
      80e3440721 net: core: Add a UID field to struct sock.
      fa8cc2c30c Revert "net: core: Support UID-based routing."
      b585141890 Revert "Handle 'sk' being NULL in UID-based routing."
      5115ab7514 Revert "net: core: fix UID-based routing build"
      f9f4281f79 Revert "ANDROID: net: fib: remove duplicate assignment"

    android-4.4:
      341965cf10 net: ipv4: Don't crash if passing a null sk to ip_rt_update_pmtu.
      344afd627c net: inet: Support UID-based routing in IP protocols.
      03441d56d8 net: core: add UID to flows, rules, and routes
      eb964bdba7 net: core: Add a UID field to struct sock.
      9789b697c6 Revert "net: core: Support UID-based routing."
  """

  def GetRulesAtPriority(self, version, priority):
    rules = self.iproute.DumpRules(version)
    out = [(rule, attributes) for rule, attributes in rules
           if attributes.get("FRA_PRIORITY", 0) == priority]
    return out

  def CheckInitialTablesHaveNoUIDs(self, version):
    rules = []
    for priority in [0, 32766, 32767]:
      rules.extend(self.GetRulesAtPriority(version, priority))
    for _, attributes in rules:
      self.assertNotIn("FRA_UID_RANGE", attributes)

  def testIPv4InitialTablesHaveNoUIDs(self):
    self.CheckInitialTablesHaveNoUIDs(4)

  def testIPv6InitialTablesHaveNoUIDs(self):
    self.CheckInitialTablesHaveNoUIDs(6)

  @staticmethod
  def _Random():
    return random.randint(1000000, 2000000)

  def CheckGetAndSetRules(self, version):
    start, end = tuple(sorted([self._Random(), self._Random()]))
    table = self._Random()
    priority = self._Random()

    # Can't create a UID range to UID -1 because -1 is INVALID_UID...
    self.assertRaisesErrno(
        errno.EINVAL,
        self.iproute.UidRangeRule, version, True, 100, 0xffffffff, table,
        priority)

    # ... but -2 is valid.
    self.iproute.UidRangeRule(version, True, 100, 0xfffffffe, table, priority)
    self.iproute.UidRangeRule(version, False, 100, 0xfffffffe, table, priority)

    try:
      # Create a UID range rule.
      self.iproute.UidRangeRule(version, True, start, end, table, priority)

      # Check that deleting the wrong UID range doesn't work.
      self.assertRaisesErrno(
          errno.ENOENT,
          self.iproute.UidRangeRule, version, False, start, end + 1, table,
          priority)
      self.assertRaisesErrno(errno.ENOENT,
        self.iproute.UidRangeRule, version, False, start + 1, end, table,
        priority)

      # Check that the UID range appears in dumps.
      rules = self.GetRulesAtPriority(version, priority)
      self.assertTrue(rules)
      _, attributes = rules[-1]
      self.assertEqual(priority, attributes["FRA_PRIORITY"])
      uidrange = attributes["FRA_UID_RANGE"]
      self.assertEqual(start, uidrange.start)
      self.assertEqual(end, uidrange.end)
      self.assertEqual(table, attributes["FRA_TABLE"])
    finally:
      self.iproute.UidRangeRule(version, False, start, end, table, priority)
      self.assertRaisesErrno(
          errno.ENOENT,
          self.iproute.UidRangeRule, version, False, start, end, table,
          priority)

    fwmask = 0xfefefefe
    try:
      # Create a rule without a UID range.
      self.iproute.FwmarkRule(version, True, 300, fwmask, 301, priority + 1)

      # Check it doesn't have a UID range.
      rules = self.GetRulesAtPriority(version, priority + 1)
      self.assertTrue(rules)
      for _, attributes in rules:
        self.assertIn("FRA_TABLE", attributes)
        self.assertNotIn("FRA_UID_RANGE", attributes)
    finally:
      self.iproute.FwmarkRule(version, False, 300, fwmask, 301, priority + 1)

    # Test that EEXIST worksfor UID range rules too.
    ranges = [(100, 101), (100, 102), (99, 101), (1234, 5678)]
    dup = ranges[0]
    try:
      # Check that otherwise identical rules with different UID ranges can be
      # created without EEXIST.
      for start, end in ranges:
        self.iproute.UidRangeRule(version, True, start, end, table, priority)
      # ... but EEXIST is returned if the UID range is identical.
      self.assertRaisesErrno(
        errno.EEXIST,
        self.iproute.UidRangeRule, version, True, dup[0], dup[1], table,
        priority)
    finally:
      # Clean up.
      for start, end in ranges + [dup]:
        try:
          self.iproute.UidRangeRule(version, False, start, end, table,
                                    priority)
        except IOError:
          pass

  def testIPv4GetAndSetRules(self):
    self.CheckGetAndSetRules(4)

  def testIPv6GetAndSetRules(self):
    self.CheckGetAndSetRules(6)

  def testDeleteErrno(self):
    for version in [4, 6]:
      table = self._Random()
      priority = self._Random()
      self.assertRaisesErrno(
          errno.EINVAL,
          self.iproute.UidRangeRule, version, False, 100, 0xffffffff, table,
          priority)

  def ExpectNoRoute(self, addr, oif, mark, uid):
    # The lack of a route may be either an error, or an unreachable route.
    try:
      routes = self.iproute.GetRoutes(addr, oif, mark, uid)
      rtmsg, _ = routes[0]
      self.assertEqual(iproute.RTN_UNREACHABLE, rtmsg.type)
    except IOError as e:
      if int(e.errno) != int(errno.ENETUNREACH):
        raise e

  def ExpectRoute(self, addr, oif, mark, uid):
    routes = self.iproute.GetRoutes(addr, oif, mark, uid)
    rtmsg, _ = routes[0]
    self.assertEqual(iproute.RTN_UNICAST, rtmsg.type)

  def CheckGetRoute(self, version, addr):
    self.ExpectNoRoute(addr, 0, 0, 0)
    for netid in self.NETIDS:
      uid = self.UidForNetid(netid)
      self.ExpectRoute(addr, 0, 0, uid)
    self.ExpectNoRoute(addr, 0, 0, 0)

  def testIPv4RouteGet(self):
    self.CheckGetRoute(4, net_test.IPV4_ADDR)

  def testIPv6RouteGet(self):
    self.CheckGetRoute(6, net_test.IPV6_ADDR)

  def testChangeFdAttributes(self):
    netid = random.choice(self.NETIDS)
    uid = self._Random()
    table = self._TableForNetid(netid)
    remoteaddr = self.GetRemoteAddress(6)
    s = socket(AF_INET6, SOCK_DGRAM, 0)

    def CheckSendFails():
      self.assertRaisesErrno(errno.ENETUNREACH,
                             s.sendto, b"foo", (remoteaddr, 53))
    def CheckSendSucceeds():
      self.assertEqual(len(b"foo"), s.sendto(b"foo", (remoteaddr, 53)))

    CheckSendFails()
    self.iproute.UidRangeRule(6, True, uid, uid, table, self.PRIORITY_UID)
    try:
      CheckSendFails()
      os.fchown(s.fileno(), uid, -1)
      CheckSendSucceeds()
      os.fchown(s.fileno(), -1, -1)
      CheckSendSucceeds()
      os.fchown(s.fileno(), -1, 12345)
      CheckSendSucceeds()
      os.fchmod(s.fileno(), 0o777)
      CheckSendSucceeds()
      os.fchown(s.fileno(), 0, -1)
      CheckSendFails()
    finally:
      self.iproute.UidRangeRule(6, False, uid, uid, table, self.PRIORITY_UID)


class RulesTest(net_test.NetworkTest):

  RULE_PRIORITY = 99999
  FWMASK = 0xffffffff

  def setUp(self):
    self.iproute = iproute.IPRoute()
    for version in [4, 6]:
      self.iproute.DeleteRulesAtPriority(version, self.RULE_PRIORITY)

  def tearDown(self):
    for version in [4, 6]:
      self.iproute.DeleteRulesAtPriority(version, self.RULE_PRIORITY)

  def testRuleDeletionMatchesTable(self):
    for version in [4, 6]:
      # Add rules with mark 300 pointing at tables 301 and 302.
      # This checks for a kernel bug where deletion request for tables > 256
      # ignored the table.
      self.iproute.FwmarkRule(version, True, 300, self.FWMASK, 301,
                              priority=self.RULE_PRIORITY)
      self.iproute.FwmarkRule(version, True, 300, self.FWMASK, 302,
                              priority=self.RULE_PRIORITY)
      # Delete rule with mark 300 pointing at table 302.
      self.iproute.FwmarkRule(version, False, 300, self.FWMASK, 302,
                              priority=self.RULE_PRIORITY)
      # Check that the rule pointing at table 301 is still around.
      attributes = [a for _, a in self.iproute.DumpRules(version)
                    if a.get("FRA_PRIORITY", 0) == self.RULE_PRIORITY]
      self.assertEqual(1, len(attributes))
      self.assertEqual(301, attributes[0]["FRA_TABLE"])


if __name__ == "__main__":
  unittest.main()