resources/traffic_scripts/ipv6_ns.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Cisco and/or its affiliates.
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 """Traffic script for IPv6 Neighbor Solicitation test."""
  17
  18 import sys
  19 import logging
  20 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
  21 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  22 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  23 from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS
  24 from scapy.layers.inet6 import ICMPv6NDOptDstLLAddr, ICMPv6NDOptSrcLLAddr
  25 from scapy.all import Ether
  26
  27
  28 def main():
  29     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip'])
  30
  31     rxq = RxQueue(args.get_arg('rx_if'))
  32     txq = TxQueue(args.get_arg('tx_if'))
  33
  34     src_mac = args.get_arg('src_mac')
  35     dst_mac = args.get_arg('dst_mac')
  36     src_ip = args.get_arg('src_ip')
  37     dst_ip = args.get_arg('dst_ip')
  38
  39     sent_packets = []
  40
  41     # send ICMPv6 neighbor solicitation message
  42     pkt_send = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') /
  43                       IPv6(src=src_ip, dst='ff02::1:ff00:2') /
  44                       ICMPv6ND_NS(tgt=dst_ip) /
  45                       ICMPv6NDOptSrcLLAddr(lladdr=src_mac))
  46     sent_packets.append(pkt_send)
  47     txq.send(pkt_send)
  48
  49     # receive ICMPv6 neighbor advertisement message
  50     ether = rxq.recv(2, sent_packets)
  51     if ether is None:
  52         raise RuntimeError('ICMPv6 echo reply Rx timeout')
  53
  54     if not ether.haslayer(IPv6):
  55         raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format(
  56             ether.__repr__()))
  57
  58     ipv6 = ether['IPv6']
  59
  60     if not ipv6.haslayer(ICMPv6ND_NA):
  61         raise RuntimeError(
  62             'Unexpected packet with no ICMPv6 ND-NA received {0}'.format(
  63                 ipv6.__repr__()))
  64
  65     icmpv6_na = ipv6['ICMPv6 Neighbor Discovery - Neighbor Advertisement']
  66
  67     # verify target address
  68     if icmpv6_na.tgt != dst_ip:
  69         raise RuntimeError('Invalid target address {0} should be {1}'.format(
  70             icmpv6_na.tgt, dst_ip))
  71
  72     if not icmpv6_na.haslayer(ICMPv6NDOptDstLLAddr):
  73         raise RuntimeError(
  74             'Missing Destination Link-Layer Address option in ICMPv6 ' +
  75             'Neighbor Advertisement {0}'.format(icmpv6_na.__repr__()))
  76
  77     option = 'ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address'
  78     dst_ll_addr = icmpv6_na[option]
  79
  80     # verify destination link-layer address field
  81     if dst_ll_addr.lladdr != dst_mac:
  82         raise RuntimeError('Invalid lladdr {0} should be {1}'.format(
  83             dst_ll_addr.lladdr, dst_mac))
  84
  85     # verify checksum
  86     cksum = icmpv6_na.cksum
  87     del icmpv6_na.cksum
  88     tmp = ICMPv6ND_NA(str(icmpv6_na))
  89     if tmp.cksum != cksum:
  90         raise RuntimeError(
  91             'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum))
  92
  93     sys.exit(0)
  94
  95 if __name__ == "__main__":
  96     main()