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         rxq._proc.terminate()
  53         raise RuntimeError('ICMPv6 echo reply Rx timeout')
  54
  55     if not ether.haslayer(IPv6):
  56         rxq._proc.terminate()
  57         raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format(
  58             ether.__repr__()))
  59
  60     ipv6 = ether['IPv6']
  61
  62     if not ipv6.haslayer(ICMPv6ND_NA):
  63         rxq._proc.terminate()
  64         raise RuntimeError(
  65             'Unexpected packet with no ICMPv6 ND-NA received {0}'.format(
  66                 ipv6.__repr__()))
  67
  68     icmpv6_na = ipv6['ICMPv6 Neighbor Discovery - Neighbor Advertisement']
  69
  70     # verify target address
  71     if icmpv6_na.tgt != dst_ip:
  72         rxq._proc.terminate()
  73         raise RuntimeError('Invalid target address {0} should be {1}'.format(
  74             icmpv6_na.tgt, dst_ip))
  75
  76     if not icmpv6_na.haslayer(ICMPv6NDOptDstLLAddr):
  77         rxq._proc.terminate()
  78         raise RuntimeError(
  79             'Missing Destination Link-Layer Address option in ICMPv6 ' +
  80             'Neighbor Advertisement {0}'.format(icmpv6_na.__repr__()))
  81
  82     option = 'ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address'
  83     dst_ll_addr = icmpv6_na[option]
  84
  85     # verify destination link-layer address field
  86     if dst_ll_addr.lladdr != dst_mac:
  87         rxq._proc.terminate()
  88         raise RuntimeError('Invalid lladdr {0} should be {1}'.format(
  89             dst_ll_addr.lladdr, dst_mac))
  90
  91     # verify checksum
  92     cksum = icmpv6_na.cksum
  93     del icmpv6_na.cksum
  94     tmp = ICMPv6ND_NA(str(icmpv6_na))
  95     if tmp.cksum != cksum:
  96         rxq._proc.terminate()
  97         raise RuntimeError(
  98             'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum))
  99
 100     rxq._proc.terminate()
 101     sys.exit(0)
 102
 103 if __name__ == "__main__":
 104     main()