feat: one-click upstream auth refresh from custom page viewer

- Add linked_upstream_id to CustomPage model with DB migration
- New POST /api/custom-pages/{pid}/refresh-auth endpoint extracts
  credentials from active remote browser and updates linked upstream
- PageViewer toolbar shows key icon button when page has linked upstream
- CustomPages form adds upstream dropdown for remote_browser pages
- Auth capture extracts New-Api-User from localStorage uid/user/self API
- Upstream client sends New-Api-User header in cookie auth mode
- Fix auth capture dialog: transparent background, field persistence,
  login URL defaults to base_url/login, focus on click for keyboard input
- Fix upstream test ASCII encoding with non-header characters validation

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
SmartUp Developer
2026-05-19 09:27:14 +08:00
parent 7cb0ff1608
commit 4c71148ff9
13 changed files with 462 additions and 53 deletions
+71 -3
View File
@@ -59,6 +59,32 @@ async def extract_session_storage(page: Any) -> dict[str, str]:
return {}
async def extract_new_api_user_id(page: Any) -> str:
try:
value = await page.evaluate("""
async () => {
const uid = localStorage.getItem('uid')
if (uid) return uid
const userRaw = localStorage.getItem('user')
if (userRaw) {
try {
const user = JSON.parse(userRaw)
if (user?.id) return String(user.id)
} catch {}
}
const response = await fetch('/api/user/self', { credentials: 'include' })
if (!response.ok) return ''
const payload = await response.json()
const data = payload?.data || payload
return data?.id ? String(data.id) : ''
}
""")
return str(value or "").strip()
except Exception as exc:
logger.debug("New-API user id extraction failed: %s", exc)
return ""
async def extract_request_headers(session: Any) -> list[dict[str, str]]:
"""Return Authorization / API-Key headers captured continuously by CDP.
@@ -83,7 +109,8 @@ async def extract_all(session: Any) -> dict[str, Any]:
local_storage = await extract_local_storage(page)
session_storage = await extract_session_storage(page)
auth_headers = await extract_request_headers(session)
candidates = _curate_candidates(cookies, local_storage, session_storage, auth_headers)
new_api_user = _find_new_api_user(local_storage, session_storage) or await extract_new_api_user_id(page)
candidates = _curate_candidates(cookies, local_storage, session_storage, auth_headers, new_api_user)
return {
"cookies": cookies,
@@ -99,6 +126,7 @@ def _curate_candidates(
local_storage: dict[str, str],
session_storage: dict[str, str],
auth_headers: list[dict[str, str]],
new_api_user: str = "",
) -> list[dict[str, Any]]:
"""Scan extracted data for likely credentials with confidence scoring."""
candidates: list[dict[str, Any]] = []
@@ -148,19 +176,59 @@ def _curate_candidates(
_add(candidates, "bearer_token", f"{store_name}.{key}", val, _preview(val),
f"{store_name}.{key} (API Key)", 90)
if not new_api_user:
new_api_user = _find_new_api_user(local_storage, session_storage)
# 3. Session cookies
for c in cookies:
cname = c["name"].lower()
if any(k in cname for k in SESSION_COOKIE_NAMES):
preview = _preview(c["value"])
cookie_val = f"{c['name']}={c['value']}"
confidence = 99 if cname == "session" else 85
extra = {"cookie_name": c["name"], "cookie_value": c["value"]}
if cname == "session" and new_api_user:
extra["new_api_user"] = new_api_user
_add(candidates, "cookie", f"cookie:{c['name']}", cookie_val, preview,
f"🍪 {c['name']} ({c['domain']})", 75,
extra={"cookie_name": c["name"], "cookie_value": c["value"]})
f"Cookie {c['name']} ({c['domain']})", confidence,
extra=extra)
candidates.sort(key=lambda item: (
0 if item.get("type") == "cookie" and item.get("cookie_name") == "session" else
1 if item.get("type") == "cookie" else
2,
-int(item.get("confidence") or 0),
))
return candidates
def _find_storage_value(*stores: dict[str, str], key: str) -> str:
for store in stores:
value = store.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
return ""
def _find_new_api_user(*stores: dict[str, str]) -> str:
uid = _find_storage_value(*stores, key="uid")
if uid:
return uid
user_raw = _find_storage_value(*stores, key="user")
if not user_raw:
return ""
try:
user = json.loads(user_raw)
except Exception:
return ""
if isinstance(user, dict):
for key in ("id", "user_id", "userId"):
value = user.get(key)
if value is not None:
return str(value).strip()
return ""
def _add(
candidates: list[dict[str, Any]],
ctype: str,