fake_keyauth.py

"""
Local fake KeyAuth.win server.
Listens on http://127.0.0.1:1337/api/x/ and impersonates KeyAuth API v1.2.
Knows the application's `secret`, so it produces correctly encrypted +
HMAC-signed responses that pass all client-side checks.

Required Python deps:
  pip install pycryptodome
"""
import base64
import hashlib
import hmac
import json
import os
import sys
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, parse_qsl

try:
    import urllib.request
    import ssl
except ImportError:
    pass

try:
    from Crypto.Cipher import AES
    from Crypto.Util.Padding import pad, unpad
except ImportError:
    sys.exit("missing pycryptodome. run: pip install pycryptodome")

# Constants harvested from TDataChecker[PRO].dll string table.
APP_NAME = "TDATA_checker[PRO]"
OWNER_ID = "90de0eb55024d31fa97c80792ac0fa2eb6dd17d83435b1e090fa4f9e25d8ce87"
SECRET   = "SGXgn-UqNm80dV-0"

# ---------------------------------------------------------------------------
# KeyAuth crypto helpers (replica of the official KeyAuth-Python-Example).
# ---------------------------------------------------------------------------
def _key_iv(secret: str, init_iv: str):
    k = hashlib.sha256(secret.encode()).hexdigest()[:32].encode()
    v = hashlib.sha256(init_iv.encode()).hexdigest()[:16].encode()
    return k, v

def kc_encrypt(plain: str, secret: str, init_iv: str) -> str:
    k, v = _key_iv(secret, init_iv)
    a = AES.new(k, AES.MODE_CBC, v)
    ct = a.encrypt(pad(plain.encode(), 16))
    return ct.hex()

def kc_decrypt(cipher_hex: str, secret: str, init_iv: str) -> str:
    k, v = _key_iv(secret, init_iv)
    a = AES.new(k, AES.MODE_CBC, v)
    pt = unpad(a.decrypt(bytes.fromhex(cipher_hex)), 16)
    return pt.decode()

# ---------------------------------------------------------------------------
# Cryptor: AES-128-ECB + base64  (discovered from DLL constants analysis)
# SECRET_KEY = SECRET = "SGXgn-UqNm80dV-0" (exactly 16 bytes)
# ---------------------------------------------------------------------------
def _ecb_key(idx=0):
    """Return AES key bytes for ECB mode. ECB_KEY env selects variant."""
    ek = SESSION.get("enckey", "")
    sid = SESSION.get("id", "")
    keys = [
        ("SECRET raw 16B",               SECRET.encode()),
        ("sha256(SECRET).digest[:16]",    hashlib.sha256(SECRET.encode()).digest()[:16]),
        ("sha256(SECRET).hex[:16]",       hashlib.sha256(SECRET.encode()).hexdigest()[:16].encode()),
        ("sha256(SECRET).hex[:32]",       hashlib.sha256(SECRET.encode()).hexdigest()[:32].encode()),
        ("OWNER_ID[:16]",                 OWNER_ID[:16].encode()),
        ("OWNER_ID[:32]",                 OWNER_ID[:32].encode()),
        ("sha256(OWNER).hex[:16]",        hashlib.sha256(OWNER_ID.encode()).hexdigest()[:16].encode()),
        ("sha256(OWNER).hex[:32]",        hashlib.sha256(OWNER_ID.encode()).hexdigest()[:32].encode()),
        ("sha256(OWNER).digest[:16]",     hashlib.sha256(OWNER_ID.encode()).digest()[:16]),
        ("sha256(enckey).hex[:16]",       hashlib.sha256(ek.encode()).hexdigest()[:16].encode() if ek else b"0"*16),
        ("sha256(sessionid).hex[:16]",    hashlib.sha256(sid.encode()).hexdigest()[:16].encode() if sid else b"0"*16),
    ]
    label, key = keys[idx % len(keys)]
    return label, key

def ecb_encrypt(plain: str) -> str:
    idx = int(os.environ.get("ECB_KEY", "0"))
    label, key = _ecb_key(idx)
    cipher = AES.new(key, AES.MODE_ECB)
    ct = cipher.encrypt(pad(plain.encode(), 16))
    b64 = base64.b64encode(ct).decode("ascii")
    sys.stderr.write(f"[srv]    ECB key #{idx}: {label} (len={len(key)})\n")
    return b64

def ecb_decrypt(enc_b64: str) -> str:
    idx = int(os.environ.get("ECB_KEY", "0"))
    _, key = _ecb_key(idx)
    cipher = AES.new(key, AES.MODE_ECB)
    pt = unpad(cipher.decrypt(base64.b64decode(enc_b64)), 16)
    return pt.decode()

def kc_sign(body: str, key: str = "", params=None) -> str:
    """Sign the response body with HMAC-SHA256 using OWNER_ID as the key.

    Discovered via proxy capture + exhaustive analysis:
      signature = hmac.new(OWNER_ID.encode(), body.encode(), sha256).hexdigest()
    where OWNER_ID is the 64-char hex string embedded in the DLL.
    """
    return hmac.new(OWNER_ID.encode(), body.encode(), hashlib.sha256).hexdigest()

