Home | History | Annotate | Download | only in net_test
      1 #!/usr/bin/python
      2 #
      3 # Copyright 2015 The Android Open Source Project
      4 #
      5 # Licensed under the Apache License, Version 2.0 (the "License");
      6 # you may not use this file except in compliance with the License.
      7 # You may obtain a copy of the License at
      8 #
      9 # http://www.apache.org/licenses/LICENSE-2.0
     10 #
     11 # Unless required by applicable law or agreed to in writing, software
     12 # distributed under the License is distributed on an "AS IS" BASIS,
     13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     14 # See the License for the specific language governing permissions and
     15 # limitations under the License.
     16 
     17 import errno
     18 import random
     19 from socket import *  # pylint: disable=wildcard-import
     20 import time
     21 import unittest
     22 
     23 from scapy import all as scapy
     24 
     25 import multinetwork_base
     26 import net_test
     27 
     28 
     29 RTMGRP_NEIGH = 4
     30 
     31 NUD_INCOMPLETE = 0x01
     32 NUD_REACHABLE = 0x02
     33 NUD_STALE = 0x04
     34 NUD_DELAY = 0x08
     35 NUD_PROBE = 0x10
     36 NUD_FAILED = 0x20
     37 NUD_PERMANENT = 0x80
     38 
     39 
     40 # TODO: Support IPv4.
     41 class NeighbourTest(multinetwork_base.MultiNetworkBaseTest):
     42 
     43   # Set a 100-ms retrans timer so we can test for ND retransmits without
     44   # waiting too long. Apparently this cannot go below 500ms.
     45   RETRANS_TIME_MS = 500
     46 
     47   # This can only be in seconds, so 1000 is the minimum.
     48   DELAY_TIME_MS = 1000
     49 
     50   # Unfortunately, this must be above the delay timer or the kernel ND code will
     51   # not behave correctly (e.g., go straight from REACHABLE into DELAY). This is
     52   # is fuzzed by the kernel from 0.5x to 1.5x of its value, so we need a value
     53   # that's 2x the delay timer.
     54   REACHABLE_TIME_MS = 2 * DELAY_TIME_MS
     55 
     56   @classmethod
     57   def setUpClass(cls):
     58     super(NeighbourTest, cls).setUpClass()
     59     for netid in cls.tuns:
     60       iface = cls.GetInterfaceName(netid)
     61       # This can't be set in an RA.
     62       cls.SetSysctl(
     63           "/proc/sys/net/ipv6/neigh/%s/delay_first_probe_time" % iface,
     64           cls.DELAY_TIME_MS / 1000)
     65 
     66   def setUp(self):
     67     super(NeighbourTest, self).setUp()
     68 
     69     for netid in self.tuns:
     70       # Clear the ND cache entries for all routers, so each test starts with
     71       # the IPv6 default router in state STALE.
     72       addr = self._RouterAddress(netid, 6)
     73       ifindex = self.ifindices[netid]
     74       self.iproute.UpdateNeighbour(6, addr, None, ifindex, NUD_FAILED)
     75 
     76       # Configure IPv6 by sending an RA.
     77       self.SendRA(netid,
     78                   retranstimer=self.RETRANS_TIME_MS,
     79                   reachabletime=self.REACHABLE_TIME_MS)
     80 
     81     self.sock = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE)
     82     self.sock.bind((0, RTMGRP_NEIGH))
     83     net_test.SetNonBlocking(self.sock)
     84 
     85     self.netid = random.choice(self.tuns.keys())
     86     self.ifindex = self.ifindices[self.netid]
     87 
     88   def GetNeighbour(self, addr):
     89     version = 6 if ":" in addr else 4
     90     for msg, args in self.iproute.DumpNeighbours(version):
     91       if args["NDA_DST"] == addr:
     92         return msg, args
     93 
     94   def GetNdEntry(self, addr):
     95     return self.GetNeighbour(addr)
     96 
     97   def CheckNoNdEvents(self):
     98     self.assertRaisesErrno(errno.EAGAIN, self.sock.recvfrom, 4096, MSG_PEEK)
     99 
    100   def assertNeighbourState(self, state, addr):
    101     self.assertEquals(state, self.GetNdEntry(addr)[0].state)
    102 
    103   def assertNeighbourAttr(self, addr, name, value):
    104     self.assertEquals(value, self.GetNdEntry(addr)[1][name])
    105 
    106   def ExpectNeighbourNotification(self, addr, state, attrs=None):
    107     msg = self.sock.recv(4096)
    108     msg, actual_attrs = self.iproute.ParseNeighbourMessage(msg)
    109     self.assertEquals(addr, actual_attrs["NDA_DST"])
    110     self.assertEquals(state, msg.state)
    111     if attrs:
    112       for name in attrs:
    113         self.assertEquals(attrs[name], actual_attrs[name])
    114 
    115   def ExpectProbe(self, is_unicast, addr):
    116     version = 6 if ":" in addr else 4
    117     if version == 6:
    118       llsrc = self.MyMacAddress(self.netid)
    119       if is_unicast:
    120         src = self.MyLinkLocalAddress(self.netid)
    121         dst = addr
    122       else:
    123         solicited = inet_pton(AF_INET6, addr)
    124         last3bytes = tuple([ord(b) for b in solicited[-3:]])
    125         dst = "ff02::1:ff%02x:%02x%02x" % last3bytes
    126         src = self.MyAddress(6, self.netid)
    127       expected = (
    128           scapy.IPv6(src=src, dst=dst) /
    129           scapy.ICMPv6ND_NS(tgt=addr) /
    130           scapy.ICMPv6NDOptSrcLLAddr(lladdr=llsrc)
    131       )
    132       msg = "%s probe" % ("Unicast" if is_unicast else "Multicast")
    133       self.ExpectPacketOn(self.netid, msg, expected)
    134     else:
    135       raise NotImplementedError
    136 
    137   def ExpectUnicastProbe(self, addr):
    138     self.ExpectProbe(True, addr)
    139 
    140   def ExpectMulticastNS(self, addr):
    141     self.ExpectProbe(False, addr)
    142 
    143   def ReceiveUnicastAdvertisement(self, addr, mac, srcaddr=None, dstaddr=None,
    144                                   S=1, O=0, R=1):
    145     version = 6 if ":" in addr else 4
    146     if srcaddr is None:
    147       srcaddr = addr
    148     if dstaddr is None:
    149       dstaddr = self.MyLinkLocalAddress(self.netid)
    150     if version == 6:
    151       packet = (
    152           scapy.Ether(src=mac, dst=self.MyMacAddress(self.netid)) /
    153           scapy.IPv6(src=srcaddr, dst=dstaddr) /
    154           scapy.ICMPv6ND_NA(tgt=addr, S=S, O=O, R=R) /
    155           scapy.ICMPv6NDOptDstLLAddr(lladdr=mac)
    156       )
    157       self.ReceiveEtherPacketOn(self.netid, packet)
    158     else:
    159       raise NotImplementedError
    160 
    161   def MonitorSleepMs(self, interval, addr):
    162     slept = 0
    163     while slept < interval:
    164       sleep_ms = min(100, interval - slept)
    165       time.sleep(sleep_ms / 1000.0)
    166       slept += sleep_ms
    167       print self.GetNdEntry(addr)
    168 
    169   def MonitorSleep(self, intervalseconds, addr):
    170     self.MonitorSleepMs(intervalseconds * 1000, addr)
    171 
    172   def SleepMs(self, ms):
    173     time.sleep(ms / 1000.0)
    174 
    175   def testNotifications(self):
    176     """Tests neighbour notifications.
    177 
    178     Relevant kernel commits:
    179       upstream net-next:
    180         765c9c6 neigh: Better handling of transition to NUD_PROBE state
    181         53385d2 neigh: Netlink notification for administrative NUD state change
    182           (only checked on kernel v3.13+, not on v3.10)
    183 
    184       android-3.10:
    185         e4a6d6b neigh: Better handling of transition to NUD_PROBE state
    186 
    187       android-3.18:
    188         2011e72 neigh: Better handling of transition to NUD_PROBE state
    189     """
    190 
    191     router4 = self._RouterAddress(self.netid, 4)
    192     router6 = self._RouterAddress(self.netid, 6)
    193     self.assertNeighbourState(NUD_PERMANENT, router4)
    194     self.assertNeighbourState(NUD_STALE, router6)
    195 
    196     # Send a packet and check that we go into DELAY.
    197     routing_mode = random.choice(["mark", "oif", "uid"])
    198     s = self.BuildSocket(6, net_test.UDPSocket, self.netid, routing_mode)
    199     s.connect((net_test.IPV6_ADDR, 53))
    200     s.send(net_test.UDP_PAYLOAD)
    201     self.assertNeighbourState(NUD_DELAY, router6)
    202 
    203     # Wait for the probe interval, then check that we're in PROBE, and that the
    204     # kernel has notified us.
    205     self.SleepMs(self.DELAY_TIME_MS)
    206     self.ExpectNeighbourNotification(router6, NUD_PROBE)
    207     self.assertNeighbourState(NUD_PROBE, router6)
    208     self.ExpectUnicastProbe(router6)
    209 
    210     # Respond to the NS and verify we're in REACHABLE again.
    211     self.ReceiveUnicastAdvertisement(router6, self.RouterMacAddress(self.netid))
    212     self.assertNeighbourState(NUD_REACHABLE, router6)
    213     if net_test.LINUX_VERSION >= (3, 13, 0):
    214       # commit 53385d2 (v3.13) "neigh: Netlink notification for administrative
    215       # NUD state change" produces notifications for NUD_REACHABLE, but these
    216       # are not generated on earlier kernels.
    217       self.ExpectNeighbourNotification(router6, NUD_REACHABLE)
    218 
    219     # Wait until the reachable time has passed, and verify we're in STALE.
    220     self.SleepMs(self.REACHABLE_TIME_MS * 1.5)
    221     self.assertNeighbourState(NUD_STALE, router6)
    222     self.ExpectNeighbourNotification(router6, NUD_STALE)
    223 
    224     # Send a packet, and verify we go into DELAY and then to PROBE.
    225     s.send(net_test.UDP_PAYLOAD)
    226     self.assertNeighbourState(NUD_DELAY, router6)
    227     self.SleepMs(self.DELAY_TIME_MS)
    228     self.assertNeighbourState(NUD_PROBE, router6)
    229     self.ExpectNeighbourNotification(router6, NUD_PROBE)
    230 
    231     # Wait for the probes to time out, and expect a FAILED notification.
    232     self.assertNeighbourAttr(router6, "NDA_PROBES", 1)
    233     self.ExpectUnicastProbe(router6)
    234 
    235     self.SleepMs(self.RETRANS_TIME_MS)
    236     self.ExpectUnicastProbe(router6)
    237     self.assertNeighbourAttr(router6, "NDA_PROBES", 2)
    238 
    239     self.SleepMs(self.RETRANS_TIME_MS)
    240     self.ExpectUnicastProbe(router6)
    241     self.assertNeighbourAttr(router6, "NDA_PROBES", 3)
    242 
    243     self.SleepMs(self.RETRANS_TIME_MS)
    244     self.assertNeighbourState(NUD_FAILED, router6)
    245     self.ExpectNeighbourNotification(router6, NUD_FAILED, {"NDA_PROBES": 3})
    246 
    247   def testRepeatedProbes(self):
    248     router4 = self._RouterAddress(self.netid, 4)
    249     router6 = self._RouterAddress(self.netid, 6)
    250     routermac = self.RouterMacAddress(self.netid)
    251     self.assertNeighbourState(NUD_PERMANENT, router4)
    252     self.assertNeighbourState(NUD_STALE, router6)
    253 
    254     def ForceProbe(addr, mac):
    255       self.iproute.UpdateNeighbour(6, addr, None, self.ifindex, NUD_PROBE)
    256       self.assertNeighbourState(NUD_PROBE, addr)
    257       self.SleepMs(1)  # TODO: Why is this necessary?
    258       self.assertNeighbourState(NUD_PROBE, addr)
    259       self.ExpectUnicastProbe(addr)
    260       self.ReceiveUnicastAdvertisement(addr, mac)
    261       self.assertNeighbourState(NUD_REACHABLE, addr)
    262 
    263     for _ in xrange(5):
    264       ForceProbe(router6, routermac)
    265 
    266   def testIsRouterFlag(self):
    267     router6 = self._RouterAddress(self.netid, 6)
    268     self.assertNeighbourState(NUD_STALE, router6)
    269 
    270     # Get into FAILED.
    271     ifindex = self.ifindices[self.netid]
    272     self.iproute.UpdateNeighbour(6, router6, None, ifindex, NUD_FAILED)
    273     self.ExpectNeighbourNotification(router6, NUD_FAILED)
    274     self.assertNeighbourState(NUD_FAILED, router6)
    275 
    276     time.sleep(1)
    277 
    278     # Send another packet and expect a multicast NS.
    279     routing_mode = random.choice(["mark", "oif", "uid"])
    280     s = self.BuildSocket(6, net_test.UDPSocket, self.netid, routing_mode)
    281     s.connect((net_test.IPV6_ADDR, 53))
    282     s.send(net_test.UDP_PAYLOAD)
    283     self.ExpectMulticastNS(router6)
    284 
    285     # Receive a unicast NA with the R flag set to 0.
    286     self.ReceiveUnicastAdvertisement(router6, self.RouterMacAddress(self.netid),
    287                                      srcaddr=self._RouterAddress(self.netid, 6),
    288                                      dstaddr=self.MyAddress(6, self.netid),
    289                                      S=1, O=0, R=0)
    290 
    291     # Expect that this takes us to REACHABLE.
    292     self.ExpectNeighbourNotification(router6, NUD_REACHABLE)
    293     self.assertNeighbourState(NUD_REACHABLE, router6)
    294 
    295 
    296 if __name__ == "__main__":
    297   unittest.main()
    298