fix #524: login with robinhood mfa

[pulkitcollier]

Prompt for mfa code after first request to resolve Issue #524

[regholl2023]

can some one tell me whats wrong with my code that i keep getting KeyError: 'status',
import getpass
import os
import pickle
import random

from robin_stocks.robinhood.helper import *
from robin_stocks.robinhood.urls import *

def generate_device_token():
"""This function will generate a token used when loggin on.

:returns: A string representing the token.

"""
rands = []
for i in range(0, 16):
    r = random.random()
    rand = 4294967296.0 * r
    rands.append((int(rand) >> ((3 & i) << 3)) & 255)

hexa = [hex(i+256).lstrip("0x").rstrip("L")[1:] for i in range(0, 256)]
id = ""
for i in range(0, 16):
    id += hexa[rands[i]]

    if i in [3, 5, 7, 9]:
        id += "-"

return(id)

def respond_to_challenge(challenge_id, sms_code):
"""This function will post to the challenge url.

:param challenge_id: The challenge id.
:type challenge_id: str
:param sms_code: The sms code.
:type sms_code: str
:returns:  The response from requests.

"""
url = challenge_url(challenge_id)
payload = {
    'response': sms_code
}
return(request_post(url, payload))

def login(username=None, password=None, expiresIn=86400, scope='internal', by_sms=True, store_session=True, mfa_code=None, pickle_path="", pickle_name=""):
device_token = generate_device_token()
home_dir = os.path.expanduser("~")
data_dir = os.path.join(home_dir, ".tokens")
if pickle_path:
if not os.path.isabs(pickle_path):
# normalize relative paths
pickle_path = os.path.normpath(os.path.join(os.getcwd(), pickle_path))
data_dir = pickle_path
if not os.path.exists(data_dir):
os.makedirs(data_dir)
creds_file = f"robinhood{pickle_name}.pickle"
pickle_path = os.path.join(data_dir, creds_file)
# Challenge type is used if not logging in with two-factor authentication.
challenge_type = "sms" if by_sms else "email"
url = login_url()
payload = {
'client_id': 'c82SH0WZOsabOXGP2sxqcj34FxkvfnWRZBKlBjFS',
'expires_in': expiresIn,
'grant_type': 'password',
'password': password,
'scope': scope,
'username': username,
'challenge_type': challenge_type,
'device_token': device_token,
'try_passkeys': False,
'token_request_path':'/login',
'create_read_only_secondary_token':True,
'request_id': '848bd19e-02bc-45d9-99b5-01bce5a79ea7'
}

if mfa_code:
    payload['mfa_code'] = mfa_code
# If authentication has been stored in pickle file then load it. Stops login server from being pinged so much.
if os.path.isfile(pickle_path):
    # If store_session has been set to false then delete the pickle file, otherwise try to load it.
    # Loading pickle file will fail if the acess_token has expired.
    if store_session:
        try:
            with open(pickle_path, 'rb') as f:
                pickle_data = pickle.load(f)
                access_token = pickle_data['access_token']
                token_type = pickle_data['token_type']
                refresh_token = pickle_data['refresh_token']
                # Set device_token to be the original device token when first logged in.
                pickle_device_token = pickle_data['device_token']
                payload['device_token'] = pickle_device_token
                # Set login status to True in order to try and get account info.
                set_login_state(True)
                update_session(
                    'Authorization', '{0} {1}'.format(token_type, access_token))
                # Try to load account profile to check that authorization token is still valid.
                res = request_get(
                    positions_url(), 'pagination', {'nonzero': 'true'}, jsonify_data=False)
                # Raises exception is response code is not 200.
                res.raise_for_status()
                return({'access_token': access_token, 'token_type': token_type,
                        'expires_in': expiresIn, 'scope': scope, 'detail': 'logged in using authentication in {0}'.format(creds_file),
                        'backup_code': None, 'refresh_token': refresh_token})
        except Exception:
            print(
                "ERROR: There was an issue loading pickle file. Authentication may be expired - logging in normally.", file=get_output())
            set_login_state(False)
            update_session('Authorization', None)
    else:
        os.remove(pickle_path)
# Try to log in normally.
if not username:
    username = input("Robinhood username: ")
    payload['username'] = username
if not password:
    password = getpass.getpass("Robinhood password: ")
    payload['password'] = password
data = request_post(url, payload)
if not data:
    raise Exception('Error: Trouble connecting to robinhood API. Check internet connection.')
if 'mfa_required' in data:
    mfa_token = input("Please type in the MFA code: ")
    payload['mfa_code'] = mfa_token
    res = request_post(url, payload, jsonify_data=False)
    while (res.status_code != 200):
        mfa_token = input(
            "That MFA code was not correct. Please type in another MFA code: ")
        payload['mfa_code'] = mfa_token
        res = request_post(url, payload, jsonify_data=False)
    data = res.json()