# ---------------------------------------------------------------------------
# Stable session: when the app re-invokes /init we hand it the same id back.
# ---------------------------------------------------------------------------
SESSION = {
    "id": uuid.uuid4().hex,
    "now": int(time.time()),
}
FAR_FUTURE = "9999999999"  # year ~2286

def fake_init() -> dict:
    return {
        "success": True,
        "sessionid": SESSION["id"],
        "appinfo": {
            "numUsers":       "1",
            "numOnlineUsers": "1",
            "numKeys":        "1",
            "version":        "1.0",
            "customerPanelLink": "https://localhost/",
        },
        "message": "Initialized",
        "newSession": False,
        "nonce": uuid.uuid4().hex,
    }

def _userdata():
    return {
        "username": "cracked",
        "ip":       "127.0.0.1",
        "hwid":     "*",
        "createdate":  str(SESSION["now"]),
        "lastlogin":   str(SESSION["now"]),
        "subscriptions": [{
            "subscription": "default",
            "expiry":   FAR_FUTURE,
            "timeleft": 99999999,
        }],
    }

def fake_license() -> dict:
    return {
        "success": True,
        "message": "Logged in!",
        "info":    _userdata(),
        "nonce":   uuid.uuid4().hex,
    }

def fake_check() -> dict:
    return {"success": True, "message": "Session is valid.", "nonce": uuid.uuid4().hex}

def fake_log() -> dict:
    return {"success": True, "message": "Logged."}

def fake_default(req_type: str) -> dict:
    return {"success": True, "message": f"ok ({req_type})", "info": _userdata(),
            "nonce": uuid.uuid4().hex}

DISPATCH = {
    "init":     fake_init,
    "license":  fake_license,
    "check":    fake_check,
    "log":      fake_log,
}

# ---------------------------------------------------------------------------
# HTTP handler
# ---------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *a):
        sys.stderr.write("[srv] " + (fmt % a) + "\n")

    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        raw = self.rfile.read(length).decode("utf-8", errors="replace")
        sys.stderr.write(f"[srv] RAW POST {self.path} ({length}b): {raw}\n")
        params = dict(parse_qsl(raw, keep_blank_values=True))
        sys.stderr.write(f"[srv]    parsed keys: {list(params.keys())}\n")

        req_type = params.get("type", "")
        enckey = params.get("enckey", "")
        sessionid = params.get("sessionid", "")

        # Save enckey from init for encrypting subsequent responses.
        if req_type == "init" and enckey:
            SESSION["enckey"] = enckey
            sys.stderr.write(f"[srv]    saved enckey={enckey!r}\n")

        sys.stderr.write(f"[srv] -> type={req_type!r}\n")

        builder = DISPATCH.get(req_type)
        payload = builder() if builder else fake_default(req_type)
        plain = json.dumps(payload, separators=(",", ":"))

        # All responses are plain JSON.
        # The wrapper patches hmac.compare_digest → True and os._exit → no-op,
        # so encryption and signature correctness are irrelevant.
        body = plain
        sys.stderr.write(f"[srv]    body: {plain[:120]}\n")

        signature = kc_sign(body)

        body_bytes = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body_bytes)))
        self.send_header("signature", signature)
        self.end_headers()
        self.wfile.write(body_bytes)

    def do_GET(self):
        # Some KeyAuth clients ping the root for sanity. Just say hi.
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(b"fake-keyauth alive\n")


# ---------------------------------------------------------------------------
# PROXY mode: forward to real keyauth.win, log raw responses, analyze sig.
# Set env PROXY=1 to enable.
# ---------------------------------------------------------------------------
REAL_URL = "https://keyauth.win/api/1.2/"

def _proxy_request(raw_body: str, params: dict) -> tuple:
    """Forward request to real keyauth.win. Returns (status, body_str, sig_header)."""
    ctx = ssl.create_default_context()
    req = urllib.request.Request(
        REAL_URL,
        data=raw_body.encode("utf-8"),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, context=ctx, timeout=15) as resp:
            status = resp.status
            body = resp.read()
            sig = resp.headers.get("signature", "")
            return status, body, sig
    except Exception as ex:
        sys.stderr.write(f"[proxy] ERROR: {ex}\n")
        return 502, b'{"success":false,"message":"proxy error"}', ""


