SteamGuidCapture.py

"""
capture_tool.py

One-click Steam credential capture for RotMG, replacing the manual
Fiddler workflow described here:
https://github.com/jakcodex/muledump/wiki/Steam-Users-Setup-Guide

Under the hood this does exactly what Fiddler does:
  1. Starts a local HTTPS-intercepting proxy (mitmproxy).
  2. Points Windows' per-user proxy setting at it (the same registry
     value Fiddler's "Capture Traffic" toggles).
  3. Uses mitmproxy's own trusted root certificate to decrypt HTTPS,
     same role Fiddler's certificate plays.
  4. Loads the capture addon from this same Python file. The addon
     watches for the /steamworks/getcredentials response and reports
     whatever fields it contains back to this window.

One-time setup:
    pip install -r requirements.txt
    python capture_tool.py
    -> click "Enable Capture" once (this makes mitmproxy generate its
       root certificate on first run)
    -> click "Disable"
    -> click "Trust CA (one-time)"
    You only need to do this once per machine.

Normal usage after that:
    python capture_tool.py
    -> click "Enable Capture"
    -> launch Steam + RotMG, get to the main menu
    -> credentials appear in the window automatically, proxy is
       disabled again for you
"""

import ctypes
import json
import os
import subprocess
import sys
import threading
import time
import tkinter as tk
from tkinter import scrolledtext
from typing import Optional, Tuple

IS_WINDOWS = sys.platform.startswith("win")

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
ADDON_PATH = os.path.abspath(__file__)  # This file also acts as the mitmproxy addon.
RESULT_PATH = os.path.join(SCRIPT_DIR, "captured_credentials.json")

PROXY_HOST = "127.0.0.1"
PROXY_PORT = 8888

MITMPROXY_CA_CERT = os.path.join(os.path.expanduser("~"), ".mitmproxy", "mitmproxy-ca-cert.pem")


# --------------------------------------------------------------------------
# mitmproxy addon (kept in this same file)
# --------------------------------------------------------------------------

try:
    from mitmproxy import http as mitm_http
except ImportError:
    # Keep the GUI importable even before mitmproxy is installed. The capture
    # process will surface the missing dependency when mitmdump is started.
    mitm_http = None

TARGET_PATH_FRAGMENT = "steamworks/getcredentials"


def _flatten_xml(elem, out):
    """Recursively collect tag -> text pairs from an XML element tree."""
    text = (elem.text or "").strip()
    if text:
        out[elem.tag] = text
    for child in elem:
        _flatten_xml(child, out)


if mitm_http is not None:
    import re
    import xml.etree.ElementTree as ET

    class SteamCredentialCapture:
        def __init__(self):
            self.captured = False

        def response(self, flow: mitm_http.HTTPFlow) -> None:
            if self.captured:
                return

            if TARGET_PATH_FRAGMENT not in flow.request.path.lower():
                return

            body = flow.response.get_text(strict=False) or ""
            if not body.strip():
                return

            parsed = {}
            try:
                root = ET.fromstring(body)
                _flatten_xml(root, parsed)
            except ET.ParseError:
                # Not clean XML -- fall back to a generic tag sweep so we
                # still surface something useful.
                for match in re.finditer(
                    r"<(\w+)>(.*?)</\1>",
                    body,
                    re.IGNORECASE | re.DOTALL,
                ):
                    parsed[match.group(1)] = match.group(2).strip()

            result = {
                "url": flow.request.pretty_url,
                "raw_response": body,
                "parsed_fields": parsed,
            }

            with open(RESULT_PATH, "w", encoding="utf-8") as f:
                json.dump(result, f, indent=2)

            self.captured = True
            print("\n" + "=" * 60)
            print("[+] Captured /steamworks/getcredentials response!")
            for k, v in parsed.items():
                print(f"    {k}: {v}")
            print(f"[+] Full details saved to: {RESULT_PATH}")
            print("=" * 60 + "\n")


    addons = [SteamCredentialCapture()]


# --------------------------------------------------------------------------
# Windows proxy + certificate helpers
# --------------------------------------------------------------------------

