为角色授予指定工作流的运行权限¶
先列出工作区中的角色和工作流，选择目标后校验授权请求，再按需保存。示例授予指定工作流的运行权限，不创建角色、不修改成员绑定，也不启动工作流。
准备¶
目标角色 ID：首次运行下方示例，从角色输出中选择允许修改策略的普通角色。
工作流 ID：首次运行下方示例，从工作流列表中复制目标 ID。
授权决定：由工作区权限管理者确认目标角色确实需要运行该工作流。
首次运行时，清除示例的目标角色、工作流和保存开关，仅列出可选对象：
unset TARGET_ROLE_ID WORKFLOW_ID APPLY_GRANT
列出角色和工作流，再校验授权¶
将所选语言代码保存为本地程序后运行。未设置目标角色和工作流时，程序只列出角色的 ID、名称、生命周期状态、是否允许修改策略，以及工作流的 ID、名称和列表总数。列表为空或未包含目标时，不要自行构造 ID；工作流列表只代表当前返回的一页。
选定对象后，按配置目标并保存授权设置环境变量并重新运行。程序读取当前权限结构与目标角色的策略版本、生命周期版本，构造具体的工作流运行授权请求；默认只校验，不保存。
import os
import uuid
import moi_product_sdk as sdk
def main():
client = sdk.new_with_personal_access_token(
os.environ["PRODUCT_API_BASE_URL"], os.environ["PRODUCT_API_KEY"]
)
workspace = client.workspace(os.environ["WORKSPACE_ID"])
iam = workspace.iam()
for role in iam.roles().roles:
print("role:", role.role_id, role.role_name, role.lifecycle_state,
"editable:", role.policy_mutation_allowed)
page = workspace.workflows().list()
print("workflow total:", page.total)
for workflow in page.workflows:
print("workflow:", workflow.id, workflow.name)
role_id = os.environ.get("TARGET_ROLE_ID", "")
workflow_id = os.environ.get("WORKFLOW_ID", "")
if not role_id or not workflow_id:
print("选择上方角色与工作流，设置 TARGET_ROLE_ID、WORKFLOW_ID 后重新运行。")
return
role = iam.role_permission_view(role_id)
if role.protected_role or not role.policy_mutation_allowed:
raise RuntimeError("该角色不允许修改策略")
schema = iam.permission_schema()
before = iam.object_permission_view("workflow", workflow_id)
print("owner role:", before.owner.role_id)
request = sdk.IAMObjectGrantsMutationRequest(
schema_version=schema.schema_version,
expected_policy_version=role.policy_version,
expected_role_lifecycle_version=role.expected_role_lifecycle_version,
request_id=str(uuid.uuid4()),
changes=[sdk.IAMObjectGrantChange(action_id="workflow.run", remove=False)],
)
checked = iam.validate_object_grants("workflow", workflow_id, role_id, request)
if not checked.validate_only:
raise RuntimeError("未收到校验响应")
print("validate_only:", checked.validate_only, "would add:", checked.added_count,
"would remove:", checked.removed_count)
if os.environ.get("APPLY_GRANT") != "1":
print("仅校验，未保存授权。确认后设置 APPLY_GRANT=1 再运行。")
return
saved = iam.apply_object_grants("workflow", workflow_id, role_id, request)
print("saved role:", saved.target_role_id, "policy version:", saved.policy_version,
"added:", saved.added_count, "removed:", saved.removed_count,
"replay:", saved.idempotent_replay)
current = iam.object_permission_view("workflow", workflow_id)
found = any(g.role_id == role_id and g.action_id == "workflow.run"
and g.resource_id == workflow_id for g in current.direct_grants)
if saved.validate_only or not found:
raise RuntimeError("尚未确认目标角色的工作流运行直接授权，请重新读取权限视图")
print("已确认目标角色对指定工作流的运行直接授权。")
if __name__ == "__main__":
main()
package main
import (
"context"
"fmt"
"os"
"time"
sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)
func main() {
ctx := context.Background()
client, err := sdk.NewWithPersonalAccessToken(os.Getenv("PRODUCT_API_BASE_URL"), os.Getenv("PRODUCT_API_KEY"))
if err != nil {
panic(err)
}
defer client.Close()
workspace, err := client.Workspace(os.Getenv("WORKSPACE_ID"))
if err != nil {
panic(err)
}
iam := workspace.IAM()
roles, err := iam.Roles(ctx)
if err != nil {
panic(err)
}
for _, role := range roles.GetRoles() {
fmt.Println("role:", role.GetRoleId(), role.GetRoleName(), role.GetLifecycleState(), "editable:", role.GetPolicyMutationAllowed())
}
page, err := workspace.Workflows().List(ctx)
if err != nil {
panic(err)
}
fmt.Println("workflow total:", page.GetTotal())
for _, workflow := range page.GetWorkflows() {
fmt.Println("workflow:", workflow.GetId(), workflow.GetName())
}
roleID, workflowID := os.Getenv("TARGET_ROLE_ID"), os.Getenv("WORKFLOW_ID")
if roleID == "" || workflowID == "" {
fmt.Println("选择上方角色与工作流，设置 TARGET_ROLE_ID、WORKFLOW_ID 后重新运行。")
return
}
role, err := iam.RolePermissionView(ctx, roleID)
if err != nil {
panic(err)
}
if role.GetProtectedRole() || !role.GetPolicyMutationAllowed() {
panic("该角色不允许修改策略")
}
schema, err := iam.PermissionSchema(ctx)
if err != nil {
panic(err)
}
before, err := iam.ObjectPermissionView(ctx, "workflow", workflowID)
if err != nil {
panic(err)
}
fmt.Println("owner role:", before.GetOwner().GetRoleId())
request := &sdk.IAMObjectGrantsMutationRequest{
SchemaVersion: schema.GetSchemaVersion(),
ExpectedPolicyVersion: role.GetPolicyVersion(),
ExpectedRoleLifecycleVersion: role.GetExpectedRoleLifecycleVersion(),
RequestId: fmt.Sprintf("workflow-grant-%d", time.Now().UnixNano()),
Changes: []*sdk.IAMObjectGrantChange{{ActionId: "workflow.run", Remove: false}},
}
checked, err := iam.ValidateObjectGrants(ctx, "workflow", workflowID, roleID, request)
if err != nil {
panic(err)
}
if !checked.GetValidateOnly() {
panic("未收到校验响应")
}
fmt.Println("validate_only:", checked.GetValidateOnly(), "would add:", checked.GetAddedCount(), "would remove:", checked.GetRemovedCount())
if os.Getenv("APPLY_GRANT") != "1" {
fmt.Println("仅校验，未保存授权。确认后设置 APPLY_GRANT=1 再运行。")
return
}
saved, err := iam.ApplyObjectGrants(ctx, "workflow", workflowID, roleID, request)
if err != nil {
panic(err)
}
fmt.Println("saved role:", saved.GetTargetRoleId(), "policy version:", saved.GetPolicyVersion(), "added:", saved.GetAddedCount(), "removed:", saved.GetRemovedCount(), "replay:", saved.GetIdempotentReplay())
current, err := iam.ObjectPermissionView(ctx, "workflow", workflowID)
if err != nil {
panic(err)
}
found := false
for _, grant := range current.GetDirectGrants() {
if grant.GetRoleId() == roleID && grant.GetActionId() == "workflow.run" && grant.GetResourceId() == workflowID {
found = true
}
}
if saved.GetValidateOnly() || !found {
panic("尚未确认目标角色的工作流运行直接授权，请重新读取权限视图")
}
fmt.Println("已确认目标角色对指定工作流的运行直接授权。")
}
Python 保存为 iam_grant.py 后运行 python iam_grant.py；Go 保存为 main.go，在已引入 SDK 的模块中运行 go run .。首次运行后，从输出中选择目标普通角色和工作流。
配置目标并保存授权¶
将上一步输出的角色 ID 和工作流 ID 填入以下变量，然后重新运行同一程序。此时仍只校验授权意图，尚未保存变更。
export TARGET_ROLE_ID='<角色输出中的 ID>'
export WORKFLOW_ID='<工作流输出中的 ID>'
校验返回后，输出会标明仅校验，以及预计新增和移除的授权条目数量。确认对象和变更符合要求后，设置以下变量并再次运行同一程序。此次会重新读取版本、重新校验，然后保存授权。
export APPLY_GRANT='1'
保存会影响目标角色对该工作流的访问。完成后执行 unset APPLY_GRANT，避免下一次运行时继续提交变更。
确认授权结果¶
校验返回预计增删数量；保存后，程序重新读取权限视图，确认所选角色具有目标工作流的直接运行授权。授权已存在时，新增数量可以为零。
并发修改引起版本冲突时，重新读取角色版本并核对变更，再提交。校验不会锁定后续保存。工作流运行所依赖的数据表、连接器等资源需另行授权。
最后更新于