tests/net_test/neighbour_test.py

   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()