Compare commits

..
Author SHA1 Message Date
chichuan 43b1936b65 Merge pull request #950 from DingTalk-Real-AI/codex/changelog-v1.0.58-beta.3
docs: seal v1.0.58-beta.3 changelog
2026-08-11 17:50:03 +08:00
chichuan dbe47d58fb docs: seal v1.0.58-beta.3 changelog 2026-08-11 17:45:44 +08:00
github-actions[bot] ec7593dabb Merge pull request #936 from wxianfeng/feature/aone85277391-event-runtime-token-handoff
fix(event): securely hand off runtime token to detached bus
2026-08-11 09:32:24 +00:00
chichuan 1df4cc95a6 Merge branch 'main' into feature/aone85277391-event-runtime-token-handoff 2026-08-11 17:15:15 +08:00
github-actions[bot] d052c104d9 Merge pull request #948 from cywan1998/docs/sync-calendar-skill-mono-multi
docs(skills): sync calendar reference between mono and multi layouts
2026-08-11 08:39:55 +00:00
fengbai fdf3e8cc3b docs(skills): sync calendar reference between mono and multi layouts 2026-08-11 16:21:20 +08:00
github-actions[bot] 0fb332c3f3 Merge pull request #934 from DingTalk-Real-AI/feat/eval-dispatch
ci: add /eval PR comment dispatch for internal MCP evaluation
2026-08-11 16:15:19 +08:00
chichuan 27b16b190f Merge branch 'main' into feat/eval-dispatch 2026-08-11 15:47:54 +08:00
github-actions[bot] de1e1aaf6c Merge pull request #913 from DingTalk-Real-AI/codex/fix-im-reliability
fix(chat): harden IM search, card updates, and message workflows
2026-08-11 15:47:27 +08:00
chichuan 20d1f7c614 feat(eval-dispatch): optional sha= for own-PR dispatch; structural cases ref validation
- /eval on one's own PR may omit sha=: the guard auto-pins the
  dispatch-time head (commenter == PR author leaves no third-party
  swap window); dispatching another author's PR still requires the
  explicit reviewed SHA (keeps the P1-2 TOCTOU remedy where the
  threat lives)
- cases= is now validated structurally per git check-ref-format
  semantics (leading/trailing//double slashes, '..', dot-leading
  components, .lock suffixes) and rejects '-'-leading values to
  prevent git fetch option injection (review P2)
