sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(api): routes_strategy全端点挂鉴权+文件名白名单——架构审计P0-2用户拍板限修: router级verify_token(原MVP裸奔=写.py+动态import执行面绑0.0.0.0公网直达); validate_py_name白名单^[A-Za-z0-9_-]+.py$堵os.path.join穿越/绝对路径逃逸(read/write/code-versions三口); 顺手收敛routes_portfolio/live两份分叉鉴权副本→routes.py正主(副本缺Bearer前缀时split越界500); 测试=无token 401+坏token 401+白名单parametrize+连坐两文件补帽 [vps] [no-doc]

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Dev
2026-10-01 09:51:39 +08:00
parent aa607d7e82
commit 2e30bb67ad
7 changed files with 148 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sanguo_api/routes_live.py
+3 -6
View File
@@ -11,10 +11,10 @@ from __future__ import annotations
import os
import time
from fastapi import APIRouter, Depends, Header, HTTPException
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from .auth import verify_token as verify_token_impl
from .routes import verify_token
router = APIRouter()
_db_path = {"path": None}
@@ -31,10 +31,7 @@ def set_db_path(p):
init_db(p) # app 启动建表(幂等)
async def verify_token(authorization: str | None = Header(None)):
if authorization is None or not authorization.startswith("Bearer "):
raise HTTPException(401, "Missing/invalid authorization")
return verify_token_impl(authorization.split(" ", 1)[1])
# verify_token 收敛(2026-10-01 P0-2 同笔): 本地分叉副本删除, 统一走 routes.py 正主。
class LiveCreateRequest(BaseModel):
sanguo_api/routes_portfolio.py
+4 -8
View File
@@ -11,21 +11,17 @@ from __future__ import annotations
import logging
from typing import Any
from fastapi import APIRouter, Depends, Header, HTTPException
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from .auth import verify_token as verify_token_impl
from .routes import get_orchestrator
from .routes import get_orchestrator, verify_token
from .validation import validate_portfolio_request
logger = logging.getLogger(__name__)
router = APIRouter()
async def verify_token(authorization: str | None = Header(None)):
if authorization is None or not authorization.startswith("Bearer "):
raise HTTPException(401, "Missing/invalid authorization")
return verify_token_impl(authorization.split(" ", 1)[1])
# verify_token 收敛(2026-10-01 P0-2 同笔): 本地分叉副本(缺 Bearer 前缀时
# split 越界 500 等)删除, 统一走 routes.py 正主(401 语义与全仓一致)。
class PortfolioBacktestRequest(BaseModel):
sanguo_api/routes_strategy.py
+16 -4
View File
@@ -1,17 +1,21 @@
"""策略管理路由：代码文件 CRUD + 实例 CRUD（spec §12 三层模型）。
鉴权说明：MVP 不挂 verify_token，便于本地联调与单测；敏感写操作（写文件）
已由 write_strategy_file 的 py_compile 语法门禁兜底。
鉴权说明（2026-10-01 P0-2 修正）：router 级 verify_token 全端点生效——
本组含「写 .py 文件 + 动态 import 执行」链, 曾是唯一无鉴权直达面
(绑定 0.0.0.0 + 公网域名可达); 文件名另有 validate_py_name 白名单
(防路径穿越/绝对路径逃逸), py_compile 语法门禁仅是第三道兜底。
"""
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from .routes import verify_token
from .strategy_registry import (
list_strategy_files, read_strategy_file, write_strategy_file,
validate_py_name,
)
from . import instance_store
router = APIRouter()
router = APIRouter(dependencies=[Depends(verify_token)])
class FileWriteRequest(BaseModel):
@@ -199,6 +203,10 @@ def get_code_versions(file: str):
"""该策略文件的全部发起快照（新→旧）。"""
from .code_versions import list_versions
try:
validate_py_name(file)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return {"file": file, "versions": list_versions(file)}
@@ -207,6 +215,10 @@ def get_code_version(file: str, h8: str):
"""读某版本全文（编辑器 diff 视图左栏）。"""
from .code_versions import read_version
try:
validate_py_name(file)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
code = read_version(file, h8)
if code is None:
raise HTTPException(status_code=404, detail="版本不存在")
sanguo_api/strategy_registry.py
+16
View File
@@ -12,6 +12,7 @@ import importlib
import logging
import os
import pkgutil
import re
logger = logging.getLogger(__name__)
@@ -174,8 +175,22 @@ def list_strategy_files() -> dict:
return {"files": _scan_self_owned()}
_PY_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+\.py$")
def validate_py_name(name: str) -> str:
"""策略文件名白名单(10-01 审计 P0-2): 仅单段 [A-Za-z0-9_-]+.py。
路径分隔符/绝对路径/盘符/点穿越全被字符集排除(os.path.join 的
第二参数带分隔符即逃逸 rel, 绝对名直接弃前缀——两层都被此正则堵死)。"""
if not _PY_NAME_RE.match(name or ""):
raise ValueError(f"非法策略文件名(仅允许 [A-Za-z0-9_-]+.py): {name!r}")
return name
def read_strategy_file(name: str) -> dict:
"""按文件名读单个策略文件全量（含 code）。找不到 → ValueError。"""
validate_py_name(name)
root = _repo_root()
for rel in SELF_OWNED_DIRS:
path = os.path.join(root, rel, name)
@@ -197,6 +212,7 @@ def read_strategy_file(name: str) -> dict:
def write_strategy_file(name: str, code: str) -> None:
"""写策略文件。py_compile 语法校验失败 → ValueError（拒绝坏代码落盘）。"""
validate_py_name(name)
import py_compile
import tempfile
with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as tf:
tests/api/test_code_versions.py
+5 -5
View File
@@ -95,11 +95,11 @@ def test_enriched_code_changed_flag(tmp_path, monkeypatch):
"params": {}}],
"start": "2024-01-01", "end": "",
}, headers={"Authorization": f"Bearer {token}"})
insts = c.get("/api/v1/strategy/instances/enriched").json()["instances"]
insts = c.get("/api/v1/strategy/instances/enriched", headers={"Authorization": f"Bearer {token}"}).json()["instances"]
assert insts[0]["code_changed"] in (False, None) # 刚发起必然一致/不可判
# 改当前文件哈希（模拟代码变更）
monkeypatch.setattr(CV, "current_hash", lambda f: "deadbeef")
insts = c.get("/api/v1/strategy/instances/enriched").json()["instances"]
insts = c.get("/api/v1/strategy/instances/enriched", headers={"Authorization": f"Bearer {token}"}).json()["instances"]
me = insts[0]
if me["running_accounts"]:
acc = me["running_accounts"][0]
@@ -115,8 +115,8 @@ def test_code_version_endpoints(tmp_path, monkeypatch):
return {"name": name, "code": "x = 1\n"}
monkeypatch.setattr("sanguo_api.strategy_registry.read_strategy_file", _FakeFile())
snap = CV.snapshot_code("demo.py")
lst = c.get("/api/v1/strategy/code-versions?file=demo.py").json()
lst = c.get("/api/v1/strategy/code-versions?file=demo.py", headers={"Authorization": f"Bearer {token}"}).json()
assert lst["versions"] and lst["versions"][0]["code_version"] == snap["code_version"]
one = c.get(f"/api/v1/strategy/code-versions/demo.py/{snap['code_version']}").json()
one = c.get(f"/api/v1/strategy/code-versions/demo.py/{snap['code_version']}", headers={"Authorization": f"Bearer {token}"}).json()
assert one["code"] == "x = 1\n"
assert c.get("/api/v1/strategy/code-versions/demo.py/00000000").status_code == 404
assert c.get("/api/v1/strategy/code-versions/demo.py/00000000", headers={"Authorization": f"Bearer {token}"}).status_code == 404
tests/api/test_instance_binding.py
+7 -7
View File
@@ -48,7 +48,7 @@ def test_auto_create_instance_on_launch(tmp_path, monkeypatch):
c, token = _setup(tmp_path, monkeypatch)
r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token))
assert r.status_code == 200
insts = c.get("/api/v1/strategy/instances").json()["instances"]
insts = c.get("/api/v1/strategy/instances", headers=_auth(token)).json()["instances"]
assert len(insts) == 1
iid = insts[0]["id"]
assert insts[0]["params"] == {"fast_window": 5}
@@ -93,7 +93,7 @@ def test_sync_params_to_running_accounts(tmp_path, monkeypatch):
}, headers=_auth(token))
# 档案改参 → 漂移出现
instance_store.update_instance(iid, {"params": {"fast_window": 20}})
enriched = c.get("/api/v1/strategy/instances/enriched").json()["instances"]
enriched = c.get("/api/v1/strategy/instances/enriched", headers=_auth(token)).json()["instances"]
me = [i for i in enriched if i["id"] == iid][0]
assert me["drift"] is True
# 同步 → 两个模拟账户参数齐刷
@@ -111,7 +111,7 @@ def test_delete_protection_with_running_account(tmp_path, monkeypatch):
c, token = _setup(tmp_path, monkeypatch)
r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token))
assert r.status_code == 200
iid = c.get("/api/v1/strategy/instances").json()["instances"][0]["id"]
iid = c.get("/api/v1/strategy/instances", headers=_auth(token)).json()["instances"][0]["id"]
resp = c.delete(f"/api/v1/strategy/instances/{iid}", headers=_auth(token))
assert resp.status_code == 409
assert "运行中" in resp.json()["detail"]
@@ -124,8 +124,8 @@ def test_enriched_and_overview(tmp_path, monkeypatch):
"""读时聚合（四格实走格 running）+ P1 全景（runs + 合并持仓归因）。"""
c, token = _setup(tmp_path, monkeypatch)
r = c.post("/api/v1/paper/create", json=_LIVE_BODY, headers=_auth(token))
iid = c.get("/api/v1/strategy/instances").json()["instances"][0]["id"]
enriched = c.get("/api/v1/strategy/instances/enriched").json()["instances"]
iid = c.get("/api/v1/strategy/instances", headers=_auth(token)).json()["instances"][0]["id"]
enriched = c.get("/api/v1/strategy/instances/enriched", headers=_auth(token)).json()["instances"]
me = [i for i in enriched if i["id"] == iid][0]
assert me["status"]["paper_live"] == "running"
assert len(me["running_accounts"]) == 1
@@ -137,7 +137,7 @@ def test_enriched_and_overview(tmp_path, monkeypatch):
save_positions(db, 1, "live",
{"600000": {"volume": 100, "frozen": 0, "avg_price": 10.5}},
date="2026-08-15")
ov = c.get(f"/api/v1/strategy/instances/{iid}/overview").json()
ov = c.get(f"/api/v1/strategy/instances/{iid}/overview", headers=_auth(token)).json()
assert ov["instance"]["name"]
assert len(ov["runs"]) == 1 and ov["runs"][0]["kind"] == "paper"
assert ov["positions"] and ov["positions"][0]["symbol"] == "600000"
@@ -157,7 +157,7 @@ def test_live_create_auto_instance(tmp_path, monkeypatch):
ldb = os.path.join(str(tmp_path), "live.db")
accs = list_accounts(ldb)
assert accs and accs[0].get("instance_id")
insts = c.get("/api/v1/strategy/instances").json()["instances"]
insts = c.get("/api/v1/strategy/instances", headers=_auth(token)).json()["instances"]
assert insts[0]["id"] == accs[0]["instance_id"]
tests/api/test_routes_strategy.py
+97 -23
View File
@@ -1,21 +1,31 @@
"""Tests for sanguo_api.routes_strategy（策略代码 + 实例 路由组）。
需 fastapi/httpx → 在 NAS docker 容器或装齐后端的 venv 跑（Mac venv310 无 fastapi）。
2026-10-01 P0-2: router 级 verify_token 生效——全部调用带 Bearer token,
并钉 401(无 token)与文件名白名单(路径穿越面)回归。
"""
import pytest
from fastapi.testclient import TestClient
from sanguo_api.app import create_app
from sanguo_api.auth import create_token, set_jwt_config
from sanguo_api import instance_store
from sanguo_api.strategy_registry import validate_py_name
def _client(monkeypatch, tmp_path):
monkeypatch.setattr(instance_store, "_STORE_PATH", str(tmp_path / "i.json"))
set_jwt_config(secret="t", expire_minutes=60)
app = create_app(db_path=str(tmp_path / "test.db"))
return TestClient(app)
return TestClient(app), create_token("admin")
def _auth(token):
return {"Authorization": f"Bearer {token}"}
def test_files_endpoint_shape(monkeypatch, tmp_path):
c = _client(monkeypatch, tmp_path)
r = c.get("/api/v1/strategy/files")
c, token = _client(monkeypatch, tmp_path)
r = c.get("/api/v1/strategy/files", headers=_auth(token))
assert r.status_code == 200
body = r.json()
assert "files" in body and isinstance(body["files"], list)
@@ -26,8 +36,8 @@ def test_files_endpoint_shape(monkeypatch, tmp_path):
def test_instance_crud_via_http(monkeypatch, tmp_path):
c = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/instances", json={
c, token = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/instances", headers=_auth(token), json={
"code_file": "double_ma.py", "name": "HTTP实例", "type": "cta",
"params": {"fast_window": 5}, "symbol_or_pool": "600519.SH",
"interval": "d", "match_session": "next_open",
@@ -35,55 +45,64 @@ def test_instance_crud_via_http(monkeypatch, tmp_path):
assert r.status_code == 200, r.text
new_id = r.json()["id"]
assert any(i["id"] == new_id for i in c.get("/api/v1/strategy/instances").json()["instances"])
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"]["name"] == "HTTP实例"
assert any(i["id"] == new_id for i in c.get(
"/api/v1/strategy/instances", headers=_auth(token)).json()["instances"])
assert c.get(f"/api/v1/strategy/instances/{new_id}",
headers=_auth(token)).json()["instance"]["name"] == "HTTP实例"
# update
upd = c.put(f"/api/v1/strategy/instances/{new_id}", json={"name": "改名", "params": {"fast_window": 20}})
upd = c.put(f"/api/v1/strategy/instances/{new_id}", headers=_auth(token),
json={"name": "改名", "params": {"fast_window": 20}})
assert upd.status_code == 200
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"]["name"] == "改名"
assert c.get(f"/api/v1/strategy/instances/{new_id}",
headers=_auth(token)).json()["instance"]["name"] == "改名"
# delete
assert c.delete(f"/api/v1/strategy/instances/{new_id}").status_code == 200
assert c.get(f"/api/v1/strategy/instances/{new_id}").json()["instance"] is None
assert c.delete(f"/api/v1/strategy/instances/{new_id}",
headers=_auth(token)).status_code == 200
assert c.get(f"/api/v1/strategy/instances/{new_id}",
headers=_auth(token)).json()["instance"] is None
def test_file_write_rejects_syntax_error(monkeypatch, tmp_path):
c = _client(monkeypatch, tmp_path)
c, token = _client(monkeypatch, tmp_path)
# 故意写语法错误的代码 → 400（py_compile 门禁）
r = c.post("/api/v1/strategy/file/__probe_bad.py", json={"code": "def (:"})
r = c.post("/api/v1/strategy/file/__probe_bad.py",
headers=_auth(token), json={"code": "def (:"})
assert r.status_code == 400
def test_file_read_returns_code(monkeypatch, tmp_path):
c = _client(monkeypatch, tmp_path)
files = c.get("/api/v1/strategy/files").json()["files"]
c, token = _client(monkeypatch, tmp_path)
files = c.get("/api/v1/strategy/files",
headers=_auth(token)).json()["files"]
if not files:
return
name = files[0]["name"]
r = c.get(f"/api/v1/strategy/file/{name}")
r = c.get(f"/api/v1/strategy/file/{name}", headers=_auth(token))
assert r.status_code == 200, r.text
body = r.json()
assert isinstance(body["code"], str) and body["code"] # 编辑器空白 bug 的回归测试
assert body["name"] == name
assert c.get("/api/v1/strategy/file/__no_such__.py").status_code == 404
assert c.get("/api/v1/strategy/file/__no_such__.py",
headers=_auth(token)).status_code == 404
# ===== 语法检查端点(编辑器「语法检查」按钮真接线)=====
def test_check_syntax_ok(monkeypatch, tmp_path):
c = _client(monkeypatch, tmp_path)
c, token = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/file/foo.py/check",
json={"code": "x = [1, 2, 3]\nprint(x)\n"})
headers=_auth(token), json={"code": "x = [1, 2, 3]\nprint(x)\n"})
assert r.status_code == 200
assert r.json() == {"ok": True}
def test_check_syntax_catches_missing_bracket(monkeypatch, tmp_path):
"""删掉一个中括号必须报错并给行号(2026-08-15 用户实况:删括号仍报通过)。"""
c = _client(monkeypatch, tmp_path)
c, token = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/file/foo.py/check",
json={"code": "x = [1, 2, 3\nprint(x)\n"})
headers=_auth(token), json={"code": "x = [1, 2, 3\nprint(x)\n"})
assert r.status_code == 200
body = r.json()
assert body["ok"] is False
@@ -93,7 +112,62 @@ def test_check_syntax_catches_missing_bracket(monkeypatch, tmp_path):
def test_check_syntax_empty_code_ok(monkeypatch, tmp_path):
"""空代码/纯注释可编译 → ok(编辑器允许半成品里写注释)。"""
c = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/file/foo.py/check", json={"code": "# 只有注释\n"})
c, token = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/file/foo.py/check",
headers=_auth(token), json={"code": "# 只有注释\n"})
assert r.status_code == 200
assert r.json()["ok"] is True
# ===== 2026-10-01 P0-2: 鉴权 + 文件名白名单 =====
def test_no_token_401(monkeypatch, tmp_path):
"""无 token 全端点 401(修前=无鉴权直达, 写 .py+动态执行面)。"""
c, _ = _client(monkeypatch, tmp_path)
assert c.get("/api/v1/strategy/files").status_code == 401
assert c.post("/api/v1/strategy/file/x.py",
json={"code": "x=1"}).status_code == 401
assert c.get("/api/v1/strategy/instances").status_code == 401
assert c.post("/api/v1/strategy/instances",
json={}).status_code == 401
# 坏 token 同拒
assert c.get("/api/v1/strategy/files",
headers={"Authorization": "Bearer bogus"}).status_code == 401
def test_bad_token_rejected_good_token_passes(monkeypatch, tmp_path):
c, token = _client(monkeypatch, tmp_path)
set_jwt_config(secret="other", expire_minutes=60) # 换密钥=旧 token 失效
assert c.get("/api/v1/strategy/files",
headers=_auth(token)).status_code == 401
set_jwt_config(secret="t", expire_minutes=60)
assert c.get("/api/v1/strategy/files",
headers=_auth(token)).status_code == 200
@pytest.mark.parametrize("bad", [
"../evil.py", "a/b.py", r"a\b.py", "/abs/x.py", r"C:\abs\x.py",
"no_ext", "double..py", ".py", "", "sp ace.py", "中文.py",
])
def test_validate_py_name_rejects_traversal_and_odd_names(bad):
with pytest.raises(ValueError):
validate_py_name(bad)
@pytest.mark.parametrize("good", ["double_ma.py", "__probe_bad.py",
"value_selection_ex.py", "x-1.py"])
def test_validate_py_name_accepts_normal_names(good):
assert validate_py_name(good) == good
def test_write_endpoint_rejects_bad_filename(monkeypatch, tmp_path):
"""HTTP 层: 无扩展名/白名单外名 → 400(路径穿越面, 修前 os.path.join 直拼)。"""
c, token = _client(monkeypatch, tmp_path)
r = c.post("/api/v1/strategy/file/noext",
headers=_auth(token), json={"code": "x=1"})
assert r.status_code == 400
assert "非法策略文件名" in r.json()["detail"]
# 版本列表端点同守卫(query 参数可带任意串)
r2 = c.get("/api/v1/strategy/code-versions",
headers=_auth(token), params={"file": "../evil.py"})
assert r2.status_code == 400