Skip to content

Instantly share code, notes, and snippets.

@xshill xshill/ Secret
Last active Jun 23, 2019

What would you like to do?
Sample script that was used to create a replay file from a pcap using PyRDP and scapy
import logging
import os
from progressbar import progressbar
from scapy.all import *
from pyrdp.enum import ParserMode, PlayerMessageType
from pyrdp.layer import SegmentationLayer, TPKTLayer, X224Layer, MCSLayer, TLSSecurityLayer, SlowPathLayer, FastPathLayer, VirtualChannelLayer, ClipboardLayer
from pyrdp.mcs import MCSRouter
from pyrdp.parser import ClientInfoParser, BasicFastPathParser
from pyrdp.pdu import FormatDataResponsePDU
from pyrdp.recording import FileLayer, Recorder, RecordingFastPathObserver, RecordingSlowPathObserver
def bytesToIP(data: bytes):
return ".".join(str(b) for b in data)
def parseExportedPdu(packet: packet.Raw):
source = packet.load[12 : 16]
source = bytesToIP(source)
destination = packet.load[20 : 24]
destination = bytesToIP(destination)
data = packet.load[60 :]
return source, destination, data
class CustomRecorder(Recorder):
currentTimeStamp: float = None
def getCurrentTimeStamp(self) -> float:
return self.currentTimeStamp
def setTimeStamp(self, timeStamp: float):
self.currentTimeStamp = timeStamp
class RDP(MCSRouter):
def __init__(self, mode: ParserMode, recorder: Recorder):
self.recorder = recorder
self.currentClipboardData = None
self.segmentation = SegmentationLayer()
self.tpkt = TPKTLayer()
self.x224 = X224Layer()
self.mcs = MCSLayer() = TLSSecurityLayer() = SlowPathLayer()
self.fastPath = FastPathLayer(BasicFastPathParser(mode))
self.clipboardSecurity = TLSSecurityLayer()
self.virtualChannelLayer = VirtualChannelLayer()
self.clipboardLayer = ClipboardLayer()
self.segmentation.attachLayer(0, self.fastPath)
self.segmentation.attachLayer(3, self.tpkt) = True = self.onClientInfoReceived, onLicensingDataReceived = self.onLicensingDataReceived)
self.fastPath.addObserver(RecordingFastPathObserver(recorder, PlayerMessageType.FAST_PATH_OUTPUT if mode == ParserMode.CLIENT else PlayerMessageType.FAST_PATH_INPUT))
self.clipboardLayer.createObserver(onPDUReceived = self.onClipboard)
MCSRouter.__init__(self, self.mcs)
def recv(self, data: bytes):
def recvChannel(self, pdu):
if pdu.channelID == 1003:
# I/O channel
elif pdu.channelID == 1004:
# Clipboard channel
def onSendDataRequest(self, pdu):
def onSendDataIndication(self, pdu):
def onClientInfoReceived(self, data):
pdu = ClientInfoParser().parse(data)
recorder.record(pdu, PlayerMessageType.CLIENT_INFO)
def onLicensingDataReceived(self, pdu): = False
def onClipboard(self, pdu):
if isinstance(pdu, FormatDataResponsePDU):
if pdu.requestedFormatData != self.currentClipboardData:
self.recorder.record(pdu, PlayerMessageType.CLIPBOARD_DATA)
self.currentClipboardData = pdu.requestedFormatData
logging.basicConfig(level = logging.CRITICAL)
directory = os.path.dirname(os.path.realpath(__file__))
packets = rdpcap(directory + "capture.pcap")
replayFile = open("replay.pyrdp", "wb")
fileLayer = FileLayer(replayFile)
recorder = CustomRecorder([fileLayer])
client = RDP(ParserMode.CLIENT, recorder)
server = RDP(ParserMode.SERVER, recorder)
rdp = {
CLIENT: client,
SERVER: server
# The last packet of this pcap is incomplete
for packet in progressbar(packets[: -1]):
# The packets start with a Wireshark exported PDU structure
source, destination, data = parseExportedPdu(packet)
rdp[destination].recv(data) =
except NotImplementedError:
recorder.record(None, PlayerMessageType.CONNECTION_CLOSE)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.