Add --idle-skip option: input-based idle detection for MP4 export

Detect idle periods by absence of FAST_PATH_INPUT PDUs (keyboard/mouse)
instead of screen changes. Cursor blinks and screen refreshes no longer
prevent idle detection. During idle, PTS is frozen with one frame
encoded every 10s for state capture. True PDU gaps are also compressed.

Author: jmvermeulen

pyrdp/bin/convert.py

@@ -57,6 +57,14 @@ def main():
         "otherwise the result is output next to the source file with the proper extension. "
         "However if the source of the conversion is a .pcap then this option will create a directory where all files will be stored.",
     )
+    parser.add_argument(
+        "--idle-skip",
+        type=int,
+        default=0,
+        metavar="N",
+        help="Skip idle periods longer than N seconds in MP4 output (0 = disabled, default: 0)",
+    )
+
 
     args = parser.parse_args()
 
@@ -83,15 +91,19 @@ def main():
     else:
         outputPrefix = ""
 
+    handler_kwargs = {}
+    if args.idle_skip > 0:
+        handler_kwargs['idle_skip'] = args.idle_skip
+
     if inputFile.suffix in [".pcap"]:
         secrets = loadSecrets(args.secrets) if args.secrets else None
-        converter = PCAPConverter(inputFile, outputPrefix, args.format, secrets=secrets, srcFilter=args.src, dstFilter=args.dst, listOnly=args.list_only)
+        converter = PCAPConverter(inputFile, outputPrefix, args.format, secrets=secrets, srcFilter=args.src, dstFilter=args.dst, listOnly=args.list_only, handler_kwargs=handler_kwargs)
     elif inputFile.suffix in [".pyrdp"]:
         if args.format == "replay":
             sys.stderr.write("Refusing to convert a replay file to a replay file. Choose another format.")
             sys.exit(1)
 
-        converter = ReplayConverter(inputFile, outputPrefix, args.format)
+        converter = ReplayConverter(inputFile, outputPrefix, args.format, handler_kwargs=handler_kwargs)
     else:
         sys.stderr.write("Unknown file extension. (Supported: .pcap, .pyrdp)")
         sys.exit(1)

pyrdp/convert/Converter.py

@@ -7,10 +7,11 @@
 
 
 class Converter:
-    def __init__(self, inputFile: Path, outputPrefix: str, format: str):
+    def __init__(self, inputFile: Path, outputPrefix: str, format: str, handler_kwargs: dict = None):
         self.inputFile = inputFile
         self.outputPrefix = outputPrefix
         self.format = format
+        self.handler_kwargs = handler_kwargs or {}
 
     def process(self):
         raise NotImplementedError("Converter.process is not implemented")

pyrdp/convert/MP4EventHandler.py

@@ -4,7 +4,7 @@
 # Licensed under the GPLv3 or later.
 #
 
-from pyrdp.enum import CapabilityType
+from pyrdp.enum import CapabilityType, PlayerPDUType
 from pyrdp.pdu import PlayerPDU
 from pyrdp.player.ImageHandler import ImageHandler
 from pyrdp.player.RenderingEventHandler import RenderingEventHandler
@@ -38,13 +38,14 @@ def screen(self) -> QImage:
 
 class MP4EventHandler(RenderingEventHandler):
 
-    def __init__(self, filename: str, fps=10, progress=None):
+    def __init__(self, filename: str, fps=10, progress=None, idle_skip=0):
         """
         Construct an event handler that outputs to an Mp4 file.
 
         :param filename: The output file to write to.
         :param fps: The frame rate (10 recommended for forensic captures).
         :param progress: An optional callback (sig: `() -> ()`) whenever a frame is muxed.
+        :param idle_skip: Seconds of inactivity before compressing idle gaps (0 = disabled).
         """
         self.filename = filename
         # faststart moves the moov atom to the front for seekable playback.
@@ -66,6 +67,13 @@ def __init__(self, filename: str, fps=10, progress=None):
         self.pts = 0
         # Track whether the surface has changed since the last encoded frame
         self.dirty = False
+        # Idle skip: based on user input (keyboard/mouse), not screen changes
+        self.idle_skip_ms = idle_skip * 1000 if idle_skip > 0 else 0
+        self.total_skipped_ms = 0
+        self.lastInputTimestamp = None
+        self._in_idle = False
+        self._idle_enter_ts = None
+        self._last_idle_frame_ts = None
 
         super().__init__(MP4Image())
 
@@ -78,16 +86,56 @@ def onPDUReceived(self, pdu: PlayerPDU):
 
         ts = pdu.timestamp
         self.timestamp = ts
+        is_input = pdu.header == PlayerPDUType.FAST_PATH_INPUT
+
+        # Track user input for idle detection
+        if is_input:
+            self.lastInputTimestamp = ts
 
         if self.prevTimestamp is None:
-            # First PDU: encode if surface was rendered
+            # First PDU: assume active at start
+            if self.lastInputTimestamp is None:
+                self.lastInputTimestamp = ts
+            if self.dirty:
+                self.writeFrame()
+                self.dirty = False
+            self.prevTimestamp = ts
+            return
+
+        # Check input-idle state
+        input_idle_ms = ts - self.lastInputTimestamp
+        now_idle = self.idle_skip_ms > 0 and input_idle_ms > self.idle_skip_ms
+
+        if now_idle and not self._in_idle:
+            # Entering idle: encode last dirty frame, then freeze
+            self._in_idle = True
+            self._idle_enter_ts = ts
+            self._last_idle_frame_ts = ts
             if self.dirty:
                 self.writeFrame()
                 self.dirty = False
