nitter/tools/get_account.py

#!/usr/bin/env python3
from sys import argv
import requests
import pyotp
import json

TW_CONSUMER_KEY = "3nVuSoBZnx6U4vzUxf5w"
TW_CONSUMER_SECRET = "Bcs59EFbbsdF6Sl9Ng71smgStWEGwXXKSjYvPVt7qys"


def auth(username, password, otp_secret):
    bearer_token_req = requests.post(
        "https://api.twitter.com/oauth2/token",
        auth=(TW_CONSUMER_KEY, TW_CONSUMER_SECRET),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        data="grant_type=client_credentials",
    ).json()
    bearer_token = " ".join(str(x) for x in bearer_token_req.values())

    guest_token = (
        requests.post(
            "https://api.twitter.com/1.1/guest/activate.json",
            headers={"Authorization": bearer_token},
        )
        .json()
        .get("guest_token")
    )

    if not guest_token:
        print("Failed to obtain guest token.")
        exit(1)

    twitter_header = {
        "Authorization": bearer_token,
        "Content-Type": "application/json",
        "User-Agent": "TwitterAndroid/10.21.0-release.0 (310210000-r-0) ONEPLUS+A3010/9 (OnePlus;ONEPLUS+A3010;OnePlus;OnePlus3;0;;1;2016)",
        "X-Twitter-API-Version": "5",
        "X-Twitter-Client": "TwitterAndroid",
        "X-Twitter-Client-Version": "10.21.0-release.0",
        "OS-Version": "28",
        "System-User-Agent": "Dalvik/2.1.0 (Linux; U; Android 9; ONEPLUS A3010 Build/PKQ1.181203.001)",
        "X-Twitter-Active-User": "yes",
        "X-Guest-Token": guest_token,
        "X-Twitter-Client-DeviceID": "",
    }

    session = requests.Session()
    session.headers = twitter_header

    task1 = session.post(
        "https://api.twitter.com/1.1/onboarding/task.json",
        params={
            "flow_name": "login",
            "api_version": "1",
            "known_device_token": "",
            "sim_country_code": "us",
        },
        json={
            "flow_token": None,
            "input_flow_data": {
                "country_code": None,
                "flow_context": {
                    "referrer_context": {
                        "referral_details": "utm_source=google-play&utm_medium=organic",
                        "referrer_url": "",
                    },
                    "start_location": {"location": "deeplink"},
                },
                "requested_variant": None,
                "target_user_id": 0,
            },
        },
    )

    session.headers["att"] = task1.headers.get("att")

    task2 = session.post(
        "https://api.twitter.com/1.1/onboarding/task.json",
        json={
            "flow_token": task1.json().get("flow_token"),
            "subtask_inputs": [
                {
                    "enter_text": {
                        "suggestion_id": None,
                        "text": username,
                        "link": "next_link",
                    },
                    "subtask_id": "LoginEnterUserIdentifier",
                }
            ],
        },
    )

    task3 = session.post(
        "https://api.twitter.com/1.1/onboarding/task.json",
        json={
            "flow_token": task2.json().get("flow_token"),
            "subtask_inputs": [
                {
                    "enter_password": {"password": password, "link": "next_link"},
                    "subtask_id": "LoginEnterPassword",
                }
            ],
        },
    )

    for t3_subtask in task3.json().get("subtasks", []):
        if "open_account" in t3_subtask:
            return t3_subtask["open_account"]
        elif "enter_text" in t3_subtask:
            response_text = t3_subtask["enter_text"]["hint_text"]
            totp = pyotp.TOTP(otp_secret)
            generated_code = totp.now()
            task4resp = session.post(
                "https://api.twitter.com/1.1/onboarding/task.json",
                json={
                    "flow_token": task3.json().get("flow_token"),
                    "subtask_inputs": [
                        {
                            "enter_text": {
                                "suggestion_id": None,
                                "text": generated_code,
                                "link": "next_link",
                            },
                            "subtask_id": "LoginTwoFactorAuthChallenge",
                        }
                    ],
                },
            )
            task4 = task4resp.json()
            for t4_subtask in task4.get("subtasks", []):
                if "open_account" in t4_subtask:
                    return t4_subtask["open_account"]

    return None


if __name__ == "__main__":
    if len(argv) != 4:
        print("Usage: %s <username> <password> <2fa secret>" % argv[0])
        exit(1)

    username = argv[1]
    password = argv[2]
    otp_secret = argv[3]

    result = auth(username, password, otp_secret)
    if result is None:
        print("Authentication failed.")
        exit(1)

    print(
        json.dumps(
            {
                "oauth_token": result.get("oauth_token"),
                "oauth_token_secret": result.get("oauth_token_secret"),
            }
        )
    )