def _analyze_sig(body_str: str, sig_real: str, params: dict):
    """Try every known formula against the real sig and print matches."""
    b = body_str.encode()
    s = SECRET.encode()
    enckey = (params.get("enckey", "") or "").encode()
    ownerid = (params.get("ownerid", "") or "").encode()
    name = (params.get("name", "") or "").encode()
    h = (params.get("hash", "") or "").encode()
    sessionid = (params.get("sessionid", "") or "").encode()
    sha = lambda x: hashlib.sha256(x).hexdigest().encode()

    candidates = {
        "hmac(enckey, body)":        hmac.new(enckey, b, hashlib.sha256).hexdigest(),
        "hmac(secret, body)":        hmac.new(s, b, hashlib.sha256).hexdigest(),
        "hmac(ownerid, body)":       hmac.new(ownerid, b, hashlib.sha256).hexdigest() if ownerid else "",
        "hmac(name, body)":          hmac.new(name, b, hashlib.sha256).hexdigest() if name else "",
        "hmac(hash, body)":          hmac.new(h, b, hashlib.sha256).hexdigest() if h else "",
        "sha256(body)":              hashlib.sha256(b).hexdigest(),
        "sha256(body+secret)":       hashlib.sha256(b + s).hexdigest(),
        "sha256(secret+body)":       hashlib.sha256(s + b).hexdigest(),
        "sha256(s+b+s)":             hashlib.sha256(s + b + s).hexdigest(),
        "sha256(body+enckey)":       hashlib.sha256(b + enckey).hexdigest(),
        "sha256(enckey+body)":       hashlib.sha256(enckey + b).hexdigest(),
        "sha256(body+ownerid)":      hashlib.sha256(b + ownerid).hexdigest(),
        "sha256(ownerid+body)":      hashlib.sha256(ownerid + b).hexdigest(),
        "sha256(body+name)":         hashlib.sha256(b + name).hexdigest(),
        "sha256(body+hash)":         hashlib.sha256(b + h).hexdigest(),
        "sha256(hash+body)":         hashlib.sha256(h + b).hexdigest(),
        "sha256(s+enckey+body)":     hashlib.sha256(s + enckey + b).hexdigest(),
        "sha256(enckey+s+body)":     hashlib.sha256(enckey + s + b).hexdigest(),
        "sha256(body+s+enckey)":     hashlib.sha256(b + s + enckey).hexdigest(),
        "sha256(s+ownerid+body)":    hashlib.sha256(s + ownerid + b).hexdigest(),
        "sha256(ownerid+s+body)":    hashlib.sha256(ownerid + s + b).hexdigest(),
        "hmac(sha(s), body)":        hmac.new(sha(s), b, hashlib.sha256).hexdigest(),
        "hmac(sha(enckey), body)":   hmac.new(sha(enckey), b, hashlib.sha256).hexdigest(),
        "hmac(s+enckey, body)":      hmac.new(s + enckey, b, hashlib.sha256).hexdigest(),
        "hmac(enckey+s, body)":      hmac.new(enckey + s, b, hashlib.sha256).hexdigest(),
        "hmac(s+ownerid, body)":     hmac.new(s + ownerid, b, hashlib.sha256).hexdigest(),
    }
    sys.stderr.write(f"[analyze] real sig = {sig_real}\n")
    sys.stderr.write(f"[analyze] body[:120] = {body_str[:120]}\n")
    found = False
    for label, computed in candidates.items():
        if computed and computed == sig_real:
            sys.stderr.write(f"[analyze] *** MATCH: {label} ***\n")
            found = True
        # Also print first 16 chars for manual comparison
    if not found:
        sys.stderr.write(f"[analyze] NO MATCH among {len(candidates)} formulas\n")
        # Print all for manual comparison
        for label, computed in candidates.items():
            if computed:
                sys.stderr.write(f"[analyze]   {label:35s} = {computed[:16]}...\n")


class ProxyHandler(BaseHTTPRequestHandler):
    """Transparent proxy to real keyauth.win with response logging."""
    def log_message(self, fmt, *a):
        sys.stderr.write("[proxy] " + (fmt % a) + "\n")

    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        raw = self.rfile.read(length).decode("utf-8", errors="replace")
        params = dict(parse_qsl(raw, keep_blank_values=True))
        req_type = params.get("type", "?")
        sys.stderr.write(f"[proxy] >>> POST {self.path} type={req_type} ({length}b)\n")
        sys.stderr.write(f"[proxy]     params: {params}\n")

        status, body_bytes, sig = _proxy_request(raw, params)
        body_str = body_bytes.decode("utf-8", errors="replace")

        sys.stderr.write(f"[proxy] <<< status={status} body_len={len(body_bytes)} sig={sig[:32]}...\n")
        sys.stderr.write(f"[proxy]     body: {body_str[:300]}\n")

        # Analyze the signature
        _analyze_sig(body_str, sig, params)

        # Return exact response to the app
        self.send_response(status)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body_bytes)))
        if sig:
            self.send_header("signature", sig)
        self.end_headers()
        self.wfile.write(body_bytes)

    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(b"proxy-keyauth alive\n")


def main():
    proxy_mode = os.environ.get("PROXY") == "1"
    host, port = "127.0.0.1", 1337
    handler = ProxyHandler if proxy_mode else Handler
    srv = ThreadingHTTPServer((host, port), handler)
    mode_label = "PROXY -> keyauth.win" if proxy_mode else "FAKE (local)"
    sys.stderr.write(f"[srv] KeyAuth server listening on http://{host}:{port}/  mode={mode_label}\n")
    sys.stderr.write(f"[srv] secret = {SECRET}\n")
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        sys.stderr.write("[srv] bye\n")


if __name__ == "__main__":
    main()