+
+        if self._in_idle and not now_idle:
+            # Exiting idle (input resumed)
+            self._in_idle = False
+            self.total_skipped_ms += ts - self._idle_enter_ts
+            self.pts += self.fps  # 1s pause in output
+            self.prevTimestamp = ts
+            return
+
+        if self._in_idle:
+            # During idle: encode 1 frame every 10s to capture screen state
+            if self.dirty and (ts - self._last_idle_frame_ts) >= 10000:
+                self.writeFrame()
+                self.dirty = False
+                self._last_idle_frame_ts = ts
+            else:
+                self.dirty = False
             self.prevTimestamp = ts
             return
 
-        dt = self.timestamp - self.prevTimestamp  # ms
+        # Normal (non-idle) processing
+        dt = ts - self.prevTimestamp  # ms
         nframes = (dt // self.delta)
 
         if nframes > 0:
@@ -97,11 +145,23 @@ def onPDUReceived(self, pdu: PlayerPDU):
                 self.writeFrame()
                 self.dirty = False
                 nframes -= 1  # One frame was just encoded
+
+            # True gap (no PDUs at all): compress
+            gap_threshold = int(self.idle_skip_ms / self.delta) if self.idle_skip_ms > 0 else 0
+            if gap_threshold > 0 and nframes > gap_threshold:
+                self.total_skipped_ms += nframes * self.delta
+                nframes = self.fps  # Replace gap with 1s pause
+
             # Skip remaining frames (player holds last frame)
             self.pts += nframes
             self.prevTimestamp = ts
 
     def cleanup(self):
+        # Close out idle state if capture ends while idle
+        if self._in_idle and self._idle_enter_ts and self.timestamp:
+            self.total_skipped_ms += self.timestamp - self._idle_enter_ts
+            self._in_idle = False
+
         # Flush any pending dirty frame
         if self.dirty:
             self.writeFrame()
@@ -116,6 +176,13 @@ def cleanup(self):
             if self.progress:
                 self.progress()
             self.mp4.mux(pkt)
+
+        if self.total_skipped_ms > 0:
+            skipped_s = self.total_skipped_ms / 1000
+            m, s = divmod(int(skipped_s), 60)
+            h, m = divmod(m, 60)
+            self.log.info('Total idle time skipped: %dh %dm %ds (%.1fs)', h, m, s, skipped_s)
+
         self.log.info('Export completed.')
         self.mp4.close()
 

pyrdp/convert/PCAPConverter.py

@@ -24,8 +24,8 @@
 class PCAPConverter(Converter):
     SESSIONID_FORMAT = "{timestamp}_{src}-{dst}"
 
-    def __init__(self, inputFile: Path, outputPrefix: str, format: str, secrets: Dict = None, srcFilter = None, dstFilter = None, listOnly = False):
-        super().__init__(inputFile, outputPrefix, format)
+    def __init__(self, inputFile: Path, outputPrefix: str, format: str, secrets: Dict = None, srcFilter = None, dstFilter = None, listOnly = False, handler_kwargs: dict = None):
+        super().__init__(inputFile, outputPrefix, format, handler_kwargs=handler_kwargs)
         self.secrets = secrets if secrets is not None else {}
         self.srcFilter = srcFilter if srcFilter is not None else srcFilter
         self.dstFilter = dstFilter if dstFilter is not None else dstFilter
@@ -106,7 +106,7 @@ def processStream(self, startTimeStamp: int, stream: PCAPStream):
         })
         sessionID = sessionID.replace(":", "_")
 
-        handler, _ = createHandler(self.format, self.outputPrefix + sessionID)
+        handler, _ = createHandler(self.format, self.outputPrefix + sessionID, **self.handler_kwargs)
         replayer = RDPReplayer(handler, self.outputPrefix, sessionID)
 
         print(f"[*] Processing {stream.client} -> {stream.server}")

pyrdp/convert/ReplayConverter.py

@@ -20,7 +20,7 @@ def process(self):
             print(f"[*] Converting '{self.inputFile}' to {self.format.upper()}")
 
             outputFileBase = self.outputPrefix + self.inputFile.stem
-            handler, outputPath = createHandler(self.format, outputFileBase)
+            handler, outputPath = createHandler(self.format, outputFileBase, **self.handler_kwargs)
 
             if not handler:
                 print("The input file is already a replay file. Nothing to do.")

pyrdp/convert/utils.py

@@ -74,7 +74,7 @@ def extractInetAddressesFromPDUPacket(packet) -> Tuple[InetAddress, InetAddress]
     return (InetAddress(x.src, x.sport), InetAddress(x.dst, x.dport))
 
 
-def createHandler(format: str, outputFileBase: str, progress=None) -> Tuple[str, str]:
+def createHandler(format: str, outputFileBase: str, progress=None, **kwargs) -> Tuple[str, str]:
     """
     Gets the appropriate handler and returns the filename with extension.
     Returns None if the format is replay.
@@ -87,7 +87,9 @@ def createHandler(format: str, outputFileBase: str, progress=None) -> Tuple[str,
 
     HandlerClass, ext = HANDLERS[format]
     outputFileBase += f".{ext}"
-    return HandlerClass(outputFileBase, progress=progress) if HandlerClass else None, outputFileBase
+    if HandlerClass:
+        return HandlerClass(outputFileBase, progress=progress, **kwargs), outputFileBase
+    return None, outputFileBase
 
 
 class ExportedPDU(Packet):