2026-08-11 15:46:16 +08:00
chichuan 233e0359e4 chore(eval-dispatch): seed allowlist with 53 internal contributors 2026-08-11 15:45:33 +08:00
chichuan ad6837d694 feat(eval-dispatch): allowlist tier for self-service PR evaluation
Users listed in .github/eval-allowlist.txt (default branch, PR-reviewed)
may dispatch /eval for their own PRs only; write/maintain/admin retain
dispatch for any PR. Fail-closed on permission API 404/network errors.
2026-08-11 15:45:33 +08:00
chichuan 8619d90119 Merge remote-tracking branch 'origin/main' into feat/eval-dispatch 2026-08-11 14:16:33 +08:00
chichuan 42e764a7a8 fix(ci): harden eval dispatch authorization 2026-08-11 13:57:42 +08:00
wxianfeng bab7c8879b Merge remote-tracking branch 'upstream/main' into feature/aone85277391-event-runtime-token-handoff 2026-08-11 11:48:08 +08:00
wxianfeng 82b17ced32 Merge upstream/main into feature/aone85277391-event-runtime-token-handoff 2026-08-11 11:37:37 +08:00
wxianfeng 63dbf98cdf test(event): cover runtime token rejection on Windows to #85277391 2026-08-11 11:22:48 +08:00
chichuan 69cef74e1d Merge branch 'main' into feat/eval-dispatch 2026-08-11 11:10:05 +08:00
wxianfeng 5004ed8ae6 fix(event): securely hand off runtime token to detached bus to #85277391 2026-08-10 15:59:11 +08:00
chichuan a5672152a7 ci: add /eval comment dispatch workflow for internal MCP evaluation (Aone JSON trigger contract) 2026-08-10 15:34:01 +08:00
chichuan 2d38abe681 feat(ci): PR 评论 /eval 触发内网 MCP 评测的 dispatch workflow
- issue_comment 触发,author_association ∈ OWNER/MEMBER/COLLABORATOR 门控
- 不 checkout、不执行 PR 代码;触发通道与凭证全部经 secrets 注入
- scripts/ci/eval_comment_parse.py 解析 /eval <products> [cases=<ref>](10 个单测)
2026-08-10 15:34:01 +08:00
41 changed files with 5765 additions and 220 deletions
+61
View File
@@ -0,0 +1,61 @@
# /eval 自助触发允许名单
#
# 名单内的 GitHub 登录名可对【自己创建的 PR】触发 /eval 评测;
# 对任意 PR 触发仍需仓库 write/maintain/admin 权限(维护者背书)。
# 授权读取的始终是默认分支上的本文件,PR 无法修改自身授权。
#
# 变更本文件必须走 PR 评审。每行一个 GitHub login,# 开头为注释。
aftersss
notable-open
EdgarWang0925
ayunya
yutongshe
qingyang1014
caiTriumph
xlb1130
Anonymity-0
FuShu-Yang
guimingyue
AlwaysLee
TaoJikun
zengyoulingzyl-stack
liyuan333
huangyoo
lifeihong
nitonitori
cywan1998
gangwn
junlonghuo2
aqruan
Freda0909
ShawnWhite777
PeterGuy326
abucraft
pengzhihan47-star
rainyak8
gongrongyun
huangyuanzhuo-coder
ybcstudy
bigqy
liwang-ai
meng93
wxianfeng
Patrick-Star-CN
rossluo28-hz
dxy704330469
gtezg30062
Neige-Premaire
zhuoyu20
avicii-chen
typefield
Haofeng0705
Huwenjiao
liuzeyang
maoqxxmm
FloralTide
lingyun9833
dxb121
C0922
xiaoji121
H3java
+139
View File
@@ -0,0 +1,139 @@
name: PR Eval Dispatch
# `/eval <products> [sha=<full-head-sha>] [cases=<ref>]` PR 评论 → 触发内网评测流水线,报告由内网 bot 回贴。
# 本 workflow 只在默认分支上下文运行,不 checkout、不执行 PR 代码。
# 审核 SHA 规则:评测他人 PR 必须显式携带 sha=(审阅背书凭据,验证
# 其恰为当前 open head);评测自己创建的 PR 可省略,自动钉住派发时刻
# 的当前 head(作者自背书,无第三方偷换窗口);内网 CI 另以
# FETCH_HEAD 校验兜底派发后的变更。
# 授权两级:仓库 write/maintain/admin 可派发任意 PR;默认分支
# .github/eval-allowlist.txt 名单内的用户仅可派发自己创建的 PR。
# 触发通道与凭证全部经 secrets 注入,文件内不出现任何内网信息。
on:
issue_comment:
types:
- created
permissions: {}
concurrency:
group: eval-dispatch-${{ github.event.issue.number }}
cancel-in-progress: false
jobs:
dispatch:
name: Dispatch internal evaluation
if: >-
github.event.issue.pull_request &&
startsWith(github.event.comment.body, '/eval')
runs-on: ubuntu-latest
timeout-minutes: 5
permissions:
contents: read
issues: write
pull-requests: read
steps:
- name: Check out default branch tooling
uses: actions/checkout@v4
- name: Verify commenter dispatch authorization
env:
GH_TOKEN: ${{ github.token }}
COMMENTER: ${{ github.event.comment.user.login }}
PR_AUTHOR: ${{ github.event.issue.user.login }}
EVAL_ALLOWLIST_PATH: .github/eval-allowlist.txt
run: |
# 不用 --fail:非协作者查权限返回 404 错误体,交由 guard 走名单分支;硬网络错误降级为空对象同样 fail-closed
permission_json="$(curl --silent --show-error \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/collaborators/${COMMENTER}/permission")" || permission_json='{}'
printf '%s' "$permission_json" | python3 scripts/ci/eval_dispatch_guard.py permission
- name: Parse /eval command
id: parse
continue-on-error: true
env:
COMMENT_BODY: ${{ github.event.comment.body }}
run: python3 scripts/ci/eval_comment_parse.py
- name: Reply usage on parse failure
if: steps.parse.outcome == 'failure'
env:
GH_TOKEN: ${{ github.token }}
PR_NUMBER: ${{ github.event.issue.number }}
PARSE_ERROR: ${{ steps.parse.outputs.error }}
run: |
body="❌ /eval 命令解析失败:${PARSE_ERROR}"
jq -n --arg body "$body" '{body: $body}' | curl --fail --silent --show-error \
-X POST \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments" > /dev/null
exit 1
- name: Verify reviewed PR head
id: pr
env:
GH_TOKEN: ${{ github.token }}
PR_NUMBER: ${{ github.event.issue.number }}
EXPECTED_PR_NUMBER: ${{ github.event.issue.number }}
REVIEWED_SHA: ${{ steps.parse.outputs.reviewed_sha }}
COMMENTER: ${{ github.event.comment.user.login }}
run: |
pr_json="$(curl --fail --silent --show-error \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}")"
printf '%s' "$pr_json" \
| python3 scripts/ci/eval_dispatch_guard.py head \
>> "$GITHUB_OUTPUT"
- name: Trigger internal evaluation pipeline
env:
EVAL_TRIGGER_TOKEN: ${{ secrets.EVAL_TRIGGER_TOKEN }}
EVAL_TRIGGER_URL: ${{ secrets.EVAL_TRIGGER_URL }}
PR_NUMBER: ${{ github.event.issue.number }}
PR_HEAD_SHA: ${{ steps.pr.outputs.head_sha }}
PRODUCTS: ${{ steps.parse.outputs.products }}
CASES_REF: ${{ steps.parse.outputs.cases_ref }}
run: |
if [ -z "$EVAL_TRIGGER_TOKEN" ] || [ -z "$EVAL_TRIGGER_URL" ]; then
echo "EVAL_TRIGGER_URL / EVAL_TRIGGER_TOKEN not configured; cannot dispatch." >&2
exit 1
fi
jq -n \
--arg pr "$PR_NUMBER" \
--arg sha "$PR_HEAD_SHA" \
--arg products "$PRODUCTS" \
--arg cases "$CASES_REF" \
'{branch: "main", params: {pr_number: $pr, pr_head_sha: $sha, products: $products, cases_ref: $cases}}' \
| curl --fail --silent --show-error \
-X POST \
-H "private-token: ${EVAL_TRIGGER_TOKEN}" \
-H "Content-Type: application/json" \
--data @- \
"$EVAL_TRIGGER_URL"
echo "Internal evaluation dispatched."
- name: Acknowledge on PR
env:
GH_TOKEN: ${{ github.token }}
PR_NUMBER: ${{ github.event.issue.number }}
PR_HEAD_SHA: ${{ steps.pr.outputs.head_sha }}
PRODUCTS: ${{ steps.parse.outputs.products }}
CASES_REF: ${{ steps.parse.outputs.cases_ref }}
run: |
cases_note=""
if [ -n "$CASES_REF" ]; then
cases_note=",用例版本 \`${CASES_REF}\`"
fi
body="🛰️ /eval 已受理:产品集 \`${PRODUCTS}\`${cases_note},评测对象 \`${PR_HEAD_SHA}\`。内网评测流水线运行结束后将由 bot 回贴报告(首行为基线对比头条)。"
jq -n --arg body "$body" '{body: $body}' | curl --fail --silent --show-error \
-X POST \
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
--data @- \
"https://api.github.com/repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments" > /dev/null
+30
View File
@@ -6,6 +6,8 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
## [Unreleased]
## [1.0.58-beta.3] - 2026-08-11
### Added
- **Aitable workflow execution and history** — adds `dws aitable workflow run` for confirmed asynchronous execution of scheduled or record-triggered workflows, plus `dws aitable workflow history` for status-, time-, and page-filtered execution records. The commands map directly to `aitable/run_workflow` and `aitable/get_flow_record_list`, validate trigger-specific arguments locally, and document the `executionId` / `instanceId` correlation.
@@ -13,6 +15,9 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
`--at-open-dingtalk-ids` and `--at-all` for group cards, passing mention
targets to the initial card-creation request and prepending its returned
`atTag` to the automatic streaming update.
- **Personal OA approval events** — personal event consumers now support task
creation, completion, redirection, instance start, termination, and
completion events, with typed output and matching usage documentation.
### Fixed
@@ -22,6 +27,18 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
a successful local write. The result includes the saved path and byte size;
document exports additionally report the node, requested format, job/task
ID, and final status.
- **IM search and card-write safety** — conversation-scoped search now fails
closed when the target cannot be verified, and streaming-card updates require
business evidence rather than a transport-only success response.
- **Document shortcut reliability** — document write, readback verification,
pagination, template/version discovery, export, media, and local-file
workflows now preserve compatibility while rejecting ambiguous write results.
- **Event runtime-token handoff** — personal `event consume`, `status`,
`stop`, and `+listen-im` honor the root `--token` without falling back to a
stale OAuth profile. Detached buses negotiate an owner-only, memory-only IPC
credential channel; tokens are never placed in child argv, environment,
profiles, logs, or run-state files.
### Changed
- **Minutes `permission apply --policy` type** — `--policy` is now declared as
@@ -31,6 +48,15 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
- **Minutes skill references** — document `permission apply` in both Minutes
skill references: list it in the command trees, describe its policy values and
how it differs from `permission add`, and add its intent routing.
- **Chat paging guidance** — typed chat message commands now document
`--page-all`, aggregate result shapes, and cursor behavior in CLI Help and
Agent selection examples.
- **Calendar skill parity** — mono and multi Calendar references are aligned to
prevent documentation drift without changing CLI behavior.
- **Release engineering** — CI now shards helper-package changes through the
full race suite, widens a flaky stdio idempotency test budget, governs exact
reviewed CLI/Schema type migrations, and lets authorized maintainers trigger
internal MCP evaluation with a reviewed `/eval` PR comment.
## [1.0.58-beta.2] - 2026-08-10
@@ -60,6 +86,10 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
- **CLI 接口兼容门禁支持 reviewed flag 类型豁免**(无用户可见变更)— `authoritative-interface-integrity` 与 `check-command-compatibility.sh` 此前一律拒绝历史命令的 flag 类型变更,即使新类型只是把同一套校验从 RunE 前移到解析期,也没有任何评审通道。现在两道门禁各带一张精确豁免表:命令路径 + flag 名 + 旧类型 → 新类型四元组全等才命中、方向敏感(`string`→`int` 与 `int`→`string` 是两个不同的键,只有被评审的方向可用),且仅当该 flag 的其他契约(shorthand / required / hidden / no-opt / scope)纹丝不动时才放行,因此豁免夹带不了别的破坏。首条也是目前唯一一条登记的是 `dws minutes permission apply --policy` 的 `string` → `int`(配合 #912):旧实现在 RunE 里做 `strconv.ParseInt(v, 10, 64)` 再校验 `[2,4]`,新实现由 pflag 以 `strconv.ParseInt(s, 0, 64)` 解析后仍校验 `[2,4]`,**历史上能成功的调用集是新调用集的子集**(base 0 额外接受 `0x3` 这类写法,只放宽不收紧),非法值依然失败、只是报错文案与时机前移;flag 默认值由 `""` 变 `"0"` 是类型的必然结果,两道门禁都不比较默认值,且该 flag 必须显式给出、默认值不可达。两张表必须逐字一致并有守卫测试锚定漂移——重复是被迫的而非选择:`check-authoritative-interface-baselines.sh` 会把整个 `scripts/policy/interface-baseline` 目录复制进检出历史版本的 worktree 再编译,那份拷贝不能 import 本分支新增的包。
- **Schema 兼容门禁支持 reviewed 参数类型豁免**(无用户可见变更)— 接上一条。`schema-compatibility` 是同一个 `Interface Integrity` job 里排在两道 CLI 接口门禁之后的第三道检查,此前也一律拒绝已发布参数的 `type` 变更。由于前两道先失败、`set -e` 让它从未在 CI 上暴露,上一条豁免只解决了三分之二。现在 `checkParameterCompatibility` 也带一张精确豁免表:`<product>/<tool id>` + 参数名 + 旧类型 + 新类型四元组全等才命中、方向敏感,且仅当该参数**除 `type` 外的全部已发布字段逐字段相等**时才放行。这里刻意用相等性比较而非「没有产生其他兼容性错误」:放宽 `required` / `cli_required`、清空 `required_when`、扩宽 `enum`、清空 `interface_type`、经 reviewed mapping exclusion 清空 `property`——这些变化单独看都是兼容的、根本不产生错误,若以错误列表代替相等性检查,它们就能搭着一次已评审的类型迁移一起蒙混过关。结构体整体比较还意味着将来给 `parameterSchema` 新增字段时会自动纳入守卫,而不是悄悄放宽每一条既有条目。唯一条目是 `minutes/minutes.apply_minutes_permission` 的 `policy` 由 `"string"` 迁移到 `"integer"`(配合 #912):该 `type` 由 Cobra flag 类型投影而来(provenance `cobra_flag_type`),描述的是 CLI 如何接受取值;消费方据此拼装的是命令行,而 `--policy 4` 在两种声明下是同一个 argv,加引号的 `--policy "4"` 到 pflag 仍是 4,RunE 也仍校验 `[2,4]`——而且该参数映射的 property `policyId` 一直以数字上报,新声明比旧声明更贴近真实请求。表里的类型值必须是 `schemaType` 实际产出的带引号形态(`"string"` 而非裸 `string`),守卫测试用 `schemaType` 复算并校验类型名属于 JSON Schema 的封闭取值集合——`reviewedInterfaceRefRedirect` 曾因键的书写形态错误两次静默失效,这里不重犯。
### Fixed
- **Event runtime-token handoff** — personal `event consume`, `status`, `stop`, and `+listen-im` now honor the existing root `--token` instead of falling back to a stale local OAuth profile. Detached personal-event buses negotiate the credential only after an additive capability handshake, receive and rotate it through owner-only local IPC, and keep it in memory; the token is never forwarded through child argv, environment variables, profiles, logs, or run-state files. Existing OAuth and multi-profile behavior is unchanged when `--token` is absent. A new client refuses to send a runtime token to an older bus and leaves its existing consumers and subscriptions untouched; the recovery message asks users to inspect `event status --as user`, preview `event stop --as user --all --dry-run`, and explicitly confirm `event stop --as user --all --yes` before retrying.
## [1.0.58-beta.1] - 2026-08-07
### Added
+41 -1
View File
@@ -4,7 +4,7 @@ Defines the stable `dws event consume` subprocess contract so an
orchestrator can determine when the consumer is ready, stop it cleanly,
and machine-read why it exited.
Scope of this branch: the five **contract** items below. Reconnect
Scope of this branch: the six **contract** items below. Reconnect
resilience (keeping the stream alive across a transient upstream drop) is
tracked separately and intentionally out of scope here.
@@ -159,6 +159,46 @@ marker; reconnecting an established Stream remains a separate mechanism.
`terminal_hold`, and identity-scoped cleanup; skill/docs tests pin the
operational recovery instructions.
### 6. Host runtime-token handoff
When the root command carries an explicit host-supplied `--token`, personal
event control requests and the foreground Stream use that token with higher
priority than local OAuth. A detached bus receives it only through the
owner-only local IPC transport:
1. The child starts in runtime-token mode with non-sensitive identity and
ticket metadata only; neither its argv nor environment contains the token.
2. The consumer sends `Hello` with `credential_mode=runtime_token`.
3. The bus advertises the additive `runtime_token_v1` capability and its
in-memory credential generation in `HelloAck`.
4. Only after that capability is confirmed does the consumer send a bounded
`credential_update` frame. The bus applies it with generation CAS, replies
with `credential_update_ack`, and registers the consumer only on success.
The bus blocks ticket acquisition until the first runtime credential arrives.
A later invocation may rotate Token A to Token B on a compatible existing bus;
the current WebSocket remains connected and the next ticket request or natural
reconnect uses B. If a 401 rejects the current runtime token, only an already
installed newer generation is retried; the runtime path never refreshes or
falls back to a local OAuth profile and never suggests `dws auth login`.
Clients do not send a token to a bus that lacks the capability, do not stop
other consumers automatically, and fail before printing the ready marker. With
no explicit `--token`, the original OAuth, refresh, profile, and old-client to
new-bus protocol behavior remains unchanged.
**Verification**
- T6a: a stale local Token A and root Token B produce control and ticket
requests authenticated only with B.
- T6b: compatible bus reuse supports A-to-B rotation and generation conflicts;
401 retries only an already-installed newer runtime token.
- T6c: an old bus receives no credential and remains running; the new consumer
exits before its ready marker.
- T6d: a canary credential is absent from child argv/environment, dry-run,
stdout/stderr, `bus.meta`, `bus.log`, run state, and returned errors.
- T6e: no-token OAuth, refresh, multi-profile, marker/cache, and bus-reuse tests
continue to pass.
## Out of scope (next branch)
**Reconnect resilience** — today `personal source` retries only
+84 -24
View File
@@ -38,6 +38,7 @@ import (
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/registry"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
@@ -74,7 +75,7 @@ var (
// newEventCommand returns the `event` parent command and all its subcommands.
// Wired into root.go's utilityCommands list.
func newEventCommand() *cobra.Command {
func newEventCommand(globalFlags ...*GlobalFlags) *cobra.Command {
// Product-level Agent routing Decl (migrated from selection/event.json
// products.event). Catalog assembly stamps provenance contract_final.
contract.RegisterProductDecl(contract.ProductDecl{
@@ -99,12 +100,12 @@ func newEventCommand() *cobra.Command {
RunE: func(c *cobra.Command, _ []string) error { return c.Help() },
}
cmd.AddCommand(
newEventListenIMCommand(),
newEventConsumeCommand(),
newEventListenIMCommand(globalFlags...),
newEventConsumeCommand(globalFlags...),
newEventListCommand(),
newEventSchemaCommand(),
newEventStatusCommand(),
newEventStopCommand(),
newEventStatusCommandWithFlags(globalFlags...),
newEventStopCommandWithFlags(globalFlags...),
newEventBusCommand(),
)
return cmd
@@ -114,7 +115,7 @@ func newEventCommand() *cobra.Command {
// event consume
// ─────────────────────────────────────────────────────────────────────
func newEventConsumeCommand() *cobra.Command {
func newEventConsumeCommand(globalFlags ...*GlobalFlags) *cobra.Command {
var (
eventTypes []string
filter string
@@ -170,6 +171,8 @@ SIGTERM、关 stdin,或先用 dws event stop <subscribe_id> --dry-run 预览
return err
}
if as == "user" {
personalOpts.ExplicitToken = eventExplicitToken(globalFlags)
personalOpts.ClientIDOverride = eventExplicitClientID(globalFlags)
personalOpts.EventKeys = dedupePersonalEventKeys(args)
personalOpts.EventKey = firstArg(personalOpts.EventKeys)
personalOpts.Flatten = flatten
@@ -563,6 +566,8 @@ func newEventBusCommand() *cobra.Command {
clientIDOverride string
idleTimeout time.Duration
sourceKindRaw string
runtimeTokenMode bool
identityHashFlag string
streamOpts eventStreamTicketOptions
)
cmd := &cobra.Command{
@@ -599,23 +604,45 @@ func newEventBusCommand() *cobra.Command {
sourceKind = dwsevent.SourceKindAppStream
}
if sourceKind == dwsevent.SourceKindPersonalStream {
identity, err := eventResolvePersonal(ctx, configDir, streamOpts.SourceID)
if err != nil {
return failEarly(fmt.Errorf("event _bus: %w", err))
var (
identity personal.Identity
identityHash string
)
if runtimeTokenMode {
identityHash = strings.TrimSpace(identityHashFlag)
if !validPersonalIdentityHash(identityHash) {
return failEarly(errors.New("event _bus: --identity-hash must be a 16-character hexadecimal identity hash in runtime token mode"))
}
if strings.TrimSpace(clientIDOverride) == "" {
return failEarly(errors.New("event _bus: --client-id is required in runtime token mode"))
}
identity = personal.Identity{
ClientID: strings.TrimSpace(clientIDOverride),
SourceID: personalEventStreamSourceID(streamOpts.SourceID),
}
} else {
var err error
identity, err = eventResolvePersonal(ctx, configDir, streamOpts.SourceID)
if err != nil {
return failEarly(fmt.Errorf("event _bus: %w", err))
}
if clientIDOverride != "" {
identity.ClientID = clientIDOverride
}
identityHash = dwsevent.IdentityHash(identity.Key())
}
if clientIDOverride != "" {
identity.ClientID = clientIDOverride
}
identityHash := dwsevent.IdentityHash(identity.Key())
editionName := editionNameOrDefault()
workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
endpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
credentialBroker := newPersonalCredentialBroker(configDir, runtimeTokenMode, runtimeTokenMode)
src, err := eventNewPersonalSource(ctx, personalStreamSourceOptions{
ConfigDir: configDir,
Identity: identity,
TicketMode: streamOpts.Mode,
TicketURL: streamOpts.TicketURL,
ClientIDOverride: clientIDOverride,
CredentialBroker: credentialBroker,
RuntimeTokenMode: runtimeTokenMode,
})
if err != nil {
return failEarly(err)
@@ -628,17 +655,18 @@ func newEventBusCommand() *cobra.Command {
}
}
busCfg := bus.Config{
WorkDir: workDir,
IPCEndpoint: endpoint,
ClientID: identity.ClientID,
Edition: editionName,
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: identityHash,
SourceID: identity.SourceID,
Source: src,
IdleTimeout: idleTimeout,
ReadyPipe: readyPipe,
Logger: slog.Default(),
WorkDir: workDir,
IPCEndpoint: endpoint,
ClientID: identity.ClientID,
Edition: editionName,
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: identityHash,
SourceID: identity.SourceID,
Source: src,
IdleTimeout: idleTimeout,
ReadyPipe: readyPipe,
Logger: slog.Default(),
CredentialBroker: credentialBroker,
}
bus.ApplyEnvTuning(&busCfg)
return eventBusRun(ctx, busCfg)
@@ -698,12 +726,18 @@ func newEventBusCommand() *cobra.Command {
"exit after this long with zero consumers (0 = disabled)")
cmd.Flags().StringVar(&sourceKindRaw, "source-kind", string(dwsevent.SourceKindAppStream),
"event source kind: app_stream|personal_stream")
cmd.Flags().BoolVar(&runtimeTokenMode, "runtime-token-mode", false,
"use an owner-injected in-memory runtime credential")
cmd.Flags().StringVar(&identityHashFlag, "identity-hash", "",
"pre-resolved non-sensitive personal identity hash")
cmd.Flags().StringVar(&streamOpts.Mode, "stream-ticket-mode", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_MODE")),
"用户 Stream 建联模式:空=SDK app credential;normal/custom=portal 取票")
cmd.Flags().StringVar(&streamOpts.SourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
"用户 Stream sourceId;personal_stream 开源版默认 open")
cmd.Flags().StringVar(&streamOpts.TicketURL, "stream-ticket-url", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_URL")),
"用户 Stream 取票 URL;personal_stream 默认由 MCP base URL 派生")
_ = cmd.Flags().MarkHidden("runtime-token-mode")
_ = cmd.Flags().MarkHidden("identity-hash")
return cmd
}
@@ -822,6 +856,10 @@ func newEventListCommand() *cobra.Command {
// ─────────────────────────────────────────────────────────────────────
func newEventStatusCommand() *cobra.Command {
return newEventStatusCommandWithFlags()
}
func newEventStatusCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
var (
all bool
allEditions bool
@@ -847,6 +885,8 @@ func newEventStatusCommand() *cobra.Command {
return fmt.Errorf("event status: %w", err)
}
personalOpts.Format = formatRaw
personalOpts.ExplicitToken = eventExplicitToken(globalFlags)
personalOpts.ClientIDOverride = eventExplicitClientID(globalFlags)
return eventRunPersonalStatus(c, personalOpts)
}
if err := rejectChangedFlags(c, "user", "event", "status", "subscribe-id", "personal-event-base-url", "stream-source-id"); err != nil {
@@ -1138,6 +1178,10 @@ func renderStatusBlock(w io.Writer, qs busctl.EntryStatus) {
}
func newEventStopCommand() *cobra.Command {
return newEventStopCommandWithFlags()
}
func newEventStopCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
var asIdentity string
var opts personalStopOptions
cmd := &cobra.Command{
@@ -1158,6 +1202,8 @@ func newEventStopCommand() *cobra.Command {
}
if as == "user" {
opts.SubscribeID = firstArg(args)
opts.ExplicitToken = eventExplicitToken(globalFlags)
opts.ClientIDOverride = eventExplicitClientID(globalFlags)
if eventStopDryRun(c) {
return writeEventStopDryRun(c, as, opts)
}
@@ -1261,6 +1307,20 @@ func eventStopDryRun(cmd *cobra.Command) bool {
return value
}
func eventExplicitToken(globalFlags []*GlobalFlags) string {
if len(globalFlags) == 0 || globalFlags[0] == nil {
return ""
}
return strings.TrimSpace(globalFlags[0].Token)
}
func eventExplicitClientID(globalFlags []*GlobalFlags) string {
if len(globalFlags) == 0 || globalFlags[0] == nil {
return ""
}
return strings.TrimSpace(globalFlags[0].ClientID)
}
func writeEventStopDryRun(cmd *cobra.Command, identity string, opts personalStopOptions) error {
payload := map[string]any{
"dry_run": true,
+3 -1
View File
@@ -65,7 +65,7 @@ func (eventTargetReader) CallMCPData(product, tool string, params map[string]any
var eventListenIMReader = func() targetresolver.Reader { return eventTargetReader{} }
func newEventListenIMCommand() *cobra.Command {
func newEventListenIMCommand(globalFlags ...*GlobalFlags) *cobra.Command {
var opts listenIMOptions
cmd := &cobra.Command{
Use: "+listen-im",
@@ -91,6 +91,8 @@ func newEventListenIMCommand() *cobra.Command {
StreamTicketMode: opts.StreamTicketMode,
StreamTicketURL: opts.StreamTicketURL,
StreamSourceID: opts.StreamSourceID,
ExplicitToken: eventExplicitToken(globalFlags),
ClientIDOverride: eventExplicitClientID(globalFlags),
Common: commonConsumeOptions{
FormatRaw: "ndjson",
MaxEvents: opts.MaxEvents,
+20
View File
@@ -28,6 +28,7 @@ import (
apperrors "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/errors"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
)
@@ -184,6 +185,25 @@ func (r *personalSubscriptionAttemptReservation) completeSuccess() error {
return nil
}
// releaseRuntimeTokenFailure releases the in-flight claim without recording a
// cross-invocation hold. A host may supply a fresh token on the very next
// command, which must be allowed to retry immediately.
func (r *personalSubscriptionAttemptReservation) releaseRuntimeTokenFailure() error {
if r == nil {
return runtimecred.ErrRuntimeTokenRejected
}
if r.store == nil || r.claim == nil {
return personalSubscriptionGuardError(errors.Join(
runtimecred.ErrRuntimeTokenRejected,
errors.New("personal event: subscription attempt reservation is incomplete"),
))
}
if err := r.store.Release(r.claim); err != nil {
return personalSubscriptionGuardError(errors.Join(runtimecred.ErrRuntimeTokenRejected, err))
}
return runtimecred.ErrRuntimeTokenRejected
}
func (r *personalSubscriptionAttemptReservation) completeFailure(
ctx context.Context,
failedIndex int,
+432 -47
View File
@@ -14,6 +14,7 @@
package app
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
@@ -40,6 +41,7 @@ import (
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
@@ -82,6 +84,8 @@ type personalConsumeOptions struct {
StreamTicketMode string
StreamTicketURL string
StreamSourceID string
ExplicitToken string
ClientIDOverride string
}
type personalListOptions struct {
@@ -92,19 +96,23 @@ type personalListOptions struct {
}
type personalStatusOptions struct {
EventKey string
Status string
SubscribeID string
Format string
ControlBaseURL string
StreamSourceID string
EventKey string
Status string
SubscribeID string
Format string
ControlBaseURL string
StreamSourceID string
ExplicitToken string
ClientIDOverride string
}
type personalStopOptions struct {
SubscribeID string
All bool
ControlBaseURL string
StreamSourceID string
SubscribeID string
All bool
ControlBaseURL string
StreamSourceID string
ExplicitToken string
ClientIDOverride string
}
type personalStreamSourceOptions struct {
@@ -113,6 +121,8 @@ type personalStreamSourceOptions struct {
TicketMode string
TicketURL string
ClientIDOverride string
CredentialBroker *runtimecred.Broker
RuntimeTokenMode bool
}
var (
@@ -142,10 +152,19 @@ var (
personalResolveAuxiliaryAccessToken = ResolveAuxiliaryAccessToken
personalForceRefreshRejectedToken = forceRefreshRejectedAccessToken
personalLoadTokenData = authpkg.LoadTokenData
personalLoadProfiles = authpkg.LoadProfiles
personalClientID = authpkg.ClientID
personalRuntimeEventClientID = runtimePersonalEventClientID
personalResolveAppCredentialsStrict = authpkg.ResolveAppCredentialsStrict
)
func runtimePersonalEventClientID() string {
if clientID := strings.TrimSpace(edition.Get().AuthClientID); clientID != "" {
return clientID
}
return strings.TrimSpace(os.Getenv("DWS_CLIENT_ID"))
}
func newEventSchemaCommand() *cobra.Command {
var asIdentity string
var formatRaw string
@@ -283,7 +302,7 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
projector := personalEventProjector(opts.DebugRawEvents, opts.Flatten)
configDir := defaultConfigDir()
identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
if err != nil {
return fmt.Errorf("event consume --as user: %w", err)
}
@@ -291,13 +310,24 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
editionName := editionNameOrDefault()
workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
spawnProfileSelector := personalBusProfileSelector(configDir, identity)
spawnProfileSelector := ""
if strings.TrimSpace(opts.ExplicitToken) == "" {
spawnProfileSelector = personalBusProfileSelector(configDir, identity)
}
spawnArgs := personalBusSpawnArgsForToken(
identity,
identityHash,
opts.StreamTicketMode,
opts.StreamTicketURL,
spawnProfileSelector,
opts.ExplicitToken,
)
routes, err := consume.ParseRoutes(opts.Common.RoutesRaw)
if err != nil {
return fmt.Errorf("event consume --as user: %w", personalSubscriptionValidationError(err))
}
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
if opts.Common.DryRun {
if strings.TrimSpace(opts.SubscribeID) == "" {
if err := validatePersonalSubscriptionOptions(opts); err != nil {
@@ -314,7 +344,7 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
WorkDir: workDir,
IPCEndpoint: ipcEndpoint,
ClientID: identity.ClientID,
SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector),
SpawnExtraArgs: personalBusSpawnArgsForToken(identity, identityHash, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector, opts.ExplicitToken),
Compact: opts.Common.Compact,
MaxEvents: opts.Common.MaxEvents,
Duration: opts.Common.Duration,
@@ -341,7 +371,8 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
WorkDir: workDir,
IPCEndpoint: ipcEndpoint,
ClientID: identity.ClientID,
SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, opts.StreamTicketURL, spawnProfileSelector),
SpawnExtraArgs: spawnArgs,
RuntimeToken: strings.TrimSpace(opts.ExplicitToken),
Compact: opts.Common.Compact,
MaxEvents: opts.Common.MaxEvents,
Duration: opts.Common.Duration,
@@ -370,13 +401,25 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
}
}
var foregroundSource *source.PersonalSource
var (
foregroundSource *source.PersonalSource
foregroundBroker *runtimecred.Broker
)
if opts.Common.Foreground {
explicitToken := strings.TrimSpace(opts.ExplicitToken)
foregroundBroker = newPersonalCredentialBroker(configDir, explicitToken != "", false)
if explicitToken != "" {
if _, err := foregroundBroker.Update(0, explicitToken); err != nil {
return personalSubscriptionValidationError(err)
}
}
foregroundSource, err = personalNewStreamSource(ctx, personalStreamSourceOptions{
ConfigDir: configDir,
Identity: identity,
TicketMode: opts.StreamTicketMode,
TicketURL: opts.StreamTicketURL,
ConfigDir: configDir,
Identity: identity,
TicketMode: opts.StreamTicketMode,
TicketURL: opts.StreamTicketURL,
CredentialBroker: foregroundBroker,
RuntimeTokenMode: explicitToken != "",
})
if err != nil {
return personalSubscriptionValidationError(err)
@@ -398,6 +441,10 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
}
sub, eventKey, ruleType, err := personalEnsureSubscription(ctx, client, identity, opts)
if err != nil {
if strings.TrimSpace(opts.ExplicitToken) != "" && personalRuntimeTokenControlRejection(err) {
err = attempt.releaseRuntimeTokenFailure()
return fmt.Errorf("event consume --as user: %w", err)
}
err = attempt.completeFailure(ctx, 0, 0, err, nil)
return fmt.Errorf("event consume --as user: %w", err)
}
@@ -476,14 +523,15 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
}
if opts.Common.Foreground {
busCfg := bus.Config{
WorkDir: workDir,
IPCEndpoint: ipcEndpoint,
ClientID: identity.ClientID,
Edition: editionName,
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: identityHash,
SourceID: identity.SourceID,
Source: foregroundSource,
WorkDir: workDir,
IPCEndpoint: ipcEndpoint,
ClientID: identity.ClientID,
Edition: editionName,
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: identityHash,
SourceID: identity.SourceID,
Source: foregroundSource,
CredentialBroker: foregroundBroker,
}
bus.ApplyEnvTuning(&busCfg)
return personalBusRun(ctx, busCfg)
@@ -517,7 +565,7 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
ctx := c.Context()
configDir := defaultConfigDir()
identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
if err != nil {
return fmt.Errorf("event consume --as user: %w", err)
}
@@ -525,7 +573,10 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
editionName := editionNameOrDefault()
workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
spawnProfileSelector := personalBusProfileSelector(configDir, identity)
spawnProfileSelector := ""
if strings.TrimSpace(opts.ExplicitToken) == "" {
spawnProfileSelector = personalBusProfileSelector(configDir, identity)
}
routes, err := consume.ParseRoutes(opts.Common.RoutesRaw)
if err != nil {
return fmt.Errorf("event consume --as user: %w", personalSubscriptionValidationError(err))
@@ -534,7 +585,7 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
WorkDir: workDir,
IPCEndpoint: ipcEndpoint,
ClientID: identity.ClientID,
SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector),
SpawnExtraArgs: personalBusSpawnArgsForToken(identity, identityHash, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector, opts.ExplicitToken),
Compact: opts.Common.Compact,
MaxEvents: opts.Common.MaxEvents,
Duration: opts.Common.Duration,
@@ -560,8 +611,9 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
printPersonalMultiDryRun(c.ErrOrStderr(), baseCfg, plans)
return nil
}
baseCfg.RuntimeToken = strings.TrimSpace(opts.ExplicitToken)
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
attempt, err := reservePersonalSubscriptionAttempts(
workDir,
client,
@@ -598,6 +650,10 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
if personalSubscriptionCanceled(ctx, cause) {
cleanupCtx = ctx
}
if strings.TrimSpace(opts.ExplicitToken) != "" && personalRuntimeTokenControlRejection(cause) {
cleanup(cleanupCtx)
return attempt.releaseRuntimeTokenFailure()
}
completed := attempt.completeFailure(ctx, failedIndex, succeededCount, cause, override)
// Persist the hold (or release a canceled claim) before any potentially
// slow remote rollback. Otherwise the attempt lease can expire while
@@ -985,7 +1041,7 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
return err
}
configDir := defaultConfigDir()
identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
if err != nil {
return fmt.Errorf("event status --as user: %w", err)
}
@@ -1017,7 +1073,7 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
if status == "" || status == "all" {
status = ""
}
subs, err := personalListSubscriptions(newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity), ctx, personal.ListOptions{
subs, err := personalListSubscriptions(newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken), ctx, personal.ListOptions{
Status: status,
EventKey: opts.EventKey,
SubscribeID: opts.SubscribeID,
@@ -1038,6 +1094,15 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
return nil
}
func personalRuntimeTokenControlRejection(err error) bool {
var apiErr *personal.APIError
if !errors.As(err, &apiErr) || apiErr == nil {
return false
}
return apiErr.HTTPStatus == http.StatusUnauthorized ||
strings.EqualFold(strings.TrimSpace(apiErr.Code), "RUNTIME_TOKEN_REJECTED")
}
func ensurePublicPersonalEvent(eventKey string) error {
eventKey = strings.TrimSpace(eventKey)
if eventKey == "" {
@@ -1120,7 +1185,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error {
}
configDir := defaultConfigDir()
identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
if err != nil {
return fmt.Errorf("event stop --as user: %w", err)
}
@@ -1132,7 +1197,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error {
if err != nil {
return fmt.Errorf("event stop --as user: %w", err)
}
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
for _, id := range subscribeIDs {
if err := personalDeleteSubscription(client, ctx, id); err != nil {
return fmt.Errorf("event stop --as user: cancel subscription %s: %w", id, err)
@@ -1248,6 +1313,138 @@ func printPersonalStopResult(w io.Writer, subscribeIDs []string, single bool, bu
fmt.Fprintf(w, "cancelled %d personal subscription(s); %s\n", len(subscribeIDs), busState)
}
func resolvePersonalEventIdentityForToken(ctx context.Context, configDir, sourceIDOverride, explicitToken string, clientIDOverrides ...string) (personal.Identity, error) {
explicitToken = strings.TrimSpace(explicitToken)
if explicitToken == "" {
return personalResolveEventIdentity(ctx, configDir, sourceIDOverride)
}
clientIDOverride := ""
if len(clientIDOverrides) > 0 {
clientIDOverride = strings.TrimSpace(clientIDOverrides[0])
}
return resolvePersonalEventIdentityWithToken(ctx, configDir, sourceIDOverride, explicitToken, clientIDOverride)
}
// resolvePersonalEventIdentityWithToken resolves only non-sensitive identity
// metadata around a caller-supplied bearer token. It intentionally does not
// call LoadTokenData or any refresh-capable token resolver: an explicit root
// --token must never be replaced with, persisted into, or used to refresh a
// local OAuth profile.
func resolvePersonalEventIdentityWithToken(ctx context.Context, configDir, sourceIDOverride, explicitToken string, clientIDOverrides ...string) (personal.Identity, error) {
explicitToken = strings.TrimSpace(explicitToken)
if explicitToken == "" {
return resolvePersonalEventIdentity(ctx, configDir, sourceIDOverride)
}
if strings.Contains(strings.TrimSpace(authpkg.RuntimeProfile()), ",") {
return personal.Identity{}, fmt.Errorf("personal events require exactly one --profile")
}
corpID := resolveRuntimeDefault(ctx, "$corpId")
userID := resolveRuntimeDefault(ctx, "$currentUserId")
clientID := ""
if len(clientIDOverrides) > 0 {
clientID = strings.TrimSpace(clientIDOverrides[0])
}
if clientID == "" {
// An edition hook or explicit environment value is runtime identity,
// not persisted app state. Resolve it before profiles.json so a complete
// host context never depends on local OAuth metadata health.
clientID = strings.TrimSpace(personalRuntimeEventClientID())
}
explicitProfile := strings.TrimSpace(authpkg.RuntimeProfile()) != ""
if explicitProfile || corpID == "" || userID == "" || clientID == "" {
profile, err := personalEventProfileMetadata(configDir)
if err != nil {
// A user-selected --profile remains a strict contract. Without an
// explicit selector, profiles.json is optional metadata for a
// host-managed bearer: malformed or stale persisted state must not
// override complete runtime defaults or prevent the later global
// client-id fallback.
if explicitProfile {
return personal.Identity{}, fmt.Errorf("load OAuth identity metadata: %w", err)
}
profile = nil
}
if profile != nil {
if corpID == "" {
corpID = strings.TrimSpace(profile.CorpID)
}
if userID == "" {
userID = strings.TrimSpace(profile.UserID)
}
if clientID == "" {
clientID = strings.TrimSpace(profile.ClientID)
}
}
}
if clientID == "" {
// Persisted/global app credentials are only a fallback after the
// selected profile, so an old app config cannot override profile.ClientID.
clientID = strings.TrimSpace(personalClientID())
}
if clientID == "" {
if id, _, _, _, resolveErr := personalResolveAppCredentialsStrict(configDir); resolveErr == nil {
clientID = strings.TrimSpace(id)
}
}
if clientID == "" {
return personal.Identity{}, fmt.Errorf("cannot resolve OAuth client_id for personal events")
}
sourceID := strings.TrimSpace(sourceIDOverride)
if sourceID == "" {
sourceID = personalEventStreamSourceID("")
}
localSubject := ""
if corpID == "" || userID == "" {
localSubject = personalTokenSubject("access", explicitToken)
}
return personal.Identity{
LocalSubject: localSubject,
CorpID: corpID,
UserID: userID,
ClientID: clientID,
SourceID: sourceID,
}, nil
}
func personalEventProfileMetadata(configDir string) (*authpkg.Profile, error) {
cfg, err := personalLoadProfiles(configDir)
if err != nil {
return nil, err
}
selector := strings.TrimSpace(authpkg.RuntimeProfile())
explicitSelector := selector != ""
if strings.Contains(selector, ",") {
return nil, fmt.Errorf("personal events require exactly one --profile")
}
if cfg == nil || len(cfg.Profiles) == 0 {
if explicitSelector {
return nil, fmt.Errorf("profile %q not found", selector)
}
return nil, nil
}
if selector == "" {
selector = strings.TrimSpace(cfg.CurrentProfile)
}
if selector == "" {
return nil, nil
}
profile, err := selectPersonalEventProfileMetadata(cfg, selector, make(map[string]struct{}))
if err != nil && !explicitSelector {
// A stale persisted CurrentProfile must not make a host-provided bearer
// unusable. Runtime defaults and the one-way local subject are sufficient
// to isolate the event bus without consulting local OAuth credentials.
return nil, nil
}
return profile, err
}
func selectPersonalEventProfileMetadata(cfg *authpkg.ProfilesConfig, selector string, visited map[string]struct{}) (*authpkg.Profile, error) {
_ = visited // retained for the focused compatibility seam used by app tests.
return authpkg.ResolveProfileMetadata(cfg, strings.TrimSpace(selector))
}
func resolvePersonalEventIdentity(ctx context.Context, configDir string, sourceIDOverride string) (personal.Identity, error) {
accessToken, err := personalResolveAuxiliaryAccessToken(ctx, configDir, "")
if err != nil {
@@ -1302,7 +1499,11 @@ func resolvePersonalEventIdentity(ctx context.Context, configDir string, sourceI
}, nil
}
func newPersonalEventControlClient(configDir, baseURL string, identity personal.Identity) *personal.Client {
func newPersonalEventControlClient(configDir, baseURL string, identity personal.Identity, explicitTokens ...string) *personal.Client {
explicitToken := ""
if len(explicitTokens) > 0 {
explicitToken = strings.TrimSpace(explicitTokens[0])
}
identity.AccessToken = ""
client := personal.NewClient(baseURL, identity)
version := strings.TrimSpace(RawVersion())
@@ -1311,12 +1512,146 @@ func newPersonalEventControlClient(configDir, baseURL string, identity personal.
}
client.ClientVersion = version
client.UserAgent = "dws-cli/" + version
client.AccessTokenProvider = func(ctx context.Context) (string, error) {
return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
if explicitToken != "" {
client.AccessTokenProvider = func(context.Context) (string, error) { return explicitToken, nil }
client.HTTPClient.Transport = runtimeTokenControlTransport{base: http.DefaultTransport, token: explicitToken}
client.HTTPClient.CheckRedirect = runtimeTokenRedirectPolicy
} else {
client.AccessTokenProvider = func(ctx context.Context) (string, error) {
return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
}
}
return client
}
// runtimeTokenRedirectPolicy prevents Go's redirect machinery from copying
// DWS's custom x-user-access-token header to another authority. Returning
// ErrUseLastResponse keeps the 3xx response available to the caller without a
// url.Error that could echo an attacker-controlled Location value.
func runtimeTokenRedirectPolicy(req *http.Request, via []*http.Request) error {
if len(via) == 0 || req == nil || req.URL == nil || via[0] == nil || via[0].URL == nil {
return http.ErrUseLastResponse
}
origin := via[0].URL
if !strings.EqualFold(strings.TrimSpace(req.URL.Host), strings.TrimSpace(origin.Host)) {
return http.ErrUseLastResponse
}
if strings.EqualFold(origin.Scheme, "https") && !strings.EqualFold(req.URL.Scheme, "https") {
return http.ErrUseLastResponse
}
return nil
}
const runtimeTokenControlErrorBody = `{"code":"RUNTIME_TOKEN_REJECTED","message":"event runtime token was rejected; retry with a fresh host credential"}`
// runtimeTokenControlTransport scrubs an explicit bearer from every response
// body and diagnostic header before the control client decodes or logs it. A
// 401 is replaced with a fixed rejection envelope so untrusted response text
// can never escape through stderr or debug logs.
type runtimeTokenControlTransport struct {
base http.RoundTripper
token string
}
func (t runtimeTokenControlTransport) RoundTrip(req *http.Request) (*http.Response, error) {
base := t.base
if base == nil {
base = http.DefaultTransport
}
resp, err := base.RoundTrip(req)
if err != nil {
if token := strings.TrimSpace(t.token); token != "" && strings.Contains(err.Error(), token) {
return nil, errors.New("personal event: runtime-token control request failed")
}
return nil, err
}
if resp == nil {
return resp, err
}
token := strings.TrimSpace(t.token)
for key, values := range resp.Header {
for i := range values {
if token != "" {
values[i] = strings.ReplaceAll(values[i], token, "<redacted-runtime-token>")
}
}
resp.Header[key] = values
}
var responseBody []byte
if resp.Body != nil {
responseBody, err = io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
_ = resp.Body.Close()
if err != nil {
return nil, errors.New("personal event: read runtime-token control response")
}
}
if resp.StatusCode == http.StatusUnauthorized {
responseBody = []byte(runtimeTokenControlErrorBody)
} else if token != "" {
responseBody = redactRuntimeTokenResponseBody(responseBody, token)
}
resp.Body = io.NopCloser(bytes.NewReader(responseBody))
resp.ContentLength = int64(len(responseBody))
if resp.Header == nil {
resp.Header = make(http.Header)
}
resp.Header.Set("Content-Type", "application/json")
resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(responseBody)))
return resp, nil
}
func redactRuntimeTokenResponseBody(data []byte, token string) []byte {
token = strings.TrimSpace(token)
if len(data) == 0 || token == "" {
return data
}
decoder := json.NewDecoder(bytes.NewReader(data))
decoder.UseNumber()
var decoded any
if err := decoder.Decode(&decoded); err == nil {
var trailing any
if trailingErr := decoder.Decode(&trailing); errors.Is(trailingErr, io.EOF) {
if redacted, changed := redactRuntimeTokenJSONValue(decoded, token); changed {
if encoded, marshalErr := json.Marshal(redacted); marshalErr == nil {
return encoded
}
}
}
}
return bytes.ReplaceAll(data, []byte(token), []byte("<redacted-runtime-token>"))
}
func redactRuntimeTokenJSONValue(value any, token string) (any, bool) {
switch typed := value.(type) {
case string:
redacted := strings.ReplaceAll(typed, token, "<redacted-runtime-token>")
return redacted, redacted != typed
case []any:
changed := false
for i := range typed {
var itemChanged bool
typed[i], itemChanged = redactRuntimeTokenJSONValue(typed[i], token)
changed = changed || itemChanged
}
return typed, changed
case map[string]any:
changed := false
redactedMap := make(map[string]any, len(typed))
for key, item := range typed {
redactedKey := strings.ReplaceAll(key, token, "<redacted-runtime-token>")
redacted, itemChanged := redactRuntimeTokenJSONValue(item, token)
redactedMap[redactedKey] = redacted
changed = changed || itemChanged || redactedKey != key
}
if !changed {
return typed, false
}
return redactedMap, true
default:
return value, false
}
}
func personalTokenSubject(kind, token string) string {
token = strings.TrimSpace(token)
if token == "" {
@@ -1326,6 +1661,15 @@ func personalTokenSubject(kind, token string) string {
return strings.TrimSpace(kind) + ":" + hex.EncodeToString(sum[:])
}
func validPersonalIdentityHash(value string) bool {
value = strings.TrimSpace(value)
if len(value) != 16 {
return false
}
_, err := hex.DecodeString(value)
return err == nil
}
func resolveRuntimeDefault(ctx context.Context, key string) string {
if fnMap := edition.Get().RuntimeDefaults; fnMap != nil {
if fn := fnMap()[key]; fn != nil {
@@ -1363,20 +1707,42 @@ func newPersonalStreamSource(ctx context.Context, opts personalStreamSourceOptio
}
clientSecret = secret
}
credentialBroker := opts.CredentialBroker
if credentialBroker == nil {
credentialBroker = newPersonalCredentialBroker(opts.ConfigDir, false, false)
}
httpClient := &http.Client{Timeout: 30 * time.Second}
if opts.RuntimeTokenMode {
httpClient.CheckRedirect = runtimeTokenRedirectPolicy
}
_ = ctx
return source.NewPersonal(source.PersonalConfig{
AccessTokenProvider: func(ctx context.Context) (string, error) {
return personalResolveAuxiliaryAccessToken(ctx, opts.ConfigDir, "")
return credentialBroker.Resolve(ctx)
},
ForceRefreshToken: func(ctx context.Context, rejectedToken string) (string, error) {
return personalForceRefreshRejectedToken(ctx, opts.ConfigDir, rejectedToken)
return credentialBroker.RefreshRejected(ctx, rejectedToken)
},
ClassifyRetryReject: credentialBroker.ClassifyRejectedAfterRetry,
ClientID: clientID,
ClientSecret: clientSecret,
SourceID: opts.Identity.SourceID,
TicketURL: ticketURL,
TicketMode: mode,
HTTPClient: httpClient,
})
}
func newPersonalCredentialBroker(configDir string, requireSeed, requireActivation bool) *runtimecred.Broker {
return runtimecred.New(runtimecred.Config{
RequireSeed: requireSeed,
RequireActivation: requireActivation,
LocalResolve: func(ctx context.Context) (string, error) {
return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
},
LocalRefresh: func(ctx context.Context, rejectedToken string) (string, error) {
return personalForceRefreshRejectedToken(ctx, configDir, rejectedToken)
},
ClientID: clientID,
ClientSecret: clientSecret,
SourceID: opts.Identity.SourceID,
TicketURL: ticketURL,
TicketMode: mode,
HTTPClient: &http.Client{Timeout: 30 * time.Second},
})
}
@@ -1441,6 +1807,25 @@ func personalBusSpawnArgs(identity personal.Identity, ticketMode, ticketURL stri
return args
}
func personalBusSpawnArgsForToken(identity personal.Identity, identityHash, ticketMode, ticketURL, profileSelector, explicitToken string) []string {
if strings.TrimSpace(explicitToken) == "" {
return personalBusSpawnArgs(identity, ticketMode, ticketURL, profileSelector)
}
args := []string{
"--source-kind", string(dwsevent.SourceKindPersonalStream),
"--runtime-token-mode",
"--identity-hash", strings.TrimSpace(identityHash),
"--stream-source-id", strings.TrimSpace(identity.SourceID),
}
if strings.TrimSpace(ticketMode) != "" {
args = append(args, "--stream-ticket-mode", strings.TrimSpace(ticketMode))
}
if strings.TrimSpace(ticketURL) != "" {
args = append(args, "--stream-ticket-url", strings.TrimSpace(ticketURL))
}
return args
}
func personalEventTypes(eventKey string, explicit []string) []string {
if len(explicit) > 0 {
return explicit
@@ -0,0 +1,396 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
package app
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"log/slog"
"os"
"os/signal"
"path/filepath"
"runtime"
"strings"
"syscall"
"testing"
"time"
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
const (
runtimeTokenDetachedChildEnv = "DWS_EVENT_RUNTIME_TOKEN_E2E_CHILD"
runtimeTokenDetachedWorkDirEnv = "DWS_EVENT_RUNTIME_TOKEN_E2E_WORKDIR"
runtimeTokenDetachedEndpointEnv = "DWS_EVENT_RUNTIME_TOKEN_E2E_ENDPOINT"
runtimeTokenDetachedEvidenceEnv = "DWS_EVENT_RUNTIME_TOKEN_E2E_EVIDENCE"
runtimeTokenDetachedCanaryA = "dws-runtime-e2e-A-9f34c8d10b7e"
runtimeTokenDetachedCanaryB = "dws-runtime-e2e-B-2ad761e5c490"
runtimeTokenDetachedClientID = "runtime-e2e-client"
runtimeTokenDetachedIdentityHash = "90abcdef12345678"
runtimeTokenDetachedSourceID = "runtime-e2e-source"
)
// runRuntimeTokenDetachedE2EChild is called at the very start of TestMain.
// busctl.Spawn executes this test binary with production-style `event _bus`
// arguments; the env marker lets the child run a real bus daemon before the Go
// test runner attempts to parse those CLI arguments.
func runRuntimeTokenDetachedE2EChild() (int, bool) {
if os.Getenv(runtimeTokenDetachedChildEnv) != "1" {
return 0, false
}
workDir := strings.TrimSpace(os.Getenv(runtimeTokenDetachedWorkDirEnv))
endpoint := strings.TrimSpace(os.Getenv(runtimeTokenDetachedEndpointEnv))
evidence := strings.TrimSpace(os.Getenv(runtimeTokenDetachedEvidenceEnv))
if workDir == "" || endpoint == "" || evidence == "" {
return 91, true
}
argvClean := !runtimeTokenDetachedContainsCanary(strings.Join(os.Args, "\x00"))
envClean := !runtimeTokenDetachedContainsCanary(strings.Join(os.Environ(), "\x00"))
if err := appendRuntimeTokenDetachedEvidence(evidence,
fmt.Sprintf("child_start argv_clean=%t env_clean=%t", argvClean, envClean)); err != nil {
return 92, true
}
if !argvClean || !envClean {
return 93, true
}
logFile, err := os.OpenFile(filepath.Join(workDir, "bus.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
return 94, true
}
defer logFile.Close()
broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer cancel()
err = bus.Run(ctx, bus.Config{
WorkDir: workDir,
IPCEndpoint: endpoint,
ClientID: runtimeTokenDetachedClientID,
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: runtimeTokenDetachedIdentityHash,
SourceID: runtimeTokenDetachedSourceID,
Edition: "open",
SDKVersion: "runtime-e2e",
Source: &runtimeTokenDetachedSource{broker: broker, evidence: evidence},
CredentialBroker: broker,
ReadyPipe: busctl.ReadyFDFromEnv(),
Logger: slog.New(slog.NewTextHandler(logFile, nil)),
})
if err != nil && !errors.Is(err, context.Canceled) {
_ = appendRuntimeTokenDetachedEvidence(evidence, "bus_exit clean=false")
return 95, true
}
_ = appendRuntimeTokenDetachedEvidence(evidence, "bus_exit clean=true")
return 0, true
}
type runtimeTokenDetachedSource struct {
broker *runtimecred.Broker
evidence string
}
// Start models the credential-sensitive part of a reconnecting Stream source
// without network access. It resolves A for the first connection, waits until a
// second consumer rotates the broker to B, then exercises the exact 401 path:
// RefreshRejected(A) must return B and must not fall back to local OAuth.
func (s *runtimeTokenDetachedSource) Start(ctx context.Context, _ dwsevent.EmitFn) error {
first, err := s.broker.Resolve(ctx)
if err != nil {
return errors.New("runtime e2e: initial credential unavailable")
}
if first != runtimeTokenDetachedCanaryA || s.broker.Generation() != 1 {
return errors.New("runtime e2e: initial credential mismatch")
}
if err := appendRuntimeTokenDetachedEvidence(s.evidence, "resolved_a=true generation=1"); err != nil {
return errors.New("runtime e2e: record initial connection")
}
ticker := time.NewTicker(5 * time.Millisecond)
defer ticker.Stop()
for s.broker.Generation() < 2 {
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
}
rotated, err := s.broker.RefreshRejected(ctx, first)
if err != nil || rotated != runtimeTokenDetachedCanaryB {
return errors.New("runtime e2e: rotated credential unavailable")
}
if err := appendRuntimeTokenDetachedEvidence(s.evidence, "rejected_a=true resolved_b=true reconnect=true generation=2"); err != nil {
return errors.New("runtime e2e: record reconnect")
}
<-ctx.Done()
return ctx.Err()
}
func appendRuntimeTokenDetachedEvidence(path, line string) error {
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
return err
}
defer f.Close()
_, err = fmt.Fprintln(f, line)
return err
}
func runtimeTokenDetachedContainsCanary(value string) bool {
return strings.Contains(value, runtimeTokenDetachedCanaryA) ||
strings.Contains(value, runtimeTokenDetachedCanaryB)
}
func TestCrossPlatformCoverageUnixDetachedRuntimeTokenLifecycleAndCanaryLeakScan(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("real detached-process lifecycle is Unix-only; Windows named-pipe code is cross-compiled separately")
}
root, err := os.MkdirTemp("/tmp", "dws-runtime-token-e2e-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = os.RemoveAll(root) })
workDir := filepath.Join(root, "events", "open", string(dwsevent.SourceKindPersonalStream), runtimeTokenDetachedIdentityHash)
if err := os.MkdirAll(workDir, 0o700); err != nil {
t.Fatal(err)
}
endpoint := dwsevent.IPCEndpoint(workDir, "open", dwsevent.SourceKindPersonalStream, runtimeTokenDetachedIdentityHash)
evidencePath := filepath.Join(workDir, "runtime-e2e.evidence")
identity := personal.Identity{
ClientID: runtimeTokenDetachedClientID,
SourceID: runtimeTokenDetachedSourceID,
CorpID: "runtime-e2e-corp",
UserID: "runtime-e2e-user",
}
spawnArgs := personalBusSpawnArgsForToken(identity, runtimeTokenDetachedIdentityHash, "", "", "corp:user", runtimeTokenDetachedCanaryA)
assertRuntimeTokenDetachedClean(t, "spawn argv", []byte(strings.Join(spawnArgs, "\x00")))
childEnv := append([]string{}, os.Environ()...)
childEnv = append(childEnv,
runtimeTokenDetachedChildEnv+"=1",
runtimeTokenDetachedWorkDirEnv+"="+workDir,
runtimeTokenDetachedEndpointEnv+"="+endpoint,
runtimeTokenDetachedEvidenceEnv+"="+evidencePath,
)
assertRuntimeTokenDetachedClean(t, "spawn environment", []byte(strings.Join(childEnv, "\x00")))
pid, err := busctl.Spawn(busctl.SpawnConfig{
ExecPath: os.Args[0],
ClientID: runtimeTokenDetachedClientID,
ExtraArgs: spawnArgs,
Env: childEnv,
})
if err != nil {
failRuntimeTokenDetachedError(t, "spawn detached runtime bus", err)
}
stopped := false
t.Cleanup(func() {
if !stopped {
_ = busctl.Stop(busctl.StopConfig{WorkDir: workDir, Timeout: 2 * time.Second})
if proc, findErr := os.FindProcess(pid); findErr == nil {
_ = proc.Kill()
}
}
})
waitRuntimeTokenDetachedFile(t, evidencePath, "child_start argv_clean=true env_clean=true", 3*time.Second)
var stdoutA, stderrA bytes.Buffer
err = consume.Run(context.Background(), runtimeTokenDetachedConsumeConfig(
workDir, endpoint, "sub-runtime-a", runtimeTokenDetachedCanaryA, 500*time.Millisecond, &stdoutA, &stderrA,
))
if err != nil {
failRuntimeTokenDetachedError(t, "consume token A", err)
}
waitRuntimeTokenDetachedFile(t, evidencePath, "resolved_a=true generation=1", 3*time.Second)
if err := personal.UpsertRunState(workDir, personal.RunState{
SubscribeID: "sub-runtime-b",
EventKey: personal.EventMention,
ClientID: runtimeTokenDetachedClientID,
SourceID: runtimeTokenDetachedSourceID,
IdentityHash: runtimeTokenDetachedIdentityHash,
}); err != nil {
failRuntimeTokenDetachedError(t, "persist non-sensitive run state", err)
}
var stdoutB, stderrB bytes.Buffer
consumeDone := make(chan error, 1)
go func() {
consumeDone <- consume.Run(context.Background(), runtimeTokenDetachedConsumeConfig(
workDir, endpoint, "sub-runtime-b", runtimeTokenDetachedCanaryB, 5*time.Second, &stdoutB, &stderrB,
))
}()
status := waitRuntimeTokenDetachedStatus(t, endpoint, "sub-runtime-b", 3*time.Second)
if status.Bus.PID != pid || status.Bus.IdentityHash != runtimeTokenDetachedIdentityHash {
t.Fatalf("status bus identity = %#v, want pid=%d identity=%s", status.Bus, pid, runtimeTokenDetachedIdentityHash)
}
waitRuntimeTokenDetachedFile(t, evidencePath, "rejected_a=true resolved_b=true reconnect=true generation=2", 3*time.Second)
stopResp, err := busctl.StopConsumers(endpoint, []string{"sub-runtime-b"})
if err != nil {
failRuntimeTokenDetachedError(t, "targeted consumer stop", err)
}
if len(stopResp.Stopped) != 1 || stopResp.Stopped[0] != "sub-runtime-b" {
t.Fatalf("targeted stop response = %#v", stopResp)
}
select {
case err := <-consumeDone:
if err != nil {
failRuntimeTokenDetachedError(t, "consume token B after targeted stop", err)
}
case <-time.After(3 * time.Second):
t.Fatal("token B consumer did not exit after targeted stop")
}
status = waitRuntimeTokenDetachedStatus(t, endpoint, "", 3*time.Second)
if len(status.Consumers) != 0 {
t.Fatalf("status consumers after stop = %#v", status.Consumers)
}
if err := busctl.Stop(busctl.StopConfig{WorkDir: workDir, Timeout: 4 * time.Second}); err != nil {
failRuntimeTokenDetachedError(t, "stop detached bus", err)
}
stopped = true
waitRuntimeTokenDetachedFile(t, evidencePath, "bus_exit clean=true", 3*time.Second)
statusJSON, err := json.Marshal(status)
if err != nil {
t.Fatal(err)
}
stopJSON, err := json.Marshal(stopResp)
if err != nil {
t.Fatal(err)
}
for name, artifact := range map[string][]byte{
"consume A stdout": stdoutA.Bytes(),
"consume A stderr": stderrA.Bytes(),
"consume B stdout": stdoutB.Bytes(),
"consume B stderr": stderrB.Bytes(),
"status response": statusJSON,
"stop response": stopJSON,
} {
assertRuntimeTokenDetachedClean(t, name, artifact)
}
assertRuntimeTokenDetachedTreeClean(t, root)
for _, required := range []string{
filepath.Join(workDir, bus.MetaFileName),
filepath.Join(workDir, "bus.log"),
filepath.Join(workDir, personal.StateFileName),
evidencePath,
} {
if info, statErr := os.Stat(required); statErr != nil || !info.Mode().IsRegular() {
t.Fatalf("expected runtime artifact %s: info=%v err=%v", required, info, statErr)
}
}
}
func runtimeTokenDetachedConsumeConfig(workDir, endpoint, subscribeID, token string, duration time.Duration, stdout, stderr *bytes.Buffer) consume.Config {
return consume.Config{
WorkDir: workDir,
IPCEndpoint: endpoint,
ClientID: runtimeTokenDetachedClientID,
RuntimeToken: token,
EventTypes: []string{personal.EventMention},
EventKey: personal.EventMention,
SubscribeID: subscribeID,
ReadySubscribeID: subscribeID,
Duration: duration,
Format: consume.FormatNDJSON,
Stdout: stdout,
Stderr: stderr,
}
}
func waitRuntimeTokenDetachedFile(t *testing.T, path, want string, timeout time.Duration) string {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
data, err := os.ReadFile(path)
if err == nil && strings.Contains(string(data), want) {
return string(data)
}
time.Sleep(10 * time.Millisecond)
}
data, err := os.ReadFile(path)
if runtimeTokenDetachedContainsCanary(string(data)) {
t.Fatalf("runtime credential leaked into child evidence while waiting for %q", want)
}
t.Fatalf("evidence %s missing %q: data=%q err=%v", path, want, data, err)
return ""
}
func waitRuntimeTokenDetachedStatus(t *testing.T, endpoint, subscribeID string, timeout time.Duration) *transport.StatusResp {
t.Helper()
deadline := time.Now().Add(timeout)
var lastErr error
for time.Now().Before(deadline) {
status, err := busctl.QueryStatus(endpoint)
if err == nil {
if subscribeID == "" && len(status.Consumers) == 0 {
return status
}
for _, consumer := range status.Consumers {
if consumer.SubscribeID == subscribeID {
return status
}
}
}
lastErr = err
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("status never reached subscribe_id=%q: %v", subscribeID, lastErr)
return nil
}
func assertRuntimeTokenDetachedTreeClean(t *testing.T, root string) {
t.Helper()
err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
if walkErr != nil {
return walkErr
}
if entry.IsDir() || !entry.Type().IsRegular() {
return nil
}
data, err := os.ReadFile(path)
if err != nil {
return err
}
assertRuntimeTokenDetachedClean(t, path, data)
return nil
})
if err != nil {
t.Fatalf("scan runtime artifacts: %v", err)
}
}
func assertRuntimeTokenDetachedClean(t *testing.T, name string, artifact []byte) {
t.Helper()
if runtimeTokenDetachedContainsCanary(string(artifact)) {
t.Fatalf("runtime credential leaked into %s", name)
}
}
func failRuntimeTokenDetachedError(t *testing.T, step string, err error) {
t.Helper()
if err != nil && runtimeTokenDetachedContainsCanary(err.Error()) {
t.Fatalf("%s failed and exposed a runtime credential", step)
}
t.Fatalf("%s: %v", step, err)
}
@@ -0,0 +1,335 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
package app
import (
"context"
"errors"
"io"
"net/http"
"strings"
"testing"
"time"
authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
)
func TestCrossPlatformCoverageRuntimeTokenBusRejectsIncompleteIdentity(t *testing.T) {
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
for _, tc := range []struct {
name string
args []string
want string
}{
{
name: "invalid identity hash",
args: []string{"--source-kind", "personal_stream", "--runtime-token-mode", "--identity-hash", "not-a-hash", "--client-id", "client"},
want: "16-character hexadecimal identity hash",
},
{
name: "missing client id",
args: []string{"--source-kind", "personal_stream", "--runtime-token-mode", "--identity-hash", "0123456789abcdef"},
want: "--client-id is required",
},
} {
t.Run(tc.name, func(t *testing.T) {
cmd := newEventBusCommand()
cmd.SetOut(io.Discard)
cmd.SetErr(io.Discard)
cmd.SetArgs(tc.args)
err := cmd.Execute()
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("Execute() error = %v, want %q", err, tc.want)
}
})
}
}
type eventRuntimeTokenReleaseErrorStore struct {
err error
}
func (*eventRuntimeTokenReleaseErrorStore) Claim([]personal.AttemptSpec, time.Duration) (*personal.AttemptClaim, error) {
return nil, nil
}
func (*eventRuntimeTokenReleaseErrorStore) CompleteSuccess(*personal.AttemptClaim) error {
return nil
}
func (*eventRuntimeTokenReleaseErrorStore) CompleteFailure(*personal.AttemptClaim, []string, personal.AttemptFailure) (personal.AttemptHold, error) {
return personal.AttemptHold{}, nil
}
func (s *eventRuntimeTokenReleaseErrorStore) Release(*personal.AttemptClaim) error {
return s.err
}
func TestCrossPlatformCoverageRuntimeTokenAttemptReleaseGuardEdges(t *testing.T) {
var nilReservation *personalSubscriptionAttemptReservation
if err := nilReservation.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("nil reservation error = %v", err)
}
incomplete := &personalSubscriptionAttemptReservation{}
if err := incomplete.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) ||
!strings.Contains(err.Error(), "reservation is incomplete") {
t.Fatalf("incomplete reservation error = %v", err)
}
wantErr := errors.New("release failed")
reservation := &personalSubscriptionAttemptReservation{
store: &eventRuntimeTokenReleaseErrorStore{err: wantErr},
claim: &personal.AttemptClaim{AttemptID: "attempt"},
}
if err := reservation.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) ||
!errors.Is(err, wantErr) {
t.Fatalf("release failure error = %v", err)
}
}
func TestCrossPlatformCoverageRuntimeTokenConsumeRejectionAndOversizeEdges(t *testing.T) {
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
oldEdition := edition.Get()
oldProfile := authpkg.RuntimeProfile()
oldLoadProfiles := personalLoadProfiles
oldValidate := personalValidateConsumeConfig
oldConflict := personalValidateNoOutputConflict
oldAttemptStore := personalNewSubscriptionAttemptStore
oldEnsure := personalEnsureSubscription
t.Cleanup(func() {
edition.Override(oldEdition)
authpkg.SetRuntimeProfile(oldProfile)
personalLoadProfiles = oldLoadProfiles
personalValidateConsumeConfig = oldValidate
personalValidateNoOutputConflict = oldConflict
personalNewSubscriptionAttemptStore = oldAttemptStore
personalEnsureSubscription = oldEnsure
})
edition.Override(&edition.Hooks{})
authpkg.SetRuntimeProfile("")
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
personalValidateConsumeConfig = func(consume.Config) error { return nil }
personalValidateNoOutputConflict = func(consume.Config, string) error { return nil }
oversized := strings.Repeat("x", runtimecred.DefaultMaxTokenBytes+1)
err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
EventKey: personal.EventMention,
ExplicitToken: oversized,
ClientIDOverride: "runtime-client",
Common: commonConsumeOptions{Foreground: true},
})
if !errors.Is(err, runtimecred.ErrTokenTooLarge) {
t.Fatalf("oversized foreground token error = %v", err)
}
rejection := &personal.APIError{
Code: "RUNTIME_TOKEN_REJECTED",
HTTPStatus: http.StatusUnauthorized,
}
if personalRuntimeTokenControlRejection(errors.New("ordinary failure")) {
t.Fatal("ordinary error classified as runtime-token rejection")
}
singleStore := &personalRecordingAttemptStore{}
personalNewSubscriptionAttemptStore = func(string) personalSubscriptionAttemptStore { return singleStore }
personalEnsureSubscription = func(context.Context, *personal.Client, personal.Identity, personalConsumeOptions) (*personal.Subscription, string, string, error) {
return nil, "", "", rejection
}
err = runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
EventKey: personal.EventMention,
ExplicitToken: "runtime-token-single",
ClientIDOverride: "runtime-client",
ControlBaseURL: "https://control.example.test",
})
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) || singleStore.releaseCalls != 1 || singleStore.failureCalls != 0 {
t.Fatalf("single rejection = %v, release=%d failure=%d", err, singleStore.releaseCalls, singleStore.failureCalls)
}
manyStore := &personalRecordingAttemptStore{}
personalNewSubscriptionAttemptStore = func(string) personalSubscriptionAttemptStore { return manyStore }
err = runPersonalEventConsumeMany(newPersonalCoverageCommand(), personalConsumeOptions{
EventKeys: []string{personal.EventMention, personal.EventAllSingleChat},
ExplicitToken: "runtime-token-many",
ClientIDOverride: "runtime-client",
ControlBaseURL: "https://control.example.test",
})
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) || manyStore.releaseCalls != 1 || manyStore.failureCalls != 0 {
t.Fatalf("multi rejection = %v, release=%d failure=%d", err, manyStore.releaseCalls, manyStore.failureCalls)
}
}
func TestCrossPlatformCoverageRuntimeTokenIdentityFallbackEdges(t *testing.T) {
configDir := t.TempDir()
oldEdition := edition.Get()
oldProfile := authpkg.RuntimeProfile()
oldResolveIdentity := personalResolveEventIdentity
oldResolveAuxiliary := personalResolveAuxiliaryAccessToken
oldLoadTokenData := personalLoadTokenData
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
oldClientID := personalClientID
oldResolveCredentials := personalResolveAppCredentialsStrict
t.Cleanup(func() {
edition.Override(oldEdition)
authpkg.SetRuntimeProfile(oldProfile)
personalResolveEventIdentity = oldResolveIdentity
personalResolveAuxiliaryAccessToken = oldResolveAuxiliary
personalLoadTokenData = oldLoadTokenData
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
personalClientID = oldClientID
personalResolveAppCredentialsStrict = oldResolveCredentials
})
legacy := personal.Identity{ClientID: "legacy-client", SourceID: "legacy-source"}
personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) { return legacy, nil }
identity, err := resolvePersonalEventIdentityForToken(context.Background(), configDir, "", " ")
if err != nil || identity.ClientID != legacy.ClientID {
t.Fatalf("wrapper empty-token fallback = %#v, %v", identity, err)
}
personalResolveAuxiliaryAccessToken = func(context.Context, string, string) (string, error) {
return "legacy-access", nil
}
personalLoadTokenData = func(string) (*authpkg.TokenData, error) {
return &authpkg.TokenData{
CorpID: "legacy-corp", UserID: "legacy-user", ClientID: "direct-client",
}, nil
}
identity, err = resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", " ")
if err != nil || identity.ClientID != "direct-client" {
t.Fatalf("direct empty-token fallback = %#v, %v", identity, err)
}
edition.Override(&edition.Hooks{})
personalRuntimeEventClientID = func() string { return "" }
personalClientID = func() string { return "" }
wantMetadataErr := errors.New("profiles unreadable")
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, wantMetadataErr }
authpkg.SetRuntimeProfile("corp:user")
if _, err := resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "token", "runtime-client"); !errors.Is(err, wantMetadataErr) {
t.Fatalf("explicit profile metadata error = %v", err)
}
authpkg.SetRuntimeProfile("")
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
return "app-client", "", "", "", nil
}
identity, err = resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "runtime-token")
if err != nil || identity.ClientID != "app-client" || !strings.HasPrefix(identity.LocalSubject, "access:") {
t.Fatalf("app-credential fallback identity = %#v, %v", identity, err)
}
personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
return "", "", "", "", errors.New("missing app credentials")
}
if _, err := resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "runtime-token"); err == nil || !strings.Contains(err.Error(), "cannot resolve OAuth client_id") {
t.Fatalf("missing client ID error = %v", err)
}
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{
CurrentProfile: "stale",
Profiles: []authpkg.Profile{{Name: "other", CorpID: "corp", UserID: "user"}},
}, nil
}
profile, err := personalEventProfileMetadata(configDir)
if err != nil || profile != nil {
t.Fatalf("stale implicit current profile = %#v, %v", profile, err)
}
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{Profiles: []authpkg.Profile{{Name: "other"}}}, nil
}
profile, err = personalEventProfileMetadata(configDir)
if err != nil || profile != nil {
t.Fatalf("empty implicit selector = %#v, %v", profile, err)
}
authpkg.SetRuntimeProfile("corp-a:user-a,corp-b:user-b")
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
if _, err := personalEventProfileMetadata(configDir); err == nil || !strings.Contains(err.Error(), "exactly one --profile") {
t.Fatalf("multi-profile metadata error = %v", err)
}
}
type eventRuntimeTokenReadErrorBody struct{}
func (eventRuntimeTokenReadErrorBody) Read([]byte) (int, error) {
return 0, errors.New("body read failed")
}
func (eventRuntimeTokenReadErrorBody) Close() error { return nil }
func TestCrossPlatformCoverageRuntimeTokenControlTransportErrorEdges(t *testing.T) {
const token = "runtime-control-edge-canary"
unsupported, err := http.NewRequest(http.MethodGet, "unsupported://control.example.test/path", nil)
if err != nil {
t.Fatal(err)
}
if _, err := (runtimeTokenControlTransport{}).RoundTrip(unsupported); err == nil {
t.Fatal("nil base unexpectedly accepted an unsupported protocol")
}
wantTransportErr := errors.New("ordinary transport failure")
ordinary := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
return nil, wantTransportErr
})}
if _, err := ordinary.RoundTrip(unsupported); !errors.Is(err, wantTransportErr) {
t.Fatalf("ordinary transport error = %v", err)
}
leaking := runtimeTokenControlTransport{
token: token,
base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
return nil, errors.New("reflected " + token)
}),
}
if _, err := leaking.RoundTrip(unsupported); err == nil || strings.Contains(err.Error(), token) ||
err.Error() != "personal event: runtime-token control request failed" {
t.Fatalf("redacted transport error = %v", err)
}
nilResponse := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
return nil, nil
})}
if resp, err := nilResponse.RoundTrip(unsupported); resp != nil || err != nil {
t.Fatalf("nil response = %#v, %v", resp, err)
}
readFailure := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusInternalServerError,
Header: make(http.Header),
Body: eventRuntimeTokenReadErrorBody{},
Request: req,
}, nil
})}
if _, err := readFailure.RoundTrip(unsupported); err == nil || !strings.Contains(err.Error(), "read runtime-token control response") {
t.Fatalf("body read error = %v", err)
}
nilHeader := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{StatusCode: http.StatusOK, Request: req}, nil
})}
resp, err := nilHeader.RoundTrip(unsupported)
if err != nil || resp == nil || resp.Header == nil || resp.Header.Get("Content-Type") != "application/json" {
t.Fatalf("nil-header response = %#v, %v", resp, err)
}
if got := redactRuntimeTokenResponseBody(nil, token); len(got) != 0 {
t.Fatalf("empty response redaction = %q", got)
}
value, changed := redactRuntimeTokenJSONValue([]any{"plain", "prefix-" + token}, token)
items, ok := value.([]any)
if !ok || !changed || len(items) != 2 || strings.Contains(items[1].(string), token) {
t.Fatalf("array redaction = %#v changed=%t", value, changed)
}
}
@@ -0,0 +1,917 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
package app
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
"github.com/spf13/cobra"
)
func TestCrossPlatformCoverageEventCommandsWireTrimmedRootRuntimeToken(t *testing.T) {
oldConsume := eventRunPersonalConsume
oldStatus := eventRunPersonalStatus
oldStop := eventRunPersonalStop
t.Cleanup(func() {
eventRunPersonalConsume = oldConsume
eventRunPersonalStatus = oldStatus
eventRunPersonalStop = oldStop
})
flags := &GlobalFlags{Token: " runtime-canary ", ClientID: " root-client "}
assertIdentity := func(token, clientID string) {
t.Helper()
if token != "runtime-canary" || clientID != "root-client" {
t.Fatalf("runtime identity = token %q client %q", token, clientID)
}
}
eventRunPersonalConsume = func(_ *cobra.Command, opts personalConsumeOptions) error {
assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
return nil
}
consumeCmd := newEventConsumeCommand(flags)
if err := consumeCmd.RunE(consumeCmd, []string{personal.EventMention}); err != nil {
t.Fatalf("consume RunE() error = %v", err)
}
eventRunPersonalStatus = func(_ *cobra.Command, opts personalStatusOptions) error {
assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
return nil
}
statusCmd := newEventStatusCommandWithFlags(flags)
if err := statusCmd.RunE(statusCmd, nil); err != nil {
t.Fatalf("status RunE() error = %v", err)
}
eventRunPersonalStop = func(_ *cobra.Command, opts personalStopOptions) error {
assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
return nil
}
stopCmd := newEventStopCommandWithFlags(flags)
stopRoot := &cobra.Command{Use: "dws"}
stopRoot.PersistentFlags().Bool("yes", true, "")
stopRoot.AddCommand(stopCmd)
if err := stopCmd.RunE(stopCmd, []string{"sub-runtime"}); err != nil {
t.Fatalf("stop RunE() error = %v", err)
}
listenCmd := newEventListenIMCommand(flags)
if err := listenCmd.RunE(listenCmd, nil); err != nil {
t.Fatalf("listen-im RunE() error = %v", err)
}
}
func TestCrossPlatformCoverageEventConsumeParsesRootRuntimeTokenBeforeAndAfterSubcommand(t *testing.T) {
oldConsume := eventRunPersonalConsume
t.Cleanup(func() { eventRunPersonalConsume = oldConsume })
for _, tc := range []struct {
name string
args []string
}{
{name: "before", args: []string{"--token", "runtime-before", "event", "consume", personal.EventMention}},
{name: "after", args: []string{"event", "consume", personal.EventMention, "--token", "runtime-after"}},
} {
t.Run(tc.name, func(t *testing.T) {
flags := &GlobalFlags{}
root := &cobra.Command{Use: "dws", SilenceErrors: true, SilenceUsage: true}
bindPersistentFlags(root, flags)
root.AddCommand(newEventCommand(flags))
var got string
eventRunPersonalConsume = func(_ *cobra.Command, opts personalConsumeOptions) error {
got = opts.ExplicitToken
return nil
}
root.SetArgs(tc.args)
if err := root.Execute(); err != nil {
t.Fatalf("Execute() error = %v", err)
}
want := "runtime-" + tc.name
if got != want {
t.Fatalf("ExplicitToken = %q, want %q", got, want)
}
})
}
}
func TestCrossPlatformCoverageResolvePersonalEventIdentityWithTokenUsesMetadataOnly(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldLoadToken := personalLoadTokenData
oldAux := personalResolveAuxiliaryAccessToken
oldClientID := personalClientID
oldCredentials := personalResolveAppCredentialsStrict
previousProfile := authpkg.RuntimeProfile()
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalLoadTokenData = oldLoadToken
personalResolveAuxiliaryAccessToken = oldAux
personalClientID = oldClientID
personalResolveAppCredentialsStrict = oldCredentials
authpkg.SetRuntimeProfile(previousProfile)
})
personalLoadTokenData = func(string) (*authpkg.TokenData, error) {
t.Fatal("explicit token identity read sensitive TokenData")
return nil, nil
}
personalResolveAuxiliaryAccessToken = func(context.Context, string, string) (string, error) {
t.Fatal("explicit token identity resolved local OAuth")
return "", nil
}
personalClientID = func() string {
t.Fatal("explicit root client ID was not preferred")
return ""
}
personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
t.Fatal("explicit root client ID unexpectedly fell back to app credentials")
return "", "", "", "", nil
}
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{
Version: 2,
Profiles: []authpkg.Profile{{
Name: "Runtime profile",
CorpID: "profile-corp",
CorpName: "Runtime Org",
UserID: "profile-user",
UserName: "Runtime User",
ClientID: "profile-client",
}},
}, nil
}
authpkg.SetRuntimeProfile("Runtime Org:Runtime User")
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "", false },
}
}})
identity, err := resolvePersonalEventIdentityWithToken(
context.Background(), "unused", "runtime-source", " runtime-canary ", "root-client",
)
if err != nil {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
if identity.CorpID != "runtime-corp" || identity.UserID != "profile-user" || identity.ClientID != "root-client" {
t.Fatalf("identity metadata = %#v", identity)
}
if identity.AccessToken != "" {
t.Fatalf("identity retained raw runtime token: %q", identity.AccessToken)
}
if identity.LocalSubject != "" {
t.Fatalf("complete identity LocalSubject = %q, want empty", identity.LocalSubject)
}
}
func TestCrossPlatformCoverageResolvePersonalEventIdentityWithCompleteRuntimeMetadataSkipsProfiles(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
})
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
t.Fatal("complete host metadata unexpectedly read profiles.json")
return nil, nil
}
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
personalRuntimeEventClientID = func() string { return "edition-client" }
identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary", "root-client")
if err != nil {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "root-client" {
t.Fatalf("identity = %#v", identity)
}
}
func TestCrossPlatformCoverageRuntimeEventClientIDPrefersEditionBeforeEnvironment(t *testing.T) {
oldEdition := edition.Get()
t.Cleanup(func() { edition.Override(oldEdition) })
t.Setenv("DWS_CLIENT_ID", "environment-client")
edition.Override(&edition.Hooks{AuthClientID: "edition-client"})
if got := runtimePersonalEventClientID(); got != "edition-client" {
t.Fatalf("runtime client ID = %q, want edition hook", got)
}
edition.Override(&edition.Hooks{})
if got := runtimePersonalEventClientID(); got != "environment-client" {
t.Fatalf("runtime client ID = %q, want environment fallback", got)
}
}
func TestCrossPlatformCoverageCompleteRuntimeIdentityUsesEditionClientBeforeProfiles(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
})
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
t.Fatal("complete host metadata unexpectedly read profiles.json")
return nil, errors.New("unreachable")
}
personalRuntimeEventClientID = func() string { return "edition-client" }
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
if err != nil {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "edition-client" {
t.Fatalf("identity = %#v", identity)
}
}
func TestCrossPlatformCoverageSelectedProfileClientPrecedesPersistedGlobalClient(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
oldClientID := personalClientID
previousProfile := authpkg.RuntimeProfile()
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
personalClientID = oldClientID
authpkg.SetRuntimeProfile(previousProfile)
})
edition.Override(&edition.Hooks{})
personalRuntimeEventClientID = func() string { return "" }
personalClientID = func() string { return "stale-global-client" }
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{
Version: 3,
CurrentProfile: "corp:user",
Profiles: []authpkg.Profile{{
Name: "Selected", CorpID: "corp", UserID: "user", ClientID: "profile-client",
}},
}, nil
}
authpkg.SetRuntimeProfile("corp:user")
identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
if err != nil {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
if identity.ClientID != "profile-client" {
t.Fatalf("ClientID = %q, want selected profile client", identity.ClientID)
}
}
func TestCrossPlatformCoverageMalformedPersistedProfilesDoNotBlockRuntimeDefaultsAndGlobalClient(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
oldClientID := personalClientID
previousProfile := authpkg.RuntimeProfile()
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
personalClientID = oldClientID
authpkg.SetRuntimeProfile(previousProfile)
})
authpkg.SetRuntimeProfile("")
personalRuntimeEventClientID = func() string { return "" }
personalClientID = func() string { return "global-client" }
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return nil, errors.New("malformed persisted profiles")
}
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
if err != nil {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "global-client" {
t.Fatalf("identity = %#v", identity)
}
}
func TestCrossPlatformCoverageResolvePersonalEventIdentityWithTokenRejectsMultipleProfilesBeforeMetadata(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
previousProfile := authpkg.RuntimeProfile()
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
authpkg.SetRuntimeProfile(previousProfile)
})
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
t.Fatal("multiple runtime profiles unexpectedly reached metadata loading")
return nil, nil
}
authpkg.SetRuntimeProfile("corp-a:user-a,corp-b:user-b")
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
_, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary", "root-client")
if err == nil || !strings.Contains(err.Error(), "exactly one --profile") {
t.Fatalf("multiple-profile error = %v", err)
}
}
func TestCrossPlatformCoverageExplicitProfileRequiresMetadataRegistry(t *testing.T) {
oldLoadProfiles := personalLoadProfiles
defer func() { personalLoadProfiles = oldLoadProfiles }()
oldProfile := authpkg.RuntimeProfile()
authpkg.SetRuntimeProfile("missing-profile")
defer authpkg.SetRuntimeProfile(oldProfile)
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{}, nil
}
_, err := personalEventProfileMetadata(t.TempDir())
if err == nil || !strings.Contains(err.Error(), `profile "missing-profile" not found`) {
t.Fatalf("personalEventProfileMetadata() error = %v", err)
}
}
func TestCrossPlatformCoverageCompleteRuntimeIdentityStillValidatesExplicitProfile(t *testing.T) {
oldEdition := edition.Get()
oldLoadProfiles := personalLoadProfiles
oldRuntimeClientID := personalRuntimeEventClientID
oldProfile := authpkg.RuntimeProfile()
t.Cleanup(func() {
edition.Override(oldEdition)
personalLoadProfiles = oldLoadProfiles
personalRuntimeEventClientID = oldRuntimeClientID
authpkg.SetRuntimeProfile(oldProfile)
})
authpkg.SetRuntimeProfile("missing-profile")
personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
return &authpkg.ProfilesConfig{}, nil
}
personalRuntimeEventClientID = func() string { return "runtime-client" }
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
_, err := resolvePersonalEventIdentityWithToken(context.Background(), t.TempDir(), "source", "canary")
if err == nil || !strings.Contains(err.Error(), `profile "missing-profile" not found`) {
t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
}
}
func TestCrossPlatformCoveragePersonalProfileMetadataOrganizationCurrentBeatsUnresolved(t *testing.T) {
cfg := &authpkg.ProfilesConfig{
Version: 3,
Profiles: []authpkg.Profile{
{Name: "Historical", CorpID: "corp-1"},
{Name: "Exact", CorpID: "corp-1", UserID: "user-1"},
},
OrgCurrentProfiles: map[string]string{"corp-1": "corp-1:user-1"},
}
profile, err := selectPersonalEventProfileMetadata(cfg, "corp-1", make(map[string]struct{}))
if err != nil {
t.Fatalf("selectPersonalEventProfileMetadata() error = %v", err)
}
if profile == nil || profile.UserID != "user-1" {
t.Fatalf("selected profile = %#v, want organization current account", profile)
}
}
func TestCrossPlatformCoverageExplicitTokenControlClientRedactsReflected401(t *testing.T) {
const token = "runtime-control-canary"
oldLogger := slog.Default()
var logs bytes.Buffer
slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
t.Cleanup(func() { slog.SetDefault(oldLogger) })
client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
ClientID: "client", SourceID: "source",
}, token)
wrapped, ok := client.HTTPClient.Transport.(runtimeTokenControlTransport)
if !ok {
t.Fatalf("control transport = %T, want runtimeTokenControlTransport", client.HTTPClient.Transport)
}
var authorization string
wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
authorization = req.Header.Get("Authorization")
body := `{"code":"UNAUTHORIZED","message":"rejected ` + token + `"}`
header := make(http.Header)
header.Set("X-Request-Id", "request-"+token)
header.Set("X-Trace-Id", "trace-"+token)
return &http.Response{
StatusCode: http.StatusUnauthorized,
Header: header,
Body: io.NopCloser(strings.NewReader(body)),
Request: req,
}, nil
})
client.HTTPClient.Transport = wrapped
_, err := client.ListSubscriptions(context.Background(), personal.ListOptions{})
if err == nil {
t.Fatal("ListSubscriptions() unexpectedly succeeded")
}
if authorization != "Bearer "+token {
t.Fatalf("Authorization = %q", authorization)
}
if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
t.Fatalf("runtime token leaked: error=%q logs=%q", err, logs.String())
}
if !strings.Contains(err.Error(), "RUNTIME_TOKEN_REJECTED") {
t.Fatalf("error = %q, want fixed runtime token rejection", err)
}
}
func TestCrossPlatformCoverageRuntimeTokenRedirectGuardDoesNotForwardCustomHeader(t *testing.T) {
const token = "runtime-redirect-canary"
var controlTargetHits, ticketTargetHits atomic.Int32
controlTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
controlTargetHits.Add(1)
if r.Header.Get("x-user-access-token") == token {
t.Error("control redirect forwarded runtime token")
}
w.WriteHeader(http.StatusNoContent)
}))
defer controlTarget.Close()
controlOrigin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("x-user-access-token") != token {
t.Error("control origin did not receive runtime token")
}
http.Redirect(w, r, controlTarget.URL, http.StatusFound)
}))
defer controlOrigin.Close()
client := newPersonalEventControlClient("unused", controlOrigin.URL, personal.Identity{
ClientID: "client", SourceID: "source",
}, token)
_, controlErr := client.ListSubscriptions(context.Background(), personal.ListOptions{})
if controlErr == nil {
t.Fatal("cross-host control redirect unexpectedly succeeded")
}
if controlTargetHits.Load() != 0 || strings.Contains(controlErr.Error(), token) {
t.Fatalf("control redirect hits=%d error=%q", controlTargetHits.Load(), controlErr)
}
ticketTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
ticketTargetHits.Add(1)
if r.Header.Get("x-user-access-token") == token {
t.Error("ticket redirect forwarded runtime token")
}
w.WriteHeader(http.StatusNoContent)
}))
defer ticketTarget.Close()
ticketOrigin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("x-user-access-token") != token {
t.Error("ticket origin did not receive runtime token")
}
http.Redirect(w, r, ticketTarget.URL, http.StatusFound)
}))
defer ticketOrigin.Close()
broker := runtimecred.New(runtimecred.Config{RequireSeed: true})
if _, err := broker.Update(0, token); err != nil {
t.Fatal(err)
}
src, err := newPersonalStreamSource(context.Background(), personalStreamSourceOptions{
ConfigDir: "unused",
Identity: personal.Identity{ClientID: "client", SourceID: "source"},
TicketURL: ticketOrigin.URL,
CredentialBroker: broker,
RuntimeTokenMode: true,
})
if err != nil {
t.Fatal(err)
}
err = src.Start(context.Background(), func(*dwsevent.RawEvent) {})
if err == nil {
t.Fatal("cross-host ticket redirect unexpectedly succeeded")
}
if ticketTargetHits.Load() != 0 || strings.Contains(err.Error(), token) {
t.Fatalf("ticket redirect hits=%d error=%q", ticketTargetHits.Load(), err)
}
}
func TestCrossPlatformCoverageRuntimeTokenRedirectPolicyBranches(t *testing.T) {
origin, _ := http.NewRequest(http.MethodGet, "https://control.example/start", nil)
sameHost, _ := http.NewRequest(http.MethodGet, "https://control.example/next", nil)
if err := runtimeTokenRedirectPolicy(sameHost, []*http.Request{origin}); err != nil {
t.Fatalf("same-host HTTPS redirect rejected: %v", err)
}
for name, request := range map[string]*http.Request{
"cross-host": func() *http.Request {
r, _ := http.NewRequest(http.MethodGet, "https://other.example/next", nil)
return r
}(),
"downgrade": func() *http.Request {
r, _ := http.NewRequest(http.MethodGet, "http://control.example/next", nil)
return r
}(),
} {
if err := runtimeTokenRedirectPolicy(request, []*http.Request{origin}); !errors.Is(err, http.ErrUseLastResponse) {
t.Fatalf("%s redirect policy error = %v", name, err)
}
}
if err := runtimeTokenRedirectPolicy(nil, nil); !errors.Is(err, http.ErrUseLastResponse) {
t.Fatalf("empty redirect chain error = %v", err)
}
}
func TestCrossPlatformCoverageExplicitTokenControlClientRedactsEveryErrorEnvelope(t *testing.T) {
const token = "runtime-control-all-status-canary"
tests := []struct {
name string
status int
body string
}{
{name: "bad-request", status: http.StatusBadRequest, body: `{"code":"BAD_REQUEST","message":"` + token + `"}`},
{name: "server-error", status: http.StatusInternalServerError, body: `{"code":"INTERNAL","message":"` + token + `"}`},
{name: "success-false", status: http.StatusOK, body: `{"success":false,"errorCode":"DENIED","errorMsg":"` + token + `"}`},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
oldLogger := slog.Default()
var logs bytes.Buffer
slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
t.Cleanup(func() { slog.SetDefault(oldLogger) })
client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
ClientID: "client", SourceID: "source",
}, token)
wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
header := make(http.Header)
header.Set("X-Request-Id", "request-"+token)
header.Set("X-Trace-Id", "trace-"+token)
return &http.Response{
StatusCode: tc.status,
Header: header,
Body: io.NopCloser(strings.NewReader(tc.body)),
Request: req,
}, nil
})
client.HTTPClient.Transport = wrapped
_, err := client.ListSubscriptions(context.Background(), personal.ListOptions{})
if err == nil {
t.Fatal("ListSubscriptions() unexpectedly succeeded")
}
if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
t.Fatalf("runtime token leaked: error=%q logs=%q", err, logs.String())
}
})
}
}
func TestCrossPlatformCoverageExplicitTokenControlTransportPreservesSuccessfulResponse(t *testing.T) {
client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
ClientID: "client", SourceID: "source",
}, "runtime-success-canary")
wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusOK,
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(`{"success":true,"result":{"items":[],"total":0}}`)),
Request: req,
}, nil
})
client.HTTPClient.Transport = wrapped
if _, err := client.ListSubscriptions(context.Background(), personal.ListOptions{}); err != nil {
t.Fatalf("ListSubscriptions() successful response error = %v", err)
}
}
func TestCrossPlatformCoverageExplicitTokenControlClientRedactsJSONEscapedToken(t *testing.T) {
const token = "runtime<escaped>&canary"
body, err := json.Marshal(map[string]any{"code": "BAD_REQUEST", "message": "rejected " + token})
if err != nil {
t.Fatal(err)
}
if bytes.Contains(body, []byte(token)) {
t.Fatalf("fixture was not JSON-escaped: %s", body)
}
oldLogger := slog.Default()
var logs bytes.Buffer
slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
t.Cleanup(func() { slog.SetDefault(oldLogger) })
client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
ClientID: "client", SourceID: "source",
}, token)
wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusBadRequest,
Header: make(http.Header),
Body: io.NopCloser(bytes.NewReader(body)),
Request: req,
}, nil
})
client.HTTPClient.Transport = wrapped
_, err = client.ListSubscriptions(context.Background(), personal.ListOptions{})
if err == nil {
t.Fatal("ListSubscriptions() unexpectedly succeeded")
}
if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
t.Fatalf("escaped runtime token leaked: error=%q logs=%q", err, logs.String())
}
}
func TestCrossPlatformCoverageRuntimeTokenBusModeSkipsLocalOAuthIdentity(t *testing.T) {
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
oldResolve := eventResolvePersonal
oldSource := eventNewPersonalSource
oldRun := eventBusRun
t.Cleanup(func() {
eventResolvePersonal = oldResolve
eventNewPersonalSource = oldSource
eventBusRun = oldRun
})
resolvedLocal := false
eventResolvePersonal = func(context.Context, string, string) (personal.Identity, error) {
resolvedLocal = true
return personal.Identity{}, nil
}
var sourceOpts personalStreamSourceOptions
eventNewPersonalSource = func(_ context.Context, opts personalStreamSourceOptions) (*source.PersonalSource, error) {
sourceOpts = opts
return nil, nil
}
var busCfg bus.Config
eventBusRun = func(_ context.Context, cfg bus.Config) error {
busCfg = cfg
return nil
}
cmd := newEventBusCommand()
cmd.SetArgs([]string{
"--source-kind", "personal_stream",
"--runtime-token-mode",
"--identity-hash", "0123456789abcdef",
"--client-id", "runtime-client",
"--stream-source-id", "runtime-source",
"--idle-timeout", "0",
})
if err := cmd.Execute(); err != nil {
t.Fatalf("event _bus runtime mode error = %v", err)
}
if resolvedLocal {
t.Fatal("runtime token bus resolved local OAuth identity")
}
if sourceOpts.CredentialBroker == nil || busCfg.CredentialBroker != sourceOpts.CredentialBroker {
t.Fatal("personal source and bus did not share one credential broker")
}
if busCfg.IdentityHash != "0123456789abcdef" || busCfg.ClientID != "runtime-client" || busCfg.SourceID != "runtime-source" {
t.Fatalf("bus identity = %#v", busCfg)
}
generation, err := sourceOpts.CredentialBroker.Update(0, "detached-activation-canary")
if err != nil {
t.Fatalf("seed detached broker: %v", err)
}
waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
if _, err := sourceOpts.CredentialBroker.Resolve(waitCtx); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("detached broker resolved before consumer activation: %v", err)
}
if _, err := sourceOpts.CredentialBroker.Activate(generation); err != nil {
t.Fatalf("activate detached broker: %v", err)
}
if resolved, err := sourceOpts.CredentialBroker.Resolve(context.Background()); err != nil || resolved == "" {
t.Fatalf("detached broker did not resolve after activation: %v", err)
}
}
func TestCrossPlatformCoverageForegroundRuntimeBrokerDoesNotRequireActivation(t *testing.T) {
broker := newPersonalCredentialBroker(t.TempDir(), true, false)
if _, err := broker.Update(0, "foreground-activation-canary"); err != nil {
t.Fatalf("seed foreground broker: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if resolved, err := broker.Resolve(ctx); err != nil || resolved == "" {
t.Fatalf("foreground broker unexpectedly waited for activation: %v", err)
}
}
func TestCrossPlatformCoveragePersonalRuntimeBusSpawnArgsContainNoSecretOrProfile(t *testing.T) {
const token = "runtime-spawn-canary"
args := personalBusSpawnArgsForToken(personal.Identity{
ClientID: "client", SourceID: "source", CorpID: "corp", UserID: "user",
}, "identity-hash", "normal", "https://ticket.invalid", "corp:user", token)
joined := strings.Join(args, " ")
for _, forbidden := range []string{token, "--profile", "corp:user"} {
if strings.Contains(joined, forbidden) {
t.Fatalf("spawn args leaked %q: %q", forbidden, joined)
}
}
for _, required := range []string{"--runtime-token-mode", "--identity-hash", "identity-hash", "--stream-source-id", "source"} {
if !strings.Contains(joined, required) {
t.Fatalf("spawn args %q missing %q", joined, required)
}
}
}
func TestCrossPlatformCoverageUnsupportedOldBusDoesNotDeleteReusedSubscription(t *testing.T) {
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
oldEdition := edition.Get()
oldEnsure := personalEnsureSubscription
oldUpsert := personalUpsertRunState
oldDelete := personalDeleteSubscription
oldRemove := personalRemoveRunStates
oldConsume := personalConsumeRun
oldValidate := personalValidateConsumeConfig
oldConflict := personalValidateNoOutputConflict
t.Cleanup(func() {
edition.Override(oldEdition)
personalEnsureSubscription = oldEnsure
personalUpsertRunState = oldUpsert
personalDeleteSubscription = oldDelete
personalRemoveRunStates = oldRemove
personalConsumeRun = oldConsume
personalValidateConsumeConfig = oldValidate
personalValidateNoOutputConflict = oldConflict
})
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
personalEnsureSubscription = func(context.Context, *personal.Client, personal.Identity, personalConsumeOptions) (*personal.Subscription, string, string, error) {
return &personal.Subscription{SubscribeID: "sub-existing"}, personal.EventMention, "at", nil
}
personalUpsertRunState = func(string, personal.RunState) error { return nil }
deleteCalls := 0
personalDeleteSubscription = func(*personal.Client, context.Context, string) error {
deleteCalls++
return nil
}
var removed []string
personalRemoveRunStates = func(_ string, ids []string) error {
removed = append(removed, ids...)
return nil
}
personalValidateConsumeConfig = func(consume.Config) error { return nil }
personalValidateNoOutputConflict = func(consume.Config, string) error { return nil }
personalConsumeRun = func(_ context.Context, cfg consume.Config) error {
if strings.TrimSpace(cfg.RuntimeToken) == "" {
t.Fatal("runtime token was not wired to consume")
}
return &consume.RuntimeTokenUnsupportedError{BusPID: 72}
}
err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
SubscribeID: "sub-existing",
ExplicitToken: "old-bus-cleanup-canary",
ClientIDOverride: "runtime-client",
})
if !errors.Is(err, consume.ErrRuntimeTokenUnsupported) {
t.Fatalf("consume error = %v", err)
}
if deleteCalls != 0 {
t.Fatalf("reused remote subscription was deleted %d time(s)", deleteCalls)
}
if len(removed) != 0 {
t.Fatalf("reused local run-state was removed: %#v", removed)
}
}
func TestCrossPlatformCoverageRuntimeTokenReusedDryRunUsesExplicitControlCredential(t *testing.T) {
const token = "runtime-dry-run-control-canary"
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
oldEdition := edition.Get()
oldEnsure := personalEnsureSubscription
oldUpsert := personalUpsertRunState
oldConsume := personalConsumeRun
oldBusRun := personalBusRun
t.Cleanup(func() {
edition.Override(oldEdition)
personalEnsureSubscription = oldEnsure
personalUpsertRunState = oldUpsert
personalConsumeRun = oldConsume
personalBusRun = oldBusRun
})
edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
return map[string]edition.RuntimeDefaultFn{
"$corpId": func(context.Context) (string, bool) { return "runtime-corp", true },
"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
}
}})
personalEnsureSubscription = func(ctx context.Context, client *personal.Client, _ personal.Identity, _ personalConsumeOptions) (*personal.Subscription, string, string, error) {
if _, ok := client.HTTPClient.Transport.(runtimeTokenControlTransport); !ok {
t.Fatalf("control transport = %T, want runtimeTokenControlTransport", client.HTTPClient.Transport)
}
got, err := client.AccessTokenProvider(ctx)
if err != nil || got != token {
t.Fatalf("control token = %q, %v", got, err)
}
return &personal.Subscription{SubscribeID: "sub-existing"}, personal.EventMention, "at", nil
}
personalUpsertRunState = func(string, personal.RunState) error {
t.Fatal("dry-run unexpectedly persisted run state")
return nil
}
consumeCalls := 0
personalConsumeRun = func(_ context.Context, cfg consume.Config) error {
consumeCalls++
if !cfg.DryRun {
t.Fatal("consume config is not dry-run")
}
if strings.Contains(strings.Join(cfg.SpawnExtraArgs, " "), token) {
t.Fatal("dry-run spawn args leaked runtime token")
}
return nil
}
personalBusRun = func(context.Context, bus.Config) error {
t.Fatal("dry-run unexpectedly started a bus")
return nil
}
err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
SubscribeID: "sub-existing",
ExplicitToken: token,
ClientIDOverride: "runtime-client",
Common: commonConsumeOptions{DryRun: true},
})
if err != nil {
t.Fatalf("dry-run consume error = %v", err)
}
if consumeCalls != 1 {
t.Fatalf("dry-run consume calls = %d, want 1", consumeCalls)
}
}
func TestCrossPlatformCoverageRuntimeTokenControlRejectionReleasesSubscriptionClaim(t *testing.T) {
store := &personalRecordingAttemptStore{}
reservation := &personalSubscriptionAttemptReservation{
store: store,
claim: &personal.AttemptClaim{AttemptID: "runtime-token-attempt"},
items: []personalSubscriptionAttemptItem{{eventKey: personal.EventMention, fingerprint: strings.Repeat("a", 64)}},
}
cause := &personal.APIError{
Code: "RUNTIME_TOKEN_REJECTED",
Message: "event runtime token was rejected; retry with a fresh host credential",
HTTPStatus: http.StatusUnauthorized,
}
if !personalRuntimeTokenControlRejection(cause) {
t.Fatal("runtime token control rejection was not classified")
}
err := reservation.releaseRuntimeTokenFailure()
if err == nil || !strings.Contains(err.Error(), "runtime token was rejected") {
t.Fatalf("releaseRuntimeTokenFailure() error = %v", err)
}
if store.releaseCalls != 1 || store.failureCalls != 0 {
t.Fatalf("attempt store release=%d failure=%d, want release only", store.releaseCalls, store.failureCalls)
}
}
type eventRuntimeRoundTripFunc func(*http.Request) (*http.Response, error)
func (f eventRuntimeRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
return f(req)
}
+1 -1
View File
@@ -472,7 +472,7 @@ func newRootCommandWithEngine(rootCtx context.Context, engine *pipeline.Engine,
newConfigCommand(),
newDoctorCommand(),
newRecoveryCommand(),
newEventCommand(),
newEventCommand(flags),
newAuditCommand(),
newCompletionCommand(root),
newUpgradeCommand(),
+4
View File
@@ -44,6 +44,10 @@ import (
// test binary never launches a page on the developer's machine; tests that
// need to assert the URL can still replace openBrowserFunc locally.
func TestMain(m *testing.M) {
if code, ok := runRuntimeTokenDetachedE2EChild(); ok {
os.Exit(code)
}
tmpDir, err := os.MkdirTemp("", "dws-app-test-keychain-")
if err != nil {
panic("create test keychain tempdir: " + err.Error())
+40
View File
@@ -0,0 +1,40 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
package auth
import "testing"
func TestCrossPlatformCoverageResolveProfileMetadataUsesSelectorGrammarAndReturnsCopy(t *testing.T) {
cfg := &ProfilesConfig{
Version: 3,
Profiles: []Profile{
{Name: "Historical", CorpID: "corp-1", CorpName: "Example Org"},
{Name: "Exact", CorpID: "corp-1", CorpName: "Example Org", UserID: "user-1", UserName: "Example User", ClientID: "client-1"},
},
OrgCurrentProfiles: map[string]string{"corp-1": "corp-1:user-1"},
}
for _, selector := range []string{"corp-1", "Example Org", "corp-1:Example User", "Example Org:Example User"} {
profile, err := ResolveProfileMetadata(cfg, selector)
if err != nil {
t.Fatalf("ResolveProfileMetadata(%q) error = %v", selector, err)
}
if profile == nil || profile.UserID != "user-1" || profile.ClientID != "client-1" {
t.Fatalf("ResolveProfileMetadata(%q) = %#v", selector, profile)
}
}
profile, err := ResolveProfileMetadata(cfg, "corp-1:user-1")
if err != nil {
t.Fatalf("ResolveProfileMetadata(exact) error = %v", err)
}
profile.Name = "mutated copy"
if cfg.Profiles[1].Name != "Exact" {
t.Fatalf("ResolveProfileMetadata returned registry-owned pointer")
}
if _, err := ResolveProfileMetadata(cfg, "missing"); err == nil {
t.Fatal("ResolveProfileMetadata(missing) unexpectedly succeeded")
}
}
+14
View File
@@ -738,6 +738,20 @@ func ResolveProfileWithScope(configDir, selector string) (*Profile, bool, error)
return result, exact, err
}
// ResolveProfileMetadata applies the public profile-selector grammar to an
// already-loaded, non-sensitive profiles registry. It performs no migration,
// keychain access, token loading, or persistence, making it suitable for
// callers that carry an externally managed bearer credential and need only
// corp/user/client identity metadata.
func ResolveProfileMetadata(cfg *ProfilesConfig, selector string) (*Profile, error) {
profile, _, err := resolveProfileSelection("", cfg, selector)
if err != nil || profile == nil {
return nil, err
}
copy := *profile
return &copy, nil
}
func resolveProfileWithScopeLocked(configDir, selector string) (*Profile, bool, error) {
if err := profilesEnsureMigration(configDir); err != nil {
return nil, false, err
+16
View File
@@ -17,6 +17,7 @@ import (
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
eventlock "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/lock"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
@@ -324,6 +325,21 @@ func TestCrossPlatformCoverageRunStartupAndSourceEdges(t *testing.T) {
if err := Run(context.Background(), base); !errors.Is(err, errBusInjected) {
t.Fatalf("source error = %v", err)
}
runtimeWorkDir := shortTempDir(t)
base.WorkDir = runtimeWorkDir
base.IPCEndpoint = dwsevent.IPCEndpoint(
runtimeWorkDir,
"open",
dwsevent.SourceKindPersonalStream,
dwsevent.IdentityHash(runtimeWorkDir),
)
base.Source = edgeSource{start: func(context.Context, dwsevent.EmitFn) error {
return runtimecred.ErrRuntimeTokenRejected
}}
if err := Run(context.Background(), base); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("runtime source error = %v", err)
}
}
type scriptedListener struct {
@@ -0,0 +1,275 @@
package bus
import (
"context"
"errors"
"io"
"log/slog"
"net"
"runtime"
"testing"
"time"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
func eventCoreDaemon(broker *runtimecred.Broker) *daemon {
return &daemon{
cfg: Config{CredentialBroker: broker, IdleTimeout: time.Second},
log: slog.New(slog.NewTextHandler(io.Discard, nil)),
hub: NewHub(4),
started: time.Now(),
idleStop: make(chan struct{}),
}
}
func eventCoreConnection(d *daemon, wrap func(net.Conn) net.Conn) (net.Conn, *transport.Writer, *transport.Reader, <-chan struct{}) {
server, client := net.Pipe()
if wrap != nil {
server = wrap(server)
}
done := make(chan struct{})
go func() {
d.handleConnection(context.Background(), server)
close(done)
}()
return client, transport.NewWriter(client), transport.NewReader(client), done
}
func eventCoreWaitDone(t *testing.T, done <-chan struct{}) {
t.Helper()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("connection handler did not stop")
}
}
type eventCoreWriteHookConn struct {
net.Conn
writes int
hook func(int)
}
func (c *eventCoreWriteHookConn) Write(p []byte) (int, error) {
n, err := c.Conn.Write(p)
c.writes++
if err == nil && c.hook != nil {
c.hook(c.writes)
}
return n, err
}
func TestCrossPlatformCoverageEventCoreDaemonHandshakeEdges(t *testing.T) {
t.Run("incompatible ack write failure", func(t *testing.T) {
d := eventCoreDaemon(nil)
client, w, _, done := eventCoreConnection(d, nil)
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: "unsupported"}); err != nil {
t.Fatal(err)
}
_ = client.Close()
eventCoreWaitDone(t, done)
})
t.Run("runtime ack write failure", func(t *testing.T) {
d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
client, w, _, done := eventCoreConnection(d, nil)
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
_ = client.Close()
eventCoreWaitDone(t, done)
})
t.Run("terminal runtime hello", func(t *testing.T) {
d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
client, w, r, done := eventCoreConnection(d, nil)
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil || ack.TerminalReason != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("terminal ack = %#v, %v", ack, err)
}
eventCoreWaitDone(t, done)
})
t.Run("malformed credential update", func(t *testing.T) {
d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
client, w, r, done := eventCoreConnection(d, nil)
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
if _, err := client.Write([]byte("{\n")); err != nil {
t.Fatal(err)
}
eventCoreWaitDone(t, done)
})
t.Run("unexpected credential update", func(t *testing.T) {
d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
client, w, r, done := eventCoreConnection(d, nil)
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
if err := w.WriteJSON(transport.Heartbeat{Type: transport.FrameTypeHeartbeat}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil || updateAck.ErrorCode != transport.CredentialErrorInvalid {
t.Fatalf("unexpected-frame ack = %#v, %v", updateAck, err)
}
eventCoreWaitDone(t, done)
})
t.Run("credential ack write failure", func(t *testing.T) {
d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
client, w, r, done := eventCoreConnection(d, nil)
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate, ExpectedGeneration: ack.CredentialGeneration, Token: "token",
}); err != nil {
t.Fatal(err)
}
_ = client.Close()
eventCoreWaitDone(t, done)
})
t.Run("activation conflict", func(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
d := eventCoreDaemon(broker)
client, w, r, done := eventCoreConnection(d, func(conn net.Conn) net.Conn {
return &eventCoreWriteHookConn{Conn: conn, hook: func(write int) {
if write == 2 {
_, _ = broker.Update(1, "newer-token")
}
}}
})
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
if err := w.WriteJSON(transport.CredentialUpdate{Type: transport.FrameTypeCredentialUpdate, Token: "first-token"}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil || !updateAck.Accepted {
t.Fatalf("credential ack = %#v, %v", updateAck, err)
}
var bye transport.Bye
if err := r.ReadJSON(&bye); err != nil || bye.Reason != "runtime_credential_activation_failed" {
t.Fatalf("activation failure bye = %#v, %v", bye, err)
}
eventCoreWaitDone(t, done)
})
}
func TestCrossPlatformCoverageEventCoreDaemonWriterStopEdges(t *testing.T) {
originalProcs := runtime.GOMAXPROCS(1)
t.Cleanup(func() { runtime.GOMAXPROCS(originalProcs) })
run := func(t *testing.T, queueEvent bool) {
t.Helper()
d := eventCoreDaemon(nil)
client, w, r, done := eventCoreConnection(d, nil)
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, SubscribeID: "writer-stop"}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(time.Second)
for d.hub.Len() != 1 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
time.Sleep(5 * time.Millisecond)
d.hub.mu.RLock()
var consumer *Consumer
for _, candidate := range d.hub.consumers {
consumer = candidate
}
d.hub.mu.RUnlock()
if consumer == nil {
t.Fatal("consumer not registered")
}
if queueEvent {
consumer.SendCh <- transport.Heartbeat{Type: transport.FrameTypeHeartbeat}
}
consumer.StopCh <- "writer-stop"
var bye transport.Bye
if err := r.ReadJSON(&bye); err != nil || bye.Reason != "writer-stop" {
t.Fatalf("writer stop bye = %#v, %v", bye, err)
}
eventCoreWaitDone(t, done)
}
t.Run("recheck after event", func(t *testing.T) { run(t, true) })
t.Run("blocked stop select", func(t *testing.T) { run(t, false) })
}
func TestCrossPlatformCoverageEventCoreDaemonHelpersAndStopAll(t *testing.T) {
var nilDaemon *daemon
nilDaemon.setTerminalReason("ignored")
if nilDaemon.getTerminalReason() != "" {
t.Fatal("nil daemon returned terminal reason")
}
d := eventCoreDaemon(nil)
d.setTerminalReason("ignored")
if d.getTerminalReason() != "" {
t.Fatal("invalid terminal reason was stored")
}
d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
if d.getTerminalReason() != transport.ByeReasonRuntimeTokenRejected {
t.Fatal("terminal reason was not stored")
}
if code, _ := classifyCredentialUpdateError(runtimecred.ErrEmptyToken); code != transport.CredentialErrorInvalid {
t.Fatalf("empty-token classification = %q", code)
}
if code, message := classifyCredentialUpdateError(errors.New("internal detail")); code != transport.CredentialErrorInternal || message != "runtime credential update failed" {
t.Fatalf("internal classification = %q, %q", code, message)
}
hub := NewHub(1)
consumer, err := hub.Register(transport.Hello{})
if err != nil {
t.Fatal(err)
}
if stopped := hub.StopAll(" "); stopped != 1 {
t.Fatalf("StopAll = %d", stopped)
}
select {
case reason := <-consumer.StopCh:
if reason != "shutdown" {
t.Fatalf("default stop reason = %q", reason)
}
case <-time.After(time.Second):
t.Fatal("default stop reason not delivered")
}
hub.Unregister(consumer.ID)
}
+251 -35
View File
@@ -30,6 +30,7 @@ import (
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/dedup"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
)
@@ -78,6 +79,11 @@ type Config struct {
// Source is the cloud adapter. Required.
Source SourceAdapter
// CredentialBroker enables additive runtime-token handoff over the
// owner-only local IPC transport. Nil preserves the original protocol and
// does not advertise runtime-token support.
CredentialBroker *runtimecred.Broker
// IdleTimeout: bus self-exits after this long with zero consumers.
// Zero disables (bus runs until SIGTERM).
IdleTimeout time.Duration
@@ -106,11 +112,13 @@ type Config struct {
}
var (
daemonMkdirAll = os.MkdirAll
daemonAcquire = Acquire
daemonWriteMeta = WriteMeta
daemonListen = transport.Listen
daemonShutdownTimeout = 2 * time.Second
daemonMkdirAll = os.MkdirAll
daemonAcquire = Acquire
daemonWriteMeta = WriteMeta
daemonListen = transport.Listen
daemonShutdownTimeout = 2 * time.Second
daemonByeDrainTimeout = 100 * time.Millisecond
daemonCredentialHandshakeTimeout = 10 * time.Second
)
// Run starts the bus daemon. Lifecycle (plan §4 invariant #6):
@@ -237,12 +245,22 @@ func Run(ctx context.Context, cfg Config) error {
// 6. Wait for shutdown trigger.
var exitErr error
shutdownReason := "shutdown"
select {
case <-ctx.Done():
log.Info("bus: shutdown requested by ctx", "reason", ctx.Err())
case err := <-srcErr:
log.Error("bus: source exited", "err", err)
exitErr = err
shutdownReason = sourceShutdownReason(err)
if shutdownReason == transport.ByeReasonRuntimeTokenRejected {
// A runtime token can fail immediately after Broker.Update. Serialize
// terminal publication with that handshake so the initiating consumer
// is registered (or receives a terminal HelloAck) before shutdown.
d.credentialHandoffMu.Lock()
d.setTerminalReason(shutdownReason)
d.credentialHandoffMu.Unlock()
}
case <-d.idleStop:
log.Info("bus: idle timeout reached, shutting down")
}
@@ -252,7 +270,7 @@ func Run(ctx context.Context, cfg Config) error {
// consumers. The accept-loop barrier is required before WaitGroup.Wait:
// sync.WaitGroup forbids a positive Add racing with Wait.
cancelRun()
d.shutdown(acceptDone)
d.shutdown(acceptDone, shutdownReason)
<-idleDone
<-dropWarnDone
@@ -274,6 +292,10 @@ type daemon struct {
shutdownMu sync.Mutex
shuttingDown atomic.Bool
idleStop chan struct{}
credentialHandoffMu sync.Mutex
terminalMu sync.RWMutex
terminalReason string
}
// closeOnceConn makes every connection close path idempotent. A live consumer
@@ -330,7 +352,8 @@ func (d *daemon) acceptLoop(ctx context.Context) {
}
// handleConnection processes one IPC connection's full lifecycle: read
// Hello → register with Hub → spawn writer goroutine → read until EOF/Bye.
// Hello → optional runtime credential negotiation → register with Hub → spawn
// writer goroutine → read until EOF/Bye.
// Always Unregisters and Closes on exit (plan invariant #5).
func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
conn = ensureCloseOnce(conn)
@@ -372,29 +395,141 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
return
}
// Regular consumer registration
// HelloAck — credentials_source fields are filled in by the daemon
// runner (which knows from the strict resolver) and exposed via the
// adapter for forward-compat. v1 leaves them empty here; daemon.Run
// passes them through future config if the caller wishes.
ack := d.helloAck()
handoffLocked := false
runtimeGeneration := uint64(0)
defer func() {
if handoffLocked {
d.credentialHandoffMu.Unlock()
}
}()
// Runtime credentials use a two-phase additive handshake. The first ack
// proves capability before the client sends any secret. Only a successful
// CAS and credential ack permit Hub registration.
if hello.CredentialMode != "" {
if hello.CredentialMode != transport.CredentialModeRuntimeToken || d.cfg.CredentialBroker == nil {
ack.Capabilities = nil
ack.CredentialGeneration = 0
if err := w.WriteJSON(ack); err != nil {
d.log.Warn("bus: incompatible helloack write failed", "err", err)
}
return
}
d.credentialHandoffMu.Lock()
handoffLocked = true
// Terminal state may have been published while this Hello waited for a
// concurrent credential handoff. Rebuild the ack while holding the gate.
ack = d.helloAck()
if err := w.WriteJSON(ack); err != nil {
d.log.Warn("bus: runtime helloack write failed", "err", err)
return
}
if ack.TerminalReason == transport.ByeReasonRuntimeTokenRejected {
return
}
var update transport.CredentialUpdate
_ = conn.SetReadDeadline(time.Now().Add(daemonCredentialHandshakeTimeout))
if err := r.ReadJSON(&update); err != nil {
// Do not include the decoder error: malformed JSON may contain
// fragments of the credential.
d.log.Warn("bus: malformed runtime credential update")
return
}
_ = conn.SetReadDeadline(time.Time{})
if update.Type != transport.FrameTypeCredentialUpdate {
_ = w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: false,
CredentialGeneration: d.cfg.CredentialBroker.Generation(),
ErrorCode: transport.CredentialErrorInvalid,
Error: "unexpected credential update frame",
})
return
}
// Validate registration before applying the credential or sending an
// accepted ack. Hub.Register performs the same deterministic compile
// before mutating the Hub; this preflight keeps invalid filters from
// producing a ready marker after credential negotiation.
if _, err := compileMatcher(hello.EventTypes, hello.Filter, hello.SubscribeID); err != nil {
update.Token = ""
_ = w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: false,
CredentialGeneration: d.cfg.CredentialBroker.Generation(),
ErrorCode: transport.CredentialErrorRegistration,
Error: "consumer registration validation failed",
})
d.log.Warn("bus: runtime consumer registration validation failed")
return
}
generation, updateErr := d.cfg.CredentialBroker.Update(update.ExpectedGeneration, update.Token)
runtimeGeneration = generation
update.Token = ""
credentialAck := transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: updateErr == nil,
CredentialGeneration: generation,
}
if updateErr != nil {
credentialAck.ErrorCode, credentialAck.Error = classifyCredentialUpdateError(updateErr)
}
if err := w.WriteJSON(credentialAck); err != nil {
d.log.Warn("bus: credential update ack write failed", "err", err)
return
}
if updateErr != nil {
d.log.Warn("bus: runtime credential update rejected", "error_code", credentialAck.ErrorCode)
return
}
}
// Regular consumer registration. Local clients retain the original
// register-before-HelloAck ordering; runtime clients were already acked by
// the additive handshake above.
c, err := d.hub.Register(hello)
if err != nil {
d.log.Warn("bus: register failed", "err", err, "pid", hello.ConsumerPID)
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "register_failed: " + err.Error()})
return
}
// HelloAck — credentials_source fields are filled in by the daemon
// runner (which knows from the strict resolver) and exposed via the
// adapter for forward-compat. v1 leaves them empty here; daemon.Run
// passes them through future config if the caller wishes.
idleSecs := int(d.cfg.IdleTimeout / time.Second)
if err := w.WriteJSON(transport.HelloAck{
Type: transport.FrameTypeHelloAck,
BusPID: os.Getpid(),
SourceState: "connected", // best-effort; full state machine pushed via SourceState frames
StateSource: "inferred",
IdleTimeoutSecs: idleSecs,
}); err != nil {
d.log.Warn("bus: helloack write failed", "err", err)
if handoffLocked {
// The runtime broker deliberately keeps the seed pending until the
// initiating consumer is registered. This prevents ticket acquisition
// (and an immediate 401) from racing ahead of the only connection that
// can observe the typed terminal reason.
if _, activateErr := d.cfg.CredentialBroker.Activate(runtimeGeneration); activateErr != nil {
d.log.Error("bus: runtime credential activation failed")
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "runtime_credential_activation_failed"})
d.hub.Unregister(c.ID)
return
}
}
// A local/legacy consumer can arrive after terminal publication but after
// StopAll took its snapshot. Refuse it synchronously so it cannot observe a
// clean EOF for a runtime-token rejection.
if terminalReason := d.getTerminalReason(); terminalReason != "" {
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: terminalReason})
d.hub.Unregister(c.ID)
return
}
if handoffLocked {
d.credentialHandoffMu.Unlock()
handoffLocked = false
}
if hello.CredentialMode == "" {
if err := w.WriteJSON(ack); err != nil {
d.log.Warn("bus: helloack write failed", "err", err)
d.hub.Unregister(c.ID)
return
}
}
// Writer goroutine pulls from SendCh and writes to the wire.
writerDone := make(chan struct{})
@@ -416,6 +551,16 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
if !ok {
return
}
// A stop may have arrived while both channels were ready and the
// scheduler selected the buffered event. Re-check before starting a
// potentially blocking event write so terminal reasons stay prompt.
select {
case reason := <-c.StopCh:
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: reason})
_ = conn.Close()
return
default:
}
if err := w.WriteJSON(frame); err != nil {
return
}
@@ -454,6 +599,51 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
_ = ctx // for future use (writer ctx-cancel propagation)
}
func (d *daemon) helloAck() transport.HelloAck {
ack := transport.HelloAck{
Type: transport.FrameTypeHelloAck,
BusPID: os.Getpid(),
SourceState: "connected", // best-effort; full state machine pushed via SourceState frames
StateSource: "inferred",
IdleTimeoutSecs: int(d.cfg.IdleTimeout / time.Second),
}
if d.cfg.CredentialBroker != nil {
ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
ack.CredentialGeneration = d.cfg.CredentialBroker.Generation()
}
ack.TerminalReason = d.getTerminalReason()
return ack
}
func (d *daemon) setTerminalReason(reason string) {
if d == nil || reason != transport.ByeReasonRuntimeTokenRejected {
return
}
d.terminalMu.Lock()
d.terminalReason = reason
d.terminalMu.Unlock()
}
func (d *daemon) getTerminalReason() string {
if d == nil {
return ""
}
d.terminalMu.RLock()
defer d.terminalMu.RUnlock()
return d.terminalReason
}
func classifyCredentialUpdateError(err error) (string, string) {
var conflict *runtimecred.GenerationConflictError
if errors.As(err, &conflict) {
return transport.CredentialErrorGenerationConflict, conflict.Error()
}
if errors.Is(err, runtimecred.ErrEmptyToken) || errors.Is(err, runtimecred.ErrTokenTooLarge) {
return transport.CredentialErrorInvalid, err.Error()
}
return transport.CredentialErrorInternal, "runtime credential update failed"
}
func (d *daemon) handleConsumerStopRPC(w *transport.Writer, r *transport.Reader) {
var req transport.ConsumerStopReq
if err := r.ReadJSON(&req); err != nil {
@@ -577,38 +767,64 @@ func (d *daemon) triggerShutdown(reason string) {
// 4. wait for acceptLoop to return so no future consumerWG.Add can occur
// 5. close all accepted connections and wait for handlers to drain
// 6. lock + meta cleanup via Run's defers
func (d *daemon) shutdown(acceptDone <-chan struct{}) {
func (d *daemon) shutdown(acceptDone <-chan struct{}, reasons ...string) {
d.shutdownMu.Lock()
defer d.shutdownMu.Unlock()
if !d.shuttingDown.CompareAndSwap(false, true) {
return
}
d.hub.Broadcast(transport.Bye{Type: transport.FrameTypeBye, Reason: "shutdown"})
reason := normalizedShutdownReason(reasons...)
if reason == transport.ByeReasonRuntimeTokenRejected {
d.hub.StopAll(reason)
} else {
d.hub.Broadcast(transport.Bye{Type: transport.FrameTypeBye, Reason: reason})
}
_ = d.listener.Close()
<-acceptDone
// Force-close all open IPC connections so any reader goroutine blocked
// on Read() returns with a network error and exits cleanly. Without
// this the consumerWG never drains and Run hangs forever.
d.conns.Range(func(k, _ any) bool {
if c, ok := k.(net.Conn); ok {
_ = c.Close()
}
return true
})
// Give consumers a brief moment to drain final frames before we tear
// down their channels.
// Let local consumers drain the final Bye before force-closing their
// connections. This short grace period is what makes typed shutdown
// reasons (notably runtime_token_rejected) observable instead of racing
// with EOF. Consumers close their side immediately after reading Bye.
doneCh := make(chan struct{})
go func() {
d.consumerWG.Wait()
close(doneCh)
}()
select {
case <-doneCh:
return
case <-time.After(daemonByeDrainTimeout):
}
// A wedged/old consumer may not close after Bye. Force-close remaining
// connections so the daemon still has a bounded shutdown.
d.conns.Range(func(k, _ any) bool {
if c, ok := k.(net.Conn); ok {
_ = c.Close()
}
return true
})
select {
case <-doneCh:
case <-time.After(daemonShutdownTimeout):
d.log.Warn("bus: shutdown: consumer goroutines did not drain within 2s")
}
}
func sourceShutdownReason(err error) string {
if errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
return transport.ByeReasonRuntimeTokenRejected
}
return "shutdown"
}
func normalizedShutdownReason(reasons ...string) string {
if len(reasons) > 0 && reasons[0] == transport.ByeReasonRuntimeTokenRejected {
return transport.ByeReasonRuntimeTokenRejected
}
return "shutdown"
}
// signalReady writes a single 'R' byte to the ready pipe (if provided) and
// closes it. The parent process (busctl/spawn) reads one byte and proceeds.
func signalReady(p *os.File) {
+25
View File
@@ -246,6 +246,31 @@ func (h *Hub) StopConsumers(subscribeIDs []string, reason string) []string {
return out
}
// StopAll requests a high-priority graceful close for every live consumer.
// Unlike Broadcast(Bye), this uses the writer's priority StopCh and therefore
// cannot sit behind a full event buffer during a terminal source failure.
func (h *Hub) StopAll(reason string) int {
reason = strings.TrimSpace(reason)
if reason == "" {
reason = "shutdown"
}
h.mu.RLock()
consumers := make([]*Consumer, 0, len(h.consumers))
for _, consumer := range h.consumers {
consumers = append(consumers, consumer)
}
h.mu.RUnlock()
stopped := 0
for _, consumer := range consumers {
select {
case consumer.StopCh <- reason:
stopped++
default:
}
}
return stopped
}
// Unregister removes a consumer by ID and closes its sendCh. Idempotent —
// calling twice or on an unknown ID is a no-op. closeSend shares the same
// per-consumer lock as Deliver/Broadcast, so a stale Hub snapshot cannot send
+21
View File
@@ -346,6 +346,27 @@ func TestHub_StopConsumersCoalescesQueuedStop(t *testing.T) {
}
}
func TestCrossPlatformCoverageHubStopAllBypassesFullEventBuffer(t *testing.T) {
hub := NewHub(1)
consumer, err := hub.Register(transport.Hello{Type: transport.FrameTypeHello})
if err != nil {
t.Fatal(err)
}
consumer.SendCh <- transport.Event{Type: transport.FrameTypeEvent}
if stopped := hub.StopAll(transport.ByeReasonRuntimeTokenRejected); stopped != 1 {
t.Fatalf("StopAll() = %d, want 1", stopped)
}
select {
case reason := <-consumer.StopCh:
if reason != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("StopCh reason = %q", reason)
}
default:
t.Fatal("terminal stop was blocked behind the full event buffer")
}
}
func TestHub_ConcurrentStopConsumersRegisterUnregister(t *testing.T) {
h := NewHub(4)
const workers = 32
@@ -0,0 +1,407 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
package bus
import (
"bytes"
"context"
"errors"
"io"
"log/slog"
"net"
"path/filepath"
"strings"
"testing"
"time"
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
type runtimeCredentialRejectSource struct {
broker *runtimecred.Broker
}
func (s *runtimeCredentialRejectSource) Start(ctx context.Context, _ dwsevent.EmitFn) error {
if _, err := s.broker.Resolve(ctx); err != nil {
return err
}
return runtimecred.ErrRuntimeTokenRejected
}
func runtimeCredentialDaemon(broker *runtimecred.Broker, logOutput io.Writer) *daemon {
if logOutput == nil {
logOutput = io.Discard
}
return &daemon{
cfg: Config{CredentialBroker: broker},
log: slog.New(slog.NewTextHandler(logOutput, nil)),
hub: NewHub(2),
started: time.Now(),
idleStop: make(chan struct{}),
}
}
func runRuntimeCredentialConnection(t *testing.T, d *daemon) (net.Conn, *transport.Writer, *transport.Reader, <-chan struct{}) {
t.Helper()
server, client := net.Pipe()
done := make(chan struct{})
go func() {
d.handleConnection(context.Background(), server)
close(done)
}()
return client, transport.NewWriter(client), transport.NewReader(client), done
}
func TestCrossPlatformCoverageDaemonRuntimeCredentialHandshakeBeforeRegister(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
d := runtimeCredentialDaemon(broker, nil)
client, w, r, done := runRuntimeCredentialConnection(t, d)
defer client.Close()
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
ConsumerPID: 42,
CredentialMode: transport.CredentialModeRuntimeToken,
}); err != nil {
t.Fatal(err)
}
var helloAck transport.HelloAck
if err := r.ReadJSON(&helloAck); err != nil {
t.Fatal(err)
}
if !hasTransportCapability(helloAck.Capabilities, transport.CapabilityRuntimeTokenV1) || helloAck.CredentialGeneration != 0 {
t.Fatalf("hello ack = %#v", helloAck)
}
if d.hub.Len() != 0 {
t.Fatalf("consumer registered before credential update: %d", d.hub.Len())
}
const canary = "ipc-canary-runtime-token"
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: helloAck.CredentialGeneration,
Token: canary,
}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil {
t.Fatal(err)
}
if !updateAck.Accepted || updateAck.CredentialGeneration != 1 {
t.Fatalf("credential ack = %#v", updateAck)
}
deadline := time.Now().Add(time.Second)
for d.hub.Len() != 1 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
if d.hub.Len() != 1 {
t.Fatal("consumer was not registered after credential ack")
}
if token, err := broker.Resolve(context.Background()); err != nil || token != canary {
t.Fatalf("broker did not resolve installed runtime token: %v", err)
}
if err := w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"}); err != nil {
t.Fatal(err)
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("connection did not close")
}
}
func TestCrossPlatformCoverageDaemonCompatibleBusRotatesRuntimeCredentialAcrossConnections(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
d := runtimeCredentialDaemon(broker, nil)
handshake := func(token string, wantHelloGeneration, wantAckGeneration uint64) {
t.Helper()
client, w, r, done := runRuntimeCredentialConnection(t, d)
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
CredentialMode: transport.CredentialModeRuntimeToken,
}); err != nil {
t.Fatal(err)
}
var helloAck transport.HelloAck
if err := r.ReadJSON(&helloAck); err != nil {
t.Fatal(err)
}
if helloAck.CredentialGeneration != wantHelloGeneration {
t.Fatalf("hello generation = %d, want %d", helloAck.CredentialGeneration, wantHelloGeneration)
}
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: helloAck.CredentialGeneration,
Token: token,
}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil {
t.Fatal(err)
}
if !updateAck.Accepted || updateAck.CredentialGeneration != wantAckGeneration {
t.Fatalf("credential ack accepted=%v generation=%d, want true/%d", updateAck.Accepted, updateAck.CredentialGeneration, wantAckGeneration)
}
if err := w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"}); err != nil {
t.Fatal(err)
}
_ = client.Close()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("connection did not close")
}
}
handshake("runtime-token-a", 0, 1)
if token, err := broker.Resolve(context.Background()); err != nil || token != "runtime-token-a" {
t.Fatalf("broker did not retain first token: %v", err)
}
handshake("runtime-token-b", 1, 2)
if token, err := broker.Resolve(context.Background()); err != nil || token != "runtime-token-b" {
t.Fatalf("broker did not rotate to second token: %v", err)
}
}
func TestCrossPlatformCoverageDaemonRuntimeCredentialMissingCapabilityDoesNotRegister(t *testing.T) {
d := runtimeCredentialDaemon(nil, nil)
client, w, r, done := runRuntimeCredentialConnection(t, d)
defer client.Close()
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
CredentialMode: transport.CredentialModeRuntimeToken,
}); err != nil {
t.Fatal(err)
}
var ack transport.HelloAck
if err := r.ReadJSON(&ack); err != nil {
t.Fatal(err)
}
if hasTransportCapability(ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
t.Fatalf("unsupported daemon advertised capability: %#v", ack)
}
if d.hub.Len() != 0 {
t.Fatalf("unsupported daemon registered consumer: %d", d.hub.Len())
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("unsupported connection did not close")
}
}
func TestCrossPlatformCoverageDaemonTerminalStateRejectsLateLegacyConsumer(t *testing.T) {
d := runtimeCredentialDaemon(nil, nil)
d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
client, w, r, done := runRuntimeCredentialConnection(t, d)
defer client.Close()
if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, ConsumerPID: 7}); err != nil {
t.Fatal(err)
}
var bye transport.Bye
if err := r.ReadJSON(&bye); err != nil {
t.Fatal(err)
}
if bye.Type != transport.FrameTypeBye || bye.Reason != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("late consumer frame = %#v", bye)
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("late consumer connection did not close")
}
if d.hub.Len() != 0 {
t.Fatalf("late terminal consumer remained registered: %d", d.hub.Len())
}
}
func TestCrossPlatformCoverageDaemonRuntimeCredentialConflictDoesNotLeakOrRegister(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{})
if _, err := broker.Update(0, "installed-secret"); err != nil {
t.Fatal(err)
}
var logs bytes.Buffer
d := runtimeCredentialDaemon(broker, &logs)
client, w, r, done := runRuntimeCredentialConnection(t, d)
defer client.Close()
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
CredentialMode: transport.CredentialModeRuntimeToken,
}); err != nil {
t.Fatal(err)
}
var helloAck transport.HelloAck
if err := r.ReadJSON(&helloAck); err != nil {
t.Fatal(err)
}
const canary = "rejected-canary-secret"
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: 0,
Token: canary,
}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil {
t.Fatal(err)
}
if updateAck.Accepted || updateAck.ErrorCode != transport.CredentialErrorGenerationConflict || updateAck.CredentialGeneration != 1 {
t.Fatalf("conflict ack = %#v", updateAck)
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("rejected connection did not close")
}
if strings.Contains(updateAck.Error, canary) || strings.Contains(logs.String(), canary) {
t.Fatal("credential appeared in acknowledgement or logs")
}
if d.hub.Len() != 0 {
t.Fatalf("rejected consumer registered: %d", d.hub.Len())
}
}
func TestCrossPlatformCoverageDaemonRuntimeCredentialInvalidFilterRejectedBeforeUpdate(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{})
d := runtimeCredentialDaemon(broker, nil)
client, w, r, done := runRuntimeCredentialConnection(t, d)
defer client.Close()
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
CredentialMode: transport.CredentialModeRuntimeToken,
Filter: "[",
}); err != nil {
t.Fatal(err)
}
var helloAck transport.HelloAck
if err := r.ReadJSON(&helloAck); err != nil {
t.Fatal(err)
}
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: helloAck.CredentialGeneration,
Token: "filter-canary-secret",
}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil {
t.Fatal(err)
}
if updateAck.Accepted || updateAck.ErrorCode != transport.CredentialErrorRegistration {
t.Fatalf("invalid filter ack = %#v", updateAck)
}
if broker.Generation() != 0 || d.hub.Len() != 0 {
t.Fatalf("invalid filter mutated state: generation=%d consumers=%d", broker.Generation(), d.hub.Len())
}
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("invalid filter connection did not close")
}
}
func TestCrossPlatformCoverageRuntimeCredentialShutdownReason(t *testing.T) {
if got := sourceShutdownReason(runtimecred.ErrRuntimeTokenRejected); got != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("runtime source shutdown reason = %q", got)
}
if got := sourceShutdownReason(errors.New("local source failed")); got != "shutdown" {
t.Fatalf("local source shutdown reason = %q", got)
}
if got := normalizedShutdownReason(transport.ByeReasonRuntimeTokenRejected); got != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("normalized runtime shutdown reason = %q", got)
}
for _, reasons := range [][]string{nil, {"peer-controlled"}} {
if got := normalizedShutdownReason(reasons...); got != "shutdown" {
t.Fatalf("normalized untrusted shutdown reason = %q", got)
}
}
}
func TestCrossPlatformCoverageDaemonRuntimeCredentialSourceRejectionBroadcastsTypedBye(t *testing.T) {
skipOnWindows(t, "uses Unix socket dial")
workDir := shortTempDir(t)
sockPath := filepath.Join(workDir, "bus.sock")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
runDone := make(chan error, 1)
go func() {
runDone <- Run(ctx, Config{
WorkDir: workDir,
IPCEndpoint: sockPath,
ClientID: "runtime-client",
Edition: "open",
SourceKind: dwsevent.SourceKindPersonalStream,
IdentityHash: "0123456789abcdef",
SourceID: "open",
Source: &runtimeCredentialRejectSource{broker: broker},
CredentialBroker: broker,
})
}()
waitForFile(t, sockPath, 2*time.Second)
conn, err := transport.Dial(sockPath)
if err != nil {
t.Fatal(err)
}
defer conn.Close()
r, w := transport.NewReader(conn), transport.NewWriter(conn)
if err := w.WriteJSON(transport.Hello{
Type: transport.FrameTypeHello,
CredentialMode: transport.CredentialModeRuntimeToken,
}); err != nil {
t.Fatal(err)
}
var helloAck transport.HelloAck
if err := r.ReadJSON(&helloAck); err != nil {
t.Fatal(err)
}
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: helloAck.CredentialGeneration,
Token: "runtime-rejected-canary",
}); err != nil {
t.Fatal(err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil || !updateAck.Accepted {
t.Fatalf("credential update ack = %#v, %v", updateAck, err)
}
var bye transport.Bye
if err := r.ReadJSON(&bye); err != nil {
t.Fatalf("read typed shutdown: %v", err)
}
if bye.Type != transport.FrameTypeBye || bye.Reason != transport.ByeReasonRuntimeTokenRejected {
t.Fatalf("shutdown frame = %#v", bye)
}
_ = conn.Close()
select {
case err := <-runDone:
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("Run() error = %v", err)
}
case <-time.After(3 * time.Second):
t.Fatal("runtime credential bus did not stop")
}
}
func hasTransportCapability(capabilities []string, want string) bool {
for _, capability := range capabilities {
if capability == want {
return true
}
}
return false
}
@@ -0,0 +1,101 @@
package consume
import (
"bytes"
"context"
"errors"
"io"
"strings"
"testing"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
func eventCoreCredentialReader(t *testing.T, frame any) *transport.Reader {
t.Helper()
var buffer bytes.Buffer
if err := transport.NewWriter(&buffer).WriteJSON(frame); err != nil {
t.Fatal(err)
}
return transport.NewReader(&buffer)
}
func TestCrossPlatformCoverageEventCoreRuntimeNegotiationEdges(t *testing.T) {
capable := transport.HelloAck{Capabilities: []string{transport.CapabilityRuntimeTokenV1}}
if err := negotiateRuntimeToken(
transport.NewWriter(io.Discard),
transport.NewReader(strings.NewReader("")),
transport.HelloAck{TerminalReason: transport.ByeReasonRuntimeTokenRejected},
"token",
); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("terminal hello error = %v", err)
}
wantWriteErr := errors.New("write failed")
if err := negotiateRuntimeToken(
transport.NewWriter(errorWriter{err: wantWriteErr}),
transport.NewReader(strings.NewReader("")),
capable,
"token",
); !errors.Is(err, wantWriteErr) {
t.Fatalf("credential write error = %v", err)
}
if err := negotiateRuntimeToken(
transport.NewWriter(io.Discard),
transport.NewReader(strings.NewReader("")),
capable,
"token",
); !errors.Is(err, io.EOF) {
t.Fatalf("credential ack read error = %v", err)
}
if err := negotiateRuntimeToken(
transport.NewWriter(io.Discard),
eventCoreCredentialReader(t, transport.Heartbeat{Type: transport.FrameTypeHeartbeat}),
capable,
"token",
); err == nil || !strings.Contains(err.Error(), "unexpected runtime credential response") {
t.Fatalf("unexpected credential frame error = %v", err)
}
if err := negotiateRuntimeToken(
transport.NewWriter(io.Discard),
eventCoreCredentialReader(t, transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: false,
ErrorCode: transport.CredentialErrorRuntimeRejected,
}),
capable,
"token",
); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("runtime rejected ack error = %v", err)
}
for _, code := range []string{
transport.CredentialErrorGenerationConflict,
transport.CredentialErrorInvalid,
transport.CredentialErrorRegistration,
transport.CredentialErrorRuntimeRejected,
transport.CredentialErrorInternal,
} {
if got := safeCredentialErrorCode(code); got != code {
t.Fatalf("safeCredentialErrorCode(%q) = %q", code, got)
}
}
if got := safeCredentialErrorCode("peer-controlled"); got != transport.CredentialErrorInternal {
t.Fatalf("unknown credential error code = %q", got)
}
}
func TestCrossPlatformCoverageEventCoreRunManyHandshakeFailure(t *testing.T) {
bus := newManyFakeBus(901, nil)
installManyDiscover(t, bus)
cfg := manyTestConfig(io.Discard, io.Discard)
cfg.RuntimeToken = "runtime-token"
err := RunMany(context.Background(), cfg, manyTestSpecs())
if !errors.Is(err, ErrRuntimeTokenUnsupported) || !strings.Contains(err.Error(), "runtime credential handshake") {
t.Fatalf("RunMany handshake error = %v", err)
}
}
+103
View File
@@ -21,9 +21,11 @@ import (
"io"
"net"
"os"
"strings"
"time"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
@@ -45,6 +47,11 @@ type Config struct {
// personal_stream.
SpawnExtraArgs []string
// RuntimeToken is a host-supplied credential handed to a compatible bus
// only after capability negotiation over owner-only local IPC. It is never
// included in dry-run output, child argv, environment, or persisted state.
RuntimeToken string `json:"-" yaml:"-"`
// EventTypes / Filter / Compact are forwarded to the bus via Hello
// for server-side pushdown filtering.
EventTypes []string
@@ -132,6 +139,36 @@ type Config struct {
var discoverBus = busctl.Discover
var (
ErrRuntimeTokenUnsupported = errors.New("consume: event bus does not support secure runtime-token handoff")
ErrRuntimeTokenUpdate = errors.New("consume: event bus rejected runtime-token update")
)
// RuntimeTokenUnsupportedError is returned before the token is sent when the
// connected bus lacks the runtime_token_v1 capability.
type RuntimeTokenUnsupportedError struct {
BusPID int
}
func (e *RuntimeTokenUnsupportedError) Error() string {
return fmt.Sprintf("consume: running event bus (pid %d) does not support secure runtime-token handoff; let existing consumers exit, inspect with `dws event status --as user --format json`, preview cleanup with `dws event stop --as user --all --dry-run`, then confirm with `dws event stop --as user --all --yes` and retry", e.BusPID)
}
func (e *RuntimeTokenUnsupportedError) Unwrap() error { return ErrRuntimeTokenUnsupported }
// RuntimeTokenUpdateError reports a rejected credential CAS without carrying
// either the credential or peer-provided free-form error text.
type RuntimeTokenUpdateError struct {
Code string
Generation uint64
}
func (e *RuntimeTokenUpdateError) Error() string {
return fmt.Sprintf("consume: event bus rejected runtime-token update (code=%s, generation=%d)", e.Code, e.Generation)
}
func (e *RuntimeTokenUpdateError) Unwrap() error { return ErrRuntimeTokenUpdate }
// Run dials the bus (forking one if necessary), sends Hello, and writes
// each received Event frame as one NDJSON line to stdout. Blocks until
// ctx is cancelled, MaxEvents is reached, the bus sends Bye, or the
@@ -144,6 +181,7 @@ func Run(ctx context.Context, cfg Config) error {
if cfg.WorkDir == "" || cfg.IPCEndpoint == "" || cfg.ClientID == "" {
return errors.New("consume: WorkDir, IPCEndpoint, and ClientID are required")
}
cfg.RuntimeToken = strings.TrimSpace(cfg.RuntimeToken)
if cfg.Stdout == nil {
cfg.Stdout = os.Stdout
}
@@ -225,6 +263,9 @@ func Run(ctx context.Context, cfg Config) error {
SubscribeID: cfg.SubscribeID,
Compact: cfg.Compact,
}
if cfg.RuntimeToken != "" {
hello.CredentialMode = transport.CredentialModeRuntimeToken
}
if err := w.WriteJSON(hello); err != nil {
return fmt.Errorf("consume: write hello: %w", err)
}
@@ -236,6 +277,9 @@ func Run(ctx context.Context, cfg Config) error {
if ack.Type != transport.FrameTypeHelloAck {
return fmt.Errorf("consume: unexpected first frame type %q", ack.Type)
}
if err := negotiateRuntimeToken(w, r, ack, cfg.RuntimeToken); err != nil {
return err
}
if !cfg.Quiet {
// Contract: a fixed ready line on stderr BEFORE any stdout event.
// Parents block on stderr until this appears, then read stdout.
@@ -331,6 +375,9 @@ func Run(ctx context.Context, cfg Config) error {
case transport.FrameTypeBye:
var bye transport.Bye
_ = json.Unmarshal(raw, &bye)
if bye.Reason == transport.ByeReasonRuntimeTokenRejected {
return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
}
if !cfg.Quiet {
fmt.Fprintf(cfg.Stderr, "[event] bus closing: %s\n", bye.Reason)
}
@@ -350,6 +397,62 @@ func Run(ctx context.Context, cfg Config) error {
}
}
func negotiateRuntimeToken(w *transport.Writer, r *transport.Reader, ack transport.HelloAck, token string) error {
if token == "" {
return nil
}
if ack.TerminalReason == transport.ByeReasonRuntimeTokenRejected {
return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
}
if !hasCapability(ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
return &RuntimeTokenUnsupportedError{BusPID: ack.BusPID}
}
if err := w.WriteJSON(transport.CredentialUpdate{
Type: transport.FrameTypeCredentialUpdate,
ExpectedGeneration: ack.CredentialGeneration,
Token: token,
}); err != nil {
return fmt.Errorf("consume: write runtime credential update: %w", err)
}
var updateAck transport.CredentialUpdateAck
if err := r.ReadJSON(&updateAck); err != nil {
return fmt.Errorf("consume: read runtime credential update ack: %w", err)
}
if updateAck.Type != transport.FrameTypeCredentialUpdateAck {
return errors.New("consume: unexpected runtime credential response frame")
}
if !updateAck.Accepted {
code := safeCredentialErrorCode(updateAck.ErrorCode)
if code == transport.CredentialErrorRuntimeRejected {
return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
}
return &RuntimeTokenUpdateError{Code: code, Generation: updateAck.CredentialGeneration}
}
return nil
}
func safeCredentialErrorCode(code string) string {
switch code {
case transport.CredentialErrorGenerationConflict,
transport.CredentialErrorInvalid,
transport.CredentialErrorRegistration,
transport.CredentialErrorRuntimeRejected,
transport.CredentialErrorInternal:
return code
default:
return transport.CredentialErrorInternal
}
}
func hasCapability(capabilities []string, want string) bool {
for _, capability := range capabilities {
if capability == want {
return true
}
}
return false
}
// closeOnContext spawns a goroutine that closes conn when ctx is done.
// This unblocks any pending Read on conn so the main loop can return.
func closeOnContext(ctx context.Context, conn net.Conn) {
+18 -2
View File
@@ -16,6 +16,7 @@ import (
"time"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
@@ -53,6 +54,7 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
if cfg.WorkDir == "" || cfg.IPCEndpoint == "" || cfg.ClientID == "" {
return errors.New("consume: WorkDir, IPCEndpoint, and ClientID are required")
}
cfg.RuntimeToken = strings.TrimSpace(cfg.RuntimeToken)
if len(specs) < 2 {
return errors.New("consume: RunMany requires at least two consumers")
}
@@ -127,14 +129,18 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
}
sessions = append(sessions, session)
closeOnContext(ctx, session.conn)
if err := session.w.WriteJSON(transport.Hello{
hello := transport.Hello{
Type: transport.FrameTypeHello,
ConsumerPID: os.Getpid(),
EventTypes: spec.EventTypes,
Filter: spec.Filter,
SubscribeID: spec.SubscribeID,
Compact: cfg.Compact,
}); err != nil {
}
if cfg.RuntimeToken != "" {
hello.CredentialMode = transport.CredentialModeRuntimeToken
}
if err := session.w.WriteJSON(hello); err != nil {
return fmt.Errorf("consume: write hello for %s: %w", spec.EventKey, err)
}
if err := session.r.ReadJSON(&session.ack); err != nil {
@@ -143,9 +149,16 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
if session.ack.Type != transport.FrameTypeHelloAck {
return fmt.Errorf("consume: unexpected first frame type %q for %s", session.ack.Type, spec.EventKey)
}
// Verify that every connection reached the same bus before handing a
// runtime credential to it. Discovery is expected to converge on one
// daemon, but a stale endpoint/race must not propagate the host token to
// an unrelated process merely so we can report the PID mismatch later.
if len(sessions) > 1 && session.ack.BusPID != sessions[0].ack.BusPID {
return fmt.Errorf("consume: consumers connected to different bus processes (%d and %d)", sessions[0].ack.BusPID, session.ack.BusPID)
}
if err := negotiateRuntimeToken(session.w, session.r, session.ack, cfg.RuntimeToken); err != nil {
return fmt.Errorf("consume: runtime credential handshake for %s: %w", spec.EventKey, err)
}
}
if !cfg.Quiet {
@@ -225,6 +238,9 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
case transport.FrameTypeBye:
var bye transport.Bye
_ = json.Unmarshal(frame.raw, &bye)
if bye.Reason == transport.ByeReasonRuntimeTokenRejected {
return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
}
if bye.Reason == transport.ByeReasonSubscriptionStopped {
delete(active, frame.index)
_ = sessions[frame.index].conn.Close()
+117 -12
View File
@@ -16,6 +16,7 @@ import (
"time"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
@@ -37,23 +38,25 @@ func (b *synchronizedBuffer) String() string {
}
type manyFakeBus struct {
client net.Conn
server net.Conn
hello chan transport.Hello
acked chan struct{}
ackGate <-chan struct{}
ack transport.HelloAck
send chan any
client net.Conn
server net.Conn
hello chan transport.Hello
credentialUpdate chan transport.CredentialUpdate
acked chan struct{}
ackGate <-chan struct{}
ack transport.HelloAck
send chan any
}
func newManyFakeBus(busPID int, ackGate <-chan struct{}) *manyFakeBus {
client, server := net.Pipe()
f := &manyFakeBus{
client: client,
server: server,
hello: make(chan transport.Hello, 1),
acked: make(chan struct{}),
ackGate: ackGate,
client: client,
server: server,
hello: make(chan transport.Hello, 1),
credentialUpdate: make(chan transport.CredentialUpdate, 1),
acked: make(chan struct{}),
ackGate: ackGate,
ack: transport.HelloAck{
Type: transport.FrameTypeHelloAck,
BusPID: busPID,
@@ -85,6 +88,23 @@ func (f *manyFakeBus) serve() {
if err := w.WriteJSON(f.ack); err != nil {
return
}
if hello.CredentialMode == transport.CredentialModeRuntimeToken {
if !hasCapability(f.ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
return
}
var update transport.CredentialUpdate
if err := r.ReadJSON(&update); err != nil {
return
}
f.credentialUpdate <- update
if err := w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: true,
CredentialGeneration: f.ack.CredentialGeneration + 1,
}); err != nil {
return
}
}
close(f.acked)
go func() {
for {
@@ -221,6 +241,91 @@ func TestRunManyWaitsForAllConsumersAndStopsOneAtATime(t *testing.T) {
}
}
func TestCrossPlatformCoverageRunManyRuntimeTokenUsesEachConnectionGeneration(t *testing.T) {
const canary = "many-runtime-canary"
busA := newManyFakeBus(111, nil)
busB := newManyFakeBus(111, nil)
busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
busA.ack.CredentialGeneration = 2
busB.ack.CredentialGeneration = 8
installManyDiscover(t, busA, busB)
cfg := manyTestConfig(io.Discard, io.Discard)
cfg.RuntimeToken = canary
done := make(chan error, 1)
go func() { done <- RunMany(context.Background(), cfg, manyTestSpecs()) }()
updateA := <-busA.credentialUpdate
updateB := <-busB.credentialUpdate
if updateA.Token != canary || updateA.ExpectedGeneration != 2 {
t.Fatal("first connection used the wrong credential or generation")
}
if updateB.Token != canary || updateB.ExpectedGeneration != 8 {
t.Fatal("second connection used the wrong credential or generation")
}
busA.send <- transport.Bye{Type: transport.FrameTypeBye, Reason: "shutdown"}
select {
case err := <-done:
if err != nil {
t.Fatal(err)
}
case <-time.After(2 * time.Second):
t.Fatal("RunMany did not stop")
}
}
func TestCrossPlatformCoverageRunManyRuntimeTokenRejectsDifferentBusBeforeSecondCredential(t *testing.T) {
const canary = "many-mismatched-bus-canary"
busA := newManyFakeBus(111, nil)
busB := newManyFakeBus(222, nil)
busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
installManyDiscover(t, busA, busB)
cfg := manyTestConfig(io.Discard, io.Discard)
cfg.RuntimeToken = canary
err := RunMany(context.Background(), cfg, manyTestSpecs())
if err == nil || !strings.Contains(err.Error(), "different bus processes") {
t.Fatalf("RunMany() error = %v", err)
}
if update := <-busA.credentialUpdate; update.Token != canary {
t.Fatal("first bus did not receive the negotiated credential")
}
select {
case update := <-busB.credentialUpdate:
t.Fatalf("mismatched second bus received credential: generation=%d", update.ExpectedGeneration)
default:
}
}
func TestCrossPlatformCoverageRunManyRuntimeTokenRejectedByeReturnsTypedError(t *testing.T) {
busA := newManyFakeBus(333, nil)
busB := newManyFakeBus(333, nil)
busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
installManyDiscover(t, busA, busB)
cfg := manyTestConfig(io.Discard, io.Discard)
cfg.RuntimeToken = "many-runtime-rejected-canary"
done := make(chan error, 1)
go func() { done <- RunMany(context.Background(), cfg, manyTestSpecs()) }()
<-busA.credentialUpdate
<-busB.credentialUpdate
busA.send <- transport.Bye{
Type: transport.FrameTypeBye,
Reason: transport.ByeReasonRuntimeTokenRejected,
}
select {
case err := <-done:
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("RunMany() error = %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("RunMany did not return runtime credential rejection")
}
}
func TestRunManyMaxEventsIsSharedAcrossConsumers(t *testing.T) {
busA := newManyFakeBus(202, nil)
busB := newManyFakeBus(202, nil)
@@ -0,0 +1,242 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
package consume
import (
"bytes"
"context"
"errors"
"io"
"net"
"strings"
"testing"
"time"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
)
func installRuntimeCredentialDiscover(t *testing.T, serve func(net.Conn)) {
t.Helper()
oldDiscover := discoverBus
done := make(chan struct{})
discoverBus = func(busctl.DiscoverConfig) (net.Conn, error) {
client, server := net.Pipe()
go func() {
defer close(done)
defer server.Close()
serve(server)
}()
return client, nil
}
t.Cleanup(func() {
discoverBus = oldDiscover
select {
case <-done:
case <-time.After(time.Second):
t.Error("fake runtime credential bus did not stop")
}
})
}
func TestCrossPlatformCoverageRunRuntimeTokenNegotiatesBeforeReady(t *testing.T) {
const canary = "consume-runtime-canary"
helloSeen := make(chan transport.Hello, 1)
updateSeen := make(chan transport.CredentialUpdate, 1)
installRuntimeCredentialDiscover(t, func(conn net.Conn) {
r, w := transport.NewReader(conn), transport.NewWriter(conn)
var hello transport.Hello
if err := r.ReadJSON(&hello); err != nil {
return
}
helloSeen <- hello
_ = w.WriteJSON(transport.HelloAck{
Type: transport.FrameTypeHelloAck,
BusPID: 71,
Capabilities: []string{transport.CapabilityRuntimeTokenV1},
CredentialGeneration: 3,
})
var update transport.CredentialUpdate
if err := r.ReadJSON(&update); err != nil {
return
}
updateSeen <- update
_ = w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: true,
CredentialGeneration: 4,
})
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"})
})
var stderr bytes.Buffer
cfg := validRunConfig()
cfg.RuntimeToken = " " + canary + " "
cfg.Stderr = &stderr
if err := Run(context.Background(), cfg); err != nil {
t.Fatal(err)
}
hello := <-helloSeen
if hello.CredentialMode != transport.CredentialModeRuntimeToken {
t.Fatalf("credential mode = %q", hello.CredentialMode)
}
update := <-updateSeen
if update.ExpectedGeneration != 3 || update.Token != canary {
t.Fatal("credential update used the wrong token or generation")
}
if !strings.Contains(stderr.String(), "[event] ready bus_pid=71") {
t.Fatalf("ready marker missing: %s", stderr.String())
}
}
func TestCrossPlatformCoverageRunRuntimeTokenMissingCapabilityFailsBeforeSendingSecretOrReady(t *testing.T) {
const canary = "unsupported-canary-secret"
peerBytes := make(chan string, 1)
installRuntimeCredentialDiscover(t, func(conn net.Conn) {
r, w := transport.NewReader(conn), transport.NewWriter(conn)
var hello transport.Hello
if err := r.ReadJSON(&hello); err != nil {
return
}
_ = w.WriteJSON(transport.HelloAck{Type: transport.FrameTypeHelloAck, BusPID: 72})
raw, _ := io.ReadAll(conn)
peerBytes <- string(raw)
})
var stderr bytes.Buffer
cfg := validRunConfig()
cfg.RuntimeToken = canary
cfg.Stderr = &stderr
err := Run(context.Background(), cfg)
if !errors.Is(err, ErrRuntimeTokenUnsupported) {
t.Fatalf("Run error = %v", err)
}
if strings.Contains(err.Error(), canary) {
t.Fatal("unsupported-bus error contained runtime token")
}
for _, recoveryStep := range []string{
"dws event status --as user --format json",
"dws event stop --as user --all --dry-run",
"dws event stop --as user --all --yes",
} {
if !strings.Contains(err.Error(), recoveryStep) {
t.Fatalf("unsupported-bus error missing recovery step %q: %v", recoveryStep, err)
}
}
if strings.Contains(stderr.String(), "[event] ready") {
t.Fatal("ready marker was written before capability negotiation succeeded")
}
if got := <-peerBytes; strings.Contains(got, canary) || got != "" {
t.Fatal("client sent data after unsupported capability acknowledgement")
}
}
func TestCrossPlatformCoverageRunRuntimeTokenRejectedDoesNotSurfacePeerText(t *testing.T) {
const canary = "rejected-canary-secret"
installRuntimeCredentialDiscover(t, func(conn net.Conn) {
r, w := transport.NewReader(conn), transport.NewWriter(conn)
var hello transport.Hello
if err := r.ReadJSON(&hello); err != nil {
return
}
_ = w.WriteJSON(transport.HelloAck{
Type: transport.FrameTypeHelloAck,
Capabilities: []string{transport.CapabilityRuntimeTokenV1},
})
var update transport.CredentialUpdate
if err := r.ReadJSON(&update); err != nil {
return
}
_ = w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: false,
ErrorCode: "malicious-code-" + update.Token,
Error: "malicious echo " + update.Token,
})
})
cfg := validRunConfig()
cfg.RuntimeToken = canary
err := Run(context.Background(), cfg)
if !errors.Is(err, ErrRuntimeTokenUpdate) {
t.Fatalf("Run error = %v", err)
}
if strings.Contains(err.Error(), canary) {
t.Fatal("credential update error contained peer-provided token text")
}
}
func TestCrossPlatformCoverageRuntimeTokenNeverAppearsInDryRun(t *testing.T) {
const canary = "dry-run-canary-secret"
var output bytes.Buffer
cfg := validRunConfig()
cfg.RuntimeToken = canary
PrintDryRun(&output, cfg)
if strings.Contains(output.String(), canary) || strings.Contains(output.String(), "RuntimeToken") {
t.Fatal("dry-run output contained runtime-token data")
}
}
func TestCrossPlatformCoverageRunWhitespaceRuntimeTokenUsesLegacyHandshake(t *testing.T) {
helloSeen := make(chan transport.Hello, 1)
installRuntimeCredentialDiscover(t, func(conn net.Conn) {
r, w := transport.NewReader(conn), transport.NewWriter(conn)
var hello transport.Hello
if err := r.ReadJSON(&hello); err != nil {
return
}
helloSeen <- hello
_ = w.WriteJSON(transport.HelloAck{Type: transport.FrameTypeHelloAck})
_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"})
})
cfg := validRunConfig()
cfg.RuntimeToken = " "
if err := Run(context.Background(), cfg); err != nil {
t.Fatal(err)
}
if hello := <-helloSeen; hello.CredentialMode != "" {
t.Fatalf("whitespace token enabled mode %q", hello.CredentialMode)
}
}
func TestCrossPlatformCoverageRunRuntimeTokenRejectedByeReturnsTypedError(t *testing.T) {
installRuntimeCredentialDiscover(t, func(conn net.Conn) {
r, w := transport.NewReader(conn), transport.NewWriter(conn)
var hello transport.Hello
if err := r.ReadJSON(&hello); err != nil {
return
}
_ = w.WriteJSON(transport.HelloAck{
Type: transport.FrameTypeHelloAck,
BusPID: 73,
Capabilities: []string{transport.CapabilityRuntimeTokenV1},
})
var update transport.CredentialUpdate
if err := r.ReadJSON(&update); err != nil {
return
}
_ = w.WriteJSON(transport.CredentialUpdateAck{
Type: transport.FrameTypeCredentialUpdateAck,
Accepted: true,
CredentialGeneration: 1,
})
_ = w.WriteJSON(transport.Bye{
Type: transport.FrameTypeBye,
Reason: transport.ByeReasonRuntimeTokenRejected,
})
})
var stderr bytes.Buffer
cfg := validRunConfig()
cfg.RuntimeToken = "runtime-rejected-canary"
cfg.Stderr = &stderr
err := Run(context.Background(), cfg)
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("Run() error = %v", err)
}
if strings.Contains(stderr.String(), "reason: bus_shutdown") {
t.Fatalf("runtime rejection was reported as a successful exit: %s", stderr.String())
}
}
+3 -2
View File
@@ -113,8 +113,9 @@ func IsValidationError(err error) bool {
// human-readable block. Called by Run when cfg.DryRun is true. Format
// avoids JSON so users can `dws event consume --dry-run | head` cleanly.
//
// Secret-bearing fields are never present in Config (credentials never
// reach this layer), so no redaction is required here.
// RuntimeToken is the only secret-bearing Config field and is deliberately
// not read or rendered here. Keep this function allowlist-based: never switch
// it to generic struct serialization.
func PrintDryRun(w io.Writer, cfg Config) {
if w == nil {
return
+300
View File
@@ -0,0 +1,300 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Package runtimecred provides an in-memory credential broker for event bus
// processes. Runtime credentials are never persisted by this package.
package runtimecred
import (
"context"
"errors"
"fmt"
"strings"
"sync"
)
// DefaultMaxTokenBytes bounds credentials accepted over the local IPC
// channel. Access tokens are normally only a few KiB; the larger limit leaves
// ample room for future token formats while avoiding accidental large secret
// allocations.
const DefaultMaxTokenBytes = 64 << 10
var (
ErrEmptyToken = errors.New("runtime credential: token is empty")
ErrTokenTooLarge = errors.New("runtime credential: token exceeds size limit")
ErrCredentialUnavailable = errors.New("runtime credential: no credential resolver is available")
ErrLocalRefreshUnavailable = errors.New("runtime credential: no local refresh callback is available")
ErrRuntimeTokenRejected = &RuntimeTokenRejectedError{}
)
// RuntimeTokenRejectedError means the currently installed runtime token was
// rejected and no newer runtime token is available. It deliberately carries
// no token or server response data so it is safe to surface to users and logs.
type RuntimeTokenRejectedError struct{}
func (*RuntimeTokenRejectedError) Error() string {
return "event runtime token was rejected; retry with a fresh host credential"
}
func (*RuntimeTokenRejectedError) Is(target error) bool {
_, ok := target.(*RuntimeTokenRejectedError)
return ok
}
// GenerationConflictError reports a failed compare-and-swap update.
type GenerationConflictError struct {
Expected uint64
Actual uint64
}
func (e *GenerationConflictError) Error() string {
return fmt.Sprintf("runtime credential: generation conflict (expected %d, actual %d)", e.Expected, e.Actual)
}
// ResolveFunc resolves the existing local OAuth credential when no runtime
// credential has been installed.
type ResolveFunc func(context.Context) (string, error)
// RefreshFunc refreshes a rejected local OAuth credential. It is never called
// after a runtime credential has been installed.
type RefreshFunc func(context.Context, string) (string, error)
type Config struct {
LocalResolve ResolveFunc
LocalRefresh RefreshFunc
RequireSeed bool
// RequireActivation keeps the first installed runtime credential pending
// until Activate is called. Detached buses use it to register the consumer
// before ticket acquisition can emit or fail.
RequireActivation bool
MaxTokenBytes int
}
// Broker holds at most one runtime credential. All state, including the
// credential generation, is process-local and concurrency-safe.
type Broker struct {
localResolve ResolveFunc
localRefresh RefreshFunc
requireSeed bool
requireActivation bool
maxBytes int
mu sync.Mutex
token string
generation uint64
active bool
changed chan struct{}
}
func New(cfg Config) *Broker {
maxBytes := cfg.MaxTokenBytes
if maxBytes <= 0 {
maxBytes = DefaultMaxTokenBytes
}
return &Broker{
localResolve: cfg.LocalResolve,
localRefresh: cfg.LocalRefresh,
requireSeed: cfg.RequireSeed,
requireActivation: cfg.RequireActivation,
maxBytes: maxBytes,
active: !cfg.RequireActivation,
changed: make(chan struct{}),
}
}
// Generation returns the current runtime credential generation. Generation 0
// means that no runtime credential has been installed yet.
func (b *Broker) Generation() uint64 {
if b == nil {
return 0
}
b.mu.Lock()
defer b.mu.Unlock()
return b.generation
}
// Update atomically installs token when expectedGeneration matches the
// current generation. Reinstalling the same token is idempotent, including
// when another concurrent writer already installed it.
func (b *Broker) Update(expectedGeneration uint64, token string) (uint64, error) {
if b == nil {
return 0, ErrCredentialUnavailable
}
normalized, err := b.validate(token)
if err != nil {
return b.Generation(), err
}
b.mu.Lock()
defer b.mu.Unlock()
if b.token == normalized {
return b.generation, nil
}
if expectedGeneration != b.generation {
return b.generation, &GenerationConflictError{Expected: expectedGeneration, Actual: b.generation}
}
b.token = normalized
b.generation++
if b.active {
b.signalChangedLocked()
}
return b.generation, nil
}
// Activate publishes a pending first runtime credential after the bus has
// registered the initiating consumer. It is an idempotent generation-checked
// no-op for brokers that do not require activation.
func (b *Broker) Activate(expectedGeneration uint64) (uint64, error) {
if b == nil {
return 0, ErrCredentialUnavailable
}
b.mu.Lock()
defer b.mu.Unlock()
if expectedGeneration != b.generation {
return b.generation, &GenerationConflictError{Expected: expectedGeneration, Actual: b.generation}
}
if b.token == "" {
return b.generation, ErrCredentialUnavailable
}
if b.active {
return b.generation, nil
}
b.active = true
b.signalChangedLocked()
return b.generation, nil
}
// Resolve returns the runtime credential when installed. In RequireSeed mode
// it waits until Update installs one; otherwise it preserves the existing
// local resolver behavior until a runtime credential arrives.
func (b *Broker) Resolve(ctx context.Context) (string, error) {
if b == nil {
return "", ErrCredentialUnavailable
}
for {
token, wait, requireSeed, active := b.snapshot()
if token != "" && active {
return token, nil
}
if requireSeed || (token != "" && !active) {
select {
case <-ctx.Done():
return "", ctx.Err()
case <-wait:
continue
}
}
if b.localResolve == nil {
return "", ErrCredentialUnavailable
}
local, err := b.localResolve(ctx)
if err != nil {
return "", err
}
if runtime, _, _, runtimeActive := b.snapshot(); runtime != "" && runtimeActive {
return runtime, nil
}
return validateResolved(local)
}
}
// RefreshRejected returns a newer runtime token if one was installed after
// rejectedToken was used. If the installed runtime token itself was rejected,
// it returns RuntimeTokenRejectedError and never invokes local OAuth refresh.
func (b *Broker) RefreshRejected(ctx context.Context, rejectedToken string) (string, error) {
if b == nil {
return "", ErrCredentialUnavailable
}
for {
token, wait, requireSeed, active := b.snapshot()
if token != "" && active {
if token != rejectedToken {
return token, nil
}
return "", &RuntimeTokenRejectedError{}
}
if requireSeed || (token != "" && !active) {
select {
case <-ctx.Done():
return "", ctx.Err()
case <-wait:
continue
}
}
if b.localRefresh == nil {
return "", ErrLocalRefreshUnavailable
}
refreshed, err := b.localRefresh(ctx, rejectedToken)
if err != nil {
return "", err
}
if runtime, _, _, runtimeActive := b.snapshot(); runtime != "" && runtimeActive {
if runtime != rejectedToken {
return runtime, nil
}
return "", &RuntimeTokenRejectedError{}
}
return validateResolved(refreshed)
}
}
// ClassifyRejectedAfterRetry is called when a token returned by
// RefreshRejected was itself rejected. When a still newer runtime credential
// is already installed, superseded is true so the source may reconnect and
// resolve that generation without a second in-attempt retry. When the rejected
// token is still current, the fixed typed rejection is returned. With no
// runtime credential installed it preserves local OAuth behavior by returning
// (false, nil).
func (b *Broker) ClassifyRejectedAfterRetry(rejectedToken string) (superseded bool, err error) {
if b == nil {
return false, ErrCredentialUnavailable
}
token, _, _, active := b.snapshot()
if token == "" || !active {
return false, nil
}
if token != strings.TrimSpace(rejectedToken) {
return true, nil
}
return false, &RuntimeTokenRejectedError{}
}
func (b *Broker) snapshot() (string, <-chan struct{}, bool, bool) {
b.mu.Lock()
defer b.mu.Unlock()
return b.token, b.changed, b.requireSeed, b.active
}
func (b *Broker) signalChangedLocked() {
close(b.changed)
b.changed = make(chan struct{})
}
func (b *Broker) validate(token string) (string, error) {
normalized := strings.TrimSpace(token)
if normalized == "" {
return "", ErrEmptyToken
}
if len(normalized) > b.maxBytes {
return "", ErrTokenTooLarge
}
return normalized, nil
}
func validateResolved(token string) (string, error) {
token = strings.TrimSpace(token)
if token == "" {
return "", ErrCredentialUnavailable
}
return token, nil
}
+203
View File
@@ -0,0 +1,203 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
package runtimecred
import (
"context"
"errors"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
)
func TestCrossPlatformCoverageBrokerLocalFallbackAndRuntimePrecedence(t *testing.T) {
var localCalls atomic.Int32
b := New(Config{LocalResolve: func(context.Context) (string, error) {
localCalls.Add(1)
return "local", nil
}})
if got, err := b.Resolve(context.Background()); err != nil || got != "local" {
t.Fatalf("local Resolve = %q, %v", got, err)
}
gen, err := b.Update(0, " runtime ")
if err != nil || gen != 1 {
t.Fatalf("Update = %d, %v", gen, err)
}
if got, err := b.Resolve(context.Background()); err != nil || got != "runtime" {
t.Fatalf("runtime Resolve = %q, %v", got, err)
}
if localCalls.Load() != 1 {
t.Fatalf("local resolver calls = %d", localCalls.Load())
}
}
func TestCrossPlatformCoverageBrokerRequireSeedWaitsAndCancels(t *testing.T) {
b := New(Config{RequireSeed: true})
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { _, err := b.Resolve(ctx); done <- err }()
cancel()
if err := <-done; !errors.Is(err, context.Canceled) {
t.Fatalf("Resolve error = %v", err)
}
resolved := make(chan string, 1)
go func() {
token, _ := b.Resolve(context.Background())
resolved <- token
}()
select {
case <-resolved:
t.Fatal("Resolve returned before seed")
case <-time.After(20 * time.Millisecond):
}
if _, err := b.Update(0, "seed"); err != nil {
t.Fatal(err)
}
if got := <-resolved; got != "seed" {
t.Fatalf("Resolve = %q", got)
}
}
func TestCrossPlatformCoverageBrokerUpdateCASIdempotenceAndConcurrentConflict(t *testing.T) {
b := New(Config{})
if gen, err := b.Update(0, "same"); err != nil || gen != 1 {
t.Fatalf("first update = %d, %v", gen, err)
}
if gen, err := b.Update(0, "same"); err != nil || gen != 1 {
t.Fatalf("idempotent stale update = %d, %v", gen, err)
}
if _, err := b.Update(0, "different"); err == nil {
t.Fatal("expected generation conflict")
} else {
var conflict *GenerationConflictError
if !errors.As(err, &conflict) || conflict.Actual != 1 {
t.Fatalf("conflict = %#v, %v", conflict, err)
}
}
b = New(Config{})
start := make(chan struct{})
var successes atomic.Int32
var conflicts atomic.Int32
var wg sync.WaitGroup
for _, token := range []string{"a", "b"} {
wg.Add(1)
go func(token string) {
defer wg.Done()
<-start
_, err := b.Update(0, token)
if err == nil {
successes.Add(1)
return
}
var conflict *GenerationConflictError
if errors.As(err, &conflict) {
conflicts.Add(1)
}
}(token)
}
close(start)
wg.Wait()
if successes.Load() != 1 || conflicts.Load() != 1 {
t.Fatalf("successes=%d conflicts=%d", successes.Load(), conflicts.Load())
}
}
func TestCrossPlatformCoverageBrokerRefreshRejectedRuntimeNeverFallsBack(t *testing.T) {
var refreshCalls atomic.Int32
b := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
refreshCalls.Add(1)
return "local-new", nil
}})
if _, err := b.Update(0, "runtime-a"); err != nil {
t.Fatal(err)
}
if _, err := b.RefreshRejected(context.Background(), "runtime-a"); !errors.Is(err, ErrRuntimeTokenRejected) {
t.Fatalf("same token refresh error = %v", err)
}
if refreshCalls.Load() != 0 {
t.Fatalf("local refresh called %d times", refreshCalls.Load())
}
if _, err := b.Update(1, "runtime-b"); err != nil {
t.Fatal(err)
}
if got, err := b.RefreshRejected(context.Background(), "runtime-a"); err != nil || got != "runtime-b" {
t.Fatalf("rotated refresh = %q, %v", got, err)
}
if superseded, err := b.ClassifyRejectedAfterRetry("runtime-b"); superseded || !errors.Is(err, ErrRuntimeTokenRejected) {
t.Fatalf("current retry rejection = superseded %v, error %v", superseded, err)
}
if _, err := b.Update(2, "runtime-c"); err != nil {
t.Fatal(err)
}
if superseded, err := b.ClassifyRejectedAfterRetry("runtime-b"); !superseded || err != nil {
t.Fatalf("newer generation classification = superseded %v, error %v", superseded, err)
}
localOnly := New(Config{})
if superseded, err := localOnly.ClassifyRejectedAfterRetry("local"); superseded || err != nil {
t.Fatalf("local-only classification = superseded %v, error %v", superseded, err)
}
}
func TestCrossPlatformCoverageBrokerDefersSeedUntilActivation(t *testing.T) {
b := New(Config{RequireSeed: true, RequireActivation: true})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
resolved := make(chan string, 1)
errs := make(chan error, 1)
go func() {
token, err := b.Resolve(ctx)
if err != nil {
errs <- err
return
}
resolved <- token
}()
generation, err := b.Update(0, "runtime-a")
if err != nil || generation != 1 {
t.Fatalf("Update() = generation %d, err %v", generation, err)
}
select {
case token := <-resolved:
t.Fatalf("Resolve() returned before activation: %q", token)
case err := <-errs:
t.Fatalf("Resolve() failed before activation: %v", err)
case <-time.After(20 * time.Millisecond):
}
if _, err := b.Activate(0); err == nil {
t.Fatal("Activate() with stale generation unexpectedly succeeded")
}
if activeGeneration, err := b.Activate(generation); err != nil || activeGeneration != generation {
t.Fatalf("Activate() = generation %d, err %v", activeGeneration, err)
}
select {
case token := <-resolved:
if token != "runtime-a" {
t.Fatalf("Resolve() token = %q", token)
}
case err := <-errs:
t.Fatalf("Resolve() failed after activation: %v", err)
case <-time.After(time.Second):
t.Fatal("Resolve() remained blocked after activation")
}
}
func TestCrossPlatformCoverageBrokerRejectsInvalidTokensWithoutEcho(t *testing.T) {
b := New(Config{MaxTokenBytes: 4})
for _, token := range []string{" ", "secret-token"} {
_, err := b.Update(0, token)
if err == nil {
t.Fatalf("Update(%q) succeeded", token)
}
if strings.Contains(err.Error(), token) {
t.Fatal("validation error contained rejected token")
}
}
}
@@ -0,0 +1,150 @@
package runtimecred
import (
"context"
"errors"
"testing"
"time"
)
func TestCrossPlatformCoverageEventCoreBrokerEdges(t *testing.T) {
rejected := &RuntimeTokenRejectedError{}
if rejected.Error() == "" || !errors.Is(rejected, ErrRuntimeTokenRejected) || rejected.Is(errors.New("other")) {
t.Fatal("runtime rejection error contract failed")
}
conflict := &GenerationConflictError{Expected: 2, Actual: 3}
if conflict.Error() == "" {
t.Fatal("generation conflict error is empty")
}
var nilBroker *Broker
if nilBroker.Generation() != 0 {
t.Fatal("nil broker generation is non-zero")
}
if _, err := nilBroker.Update(0, "token"); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("nil Update error = %v", err)
}
if _, err := nilBroker.Activate(0); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("nil Activate error = %v", err)
}
if _, err := nilBroker.Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("nil Resolve error = %v", err)
}
if _, err := nilBroker.RefreshRejected(context.Background(), "token"); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("nil RefreshRejected error = %v", err)
}
if superseded, err := nilBroker.ClassifyRejectedAfterRetry("token"); superseded || !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("nil classification = %v, %v", superseded, err)
}
b := New(Config{})
if _, err := b.Activate(0); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("empty Activate error = %v", err)
}
if _, err := b.Update(0, "token"); err != nil {
t.Fatal(err)
}
if generation, err := b.Activate(1); err != nil || generation != 1 {
t.Fatalf("active Activate = %d, %v", generation, err)
}
if _, err := New(Config{}).Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("missing local resolver error = %v", err)
}
wantResolveErr := errors.New("resolve failed")
if _, err := New(Config{LocalResolve: func(context.Context) (string, error) {
return "", wantResolveErr
}}).Resolve(context.Background()); !errors.Is(err, wantResolveErr) {
t.Fatalf("local resolve error = %v", err)
}
if _, err := New(Config{LocalResolve: func(context.Context) (string, error) {
return " ", nil
}}).Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("empty resolved credential error = %v", err)
}
var racingResolve *Broker
racingResolve = New(Config{LocalResolve: func(context.Context) (string, error) {
_, err := racingResolve.Update(0, "runtime-wins")
return "local", err
}})
if token, err := racingResolve.Resolve(context.Background()); err != nil || token != "runtime-wins" {
t.Fatalf("runtime precedence after resolve = %q, %v", token, err)
}
seed := New(Config{RequireSeed: true})
canceled, cancel := context.WithCancel(context.Background())
cancel()
if _, err := seed.Resolve(canceled); !errors.Is(err, context.Canceled) {
t.Fatalf("seed cancellation = %v", err)
}
if _, err := New(Config{}).RefreshRejected(context.Background(), "local"); !errors.Is(err, ErrLocalRefreshUnavailable) {
t.Fatalf("missing local refresh error = %v", err)
}
wantRefreshErr := errors.New("refresh failed")
if _, err := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
return "", wantRefreshErr
}}).RefreshRejected(context.Background(), "local"); !errors.Is(err, wantRefreshErr) {
t.Fatalf("local refresh error = %v", err)
}
if _, err := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
return " ", nil
}}).RefreshRejected(context.Background(), "local"); !errors.Is(err, ErrCredentialUnavailable) {
t.Fatalf("empty refreshed credential error = %v", err)
}
var racingRefresh *Broker
racingRefresh = New(Config{LocalRefresh: func(context.Context, string) (string, error) {
_, err := racingRefresh.Update(0, "runtime-new")
return "local-new", err
}})
if token, err := racingRefresh.RefreshRejected(context.Background(), "runtime-old"); err != nil || token != "runtime-new" {
t.Fatalf("runtime precedence after refresh = %q, %v", token, err)
}
var sameRefresh *Broker
sameRefresh = New(Config{LocalRefresh: func(context.Context, string) (string, error) {
_, err := sameRefresh.Update(0, "same")
return "local-new", err
}})
if _, err := sameRefresh.RefreshRejected(context.Background(), "same"); !errors.Is(err, ErrRuntimeTokenRejected) {
t.Fatalf("same runtime after refresh error = %v", err)
}
pending := New(Config{RequireSeed: true, RequireActivation: true})
generation, err := pending.Update(0, "pending")
if err != nil {
t.Fatal(err)
}
waitCtx, waitCancel := context.WithCancel(context.Background())
waitCancel()
if _, err := pending.RefreshRejected(waitCtx, "pending"); !errors.Is(err, context.Canceled) {
t.Fatalf("pending refresh cancellation = %v", err)
}
refreshDone := make(chan error, 1)
go func() {
_, err := pending.RefreshRejected(context.Background(), "pending")
refreshDone <- err
}()
// Let RefreshRejected take the pending credential's changed-channel path
// before activation publishes it.
time.Sleep(10 * time.Millisecond)
if _, err := pending.Activate(generation); err != nil {
t.Fatal(err)
}
select {
case err := <-refreshDone:
if !errors.Is(err, ErrRuntimeTokenRejected) {
t.Fatalf("activated pending refresh error = %v", err)
}
case <-time.After(time.Second):
t.Fatal("pending refresh did not resume")
}
inactive := New(Config{RequireActivation: true})
if _, err := inactive.Update(0, "inactive"); err != nil {
t.Fatal(err)
}
if superseded, err := inactive.ClassifyRejectedAfterRetry("other"); superseded || err != nil {
t.Fatalf("inactive classification = %v, %v", superseded, err)
}
}
@@ -0,0 +1,40 @@
package source
import (
"context"
"io"
"net/http"
"strings"
"testing"
)
func TestCrossPlatformCoverageEventCorePersonalSupersededRetry(t *testing.T) {
source, err := NewPersonal(PersonalConfig{
AccessToken: "stale",
ForceRefreshToken: func(context.Context, string) (string, error) {
return "rotated", nil
},
ClassifyRetryReject: func(token string) (bool, error) {
if token != "rotated" {
t.Fatalf("classified token = %q", token)
}
return true, nil
},
ClientID: "client",
SourceID: "source",
TicketURL: "https://ticket.test",
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusUnauthorized,
Body: io.NopCloser(strings.NewReader("rejected")),
Header: make(http.Header),
}, nil
})},
})
if err != nil {
t.Fatal(err)
}
if _, err := source.fetchTicket(context.Background()); !isRetryablePersonalError(err) {
t.Fatalf("superseded retry error = %v", err)
}
}
+20 -1
View File
@@ -45,6 +45,7 @@ type PersonalConfig struct {
AccessToken string
AccessTokenProvider AccessTokenProvider
ForceRefreshToken ForceRefreshTokenFn
ClassifyRetryReject RetryRejectClassifier
ClientID string
ClientSecret string
SourceID string
@@ -66,6 +67,12 @@ type AccessTokenProvider func(context.Context) (string, error)
// when nil a 401 stays fatal, matching the previous behavior.
type ForceRefreshTokenFn func(ctx context.Context, rejectedToken string) (string, error)
// RetryRejectClassifier classifies a 401 from the one refreshed-token retry.
// superseded means a newer credential is already available and the outer
// reconnect loop should start a fresh attempt; err is a terminal typed
// rejection. A nil callback preserves the historical local OAuth behavior.
type RetryRejectClassifier func(rejectedToken string) (superseded bool, err error)
type PersonalSource struct {
cfg PersonalConfig
machine *Machine
@@ -223,7 +230,19 @@ func (s *PersonalSource) fetchTicket(ctx context.Context) (*ticketResponse, erro
return nil, refreshErr
}
// Retry once with the freshly rotated token; a second 401 stays fatal.
ticket, _, err = s.fetchTicketAttempt(ctx, refreshed)
var retryStatus int
var retryErr error
ticket, retryStatus, retryErr = s.fetchTicketAttempt(ctx, refreshed)
if retryStatus == http.StatusUnauthorized && s.cfg.ClassifyRetryReject != nil {
superseded, classifyErr := s.cfg.ClassifyRetryReject(refreshed)
if classifyErr != nil {
return nil, classifyErr
}
if superseded {
return nil, retryPersonal(retryErr)
}
}
err = retryErr
}
return ticket, err
}
@@ -14,6 +14,7 @@ import (
"time"
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
"github.com/gorilla/websocket"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
)
@@ -316,6 +317,46 @@ func TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal(t *
}
}
func TestCrossPlatformCoveragePersonalRuntimeTokenAtoBSecond401IsTyped(t *testing.T) {
broker := runtimecred.New(runtimecred.Config{})
if _, err := broker.Update(0, "runtime-a"); err != nil {
t.Fatal(err)
}
var attemptTokens []string
src, err := NewPersonal(PersonalConfig{
AccessTokenProvider: broker.Resolve,
ForceRefreshToken: broker.RefreshRejected,
ClassifyRetryReject: broker.ClassifyRejectedAfterRetry,
ClientID: "client",
SourceID: "source",
TicketURL: "https://ticket.test",
HTTPClient: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
token := req.Header.Get("x-user-access-token")
attemptTokens = append(attemptTokens, token)
if token == "runtime-a" {
if _, err := broker.Update(1, "runtime-b"); err != nil {
t.Fatalf("rotate runtime credential: %v", err)
}
}
return &http.Response{
StatusCode: http.StatusUnauthorized,
Body: io.NopCloser(strings.NewReader("must-not-surface")),
Header: make(http.Header),
}, nil
})},
})
if err != nil {
t.Fatal(err)
}
_, err = src.fetchTicket(context.Background())
if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
t.Fatalf("fetchTicket() error = %v", err)
}
if len(attemptTokens) != 2 || attemptTokens[0] != "runtime-a" || attemptTokens[1] != "runtime-b" {
t.Fatalf("attempt tokens = %v", attemptTokens)
}
}
// brokenBody simulates a response body that fails mid-read, e.g. the server
// closing the connection before the error payload is fully written.
type brokenBody struct{}
+60 -12
View File
@@ -26,16 +26,28 @@ import (
type FrameType string
const (
FrameTypeHello FrameType = "hello" // consume → bus
FrameTypeHelloAck FrameType = "hello_ack" // bus → consume
FrameTypeEvent FrameType = "event" // bus → consume
FrameTypeHeartbeat FrameType = "heartbeat" // bidirectional
FrameTypeSourceState FrameType = "source_state" // bus → consume
FrameTypeBye FrameType = "bye" // bidirectional
FrameTypeStatusReq FrameType = "status_req" // consume/ad-hoc → bus
FrameTypeStatusResp FrameType = "status_resp" // bus → consume/ad-hoc
FrameTypeConsumerStopReq FrameType = "consumer_stop_req" // ad-hoc → bus
FrameTypeConsumerStopResp FrameType = "consumer_stop_resp" // bus → ad-hoc
FrameTypeHello FrameType = "hello" // consume → bus
FrameTypeHelloAck FrameType = "hello_ack" // bus → consume
FrameTypeEvent FrameType = "event" // bus → consume
FrameTypeHeartbeat FrameType = "heartbeat" // bidirectional
FrameTypeSourceState FrameType = "source_state" // bus → consume
FrameTypeBye FrameType = "bye" // bidirectional
FrameTypeStatusReq FrameType = "status_req" // consume/ad-hoc → bus
FrameTypeStatusResp FrameType = "status_resp" // bus → consume/ad-hoc
FrameTypeConsumerStopReq FrameType = "consumer_stop_req" // ad-hoc → bus
FrameTypeConsumerStopResp FrameType = "consumer_stop_resp" // bus → ad-hoc
FrameTypeCredentialUpdate FrameType = "credential_update" // consume → bus
FrameTypeCredentialUpdateAck FrameType = "credential_update_ack" // bus → consume
)
// CredentialMode declares that a consumer needs an additive credential
// negotiation before it can register for events. The zero value preserves the
// original protocol.
type CredentialMode string
const (
CredentialModeRuntimeToken CredentialMode = "runtime_token"
CapabilityRuntimeTokenV1 = "runtime_token_v1"
)
// Hello is the first frame a consumer sends after dialing the bus. The bus
@@ -51,7 +63,8 @@ type Hello struct {
// Role distinguishes a real consumer (registered for events) from an
// ad-hoc tooling connection (status/list/stop). Ad-hoc connections do
// NOT register with the Hub.
Role HelloRole `json:"role,omitempty"`
Role HelloRole `json:"role,omitempty"`
CredentialMode CredentialMode `json:"credential_mode,omitempty"`
}
// HelloRole tags the purpose of a Hello connection.
@@ -76,6 +89,38 @@ type HelloAck struct {
ClientIDSource string `json:"client_id_source"` // auth.CredentialSource string
ClientSecretSource string `json:"client_secret_source"` // auth.CredentialSource string
IdleTimeoutSecs int `json:"idle_timeout_secs,omitempty"` // bus's IdleTimeout for diagnostics
Capabilities []string `json:"capabilities,omitempty"`
// CredentialGeneration is process-local and contains no secret data.
CredentialGeneration uint64 `json:"credential_generation"`
// TerminalReason is a fixed, non-sensitive bus terminal state. A runtime
// client checks it before sending credential material.
TerminalReason string `json:"terminal_reason,omitempty"`
}
// CredentialUpdate installs a host-supplied runtime token into a compatible
// bus after the bus has advertised CapabilityRuntimeTokenV1. Token is carried
// only over the owner-only local IPC connection and must never be logged.
type CredentialUpdate struct {
Type FrameType `json:"type"`
ExpectedGeneration uint64 `json:"expected_generation"`
Token string `json:"token"`
}
const (
CredentialErrorGenerationConflict = "generation_conflict"
CredentialErrorInvalid = "invalid_credential"
CredentialErrorRegistration = "registration_failed"
CredentialErrorRuntimeRejected = "runtime_token_rejected"
CredentialErrorInternal = "internal_error"
)
// CredentialUpdateAck reports the CAS result without echoing any credential.
type CredentialUpdateAck struct {
Type FrameType `json:"type"`
Accepted bool `json:"accepted"`
CredentialGeneration uint64 `json:"credential_generation"`
ErrorCode string `json:"error_code,omitempty"`
Error string `json:"error,omitempty"`
}
// Event wraps one delivered RawEvent for the wire. We keep the payload as
@@ -126,7 +171,10 @@ type Bye struct {
Reason string `json:"reason"`
}
const ByeReasonSubscriptionStopped = "subscription_stopped"
const (
ByeReasonSubscriptionStopped = "subscription_stopped"
ByeReasonRuntimeTokenRejected = "runtime_token_rejected"
)
// ConsumerStopReq asks the bus to close consumers whose exact personal
// subscription IDs match. It is an additive local IPC control operation;
+72 -31
View File
@@ -23,21 +23,23 @@ import (
// to guard against accidental wire-format changes (the JSON field tags
// double as part of the cross-version protocol contract).
func TestFrameType_StableWireValues(t *testing.T) {
func TestCrossPlatformCoverageFrameTypeStableWireValues(t *testing.T) {
// If any of these strings change we've made a protocol-breaking
// change. The test value list is duplicated here on purpose so a
// reviewer renaming a constant is forced to also update the test.
wants := map[FrameType]string{
FrameTypeHello: "hello",
FrameTypeHelloAck: "hello_ack",
FrameTypeEvent: "event",
FrameTypeHeartbeat: "heartbeat",
FrameTypeSourceState: "source_state",
FrameTypeBye: "bye",
FrameTypeStatusReq: "status_req",
FrameTypeStatusResp: "status_resp",
FrameTypeConsumerStopReq: "consumer_stop_req",
FrameTypeConsumerStopResp: "consumer_stop_resp",
FrameTypeHello: "hello",
FrameTypeHelloAck: "hello_ack",
FrameTypeEvent: "event",
FrameTypeHeartbeat: "heartbeat",
FrameTypeSourceState: "source_state",
FrameTypeBye: "bye",
FrameTypeStatusReq: "status_req",
FrameTypeStatusResp: "status_resp",
FrameTypeConsumerStopReq: "consumer_stop_req",
FrameTypeConsumerStopResp: "consumer_stop_resp",
FrameTypeCredentialUpdate: "credential_update",
FrameTypeCredentialUpdateAck: "credential_update_ack",
}
for ft, want := range wants {
if string(ft) != want {
@@ -104,54 +106,91 @@ func roundTrip(t *testing.T, in any, dst any) {
}
}
func TestHello_Roundtrip(t *testing.T) {
func TestCrossPlatformCoverageHelloRoundtrip(t *testing.T) {
in := Hello{
Type: FrameTypeHello,
ConsumerPID: 42,
EventTypes: []string{"im.*", "approval.task"},
Filter: `^im\.`,
Compact: true,
Role: HelloRoleStatus,
Type: FrameTypeHello,
ConsumerPID: 42,
EventTypes: []string{"im.*", "approval.task"},
Filter: `^im\.`,
Compact: true,
Role: HelloRoleStatus,
CredentialMode: CredentialModeRuntimeToken,
}
var out Hello
roundTrip(t, in, &out)
if out.Type != in.Type || out.ConsumerPID != in.ConsumerPID || out.Filter != in.Filter ||
out.Compact != in.Compact || out.Role != in.Role || len(out.EventTypes) != len(in.EventTypes) {
out.Compact != in.Compact || out.Role != in.Role || out.CredentialMode != in.CredentialMode || len(out.EventTypes) != len(in.EventTypes) {
t.Fatalf("roundtrip mismatch: %+v != %+v", out, in)
}
}
func TestHello_OmitemptyForDefaults(t *testing.T) {
func TestCrossPlatformCoverageHelloOmitemptyForDefaults(t *testing.T) {
// Default values should NOT appear in the wire form so old/new readers
// stay tolerant of each other (each new field comes in with its
// zero value by default).
in := Hello{Type: FrameTypeHello, ConsumerPID: 1}
b, _ := json.Marshal(in)
s := string(b)
for _, k := range []string{`"event_types"`, `"filter"`, `"compact"`, `"role"`} {
for _, k := range []string{`"event_types"`, `"filter"`, `"compact"`, `"role"`, `"credential_mode"`} {
if strings.Contains(s, k) {
t.Errorf("zero-value field %s leaked into wire form: %s", k, s)
}
}
}
func TestHelloAck_Roundtrip(t *testing.T) {
func TestCrossPlatformCoverageHelloAckRoundtrip(t *testing.T) {
in := HelloAck{
Type: FrameTypeHelloAck,
BusPID: 12345,
SourceState: "connected",
StateSource: "inferred",
ClientIDSource: "env",
ClientSecretSource: "env",
IdleTimeoutSecs: 300,
Type: FrameTypeHelloAck,
BusPID: 12345,
SourceState: "connected",
StateSource: "inferred",
ClientIDSource: "env",
ClientSecretSource: "env",
IdleTimeoutSecs: 300,
Capabilities: []string{CapabilityRuntimeTokenV1},
CredentialGeneration: 7,
}
var out HelloAck
roundTrip(t, in, &out)
if out != in {
if out.Type != in.Type || out.BusPID != in.BusPID || out.SourceState != in.SourceState ||
out.StateSource != in.StateSource || out.ClientIDSource != in.ClientIDSource ||
out.ClientSecretSource != in.ClientSecretSource || out.IdleTimeoutSecs != in.IdleTimeoutSecs ||
out.CredentialGeneration != in.CredentialGeneration || strings.Join(out.Capabilities, ",") != CapabilityRuntimeTokenV1 {
t.Fatalf("HelloAck roundtrip: %+v != %+v", out, in)
}
}
func TestCrossPlatformCoverageCredentialFramesRoundtripWithoutTokenInAck(t *testing.T) {
update := CredentialUpdate{
Type: FrameTypeCredentialUpdate,
ExpectedGeneration: 4,
Token: "canary-secret",
}
var decodedUpdate CredentialUpdate
roundTrip(t, update, &decodedUpdate)
if decodedUpdate != update {
t.Fatal("CredentialUpdate roundtrip mismatch")
}
ack := CredentialUpdateAck{
Type: FrameTypeCredentialUpdateAck,
Accepted: true,
CredentialGeneration: 5,
}
var decodedAck CredentialUpdateAck
roundTrip(t, ack, &decodedAck)
if decodedAck != ack {
t.Fatalf("CredentialUpdateAck roundtrip = %#v", decodedAck)
}
b, err := json.Marshal(ack)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(b), update.Token) {
t.Fatal("credential acknowledgement contained token")
}
}
func TestEvent_Roundtrip(t *testing.T) {
in := Event{
Type: FrameTypeEvent,
@@ -245,7 +284,7 @@ func TestStatusResp_Roundtrip(t *testing.T) {
// PeekType is used by the daemon to dispatch incoming frames before
// fully decoding into the typed struct. Behaviour at boundaries matters.
func TestPeekType_EachFrameVariant(t *testing.T) {
func TestCrossPlatformCoveragePeekTypeEachFrameVariant(t *testing.T) {
cases := []struct {
v any
typ FrameType
@@ -258,6 +297,8 @@ func TestPeekType_EachFrameVariant(t *testing.T) {
{Bye{Type: FrameTypeBye, Reason: "x"}, FrameTypeBye},
{StatusReq{Type: FrameTypeStatusReq}, FrameTypeStatusReq},
{StatusResp{Type: FrameTypeStatusResp}, FrameTypeStatusResp},
{CredentialUpdate{Type: FrameTypeCredentialUpdate}, FrameTypeCredentialUpdate},
{CredentialUpdateAck{Type: FrameTypeCredentialUpdateAck}, FrameTypeCredentialUpdateAck},
}
for _, c := range cases {
b, _ := json.Marshal(c.v)
+116
View File
@@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""解析 PR 评论中的 `/eval <products> [sha=<full-head-sha>] [cases=<ref>]` 触发命令。
供 .github/workflows/eval-dispatch.yml 调用:评论体经环境变量 COMMENT_BODY
传入(避免 argv 注入),解析结果以 GitHub Actions output
(products / cases_ref / reviewed_sha)写出;格式非法时把单行错误写入
output `error` 并以非零退出。
产品集显式必填;`sha=` 在评测他人 PR 时必填(审核背书凭据,由 guard 校验),
评测自己创建的 PR 时可省略(自助模式,自动钉住派发时刻的当前 head);
`cases=<ref>` 为用例仓库版本逃生舱(破坏性变更配对验证)。
"""
import os
import re
import sys
PRODUCT_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
REF_CHARSET_RE = re.compile(r"^[A-Za-z0-9._/-]+$")
SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$")
USAGE = (
"用法: /eval <product>[,<product>...] [sha=<40位PR-head-SHA>] [cases=<ref>];"
"评测他人 PR 时 sha= 必填,自己的 PR 可省略;"
"示例: /eval drive,doc sha=0123456789abcdef0123456789abcdef01234567"
)
def _is_valid_cases_ref(ref: str) -> bool:
"""等价于 git check-ref-format --allow-onelevel 的结构校验(不执行 git)。
字符白名单之上补齐结构规则:禁止首尾斜杠与连续斜杠、尾部点号、
任意位置的 `..`、以点开头或以 .lock 结尾的路径分量;另禁止 `-`
开头,避免被下游 `git fetch origin <ref>` 当作选项解析。
"""
if not ref or not REF_CHARSET_RE.match(ref):
return False
if ref.startswith(("-", "/")) or ref.endswith(("/", ".")):
return False
if ".." in ref or "//" in ref:
return False
for component in ref.split("/"):
if component.startswith(".") or component.endswith(".lock"):
return False
return True
def parse(body: str):
lines = [line.strip() for line in (body or "").splitlines() if line.strip()]
if not lines:
raise ValueError(f"评论为空。{USAGE}")
tokens = lines[0].split()
if tokens[0] != "/eval":
raise ValueError(f"首行必须以 /eval 命令开头。{USAGE}")
if len(tokens) < 2:
raise ValueError(f"产品集显式必填。{USAGE}")
products = [p for p in tokens[1].split(",") if p]
if not products:
raise ValueError(f"产品集显式必填。{USAGE}")
for product in products:
if not PRODUCT_RE.match(product):
raise ValueError(f"产品名非法: {product}。{USAGE}")
cases_ref = ""
reviewed_sha = ""
for extra in tokens[2:]:
if extra.startswith("cases="):
if cases_ref:
raise ValueError(f"cases 引用只能指定一次。{USAGE}")
cases_ref = extra[len("cases="):]
if not _is_valid_cases_ref(cases_ref):
raise ValueError(f"cases 引用非法: {extra}。{USAGE}")
elif extra.startswith("sha="):
if reviewed_sha:
raise ValueError(f"审核 SHA 只能指定一次。{USAGE}")
reviewed_sha = extra[len("sha="):]
if not SHA_RE.match(reviewed_sha):
raise ValueError(f"审核 SHA 非法: {extra}。{USAGE}")
else:
raise ValueError(f"未知参数: {extra}。{USAGE}")
if not reviewed_sha:
# 是否允许省略由 guard 按“评论者是否 PR 作者”裁决,解析层不拦
return ",".join(products), cases_ref, ""
return ",".join(products), cases_ref, reviewed_sha.lower()
def _write_outputs(pairs):
lines = [f"{key}={value}" for key, value in pairs]
output_path = os.environ.get("GITHUB_OUTPUT", "")
if output_path:
with open(output_path, "a", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
print("\n".join(lines))
def main() -> int:
try:
products, cases_ref, reviewed_sha = parse(os.environ.get("COMMENT_BODY", ""))
except ValueError as exc:
_write_outputs([("error", str(exc))])
print(str(exc), file=sys.stderr)
return 1
_write_outputs(
[
("products", products),
("cases_ref", cases_ref),
("reviewed_sha", reviewed_sha),
]
)
return 0
if __name__ == "__main__":
sys.exit(main())
+92
View File
@@ -0,0 +1,92 @@
import pytest
from eval_comment_parse import parse
SHA = "0123456789abcdef0123456789abcdef01234567"
def test_single_product():
assert parse(f"/eval drive sha={SHA}") == ("drive", "", SHA)
def test_multiple_products():
assert parse(f"/eval drive,doc,edu-app sha={SHA}") == ("drive,doc,edu-app", "", SHA)
def test_cases_ref_escape_hatch():
assert parse(f"/eval drive sha={SHA} cases=feat/drive-latest") == (
"drive",
"feat/drive-latest",
SHA,
)
def test_extra_lines_after_command_are_ignored():
body = f"/eval drive sha={SHA}\n\n顺便说明:这个 PR 只动了 drive 的 --latest。"
assert parse(body) == ("drive", "", SHA)
def test_missing_products_rejected():
with pytest.raises(ValueError, match="产品集显式必填"):
parse("/eval")
def test_omitted_reviewed_sha_defers_to_guard():
# 是否允许省略由 guard 按“评论者是否 PR 作者”裁决,解析层放行并输出空 reviewed_sha
assert parse("/eval drive") == ("drive", "", "")
def test_similar_command_prefix_rejected():
with pytest.raises(ValueError, match="/eval 命令开头"):
parse(f"/evaluate drive sha={SHA}")
def test_illegal_product_name_rejected():
with pytest.raises(ValueError, match="产品名非法"):
parse(f"/eval drive;rm sha={SHA}")
def test_unknown_extra_token_rejected():
with pytest.raises(ValueError, match="未知参数"):
parse(f"/eval drive sha={SHA} --force")
def test_illegal_cases_ref_rejected():
with pytest.raises(ValueError, match="cases 引用非法"):
parse(f"/eval drive sha={SHA} cases=$(whoami)")
def test_structurally_invalid_cases_refs_rejected():
# git check-ref-format 结构规则:字符合法但结构非法的引用必须拒绝
invalid_refs = [
"..", # 以点开头且含连续点号
"/main", # 首部斜杠
"feature//x", # 连续斜杠
"feature/", # 尾部斜杠
"release.", # 尾部点号
"a..b", # 任意位置连续点号
".hidden", # 分量以点开头
"a/.hidden", # 非首分量以点开头
"a.lock", # 分量以 .lock 结尾
"a/b.lock", # 非首分量以 .lock 结尾
"-flag", # 以 - 开头,防 git fetch 选项注入
]
for ref in invalid_refs:
with pytest.raises(ValueError, match="cases 引用非法"):
parse(f"/eval drive sha={SHA} cases={ref}")
def test_structurally_valid_cases_refs_accepted():
valid_refs = ["main", "feat/drive-latest", "release/v1.0", "a.b/c-d_e", SHA]
for ref in valid_refs:
assert parse(f"/eval drive sha={SHA} cases={ref}") == ("drive", ref, SHA)
def test_short_reviewed_sha_rejected():
with pytest.raises(ValueError, match="审核 SHA 非法"):
parse("/eval drive sha=0123456")
def test_empty_comment_rejected():
with pytest.raises(ValueError, match="评论为空"):
parse(" \n ")
+127
View File
@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""校验 `/eval` 派发的仓库权限与已审核 PR head。"""
import json
import os
import re
import sys
TRUSTED_PERMISSIONS = frozenset({"write", "maintain", "admin"})
FULL_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$")
def _read_api_response():
try:
value = json.load(sys.stdin)
except (json.JSONDecodeError, OSError) as exc:
raise ValueError(f"GitHub API response is not valid JSON: {exc}") from exc
if not isinstance(value, dict):
raise ValueError("GitHub API response must be a JSON object")
return value
def load_allowlist(path: str) -> frozenset:
"""读取自助触发允许名单(默认分支文件;# 注释与空行忽略;大小写不敏感)。"""
if not path:
return frozenset()
try:
with open(path, encoding="utf-8") as f:
lines = f.read().splitlines()
except FileNotFoundError:
return frozenset()
entries = set()
for line in lines:
entry = line.split("#", 1)[0].strip()
if entry:
entries.add(entry.lower())
return frozenset(entries)
def authorize_dispatch(response, commenter: str, pr_author: str, allowlist: frozenset):
"""两级授权:仓库写权限可派发任意 PR;名单内用户仅可派发自己创建的 PR。"""
permission = response.get("permission")
if permission in TRUSTED_PERMISSIONS:
return f"maintainer:{permission}"
commenter_key = (commenter or "").strip().lower()
author_key = (pr_author or "").strip().lower()
if commenter_key and commenter_key in allowlist:
if commenter_key == author_key:
return "allowlisted-author"
raise PermissionError(
f"{commenter or 'commenter'} is allowlisted for self-service only "
"and may not dispatch other authors' PRs"
)
raise PermissionError(
f"{commenter or 'commenter'} does not have write, maintain, or admin permission "
"and is not an allowlisted PR author"
)
def require_reviewed_current_head(response, expected_pr_number: str, reviewed_sha: str, commenter: str):
try:
expected_number = int(expected_pr_number)
except (TypeError, ValueError) as exc:
raise ValueError("expected PR number is invalid") from exc
if response.get("number") != expected_number:
raise ValueError("GitHub API response does not match the requested PR")
if response.get("state") != "open":
raise ValueError("PR is not open")
head = response.get("head")
current_sha = head.get("sha") if isinstance(head, dict) else ""
if not FULL_SHA_RE.fullmatch(current_sha or ""):
raise ValueError("GitHub API response does not contain a valid PR head SHA")
if reviewed_sha:
if not FULL_SHA_RE.fullmatch(reviewed_sha):
raise ValueError("reviewed SHA must be a full 40-character commit SHA")
if current_sha.lower() != reviewed_sha.lower():
raise ValueError(
f"PR head changed after review: reviewed {reviewed_sha.lower()}, "
f"current {current_sha.lower()}"
)
return reviewed_sha.lower()
# sha 省略:仅限评论者本人创建的 PR(自助模式)——作者对自己分支的
# tip 背书不存在第三方偷换窗口;钉住派发时刻的当前 head,内网侧
# FETCH_HEAD 校验继续兜底派发→执行窗口。评测他人 PR 必须显式带 sha=。
author = ((response.get("user") or {}).get("login") or "").strip().lower()
commenter_key = (commenter or "").strip().lower()
if not author or commenter_key != author:
raise ValueError(
"dispatching another author's PR requires an explicit reviewed sha=<full-head-sha>"
)
return current_sha.lower()
def main() -> int:
mode = sys.argv[1] if len(sys.argv) == 2 else ""
try:
response = _read_api_response()
if mode == "permission":
role = authorize_dispatch(
response,
os.environ.get("COMMENTER", ""),
os.environ.get("PR_AUTHOR", ""),
load_allowlist(os.environ.get("EVAL_ALLOWLIST_PATH", "")),
)
print(f"authorized={role}")
return 0
if mode == "head":
head_sha = require_reviewed_current_head(
response,
os.environ.get("EXPECTED_PR_NUMBER", ""),
os.environ.get("REVIEWED_SHA", ""),
os.environ.get("COMMENTER", ""),
)
print(f"head_sha={head_sha}")
return 0
raise ValueError(f"unknown guard mode: {mode}")
except (PermissionError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
+127 -49
View File
@@ -5,7 +5,8 @@
- **二级子命令(必选其一)**:`event`(日程)、`attendee`(参会人)、`room`(会议室)、`busy`(闲忙)、`attachment`(日程附件)、`book`(我能看哪些日历本)、`acl`(我的日历共享给了谁)。`dws calendar` 后**必须**紧跟上述之一;**禁止**只执行 `dws calendar`(无子命令)。
- **个人日程 / 给自己留时间块 / 专注时段**:统一走 **`dws calendar event create`**。当前**没有**单独的 `personal schedule create` / `calendar create` 命令。
- **查日程列表**:`dws calendar event list --start "<ISO-8601>" --end "<ISO-8601>" --format json`,或优先使用脚本 `python scripts/calendar_today_agenda.py [today|tomorrow|week]`(见文末「自动化脚本」)。
- **查用户日历本列表**:`dws calendar book list`(返回字段为 `calendarId`,主日历 `calendarId == "primary"` 等;注意无 `id` 字段,按 `.id` 提取会取空)。**重要**可以查询他人共享给自己的日历本,根据日历本id可以进一步查询对方的日程信息。
- **查循环日程实例**:`dws calendar event instances --id <EVENT_ID> --start "<ISO-8601>" --end "<ISO-8601>" --format json`,用于按时间范围展开重复日程的每一个实例。**注意:此接口只能查询重复性日程;普通非循环日程将查不到任何实例信息。**
- **查用户日历本列表**:`dws calendar book list`(返回主日历 `id == "primary"` 等)。**重要**可以查询他人共享给自己的日历本,根据日历本id可以进一步查询对方的日程信息。
- **CLI 不存在**独立的 `dws calendar list`;若误跑无子命令的 `dws calendar`,会打印整段 Usage,**切勿**将该段 help 当作工具结果再次塞进对话(会急剧增加 token 与首字延迟)。
- **必须**遵循指令说明进行调用。**绝对禁止**使用虚构指令,使用虚构参数。
@@ -13,7 +14,6 @@
1. **禁止**执行 `dws calendar` 且不带二级子命令(会刷出大量帮助文本)。合法二级子命令:`event` / `attendee` / `room` / `busy` / `attachment` / `book` / `acl`。
2. **禁止**使用不存在的子命令试探(如臆造 `dws calendar list`);需要日程列表时一律使用 **`dws calendar event list`**(带 `--start` / `--end`,见下文「查询日程列表」示例);需要日历本列表时使用 **`dws calendar book list`**。
3. **禁止**将完整 `--help`/Usage 输出作为「观察」重复提交给模型;若误触,应直接改用本节黄金路径中的合法命令并重试。
4. **禁止**继续使用已弃用的 `--max-results` 参数 —— 它们仍可被解析但**会被丢弃**,不再透传到 MCP,模型生成命令时也不要再带。
5. **禁止**为已有日程重新创建日程来预订会议室。若日程已存在(同一会话中刚创建、或用户明确指向某日程),必须使用 `room add --event <已有EVENT_ID> --rooms <ROOM_ID>` 追加会议室,**绝不能**再调一次 `event create --rooms`(会创建重复日程)。
6. **禁止**用 `--location` 替代会议室预订。`--location` 是纯文本地点备注字段,填入会议室名称**不会**完成任何预订或占用。预订会议室必须通过 `room add --rooms <roomId>` 或 `event create --rooms <roomId>`,roomId 来自 `room search` 返回;`--location` 与 `--rooms` 是两个独立字段,用途完全不同。
7. **禁止**只传 `--recurrence-*` 部分 flag **并不是彼此独立的参数**:只传其中一项(比如只改 `--recurrence-count`、只设 `--recurrence-type`)会让服务端收到不完整的 recurrence 结构,CLI 现已前置校验并直接拒绝这类调用。**修改已有周期日程的任何一个循环字段时,都必须重新提供完整的 pattern+range 字段集合**——必要时先 `event get` 读取现有 `recurrence`,再在命令中整体重传。
@@ -26,7 +26,7 @@
日程实例(event instance):日程的具体时间实例,可以通过event list指令查询时间段内的所有实例。1个普通日程和对应1个Instance,而1个重复性日程(SeriesMaster)对应N个Instance(同属一个日程序列)。
- 同一个日程序列具有相同的iCalUid,并且重复性日程,其eventId和iCalUid的值相同。因此可以通过重复性日程实例的iCalUid得到重复性日程(SeriesMaster)的eventId
重复规则(recurrence rule):定义重复性日程的重复规则。
参会人(attendee):日程的参与者。常用通讯录工具查询userId,dws contact user search --keyword "姓名"。
参会人(attendee):日程的参与者。按姓名统一使用 `dws aisearch person --keyword "姓名" --dimension name --format json` 查询 userId。
响应状态(response):参会人对日程的回应,包括:未响应、接受、待定、拒绝。
忙闲时间(busy):查询用户在指定时间段的忙闲状态,查询会议室在指定时间段的预定状态,用于会议时间协调。
会议室(room):room是 会议室 ,room可视为日程的资源类参会人,需要加入日程完成预订。注意和location区分,location只是地点,和room不同。
@@ -42,6 +42,8 @@
dws calendar event [create|update|get|delete|respond] [flags]
# 按时间范围批量查询
dws calendar event list [flags]
# 查询循环日程的实例列表(按时间范围展开重复日程)
dws calendar event instances [flags]
# 对于非明确时间或一段时间范围的约会场景,可基于所有参会人的忙闲状态,推荐多个可用的时间块方案
dws calendar event suggest [flags]
```
@@ -64,7 +66,6 @@ dws calendar room add [flags]
dws calendar room delete [flags]
```
> room是会议室,用于线下开会场景。
> **组织限制**:部分企业使用自建会议室系统,未接入钉钉会议室能力。此时 `room search` / `room list-groups` 会返回业务错误 `400056`("所选组织不支持预定钉钉会议室")。这是组织级配置限制,不是命令用法问题;应直接告知用户该组织需在钉钉客户端手动预订会议室,不要重试或换参。
### busy 相关三级子命令
```
@@ -109,17 +110,21 @@ Usage:
dws calendar event list [flags]
Example:
dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T18:00:00+08:00"
dws calendar event list --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00" --limit 50
dws calendar event list --calendar-id primary
dws calendar event list --cursor "<nextCursor从上一次查询结果获取>"
Flags:
--calendar-id string 日历 ID (默认 primary 主日历,仅在查询其他日历本时填写;通过 `book list` 获取)
--cursor string 分页游标 (从上一次返回的 nextCursor 获取,首次查询无需传入)
--end string 结束时间 ISO-8601 (例如 2026-03-10T18:00:00+08:00)
--limit int 每页返回条数 (默认 100,最大 100)
--start string 开始时间 ISO-8601 (例如 2026-03-10T14:00:00+08:00)
```
> `--max-results` 已弃用:MCP 不再支持,CLI 仍接受但参数会被丢弃,结果固定最多返回 100 条。
**默认行为**:不传 `--start` / `--end` 时,默认返回今天的日程(00:00:00 ~ 23:59:59)。
**权限**:查询共享日历下的日程时,至少要有reader权限。
**分页**:单次最多返回 `--limit` 指定的条数(默认/最大 100);当结果超过 limit 时,返回体包含 `nextCursor` 字段。首次查询无需传 `--cursor`,仅在翻页时将上一次返回的 `nextCursor` 作为 `--cursor` 传入。
### 获取日程详情
```
@@ -133,6 +138,28 @@ Flags:
--calendar-id string 日历 ID (默认 primary 主日历)
```
### 查询循环日程实例
```
Usage:
dws calendar event instances [flags]
Example:
dws calendar event instances --id <EVENT_ID>
dws calendar event instances --id <EVENT_ID> --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00"
dws calendar event instances --id <EVENT_ID> --limit 50
dws calendar event instances --id <EVENT_ID> --cursor "<nextCursor>"
Flags:
--id string 日程 ID (必填,重复性日程 SeriesMaster 的 eventId)
--calendar-id string 日历 ID (默认 primary 主日历,仅在查询其他日历本时填写;通过 `book list` 获取)
--start string 开始时间 ISO-8601 (例如 2026-03-10T00:00:00+08:00,不传则默认今天 00:00:00)
--end string 结束时间 ISO-8601 (例如 2026-03-31T23:59:59+08:00,不传则默认今天 23:59:59)
--limit int 每页返回条数 (默认 100,最大 100)
--cursor string 分页游标 (从上一次返回的 nextCursor 获取,首次查询无需传入)
```
> **说明**:用于按时间范围展开重复日程(SeriesMaster)的每一个实例。**此接口只能查询重复性日程;若传入普通非循环日程,将查不到任何实例信息。**`--id` 必须是重复性日程的 eventId,可通过 `event list` 获取。
> **默认行为**:不传 `--start` / `--end` 时,默认返回今天的实例(00:00:00 ~ 23:59:59)。
> **分页**:单次最多返回 `--limit` 指定的条数(默认/最大 100);当结果超过 limit 时,返回体包含 `nextCursor` 字段。
### 创建日程
```
Usage:
@@ -149,10 +176,17 @@ Example:
dws calendar event create --title "每日站会" \
--start "2026-03-10T09:00:00+08:00" --end "2026-03-10T09:30:00+08:00" \
--recurrence-type daily --recurrence-interval 1 --recurrence-range-type numbered --recurrence-count 10
dws calendar event create --title "团队周会" \
--start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" \
--calendar-id <SHARED_CALENDAR_ID> # 在指定日历本下创建日程
dws calendar event create --title "重要会议" \
--start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" \
--remind-minutes 5,10 # 开始前5分钟和10分钟各提醒一次
Flags:
--title string 日程标题 (必填,最大2048字符)
--start string 开始时间 ISO-8601 (必填,例如 2026-03-10T14:00:00+08:00)
--end string 结束时间 ISO-8601 (必填,例如 2026-03-10T15:00:00+08:00)
--calendar-id string 日历 ID (可选,默认 primary 主日历;仅在共享/订阅日历本下创建时填写,通过 `book list` 获取)
--timezone string 时区 IANA 格式 (例如 Asia/Shanghai,默认 Asia/Shanghai)
--desc string 日程描述 (最大5000字符)
--attendees string 参会人 userId 列表,逗号分隔 (最多500人) 日程组织人自动放入参会人列表,无需传入userId
@@ -172,6 +206,7 @@ Flags:
--rich-text-desc string html格式的富文本类型日程描述,用于复杂内容的展示
--location string 地点信息(纯文本备注,如‘3号楼A区’;**不等于**预订会议室)
--free-busy string 此日程的忙碌状态,默认值为busy。busy - 在忙闲视图中,此日程时间段为忙碌; free - 此日程不占用忙闲
--remind-minutes string 日程开始前提醒,逗号分隔分钟数 (可选,例如 5,10,15 表示开始前5/10/15分钟提醒;不传则默认15分钟提醒)
```
> **说明**:个人日程也走 `event create`。如果只是给自己安排时间,不传 `--attendees` / `--open-dingtalk-ids` 即可。
@@ -185,8 +220,10 @@ Example:
dws calendar event update --id <EVENT_ID> --desc "新描述" --timezone Asia/Tokyo
dws calendar event update --id <EVENT_ID> --recurrence-type daily --recurrence-interval 1 \
--recurrence-range-type numbered --recurrence-count 5
dws calendar event update --id <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID> --title "新标题" # 修改其他日历本下的日程
Flags:
--id string 日程 ID (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取)
--title string 新标题
--start string 新开始时间 ISO-8601
--end string 新结束时间 ISO-8601
@@ -218,8 +255,10 @@ Usage:
dws calendar event delete [flags]
Example:
dws calendar event delete --id <EVENT_ID> --yes
dws calendar event delete --id <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID> --yes # 删除其他日历本下的日程
Flags:
--id string 日程 ID (必填)
--id string 日程 ID (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取)
```
### 查看参会人
@@ -228,8 +267,10 @@ Usage:
dws calendar attendee list [flags]
Example:
dws calendar attendee list --event <EVENT_ID>
dws calendar attendee list --event <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID> # 查看其他日历本下日程的参会人
Flags:
--event string 日程 ID (必填)
--event string 日程 ID (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取, 注意:订阅日历下的日程无参会人,因此不可查看)
```
### 添加参会人
@@ -239,10 +280,12 @@ Usage:
Example:
dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID_1>,<USER_ID_2>
dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID> --optional
dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID> --calendar-id <SHARED_CALENDAR_ID> # 给其他日历本下的日程添加参会人
Flags:
--event string 日程 ID (必填)
--attendees string 参会人 userId 列表,逗号分隔 (必填,最多500人)
--optional 参会人可选 (默认必选参会人)
--event string 日程 ID (必填)
--attendees string 参会人 userId 列表,逗号分隔 (必填,最多500人)
--optional 参会人可选 (默认必选参会人)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取,注意:订阅日历下的日程无参会人,因此不可添加)
```
### 移除参会人
@@ -254,36 +297,54 @@ Usage:
dws calendar attendee delete [flags]
Example:
dws calendar attendee delete --event <EVENT_ID> --attendees <USER_ID> --yes
dws calendar attendee delete --event <EVENT_ID> --attendees <USER_ID> --calendar-id <SHARED_CALENDAR_ID> --yes # 移除其他日历本下日程的参会人
Flags:
--event string 日程 ID (必填)
--attendees string 参会人 userId 列表,逗号分隔 (必填)
--event string 日程 ID (必填)
--attendees string 参会人 userId 列表,逗号分隔 (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取,注意:订阅日历下的日程无参会人,因此不可移除)
```
### 搜索会议室
> 此指令可用于搜索当前用户可用的会议室。**注意**,大部分会议室**仅在工作时间**可用,如果检索时间不在工作时间可能查不到任何结果。
> 此指令支持两种模式:**按名称搜索**(不传 --start/--end)和**按时间段搜索可用会议室**(传 --start/--end 或不传任何参数)。
> 此指令搜索到的会议室结果中,有两个值需要注意:
- customApprovalProcess: true - 表示该会议室设置了自定义审批流程,只能通过客户端完成预订。
- supportRecurring: true - 表示该会议室支持循环预定;false - 表示不支持循环预定,直接加入到循环日程会失败。
> - customApprovalProcess: true - 表示该会议室设置了自定义审批流程,只能通过客户端完成预订。
> - supportRecurring: true - 表示该会议室支持循环预定;false - 表示不支持循环预定,直接加入到循环日程会失败。
**模式路由规则**:
- **按名称搜索**:仅传 `--room-name`,不传 `--start`/`--end` → 返回所有匹配名称的会议室,**不检查可用性**。适用于「找到某个会议室」的场景。
- **按时间段搜索可用会议室**:传 `--start`/`--end`,或不传任何参数 → 返回指定时间段内**可用**的会议室。不传时间时默认当前时间起 1 小时。
```
Usage:
dws calendar room search [flags]
Example:
# 按名称搜索(不检查可用性,返回所有匹配的会议室)
dws calendar room search --room-name 永澄亭 # 注意:用户即使说「永澄亭会议室」,也应仅传「永澄亭」
# 按时间段搜索可用会议室
dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00"
dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --group-id <GROUP_ID>
dws calendar room search --room-name 永澄亭 # 注意:用户即使说「永澄亭会议室」,也应仅传「永澄亭」
dws calendar room search # 不传 --start/--end 时默认当前时间起 1 小时
# 名称 + 时间段:搜索指定名称的可用会议室
dws calendar room search --room-name 永澄亭 --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00"
# 分页(仅按时间段搜索时有效)
dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --limit 20 --page 0
Flags:
--start string 开始时间 ISO-8601 (可选,不传则默认当前时间+1分钟缓冲)
--end string 结束时间 ISO-8601 (可选,不传则默认当前时间+1 小时)
--group-id string 会议室分组ID(可选,留空查根目录;超100条时需按分组查询)
--room-name string 按会议室名称过滤(可选,服务端模糊匹配;传入前必须由调用方精简,只保留核心专名)
--group-id string 会议室分组ID(可选,留空查根目录;超100条时需按分组查询;仅按时间段搜索时有效)
--room-name string 会议室名称(按名称搜索时必填;按时间段搜索时可选,用于过滤)
--limit string 页大小 (可选,不填默认 100,超过 100 按 100 处理;仅按时间段搜索时有效)
--page string 分页起始位置 (可选,不填默认 0;仅按时间段搜索时有效)
```
> **时间约束(API 限制)**:`start` 必须是未来的时间(服务端校验:start can not less current time)。
> - 若传入的 `--start` 早于当前时间,CLI 会自动修正为 `now + 1min`,调用方无需额外处理。
> - 若传入的 `--end` 早于当前时间,CLI 直接报错——无法检索已过去的时间段。
> - **最佳实践**:调用方在组装时间参数时应确保 start/end 都是未来时间;若不确定,可省略 `--start`/`--end` 让 CLI 使用默认值(当前时间起 1 小时)。
> - **注意**:时间约束仅在「按时间段搜索」模式下生效。按名称搜索(仅传 `--room-name`)不受时间约束。
**名称过滤使用规范**:`--room-name` 适用于用户说「预定永澄亭」「约西湖厅」这类按名找会议室的场景。
- **服务端是模糊匹配,但匹配词越精简命中率越高**,关键疗法:**调用方必须在调用 CLI 前自行精简名称,CLI 不会再做任何删减**。
@@ -293,13 +354,16 @@ Flags:
- 用户:「西湖厅有空吗」 → `--room-name 西湖厅`(本身就是专名,不删即可)
- 用户:「预定贡嘎山大会议室」 → `--room-name 贡嘎山`
**优先路径**:当用户给了会议室中文名时,**优先用 `room search --room-name <核心专名> --start <开始时间> --end <结束时间>`**代替「先走 `list-groups`再遍历」的旧流程。`--room-name` 可与 `--group-id` 同时使用,表示「在指定分组内按名称过滤」。若返回空列表,再降级使用 `list-groups` 定位分组再查。
**优先路径**:
- 当用户仅给出会议室名称、未指定时间段时,用 `room search --room-name <核心专名>` 快速找到会议室(不检查可用性)。
- 当用户给出会议室名称且指定了时间段时,用 `room search --room-name <核心专名> --start <开始时间> --end <结束时间>` 查询该名称的可用会议室。
- 若返回空列表,再降级使用 `list-groups` 定位分组再查。`--room-name` 可与 `--group-id` 同时使用(仅按时间段搜索时),表示「在指定分组内按名称过滤」。
**`roomId` 与用户说的话不是一回事**:用户说的「C6-4-06-N 贡嘎山」等是**展示名/编号文案**,**绝不能**直接填进 `room add --rooms`。`--rooms` 只接受上一步 `room search`(或同类接口)返回 JSON 里的 **`rooms[].roomId`**。形态上多为**小写十六进制串**(长度以接口为准,例如 `e6b7b65b8b30fb707afcf6c3b699f028003e6834fdd7fee7`)。含**中文、空格、连字符拼接的楼层编号**、或凭空调 UUID/纯数字「试格式」——一律视为非法,必须先搜房再取返回字段。
> 如果知道roomId,想查该会议室的预订记录,直接用dws calendar busy search 指令
---
---
### 预定会议室
```
@@ -307,9 +371,11 @@ Usage:
dws calendar room add [flags]
Example:
dws calendar room add --event <EVENT_ID> --rooms <ROOM_ID>
dws calendar room add --event <EVENT_ID> --rooms <ROOM_ID> --calendar-id <SHARED_CALENDAR_ID> # 给其他日历本下的日程预定会议室
Flags:
--event string 日程 ID (必填)
--rooms string 会议室 ID 列表 (必填)
--event string 日程 ID (必填)
--rooms string 会议室 ID 列表 (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取,注意:订阅日历下的日程不可添加会议室)
```
> room是会议室,用于线下开会场景。将room加入到日程完成预订
> 重复性日程,预订会议室时,必须设置 循环结束时间(recurrence-end-date),noEnd 或者 指定循环次数 都无法完成预定。
@@ -324,9 +390,11 @@ Usage:
dws calendar room delete [flags]
Example:
dws calendar room delete --event <EVENT_ID> --rooms <ROOM_ID> --yes
dws calendar room delete --event <EVENT_ID> --rooms <ROOM_ID> --calendar-id <SHARED_CALENDAR_ID> --yes # 移除其他日历本下日程的会议室
Flags:
--event string 日程 ID (必填)
--rooms string 会议室 ID 列表 (必填)
--event string 日程 ID (必填)
--rooms string 会议室 ID 列表 (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取。注意:订阅日历下的日程不可添加会议室)
```
### 会议室分组列表
@@ -347,22 +415,24 @@ Usage:
dws calendar attachment add [flags]
Example:
dws calendar attachment add --event <EVENT_ID> --files <FILE_ID>:report.pdf,<FILE_ID2>:slides.pptx
dws calendar attachment add --event <EVENT_ID> --files <FILE_ID>:report.pdf --calendar-id <SHARED_CALENDAR_ID> # 给其他日历本下的日程添加附件
Flags:
--event string 日程 ID (必填)
--files string 附件列表,格式 <fileId>:<name>,多项逗号分隔 (必填)
--event string 日程 ID (必填)
--files string 附件列表,格式 <fileId>:<name>,多项逗号分隔 (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取,注意:订阅日历下的日程不可添加附件)
```
> 上传文件得到 `fileId` 需配合钉盘相关流程;本命令只负责把已上传的文件挂载到日程上。
### 查询用户日历列表
### 查询我能看的所有日历本
```
Usage:
dws calendar book list [flags]
Example:
dws calendar book list
```
> 通过此接口可查询当前用户的日历列表,包含 主日历本、他人共享的日历、订阅的公共/团队日历。查的是"我能看哪些日历本";注意区分:`acl list` 是查"我的日历共享**给了谁**",方向相反。
> 共享日历本中有来自 xxx 的,且权限大于reader,那么通过 `event list --calendar-id <xxx的日历本id> `可查到xxx完整的日程安排
> 查询"我能看哪些日历本",包含:我的主日历、他人共享**给我**的日历、订阅的公共/团队日历。注意区分:`acl list` 是查"我的日历共享**给了谁**",方向相反。
> 共享日历本中有来自 xxx 的,且权限大于reader,那么通过 `event list --calendar-id <xxx的日历本id> `可查到xxx完整的日程安排。
> 主日历 `id` 固定为 `primary`,绝大多数日程操作都默认走主日历,只有当用户明确要求查/写其他日历本时才需要带 `--calendar-id`。
### 查询指定日历本
@@ -495,9 +565,11 @@ Example:
dws calendar event respond --id <EVENT_ID> --status accepted
dws calendar event respond --id <EVENT_ID> --status declined
dws calendar event respond --id <EVENT_ID> --status tentative
dws calendar event respond --id <EVENT_ID> --status accepted --calendar-id <SHARED_CALENDAR_ID> # 响应其他日历本下的日程
Flags:
--id string 日程 ID (必填)
--status string 响应状态: needsAction(未操作)|accepted(接受)|declined(拒绝)|tentative(暂定) (必填)
--id string 日程 ID (必填)
--status string 响应状态: needsAction(未操作)|accepted(接受)|declined(拒绝)|tentative(暂定) (必填)
--calendar-id string 日历 ID (可选,默认 primary 主日历;指定其他日历本时填写,可通过 `book list` 获取。注意:订阅日历下的日程无参会人,因此不可响应)
```
> **说明**:作为日程参会人,设置自己的响应状态(接受、拒绝、暂定)。`--status` 可选值:`needsAction`(未操作,默认值)、`accepted`(接受)、`declined`(拒绝)、`tentative`(暂定)。
@@ -507,11 +579,12 @@ Flags:
用户说"日程/会议/约会/日历":
- 查看 → `event list`
- 详情 → `event get`
- 创建/约/给自己留时间块/个人日程 → `event create`(带参会人时加 `--attendees`,循环日程加 `--recurrence-*`)
- 创建/约/给自己留时间块/个人日程 → `event create`(带参会人时加 `--attendees`,循环日程加 `--recurrence-*`,自定义提醒加 `--remind-minutes`)
- 修改/改时间/改描述 → `event update`(支持修改标题、时间、描述、时区、循环规则)
- 取消/删除 → `event delete`
- 推荐时间/什么时候有空/协调时间 → `event suggest`
- 接受/拒绝/暂定日程 → `event respond`
- 查询循环日程/重复日程的每次实例/展开循环日程 → `event instances`
用户说"参会人/与会者":
- 查看 → `attendee list`
@@ -519,13 +592,14 @@ Flags:
- 移除 → `attendee delete --attendees <USER_ID>`
用户说"会议室/订会议室":
- 哪个空闲 → `room search`
- 按名找会议室(如「永澄亭」「永澄亭会议室」「约西湖厅」)→ 先在模型层精简名称(剔除「会议室」等通用后缀),再用 `room search --room-name <核心专名>`
- 哪个空闲 → `room search`(默认查当前时间起 1 小时内可用会议室)
- 按名找会议室(如「永澄亭」「永澄亭会议室」「约西湖厅」,未提时间段)→ 先在模型层精简名称(剔除「会议室」等通用后缀),再用 `room search --room-name <核心专名>`(按名称搜索,不检查可用性)
- 按名找可用会议室(如「永澄亭下午 2 点有空吗」)→ `room search --room-name <核心专名> --start <开始时间> --end <结束时间>`(按名称+时间段搜索可用会议室)
- 预订
- 给已有日程订会议室 → `room add --event <已有EVENT_ID> --rooms <ROOM_ID>`
- 创建新日程并订会议室 → `event create --rooms`(仅当日程尚不存在时)
- 取消预定 → `room delete`
- 分组 → `room list-groups`,取 groupId 后 `room search --group-id`(可再叠加 `--room-name` 在分组内过滤)
- 分组 → `room list-groups`,取 groupId 后 `room search --group-id`(需配合 `--start`/`--end` 按时间段搜索;可再叠加 `--room-name` 在分组内过滤)
用户说"有空吗/忙不忙/闲忙":
- 查询用户闲忙 → `busy search --users <USER_ID>`
@@ -580,11 +654,11 @@ dws calendar event create --title "Q1 复盘会" \
**方式二:先创建日程,再单独添加参会人 / 会议室**
```bash
# Step 1: 创建日程 — 提取 eventId
# Step 1: 创建日程 — 从 result.id 提取日程 ID
dws calendar event create --title "Q1 复盘会" \
--start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --format json
# Step 2: 添加参会人(必须用 Step 1 返回的 eventId)
# Step 2: 添加参会人(必须用 Step 1 返回的 result.id)
dws calendar attendee add --event <EVENT_ID> --attendees userId1,userId2 --format json
# Step 3: 搜索空闲会议室
@@ -612,13 +686,14 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
| 操作 | 从返回中提取 | 用于 |
|------|-------------|------|
| `event create` | `eventId` | attendee/room/attachment 操作的 --event |
| `event list` | `events[].eventId` | event get/update/delete/respond 的 --id |
| `event create` | `result.id` | attendee/room/attachment 操作的 --event |
| `event list` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id;下一页 --cursor |
| `event suggest` | 推荐的时间段 | event create 的 --start/--end |
| `event respond` | 响应结果 | — |
| `event instances` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id;下一页 --cursor |
| `room search` | `rooms[].roomId` | room add 的 --rooms 或 event create 的 --rooms |
| `room list-groups` | `groups[].groupId` | room search 的 --group-id |
| `book list` | `calendarId`(如 `primary`) | event list/get 的 --calendar-id, book get/update 的 --id |
| `book list` | `id`(如 `primary`) | event list/get 的 --calendar-id, book get/update 的 --id |
| `book get` | 日历详细信息 | — |
| `book search` | 匹配的日历列表 | book get/update 的 --id |
| `acl list` | `aclId` | acl delete 的 --acl-id |
@@ -638,20 +713,23 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
- `room list-groups` 支持 `--limit` / `--page` 分页(schema 类型为字符串)
- **`event create --rooms` / `room add --rooms` 的唯一合法来源**:最近一次(同一会话、同一时段窗口)`room search` 返回体中的 `roomId`;禁止把用户自然语言会议室名当 `roomId` 传入(否则会 `roomId invalid` 等错误)
- **搜房无结果**:在符合早停/用户限定范围内,`room search`(含按分组逐组查)全部返回空或无空闲 → 应**直接向用户报错/说明失败**并结束订房;**禁止**假设 roomId、禁止无合法 `roomId` 时调用 `room add` / `event create --rooms` 试探、禁止用 `event get` 等绕路推断 roomId
- **评测 / 自动化断言**:凡涉及 `room add` / `event create --rooms` 的流程,`--rooms` 只能填上游 `room search`(或等价接口)返回 JSON 中的 **`rooms[].roomId`**;不得以会议室展示名、楼层文案或用户口语当作 `roomId`
- **自动化校验**:凡涉及 `room add` / `event create --rooms` 的流程,`--rooms` 只能填上游 `room search`(或等价接口)返回 JSON 中的 **`rooms[].roomId`**;不得以会议室展示名、楼层文案或用户口语当作 `roomId`
- **附件**:`attachment add` 仅负责挂载,**不上传**文件;fileId 必须先通过钉盘流程取得;`--files` 多附件用 `<fileId>:<name>` 元素逗号分隔
- **日历本**:`book list` 返回的 `calendarId` 字段才是合法日历 id(无 `id` 字段);如无明确说明,`event list` / `event get` 都不要带 `--calendar-id`,让接口默认走 primary 主日历
- **已弃用入参**:`--max-results`(event list)在新 MCP schema 中被移除;CLI 仍接受但**不会**透传到 MCP;模型生成命令时**禁止**继续使用
- **日历本**:`book list` 返回的 `id` 才是合法 `calendarId`;如无明确说明,`event list` / `event get` 都不要带 `--calendar-id`,让接口默认走 primary 主日历
- **分页查询**:`event list` / `event instances` 均支持 `--limit`(控制每页条数,默认/最大 100)和 `--cursor`(翻页游标);**首次查询无需传 `--cursor`**,仅当返回体中包含 `nextCursor` 时,将其作为 `--cursor` 传入可获取下一页
- **循环日程实例**:`event instances` 用于按时间范围展开重复日程(SeriesMaster)的每一个实例;**普通非循环日程调用该命令将查不到任何实例信息**
- **日程提醒**:`event create` 支持 `--remind-minutes` 设置开始前提醒,逗号分隔多个分钟数(如 `--remind-minutes 5,10,15`),不传则默认15分钟提醒
- **会议室分页**:`room search` 支持 `--limit`(每页条数,默认100,最大100)和 `--page`(分页起始位置,默认0),与 `room list-groups` 分页风格一致
## 自动化脚本
| 脚本 | 场景 | 用法 |
|------|------|------|
| [calendar_today_agenda.py](../../scripts/calendar_today_agenda.py) | 查看今天/明天/本周日程安排 | `python calendar_today_agenda.py today` |
| [calendar_schedule_meeting.py](../../scripts/calendar_schedule_meeting.py) | 一键创建日程+添加参会人+预定会议室;搜房失败时输出明确原因并返回非零退出码 | `python calendar_schedule_meeting.py --title "复盘会" --start "2026-03-15T14:00" --end "2026-03-15T15:00" --users userId1 --book-room` |
| [calendar_free_slot_finder.py](../../scripts/calendar_free_slot_finder.py) | 查询多人共同空闲时段 | `python calendar_free_slot_finder.py --users userId1,userId2 --date 2026-03-15` |
| [calendar_today_agenda.py](../scripts/calendar_today_agenda.py) | 查看今天/明天/本周日程安排 | `python calendar_today_agenda.py today` |
| [calendar_schedule_meeting.py](../scripts/calendar_schedule_meeting.py) | 一键创建日程+添加参会人+预定会议室;搜房失败时输出明确原因并返回非零退出码 | `python calendar_schedule_meeting.py --title "复盘会" --start "2026-03-15T14:00" --end "2026-03-15T15:00" --users userId1 --book-room` |
| [calendar_free_slot_finder.py](../scripts/calendar_free_slot_finder.py) | 查询多人共同空闲时段 | `python calendar_free_slot_finder.py --users userId1,userId2 --date 2026-03-15` |
## 相关产品
- conference 视频会议 — 当前 CLI 不支持发起/预约/邀请入会/会中控制;请在钉钉客户端操作
- [contact](./contact.md) — 搜索同事 userId,用于 attendee add --attendees
- conference(视频会议预约) — 仅视频会议预约(返回入会链接),不含参会人/会议室管理
- [contact](../../dingtalk-contact/references/contact.md) — 搜索同事 userId,用于 attendee add --attendees
@@ -5,6 +5,7 @@
- **二级子命令(必选其一)**:`event`(日程)、`attendee`(参会人)、`room`(会议室)、`busy`(闲忙)、`attachment`(日程附件)、`book`(我能看哪些日历本)、`acl`(我的日历共享给了谁)。`dws calendar` 后**必须**紧跟上述之一;**禁止**只执行 `dws calendar`(无子命令)。
- **个人日程 / 给自己留时间块 / 专注时段**:统一走 **`dws calendar event create`**。当前**没有**单独的 `personal schedule create` / `calendar create` 命令。
- **查日程列表**:`dws calendar event list --start "<ISO-8601>" --end "<ISO-8601>" --format json`,或优先使用脚本 `python scripts/calendar_today_agenda.py [today|tomorrow|week]`(见文末「自动化脚本」)。
- **查循环日程实例**:`dws calendar event instances --id <EVENT_ID> --start "<ISO-8601>" --end "<ISO-8601>" --format json`,用于按时间范围展开重复日程的每一个实例。**注意:此接口只能查询重复性日程;普通非循环日程将查不到任何实例信息。**
- **查用户日历本列表**:`dws calendar book list`(返回主日历 `id == "primary"` 等)。**重要**可以查询他人共享给自己的日历本,根据日历本id可以进一步查询对方的日程信息。
- **CLI 不存在**独立的 `dws calendar list`;若误跑无子命令的 `dws calendar`,会打印整段 Usage,**切勿**将该段 help 当作工具结果再次塞进对话(会急剧增加 token 与首字延迟)。
- **必须**遵循指令说明进行调用。**绝对禁止**使用虚构指令,使用虚构参数。
@@ -41,6 +42,8 @@
dws calendar event [create|update|get|delete|respond] [flags]
# 按时间范围批量查询
dws calendar event list [flags]
# 查询循环日程的实例列表(按时间范围展开重复日程)
dws calendar event instances [flags]
# 对于非明确时间或一段时间范围的约会场景,可基于所有参会人的忙闲状态,推荐多个可用的时间块方案
dws calendar event suggest [flags]
```
@@ -135,6 +138,28 @@ Flags:
--calendar-id string 日历 ID (默认 primary 主日历)
```
### 查询循环日程实例
```
Usage:
dws calendar event instances [flags]
Example:
dws calendar event instances --id <EVENT_ID>
dws calendar event instances --id <EVENT_ID> --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00"
dws calendar event instances --id <EVENT_ID> --limit 50
dws calendar event instances --id <EVENT_ID> --cursor "<nextCursor>"
Flags:
--id string 日程 ID (必填,重复性日程 SeriesMaster 的 eventId)
--calendar-id string 日历 ID (默认 primary 主日历,仅在查询其他日历本时填写;通过 `book list` 获取)
--start string 开始时间 ISO-8601 (例如 2026-03-10T00:00:00+08:00,不传则默认今天 00:00:00)
--end string 结束时间 ISO-8601 (例如 2026-03-31T23:59:59+08:00,不传则默认今天 23:59:59)
--limit int 每页返回条数 (默认 100,最大 100)
--cursor string 分页游标 (从上一次返回的 nextCursor 获取,首次查询无需传入)
```
> **说明**:用于按时间范围展开重复日程(SeriesMaster)的每一个实例。**此接口只能查询重复性日程;若传入普通非循环日程,将查不到任何实例信息。**`--id` 必须是重复性日程的 eventId,可通过 `event list` 获取。
> **默认行为**:不传 `--start` / `--end` 时,默认返回今天的实例(00:00:00 ~ 23:59:59)。
> **分页**:单次最多返回 `--limit` 指定的条数(默认/最大 100);当结果超过 limit 时,返回体包含 `nextCursor` 字段。
### 创建日程
```
Usage:
@@ -559,6 +584,7 @@ Flags:
- 取消/删除 → `event delete`
- 推荐时间/什么时候有空/协调时间 → `event suggest`
- 接受/拒绝/暂定日程 → `event respond`
- 查询循环日程/重复日程的每次实例/展开循环日程 → `event instances`
用户说"参会人/与会者":
- 查看 → `attendee list`
@@ -664,6 +690,7 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
| `event list` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id;下一页 --cursor |
| `event suggest` | 推荐的时间段 | event create 的 --start/--end |
| `event respond` | 响应结果 | — |
| `event instances` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id;下一页 --cursor |
| `room search` | `rooms[].roomId` | room add 的 --rooms 或 event create 的 --rooms |
| `room list-groups` | `groups[].groupId` | room search 的 --group-id |
| `book list` | `id`(如 `primary`) | event list/get 的 --calendar-id, book get/update 的 --id |
@@ -689,7 +716,8 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
- **自动化校验**:凡涉及 `room add` / `event create --rooms` 的流程,`--rooms` 只能填上游 `room search`(或等价接口)返回 JSON 中的 **`rooms[].roomId`**;不得以会议室展示名、楼层文案或用户口语当作 `roomId`
- **附件**:`attachment add` 仅负责挂载,**不上传**文件;fileId 必须先通过钉盘流程取得;`--files` 多附件用 `<fileId>:<name>` 元素逗号分隔
- **日历本**:`book list` 返回的 `id` 才是合法 `calendarId`;如无明确说明,`event list` / `event get` 都不要带 `--calendar-id`,让接口默认走 primary 主日历
- **分页查询**:`event list` 支持 `--limit`(控制每页条数,默认/最大 100)和 `--cursor`(翻页游标);**首次查询无需传 `--cursor`**,仅当返回体中包含 `nextCursor` 时,将其作为 `--cursor` 传入可获取下一页
- **分页查询**:`event list` / `event instances` 均支持 `--limit`(控制每页条数,默认/最大 100)和 `--cursor`(翻页游标);**首次查询无需传 `--cursor`**,仅当返回体中包含 `nextCursor` 时,将其作为 `--cursor` 传入可获取下一页
- **循环日程实例**:`event instances` 用于按时间范围展开重复日程(SeriesMaster)的每一个实例;**普通非循环日程调用该命令将查不到任何实例信息**
- **日程提醒**:`event create` 支持 `--remind-minutes` 设置开始前提醒,逗号分隔多个分钟数(如 `--remind-minutes 5,10,15`),不传则默认15分钟提醒
- **会议室分页**:`room search` 支持 `--limit`(每页条数,默认100,最大100)和 `--page`(分页起始位置,默认0),与 `room list-groups` 分页风格一致
@@ -703,5 +731,5 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
## 相关产品
- 视频会议(conference)— 当前 CLI 不支持发起/预约/邀请入会/会中控制;请在钉钉客户端操作
- conference(视频会议预约) — 仅视频会议预约(返回入会链接),不含参会人/会议室管理
- [contact](../../dingtalk-contact/references/contact.md) — 搜索同事 userId,用于 attendee add --attendees
+291
View File
@@ -0,0 +1,291 @@
// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
package scripts_test
import (
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
func TestEvalDispatchWorkflowUsesRepositoryPermissionAndReviewedSHA(t *testing.T) {
t.Parallel()
root, err := filepath.Abs(filepath.Join("..", ".."))
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
data, err := os.ReadFile(filepath.Join(root, ".github", "workflows", "eval-dispatch.yml"))
if err != nil {
t.Fatalf("read eval-dispatch workflow: %v", err)
}
workflow := string(data)
for _, want := range []string{
"/collaborators/${COMMENTER}/permission",
"eval_dispatch_guard.py permission",
"PR_AUTHOR: ${{ github.event.issue.user.login }}",
"EVAL_ALLOWLIST_PATH: .github/eval-allowlist.txt",
"REVIEWED_SHA: ${{ steps.parse.outputs.reviewed_sha }}",
"eval_dispatch_guard.py head",
} {
if !strings.Contains(workflow, want) {
t.Errorf("eval-dispatch workflow missing security contract %q", want)
}
}
if strings.Contains(workflow, "author_association") {
t.Error("eval-dispatch workflow must not authorize from author_association")
}
if _, err := os.Stat(filepath.Join(root, ".github", "eval-allowlist.txt")); err != nil {
t.Errorf("eval allowlist file missing on default branch: %v", err)
}
}
func TestEvalDispatchRejectsLowRepositoryPermissions(t *testing.T) {
t.Parallel()
for _, permission := range []string{"read", "triage", "none"} {
permission := permission
t.Run(permission, func(t *testing.T) {
t.Parallel()
output, err := runEvalDispatchGuard(t, "permission", `{"permission":"`+permission+`"}`, "COMMENTER=low-privilege-user")
if err == nil {
t.Fatalf("permission %q unexpectedly allowed; output=%s", permission, output)
}
if !strings.Contains(output, "does not have write, maintain, or admin permission") {
t.Fatalf("permission %q rejection = %q, want trusted-permission error", permission, output)
}
})
}
}
func TestEvalDispatchAllowsTrustedRepositoryPermissions(t *testing.T) {
t.Parallel()
for _, permission := range []string{"write", "maintain", "admin"} {
permission := permission
t.Run(permission, func(t *testing.T) {
t.Parallel()
output, err := runEvalDispatchGuard(t, "permission", `{"permission":"`+permission+`"}`, "COMMENTER=maintainer")
if err != nil {
t.Fatalf("permission %q rejected: %v\n%s", permission, err, output)
}
})
}
}
func TestEvalDispatchRejectsChangedPRHead(t *testing.T) {
t.Parallel()
const reviewedSHA = "1111111111111111111111111111111111111111"
const currentSHA = "2222222222222222222222222222222222222222"
pr := `{"number":934,"state":"open","head":{"sha":"` + currentSHA + `"}}`
output, err := runEvalDispatchGuard(
t,
"head",
pr,
"EXPECTED_PR_NUMBER=934",
"REVIEWED_SHA="+reviewedSHA,
)
if err == nil {
t.Fatalf("changed PR head unexpectedly allowed; output=%s", output)
}
if !strings.Contains(output, "PR head changed after review") {
t.Fatalf("changed-head rejection = %q, want explicit stale-review error", output)
}
}
func TestEvalDispatchAcceptsReviewedCurrentPRHead(t *testing.T) {
t.Parallel()
const reviewedSHA = "1111111111111111111111111111111111111111"
pr := `{"number":934,"state":"open","head":{"sha":"` + reviewedSHA + `"}}`
output, err := runEvalDispatchGuard(
t,
"head",
pr,
"EXPECTED_PR_NUMBER=934",
"REVIEWED_SHA="+reviewedSHA,
)
if err != nil {
t.Fatalf("reviewed current PR head rejected: %v\n%s", err, output)
}
if !strings.Contains(output, "head_sha="+reviewedSHA) {
t.Fatalf("head guard output = %q, want pinned head SHA", output)
}
}
func TestEvalCommentRequiresExplicitReviewedSHA(t *testing.T) {
t.Parallel()
const reviewedSHA = "1111111111111111111111111111111111111111"
output, err := runEvalCommentParser(t, "/eval drive sha="+reviewedSHA)
if err != nil {
t.Fatalf("explicit reviewed SHA rejected: %v\n%s", err, output)
}
for _, want := range []string{"products=drive", "reviewed_sha=" + reviewedSHA} {
if !strings.Contains(output, want) {
t.Errorf("parser output = %q, want %q", output, want)
}
}
// sha 省略在解析层放行(空 reviewed_sha),由 guard 按评论者是否 PR 作者裁决
output, err = runEvalCommentParser(t, "/eval drive")
if err != nil {
t.Fatalf("omitted reviewed SHA rejected at parse layer: %v\n%s", err, output)
}
if !strings.Contains(output, "reviewed_sha=\n") && !strings.HasSuffix(output, "reviewed_sha=") {
t.Fatalf("parser output = %q, want empty reviewed_sha passthrough", output)
}
}
func TestEvalDispatchAutoPinsOwnPRHeadWhenShaOmitted(t *testing.T) {
t.Parallel()
const currentSHA = "3333333333333333333333333333333333333333"
pr := `{"number":934,"state":"open","user":{"login":"Internal-Contributor"},"head":{"sha":"` + currentSHA + `"}}`
output, err := runEvalDispatchGuard(
t,
"head",
pr,
"EXPECTED_PR_NUMBER=934",
"REVIEWED_SHA=",
"COMMENTER=internal-contributor",
)
if err != nil {
t.Fatalf("own-PR auto pin rejected: %v\n%s", err, output)
}
if !strings.Contains(output, "head_sha="+currentSHA) {
t.Fatalf("guard output = %q, want auto-pinned current head", output)
}
}
func TestEvalDispatchRequiresShaForOthersPRs(t *testing.T) {
t.Parallel()
const currentSHA = "3333333333333333333333333333333333333333"
pr := `{"number":934,"state":"open","user":{"login":"someone-else"},"head":{"sha":"` + currentSHA + `"}}`
output, err := runEvalDispatchGuard(
t,
"head",
pr,
"EXPECTED_PR_NUMBER=934",
"REVIEWED_SHA=",
"COMMENTER=internal-contributor",
)
if err == nil {
t.Fatalf("other author's PR without sha unexpectedly allowed; output=%s", output)
}
if !strings.Contains(output, "requires an explicit reviewed sha") {
t.Fatalf("missing-sha rejection = %q, want explicit-sha requirement", output)
}
}
func TestEvalDispatchAllowsAllowlistedAuthorOnOwnPR(t *testing.T) {
t.Parallel()
allowlist := writeEvalAllowlist(t, "# 自助评测名单\nInternal-Contributor # 内部贡献者\n")
output, err := runEvalDispatchGuard(
t,
"permission",
`{"message":"Not Found"}`,
"COMMENTER=internal-contributor",
"PR_AUTHOR=Internal-Contributor",
"EVAL_ALLOWLIST_PATH="+allowlist,
)
if err != nil {
t.Fatalf("allowlisted author rejected on own PR: %v\n%s", err, output)
}
if !strings.Contains(output, "authorized=allowlisted-author") {
t.Fatalf("guard output = %q, want allowlisted-author authorization", output)
}
}
func TestEvalDispatchRejectsAllowlistedUserOnOthersPR(t *testing.T) {
t.Parallel()
allowlist := writeEvalAllowlist(t, "internal-contributor\n")
output, err := runEvalDispatchGuard(
t,
"permission",
`{"message":"Not Found"}`,
"COMMENTER=internal-contributor",
"PR_AUTHOR=someone-else",
"EVAL_ALLOWLIST_PATH="+allowlist,
)
if err == nil {
t.Fatalf("allowlisted user unexpectedly dispatched another author's PR; output=%s", output)
}
if !strings.Contains(output, "self-service only") {
t.Fatalf("cross-PR rejection = %q, want self-service-only error", output)
}
}
func TestEvalDispatchRejectsUnlistedCommenterWithoutWrite(t *testing.T) {
t.Parallel()
allowlist := writeEvalAllowlist(t, "# 空名单\n")
output, err := runEvalDispatchGuard(
t,
"permission",
`{"permission":"read"}`,
"COMMENTER=stranger",
"PR_AUTHOR=stranger",
"EVAL_ALLOWLIST_PATH="+allowlist,
)
if err == nil {
t.Fatalf("unlisted commenter unexpectedly allowed; output=%s", output)
}
if !strings.Contains(output, "not an allowlisted PR author") {
t.Fatalf("unlisted rejection = %q, want allowlist-aware error", output)
}
}
func writeEvalAllowlist(t *testing.T, content string) string {
t.Helper()
path := filepath.Join(t.TempDir(), "eval-allowlist.txt")
if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
t.Fatalf("write allowlist fixture: %v", err)
}
return path
}
func runEvalDispatchGuard(t *testing.T, mode, input string, env ...string) (string, error) {
t.Helper()
root, err := filepath.Abs(filepath.Join("..", ".."))
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
cmd := exec.Command("python3", filepath.Join(root, "scripts", "ci", "eval_dispatch_guard.py"), mode)
cmd.Stdin = strings.NewReader(input)
cmd.Env = append(os.Environ(), env...)
output, runErr := cmd.CombinedOutput()
return string(output), runErr
}
func runEvalCommentParser(t *testing.T, comment string) (string, error) {
t.Helper()
root, err := filepath.Abs(filepath.Join("..", ".."))
if err != nil {
t.Fatalf("resolve repository root: %v", err)
}
outputPath := filepath.Join(t.TempDir(), "github-output")
cmd := exec.Command("python3", filepath.Join(root, "scripts", "ci", "eval_comment_parse.py"))
cmd.Env = append(
os.Environ(),
"COMMENT_BODY="+comment,
"GITHUB_OUTPUT="+outputPath,
)
combined, runErr := cmd.CombinedOutput()
fileOutput, readErr := os.ReadFile(outputPath)
if readErr != nil && !os.IsNotExist(readErr) {
t.Fatalf("read parser GitHub output: %v", readErr)
}
return string(combined) + string(fileOutput), runErr
}