resources/traffic_scripts/ipv6_nd_proxy_check.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2016 Cisco and/or its affiliates.
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 """Traffic script that sends DHCPv6 proxy packets."""
  16
  17 from scapy.layers.inet import Ether
  18 from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS
  19 from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply
  20
  21
  22 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  23 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  24
  25
  26 def imcpv6nd_solicit(tx_if, src_mac, dst_mac, src_ip, dst_ip):
  27     """Send ICMPv6 Neighbor Solicitation packet and expect a response
  28      from the proxy.
  29
  30     :param tx_if: Interface on TG.
  31     :param src_mac: MAC address of TG interface.
  32     :param dst_mac: MAC address of proxy interface.
  33     :param src_ip: IP address of TG interface.
  34     :param dst_ip: IP address of proxied interface.
  35     :type tx_if: str
  36     :type src_mac: str
  37     :type dst_mac: str
  38     :type src_ip: str
  39     :type dst_ip: str
  40     :raises RuntimeError: If the received packet is not correct.
  41     """
  42
  43     rxq = RxQueue(tx_if)
  44     txq = TxQueue(tx_if)
  45
  46     sent_packets = []
  47
  48     icmpv6nd_solicit_pkt = (Ether(src=src_mac, dst=dst_mac) /
  49                             IPv6(src=src_ip) /
  50                             ICMPv6ND_NS(tgt=dst_ip))
  51
  52     sent_packets.append(icmpv6nd_solicit_pkt)
  53     txq.send(icmpv6nd_solicit_pkt)
  54
  55     ether = None
  56     for _ in range(5):
  57         ether = rxq.recv(3, ignore=sent_packets)
  58         if not ether:
  59             continue
  60         if ether.haslayer(ICMPv6ND_NS):
  61             # read another packet in the queue in case of ICMPv6ND_NS packet
  62             continue
  63         else:
  64             # otherwise process the current packet
  65             break
  66
  67     if ether is None:
  68         raise RuntimeError('ICMPv6ND Proxy response timeout.')
  69
  70     if ether.src != dst_mac:
  71         raise RuntimeError("Source MAC address error: {} != {}".
  72                            format(ether.src, dst_mac))
  73     print "Source MAC address: OK."
  74
  75     if ether.dst != src_mac:
  76         raise RuntimeError("Destination MAC address error: {} != {}".
  77                            format(ether.dst, src_mac))
  78     print "Destination MAC address: OK."
  79
  80     if ether[IPv6].src != dst_ip:
  81         raise RuntimeError("Source IP address error: {} != {}".
  82                            format(ether[IPv6].src, dst_ip))
  83     print "Source IP address: OK."
  84
  85     if ether[IPv6].dst != src_ip:
  86         raise RuntimeError("Destination IP address error: {} != {}".
  87                            format(ether[IPv6].dst, src_ip))
  88     print "Destination IP address: OK."
  89
  90     try:
  91         target_addr = ether[IPv6][ICMPv6ND_NA].tgt
  92     except (KeyError, AttributeError):
  93         raise RuntimeError("Not an ICMPv6ND Neighbor Advertisement packet.")
  94
  95     if target_addr != dst_ip:
  96         raise RuntimeError("ICMPv6 field 'Target address' error: {} != {}".
  97                            format(target_addr, dst_ip))
  98     print "Target address field: OK."
  99
 100
 101 def ipv6_ping(src_if, dst_if, src_mac, dst_mac,
 102               proxy_to_src_mac, proxy_to_dst_mac, src_ip, dst_ip):
 103     """Sends ICMPv6 Echo Request, receive it and send a reply.
 104
 105     :param src_if: First TG interface on link to DUT.
 106     :param dst_if: Second TG interface on link to DUT.
 107     :param src_mac: MAC address of first interface.
 108     :param dst_mac: MAC address of second interface.
 109     :param proxy_to_src_mac: MAC address of first proxy interface on DUT.
 110     :param proxy_to_dst_mac: MAC address of second proxy interface on DUT.
 111     :param src_ip: IP address of first interface.
 112     :param dst_ip: IP address of second interface.
 113     :type src_if: str
 114     :type dst_if: str
 115     :type src_mac: str
 116     :type dst_mac: str
 117     :type proxy_to_src_mac: str
 118     :type proxy_to_dst_mac: str
 119     :type src_ip: str
 120     :type dst_ip: str
 121     :raises RuntimeError: If a received packet is not correct.
 122     """
 123     rxq = RxQueue(dst_if)
 124     txq = TxQueue(src_if)
 125
 126     icmpv6_ping_pkt = (Ether(src=src_mac, dst=proxy_to_src_mac) /
 127                        IPv6(src=src_ip, dst=dst_ip) /
 128                        ICMPv6EchoRequest())
 129
 130     txq.send(icmpv6_ping_pkt)
 131
 132     ether = None
 133     while True:
 134         ether = rxq.recv(3)
 135         if not ether:
 136             continue
 137         if ether.haslayer(ICMPv6ND_NS):
 138             # read another packet in the queue in case of ICMPv6ND_NS packet
 139             continue
 140         else:
 141             # otherwise process the current packet
 142             break
 143
 144     if ether is None:
 145         raise RuntimeError('ICMPv6 Echo Request timeout.')
 146     try:
 147         ether[IPv6]["ICMPv6 Echo Request"]
 148     except KeyError:
 149         raise RuntimeError("Received packet is not an ICMPv6 Echo Request.")
 150     print "ICMP Echo: OK."
 151
 152     rxq = RxQueue(src_if)
 153     txq = TxQueue(dst_if)
 154
 155     icmpv6_ping_pkt = (Ether(src=dst_mac, dst=proxy_to_dst_mac) /
 156                        IPv6(src=dst_ip, dst=src_ip) /
 157                        ICMPv6EchoReply())
 158
 159     txq.send(icmpv6_ping_pkt)
 160
 161     ether = None
 162     while True:
 163         ether = rxq.recv(3)
 164         if not ether:
 165             continue
 166         if ether.haslayer(ICMPv6ND_NS):
 167             # read another packet in the queue in case of ICMPv6ND_NS packet
 168             continue
 169         else:
 170             # otherwise process the current packet
 171             break
 172
 173     if ether is None:
 174         raise RuntimeError('DHCPv6 SOLICIT timeout')
 175     try:
 176         ether[IPv6]["ICMPv6 Echo Reply"]
 177     except KeyError:
 178         raise RuntimeError("Received packet is not an ICMPv6 Echo Reply.")
 179
 180     print "ICMP Reply: OK."
 181
 182
 183 def main():
 184     """Send DHCPv6 proxy messages."""
 185
 186     args = TrafficScriptArg(['src_ip', 'dst_ip', 'src_mac', 'dst_mac',
 187                              'proxy_to_src_mac', 'proxy_to_dst_mac'])
 188
 189     src_if = args.get_arg('tx_if')
 190     dst_if = args.get_arg('rx_if')
 191     src_ip = args.get_arg("src_ip")
 192     dst_ip = args.get_arg("dst_ip")
 193     src_mac = args.get_arg('src_mac')
 194     dst_mac = args.get_arg('dst_mac')
 195     proxy_to_src_mac = args.get_arg('proxy_to_src_mac')
 196     proxy_to_dst_mac = args.get_arg('proxy_to_dst_mac')
 197
 198     # Neighbor solicitation
 199     imcpv6nd_solicit(src_if, src_mac, proxy_to_src_mac, src_ip, dst_ip)
 200
 201     # Verify route (ICMP echo/reply)
 202     ipv6_ping(src_if, dst_if, src_mac, dst_mac, proxy_to_src_mac,
 203               proxy_to_dst_mac, src_ip, dst_ip)
 204
 205
 206 if __name__ == "__main__":
 207     main()