Receive token update from Chrome extension (no admin auth required, uses connection_token)
(request: dict, authorization: Optional[str] = Header(None))
| 2342 | |
| 2343 | @router.post("/api/plugin/update-token") |
| 2344 | async def plugin_update_token(request: dict, authorization: Optional[str] = Header(None)): |
| 2345 | """Receive token update from Chrome extension (no admin auth required, uses connection_token)""" |
| 2346 | await _verify_plugin_connection_token(authorization) |
| 2347 | plugin_config = await db.get_plugin_config() |
| 2348 | |
| 2349 | # Extract session token from request |
| 2350 | session_token = request.get("session_token") |
| 2351 | |
| 2352 | if not session_token: |
| 2353 | raise HTTPException(status_code=400, detail="Missing session_token") |
| 2354 | |
| 2355 | # Step 1: Convert ST to AT to get user info (including email) |
| 2356 | try: |
| 2357 | result = await token_manager.flow_client.st_to_at(session_token) |
| 2358 | at = result["access_token"] |
| 2359 | expires = result.get("expires") |
| 2360 | user_info = result.get("user", {}) |
| 2361 | email = user_info.get("email", "") |
| 2362 | |
| 2363 | if not email: |
| 2364 | raise HTTPException(status_code=400, detail="Failed to get email from session token") |
| 2365 | |
| 2366 | # Parse expiration time |
| 2367 | from datetime import datetime |
| 2368 | at_expires = None |
| 2369 | if expires: |
| 2370 | try: |
| 2371 | at_expires = datetime.fromisoformat(expires.replace('Z', '+00:00')) |
| 2372 | except: |
| 2373 | pass |
| 2374 | |
| 2375 | except Exception as e: |
| 2376 | raise HTTPException(status_code=400, detail=f"Invalid session token: {str(e)}") |
| 2377 | |
| 2378 | # Step 2: Check if token with this email exists |
| 2379 | existing_token = await db.get_token_by_email(email) |
| 2380 | |
| 2381 | if existing_token: |
| 2382 | # Update existing token |
| 2383 | try: |
| 2384 | # Update token |
| 2385 | await token_manager.update_token( |
| 2386 | token_id=existing_token.id, |
| 2387 | st=session_token, |
| 2388 | at=at, |
| 2389 | at_expires=at_expires, |
| 2390 | protocol_mode=request.get("protocol_mode"), |
| 2391 | google_cookies=request.get("google_cookies"), |
| 2392 | login_account=request.get("login_account"), |
| 2393 | login_password=request.get("login_password"), |
| 2394 | proxy_url=request.get("proxy_url"), |
| 2395 | auto_refresh_enabled=request.get("auto_refresh_enabled"), |
| 2396 | refresh_interval_minutes=request.get("refresh_interval_minutes"), |
| 2397 | ) |
| 2398 | |
| 2399 | # Check if auto-enable is enabled and token is disabled |
| 2400 | if plugin_config.auto_enable_on_update and not existing_token.is_active: |
| 2401 | await token_manager.enable_token(existing_token.id) |
nothing calls this directly
no test coverage detected