Feature request - Login with 2fa enabled

[haoyanglu]

From my observation Twitter accounts with 2fa enabled have lower chances to get suspended / locked. While it is easy to use PyOTP to generate a temporary token, I don't have clue on how to pass it back to Twitter server (via GraphQL?) for verification. Is it possible to enable this feature? Or if the answer is no, could you shed some light on how to achieve that? Thank you.

[ghost]

From my observation Twitter accounts with 2fa enabled have lower chances to get suspended / locked.

That is complete nonsense.

[alenkimov]

implemented here https://github.com/alenkimov/tweepy-self/blob/f2475715a90b985cb366d993ab5aa83cbc664629/twitter/client.py#L1936

[vakandi]

Can someone push the tweepy feature 2fa into the pull request so @trevorhobenshield can review it?
Or help me develop it

[jonathanrstern]

@haoyanglu @alenkimov @vakandi @trevorhobenshield

any progress on this?

[vakandi]

@jonathanrstern not yet, I was busy lately, but I will need it in September for the software I'm building, when I start I'll post a message here in case someone is down to help me implement it (mid September)

[jonathanrstern]

Sounds great - I just built it myself actually. Working well. Shouldn't be too tricky for you to implement.

Let me know if you want the code.

[vakandi]

Hi man,
Well yeah it's awesome thanks for your work, I definitely would like the code haha so I can test it, if you could share a repo or collaborator access that will be great

[vakandi]

@jonathanrstern let me know I could really use it right now 👌🏻 thanks

[jonathanrstern]

stand by @vakandi

[jonathanrstern]

➜ twitter-api-client git:(twitter-2fa-login) git push --set-upstream origin twitter-2fa-login
remote: Permission to trevorhobenshield/twitter-api-client.git denied to jonathanrstern.
fatal: unable to access 'https://github.com/trevorhobenshield/twitter-api-client.git/': The requested URL returned error: 403

[jonathanrstern]

@trevorhobenshield if you give me permission I will make the PR

[jonathanrstern]

Here's the code:

import sys

from httpx import Client

from .constants import YELLOW, RED, BOLD, RESET, USER_AGENTS
from .util import find_key
import pyotp

def update_token(client: Client, key: str, url: str, **kwargs) -> Client:
    caller_name = sys._getframe(1).f_code.co_name
    try:
        headers = {
            'x-guest-token': client.cookies.get('guest_token', ''),
            'x-csrf-token': client.cookies.get('ct0', ''),
            'x-twitter-auth-type': 'OAuth2Client' if client.cookies.get('auth_token') else '',
        }
        client.headers.update(headers)
        r = client.post(url, **kwargs)
        info = r.json()

        for task in info.get('subtasks', []):
            if task.get('enter_text', {}).get('keyboard_type') == 'email':
                print(f"[{YELLOW}warning{RESET}] {' '.join(find_key(task, 'text'))}")
                client.cookies.set('confirm_email', 'true')  # signal that email challenge must be solved

            if task.get('subtask_id') == 'LoginAcid':
                if task['enter_text']['hint_text'].casefold() == 'confirmation code':
                    print(f"[{YELLOW}warning{RESET}] email confirmation code challenge.")
                    client.cookies.set('confirmation_code', 'true')

            if task.get('subtask_id') == 'LoginTwoFactorAuthChallenge':
                print(f"[{YELLOW}info{RESET}] 2FA challenge detected.")
                client.cookies.set('two_factor_auth', 'true')

        client.cookies.set(key, info[key])

    except KeyError as e:
        client.cookies.set('flow_errors', 'true')  # signal that an error occurred somewhere in the flow
        print(f'[{RED}error{RESET}] failed to update token at {BOLD}{caller_name}{RESET}\n{e}')
    return client


def init_guest_token(client: Client) -> Client:
    return update_token(client, 'guest_token', 'https://api.twitter.com/1.1/guest/activate.json')


def flow_start(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json',
                        params={'flow_name': 'login'},
                        json={
                            "input_flow_data": {
                                "flow_context": {
                                    "debug_overrides": {},
                                    "start_location": {"location": "splash_screen"}
                                }
                            }, "subtask_versions": {}
                        })


def flow_instrumentation(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
        "flow_token": client.cookies.get('flow_token'),
        "subtask_inputs": [{
            "subtask_id": "LoginJsInstrumentationSubtask",
            "js_instrumentation": {"response": "{}", "link": "next_link"}
        }],
    })


def flow_username(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
        "flow_token": client.cookies.get('flow_token'),
        "subtask_inputs": [{
            "subtask_id": "LoginEnterUserIdentifierSSO",
            "settings_list": {
                "setting_responses": [{
                    "key": "user_identifier",
                    "response_data": {"text_data": {"result": client.cookies.get('username')}}
                }], "link": "next_link"}}],
    })


def flow_password(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
        "flow_token": client.cookies.get('flow_token'),
        "subtask_inputs": [{
            "subtask_id": "LoginEnterPassword",
            "enter_password": {"password": client.cookies.get('password'), "link": "next_link"}}]
    })


def flow_duplication_check(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
        "flow_token": client.cookies.get('flow_token'),
        "subtask_inputs": [{
            "subtask_id": "AccountDuplicationCheck",
            "check_logged_in_account": {"link": "AccountDuplicationCheck_false"},
        }],
    })


