quantfortune/SIP_GoldBees_Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add OAuth completion step after twofa

Browse Source 
After twofa verifies identity, re-visit connect/login as authenticated user.
Zerodha then redirects to the registered callback URL with request_token.
This mirrors what the browser JS does to complete the OAuth flow.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Thigazhezhilan J 2026-05-26 21:55:19 +05:30
parent 9d1126b84d
commit bec31b9f9d
1 changed files with 26 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
backend/app/services/auto_login_service.py
View File

@ -258,9 +258,8 @@ def _perform_zerodha_login(
flush=True,
)
# Step 4: Extract request_token.
# Modern Zerodha (SPA): returns 200 JSON with data.redirect_url.
# Older behavior: 302 Location header redirect.
# Step 4: Extract request_token from twofa response (some Zerodha versions
# return redirect_url directly in the JSON body).
request_token = None
try:
@ -273,18 +272,38 @@ def _perform_zerodha_login(
except Exception:
pass
# Also check twofa Location header
if not request_token:
location = twofa_resp.headers.get("Location", "")
for _ in range(10):
if "request_token" in location:
parsed = urlparse(location)
params = parse_qs(parsed.query)
request_token = params.get("request_token", [None])[0]
# Step 5: Complete OAuth flow.
# In a browser, after twofa succeeds, Zerodha's JavaScript re-visits the
# connect/login URL as an authenticated user. Zerodha detects the valid
# session and redirects to the registered callback URL with request_token.
# We follow the same redirect chain here to capture the token.
if not request_token:
next_url = f"https://kite.zerodha.com/connect/login?v=3&api_key={api_key}"
for attempt in range(10):
step_resp = session.get(next_url, timeout=15, allow_redirects=False)
location = step_resp.headers.get("Location", "")
print(
f"[AUTO-LOGIN-DEBUG] oauth_complete[{attempt}] "
f"status={step_resp.status_code} "
f"location={location[:300] if location else 'NONE'}",
flush=True,
)
if "request_token" in location:
parsed = urlparse(location)
params = parse_qs(parsed.query)
request_token = params.get("request_token", [None])[0]
break
if not location or twofa_resp.status_code not in (301, 302, 303, 307, 308):
if not location or step_resp.status_code not in (301, 302, 303, 307, 308):
break
twofa_resp = session.get(location, allow_redirects=False, timeout=15)
location = twofa_resp.headers.get("Location", "")
next_url = location
if not request_token:
raise AutoLoginError(