jlocal/jingrow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化基于session cookie重构用户访问权限的实现逻辑

Browse Source
This commit is contained in:
jingrow 2026-01-25 00:24:04 +08:00
parent 0d94d18f74
commit 242adf013b
2 changed files with 16 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
apps/jingrow/jingrow/core/adapters/api_adapter.py
View File

@ -12,6 +12,7 @@ import datetime
import requests
import pytz
from jingrow.config import Config
from jingrow.utils.auth import get_request_session_cookie
class ApiAdapter:
"""API 适配器 - 通过 API 调用 Jingrow SaaS 版"""
@ -21,33 +22,26 @@ class ApiAdapter:
self.api_key = Config.jingrow_api_key
self.api_secret = Config.jingrow_api_secret
def _get_headers(self, session_cookie: Optional[str] = None):
def _get_headers(self):
"""
获取请求头
认证优先级：
1. 优先使用参数传入的 session_cookie
2. 如果没有，从 ContextVar 获取（由 SessionMiddleware 设置）
3. 如果都没有，使用 API Key（系统级权限）
1. 优先使用 ContextVar（由 SessionMiddleware 设置）
2. 如果没有 session cookie，使用 API Key（系统级权限）
"""
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
# 1. 优先使用参数传入的 session cookie
if not session_cookie:
# 2. 从 ContextVar 获取（SessionMiddleware 已设置）
try:
from jingrow.utils.auth import get_request_session_cookie
session_cookie = get_request_session_cookie()
except Exception:
pass
# 从 ContextVar 获取 session cookie
session_cookie = get_request_session_cookie()
# 优先使用 session cookie（用户权限）
if session_cookie:
headers['Cookie'] = f'sid={session_cookie}'
# 3. 如果没有 session cookie，使用 API Key（系统级权限）
# 如果没有 session cookie，使用 API Key（系统级权限）
elif self.api_key and self.api_secret:
headers['Authorization'] = f'token {self.api_key}:{self.api_secret}'
@ -156,10 +150,10 @@ class ApiAdapter:
except Exception as e:
return None
def get_agent_detail(self, name: str, session_cookie: Optional[str] = None) -> Optional[Dict[str, Any]]:
def get_agent_detail(self, name: str) -> Optional[Dict[str, Any]]:
try:
api_url = f"{self.api_url}/api/action/jingrow.ai.utils.jlocal.get_local_ai_agent_detail"
headers = self._get_headers(session_cookie=session_cookie)
headers = self._get_headers()
payload = {"name": name}
response = requests.post(api_url, headers=headers, json=payload, timeout=15)
if response.status_code == 200:
@ -174,7 +168,7 @@ class ApiAdapter:
except Exception as e:
return None
def get_pg(self, pagetype: str, name: str, session_cookie: Optional[str] = None) -> Dict[str, Any]:
def get_pg(self, pagetype: str, name: str) -> Dict[str, Any]:
try:
api_url = f"{self.api_url}/api/data/{pagetype}/{name}"
@ -187,28 +181,18 @@ class ApiAdapter:
}
if self.api_key and self.api_secret:
headers['Authorization'] = f'token {self.api_key}:{self.api_secret}'
print(f"[ADAPTER] get_pg: pagetype=PageType, forcing API Key auth")
else:
headers = self._get_headers(session_cookie=session_cookie)
print(f"[ADAPTER] get_pg: pagetype={pagetype}, name={name}")
print(f"[ADAPTER] headers keys: {list(headers.keys())}")
if 'Cookie' in headers:
print(f"[ADAPTER] Using session cookie auth")
elif 'Authorization' in headers:
print(f"[ADAPTER] Using API Key auth")
headers = self._get_headers()
response = requests.get(api_url, headers=headers, timeout=15)
print(f"[ADAPTER] Response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
if isinstance(data, dict) and 'data' in data:
return {'success': True, 'data': data['data']}
return {'success': True, 'data': data}
else:
print(f"[ADAPTER] Response error: {response.text[:200]}")
return {'success': False, 'error': f"HTTP {response.status_code}: {response.text}"}
except Exception as e:
print(f"[ADAPTER] Exception: {e}")
return {'success': False, 'error': f"获取记录异常: {str(e)}"}
def create_pg(self, pagetype: str, payload: Dict[str, Any]) -> Dict[str, Any]:

25
apps/jingrow/jingrow/main.py
View File

@ -8,7 +8,6 @@ Jingrow 应用入口
import logging
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from contextlib import asynccontextmanager
from jingrow.utils.router_auto_register import include_routers_from_package
from jingrow.services.scheduler import start_scheduler, stop_scheduler
@ -19,23 +18,6 @@ from jingrow.utils.auth import set_request_session_cookie
logger = logging.getLogger("uvicorn.error")
class SessionMiddleware(BaseHTTPMiddleware):
"""
Session 中间件：自动提取请求中的 session cookie 并存储到 ContextVar
这样在整个请求生命周期中都可以访问
"""
async def dispatch(self, request: Request, call_next):
# 提取 session cookie
session_cookie = request.cookies.get('sid')
# 存储到 ContextVar（异步安全）
set_request_session_cookie(session_cookie)
# 继续处理请求
response = await call_next(request)
return response
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
@ -83,8 +65,11 @@ def create_app():
allow_headers=["*"],
)
# 添加 Session 中间件（必须在 CORS 之后）
app.add_middleware(SessionMiddleware)
# 添加 Session 中间件：提取 session cookie 并存储到 ContextVar
@app.middleware("http")
async def session_middleware(request: Request, call_next):
set_request_session_cookie(request.cookies.get('sid'))
return await call_next(request)
# 自动注册 Jingrow 框架的静态路由（无前缀）
include_routers_from_package(app, "jingrow.api")