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     checksum_equal
  23 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  24 from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS
  25 from scapy.layers.inet6 import ICMPv6NDOptDstLLAddr, ICMPv6NDOptSrcLLAddr
  26 from scapy.all import Ether
  27
  28
  29 def main():
  30     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip'])
  31
  32     rxq = RxQueue(args.get_arg('rx_if'))
  33     txq = TxQueue(args.get_arg('tx_if'))
  34
  35     src_mac = args.get_arg('src_mac')
  36     dst_mac = args.get_arg('dst_mac')
  37     src_ip = args.get_arg('src_ip')
  38     dst_ip = args.get_arg('dst_ip')
  39
  40     sent_packets = []
  41
  42     # send ICMPv6 neighbor solicitation message
  43     pkt_send = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') /
  44                       IPv6(src=src_ip, dst='ff02::1:ff00:2') /
  45                       ICMPv6ND_NS(tgt=dst_ip) /
  46                       ICMPv6NDOptSrcLLAddr(lladdr=src_mac))
  47     sent_packets.append(pkt_send)
  48     txq.send(pkt_send)
  49
  50     # receive ICMPv6 neighbor advertisement message
  51     ether = rxq.recv(2, sent_packets)
  52     if ether is None:
  53         raise RuntimeError('ICMPv6 echo reply Rx timeout')
  54
  55     if not ether.haslayer(IPv6):
  56         raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format(
  57             ether.__repr__()))
  58
  59     ipv6 = ether['IPv6']
  60
  61     if not ipv6.haslayer(ICMPv6ND_NA):
  62         raise RuntimeError(
  63             'Unexpected packet with no ICMPv6 ND-NA received {0}'.format(
  64                 ipv6.__repr__()))
  65
  66     icmpv6_na = ipv6['ICMPv6 Neighbor Discovery - Neighbor Advertisement']
  67
  68     # verify target address
  69     if icmpv6_na.tgt != dst_ip:
  70         raise RuntimeError('Invalid target address {0} should be {1}'.format(
  71             icmpv6_na.tgt, dst_ip))
  72
  73     if not icmpv6_na.haslayer(ICMPv6NDOptDstLLAddr):
  74         raise RuntimeError(
  75             'Missing Destination Link-Layer Address option in ICMPv6 ' +
  76             'Neighbor Advertisement {0}'.format(icmpv6_na.__repr__()))
  77
  78     option = 'ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address'
  79     dst_ll_addr = icmpv6_na[option]
  80
  81     # verify destination link-layer address field
  82     if dst_ll_addr.lladdr != dst_mac:
  83         raise RuntimeError('Invalid lladdr {0} should be {1}'.format(
  84             dst_ll_addr.lladdr, dst_mac))
  85
  86     # verify checksum
  87     cksum = icmpv6_na.cksum
  88     del icmpv6_na.cksum
  89     tmp = ICMPv6ND_NA(str(icmpv6_na))
  90     if not checksum_equal(tmp.cksum, cksum):
  91         raise RuntimeError(
  92             'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum))
  93
  94     sys.exit(0)
  95
  96 if __name__ == "__main__":
  97     main()