def confirm_email(client: Client) -> Client:
    return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
        "flow_token": client.cookies.get('flow_token'),
        "subtask_inputs": [
            {
                "subtask_id": "LoginAcid",
                "enter_text": {
                    "text": client.cookies.get('email'),
                    "link": "next_link"
                }
            }]
    })


def solve_confirmation_challenge(client: Client, **kwargs) -> Client:
    if fn := kwargs.get('proton'):
        confirmation_code = fn()
        return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
            "flow_token": client.cookies.get('flow_token'),
            'subtask_inputs': [
                {
                    'subtask_id': 'LoginAcid',
                    'enter_text': {
                        'text': confirmation_code,
                        'link': 'next_link',
                    },
                },
            ],
        })


def solve_2fa_challenge(client: Client, totp_secret: str, backup_code: str) -> Client:
    try:
        totp = pyotp.TOTP(totp_secret).now()
        return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
            "flow_token": client.cookies.get('flow_token'),
            'subtask_inputs': [
                {
                    'subtask_id': 'LoginTwoFactorAuthChallenge',
                    'enter_text': {
                        'text': totp,
                        'link': 'next_link',
                    },
                },
            ],
        })
    except Exception as e:
        print(f'[{RED}error{RESET}] TOTP failed')


def execute_login_flow(client: Client, totp_secret: str = None, backup_code: str = None, **kwargs) -> Client | None:
    client = init_guest_token(client)
    for fn in [flow_start, flow_instrumentation, flow_username, flow_password, flow_duplication_check]:
        client = fn(client)

    # solve email challenge
    if client.cookies.get('confirm_email') == 'true':
        client = confirm_email(client)

    # solve confirmation challenge (Proton Mail only)
    if client.cookies.get('confirmation_code') == 'true':
        if not kwargs.get('proton'):
            print(f'[{RED}warning{RESET}] Please check your email for a confirmation code'
                  f' and log in again using the web app. If you wish to automatically solve'
                  f' email confirmation challenges, add a Proton Mail account in your account settings')
            return
        client = solve_confirmation_challenge(client, **kwargs)
    
    if client.cookies.get('two_factor_auth') == 'true':
        if not totp_secret and not backup_code:
            print(f'[{RED}error{RESET}] 2FA is enabled but no TOTP secret or backup code provided.')
            return
        client = solve_2fa_challenge(client, totp_secret, backup_code)
    
    return client


def login(email: str, username: str, password: str, totp_secret: str = None, backup_code: str = None, **kwargs) -> Client:
    client = Client(
        cookies={
            "email": email,
            "username": username,
            "password": password,
            "guest_token": None,
            "flow_token": None,
        },
        headers={
            'authorization': 'Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA',
            'content-type': 'application/json',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36',
            'x-twitter-active-user': 'yes',
            'x-twitter-client-language': 'en',
        },
        follow_redirects=True
    )
    client = execute_login_flow(client, totp_secret, backup_code, **kwargs)

    cookies = dict(client.cookies)
    print('cookies:', cookies)

    if not client or client.cookies.get('flow_errors') == 'true':
        raise Exception(f'[{RED}error{RESET}] {BOLD}{username}{RESET} login failed')
    return client

[vakandi]

Thanks a lot man, nice clean job I'll test it soon

[vakandi]

@jonathanrstern I tested your code it works nicely with 'raw' 2fa code, thanks!

Some accounts do use https://2fa.fb.rip for generating the code though, does anyone needs the implementation of it? i coded it yesterday

[vakandi]

Actually my test were using ct0 and auth_token , so my bad on that i thought it worked, i tested the code again and i actually have an error, i tested it with multiple different account, using username, mail ,password and 2fa, and every time it tries to loggin it fails with this error :
[error] failed to update token at flow_duplication_check.
I used this code to try to generate a new ct0 and auth_token from a twitter 2fa account (i added a 2fa flag):

new_account = Account(email=email, username=username, password=password, fa_flag=True, fa_code=fa_code)
new_cookies = new_account.session.cookies

How i do my test:

I tried with your function :

       totp = pyotp.TOTP(totp_secret).now()
       return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
           "flow_token": client.cookies.get('flow_token'),
           'subtask_inputs': [
               {
                   'subtask_id': 'LoginTwoFactorAuthChallenge',
                   'enter_text': {
                       'text': totp,
                       'link': 'next_link',
                   },
               },
           ],
       })

and i tried with the code given by "https://2fa.fb.rip" api:
How i retrieve the code :

url = f'https://2fa.fb.rip/api/otp/{fa_code}'
   try:
       with Client() as session:
           r = session.get(url)
           if not r.json()['ok']:
               raise Exception(f'Invalid 2fa code, 2fa.fb.rip response: {r.json()}')

How i modify the solve_2fa_challenge function by just giving it the new code in the parameter directly:

       return update_token(client, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', json={
           "flow_token": client.cookies.get('flow_token'),
           'subtask_inputs': [
               {
                   'subtask_id': 'LoginTwoFactorAuthChallenge',
                   'enter_text': {
                       'text': totp_secret,
                       'link': 'next_link',
                   },
               },
           ],
       })