def set_system_proxy(enable: bool, host: str = PROXY_HOST, port: int = PROXY_PORT):
    """Toggle the per-user Windows HTTP/HTTPS proxy -- the same setting
    Fiddler flips when you hit F12 / "Capture Traffic"."""
    if not IS_WINDOWS:
        print(f"[!] Not on Windows -- would {'enable' if enable else 'disable'} "
              f"system proxy ({host}:{port}) here.")
        return

    import winreg

    key_path = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
    with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_SET_VALUE) as key:
        winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1 if enable else 0)
        if enable:
            winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, f"{host}:{port}")

    # Tell Windows to pick the change up immediately (no logoff needed).
    INTERNET_OPTION_SETTINGS_CHANGED = 39
    INTERNET_OPTION_REFRESH = 37
    set_option = ctypes.windll.Wininet.InternetSetOptionW
    set_option(0, INTERNET_OPTION_SETTINGS_CHANGED, 0, 0)
    set_option(0, INTERNET_OPTION_REFRESH, 0, 0)


def trust_mitmproxy_ca() -> Tuple[bool, str]:
    """Install mitmproxy's root CA into the CURRENT USER trusted root
    store (the '-user' flag means no admin/UAC elevation is needed).

    Returns (success, message) so the GUI can show a specific reason.
    """
    if not IS_WINDOWS:
        msg = f"Not on Windows -- trust {MITMPROXY_CA_CERT} manually in your OS keychain."
        print(f"[!] {msg}")
        return False, msg

    if not os.path.exists(MITMPROXY_CA_CERT):
        msg = ("mitmproxy hasn't generated its certificate yet. Click "
               "'Enable Capture', wait a few seconds, click 'Disable', "
               "then try Trust CA again.")
        print(f"[!] {msg}")
        return False, msg

    try:
        subprocess.run(
            ["certutil", "-user", "-addstore", "Root", MITMPROXY_CA_CERT],
            check=True, capture_output=True, text=True,
        )
        msg = "mitmproxy root certificate trusted for current user."
        print(f"[+] {msg}")
        return True, msg
    except subprocess.CalledProcessError as e:
        msg = f"certutil failed: {(e.stderr or e.stdout or str(e)).strip()}"
        print(f"[!] {msg}")
        return False, msg


# --------------------------------------------------------------------------
# mitmdump process management
# --------------------------------------------------------------------------

class CaptureSession:
    def __init__(self, on_captured, on_error=None):
        self.proc = None
        self.watch_thread = None
        self.log_thread = None
        self.stop_watch = threading.Event()
        self.on_captured = on_captured
        self.on_error = on_error

    @staticmethod
    def _mitmdump_args():
        return [
            "-s", ADDON_PATH,
            "--listen-host", PROXY_HOST,
            "--listen-port", str(PROXY_PORT),
            "--set", "termlog_verbosity=warn",
        ]

    def start(self) -> Optional[str]:
        """Start mitmdump and enable the system proxy.

        Returns an error message if mitmdump failed to stay running.
        """
        if os.path.exists(RESULT_PATH):
            os.remove(RESULT_PATH)

        args = self._mitmdump_args()
        try:
            self.proc = subprocess.Popen(
                ["mitmdump"] + args,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
            )
        except FileNotFoundError:
            # mitmdump isn't on PATH -- fall back to running it as a module.
            self.proc = subprocess.Popen(
                [sys.executable, "-m", "mitmproxy.tools.main", "mitmdump"] + args,
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
            )

        # Give mitmdump a moment to import and bind; it crashes immediately
        # if dependencies are broken (e.g. bcrypt 5.x + passlib).
        time.sleep(1.5)
        if self.proc.poll() is not None:
            output = (self.proc.stdout.read() if self.proc.stdout else "").strip()
            self.proc = None
            hint = ""
            if "bcrypt" in output.lower() or "passlib" in output.lower():
                hint = (" Try: pip install \"bcrypt<5.0.0\" "
                        "(see requirements.txt).")
            return f"mitmdump exited immediately.{hint}\n{output or '(no output)'}"

        set_system_proxy(True)

        self.stop_watch.clear()
        self.log_thread = threading.Thread(target=self._drain_output, daemon=True)
        self.log_thread.start()
        self.watch_thread = threading.Thread(target=self._watch, daemon=True)
        self.watch_thread.start()
        return None

    def _drain_output(self):
        if not self.proc or not self.proc.stdout:
            return
        for line in self.proc.stdout:
            if self.stop_watch.is_set():
                break
            text = line.rstrip()
            if text and self.on_error:
                self.on_error(text)

    def _watch(self):
        while not self.stop_watch.is_set():
            if os.path.exists(RESULT_PATH):
                try:
                    with open(RESULT_PATH, "r", encoding="utf-8") as f:
                        data = json.load(f)
                    self.on_captured(data)
                except (json.JSONDecodeError, OSError):
                    pass
                return
            time.sleep(0.5)

    def stop(self):
        self.stop_watch.set()
        set_system_proxy(False)
        if self.proc and self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.proc.kill()
        self.proc = None