elif 'challenge' in data:
    challenge_id = data['challenge']['id']
    sms_code = input('Enter Robinhood code for validation: ')
    res = respond_to_challenge(challenge_id, sms_code)
    while 'challenge' in res and res['challenge']['remaining_attempts'] > 0:
        sms_code = input('That code was not correct. {0} tries remaining. Please type in another code: '.format(
            res['challenge']['remaining_attempts']))
        res = respond_to_challenge(challenge_id, sms_code)
    update_session(
        'X-ROBINHOOD-CHALLENGE-RESPONSE-ID', challenge_id)
    data = request_post(url, payload)
elif 'verification_workflow' in data:
    workflow_id = data['verification_workflow']['id']
    _validate_sherrif_id(device_token=device_token, workflow_id=workflow_id, mfa_code=mfa_code)
    data = request_post(url, payload)
if 'access_token' not in data:
    raise Exception(data['detail'])
token = '{0} {1}'.format(data['token_type'], data['access_token'])
update_session('Authorization', token)
set_login_state(True)
data['detail'] = "logged in with brand new authentication code."
if store_session:
    with open(pickle_path, 'wb') as f:
        pickle.dump({'token_type': data['token_type'],
                     'access_token': data['access_token'],
                     'refresh_token': data['refresh_token'],
                     'device_token': payload['device_token']}, f)

return(data)

def _validate_sherrif_id(device_token:str, workflow_id:str,mfa_code:str):
url = "https://api.robinhood.com/pathfinder/user_machine/"
payload = {
'device_id': device_token,
'flow': 'suv',
'input':{'workflow_id': workflow_id}
}
data = request_post(url=url, payload=payload,json=True )
if "id" in data:
inquiries_url = f"https://api.robinhood.com/pathfinder/inquiries/{data['id']}/user_view/"
res = request_get(inquiries_url)
challenge_id=res['type_context']["context"]["sheriff_challenge"]["id"]
challenge_url = f"https://api.robinhood.com/challenge/{challenge_id}/respond/"
challenge_payload = {
'response': mfa_code
}
challenge_response = request_post(url=challenge_url, payload=challenge_payload,json=True )
if challenge_response["status"] == "validated":
inquiries_payload = {"sequence": 0, "user_input": {"status": "continue"}}
inquiries_response = request_post(url=inquiries_url, payload=inquiries_payload, json=True)
if inquiries_response["type_context"]["result"] == "workflow_status_approved":
return
else:
raise Exception("Workflow status not approved")
else:
raise Exception("Challenge not validated")
else:
raise Exception("ID not returned in user-machine call")

def _get_sherrif_challenge(token_id:str, data:dict):

if "id" in data:
    return data["id"]
raise Exception("Id not returned in user-machine call")

@login_required
def logout():
"""Removes authorization from the session header.

:returns: None

"""
set_login_state(False)
update_session('Authorization', None)

[noahfields]

@pulkitcollier I submitted a parallel request for this fix. I think @pulkitcollier's solution looks better than mine. @jmfernandes

[qtsweeney]

@pulkitcollier @noahfields - Thanks gents for this, you both are legends! For this commit - I replaced this code with my authentication.py file. I have a test script that spits out the MFA code, and tells me to prompt it. I input the MFA code and get the below error, any thoughts?

Error:

Error during Robinhood login: 'dict' object has no attribute 'status_code'

This is my test script:

import robin_stocks.robinhood as robin # Ensure you're importing from the correct submodule
import credentials
import pyotp

def login_robinhood():
try:
# Generate the MFA code
totp = pyotp.TOTP(credentials.MFA_APP_CODE).now()
print(f"MFA Code: {totp}")

    # Attempt to login with username, password, and MFA code
    login_response = robin.login(
        username=credentials.RH_Login,
        password=credentials.RH_Password,
        mfa_code='',
        store_session=False  # This is optional; set to True if you want to store the session
    )

    # Print the entire login response to understand its structure
    print(f"Login Response: {login_response}")

    # Check if login was successful
    if login_response:
        print(f"Login successful! Response: {login_response}")
    else:
        print("Login failed. Please check your credentials and MFA code.")
    
except Exception as e:
    # Handle any errors that occur during login
    print(f"An error occurred while logging in: {e}")

Call the login function to test

login_robinhood()

[finlop]

anyone know how to make the code work for device approvals. i only get device approvals now from my iphone robinhood app. and then the code i run waits for the device approval to be approved and then checks to see if i can log in successfully. but it never sees that i approved the device.

import random
import getpass
import os
import pickle
import requests
import time

