GPL/traffic_scripts/nat.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright (c) 2021 Cisco and/or its affiliates.
   4 #
   5 # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
   6 #
   7 # Licensed under the Apache License 2.0 or
   8 # GNU General Public License v2.0 or later;  you may not use this file
   9 # except in compliance with one of these Licenses. You
  10 # may obtain a copy of the Licenses at:
  11 #
  12 #     http://www.apache.org/licenses/LICENSE-2.0
  13 #     https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
  14 #
  15 # Note: If this file is linked with Scapy, which is GPLv2+, your use of it
  16 # must be under GPLv2+.  If at any point in the future it is no longer linked
  17 # with Scapy (or other GPLv2+ licensed software), you are free to choose
  18 # Apache 2.
  19 #
  20 # Unless required by applicable law or agreed to in writing, software
  21 # distributed under the License is distributed on an "AS IS" BASIS,
  22 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  23 # See the License for the specific language governing permissions and
  24 # limitations under the License.
  25
  26 """Traffic script for NAT verification."""
  27
  28 import sys
  29
  30 from scapy.layers.inet import IP, TCP, UDP
  31 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6MLReport2, ICMPv6ND_RA
  32 from scapy.layers.l2 import Ether
  33 from scapy.packet import Raw
  34
  35 from .PacketVerifier import RxQueue, TxQueue
  36 from .TrafficScriptArg import TrafficScriptArg
  37 from .ValidIp import valid_ipv4, valid_ipv6
  38
  39
  40 def main():
  41     """Send, receive and check IP/IPv6 packets with UDP/TCP layer passing
  42     through NAT.
  43     """
  44     args = TrafficScriptArg(
  45         [
  46             u"tx_src_mac", u"rx_dst_mac", u"src_ip_in", u"src_ip_out",
  47             u"dst_ip", u"tx_dst_mac", u"rx_src_mac", u"protocol",
  48             u"src_port_in", u"src_port_out", u"dst_port"
  49         ]
  50     )
  51
  52     tx_src_mac = args.get_arg(u"tx_src_mac")
  53     tx_dst_mac = args.get_arg(u"tx_dst_mac")
  54     rx_dst_mac = args.get_arg(u"rx_dst_mac")
  55     rx_src_mac = args.get_arg(u"rx_src_mac")
  56     src_ip_in = args.get_arg(u"src_ip_in")
  57     src_ip_out = args.get_arg(u"src_ip_out")
  58     dst_ip = args.get_arg(u"dst_ip")
  59     protocol = args.get_arg(u"protocol")
  60     sport_in = int(args.get_arg(u"src_port_in"))
  61     try:
  62         sport_out = int(args.get_arg(u"src_port_out"))
  63     except ValueError:
  64         sport_out = None
  65     dst_port = int(args.get_arg(u"dst_port"))
  66
  67     tx_txq = TxQueue(args.get_arg(u"tx_if"))
  68     tx_rxq = RxQueue(args.get_arg(u"tx_if"))
  69     rx_txq = TxQueue(args.get_arg(u"rx_if"))
  70     rx_rxq = RxQueue(args.get_arg(u"rx_if"))
  71
  72     sent_packets = list()
  73     pkt_raw = Ether(src=tx_src_mac, dst=tx_dst_mac)
  74
  75     if valid_ipv4(src_ip_in) and valid_ipv4(dst_ip):
  76         ip_layer = IP
  77     elif valid_ipv6(src_ip_in) and valid_ipv6(dst_ip):
  78         ip_layer = IPv6
  79     else:
  80         raise ValueError(u"IP not in correct format")
  81     pkt_raw /= ip_layer(src=src_ip_in, dst=dst_ip)
  82
  83     if protocol == u"UDP":
  84         pkt_raw /= UDP(sport=sport_in, dport=dst_port)
  85         proto_layer = UDP
  86     elif protocol == u"TCP":
  87         # flags=0x2 => SYN flag set
  88         pkt_raw /= TCP(sport=sport_in, dport=dst_port, flags=0x2)
  89         proto_layer = TCP
  90     else:
  91         raise ValueError(u"Incorrect protocol")
  92
  93     pkt_raw /= Raw()
  94     sent_packets.append(pkt_raw)
  95     tx_txq.send(pkt_raw)
  96
  97     while True:
  98         ether = rx_rxq.recv(2)
  99
 100         if ether is None:
 101             raise RuntimeError(u"IP packet Rx timeout")
 102
 103         if ether.haslayer(ICMPv6ND_NS):
 104             # read another packet in the queue if the current one is ICMPv6ND_NS
 105             continue
 106         elif ether.haslayer(ICMPv6MLReport2):
 107             # read another packet in the queue if the current one is
 108             # ICMPv6MLReport2
 109             continue
 110         elif ether.haslayer(ICMPv6ND_RA):
 111             # read another packet in the queue if the current one is
 112             # ICMPv6ND_RA
 113             continue
 114
 115         break
 116
 117     if rx_dst_mac != ether[Ether].dst or rx_src_mac != ether[Ether].src:
 118         raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
 119
 120     ip_pkt = ether.payload
 121     if not isinstance(ip_pkt, ip_layer):
 122         raise RuntimeError(f"Not an {ip_layer!s} packet received: {ip_pkt!r}")
 123     if ip_pkt.src != src_ip_out:
 124         raise RuntimeError(
 125             f"Matching Src IP address unsuccessful: "
 126             f"{src_ip_out} != {ip_pkt.src}"
 127         )
 128     if ip_pkt.dst != dst_ip:
 129         raise RuntimeError(
 130             f"Matching Dst IP address unsuccessful: {dst_ip} != {ip_pkt.dst}"
 131         )
 132
 133     proto_pkt = ip_pkt.payload
 134     if not isinstance(proto_pkt, proto_layer):
 135         raise RuntimeError(
 136             f"Not a {proto_layer!s} packet received: {proto_pkt!r}"
 137         )
 138     if sport_out is not None:
 139         if proto_pkt.sport != sport_out:
 140             raise RuntimeError(
 141                 f"Matching Src {proto_layer!s} port unsuccessful: "
 142                 f"{sport_out} != {proto_pkt.sport}"
 143             )
 144     else:
 145         sport_out = proto_pkt.sport
 146     if proto_pkt.dport != dst_port:
 147         raise RuntimeError(
 148             f"Matching Dst {proto_layer!s} port unsuccessful: "
 149             f"{dst_port} != {proto_pkt.dport}"
 150         )
 151     if proto_layer == TCP:
 152         if proto_pkt.flags != 0x2:
 153             raise RuntimeError(
 154                 f"Not a TCP SYN packet received: {proto_pkt!r}"
 155             )
 156
 157     pkt_raw = Ether(src=rx_dst_mac, dst=rx_src_mac)
 158     pkt_raw /= ip_layer(src=dst_ip, dst=src_ip_out)
 159     pkt_raw /= proto_layer(sport=dst_port, dport=sport_out)
 160     if proto_layer == TCP:
 161         # flags=0x12 => SYN, ACK flags set
 162         pkt_raw[TCP].flags = 0x12
 163     pkt_raw /= Raw()
 164     rx_txq.send(pkt_raw)
 165
 166     while True:
 167         ether = tx_rxq.recv(2, ignore=sent_packets)
 168
 169         if ether is None:
 170             raise RuntimeError(u"IP packet Rx timeout")
 171
 172         if ether.haslayer(ICMPv6ND_NS):
 173             # read another packet in the queue if the current one is ICMPv6ND_NS
 174             continue
 175         elif ether.haslayer(ICMPv6MLReport2):
 176             # read another packet in the queue if the current one is
 177             # ICMPv6MLReport2
 178             continue
 179         elif ether.haslayer(ICMPv6ND_RA):
 180             # read another packet in the queue if the current one is
 181             # ICMPv6ND_RA
 182             continue
 183
 184         break
 185
 186     if ether[Ether].dst != tx_src_mac or ether[Ether].src != tx_dst_mac:
 187         raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
 188
 189     ip_pkt = ether.payload
 190     if not isinstance(ip_pkt, ip_layer):
 191         raise RuntimeError(f"Not an {ip_layer!s} packet received: {ip_pkt!r}")
 192     if ip_pkt.src != dst_ip:
 193         raise RuntimeError(
 194             f"Matching Src IP address unsuccessful: {dst_ip} != {ip_pkt.src}"
 195         )
 196     if ip_pkt.dst != src_ip_in:
 197         raise RuntimeError(
 198             f"Matching Dst IP address unsuccessful: {src_ip_in} != {ip_pkt.dst}"
 199         )
 200
 201     proto_pkt = ip_pkt.payload
 202     if not isinstance(proto_pkt, proto_layer):
 203         raise RuntimeError(
 204             f"Not a {proto_layer!s} packet received: {proto_pkt!r}"
 205         )
 206     if proto_pkt.sport != dst_port:
 207         raise RuntimeError(
 208             f"Matching Src {proto_layer!s} port unsuccessful: "
 209             f"{dst_port} != {proto_pkt.sport}"
 210         )
 211     if proto_pkt.dport != sport_in:
 212         raise RuntimeError(
 213             f"Matching Dst {proto_layer!s} port unsuccessful: "
 214             f"{sport_in} != {proto_pkt.dport}"
 215         )
 216     if proto_layer == TCP:
 217         if proto_pkt.flags != 0x12:
 218             raise RuntimeError(
 219                 f"Not a TCP SYN-ACK packet received: {proto_pkt!r}"
 220             )
 221
 222     sys.exit(0)
 223
 224
 225 if __name__ == u"__main__":
 226     main()