# --------------------------------------------------------------------------
# GUI
# --------------------------------------------------------------------------

class App:
    def __init__(self, root):
        self.root = root
        root.title("Steam Credential Capture")
        root.geometry("580x440")

        self.session = None

        tk.Label(
            root,
            text="1) Enable Capture   2) Launch Steam + RotMG, reach the main menu   3) Wait",
            wraplength=540, justify="left",
        ).pack(padx=10, pady=(10, 4), anchor="w")

        btn_frame = tk.Frame(root)
        btn_frame.pack(pady=6)

        self.enable_btn = tk.Button(btn_frame, text="Enable Capture", width=18, command=self.enable)
        self.enable_btn.grid(row=0, column=0, padx=5)

        self.disable_btn = tk.Button(btn_frame, text="Disable", width=18, command=self.disable, state="disabled")
        self.disable_btn.grid(row=0, column=1, padx=5)

        self.trust_btn = tk.Button(btn_frame, text="Trust CA (one-time)", width=18, command=self.trust_ca)
        self.trust_btn.grid(row=0, column=2, padx=5)

        self.status_var = tk.StringVar(value="Idle")
        tk.Label(root, textvariable=self.status_var, fg="blue").pack(pady=4)

        self.output = scrolledtext.ScrolledText(root, height=17, wrap="word")
        self.output.pack(fill="both", expand=True, padx=10, pady=10)
        self.output.configure(state="disabled")

        root.protocol("WM_DELETE_WINDOW", self.on_close)

    def log(self, text):
        self.output.configure(state="normal")
        self.output.insert("end", text + "\n")
        self.output.see("end")
        self.output.configure(state="disabled")

    def trust_ca(self):
        ok, msg = trust_mitmproxy_ca()
        if ok:
            self.log("[+] Certificate trusted.")
        else:
            self.log(f"[!] {msg}")

    def enable(self):
        self.status_var.set(f"Capturing... system proxy set to {PROXY_HOST}:{PROXY_PORT}")
        self.enable_btn.config(state="disabled")
        self.disable_btn.config(state="normal")
        self.log("[*] Starting capture proxy and enabling system proxy...")

        self.session = CaptureSession(
            on_captured=self.handle_capture,
            on_error=lambda line: self.root.after(0, self.log, f"[mitmdump] {line}"),
        )
        err = self.session.start()
        if err:
            self.log("[!] Capture proxy failed to start:")
            for line in err.splitlines():
                self.log(f"    {line}")
            self.disable()

    def handle_capture(self, data):
        # Runs on the watcher thread -- hop back to the Tk main thread.
        self.root.after(0, self._render_capture, data)

    def _render_capture(self, data):
        self.log("\n" + "=" * 50)
        self.log("CAPTURED:")
        for k, v in data.get("parsed_fields", {}).items():
            self.log(f"  {k}: {v}")
        self.log("=" * 50)
        self.status_var.set("Captured! Restoring your normal network settings...")
        self.disable()

    def disable(self):
        if self.session:
            self.session.stop()
            self.session = None
        self.enable_btn.config(state="normal")
        self.disable_btn.config(state="disabled")
        self.status_var.set("Idle (proxy restored to normal)")
        self.log("[*] System proxy restored, capture proxy stopped.")

    def on_close(self):
        if self.session:
            self.session.stop()
        self.root.destroy()


if __name__ == "__main__":
    root = tk.Tk()
    app = App(root)
    root.mainloop()