Generate a unique device token
def generate_device_token():
"""Generates a unique device token."""
rands = []
for _ in range(16):
rand = random.randint(0, 255)
rands.append(rand)

hexa = [str(hex(i + 256)).lstrip("0x")[1:] for i in range(256)]
token = ""
for i, r in enumerate(rands):
token += hexa[r]
if i in [3, 5, 7, 9]:
token += "-"
return token
Respond to challenges
def respond_to_challenge(challenge_id, sms_code):
"""Responds to a challenge with the provided code."""
url = f"https://api.robinhood.com/challenge/{challenge_id}/respond/"
payload = {"response": sms_code}
response = requests.post(url, json=payload)
return response.json()

Save and load session data
def save_session(data, pickle_name="robinhood_session.pickle"):
"""Saves session data to a pickle file."""
with open(pickle_name, "wb") as f:
pickle.dump(data, f)

def load_session(pickle_name="robinhood_session.pickle"):
"""Loads session data from a pickle file if it exists."""
if os.path.exists(pickle_name):
with open(pickle_name, "rb") as f:
return pickle.load(f)
return None

def handle_verification_workflow(device_token, workflow_id):
"""Handles verification workflow for device approval."""
print("Handling verification workflow. Please approve the login request on your device.")
url = f"https://api.robinhood.com/pathfinder/user_machine/"
payload = {
"device_id": device_token,
"flow": "suv",
"input": {"workflow_id": workflow_id}
}
response = requests.post(url, json=payload)
if response.status_code == 200:
print("Verification workflow initiated successfully. Waiting for approval...")
else:
raise Exception(f"Failed to initiate verification workflow: {response.text}")

inquiries_url = f"https://api.robinhood.com/pathfinder/inquiries/{response.json()['id']}/user_view/"
max_attempts = 5 # Increase retries to 30 attempts (5 minutes total)
delay_between_attempts = 10 # 10 seconds between checks

for attempt in range(max_attempts):
print(f"Attempt {attempt + 1}: Checking approval status...")
try:
inquiry_response = requests.get(inquiries_url, timeout=10).json()
status = inquiry_response["type_context"]["context"]["sheriff_challenge"]["status"]
print(f"Device approval status: {status}")

    if status == "approved":
        print("Device approval successful!")
        return

    if status == "denied":
        raise Exception("Device approval was denied.")

except Exception as e:
    print(f"Error while checking approval status: {e}")

time.sleep(delay_between_attempts)

print("Device approval timed out. Please ensure you approved the login request.")

Login to Robinhood

def robinhood_login(username=None, password=None, expires_in=86400, scope="internal"):
"""Logs into Robinhood and handles verification workflows."""
device_token = generate_device_token()
print("Generated Device Token:", device_token)

url = "https://api.robinhood.com/oauth2/token/"
payload = {
"grant_type": "password",
"scope": scope,
"client_id": "c82SH0WZOsabOXGP2sxqcj34FxkvfnWRZBKlBjFS",
"device_token": device_token,
"username": username,
"password": password,
}

session = load_session()
if session:
print("Using saved session.")
return session["access_token"]

response = requests.post(url, data=payload)
data = response.json()

if response.status_code == 200:
print("Login successful.")
save_session(data)
return data["access_token"]

if "verification_workflow" in data:
workflow_id = data["verification_workflow"]["id"]
print(f"Verification workflow triggered. Workflow ID: {workflow_id}")
handle_verification_workflow(device_token, workflow_id)
response = requests.post(url, data=payload)
data = response.json()
if response.status_code == 200:
print("Login successful after verification workflow.")
save_session(data)
return data["access_token"]

print("Failed to log in:", data)
return None
Main flow
if name == "main":
username = input("Enter your Robinhood username: ")
password = getpass.getpass("Enter your Robinhood password: ")
token = robinhood_login(username=username, password=password)
if token:
print("Access token:", token)
else:
print("Failed to log in.")

Verification workflow triggered. Workflow ID: 00e60f42-de4c-4282-ad7d-e9c01a0206e1
Handling verification workflow. Please approve the login request on your device.
Verification workflow initiated successfully. Waiting for approval...
Attempt 1: Checking approval status...
Device approval status: issued
Attempt 2: Checking approval status...
Device approval status: issued
Attempt 3: Checking approval status...
Device approval status: issued
Attempt 4: Checking approval status...
Device approval status: issued
Attempt 5: Checking approval status...
Device approval status: issued
Device approval timed out. Please ensure you approved the login request.
Failed to log in: {'verification_workflow': {'id': '00e60f42-de4c-4282-ad7d-e9c01a0206e1', 'workflow_status': 'workflow_status_internal_pending'}}
Failed to log in.

[clayton-littlejohn]

❗ Noticed some potential issues with this approach:

