From c267f77eee70e7af1fc2721ff3b913c2a9956f6f Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 16:46:47 +0800 Subject: [PATCH] feat: distribute credentials in worker snapshots --- README.md | 18 +- api/proto/controlplane/v1/controlplane.proto | 10 + docs/adr/005-redis-activity-pool.md | 6 +- docs/api/control-plane.md | 10 +- docs/development/implementation-plan.md | 8 +- docs/requirements/completion-audit.md | 11 +- docs/requirements/traceability.md | 2 +- gen/controlplane/v1/controlplane.pb.go | 351 +++++++++++------- internal/controller/bootstrap/bootstrap.go | 1 + .../worker/owned_snapshot_source.go | 92 ++++- internal/controller/worker/service.go | 16 +- .../controller/worker/snapshot_source_test.go | 39 ++ internal/gateway/bootstrap/bootstrap.go | 15 +- internal/gateway/controlplane/watcher.go | 38 +- internal/gateway/controlplane/watcher_test.go | 10 +- internal/gateway/server/e2e_test.go | 1 - internal/gateway/server/handler_test.go | 1 - internal/gateway/snapshot/store.go | 164 ++++++-- internal/gateway/snapshot/store_test.go | 31 ++ 19 files changed, 618 insertions(+), 206 deletions(-) diff --git a/README.md b/README.md index ea5dd32..2df10be 100644 --- a/README.md +++ b/README.md @@ -44,12 +44,12 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/ Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行 - 有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway - Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View，并只按当前未过期 +有界退避重连。Controller 可向 Worker 下发已归属 Proxy、Gateway Routing 与按引用去重的 +凭据材料快照。Gateway 会将 Routing、Proxy 与凭据原子编译为同一内存 View，并只按当前未过期 View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待， 以及仍经过目标地址策略的 direct；`proxy-gateway` 通过独立控制面拨号地址维护 - Session，并仅在持有未过期 Snapshot 时 Ready；凭据分发仍待装配。 + Session，并仅在持有未过期 Snapshot 时 Ready；凭据材料只保留在当前节点内存 View。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -78,7 +78,7 @@ flowchart LR ## 当前完成度 -截至 **2026-07-31**，实施计划检查项为 **54 / 74（73.0%）**。详情见 +截至 **2026-07-31**，实施计划检查项为 **55 / 74（74.3%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 @@ -86,9 +86,9 @@ flowchart LR 提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL 管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 Redis 会话栅栏。 -- **部分完成**：Gateway 的带凭据 Proxy 分发与 Outcome 上报、Checker 调度与健康状态链、 - Docker Compose/Kubernetes 运行时 mTLS Overlay，以及 protobuf 契约。 -- **待完成**：带凭据 Proxy 分发、Outcome 上报、Checker 调度与健康状态链， +- **部分完成**：Gateway Outcome 上报、Checker 调度与健康状态链、Docker Compose/ + Kubernetes 运行时 mTLS Overlay，以及 protobuf 契约。 +- **待完成**：Outcome 上报、Checker 调度与健康状态链， 以及 loadgen 和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 @@ -185,8 +185,8 @@ Snapshot 时 `/readyz` 返回 `503`。Checker 与 loadgen 命令尚未实现。 ownership 索引，以及 Gateway 快照客户端。 - **P0 - Checker 健康链**：Checker 调度、实际探测、Observation reducer，以及 `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链。 -- **P1 - Gateway 与 Routing**：Gateway 进程、五种 Routing 策略与 - `onUnavailable` 已接入；带凭据 Proxy 分发、动态容量调整和 Drain 闭环待完成。 +- **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与 + `onUnavailable` 已接入；动态容量调整和 Drain 闭环待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， 以及故障转移和恢复演练。 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及 diff --git a/api/proto/controlplane/v1/controlplane.proto b/api/proto/controlplane/v1/controlplane.proto index 5474d43..79132d0 100644 --- a/api/proto/controlplane/v1/controlplane.proto +++ b/api/proto/controlplane/v1/controlplane.proto @@ -64,6 +64,16 @@ message WorkerSnapshot { bytes checksum = 5; repeated RoutingRule routing = 6; repeated OwnedProxy proxies = 7; + // Credential materials are protected by the WorkerControlPlane mTLS session + // and retained only in the Gateway's current in-memory snapshot view. + repeated SnapshotCredential credentials = 14; +} + +message SnapshotCredential { + string secret_ref = 1; + string credential_version = 2; + string username = 3; + string password = 4; } message SnapshotDelta { diff --git a/docs/adr/005-redis-activity-pool.md b/docs/adr/005-redis-activity-pool.md index a1b2ae7..a3e4197 100644 --- a/docs/adr/005-redis-activity-pool.md +++ b/docs/adr/005-redis-activity-pool.md @@ -235,9 +235,9 @@ Provider Parser 继续通过 `credentials.Store` 生成 `SecretRef` 和 在活动池状态提交之前。Distribution 所需凭据只保存在 Proxy 硬 TTL 和幂等 TTL 约束内，Extract 脚本可原子保存完整重放响应。 -Gateway Snapshot 继续只携带凭据引用；Gateway 通过控制面下发到节点内存的 -凭据材料解析引用，不在请求热路径查询 Redis。凭据分发与轮换属于独立后续 -实现，不改变本 ADR 的活动池边界。 +Gateway Snapshot 携带凭据引用及按引用去重的材料；材料只经 mTLS 控制面进入 +Gateway 当前内存 View，不在请求热路径查询 Redis，也不进入 PostgreSQL。凭据轮换的 +主动推送仍属于后续实现，不改变本 ADR 的活动池边界。 ### 库存真值 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index f9e862e..d38ad38 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -27,8 +27,9 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和 `WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照， Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership -索引构建无凭据引用的已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 -`usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。完整 Snapshot +索引构建已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 `usable_until`。Proxy 引用的 +凭据材料按 `secret_ref + credential_version` 去重，随完整 Snapshot 经 mTLS 下发，仅保留在 +Gateway 当前内存 View。完整 Snapshot 已从配置原始顺序和 Admin 当前状态合成 Gateway Routing，并与 Proxy 一起纳入 checksum； Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View，动态 Router 只匹配 当前未过期 View；派发器已按五种策略从该 View 选择上游，且在 Proxy 容量耗尽时只在该 @@ -36,7 +37,7 @@ View 的其余候选中回退。`wait_timeout` 随 `on_unavailable=WAIT` 下发 本地容量等待使用；`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。`proxy-gateway` 已装配 Register/Watch/ACK/Runtime 会话、HTTP 代理监听和 Snapshot 就绪探针；控制面中断时 保持进程运行并以有界退避重连，未取得有效 Snapshot 的 Worker 不会 Ready。增量、带凭据 -Proxy 分发、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`； +Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`； `100,000 QPS` 仍是未验证的设计目标。 `WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id` @@ -164,7 +165,8 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY，并更新 Redis 活动池， 集群环境使用 mTLS，证书身份绑定 Worker/Checker 类型和环境。服务端校验 消息中的逻辑 ID 与证书授权一致，设置单消息大小、流持续时间、并发 Stream -和上报批次上限。`secret_ref` 是受控引用，不在 Proto 中传播真实密码。 +和上报批次上限。`secret_ref` 是受控引用；完整 Snapshot 的 `credentials` 在 mTLS 会话中 +携带引用对应材料，Controller 和 Gateway 仅在内存处理，禁止写入 Redis/PostgreSQL、日志或指标。 ## 9. Gateway 启动参数 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 5b7e9b2..ae2c866 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -214,8 +214,8 @@ WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Run 身份、消息/流限制和有界停机已实现。`WatchSnapshots` 会发送当前 epoch 的基础完整 Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件与 `proxy-gateway` 进程装配。 -按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本，并可构建无凭据引用的 -已归属 Proxy payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成， +按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本，并可构建已归属 Proxy、 +去重凭据材料和 Routing 的完整 payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成， 重注册会清除旧引用，避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整 Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已为 Gateway 调用方 提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Gateway Routing @@ -223,7 +223,7 @@ payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 check Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。派发器的五种上游选择已 接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable` 的 reject、wait 与 direct 已接入 Gateway；`proxy-gateway` 已装配本地 HTTP/Metrics 监听、快照就绪探针和 -控制面重连，凭据分发与 Outcome 上报仍未实现。 +控制面重连与快照凭据分发，Outcome 上报仍未实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖 @@ -260,7 +260,7 @@ Controller 多副本共享同一计数。Client 身份只以 SHA-256 摘要进 原子回收历史字段；Redis 异常 fail-closed 并返回 503，真实额度耗尽返回 429。 Gateway 请求热路径仍只使用本地准入，不增加 Redis/PostgreSQL 调用。 WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础 -Snapshot 流和 Gateway 会话客户端已完成；权威 Proxy/Routing 发布、凭据分发、 +Snapshot 流、Gateway 会话客户端与快照凭据分发已完成；权威 Proxy/Routing 发布、 Outcome 和健康执行链仍待完成，因此 Task 10 尚未全部完成。 ## Task 11: Checker and Health Reducer diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 317d37d..60c1d9f 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -53,9 +53,10 @@ 服务端 TTL，负向 ACK 会关闭 Runtime 写入栅栏。Controller 已装配 loopback plaintext fixture 与 SPIFFE mTLS 服务端；基础 Snapshot 流和 Gateway 的 Register/Watch/ACK/Runtime 会话协调已实现。Redis 以 Worker 可下发 ownership - 索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Gateway Routing - payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 - 凭据分发、Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 +索引构建已归属 Proxy payload，并以租约收紧可用期；凭据材料按引用去重并与 Proxy/ +Routing 一起纳入 Snapshot checksum，仅在 mTLS 和 Gateway 当前内存 View 中处理。Gateway +Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 +Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot @@ -101,9 +102,9 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt 5. Worker Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、 SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、 分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。 -6. Worker 基础网络快照流、无凭据引用 Proxy/Gateway Routing payload、Gateway Snapshot +6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot 客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成； - 凭据分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 + Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 7. Checker 调度、探测器和健康 reducer。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index bc38337..3dd217e 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,7 +7,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布无凭据 Proxy/Gateway Routing 完整快照；Gateway 已将同版本 Routing 编译为动态匹配 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物与带凭据 Proxy 分发待实现 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Gateway 已将它们编译为同版本动态 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | diff --git a/gen/controlplane/v1/controlplane.pb.go b/gen/controlplane/v1/controlplane.pb.go index 405f2ca..f1f0808 100644 --- a/gen/controlplane/v1/controlplane.pb.go +++ b/gen/controlplane/v1/controlplane.pb.go @@ -604,8 +604,11 @@ type WorkerSnapshot struct { Checksum []byte `protobuf:"bytes,5,opt,name=checksum,proto3" json:"checksum,omitempty"` Routing []*RoutingRule `protobuf:"bytes,6,rep,name=routing,proto3" json:"routing,omitempty"` Proxies []*OwnedProxy `protobuf:"bytes,7,rep,name=proxies,proto3" json:"proxies,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Credential materials are protected by the WorkerControlPlane mTLS session + // and retained only in the Gateway's current in-memory snapshot view. + Credentials []*SnapshotCredential `protobuf:"bytes,14,rep,name=credentials,proto3" json:"credentials,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WorkerSnapshot) Reset() { @@ -687,6 +690,81 @@ func (x *WorkerSnapshot) GetProxies() []*OwnedProxy { return nil } +func (x *WorkerSnapshot) GetCredentials() []*SnapshotCredential { + if x != nil { + return x.Credentials + } + return nil +} + +type SnapshotCredential struct { + state protoimpl.MessageState `protogen:"open.v1"` + SecretRef string `protobuf:"bytes,1,opt,name=secret_ref,json=secretRef,proto3" json:"secret_ref,omitempty"` + CredentialVersion string `protobuf:"bytes,2,opt,name=credential_version,json=credentialVersion,proto3" json:"credential_version,omitempty"` + Username string `protobuf:"bytes,3,opt,name=username,proto3" json:"username,omitempty"` + Password string `protobuf:"bytes,4,opt,name=password,proto3" json:"password,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SnapshotCredential) Reset() { + *x = SnapshotCredential{} + mi := &file_controlplane_v1_controlplane_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SnapshotCredential) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SnapshotCredential) ProtoMessage() {} + +func (x *SnapshotCredential) ProtoReflect() protoreflect.Message { + mi := &file_controlplane_v1_controlplane_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SnapshotCredential.ProtoReflect.Descriptor instead. +func (*SnapshotCredential) Descriptor() ([]byte, []int) { + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{5} +} + +func (x *SnapshotCredential) GetSecretRef() string { + if x != nil { + return x.SecretRef + } + return "" +} + +func (x *SnapshotCredential) GetCredentialVersion() string { + if x != nil { + return x.CredentialVersion + } + return "" +} + +func (x *SnapshotCredential) GetUsername() string { + if x != nil { + return x.Username + } + return "" +} + +func (x *SnapshotCredential) GetPassword() string { + if x != nil { + return x.Password + } + return "" +} + type SnapshotDelta struct { state protoimpl.MessageState `protogen:"open.v1"` BaseVersion uint64 `protobuf:"varint,1,opt,name=base_version,json=baseVersion,proto3" json:"base_version,omitempty"` @@ -704,7 +782,7 @@ type SnapshotDelta struct { func (x *SnapshotDelta) Reset() { *x = SnapshotDelta{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[5] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -716,7 +794,7 @@ func (x *SnapshotDelta) String() string { func (*SnapshotDelta) ProtoMessage() {} func (x *SnapshotDelta) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[5] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -729,7 +807,7 @@ func (x *SnapshotDelta) ProtoReflect() protoreflect.Message { // Deprecated: Use SnapshotDelta.ProtoReflect.Descriptor instead. func (*SnapshotDelta) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{5} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{6} } func (x *SnapshotDelta) GetBaseVersion() uint64 { @@ -813,7 +891,7 @@ type RoutingRule struct { func (x *RoutingRule) Reset() { *x = RoutingRule{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[6] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -825,7 +903,7 @@ func (x *RoutingRule) String() string { func (*RoutingRule) ProtoMessage() {} func (x *RoutingRule) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[6] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -838,7 +916,7 @@ func (x *RoutingRule) ProtoReflect() protoreflect.Message { // Deprecated: Use RoutingRule.ProtoReflect.Descriptor instead. func (*RoutingRule) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{6} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{7} } func (x *RoutingRule) GetName() string { @@ -922,7 +1000,7 @@ type RoutingStrategy struct { func (x *RoutingStrategy) Reset() { *x = RoutingStrategy{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[7] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -934,7 +1012,7 @@ func (x *RoutingStrategy) String() string { func (*RoutingStrategy) ProtoMessage() {} func (x *RoutingStrategy) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[7] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -947,7 +1025,7 @@ func (x *RoutingStrategy) ProtoReflect() protoreflect.Message { // Deprecated: Use RoutingStrategy.ProtoReflect.Descriptor instead. func (*RoutingStrategy) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{7} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{8} } func (x *RoutingStrategy) GetType() StrategyType { @@ -992,7 +1070,7 @@ type OwnedProxy struct { func (x *OwnedProxy) Reset() { *x = OwnedProxy{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[8] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1004,7 +1082,7 @@ func (x *OwnedProxy) String() string { func (*OwnedProxy) ProtoMessage() {} func (x *OwnedProxy) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[8] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1017,7 +1095,7 @@ func (x *OwnedProxy) ProtoReflect() protoreflect.Message { // Deprecated: Use OwnedProxy.ProtoReflect.Descriptor instead. func (*OwnedProxy) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{8} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{9} } func (x *OwnedProxy) GetId() string { @@ -1127,7 +1205,7 @@ type AcknowledgeSnapshotRequest struct { func (x *AcknowledgeSnapshotRequest) Reset() { *x = AcknowledgeSnapshotRequest{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[9] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1139,7 +1217,7 @@ func (x *AcknowledgeSnapshotRequest) String() string { func (*AcknowledgeSnapshotRequest) ProtoMessage() {} func (x *AcknowledgeSnapshotRequest) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[9] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1152,7 +1230,7 @@ func (x *AcknowledgeSnapshotRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use AcknowledgeSnapshotRequest.ProtoReflect.Descriptor instead. func (*AcknowledgeSnapshotRequest) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{9} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{10} } func (x *AcknowledgeSnapshotRequest) GetWorkerId() string { @@ -1223,7 +1301,7 @@ type OutcomeBatch struct { func (x *OutcomeBatch) Reset() { *x = OutcomeBatch{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[10] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1235,7 +1313,7 @@ func (x *OutcomeBatch) String() string { func (*OutcomeBatch) ProtoMessage() {} func (x *OutcomeBatch) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[10] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1248,7 +1326,7 @@ func (x *OutcomeBatch) ProtoReflect() protoreflect.Message { // Deprecated: Use OutcomeBatch.ProtoReflect.Descriptor instead. func (*OutcomeBatch) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{10} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{11} } func (x *OutcomeBatch) GetWorkerId() string { @@ -1294,7 +1372,7 @@ type ProxyOutcome struct { func (x *ProxyOutcome) Reset() { *x = ProxyOutcome{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[11] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1306,7 +1384,7 @@ func (x *ProxyOutcome) String() string { func (*ProxyOutcome) ProtoMessage() {} func (x *ProxyOutcome) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[11] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1319,7 +1397,7 @@ func (x *ProxyOutcome) ProtoReflect() protoreflect.Message { // Deprecated: Use ProxyOutcome.ProtoReflect.Descriptor instead. func (*ProxyOutcome) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{11} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{12} } func (x *ProxyOutcome) GetProxyId() string { @@ -1380,7 +1458,7 @@ type ReportOutcomesResponse struct { func (x *ReportOutcomesResponse) Reset() { *x = ReportOutcomesResponse{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[12] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1392,7 +1470,7 @@ func (x *ReportOutcomesResponse) String() string { func (*ReportOutcomesResponse) ProtoMessage() {} func (x *ReportOutcomesResponse) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[12] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1405,7 +1483,7 @@ func (x *ReportOutcomesResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use ReportOutcomesResponse.ProtoReflect.Descriptor instead. func (*ReportOutcomesResponse) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{12} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{13} } func (x *ReportOutcomesResponse) GetAcceptedThroughSequence() uint64 { @@ -1430,7 +1508,7 @@ type ReportRuntimeRequest struct { func (x *ReportRuntimeRequest) Reset() { *x = ReportRuntimeRequest{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[13] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1442,7 +1520,7 @@ func (x *ReportRuntimeRequest) String() string { func (*ReportRuntimeRequest) ProtoMessage() {} func (x *ReportRuntimeRequest) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[13] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1455,7 +1533,7 @@ func (x *ReportRuntimeRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use ReportRuntimeRequest.ProtoReflect.Descriptor instead. func (*ReportRuntimeRequest) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{13} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{14} } func (x *ReportRuntimeRequest) GetWorkerId() string { @@ -1519,7 +1597,7 @@ type ProxyRuntime struct { func (x *ProxyRuntime) Reset() { *x = ProxyRuntime{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[14] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1531,7 +1609,7 @@ func (x *ProxyRuntime) String() string { func (*ProxyRuntime) ProtoMessage() {} func (x *ProxyRuntime) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[14] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1544,7 +1622,7 @@ func (x *ProxyRuntime) ProtoReflect() protoreflect.Message { // Deprecated: Use ProxyRuntime.ProtoReflect.Descriptor instead. func (*ProxyRuntime) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{14} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{15} } func (x *ProxyRuntime) GetProxyId() string { @@ -1586,7 +1664,7 @@ type ReportRuntimeResponse struct { func (x *ReportRuntimeResponse) Reset() { *x = ReportRuntimeResponse{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[15] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1598,7 +1676,7 @@ func (x *ReportRuntimeResponse) String() string { func (*ReportRuntimeResponse) ProtoMessage() {} func (x *ReportRuntimeResponse) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[15] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1611,7 +1689,7 @@ func (x *ReportRuntimeResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use ReportRuntimeResponse.ProtoReflect.Descriptor instead. func (*ReportRuntimeResponse) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{15} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{16} } func (x *ReportRuntimeResponse) GetAcceptedOwnershipEpoch() uint64 { @@ -1647,7 +1725,7 @@ type StreamCheckTasksRequest struct { func (x *StreamCheckTasksRequest) Reset() { *x = StreamCheckTasksRequest{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[16] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1659,7 +1737,7 @@ func (x *StreamCheckTasksRequest) String() string { func (*StreamCheckTasksRequest) ProtoMessage() {} func (x *StreamCheckTasksRequest) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[16] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1672,7 +1750,7 @@ func (x *StreamCheckTasksRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StreamCheckTasksRequest.ProtoReflect.Descriptor instead. func (*StreamCheckTasksRequest) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{16} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{17} } func (x *StreamCheckTasksRequest) GetCheckerId() string { @@ -1723,7 +1801,7 @@ type CheckTask struct { func (x *CheckTask) Reset() { *x = CheckTask{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[17] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1735,7 +1813,7 @@ func (x *CheckTask) String() string { func (*CheckTask) ProtoMessage() {} func (x *CheckTask) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[17] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1748,7 +1826,7 @@ func (x *CheckTask) ProtoReflect() protoreflect.Message { // Deprecated: Use CheckTask.ProtoReflect.Descriptor instead. func (*CheckTask) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{17} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{18} } func (x *CheckTask) GetTaskId() string { @@ -1845,7 +1923,7 @@ type ObservationBatch struct { func (x *ObservationBatch) Reset() { *x = ObservationBatch{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[18] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1857,7 +1935,7 @@ func (x *ObservationBatch) String() string { func (*ObservationBatch) ProtoMessage() {} func (x *ObservationBatch) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[18] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1870,7 +1948,7 @@ func (x *ObservationBatch) ProtoReflect() protoreflect.Message { // Deprecated: Use ObservationBatch.ProtoReflect.Descriptor instead. func (*ObservationBatch) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{18} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{19} } func (x *ObservationBatch) GetCheckerId() string { @@ -1905,7 +1983,7 @@ type HealthObservation struct { func (x *HealthObservation) Reset() { *x = HealthObservation{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[19] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1917,7 +1995,7 @@ func (x *HealthObservation) String() string { func (*HealthObservation) ProtoMessage() {} func (x *HealthObservation) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[19] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1930,7 +2008,7 @@ func (x *HealthObservation) ProtoReflect() protoreflect.Message { // Deprecated: Use HealthObservation.ProtoReflect.Descriptor instead. func (*HealthObservation) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{19} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{20} } func (x *HealthObservation) GetTaskId() string { @@ -2013,7 +2091,7 @@ type ReportObservationsResponse struct { func (x *ReportObservationsResponse) Reset() { *x = ReportObservationsResponse{} - mi := &file_controlplane_v1_controlplane_proto_msgTypes[20] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2025,7 +2103,7 @@ func (x *ReportObservationsResponse) String() string { func (*ReportObservationsResponse) ProtoMessage() {} func (x *ReportObservationsResponse) ProtoReflect() protoreflect.Message { - mi := &file_controlplane_v1_controlplane_proto_msgTypes[20] + mi := &file_controlplane_v1_controlplane_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2038,7 +2116,7 @@ func (x *ReportObservationsResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use ReportObservationsResponse.ProtoReflect.Descriptor instead. func (*ReportObservationsResponse) Descriptor() ([]byte, []int) { - return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{20} + return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{21} } func (x *ReportObservationsResponse) GetAccepted() uint32 { @@ -2086,7 +2164,7 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" + "\x10SnapshotEnvelope\x12@\n" + "\x04full\x18\x01 \x01(\v2*.proxy_pool.controlplane.v1.WorkerSnapshotH\x00R\x04full\x12A\n" + "\x05delta\x18\x02 \x01(\v2).proxy_pool.controlplane.v1.SnapshotDeltaH\x00R\x05deltaB\t\n" + - "\apayload\"\xf0\x02\n" + + "\apayload\"\xc2\x03\n" + "\x0eWorkerSnapshot\x12\x18\n" + "\aversion\x18\x01 \x01(\x04R\aversion\x12'\n" + "\x0fownership_epoch\x18\x02 \x01(\x04R\x0eownershipEpoch\x12=\n" + @@ -2095,7 +2173,14 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" + "validUntil\x12\x1a\n" + "\bchecksum\x18\x05 \x01(\fR\bchecksum\x12A\n" + "\arouting\x18\x06 \x03(\v2'.proxy_pool.controlplane.v1.RoutingRuleR\arouting\x12@\n" + - "\aproxies\x18\a \x03(\v2&.proxy_pool.controlplane.v1.OwnedProxyR\aproxies\"\xd7\x03\n" + + "\aproxies\x18\a \x03(\v2&.proxy_pool.controlplane.v1.OwnedProxyR\aproxies\x12P\n" + + "\vcredentials\x18\x0e \x03(\v2..proxy_pool.controlplane.v1.SnapshotCredentialR\vcredentials\"\x9a\x01\n" + + "\x12SnapshotCredential\x12\x1d\n" + + "\n" + + "secret_ref\x18\x01 \x01(\tR\tsecretRef\x12-\n" + + "\x12credential_version\x18\x02 \x01(\tR\x11credentialVersion\x12\x1a\n" + + "\busername\x18\x03 \x01(\tR\busername\x12\x1a\n" + + "\bpassword\x18\x04 \x01(\tR\bpassword\"\xd7\x03\n" + "\rSnapshotDelta\x12!\n" + "\fbase_version\x18\x01 \x01(\x04R\vbaseVersion\x12\x18\n" + "\aversion\x18\x02 \x01(\x04R\aversion\x12'\n" + @@ -2295,7 +2380,7 @@ func file_controlplane_v1_controlplane_proto_rawDescGZIP() []byte { } var file_controlplane_v1_controlplane_proto_enumTypes = make([]protoimpl.EnumInfo, 5) -var file_controlplane_v1_controlplane_proto_msgTypes = make([]protoimpl.MessageInfo, 25) +var file_controlplane_v1_controlplane_proto_msgTypes = make([]protoimpl.MessageInfo, 26) var file_controlplane_v1_controlplane_proto_goTypes = []any{ (StrategyType)(0), // 0: proxy_pool.controlplane.v1.StrategyType (UnavailableAction)(0), // 1: proxy_pool.controlplane.v1.UnavailableAction @@ -2307,87 +2392,89 @@ var file_controlplane_v1_controlplane_proto_goTypes = []any{ (*WatchSnapshotsRequest)(nil), // 7: proxy_pool.controlplane.v1.WatchSnapshotsRequest (*SnapshotEnvelope)(nil), // 8: proxy_pool.controlplane.v1.SnapshotEnvelope (*WorkerSnapshot)(nil), // 9: proxy_pool.controlplane.v1.WorkerSnapshot - (*SnapshotDelta)(nil), // 10: proxy_pool.controlplane.v1.SnapshotDelta - (*RoutingRule)(nil), // 11: proxy_pool.controlplane.v1.RoutingRule - (*RoutingStrategy)(nil), // 12: proxy_pool.controlplane.v1.RoutingStrategy - (*OwnedProxy)(nil), // 13: proxy_pool.controlplane.v1.OwnedProxy - (*AcknowledgeSnapshotRequest)(nil), // 14: proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest - (*OutcomeBatch)(nil), // 15: proxy_pool.controlplane.v1.OutcomeBatch - (*ProxyOutcome)(nil), // 16: proxy_pool.controlplane.v1.ProxyOutcome - (*ReportOutcomesResponse)(nil), // 17: proxy_pool.controlplane.v1.ReportOutcomesResponse - (*ReportRuntimeRequest)(nil), // 18: proxy_pool.controlplane.v1.ReportRuntimeRequest - (*ProxyRuntime)(nil), // 19: proxy_pool.controlplane.v1.ProxyRuntime - (*ReportRuntimeResponse)(nil), // 20: proxy_pool.controlplane.v1.ReportRuntimeResponse - (*StreamCheckTasksRequest)(nil), // 21: proxy_pool.controlplane.v1.StreamCheckTasksRequest - (*CheckTask)(nil), // 22: proxy_pool.controlplane.v1.CheckTask - (*ObservationBatch)(nil), // 23: proxy_pool.controlplane.v1.ObservationBatch - (*HealthObservation)(nil), // 24: proxy_pool.controlplane.v1.HealthObservation - (*ReportObservationsResponse)(nil), // 25: proxy_pool.controlplane.v1.ReportObservationsResponse - nil, // 26: proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry - nil, // 27: proxy_pool.controlplane.v1.RoutingRule.HeadersEntry - nil, // 28: proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry - nil, // 29: proxy_pool.controlplane.v1.OwnedProxy.TagsEntry - (*durationpb.Duration)(nil), // 30: google.protobuf.Duration - (*timestamppb.Timestamp)(nil), // 31: google.protobuf.Timestamp - (*emptypb.Empty)(nil), // 32: google.protobuf.Empty + (*SnapshotCredential)(nil), // 10: proxy_pool.controlplane.v1.SnapshotCredential + (*SnapshotDelta)(nil), // 11: proxy_pool.controlplane.v1.SnapshotDelta + (*RoutingRule)(nil), // 12: proxy_pool.controlplane.v1.RoutingRule + (*RoutingStrategy)(nil), // 13: proxy_pool.controlplane.v1.RoutingStrategy + (*OwnedProxy)(nil), // 14: proxy_pool.controlplane.v1.OwnedProxy + (*AcknowledgeSnapshotRequest)(nil), // 15: proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest + (*OutcomeBatch)(nil), // 16: proxy_pool.controlplane.v1.OutcomeBatch + (*ProxyOutcome)(nil), // 17: proxy_pool.controlplane.v1.ProxyOutcome + (*ReportOutcomesResponse)(nil), // 18: proxy_pool.controlplane.v1.ReportOutcomesResponse + (*ReportRuntimeRequest)(nil), // 19: proxy_pool.controlplane.v1.ReportRuntimeRequest + (*ProxyRuntime)(nil), // 20: proxy_pool.controlplane.v1.ProxyRuntime + (*ReportRuntimeResponse)(nil), // 21: proxy_pool.controlplane.v1.ReportRuntimeResponse + (*StreamCheckTasksRequest)(nil), // 22: proxy_pool.controlplane.v1.StreamCheckTasksRequest + (*CheckTask)(nil), // 23: proxy_pool.controlplane.v1.CheckTask + (*ObservationBatch)(nil), // 24: proxy_pool.controlplane.v1.ObservationBatch + (*HealthObservation)(nil), // 25: proxy_pool.controlplane.v1.HealthObservation + (*ReportObservationsResponse)(nil), // 26: proxy_pool.controlplane.v1.ReportObservationsResponse + nil, // 27: proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry + nil, // 28: proxy_pool.controlplane.v1.RoutingRule.HeadersEntry + nil, // 29: proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry + nil, // 30: proxy_pool.controlplane.v1.OwnedProxy.TagsEntry + (*durationpb.Duration)(nil), // 31: google.protobuf.Duration + (*timestamppb.Timestamp)(nil), // 32: google.protobuf.Timestamp + (*emptypb.Empty)(nil), // 33: google.protobuf.Empty } var file_controlplane_v1_controlplane_proto_depIdxs = []int32{ - 26, // 0: proxy_pool.controlplane.v1.RegisterWorkerRequest.labels:type_name -> proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry - 30, // 1: proxy_pool.controlplane.v1.RegisterWorkerResponse.heartbeat_interval:type_name -> google.protobuf.Duration - 30, // 2: proxy_pool.controlplane.v1.RegisterWorkerResponse.max_stale_age:type_name -> google.protobuf.Duration + 27, // 0: proxy_pool.controlplane.v1.RegisterWorkerRequest.labels:type_name -> proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry + 31, // 1: proxy_pool.controlplane.v1.RegisterWorkerResponse.heartbeat_interval:type_name -> google.protobuf.Duration + 31, // 2: proxy_pool.controlplane.v1.RegisterWorkerResponse.max_stale_age:type_name -> google.protobuf.Duration 9, // 3: proxy_pool.controlplane.v1.SnapshotEnvelope.full:type_name -> proxy_pool.controlplane.v1.WorkerSnapshot - 10, // 4: proxy_pool.controlplane.v1.SnapshotEnvelope.delta:type_name -> proxy_pool.controlplane.v1.SnapshotDelta - 31, // 5: proxy_pool.controlplane.v1.WorkerSnapshot.generated_at:type_name -> google.protobuf.Timestamp - 31, // 6: proxy_pool.controlplane.v1.WorkerSnapshot.valid_until:type_name -> google.protobuf.Timestamp - 11, // 7: proxy_pool.controlplane.v1.WorkerSnapshot.routing:type_name -> proxy_pool.controlplane.v1.RoutingRule - 13, // 8: proxy_pool.controlplane.v1.WorkerSnapshot.proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy - 31, // 9: proxy_pool.controlplane.v1.SnapshotDelta.generated_at:type_name -> google.protobuf.Timestamp - 11, // 10: proxy_pool.controlplane.v1.SnapshotDelta.upserted_routing:type_name -> proxy_pool.controlplane.v1.RoutingRule - 13, // 11: proxy_pool.controlplane.v1.SnapshotDelta.upserted_proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy - 27, // 12: proxy_pool.controlplane.v1.RoutingRule.headers:type_name -> proxy_pool.controlplane.v1.RoutingRule.HeadersEntry - 12, // 13: proxy_pool.controlplane.v1.RoutingRule.strategy:type_name -> proxy_pool.controlplane.v1.RoutingStrategy - 1, // 14: proxy_pool.controlplane.v1.RoutingRule.on_unavailable:type_name -> proxy_pool.controlplane.v1.UnavailableAction - 30, // 15: proxy_pool.controlplane.v1.RoutingRule.wait_timeout:type_name -> google.protobuf.Duration - 0, // 16: proxy_pool.controlplane.v1.RoutingStrategy.type:type_name -> proxy_pool.controlplane.v1.StrategyType - 28, // 17: proxy_pool.controlplane.v1.RoutingStrategy.weights:type_name -> proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry - 2, // 18: proxy_pool.controlplane.v1.OwnedProxy.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol - 31, // 19: proxy_pool.controlplane.v1.OwnedProxy.expires_at:type_name -> google.protobuf.Timestamp - 29, // 20: proxy_pool.controlplane.v1.OwnedProxy.tags:type_name -> proxy_pool.controlplane.v1.OwnedProxy.TagsEntry - 31, // 21: proxy_pool.controlplane.v1.OwnedProxy.usable_until:type_name -> google.protobuf.Timestamp - 16, // 22: proxy_pool.controlplane.v1.OutcomeBatch.outcomes:type_name -> proxy_pool.controlplane.v1.ProxyOutcome - 3, // 23: proxy_pool.controlplane.v1.ProxyOutcome.stage:type_name -> proxy_pool.controlplane.v1.OutcomeStage - 30, // 24: proxy_pool.controlplane.v1.ProxyOutcome.latency:type_name -> google.protobuf.Duration - 31, // 25: proxy_pool.controlplane.v1.ProxyOutcome.observed_at:type_name -> google.protobuf.Timestamp - 19, // 26: proxy_pool.controlplane.v1.ReportRuntimeRequest.counters:type_name -> proxy_pool.controlplane.v1.ProxyRuntime - 31, // 27: proxy_pool.controlplane.v1.ReportRuntimeRequest.observed_at:type_name -> google.protobuf.Timestamp - 4, // 28: proxy_pool.controlplane.v1.StreamCheckTasksRequest.supported_levels:type_name -> proxy_pool.controlplane.v1.CheckLevel - 2, // 29: proxy_pool.controlplane.v1.CheckTask.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol - 4, // 30: proxy_pool.controlplane.v1.CheckTask.level:type_name -> proxy_pool.controlplane.v1.CheckLevel - 30, // 31: proxy_pool.controlplane.v1.CheckTask.timeout:type_name -> google.protobuf.Duration - 31, // 32: proxy_pool.controlplane.v1.CheckTask.deadline:type_name -> google.protobuf.Timestamp - 24, // 33: proxy_pool.controlplane.v1.ObservationBatch.observations:type_name -> proxy_pool.controlplane.v1.HealthObservation - 4, // 34: proxy_pool.controlplane.v1.HealthObservation.level:type_name -> proxy_pool.controlplane.v1.CheckLevel - 30, // 35: proxy_pool.controlplane.v1.HealthObservation.latency:type_name -> google.protobuf.Duration - 31, // 36: proxy_pool.controlplane.v1.HealthObservation.observed_at:type_name -> google.protobuf.Timestamp - 5, // 37: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:input_type -> proxy_pool.controlplane.v1.RegisterWorkerRequest - 7, // 38: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:input_type -> proxy_pool.controlplane.v1.WatchSnapshotsRequest - 14, // 39: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:input_type -> proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest - 15, // 40: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:input_type -> proxy_pool.controlplane.v1.OutcomeBatch - 18, // 41: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:input_type -> proxy_pool.controlplane.v1.ReportRuntimeRequest - 21, // 42: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:input_type -> proxy_pool.controlplane.v1.StreamCheckTasksRequest - 23, // 43: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:input_type -> proxy_pool.controlplane.v1.ObservationBatch - 6, // 44: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:output_type -> proxy_pool.controlplane.v1.RegisterWorkerResponse - 8, // 45: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:output_type -> proxy_pool.controlplane.v1.SnapshotEnvelope - 32, // 46: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:output_type -> google.protobuf.Empty - 17, // 47: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:output_type -> proxy_pool.controlplane.v1.ReportOutcomesResponse - 20, // 48: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:output_type -> proxy_pool.controlplane.v1.ReportRuntimeResponse - 22, // 49: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:output_type -> proxy_pool.controlplane.v1.CheckTask - 25, // 50: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:output_type -> proxy_pool.controlplane.v1.ReportObservationsResponse - 44, // [44:51] is the sub-list for method output_type - 37, // [37:44] is the sub-list for method input_type - 37, // [37:37] is the sub-list for extension type_name - 37, // [37:37] is the sub-list for extension extendee - 0, // [0:37] is the sub-list for field type_name + 11, // 4: proxy_pool.controlplane.v1.SnapshotEnvelope.delta:type_name -> proxy_pool.controlplane.v1.SnapshotDelta + 32, // 5: proxy_pool.controlplane.v1.WorkerSnapshot.generated_at:type_name -> google.protobuf.Timestamp + 32, // 6: proxy_pool.controlplane.v1.WorkerSnapshot.valid_until:type_name -> google.protobuf.Timestamp + 12, // 7: proxy_pool.controlplane.v1.WorkerSnapshot.routing:type_name -> proxy_pool.controlplane.v1.RoutingRule + 14, // 8: proxy_pool.controlplane.v1.WorkerSnapshot.proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy + 10, // 9: proxy_pool.controlplane.v1.WorkerSnapshot.credentials:type_name -> proxy_pool.controlplane.v1.SnapshotCredential + 32, // 10: proxy_pool.controlplane.v1.SnapshotDelta.generated_at:type_name -> google.protobuf.Timestamp + 12, // 11: proxy_pool.controlplane.v1.SnapshotDelta.upserted_routing:type_name -> proxy_pool.controlplane.v1.RoutingRule + 14, // 12: proxy_pool.controlplane.v1.SnapshotDelta.upserted_proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy + 28, // 13: proxy_pool.controlplane.v1.RoutingRule.headers:type_name -> proxy_pool.controlplane.v1.RoutingRule.HeadersEntry + 13, // 14: proxy_pool.controlplane.v1.RoutingRule.strategy:type_name -> proxy_pool.controlplane.v1.RoutingStrategy + 1, // 15: proxy_pool.controlplane.v1.RoutingRule.on_unavailable:type_name -> proxy_pool.controlplane.v1.UnavailableAction + 31, // 16: proxy_pool.controlplane.v1.RoutingRule.wait_timeout:type_name -> google.protobuf.Duration + 0, // 17: proxy_pool.controlplane.v1.RoutingStrategy.type:type_name -> proxy_pool.controlplane.v1.StrategyType + 29, // 18: proxy_pool.controlplane.v1.RoutingStrategy.weights:type_name -> proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry + 2, // 19: proxy_pool.controlplane.v1.OwnedProxy.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol + 32, // 20: proxy_pool.controlplane.v1.OwnedProxy.expires_at:type_name -> google.protobuf.Timestamp + 30, // 21: proxy_pool.controlplane.v1.OwnedProxy.tags:type_name -> proxy_pool.controlplane.v1.OwnedProxy.TagsEntry + 32, // 22: proxy_pool.controlplane.v1.OwnedProxy.usable_until:type_name -> google.protobuf.Timestamp + 17, // 23: proxy_pool.controlplane.v1.OutcomeBatch.outcomes:type_name -> proxy_pool.controlplane.v1.ProxyOutcome + 3, // 24: proxy_pool.controlplane.v1.ProxyOutcome.stage:type_name -> proxy_pool.controlplane.v1.OutcomeStage + 31, // 25: proxy_pool.controlplane.v1.ProxyOutcome.latency:type_name -> google.protobuf.Duration + 32, // 26: proxy_pool.controlplane.v1.ProxyOutcome.observed_at:type_name -> google.protobuf.Timestamp + 20, // 27: proxy_pool.controlplane.v1.ReportRuntimeRequest.counters:type_name -> proxy_pool.controlplane.v1.ProxyRuntime + 32, // 28: proxy_pool.controlplane.v1.ReportRuntimeRequest.observed_at:type_name -> google.protobuf.Timestamp + 4, // 29: proxy_pool.controlplane.v1.StreamCheckTasksRequest.supported_levels:type_name -> proxy_pool.controlplane.v1.CheckLevel + 2, // 30: proxy_pool.controlplane.v1.CheckTask.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol + 4, // 31: proxy_pool.controlplane.v1.CheckTask.level:type_name -> proxy_pool.controlplane.v1.CheckLevel + 31, // 32: proxy_pool.controlplane.v1.CheckTask.timeout:type_name -> google.protobuf.Duration + 32, // 33: proxy_pool.controlplane.v1.CheckTask.deadline:type_name -> google.protobuf.Timestamp + 25, // 34: proxy_pool.controlplane.v1.ObservationBatch.observations:type_name -> proxy_pool.controlplane.v1.HealthObservation + 4, // 35: proxy_pool.controlplane.v1.HealthObservation.level:type_name -> proxy_pool.controlplane.v1.CheckLevel + 31, // 36: proxy_pool.controlplane.v1.HealthObservation.latency:type_name -> google.protobuf.Duration + 32, // 37: proxy_pool.controlplane.v1.HealthObservation.observed_at:type_name -> google.protobuf.Timestamp + 5, // 38: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:input_type -> proxy_pool.controlplane.v1.RegisterWorkerRequest + 7, // 39: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:input_type -> proxy_pool.controlplane.v1.WatchSnapshotsRequest + 15, // 40: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:input_type -> proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest + 16, // 41: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:input_type -> proxy_pool.controlplane.v1.OutcomeBatch + 19, // 42: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:input_type -> proxy_pool.controlplane.v1.ReportRuntimeRequest + 22, // 43: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:input_type -> proxy_pool.controlplane.v1.StreamCheckTasksRequest + 24, // 44: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:input_type -> proxy_pool.controlplane.v1.ObservationBatch + 6, // 45: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:output_type -> proxy_pool.controlplane.v1.RegisterWorkerResponse + 8, // 46: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:output_type -> proxy_pool.controlplane.v1.SnapshotEnvelope + 33, // 47: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:output_type -> google.protobuf.Empty + 18, // 48: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:output_type -> proxy_pool.controlplane.v1.ReportOutcomesResponse + 21, // 49: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:output_type -> proxy_pool.controlplane.v1.ReportRuntimeResponse + 23, // 50: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:output_type -> proxy_pool.controlplane.v1.CheckTask + 26, // 51: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:output_type -> proxy_pool.controlplane.v1.ReportObservationsResponse + 45, // [45:52] is the sub-list for method output_type + 38, // [38:45] is the sub-list for method input_type + 38, // [38:38] is the sub-list for extension type_name + 38, // [38:38] is the sub-list for extension extendee + 0, // [0:38] is the sub-list for field type_name } func init() { file_controlplane_v1_controlplane_proto_init() } @@ -2405,7 +2492,7 @@ func file_controlplane_v1_controlplane_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_controlplane_v1_controlplane_proto_rawDesc), len(file_controlplane_v1_controlplane_proto_rawDesc)), NumEnums: 5, - NumMessages: 25, + NumMessages: 26, NumExtensions: 0, NumServices: 2, }, diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index 2075a83..949368b 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -257,6 +257,7 @@ func runWithWorkerFactory( MaxSnapshotBytes: loaded.Value.ControlPlane.MaxMessageBytes, SnapshotReader: snapshotReader, RoutingSource: routingSource, + Credentials: opened.credentials, }) if serviceErr != nil { return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr) diff --git a/internal/controller/worker/owned_snapshot_source.go b/internal/controller/worker/owned_snapshot_source.go index 8aa13ea..564c53a 100644 --- a/internal/controller/worker/owned_snapshot_source.go +++ b/internal/controller/worker/owned_snapshot_source.go @@ -4,6 +4,7 @@ import ( "context" "errors" "math" + "sort" "time" controlplanev1 "proxy-pool/gen/controlplane/v1" @@ -11,6 +12,7 @@ import ( ownershipDomain "proxy-pool/internal/domain/ownership" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/workerruntime" + platformCredentials "proxy-pool/internal/platform/credentials" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" @@ -21,13 +23,18 @@ var ErrSnapshotCredentialsUnavailable = errors.New("worker snapshot credential m // OwnedSnapshotSource builds a complete Worker view from the Redis ownership // index. It only accepts a complete bounded result from the reader. type OwnedSnapshotSource struct { - epochs OwnershipEpochReader - reader ownershipDomain.SnapshotReader - validFor time.Duration - maxProxies int - maxBytes int - now func() time.Time - routing RoutingSource + epochs OwnershipEpochReader + reader ownershipDomain.SnapshotReader + validFor time.Duration + maxProxies int + maxBytes int + now func() time.Time + routing RoutingSource + credentials CredentialReader +} + +type CredentialReader interface { + Resolve(context.Context, platformCredentials.Reference) (platformCredentials.Value, error) } func NewOwnedSnapshotSource( @@ -38,6 +45,35 @@ func NewOwnedSnapshotSource( maxBytes int, now func() time.Time, routing ...RoutingSource, +) (*OwnedSnapshotSource, error) { + return newOwnedSnapshotSource(epochs, reader, validFor, maxProxies, maxBytes, now, nil, routing...) +} + +func NewOwnedSnapshotSourceWithCredentials( + epochs OwnershipEpochReader, + reader ownershipDomain.SnapshotReader, + validFor time.Duration, + maxProxies int, + maxBytes int, + now func() time.Time, + credentials CredentialReader, + routing ...RoutingSource, +) (*OwnedSnapshotSource, error) { + if credentials == nil { + return nil, ErrSnapshotsUnavailable + } + return newOwnedSnapshotSource(epochs, reader, validFor, maxProxies, maxBytes, now, credentials, routing...) +} + +func newOwnedSnapshotSource( + epochs OwnershipEpochReader, + reader ownershipDomain.SnapshotReader, + validFor time.Duration, + maxProxies int, + maxBytes int, + now func() time.Time, + credentials CredentialReader, + routing ...RoutingSource, ) (*OwnedSnapshotSource, error) { if epochs == nil || reader == nil || validFor <= 0 || maxProxies <= 0 || maxBytes <= 0 || now == nil || len(routing) > 1 || len(routing) == 1 && routing[0] == nil { @@ -45,6 +81,7 @@ func NewOwnedSnapshotSource( } source := &OwnedSnapshotSource{ epochs: epochs, reader: reader, validFor: validFor, maxProxies: maxProxies, maxBytes: maxBytes, now: now, + credentials: credentials, } if len(routing) == 1 { source.routing = routing[0] @@ -77,6 +114,7 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(source.validFor)), Proxies: make([]*controlplanev1.OwnedProxy, 0, len(proxies)), } + materials := make(map[platformCredentials.Reference]platformCredentials.Value) for _, item := range proxies { owned, validUntil, err := wireOwnedProxy(item, now) if err != nil { @@ -86,7 +124,26 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa full.ValidUntil = timestamppb.New(validUntil) } full.Proxies = append(full.Proxies, owned) + if owned.GetSecretRef() != "" { + reference := platformCredentials.Reference{SecretRef: owned.GetSecretRef(), CredentialVersion: owned.GetCredentialVersion()} + if source.credentials == nil { + return nil, ErrSnapshotCredentialsUnavailable + } + material, resolveErr := source.credentials.Resolve(ctx, reference) + if resolveErr != nil || (material.Username == "" && material.Password == "") || + (owned.GetUsername() != "" && material.Username != "" && material.Username != owned.GetUsername()) { + return nil, ErrSnapshotCredentialsUnavailable + } + if material.Username == "" { + material.Username = owned.GetUsername() + } + if existing, exists := materials[reference]; exists && existing != material { + return nil, ErrSnapshotCredentialsUnavailable + } + materials[reference] = material + } } + full.Credentials = wireSnapshotCredentials(materials) if source.routing != nil { routing, err := source.routing.Read(ctx) if err != nil { @@ -107,6 +164,23 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa return updates, nil } +func wireSnapshotCredentials(materials map[platformCredentials.Reference]platformCredentials.Value) []*controlplanev1.SnapshotCredential { + result := make([]*controlplanev1.SnapshotCredential, 0, len(materials)) + for reference, material := range materials { + result = append(result, &controlplanev1.SnapshotCredential{ + SecretRef: reference.SecretRef, CredentialVersion: reference.CredentialVersion, + Username: material.Username, Password: material.Password, + }) + } + sort.Slice(result, func(left, right int) bool { + if result[left].GetSecretRef() == result[right].GetSecretRef() { + return result[left].GetCredentialVersion() < result[right].GetCredentialVersion() + } + return result[left].GetSecretRef() < result[right].GetSecretRef() + }) + return result +} + func cloneSnapshotRouting(source []*controlplanev1.RoutingRule) []*controlplanev1.RoutingRule { result := make([]*controlplanev1.RoutingRule, len(source)) for index, rule := range source { @@ -119,7 +193,7 @@ func cloneSnapshotRouting(source []*controlplanev1.RoutingRule) []*controlplanev func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*controlplanev1.OwnedProxy, time.Time, error) { proxy := item.Proxy - if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || proxy.CredentialVersion != "" || proxy.SecretRef != "" { + if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || (proxy.CredentialVersion == "") != (proxy.SecretRef == "") { if proxy.CredentialVersion != "" || proxy.SecretRef != "" { return nil, time.Time{}, ErrSnapshotCredentialsUnavailable } @@ -144,7 +218,7 @@ func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*control } return &controlplanev1.OwnedProxy{ Id: proxy.ID, Upstream: proxy.SourceUpstream, Protocol: protocol, Host: proxy.Host, Port: uint32(proxy.Port), - Username: proxy.Username, ExpiresAt: timestamppb.New(proxy.ExpiresAt.UTC()), + Username: proxy.Username, CredentialVersion: proxy.CredentialVersion, SecretRef: proxy.SecretRef, ExpiresAt: timestamppb.New(proxy.ExpiresAt.UTC()), MaxConcurrency: uint32(proxy.MaxConcurrency), Tags: tags, OwnershipEpoch: item.OwnershipEpoch, UsableUntil: timestamppb.New(validUntil), }, validUntil, nil diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index 135217d..adfdb7f 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -10,6 +10,7 @@ import ( ownershipDomain "proxy-pool/internal/domain/ownership" "proxy-pool/internal/domain/workerruntime" + platformCredentials "proxy-pool/internal/platform/credentials" ) var ( @@ -59,6 +60,7 @@ type Options struct { MaxSnapshotBytes int SnapshotReader ownershipDomain.SnapshotReader RoutingSource RoutingSource + Credentials platformCredentials.Store SessionID func() (string, error) } @@ -136,9 +138,17 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err if options.RoutingSource != nil { routing = append(routing, options.RoutingSource) } - source, err := NewOwnedSnapshotSource( - result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, routing..., - ) + var source SnapshotSource + var err error + if options.Credentials != nil { + source, err = NewOwnedSnapshotSourceWithCredentials( + result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, options.Credentials, routing..., + ) + } else { + source, err = NewOwnedSnapshotSource( + result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, routing..., + ) + } if err != nil { return nil, errors.Join(ErrInvalidCommand, err) } diff --git a/internal/controller/worker/snapshot_source_test.go b/internal/controller/worker/snapshot_source_test.go index 0a45a25..1b92840 100644 --- a/internal/controller/worker/snapshot_source_test.go +++ b/internal/controller/worker/snapshot_source_test.go @@ -10,6 +10,7 @@ import ( "proxy-pool/internal/controlplane/snapshotwire" ownershipDomain "proxy-pool/internal/domain/ownership" proxyDomain "proxy-pool/internal/domain/proxy" + platformCredentials "proxy-pool/internal/platform/credentials" ) func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) { @@ -123,6 +124,44 @@ func TestOwnedSnapshotSourceRejectsCredentialReferenceUntilMaterialIsAvailable(t } } +func TestOwnedSnapshotSourceIncludesReferencedCredentialMaterial(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + expiresAt := now.Add(time.Minute) + usableUntil := now.Add(30 * time.Second) + store, err := platformCredentials.NewMemoryStore(2) + if err != nil { + t.Fatalf("NewMemoryStore(): %v", err) + } + reference, err := store.Put(context.Background(), "provider-a", platformCredentials.Value{Username: "upstream", Password: "secret"}) + if err != nil { + t.Fatalf("Put(): %v", err) + } + source, err := NewOwnedSnapshotSourceWithCredentials(epochReaderStub{epoch: 1}, snapshotReaderStub{proxies: []ownershipDomain.SnapshotProxy{{ + Proxy: proxyDomain.Proxy{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, SourceUpstream: "upstream-a", + Username: "upstream", SecretRef: reference.SecretRef, CredentialVersion: reference.CredentialVersion, + ExpiresAt: &expiresAt, UsableUntil: &usableUntil, MaxConcurrency: 1, State: proxyDomain.StateAvailable, + }, OwnershipEpoch: 1, LeaseExpiresAt: now.Add(time.Minute), + }}}, time.Minute, 10, 4096, func() time.Time { return now }, store) + if err != nil { + t.Fatalf("NewOwnedSnapshotSourceWithCredentials(): %v", err) + } + updates, err := source.Watch(context.Background(), SnapshotWatchRequest{WorkerID: "worker-a", SessionID: "session-a"}) + if err != nil { + t.Fatalf("Watch(): %v", err) + } + full := <-updates + if len(full.GetCredentials()) != 1 || full.GetCredentials()[0].GetSecretRef() != reference.SecretRef || + full.GetCredentials()[0].GetUsername() != "upstream" || full.GetCredentials()[0].GetPassword() != "secret" || + full.GetProxies()[0].GetCredentialVersion() != reference.CredentialVersion { + t.Fatal("snapshot credential material or proxy credential reference is invalid") + } + checksum, err := snapshotwire.Checksum(full) + if err != nil || string(checksum[:]) != string(full.GetChecksum()) { + t.Fatalf("snapshot checksum = %x, %v; want %x", full.GetChecksum(), err, checksum) + } +} + type epochReaderStub struct { epoch uint64 err error diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go index cff1911..351a4e6 100644 --- a/internal/gateway/bootstrap/bootstrap.go +++ b/internal/gateway/bootstrap/bootstrap.go @@ -23,6 +23,7 @@ import ( controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/config" + proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/gateway/controlplane" "proxy-pool/internal/gateway/dispatch" @@ -145,7 +146,7 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio if err != nil { return nil, fmt.Errorf("build gateway target policy: %w", err) } - proxyTransport := transport.New(transport.Config{}, nil) + proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store}) handler, err := server.New(server.ConfigFromListener(configuration.Gateway), server.Dependencies{ Auth: protection.Auth, Access: protection.Access, @@ -222,6 +223,18 @@ type generatedRuntimeClient struct { client controlplanev1.WorkerControlPlaneClient } +type snapshotCredentialResolver struct { + store *snapshot.Store +} + +func (resolver snapshotCredentialResolver) Resolve(ctx context.Context, selected proxyDomain.Proxy) (transport.Credentials, error) { + credential, err := resolver.store.Credential(ctx, selected) + if err != nil { + return transport.Credentials{}, err + } + return transport.Credentials{Username: credential.Username, Password: credential.Password}, nil +} + func (client generatedRuntimeClient) RegisterWorker( ctx context.Context, request *controlplanev1.RegisterWorkerRequest, diff --git a/internal/gateway/controlplane/watcher.go b/internal/gateway/controlplane/watcher.go index 967e93c..a921d7f 100644 --- a/internal/gateway/controlplane/watcher.go +++ b/internal/gateway/controlplane/watcher.go @@ -126,6 +126,10 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e if err != nil { return err } + credentials, err := wireSnapshotCredentials(full.GetCredentials(), proxies) + if err != nil { + return err + } routes, err := wireRouting(full.GetRouting()) if err != nil { return err @@ -133,12 +137,42 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e envelope := snapshot.Envelope{ ClusterID: watcher.options.ClusterID, WorkerID: watcher.options.WorkerID, Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, ValidUntil: validUntil, - Proxies: proxies, Routing: routes, + Proxies: proxies, Routing: routes, Credentials: credentials, } - envelope.Checksum = snapshot.ChecksumWithRouting(proxies, routes) + envelope.Checksum = snapshot.ChecksumWithCredentials(proxies, routes, credentials) return watcher.store.Apply(envelope) } +func wireSnapshotCredentials(source []*controlplanev1.SnapshotCredential, proxies []proxyDomain.Proxy) ([]snapshot.Credential, error) { + result := make([]snapshot.Credential, len(source)) + indexed := make(map[string]struct{}, len(source)) + for index, item := range source { + if item == nil || !workerruntime.ValidIdentifier(item.GetSecretRef()) || !workerruntime.ValidIdentifier(item.GetCredentialVersion()) || + (item.GetUsername() == "" && item.GetPassword() == "") { + return nil, ErrInvalidSnapshotWatcher + } + key := item.GetSecretRef() + "\x00" + item.GetCredentialVersion() + if _, duplicate := indexed[key]; duplicate { + return nil, ErrInvalidSnapshotWatcher + } + indexed[key] = struct{}{} + result[index] = snapshot.Credential{ + SecretRef: item.GetSecretRef(), CredentialVersion: item.GetCredentialVersion(), + Username: item.GetUsername(), Password: item.GetPassword(), + } + } + for _, proxy := range proxies { + if proxy.SecretRef == "" && proxy.CredentialVersion == "" { + continue + } + key := proxy.SecretRef + "\x00" + proxy.CredentialVersion + if _, exists := indexed[key]; !exists { + return nil, ErrInvalidSnapshotWatcher + } + } + return result, nil +} + func wireRouting(source []*controlplanev1.RoutingRule) ([]routingDomain.Rule, error) { result := make([]routingDomain.Rule, 0, len(source)) names := make(map[string]struct{}, len(source)) diff --git a/internal/gateway/controlplane/watcher_test.go b/internal/gateway/controlplane/watcher_test.go index 220fdbb..a2d178c 100644 --- a/internal/gateway/controlplane/watcher_test.go +++ b/internal/gateway/controlplane/watcher_test.go @@ -23,7 +23,11 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T) Version: 1, OwnershipEpoch: 7, GeneratedAt: timestamppb.New(time.Now()), ValidUntil: timestamppb.New(time.Now().Add(time.Minute)), Proxies: []*controlplanev1.OwnedProxy{{ Id: "proxy-a", Upstream: "upstream-a", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, - Host: "192.0.2.10", Port: 8080, MaxConcurrency: 3, ExpiresAt: timestamppb.New(time.Now().Add(time.Minute)), + Host: "192.0.2.10", Port: 8080, Username: "upstream", SecretRef: "cred_a", CredentialVersion: "v1", + MaxConcurrency: 3, ExpiresAt: timestamppb.New(time.Now().Add(time.Minute)), + }}, + Credentials: []*controlplanev1.SnapshotCredential{{ + SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream", Password: "secret", }}, Routing: []*controlplanev1.RoutingRule{{ Name: "gateway-api", Enabled: true, HostRegex: `^api\.example\.test$`, Upstreams: []string{"upstream-a"}, @@ -54,6 +58,10 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T) if client.watch.GetSessionId() != "session-a" || client.ack.GetVersion() != 1 || !client.ack.GetApplied() || string(client.ack.GetChecksum()) != string(full.GetChecksum()) { t.Fatalf("watch=%+v ack=%+v", client.watch, client.ack) } + credential, err := store.Credential(context.Background(), view.Entries[0].Proxy) + if err != nil || credential.Username != "upstream" || credential.Password != "secret" { + t.Fatalf("Credential() = (%+v, %v)", credential, err) + } } func TestSnapshotWatcherRejectsChecksumAndAcknowledgesFailure(t *testing.T) { diff --git a/internal/gateway/server/e2e_test.go b/internal/gateway/server/e2e_test.go index e02900b..32dd126 100644 --- a/internal/gateway/server/e2e_test.go +++ b/internal/gateway/server/e2e_test.go @@ -277,7 +277,6 @@ func proxyDescriptor(t *testing.T, id, rawURL string) proxyDomain.Proxy { SourceUpstream: "provider-a", MaxConcurrency: 2, State: proxyDomain.StateAvailable, - CredentialVersion: "v1", } } diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index 886eda9..f575f62 100644 --- a/internal/gateway/server/handler_test.go +++ b/internal/gateway/server/handler_test.go @@ -595,7 +595,6 @@ func dispatcherWithProxies(t *testing.T, ids ...string) (*dispatch.Dispatcher, * SourceUpstream: "provider-a", MaxConcurrency: 2, State: proxyDomain.StateAvailable, - CredentialVersion: "v1", }) } envelope := snapshot.Envelope{ diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index 6fb2df7..9032e72 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -1,6 +1,7 @@ package snapshot import ( + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -24,6 +25,8 @@ var ( ErrInvalidRuntimeReport = errors.New("invalid worker runtime report") ErrInvalidRuntimeLimit = errors.New("invalid snapshot runtime limit") ErrRuntimeLimitExceeded = errors.New("snapshot runtime limit exceeded") + ErrInvalidCredential = errors.New("invalid snapshot credential") + ErrCredentialMissing = errors.New("snapshot credential is missing") ) const defaultRuntimeLimit = 1_000_000 @@ -93,15 +96,29 @@ type runtimeRegistration struct { } type Envelope struct { - ClusterID string - WorkerID string - Epoch uint64 - Version uint64 - Full bool - Checksum string - ValidUntil time.Time - Proxies []proxyDomain.Proxy - Routing []routing.Rule + ClusterID string + WorkerID string + Epoch uint64 + Version uint64 + Full bool + Checksum string + ValidUntil time.Time + Proxies []proxyDomain.Proxy + Routing []routing.Rule + Credentials []Credential +} + +// Credential is ephemeral material received through a verified mTLS snapshot. +// Its Format method prevents accidental diagnostics from exposing either value. +type Credential struct { + SecretRef string + CredentialVersion string + Username string + Password string +} + +func (Credential) Format(state fmt.State, _ rune) { + _, _ = state.Write([]byte("snapshot.Credential{SecretRef:, CredentialVersion:, Username:, Password:}")) } type Entry struct { @@ -110,14 +127,15 @@ type Entry struct { } type View struct { - ClusterID string - WorkerID string - Epoch uint64 - Version uint64 - Checksum string - ValidUntil time.Time - Entries []Entry - routing *routing.RuleSet + ClusterID string + WorkerID string + Epoch uint64 + Version uint64 + Checksum string + ValidUntil time.Time + Entries []Entry + routing *routing.RuleSet + credentials map[string]Credential all []int byScheme map[proxyDomain.Scheme][]int @@ -229,13 +247,17 @@ func (s *Store) Apply(envelope Envelope) error { if !envelope.Full || envelope.Epoch == 0 || envelope.Version == 0 { return ErrResyncRequired } - if envelope.Checksum != ChecksumWithRouting(envelope.Proxies, envelope.Routing) { + if envelope.Checksum != ChecksumWithCredentials(envelope.Proxies, envelope.Routing, envelope.Credentials) { return ErrChecksumMismatch } routes, err := routing.Compile(envelope.Routing) if err != nil { return fmt.Errorf("compile snapshot routing: %w", err) } + credentials, err := indexCredentials(envelope.Credentials, envelope.Proxies) + if err != nil { + return err + } s.mu.Lock() defer s.mu.Unlock() @@ -315,20 +337,56 @@ func (s *Store) Apply(envelope Envelope) error { } next := &View{ - ClusterID: envelope.ClusterID, - WorkerID: envelope.WorkerID, - Epoch: envelope.Epoch, - Version: envelope.Version, - Checksum: envelope.Checksum, - ValidUntil: envelope.ValidUntil.UTC(), - Entries: entries, - routing: routes, + ClusterID: envelope.ClusterID, + WorkerID: envelope.WorkerID, + Epoch: envelope.Epoch, + Version: envelope.Version, + Checksum: envelope.Checksum, + ValidUntil: envelope.ValidUntil.UTC(), + Entries: entries, + routing: routes, + credentials: credentials, } next.buildIndexes() s.current.Store(next) return nil } +// Credential resolves one exact Proxy credential reference from the current +// immutable view. Credentialless proxies preserve their metadata username. +func (s *Store) Credential(ctx context.Context, selected proxyDomain.Proxy) (Credential, error) { + if ctx == nil { + return Credential{}, context.Canceled + } + if err := ctx.Err(); err != nil { + return Credential{}, err + } + if (selected.SecretRef == "") != (selected.CredentialVersion == "") { + return Credential{}, ErrInvalidCredential + } + if selected.SecretRef == "" { + return Credential{Username: selected.Username}, nil + } + if s == nil { + return Credential{}, ErrCredentialMissing + } + current := s.current.Load() + if current == nil { + return Credential{}, ErrCredentialMissing + } + credential, ok := current.credentials[credentialKey(selected.SecretRef, selected.CredentialVersion)] + if !ok { + return Credential{}, ErrCredentialMissing + } + if selected.Username != "" && credential.Username != "" && selected.Username != credential.Username { + return Credential{}, ErrInvalidCredential + } + if credential.Username == "" { + credential.Username = selected.Username + } + return credential, nil +} + // MatchRouting matches a request against the immutable rules published with // this proxy view. A View is the atomic consistency boundary for both sets. func (v *View) MatchRouting(request routing.Request) (routing.Rule, bool) { @@ -427,12 +485,18 @@ func Checksum(proxies []proxyDomain.Proxy) string { } func ChecksumWithRouting(proxies []proxyDomain.Proxy, rules []routing.Rule) string { + return ChecksumWithCredentials(proxies, rules, nil) +} + +func ChecksumWithCredentials(proxies []proxyDomain.Proxy, rules []routing.Rule, credentials []Credential) string { payload := struct { - Proxies []proxyDomain.Proxy `json:"proxies"` - Routing []routing.Rule `json:"routing"` + Proxies []proxyDomain.Proxy `json:"proxies"` + Routing []routing.Rule `json:"routing"` + Credentials []Credential `json:"credentials"` }{ - Proxies: cloneAndSort(proxies), - Routing: cloneRoutingRules(rules), + Proxies: cloneAndSort(proxies), + Routing: cloneRoutingRules(rules), + Credentials: cloneAndSortCredentials(credentials), } encoded, err := json.Marshal(payload) if err != nil { @@ -442,6 +506,46 @@ func ChecksumWithRouting(proxies []proxyDomain.Proxy, rules []routing.Rule) stri return hex.EncodeToString(digest[:]) } +func indexCredentials(source []Credential, proxies []proxyDomain.Proxy) (map[string]Credential, error) { + indexed := make(map[string]Credential, len(source)) + for _, credential := range source { + if credential.SecretRef == "" || credential.CredentialVersion == "" || (credential.Username == "" && credential.Password == "") { + return nil, ErrInvalidCredential + } + key := credentialKey(credential.SecretRef, credential.CredentialVersion) + if _, duplicate := indexed[key]; duplicate { + return nil, ErrInvalidCredential + } + indexed[key] = credential + } + for _, proxy := range proxies { + if (proxy.SecretRef == "") != (proxy.CredentialVersion == "") { + return nil, ErrInvalidCredential + } + if proxy.SecretRef == "" { + continue + } + credential, exists := indexed[credentialKey(proxy.SecretRef, proxy.CredentialVersion)] + if !exists || (proxy.Username != "" && credential.Username != "" && proxy.Username != credential.Username) { + return nil, ErrInvalidCredential + } + } + return indexed, nil +} + +func credentialKey(secretRef, version string) string { return secretRef + "\x00" + version } + +func cloneAndSortCredentials(source []Credential) []Credential { + cloned := append([]Credential(nil), source...) + sort.Slice(cloned, func(left, right int) bool { + if cloned[left].SecretRef == cloned[right].SecretRef { + return cloned[left].CredentialVersion < cloned[right].CredentialVersion + } + return cloned[left].SecretRef < cloned[right].SecretRef + }) + return cloned +} + func cloneRoutingRules(source []routing.Rule) []routing.Rule { result := make([]routing.Rule, len(source)) for index, rule := range source { diff --git a/internal/gateway/snapshot/store_test.go b/internal/gateway/snapshot/store_test.go index b44239b..01dad2f 100644 --- a/internal/gateway/snapshot/store_test.go +++ b/internal/gateway/snapshot/store_test.go @@ -1,9 +1,11 @@ package snapshot import ( + "context" "errors" "fmt" "reflect" + "strings" "testing" "time" @@ -126,6 +128,35 @@ func TestStorePublishesRoutingWithTheSameSnapshotVersion(t *testing.T) { } } +func TestStoreResolvesCredentialOnlyFromCurrentSnapshot(t *testing.T) { + store := NewStore("cluster-a", "worker-a") + proxy := proxyDomain.Proxy{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, MaxConcurrency: 1, + SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream", + } + envelope := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, Proxies: []proxyDomain.Proxy{proxy}, + Credentials: []Credential{{SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream", Password: "secret"}}, + } + envelope.Checksum = ChecksumWithCredentials(envelope.Proxies, envelope.Routing, envelope.Credentials) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(): %v", err) + } + credential, err := store.Credential(context.Background(), proxy) + if err != nil || credential.Username != "upstream" || credential.Password != "secret" { + t.Fatalf("Credential() = (%+v, %v)", credential, err) + } + if formatted := fmt.Sprintf("%+v", credential); strings.Contains(formatted, "secret") || strings.Contains(formatted, "upstream") { + t.Fatalf("formatted credential exposes material: %s", formatted) + } + + missing := Envelope{ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 2, Full: true, Proxies: []proxyDomain.Proxy{proxy}} + missing.Checksum = ChecksumWithCredentials(missing.Proxies, nil, nil) + if err := store.Apply(missing); !errors.Is(err, ErrInvalidCredential) { + t.Fatalf("Apply(missing credential) error = %v, want ErrInvalidCredential", err) + } +} + func TestViewSelectFiltersBySchemeUpstreamTagAndExclude(t *testing.T) { now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) expiresSoon := now.Add(5 * time.Second)