W3-d 分享链接(后端):只读、不用登录、必设到期、只显示定时出数的视图

为什么做(04 细案 C2,集成-I6):会话 cookie 是 SameSite=Lax,跨站的 iframe 里带不上,想把看板放进别的门户只能每人开账号且
必须同站;分享链接是跨站嵌入唯一走得通的办法。它把数据交出了授权范围,所以每一步往窄处取:
- view_shares(db/models_apps.ViewShare):只存 key 的 sha256 和开头 6 位;latest 跟着最近一次封存走,fixed 存建分享那一刻
  每个视图的运行 id、按建分享时的应用版本解析;到期必填;打开只计数
- POST / GET / DELETE /api/apps/{id}/shares:要应用的 manage,建要管理员开了 apps.share_links(缺省关);个人应用、没发布、
  归档的不能分享;这一页没有最近封存的视图不收;到期最长 90 天;key 只在建的时候交回一次(/s#<key>);建、撤销写审计
- GET /api/shares/page、/api/shares/report.html 是公开路由(中间件把 /api/shares/ 当公开前缀),只认请求头
  X-AgentLab-Share,地址里带的 key 不认;撤销、到期回 410 写明;开关关掉时已有的一并 403;应用归档、页面没了 410
- apps/resolve.resolve_page 加 share 旁路:只解析 latest_sealed 视图(fixed 用存下的运行,趋势也停在那一期),按需、实例、
  探索视图写明原因、从不发起运行(auto 的也不);流程组件和报告换成写明原因的说明块;页面缓存键另加分享的标识;out 另给
  各视图取到的运行和可下载的报告
- 下载报告只认这条分享此刻解析出来的运行,内部就是 export_report_html(同样遮过、照常记导出审计,导出人记成不冒充任何人的
  分享身份)
- 分享接口都不放给访问密钥(不在 KEY_ROUTES,门禁 403)

安全头:/s 用 AGENTLAB_EMBED_ORIGINS 和 no-referrer(第 1 波 web_security 预留的规则,这次在真实页面托管下测)。C3 同站门户
嵌整站只做配置和文档:写进 web_security 的模块说明和 deploy/agentlab.env 的「嵌入」一节,「分享链接还没有上线」那句改掉。
This commit is contained in:
ThearchyHelios
2026-10-16 16:30:42 +08:00
parent 86d2840e41
commit f759c5c82d
8 changed files with 882 additions and 15 deletions
+326 -5
View File
@@ -1,10 +1,331 @@
"""分享链接(docs/design/next-round/04-对外集成与触达.md C2)。
"""分享链接(docs/design/next-round/04-对外集成与触达.md C2,集成-I6):拿到链接、不用登录就能看应用的一页(只读,可以嵌进门户)。
读分享页的接口不用登录,只认请求头里的分享 key(中间件把 /api/shares/ 当公开前缀,第 3 波 W3-d 加);建、撤销分享在
/api/apps/{id}/shares 下,要应用的 manage 权限。第 3 波 W3-d 填。
为什么要有:会话 cookie 是 SameSite=Lax,跨站的 iframe 里带不上——想把一块看板放进别的门户,只能让每个看的人都开账号,
而且门户和 AgentLab 必须同站(同站门户嵌整站见 deploy/agentlab.env 的 AGENTLAB_FRAME_ANCESTORS,C3)。分享链接是跨站嵌入
唯一走得通的办法,所以做;但它等于把数据交出了授权范围,所以每一步都往窄处取:
- 管理员开关 apps.share_links,缺省关;关掉时已有的链接一并打不开(403 share_links_off),撤销照常能做;
- 只有应用的 manage 权限能建、能撤销;个人应用(personal.not_shareable)、没发布的、归档了的不能分享;
- 到期必填,最长 EXPIRES_MAX_DAYS 天;撤销、到期之后打开回 410,写明是撤销了还是到期了、什么时候;
- 只读:分享页只解析定时出数(latest_sealed)的视图,按需视图、实例、待办、发起按钮、记录列表换成写明原因的说明,从不
发起运行(apps/resolve.py 的 share 旁路);看到的就是使用者看到的那份遮罩视图(视图对所有人都遮);
- 出处不走在线的证据接口(那等于给匿名的人一个按运行读证据的身份):分享页给「下载报告」,GET /api/shares/report.html 只认
这条分享此刻解析出来的那几次运行,内部用的就是报告导出 export_report_html(同样遮过,文件自己离线点得开出处);
- key 放在链接的 # 后面(/s#<key>):打开页面时不发给服务器、不进代理日志和 Referer;读的接口只认请求头 X-AgentLab-Share,
地址里带的一概不认。库里只存 key 的 sha256 和开头一截;
- 建、撤销写审计(app.share_create / app.share_revoke);打开只计数(open_count、last_opened_at)。
接口:
- POST / GET /api/apps/{app_id}/shares、DELETE /api/apps/{app_id}/shares/{share_id}:登录即可进来,接口里要应用的 manage;
- GET /api/shares/page、GET /api/shares/report.html:公开路由(中间件把 /api/shares/ 当公开前缀,不用登录、不看会话),
只认分享 key。
都不放给访问密钥(不在 auth/keys.KEY_ROUTES 里,门禁一律 403):分享是给人看的页面,集成要拉数走视图数据接口。
安全头:分享页 /s 用嵌入白名单 AGENTLAB_EMBED_ORIGINS 和 Referrer-Policy: no-referrer(web_security.py);这一条分享没有允许
嵌入(embed_allowed 为假)时,页面在别的页面的 iframe 里不显示内容(服务端在 /s 上只能按全站的白名单发头:那一刻还不知道是
哪一条分享,key 不会发到服务器)。
"""
from __future__ import annotations
from fastapi import APIRouter
import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from typing import Any, Literal
router = APIRouter(prefix="/api/shares", tags=["shares"])
from fastapi import APIRouter, Depends, Header, Query
from pydantic import BaseModel
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.coded import CodedHTTPException
from app.api.deps import Principal, actor_of, require_role
from app.db.base import get_session, utcnow
from app.db.models_apps import App, AppVersion, ViewShare
router = APIRouter(tags=["shares"])
_manage = APIRouter(prefix="/api/apps")
_public = APIRouter(prefix="/api/shares")
#: 读接口认的请求头
SHARE_HEADER = "X-AgentLab-Share"
#: 到期最长多少天
EXPIRES_MAX_DAYS = 90
#: key 的随机字节数、库里存的开头几位
KEY_BYTES = 32
PREFIX_LEN = 6
#: 分享页的地址(key 接在 # 后面)
SHARE_PATH = "/s"
SHARE_FORMAT = "agentlab.share/1"
SHARE_LINKS_OFF = "share_links_off"
SHARE_KEY_MISSING = "share_key_missing"
SHARE_NOT_FOUND = "share_not_found"
SHARE_REVOKED = "share_revoked"
SHARE_EXPIRED = "share_expired"
SHARE_GONE = "share_gone"
SHARE_EXPIRES_INVALID = "share_expires_invalid"
SHARE_NO_SEALED_VIEW = "share_no_sealed_view"
SHARE_RUN_NOT_SHARED = "share_run_not_shared"
SHARE_ID_NOT_FOUND = "share_id_not_found"
LINKS_OFF_CREATE = "管理员没有打开分享链接:请管理员在「设置 → 偏好设置」的「应用」一节里打开「允许建分享链接」"
LINKS_OFF_OPEN = "管理员已关闭分享链接,这条链接暂时打不开"
def key_hash(key: str) -> str:
return hashlib.sha256(key.encode()).hexdigest()
def _when(moment: datetime | None) -> str:
if moment is None:
return ""
aware = moment if moment.tzinfo else moment.replace(tzinfo=timezone.utc)
return aware.astimezone().strftime("%Y-%m-%d %H:%M")
def _aware(moment: datetime) -> datetime:
return moment if moment.tzinfo else moment.replace(tzinfo=timezone.utc)
def state_of(row: ViewShare, now: datetime | None = None) -> str:
"""active 能打开;revoked 撤销了;expired 过了到期时间。"""
if row.revoked_at is not None:
return "revoked"
if _aware(row.expires_at) <= (now or utcnow()):
return "expired"
return "active"
def share_out(row: ViewShare, titles: dict[str, str] | None = None) -> dict[str, Any]:
"""一条分享的接口形状(typesApp.ts 的 ViewShareRow):key 不回(只在建的时候给一次),给开头一截认是哪一条。"""
iso = lambda m: _aware(m).isoformat() if m else None # noqa: E731
return {"id": row.id, "prefix": row.prefix, "app_id": row.app_id, "page_id": row.page_id,
"page_title": (titles or {}).get(row.page_id) or row.page_id, "mode": row.mode,
"app_version": row.app_version, "embed_allowed": row.embed_allowed, "created_by": row.created_by,
"created_by_name": row.created_by_name, "created_at": iso(row.created_at),
"expires_at": iso(row.expires_at), "revoked_at": iso(row.revoked_at), "revoked_by": row.revoked_by,
"open_count": row.open_count, "last_opened_at": iso(row.last_opened_at), "state": state_of(row)}
async def _enabled(session: AsyncSession) -> bool:
from app.api.settings import apps_settings
return bool((await apps_settings(session)).get("share_links"))
def _titles(spec: dict[str, Any] | None) -> dict[str, str]:
return {str(p.get("id")): str(p.get("title") or p.get("id")) for p in (spec or {}).get("pages") or []
if isinstance(p, dict) and p.get("id")}
# --------------------------------------------------------------------------
# 建、列、撤销(要应用的 manage)
# --------------------------------------------------------------------------
class ShareIn(BaseModel):
page_id: str
mode: Literal["latest", "fixed"] = "latest"
#: 到期时间,必填(不给回 422 share_expires_invalid);不带时区的按 UTC
expires_at: datetime | None = None
embed_allowed: bool = False
async def _managed(session: AsyncSession, app_id: str, x_actor: str | None) -> tuple[App, Principal]:
from app.api.apps import _app, _principal
principal = _principal(x_actor)
app, _ = await _app(session, app_id, principal, "manage")
return app, principal
@_manage.post("/{app_id}/shares", status_code=201)
async def create_share(app_id: str, payload: ShareIn, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
"""建一条分享链接:要应用的 manage、管理员开了 apps.share_links。key 只在这里交回一次(path 是 /s#<key>,url 是拼上
对外访问地址的完整链接,没设对外访问地址时为 null,界面按当前页面的地址拼)。"""
from app.api.apps import _current_version, _refuse_archived, _refuse_personal
from app.apps.audit import record
from app.apps.resolve import SharedView, share_runs, sealed_views
from app.apps.tree import page_of
from app.core import public_url
app, principal = await _managed(session, app_id, x_actor)
if not await _enabled(session):
raise CodedHTTPException(403, LINKS_OFF_CREATE, SHARE_LINKS_OFF)
_refuse_personal(app, "建分享链接")
_refuse_archived(app, "建分享链接")
version = await _current_version(session, app)
now = utcnow()
if payload.expires_at is None:
raise CodedHTTPException(422, f"分享链接必须设到期时间,最长 {EXPIRES_MAX_DAYS} 天", SHARE_EXPIRES_INVALID,
field="expires_at")
expires = _aware(payload.expires_at)
if expires <= now + timedelta(minutes=1):
raise CodedHTTPException(422, "到期时间要在现在之后", SHARE_EXPIRES_INVALID, field="expires_at")
# 留一小时的余量:界面按本机时间算「90 天后」,本机比服务端快一点不该被拒
if expires > now + timedelta(days=EXPIRES_MAX_DAYS, hours=1):
raise CodedHTTPException(422, f"分享链接最长 {EXPIRES_MAX_DAYS} 天到期,请选更早的到期时间",
SHARE_EXPIRES_INVALID, field="expires_at")
spec = version.spec or {}
if page_of(spec, payload.page_id) is None:
raise CodedHTTPException(404, f"应用的当前版本里没有页面「{payload.page_id}」", "app_page_not_found")
if not sealed_views(spec, payload.page_id):
raise CodedHTTPException(409, "这一页没有定时出数的视图(取最近一次封存运行的视图):分享链接只显示这种视图,"
"分享出去什么都看不到", SHARE_NO_SEALED_VIEW)
runs: dict[str, str | None] | None = None
if payload.mode == "fixed":
found = await share_runs(session, spec, payload.page_id, SharedView(mode="latest", key=""))
runs = {vid: (run.id if run is not None else None) for vid, run in found.items()}
if not any(runs.values()):
raise CodedHTTPException(409, "这一页定时出数的视图都还没有封存的运行,没有可以固定的数:先用「始终最新」,"
"或者等第一次出数之后再建", SHARE_NO_SEALED_VIEW)
key = secrets.token_urlsafe(KEY_BYTES)
row = ViewShare(key_hash=key_hash(key), prefix=key[:PREFIX_LEN], app_id=app.id, page_id=payload.page_id,
mode=payload.mode, app_version=version.version, runs=runs, embed_allowed=payload.embed_allowed,
created_by=actor_of(x_actor), created_by_name=principal.display_name or actor_of(x_actor),
created_at=now, expires_at=expires)
session.add(row)
await session.flush()
await record(session, action="app.share_create", subject_kind="view_share", subject_id=row.id,
actor=actor_of(x_actor), data={"app_id": app.id, "page_id": payload.page_id, "mode": payload.mode,
"app_version": version.version, "expires_at": expires.isoformat(),
"embed_allowed": payload.embed_allowed, "prefix": row.prefix})
await session.commit()
path = f"{SHARE_PATH}#{key}"
base = public_url.public_base_url()
return {**share_out(row, _titles(spec)), "key": key, "path": path, "url": f"{base}{path}" if base else None}
@_manage.get("/{app_id}/shares")
async def list_shares(app_id: str, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
"""这个应用的分享链接(新的在前)。enabled 是管理员开没开(关着时界面不给新建,已有的照列、能撤销);embed_origins 是部署
配没配允许嵌入分享页的站点(AGENTLAB_EMBED_ORIGINS);max_days 是到期最长几天。"""
from app.api.apps import _previous
from app.core.config import settings
app, _ = await _managed(session, app_id, x_actor)
rows = list((await session.execute(select(ViewShare).where(ViewShare.app_id == app.id)
.order_by(ViewShare.created_at.desc()))).scalars())
version = await _previous(session, app)
titles = _titles(version.spec if version is not None else None)
return {"enabled": await _enabled(session), "embed_origins": bool(settings.embed_origins),
"max_days": EXPIRES_MAX_DAYS, "items": [share_out(r, titles) for r in rows]}
@_manage.delete("/{app_id}/shares/{share_id}")
async def revoke_share(app_id: str, share_id: str, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
"""撤销:之后打开回 410「链接已撤销」。撤销过的再撤销照旧交回它(不重复记审计)。管理员关了开关也能撤销。"""
from app.apps.audit import record
app, _ = await _managed(session, app_id, x_actor)
row = await session.get(ViewShare, share_id)
if row is None or row.app_id != app.id:
raise CodedHTTPException(404, "分享链接不存在,可能已被删除", SHARE_ID_NOT_FOUND)
if row.revoked_at is None:
row.revoked_at, row.revoked_by = utcnow(), actor_of(x_actor)
await record(session, action="app.share_revoke", subject_kind="view_share", subject_id=row.id,
actor=actor_of(x_actor), data={"app_id": app.id, "page_id": row.page_id, "prefix": row.prefix})
await session.commit()
return share_out(row)
# --------------------------------------------------------------------------
# 打开(公开,只认请求头里的分享 key)
# --------------------------------------------------------------------------
def _share_principal(row: ViewShare) -> Principal:
"""分享身份:不是任何一个人。按使用者算(role=user),用户名为空:审计里不冒充谁,遮罩和使用者看到的一样。"""
from app.auth import current_mode
return Principal(username=None, role="user", display_name=f"分享链接 {row.prefix}", verified=False,
mode=current_mode(), via="share")
async def _opened(session: AsyncSession, key: str | None) -> tuple[ViewShare, App, AppVersion]:
"""按请求头里的 key 认出分享,核对开关、撤销、到期、应用还在不在,交回 (分享, 应用, 要解析的应用版本)。"""
from app.api.apps import _previous
if not key or not key.strip():
raise CodedHTTPException(400, "缺少分享 key:请用完整的分享链接打开(链接里 # 后面的那一段)", SHARE_KEY_MISSING)
row = (await session.execute(select(ViewShare).where(ViewShare.key_hash == key_hash(key.strip())))).scalar_one_or_none()
if row is None:
raise CodedHTTPException(404, "链接无效:可能复制得不完整,请向分享的人确认链接", SHARE_NOT_FOUND)
if not await _enabled(session):
raise CodedHTTPException(403, LINKS_OFF_OPEN, SHARE_LINKS_OFF)
state = state_of(row)
if state == "revoked":
raise CodedHTTPException(410, f"链接已撤销({_when(row.revoked_at)}):请向分享的人要新的链接", SHARE_REVOKED)
if state == "expired":
raise CodedHTTPException(410, f"链接已到期({_when(row.expires_at)}):请向分享的人要新的链接", SHARE_EXPIRED)
app = await session.get(App, row.app_id)
if app is None or app.status == "archived" or getattr(app, "scope", "shared") == "personal" \
or not app.published_version:
raise CodedHTTPException(410, "分享的应用已归档或不再发布,这条链接打不开了", SHARE_GONE)
version = None
if row.mode == "fixed" and row.app_version:
version = (await session.execute(select(AppVersion).where(
AppVersion.app_id == app.id, AppVersion.version == row.app_version))).scalar_one_or_none()
version = version or await _previous(session, app)
if version is None:
raise CodedHTTPException(410, "分享的应用已归档或不再发布,这条链接打不开了", SHARE_GONE)
return row, app, version
def _shared(row: ViewShare) -> Any:
from app.apps.resolve import SharedView
return SharedView(mode=row.mode, key=row.id, runs=dict(row.runs or {}) if row.mode == "fixed" else None)
@_public.get("/page")
async def share_page(session: AsyncSession = Depends(get_session),
share_key: str | None = Header(default=None, alias=SHARE_HEADER)) -> dict[str, Any]:
"""分享页(agentlab.share/1):{format, share: {页脚要的:谁分享、什么时候、什么时候到期、能不能嵌入、模式}, view:
agentlab.view/1(share 旁路解析), downloads: [{view, run_id, reports, as_of_text}](下载报告用)}。打开一次计一次数。"""
from app.apps.resolve import ViewRefused, resolve_page
row, app, version = await _opened(session, share_key)
try:
view = await resolve_page(session, app, page_id=row.page_id, principal=_share_principal(row), role=None,
version=version, share=_shared(row))
except ViewRefused as e:
if e.code == "app_page_not_found":
raise CodedHTTPException(410, "分享的页面在应用的当前版本里已经没有了,这条链接打不开了", SHARE_GONE) from e
raise CodedHTTPException(e.status_code, e.message, e.code or SHARE_GONE) from e
await session.execute(update(ViewShare).where(ViewShare.id == row.id).values(
open_count=ViewShare.open_count + 1, last_opened_at=utcnow()))
await session.commit()
downloads = (view.pop("share", None) or {}).get("runs") or []
return {"format": SHARE_FORMAT,
"share": {"mode": row.mode, "created_by": row.created_by_name or row.created_by,
"created_at": _aware(row.created_at).isoformat(), "expires_at": _aware(row.expires_at).isoformat(),
"embed_allowed": row.embed_allowed},
"view": view, "downloads": downloads}
@_public.get("/report.html")
async def share_report(run: str = Query(...), report: str | None = None, session: AsyncSession = Depends(get_session),
share_key: str | None = Header(default=None, alias=SHARE_HEADER)) -> Any:
"""下载报告:只认这条分享此刻解析出来的那几次运行(apps/resolve.share_runs),别的运行 id 一律 404。内部就是报告导出
(api/evidence.export_report_html:同样遮过,报告文件自己离线点得开出处;导出照常记 report.exported 审计,导出人记成
分享身份——用户名为空,不冒充谁)。"""
from app.api.deps import reset_principal, set_principal
from app.api.evidence import export_report_html
from app.apps.resolve import share_runs
row, _, version = await _opened(session, share_key)
found = await share_runs(session, version.spec or {}, row.page_id, _shared(row))
if run not in {r.id for r in found.values() if r is not None}:
raise CodedHTTPException(404, "这次运行不在这条分享链接里", SHARE_RUN_NOT_SHARED)
token = set_principal(_share_principal(row))
try:
return await export_report_html(run, report)
finally:
reset_principal(token)
router.include_router(_manage, dependencies=[require_role("user")])
router.include_router(_public)
+125 -4
View File
@@ -12,6 +12,14 @@
- instance:实例自己的那次运行。
印章(截至 / 口径 / 出具档位 / 来源)由这里算;定时任务改跑了新版本而应用还钉在旧版本这类错位照实写进 notes。
分享链接(docs/design/next-round/04-对外集成与触达.md C2,api/shares.py)是解析时的另一条旁路(resolve_page 的 share):
拿到链接、不用登录的人看到的就是使用者看到的那份遮罩视图,但只有定时出数的那部分——
- 只解析 latest_sealed 视图(fixed 模式取建分享时存下的那几次运行 id,SharedView.runs);按需、实例、个人应用的探索视图换成
写明原因的空态(SHARE_SEALED_ONLY),从不发起运行(按需视图不自动发起,因为它根本不按按需解析);
- 流程组件(入口按钮、表单、实例、进度、待办、会签、生效记录、记录列表)换成写明原因的说明块(SHARE_PLACEHOLDER),报告
组件也换成说明(正文和出处在「下载报告」里:分享页不开在线证据,那等于给匿名的人一个按运行读证据的身份);
- 印章照常算,界面上常显;页面缓存的键另加分享的标识。
"""
from __future__ import annotations
@@ -76,6 +84,36 @@ _PERIOD = re.compile(r"(period|week|month|quarter|year|date|day|期|周|月|季|
_DIGIT = re.compile(r"\d")
#: 分享链接里(share 旁路)不是定时出数的视图、流程组件、报告写什么
SHARE_SEALED_ONLY = "分享链接只显示定时出数的视图:这一块的数要打开应用才看得到"
SHARE_PLACEHOLDER = {
"entry_button": "分享链接是只读的,不能在这里发起",
"form": "分享链接是只读的,不能在这里换参数",
"instance_list": "分享链接不列实例:要看实例请打开应用",
"instance_header": "分享链接不显示实例:要看实例请打开应用",
"instance_timeline": "分享链接不显示实例进度:要看请打开应用",
"task_list": "分享链接不列待办:待办只给派给的人,请打开应用处理",
"task_form": "分享链接不显示待办:待办只给派给的人,请打开应用处理",
"signoffs": "分享链接不显示会签进度:要看请打开应用",
"action_status": "分享链接不显示写入外部系统的记录:要看请打开应用",
"records_list": "分享链接不列登记的记录:要看请打开应用",
"report": "报告正文和每个数的出处在「下载报告」里:下载的文件离线也能点开出处",
}
SHARE_FIXED_NEW_VIEW = "这个视图是建分享之后才加的:固定在当时那几次运行的分享链接里没有它"
SHARE_FIXED_EMPTY = "建分享时这个视图还没有封存的运行"
SHARE_FIXED_GONE = "建分享时固定的那次运行已经不在了"
@dataclass(frozen=True)
class SharedView:
"""分享链接的解析方式(api/shares.py 建好交进来)。mode:latest 始终取最近一次封存;fixed 取 runs 里存的运行
({视图 id: 运行 id,建分享时还没有封存运行的是 None})。key 进页面缓存的键(分享的 id)。"""
mode: str
key: str
runs: dict[str, str | None] | None = None
class ViewRefused(Exception):
def __init__(self, status_code: int, message: str, code: str | None = None) -> None:
super().__init__(message)
@@ -567,6 +605,8 @@ class _Page:
page: dict[str, Any] = field(default_factory=dict)
#: 蓝图的试运行(trial 旁路):草稿里所有视图来源都解析到这次运行,流程组件也对着它
trial: Run | None = None
#: 分享链接(share 旁路):只解析定时出数的视图,流程组件换成说明
share: SharedView | None = None
pins: dict[str, PinState] = field(default_factory=dict)
sources: dict[str, Source] = field(default_factory=dict)
problems: list[dict[str, Any]] = field(default_factory=list)
@@ -606,13 +646,16 @@ async def _backfilled(session: AsyncSession, app_id: str) -> int:
async def resolve_page(session: AsyncSession, app: Any, *, page_id: str, principal: Principal, role: str | None,
instance: Run | None = None, params: dict[str, Any] | None = None, draft: bool = False,
version: AppVersion | None = None, actor: str | None = None,
trial: Run | None = None) -> dict[str, Any]:
trial: Run | None = None, share: SharedView | None = None) -> dict[str, Any]:
"""解析一页,返回 agentlab.view/1。draft 为真时按草稿解析:不发起任何运行,视图取钉住版本已有的封存运行。
trial 是蓝图的一次试运行(docs/design/normal-user-apps.md N4,只和 draft 一起用):草稿里所有视图来源(最近封存、
按需、本实例)都解析到这次运行,流程组件(实例进度、待办)也对着它;印章的来源写「试运行」、trial 为真、说明里写
「这不是正式结果」;趋势只有这一个点。不改 agentlab.app/1 的视图来源,只是解析时的旁路。
share 是分享链接(见模块说明,只和发布版一起用:草稿、个人应用、试运行都不能分享):只解析定时出数的视图,流程组件和
报告换成写明原因的说明,从不发起运行;out 里另给 share.runs(各视图取到的运行、有哪几份报告可下载)。
个人应用(N7)没有应用版本:草稿就是在用的那一份,不按草稿预览时也解析它(version 为空)。作者隐藏了的组件(hidden)
不下发;按草稿解析(编辑状态)时照常解析、带上 hidden,界面画成半透明。"""
import copy
@@ -625,8 +668,11 @@ async def resolve_page(session: AsyncSession, app: Any, *, page_id: str, princip
if page is None:
raise ViewRefused(404, f"应用里没有页面「{page_id}」", "app_page_not_found")
trial = trial if draft else None
if share is not None and (draft or personal or version is None or instance is not None):
raise ViewRefused(409, "分享链接只解析已发布的共享应用", "personal.not_shareable" if personal else VIEW_REFUSED)
ctx = _Page(app=app, spec=spec, version=None if draft else version, principal=principal, role=role,
instance=instance or trial, params=params or {}, draft=draft, actor=actor, page=page, trial=trial)
instance=instance or trial, params=params or {}, draft=draft, actor=actor, page=page, trial=trial,
share=share)
ctx.pins = await load_pins(session, spec)
# 试运行的钉住项本来就没有版本(还没发布),不是「需要升级」
upgrade = needs_upgrade(ctx.pins) if trial is None else False
@@ -644,9 +690,10 @@ async def resolve_page(session: AsyncSession, app: Any, *, page_id: str, princip
tuple((a, p.workflow.published_version if p.workflow else None, p.retired)
for a, p in sorted(ctx.pins.items())),
role, principal.username, principal.role, await _mask_fingerprint(session),
await _backfilled(session, app.id))
await _backfilled(session, app.id), ("share", share.key) if share is not None else None)
# 分享链接里流程组件都换成了说明,页面上没有随人、随时变的东西:照样进缓存
cacheable = (not draft and instance is None and all(s.state in ("ok", "empty") for s in ctx.sources.values())
and not _dynamic(page))
and (share is not None or not _dynamic(page)))
if cacheable and key in _PAGE_CACHE:
_PAGE_CACHE.move_to_end(key)
return copy.deepcopy(_PAGE_CACHE[key])
@@ -673,6 +720,7 @@ async def resolve_page(session: AsyncSession, app: Any, *, page_id: str, princip
"problems": ctx.problems,
**({"trial": {"run_id": trial.id, "status": trial.status,
"status_text": STATUS_TEXT.get(trial.status, trial.status)}} if trial is not None else {}),
**({"share": {"mode": share.mode, "runs": _share_runs_out(ctx)}} if share is not None else {}),
}
if cacheable:
_PAGE_CACHE[key] = copy.deepcopy(out)
@@ -681,6 +729,49 @@ async def resolve_page(session: AsyncSession, app: Any, *, page_id: str, princip
return out
def _share_runs_out(ctx: _Page) -> list[dict[str, Any]]:
"""分享页上「下载报告」用:各视图取到的、成功封存的那次运行,带有哪几份没被改过的报告(节点 id)。同一次运行只列一次。"""
from app.api.evidence import _reports
out: list[dict[str, Any]] = []
seen: set[str] = set()
for vid, source in ctx.sources.items():
if source.state != "ok" or source.run is None or source.sealed is None or source.run.id in seen:
continue
seen.add(source.run.id)
reports = [r.node_id for r in _reports(source.sealed) if r.hash_ok is not False and r.doc is not None]
out.append({"view": vid, "run_id": source.run.id, "reports": reports,
"as_of_text": (source.stamp or {}).get("as_of_text")})
return out
async def share_runs(session: AsyncSession, spec: dict[str, Any], page_id: str, share: SharedView) -> dict[str, Run | None]:
"""这条分享链接现在解析得到的运行:{视图 id: 运行}(只有这一页上定时出数的视图,取不到的是 None)。下载报告只认这几次
(api/shares.py),和页面上 _source 的分享旁路同一个取法。"""
page = page_of(spec, page_id)
if page is None:
return {}
ctx = _Page(app=None, spec=spec, version=None, principal=Principal(username=None, role="user", via="share"),
role=None, instance=None, params={}, draft=False, actor=None, page=page, share=share)
ctx.pins = await load_pins(session, spec)
out: dict[str, Run | None] = {}
for vid in _views_of(page):
if (((spec.get("views") or {}).get(vid) or {}).get("source") or {}).get("kind") != "latest_sealed":
continue
source = await _source(session, ctx, vid)
out[vid] = source.run if source.state == "ok" else None
return out
def sealed_views(spec: dict[str, Any], page_id: str) -> list[str]:
"""这一页上取最近封存运行(定时出数)的视图:能不能建分享链接看它(一个都没有,分享出去什么都看不到)。"""
page = page_of(spec, page_id)
if page is None:
return []
return [vid for vid in _views_of(page)
if (((spec.get("views") or {}).get(vid) or {}).get("source") or {}).get("kind") == "latest_sealed"]
_DYNAMIC = frozenset({"instance_list", "task_list", "instance_header", "instance_timeline", "task_form", "signoffs",
"action_status", "entry_button", "records_list"})
@@ -724,6 +815,12 @@ async def _source(session: AsyncSession, ctx: _Page, vid: str) -> Source:
return Source(id=vid, kind="", state="error", message=f"视图「{vid}」不存在")
if ctx.trial is not None:
return await _trial_source(session, ctx, vid, kind, src)
if ctx.share is not None:
if kind != "latest_sealed":
# 分享链接只显示定时出数的视图:按需视图不发起、实例和探索视图不取
return Source(id=vid, kind=kind, state="empty", message=SHARE_SEALED_ONLY)
if ctx.share.mode == "fixed":
return await _fixed_source(session, ctx, vid, src)
state = ctx.pins.get(src.get("pin")) if kind != "instance" else None
run: Run | None = None
if kind == "instance":
@@ -755,6 +852,21 @@ async def _source(session: AsyncSession, ctx: _Page, vid: str) -> Source:
return source
async def _fixed_source(session: AsyncSession, ctx: _Page, vid: str, src: dict[str, Any]) -> Source:
"""fixed 模式的分享链接:取建分享时这个视图的那次运行(还得是成功封存的正式运行),之后封存的新运行不换上来。"""
assert ctx.share is not None
runs = ctx.share.runs or {}
state = ctx.pins.get(src.get("pin"))
if vid not in runs:
return Source(id=vid, kind="latest_sealed", state="empty", pin=state, message=SHARE_FIXED_NEW_VIEW)
if runs[vid] is None:
return Source(id=vid, kind="latest_sealed", state="empty", pin=state, message=SHARE_FIXED_EMPTY)
run = await session.get(Run, runs[vid])
if run is None or run.status != "succeeded" or run.run_class != "formal":
return Source(id=vid, kind="latest_sealed", state="empty", pin=state, message=SHARE_FIXED_GONE)
return Source(id=vid, kind="latest_sealed", state="ok", run=run, pin=state, origin=src.get("from"))
async def _exploratory_source(session: AsyncSession, ctx: _Page, vid: str, state: PinState) -> Source:
"""个人应用的视图(docs/design/normal-user-apps.md N7):这个应用、这个拥有者、这张图下最近一次成功封存的刷新
(blueprints/personal.exploratory_runs)。只在个人应用里合法(应用检查报 app.view_source_personal):共享应用里写了的
@@ -867,6 +979,10 @@ async def _node(session: AsyncSession, ctx: _Page, comp: dict[str, Any]) -> dict
return None
if comp.get("visible") and not await _visible(session, ctx, comp):
return None
if ctx.share is not None and (reason := SHARE_PLACEHOLDER.get(str(comp.get("type")))):
# 分享链接是只读的:流程组件、报告换成一块写明原因的说明(前端照说明块画),不取实例、待办、报告正文
return {"id": cid, "type": "notice", "props": {"text": reason, "tone": "info"},
"share_placeholder": comp.get("type")}
node: dict[str, Any] = {"id": cid, "type": comp.get("type"), "props": dict(comp.get("props") or {}),
**({"hidden": True} if comp.get("hidden") else {})}
entry = BY_NAME.get(str(comp.get("type")))
@@ -1082,6 +1198,11 @@ async def _series(session: AsyncSession, ctx: _Page, source: Source, cid: str, r
runs = await trend_runs(session, app_id, pin.pin.get("workflow_id"), pin.pin.get("version"),
pin.pin.get("graph_hash"), origin, n)
segments.append((int(pin.pin.get("version")), str(pin.pin.get("graph_hash")), runs))
if ctx.share is not None and ctx.share.mode == "fixed" and source.run is not None:
# 固定在建分享那一刻的分享链接:趋势也停在那一期,之后封存的不画进来
cutoff = period_of(source.run)
segments = [(v, h, [r for r in rows if cutoff is None or (period_of(r) or cutoff) <= cutoff])
for v, h, rows in segments]
series: list[dict[str, Any]] = []
unit = ""
for version, _, rows in segments:
+5 -2
View File
@@ -7,7 +7,8 @@
和加这一层之前一模一样。
开登录:
- 会话 cookie → Principal(认证过的用户名、角色、组);没有会话是匿名(username=None),X-Actor 一律不看;
- /api 下除了登录本身、健康检查、入站钩子,匿名一律 401。各路由的角色要求另由 main.py 的路由级依赖管,这里兜底:
- /api 下除了登录本身、健康检查、入站钩子、分享链接的读接口(/api/shares/,只认请求头里的分享 key,api/shares.py),匿名
一律 401。各路由的角色要求另由 main.py 的路由级依赖管,这里兜底:
哪个路由忘了挂依赖,也不会对没登录的人敞开;
- 接口说明(FastAPI 自带的 /docs、/redoc、/openapi.json)和 /api 一样拦:匿名 401,不到搭建者 403。它们是 FastAPI 自己
挂的路由,挂不上路由级依赖,所以角色也在这里判。以前只拦 /api/ 开头的,开了登录接口全貌也对匿名的人敞开;
@@ -38,6 +39,8 @@ _PUBLIC = frozenset({"/api/health", "/api/auth/me", "/api/auth/login", "/api/aut
"/api/auth/oidc/callback", "/api/auth/jwks", "/api/auth/bootstrap"})
#: 只认签名、不走会话的入站钩子(也不查 CSRF:调用方是外部系统,没有浏览器的 Origin)
_HOOKS = "/api/hooks/"
#: 分享链接的读接口(04 细案 C2):不用登录,接口里只认请求头 X-AgentLab-Share,会话有没有都不看。只有 GET:非 GET 照常查 CSRF
_SHARES = "/api/shares/"
#: FastAPI 自带的接口说明(main.py 里 FastAPI() 的缺省地址):开登录时要搭建者
_DOCS = ("/docs", "/redoc", "/openapi.json")
_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
@@ -153,7 +156,7 @@ class AuthMiddleware:
await send({"type": "websocket.close", "code": 1008})
return
if (scope["type"] == "http" and is_api and not principal.username and path not in _PUBLIC
and not path.startswith(_HOOKS)):
and not path.startswith(_HOOKS) and not path.startswith(_SHARES)):
await _reply(send, 401, "请先登录", AUTH_REQUIRED)
return
if scope["type"] == "http" and _is_docs(path) and (refused := refusal(principal, "builder")) is not None:
+34
View File
@@ -146,3 +146,37 @@ class ApiReceipt(Base):
status: Mapped[str] = mapped_column(String(20))
detail: Mapped[str | None] = mapped_column(Text, default=None)
received_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
class ViewShare(Base):
"""分享链接(docs/design/next-round/04-对外集成与触达.md C2,集成-I6):拿到链接、不用登录就能看应用的一页(只读,
可以嵌进门户)。
链接写成 /s#<分享 key>:key 在 # 后面,打开页面时不发给服务器、不进访问日志和 Referer,页面读出来再放进请求头
X-AgentLab-Share(api/shares.py)。库里只存 key 的 sha256(key_hash)和开头一截(prefix,认是哪一条),同访问密钥。
mode:latest 始终取最近一次封存;fixed 取建分享那一刻每个视图的运行(runs:{视图 id: 运行 id},还没有封存运行的是
null),解析建分享时的应用版本(app_version)。到期必填、最长 90 天;撤销、到期之后打开回 410 并写明。建、撤销写审计;
打开只计数(open_count、last_opened_at),不一次写一行。embed_allowed 为假时分享页不肯在别的页面的 iframe 里显示(服务端
在 /s 上只能按全站的嵌入白名单 AGENTLAB_EMBED_ORIGINS 发头:打开页面那一刻还不知道是哪一条分享)。"""
__tablename__ = "view_shares"
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
key_hash: Mapped[str] = mapped_column(String(64), unique=True)
prefix: Mapped[str] = mapped_column(String(16))
app_id: Mapped[str] = mapped_column(ForeignKey("apps.id", ondelete="CASCADE"), index=True)
page_id: Mapped[str] = mapped_column(String(100))
mode: Mapped[str] = mapped_column(String(20), default="latest") # latest | fixed
app_version: Mapped[int | None] = mapped_column(Integer, default=None)
runs: Mapped[dict[str, Any] | None] = mapped_column(default=None)
embed_allowed: Mapped[bool] = mapped_column(default=False)
created_by: Mapped[str | None] = mapped_column(String(100), default=None)
#: 建分享的人当时的显示名:分享页页脚写「由 X 分享」
created_by_name: Mapped[str | None] = mapped_column(String(100), default=None)
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
expires_at: Mapped[datetime] = mapped_column(UTCDateTime)
revoked_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
revoked_by: Mapped[str | None] = mapped_column(String(100), default=None)
open_count: Mapped[int] = mapped_column(Integer, default=0)
last_opened_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
+10 -1
View File
@@ -9,7 +9,7 @@
- Referrer-Policy: strict-origin-when-cross-origin(分享页 /s 用 no-referrer:分享的 key 在地址的 # 后面,本来不随
Referer 走,这里再收一道);
- Content-Security-Policy: frame-ancestors。缺省只有 'self';AGENTLAB_FRAME_ANCESTORS 列出允许嵌整站的同站门户;分享页
/s 只认 AGENTLAB_EMBED_ORIGINS(分享链接第 3 波才加,路由在那时挂,头的规则先在这里定好);
/s 只认 AGENTLAB_EMBED_ORIGINS(分享链接,api/shares.py;页面是前端不进外壳的那一页,由 mount_web 照常托管);
- 只有 frame-ancestors 是 'self' 时另加 X-Frame-Options: SAMEORIGIN,给不认 CSP 的老浏览器。XFO 写不了白名单,有白名单
时只发 CSP:两个都发的话,只认 XFO 的老浏览器会把白名单里的门户也拦掉(认 CSP 的浏览器以 frame-ancestors 为准)。
接口自己设过的同名头(文件下载的 nosniff)不覆盖、不重复。
@@ -19,6 +19,15 @@
index.html 里的内联脚本算 sha256(script_policy),只在页面这一个响应上发;观察一个版本没有误报,再改成强制执行。没配
上报地址:违规只在浏览器的控制台里看得到。接口返回的 HTML(导出的报告、可验证报告)文件里自带 meta CSP,不发它,也不动。
嵌入的两种办法(04 细案 C3,只做配置和文档):
- 同站门户嵌整站:门户和 AgentLab 在同一个可注册域名下(portal.corp.example 和 agentlab.corp.example)时,把门户的源写进
AGENTLAB_FRAME_ANCESTORS。会话 cookie 是 SameSite=Lax,同站的 iframe 里会随请求带上:看的人照常登录、单点登录照常,权限
和直接打开一样。不另做嵌入票据;
- 跨站嵌:嵌整站明确不支持——跨站的 iframe 里带不上 Lax 的会话 cookie,登录不了;改成 SameSite=None 会削弱整站防跨站请求
伪造的那一层,换来的嵌入在 Safari、Firefox 上默认仍然被拦。跨站要嵌,用分享链接(只读、必设到期、可撤销,/s 只认
AGENTLAB_EMBED_ORIGINS)。
deploy/agentlab.env 的「嵌入」一节写的是同一套。
HSTS 交给终结 https 的反向代理:这里不知道对外是不是 https、要不要连子域名一起管,发错了浏览器会按 max-age 记很久,
撤不回来。
"""
+373
View File
@@ -0,0 +1,373 @@
"""分享链接(api/shares.py + apps/resolve.py 的 share 旁路,docs/design/next-round/04-对外集成与触达.md C2)。
守的是:
- 管理员开关缺省关,关着建不了、已有的打不开;只有应用的 manage 能建;个人应用、没发布的、这一页没有定时出数的视图都建不了;
- 到期必填,最长 90 天;
- 读的接口不用登录,只认请求头 X-AgentLab-Share:地址里带 key 不认;库里只有 key 的哈希和开头一截;
- 只解析定时出数(最近封存)的视图:按需视图写明原因、从不发起运行(声明了 auto 的也不);流程组件、报告换成写明原因的说明;
- fixed 模式钉在建分享时的那几次运行上(趋势也停在那一期),latest 跟着最新的封存走;
- 下载报告只认这条分享解析出来的运行,内部就是报告导出(遮罩值写「已遮罩」);
- 撤销、到期回 410 并写明;建、撤销写审计,打开只计数;
- 访问密钥一律调不了分享的接口;
- /s 上的安全头:嵌入白名单用 AGENTLAB_EMBED_ORIGINS、Referrer-Policy: no-referrer。
应用、工作流、人名都是虚构的。
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from datetime import timedelta
from pathlib import Path
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, func, select
from app.core.config import BACKEND_DIR
from app.db.base import SessionLocal, utcnow
from app.db.models import AuditEvent, Run, Setting
from app.db.models_apps import ViewShare
from tests.fakes import auth_env
from tests.test_access_key_use import KPI, formal, published, wait, workflow
from tests.test_access_key_use import world # noqa: F401 - 开登录、建管理员、收尾清掉应用
HEADER = "X-AgentLab-Share"
@pytest.fixture(autouse=True)
async def switch(world): # noqa: F811
await set_switch(True)
yield
async with SessionLocal() as s:
await s.execute(delete(ViewShare))
await s.execute(delete(Setting).where(Setting.key == "apps"))
await s.commit()
async def set_switch(on: bool) -> None:
async with SessionLocal() as s:
row = await s.get(Setting, "apps")
if row is None:
s.add(Setting(key="apps", value={"share_links": on}))
else:
row.value = {**row.value, "share_links": on}
await s.commit()
def anonymous(**headers: str) -> AsyncClient:
"""不带会话 cookie、不带 CSRF 头的客户端:拿到链接的人就是这样打开的。"""
from app.main import app
return AsyncClient(transport=ASGITransport(app=app), base_url="http://test", headers=headers)
def board(wf: str, h: str, *, body: list | None = None, views: dict | None = None) -> dict:
return {"format": "agentlab.app/1", "name": "周报看板",
"pins": [{"alias": "main", "workflow_id": wf, "version": 1, "graph_hash": h, "use": "view"}],
"views": views or {"latest": {"source": {"kind": "latest_sealed", "pin": "main", "from": "formal"}}},
"pages": [{"id": "home", "title": "首页", "body": body or [
{"id": "gmv", "type": "metric", "props": {"label": "销售额"},
"bind": {"view": "latest", "metric": "card.gmv"}},
{"id": "trend", "type": "series", "bind": {"view": "latest", "series": "card.gmv"}}]}]}
def soon(days: float = 30) -> str:
return (utcnow() + timedelta(days=days)).isoformat()
async def share(app_id: str, **body) -> dict:
admin = await auth_env.login("root")
r = await admin.post(f"/api/apps/{app_id}/shares", json={"page_id": "home", "expires_at": soon(), **body})
assert r.status_code == 201, r.text
return r.json()
# --------------------------------------------------------------------------
# 建:开关、权限、到期
# --------------------------------------------------------------------------
async def test_switch_and_manage_are_both_required(world): # noqa: F811
wf, h = await workflow(KPI)
await auth_env.make_user("pia", "user")
app_id = await published(world, board(wf, h), [("user", "pia", "observe")])
admin = await auth_env.login("root")
await set_switch(False)
r = await admin.post(f"/api/apps/{app_id}/shares", json={"page_id": "home", "expires_at": soon()})
assert r.status_code == 403 and r.json()["code"] == "share_links_off" and "允许建分享链接" in r.json()["detail"]
listed = (await admin.get(f"/api/apps/{app_id}/shares")).json()
assert listed["enabled"] is False and listed["items"] == [] and listed["max_days"] == 90
await set_switch(True)
pia = await auth_env.login("pia")
r = await pia.post(f"/api/apps/{app_id}/shares", json={"page_id": "home", "expires_at": soon()})
assert r.status_code == 403 and r.json()["code"] == "app_forbidden", "查看全部实例也不够,要管理权限"
made = await share(app_id)
assert made["path"] == f"/s#{made['key']}" and made["state"] == "active" and made["mode"] == "latest"
async with SessionLocal() as s:
row = await s.get(ViewShare, made["id"])
assert row.key_hash != made["key"] and made["key"] not in json.dumps(
{c.name: str(getattr(row, c.name)) for c in ViewShare.__table__.columns}), "库里没有 key 的明文"
assert row.prefix == made["key"][:6] and made["prefix"] == row.prefix
listed = (await admin.get(f"/api/apps/{app_id}/shares")).json()
assert [i["id"] for i in listed["items"]] == [made["id"]] and "key" not in listed["items"][0]
assert listed["items"][0]["page_title"] == "首页"
@pytest.mark.parametrize("expires, fragment", [(None, "必须设到期时间"), (-1, "现在之后"), (91, "最长 90 天")])
async def test_expiry_is_required_and_at_most_90_days(world, expires, fragment): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
admin = await auth_env.login("root")
body = {"page_id": "home", **({"expires_at": soon(expires)} if expires is not None else {})}
r = await admin.post(f"/api/apps/{app_id}/shares", json=body)
assert r.status_code == 422 and r.json()["code"] == "share_expires_invalid" and fragment in r.json()["detail"]
ok = await admin.post(f"/api/apps/{app_id}/shares", json={"page_id": "home", "expires_at": soon(90)})
assert ok.status_code == 201
async def test_pages_without_sealed_views_and_personal_apps_cannot_be_shared(world): # noqa: F811
wf, h = await workflow(KPI)
on_demand = board(wf, h, views={"latest": {"source": {"kind": "on_demand", "pin": "main"}}},
body=[{"id": "gmv", "type": "metric", "bind": {"view": "latest", "metric": "card.gmv"}}])
app_id = await published(world, on_demand, [])
admin = await auth_env.login("root")
r = await admin.post(f"/api/apps/{app_id}/shares", json={"page_id": "home", "expires_at": soon()})
assert r.status_code == 409 and r.json()["code"] == "share_no_sealed_view"
r = await admin.post(f"/api/apps/{app_id}/shares", json={"page_id": "nope", "expires_at": soon()})
assert r.status_code == 404
draft_only = (await admin.post("/api/apps", json={"name": "草稿", "spec": board(wf, h)})).json()["id"]
world["apps"].append(draft_only)
r = await admin.post(f"/api/apps/{draft_only}/shares", json={"page_id": "home", "expires_at": soon()})
assert r.status_code == 409 and r.json()["code"] == "app_unpublished"
from app.db.models import App
async with SessionLocal() as s:
(await s.get(App, draft_only)).scope = "personal"
await s.commit()
r = await admin.post(f"/api/apps/{draft_only}/shares", json={"page_id": "home", "expires_at": soon()})
assert r.json()["code"] in ("personal.not_shareable", "app_not_found"), r.text
# --------------------------------------------------------------------------
# 打开
# --------------------------------------------------------------------------
async def test_the_page_is_public_but_only_trusts_the_header(world): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
run = await formal(wf)
made = await share(app_id)
async with anonymous() as c:
missing = await c.get("/api/shares/page", params={"key": made["key"]})
assert missing.status_code == 400 and missing.json()["code"] == "share_key_missing", "地址里带的 key 不认"
wrong = await c.get("/api/shares/page", headers={HEADER: made["key"][:-2] + "zz"})
assert wrong.status_code == 404 and wrong.json()["code"] == "share_not_found"
r = await c.get("/api/shares/page", headers={HEADER: made["key"]})
assert r.status_code == 200, r.text
body = r.json()
assert body["format"] == "agentlab.share/1"
assert body["share"]["created_by"] == "root" and body["share"]["mode"] == "latest"
view = body["view"]
assert view["format"] == "agentlab.view/1" and view["sources"]["latest"]["run_id"] == run.id
gmv = view["tree"][0]["data"]
assert gmv["state"] == "ok" and "4,321.5" in gmv["text"], "看到的就是使用者看到的那份"
assert view["sources"]["latest"]["stamp"]["sealed"] is True, "印章照常算"
assert body["downloads"] == [{"view": "latest", "run_id": run.id, "reports": [],
"as_of_text": view["sources"]["latest"]["stamp"]["as_of_text"]}]
async def test_only_sealed_views_and_flow_components_become_placeholders(world): # noqa: F811
wf, h = await workflow(KPI)
spec = board(wf, h, views={
"latest": {"source": {"kind": "latest_sealed", "pin": "main", "from": "formal"}},
"ask": {"source": {"kind": "on_demand", "pin": "main", "auto": True}}}, body=[
{"id": "gmv", "type": "metric", "bind": {"view": "latest", "metric": "card.gmv"}},
{"id": "asked", "type": "metric", "bind": {"view": "ask", "metric": "card.gmv"}},
{"id": "todo", "type": "task_list"},
{"id": "mine", "type": "instance_list"}])
app_id = await published(world, spec, [])
await formal(wf)
made = await share(app_id)
async with SessionLocal() as s:
before = await s.scalar(select(func.count(Run.id)))
async with anonymous(**{HEADER: made["key"]}) as c:
view = (await c.get("/api/shares/page")).json()["view"]
async with SessionLocal() as s:
assert await s.scalar(select(func.count(Run.id))) == before, "打开分享页从不发起运行(声明了 auto 的按需视图也不)"
by = {n["id"]: n for n in view["tree"]}
assert by["gmv"]["data"]["state"] == "ok"
assert by["asked"]["data"]["state"] == "empty" and "只显示定时出数的视图" in by["asked"]["data"]["message"]
for cid, kind in (("todo", "task_list"), ("mine", "instance_list")):
assert by[cid]["type"] == "notice" and by[cid]["share_placeholder"] == kind and by[cid]["props"]["text"]
assert "data" not in by[cid], "不取实例、待办"
async def test_fixed_mode_stays_on_the_runs_it_was_created_with(world): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
first = await formal(wf, "W1")
fixed = await share(app_id, mode="fixed")
latest = await share(app_id)
second = await formal(wf, "W2")
async with anonymous(**{HEADER: fixed["key"]}) as c:
pinned = (await c.get("/api/shares/page")).json()
other = await c.get("/api/shares/report.html", params={"run": second.id})
assert other.status_code == 404 and other.json()["code"] == "share_run_not_shared"
async with anonymous(**{HEADER: latest["key"]}) as c:
moving = (await c.get("/api/shares/page")).json()
assert pinned["view"]["sources"]["latest"]["run_id"] == first.id
points = pinned["view"]["tree"][1]["data"]["series"][-1]["points"]
assert [p["run_id"] for p in points] == [first.id], "趋势也停在建分享的那一期"
assert moving["view"]["sources"]["latest"]["run_id"] == second.id
async def test_report_download_only_for_the_shares_runs_and_masked(world, monkeypatch, tmp_path): # noqa: F811
import sqlite3
from app.data.engine import engines
from app.engine.launch import launch_run
from tests import test_report_masked_metrics as vipmod
path = tmp_path / "vip.db"
db = sqlite3.connect(path)
db.executescript("CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, phone TEXT);")
db.executemany("INSERT INTO customers VALUES (?, ?, ?)", vipmod.CUSTOMERS)
db.commit()
db.close()
source_id = await vipmod.set_source(path, {"mask_columns": ["phone"]})
await engines.invalidate(source_id)
world["sources"].append("vip")
vipmod.script(monkeypatch, vipmod.CLEAN)
wf, h = await workflow(vipmod.graph(on_violation="flag"))
spec = board(wf, h, body=[{"id": "n", "type": "metric", "bind": {"view": "latest", "metric": "card.n_cust"}},
{"id": "doc", "type": "report", "bind": {"view": "latest", "report": "write"}}])
app_id = await published(world, spec, [])
async with SessionLocal() as s:
launched = await launch_run(s, workflow_id=wf, run_class="formal", version=1, input_payload={},
actor="编辑", origin_kind="launcher")
run = await wait(launched.id)
assert run.status == "succeeded", run.error
made = await share(app_id)
kpi_wf, _ = await workflow(KPI)
stranger = await formal(kpi_wf)
async with anonymous(**{HEADER: made["key"]}) as c:
body = (await c.get("/api/shares/page")).json()
by = {n["id"]: n for n in body["view"]["tree"]}
assert by["doc"]["type"] == "notice" and "下载报告" in by["doc"]["props"]["text"], "报告正文不在线打开"
[download] = body["downloads"]
assert download["run_id"] == run.id and download["reports"] == ["write"]
report = await c.get("/api/shares/report.html", params={"run": run.id})
assert report.status_code == 200, report.text
assert "attachment" in report.headers["content-disposition"]
assert vipmod.leaks(report.text) == [], "报告里没有遮罩列的原值"
refused = await c.get("/api/shares/report.html", params={"run": stranger.id})
assert refused.status_code == 404 and refused.json()["code"] == "share_run_not_shared"
no_header = await anonymous().get("/api/shares/report.html", params={"run": run.id})
assert no_header.status_code == 400
async with SessionLocal() as s:
exported = (await s.execute(select(AuditEvent).where(AuditEvent.action == "report.exported",
AuditEvent.subject_id == run.id))).scalars().all()
assert [(e.actor, e.verified) for e in exported] == [(None, False)], "导出记审计,导出人不冒充谁"
async def test_revoked_and_expired_links_say_so_with_410(world): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
await formal(wf)
gone = await share(app_id)
old = await share(app_id)
admin = await auth_env.login("root")
r = await admin.delete(f"/api/apps/{app_id}/shares/{gone['id']}")
assert r.status_code == 200 and r.json()["state"] == "revoked" and r.json()["revoked_by"] == "root"
async with SessionLocal() as s:
(await s.get(ViewShare, old["id"])).expires_at = utcnow() - timedelta(minutes=5)
await s.commit()
async with anonymous() as c:
r = await c.get("/api/shares/page", headers={HEADER: gone["key"]})
assert r.status_code == 410 and r.json()["code"] == "share_revoked" and "链接已撤销" in r.json()["detail"]
r = await c.get("/api/shares/page", headers={HEADER: old["key"]})
assert r.status_code == 410 and r.json()["code"] == "share_expired" and "链接已到期" in r.json()["detail"]
live = await share(app_id)
await set_switch(False)
async with anonymous(**{HEADER: live["key"]}) as c:
r = await c.get("/api/shares/page")
assert r.status_code == 403 and r.json()["code"] == "share_links_off", "管理员关掉开关,已有的一并打不开"
assert (await admin.delete(f"/api/apps/{app_id}/shares/{live['id']}")).status_code == 200, "关着也能撤销"
async def test_create_and_revoke_are_audited_and_opening_only_counts(world): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
await formal(wf)
made = await share(app_id, embed_allowed=True)
async with anonymous(**{HEADER: made["key"]}) as c:
for _ in range(3):
assert (await c.get("/api/shares/page")).status_code == 200
admin = await auth_env.login("root")
await admin.delete(f"/api/apps/{app_id}/shares/{made['id']}")
await admin.delete(f"/api/apps/{app_id}/shares/{made['id']}")
async with SessionLocal() as s:
row = await s.get(ViewShare, made["id"])
actions = (await s.execute(select(AuditEvent.action, AuditEvent.actor).where(
AuditEvent.subject_id == made["id"]).order_by(AuditEvent.at))).all()
assert row.open_count == 3 and row.last_opened_at is not None and row.embed_allowed is True
assert actions == [("app.share_create", "root"), ("app.share_revoke", "root")], "打开不记审计;撤销两次只记一次"
async def test_access_keys_cannot_reach_share_routes(world): # noqa: F811
wf, h = await workflow(KPI)
app_id = await published(world, board(wf, h), [])
made = await share(app_id)
svc = await auth_env.make_service()
_, secret = await auth_env.make_key(svc)
c = auth_env.keyed(secret, **{HEADER: made["key"]})
for method, path in (("GET", "/api/shares/page"), ("GET", f"/api/apps/{app_id}/shares"),
("POST", f"/api/apps/{app_id}/shares")):
r = await c.request(method, path, json={"page_id": "home", "expires_at": soon()} if method == "POST" else None)
assert r.status_code == 403 and r.json()["code"] == "key_route_forbidden", (method, path)
# --------------------------------------------------------------------------
# /s 的安全头
# --------------------------------------------------------------------------
_PROBE = r"""
import asyncio, json, httpx
from app.main import app
async def main():
t = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=t, base_url="http://t") as c:
out = {}
for p in ("/s", "/apps/x", "/api/shares/page"):
r = await c.get(p)
out[p] = {"status": r.status_code, **{k: r.headers.get(k) for k in (
"content-security-policy", "x-frame-options", "referrer-policy")}}
print(json.dumps(out))
asyncio.run(main())
"""
def test_share_page_uses_the_embed_whitelist_and_no_referrer(tmp_path: Path) -> None:
dist = tmp_path / "dist"
(dist / "assets").mkdir(parents=True)
(dist / "index.html").write_text("<!doctype html><html><body><div id=root></div></body></html>")
env = {**os.environ, "AGENTLAB_DATA_DIR": str(tmp_path / "data"), "AGENTLAB_WEB_DIST": str(dist),
"AGENTLAB_SECRET_KEY": "test-only-not-a-real-key", "AGENTLAB_EMBED_ORIGINS": "https://bi.example.org",
"AGENTLAB_FRAME_ANCESTORS": ""}
proc = subprocess.run([sys.executable, "-c", _PROBE], cwd=BACKEND_DIR, env=env, capture_output=True, text=True,
timeout=120)
assert proc.returncode == 0, proc.stderr[-2000:]
out = json.loads(proc.stdout.strip().splitlines()[-1])
assert out["/s"]["status"] == 200, "分享页由页面托管给出(不进外壳由前端定)"
assert out["/s"]["content-security-policy"] == "frame-ancestors 'self' https://bi.example.org"
assert out["/s"]["x-frame-options"] is None and out["/s"]["referrer-policy"] == "no-referrer"
assert out["/apps/x"]["content-security-policy"] == "frame-ancestors 'self'", "别的页面不跟着放开嵌入"
assert out["/apps/x"]["x-frame-options"] == "SAMEORIGIN"
assert out["/api/shares/page"]["status"] == 400, "关登录时也只认请求头"
+1 -1
View File
@@ -4,7 +4,7 @@
- 页面、/api 的响应(连同登录检查自己回的 401)都带 nosniff、Referrer-Policy 和 CSP frame-ancestors;
- frame-ancestors 缺省 'self',只有 'self' 时另加 X-Frame-Options: SAMEORIGIN;配了 AGENTLAB_FRAME_ANCESTORS 白名单时
以 CSP 为准、不发 XFO(XFO 写不了白名单,两个都发的话老浏览器按 XFO 把白名单里的门户也拦了);
- 分享页 /s 用 AGENTLAB_EMBED_ORIGINS、Referrer-Policy: no-referrer(路由第 3 波才加,头的规则先守住);
- 分享页 /s 用 AGENTLAB_EMBED_ORIGINS、Referrer-Policy: no-referrer(真实页面托管下的头另见 test_shares.py);
- 严格的 script-src 只发 Content-Security-Policy-Report-Only,只在页面(index.html)上发,哈希就是构建产物里内联
脚本的 sha256;接口返回的 HTML(导出的报告自带 meta CSP)不带它;
- 接口自己设过的同名头不覆盖、不重复;WebSocket 不经过它;
+8 -2
View File
@@ -38,10 +38,16 @@ BIND_ADDR=0.0.0.0
# X-Frame-Options: SAMEORIGIN),别的站点嵌进去会被浏览器拦下,防点击劫持。HSTS 不在这里加,交给终结 https 的反向代理。
# 同站的门户(和 AgentLab 在同一个可注册域名下,比如 portal.example.internal 和 agentlab.example.internal)要嵌整站时,
# 把门户的地址写在下面,逗号分隔,只写 协议://主机[:端口],主机可以写 *.example.internal。填了之后只发
# frame-ancestors、不再发 X-Frame-Options。登录用的 cookie 只随同站的请求带上,跨站的门户嵌整站登录不了。
# frame-ancestors、不再发 X-Frame-Options。同站时登录用的会话 cookie(SameSite=Lax)会随 iframe 里的请求带上,看的人在
# 门户里照常登录、单点登录照常;嵌进去的就是整站,权限和直接打开一样。
# 跨站的门户(不在同一个可注册域名下)嵌整站不支持:浏览器不把会话 cookie 带进跨站的 iframe,登录不了;为此把 cookie
# 改成 SameSite=None 会削弱整站防跨站请求伪造的那一层,Safari、Firefox 默认还是会拦。跨站要嵌,请用下面的分享链接。
# 写错了(带路径、没写协议、只写 *)服务启动时报错。
# AGENTLAB_FRAME_ANCESTORS=https://portal.example.internal
# 分享页(/s)允许被哪些站点嵌,写法同上。分享页只认这一项、不认上面那一项。分享链接还没有上线,现在填了不起作用
# 分享页(/s,分享链接打开的只读页面)允许被哪些站点嵌,写法同上。分享页只认这一项、不认上面那一项;不填时只有 AgentLab
# 自己的页面能嵌。分享链接要管理员在「设置 → 偏好设置」的「应用」一节里打开「允许建分享链接」,再由管理应用的人在
# 应用编辑的「岗位与授权」里建(必设到期、最长 90 天,可撤销);建的时候没勾「允许嵌入」的那一条,在别的页面的 iframe 里
# 不显示内容。分享页另带 Referrer-Policy: no-referrer
# AGENTLAB_EMBED_ORIGINS=https://bi.example.internal
# ---- 数据 ----