1. Forcing a Numeric Code, Even If You Only Got a Link

You always prompt the user for a “Sheriff Challenge code.” But in many “verification_workflow” flows, Robinhood only sends a link or an app push notification to approve the new device (no numeric code). The user may have nothing to type. If that’s the case, the user is stuck, or they might attempt to type something meaningless and get repeated errors.

A better approach may be to inform the user that if they only see a link, they must click it in the Robinhood email/app. If the user truly has a numeric code, they type it here. Otherwise, the code will keep failing.

2. The While Loop in _validate_sherrif_id() Reprompts for MFA in the Wrong Place: Inside _validate_sherrif_id()

data = request_post(url=url, payload=payload, json=True)
while (data.status_code != 200):
    mfa_code = input("That MFA code was not correct. Please type in another MFA code: ")
    data = request_post(url, payload, jsonify_data=False)

Potential issues here:

1. This is the user_machine call, not the actual “respond to challenge” call. The payload used is:

payload = {
    'device_id': device_token,
    'flow': 'suv',
    'input': {'workflow_id': workflow_id}
}

2. That does not include the user’s numeric code in the payload. So re-typing the code doesn’t change anything about how this call is made—it’s always the same body.
   If data.status_code != 200 is due to a network error, or some other reason, you’re stuck prompting for an MFA code that doesn’t even get used in that payload.

• In other words, this loop conflates “retry the numeric code” with “retry the user_machine call.” They’re different API endpoints. In reality, the code-based challenge is handled later with:

challenge_payload = {'response': mfa_code}
challenge_response = request_post(url=challenge_url, payload=challenge_payload, json=True)

That is where a wrong code should trigger a re-prompt. But in your snippet, you do it on the user_machine call.

3. No Separate Handling for Link-Based Approvals

If Robinhood only sent an email reading “Click here to approve this device,” the user has no numeric code. They must manually click the link. Your code:

mfa_code = input("Please type in the MFA code: ")
_validate_sherrif_id(...)

4. Potential for Infinite Loops or Confusion

Because _validate_sherrif_id() uses:

while (data.status_code != 200):
    ...
    data = request_post(url, payload, jsonify_data=False)

there’s a risk of repeated failing if the code is truly wrong or if it’s a link-based challenge. You’ll keep prompting for an MFA code but never calling the actual challenge endpoint to re-check the code. The loop might never exit if the root cause isn’t a bad code but a link-based approval.

---

✔️ Improvised Patches:

A simplified fix to this without any additional complexities may be a minor alteration in the verification_workflow:

        elif 'verification_workflow' in data:
            workflow_id = data['verification_workflow']['id']
            if not mfa_code:
                print("If you received a numeric 'Sheriff Challenge' code via SMS or email, enter it below. If you only got a link, you must approve that link instead.")
                mfa_code = input("Sheriff Challenge code (if any): ")
            _validate_sherrif_id(device_token=device_token, workflow_id=workflow_id, mfa_code=mfa_code)
            data = request_post(url, payload)

• Separates the sheriff challenge logic into _validate_sherrif_id()
• Warns the user about the difference between a code-based vs. link-based challenge.
• Avoids the confusing infinite re-prompt on the “user_machine”.

Potential enhancments with this approach:

• If the user tries to proceed without a code in a link-based scenario, we'll raise an exception. That’s not a bug per se, but we might want to refine the user message or do a more graceful exit—however, that’s a design choice, and the code is still workable as is.

@noahfields approach also works well being that they prompt for a Sheriff Code in _validate_sherrif_id: #532 🏆

if mfa_code == None:
    mfa_code = input("Please type in the MFA code: ")

in _validate_sherrif_id(), you allow users who receive a numeric code via SMS or email to enter it. Previously, the library would try to validate with mfa_code=None and fail immediately.

• Avoids Hard 401
  If the user has a code-based challenge, they can now provide that code in real-time rather than being stuck.
• Maintains Old Flows
  The rest of the code for 'mfa_required' and 'challenge' blocks remains unchanged, so normal SMS/email 2FA or the older challenge flows still work as before.

❗ Improvement: Add a short clarity message. This at least clarifies that link-based verification requires a manual step. But the rest is still the user’s responsibility to click that link.

Additional lore around the origin of this issue:

• During the week of December 8th:
  • Robinhood has introduced, or made changes to, a “Sheriff Challenge” or “verification_workflow” mechanism for unrecognized devices. This does not set "mfa_required" in the JSON response, so the normal SMS/Email “challenge” block is never triggered.
  • When mfa_code is not provided, _validate_sherrif_id() fails immediately because it sends None to the endpoint. The new code-based prompt fixes this edge case for anyone receiving a numeric code.

[jmfernandes]

PR is redundant with #532 being merged in.