From 2bdc1ebda3e3b443328d0effee473d00ea7c6b7c Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 21:12:32 +0800 Subject: [PATCH] feat: add checker task executor --- README.md | 15 +- cmd/proxy-checker/main.go | 123 +++++++++ cmd/proxy-checker/main_test.go | 35 +++ docs/api/control-plane.md | 18 ++ docs/configuration/reference.md | 13 + docs/design/project-structure.md | 5 +- docs/development/implementation-plan.md | 4 +- internal/checker/bootstrap/bootstrap.go | 120 +++++++++ .../checker/controlplane/generated_client.go | 40 +++ internal/checker/controlplane/runner.go | 254 ++++++++++++++++++ internal/checker/controlplane/runner_test.go | 98 +++++++ internal/checker/probe/probe.go | 156 +++++++++++ internal/checker/probe/probe_test.go | 85 ++++++ internal/config/config.go | 14 +- internal/config/config_test.go | 8 + internal/config/validate.go | 11 +- .../controlplane/clienttransport/transport.go | 95 +++++++ internal/gateway/bootstrap/bootstrap.go | 68 +---- 18 files changed, 1089 insertions(+), 73 deletions(-) create mode 100644 cmd/proxy-checker/main.go create mode 100644 cmd/proxy-checker/main_test.go create mode 100644 internal/checker/bootstrap/bootstrap.go create mode 100644 internal/checker/controlplane/generated_client.go create mode 100644 internal/checker/controlplane/runner.go create mode 100644 internal/checker/controlplane/runner_test.go create mode 100644 internal/checker/probe/probe.go create mode 100644 internal/checker/probe/probe_test.go create mode 100644 internal/controlplane/clienttransport/transport.go diff --git a/README.md b/README.md index 0bdef02..57779df 100644 --- a/README.md +++ b/README.md @@ -153,7 +153,20 @@ go run ./cmd/proxy-gateway -config CONFIG_FILE ` 以上参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、`PROXY_POOL_CLUSTER_ID`、 `PROXY_POOL_WORKER_ID`、`PROXY_POOL_INSTANCE_ID` 与 `PROXY_POOL_ZONE` 提供。 Gateway 的 `/livez`、`/readyz`、`/metrics` 使用配置中的 `metrics.listen`；无有效 -Snapshot 时 `/readyz` 返回 `503`。Checker 与 loadgen 命令尚未实现。 +Snapshot 时 `/readyz` 返回 `503`。Checker 使用独立的逻辑/实例身份拉取有界任务： + +```powershell +go run ./cmd/proxy-checker -config CONFIG_FILE ` + -control-plane CONTROLLER_HOST:8443 ` + -checker-id CHECKER_ID -instance-id INSTANCE_ID ` + -max-in-flight 64 +``` + +Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 +`PROXY_POOL_CHECKER_ID`、`PROXY_POOL_CHECKER_INSTANCE_ID` 与 +`PROXY_POOL_CHECKER_MAX_IN_FLIGHT` 提供。它不会访问 Redis/PostgreSQL；当前生产 +Controller 尚未装配 Redis 共享任务 broker，任务流会返回 `Unavailable`，直到后续 +调度阶段接入。loadgen 命令尚未实现。 ## 关键配置与入口 diff --git a/cmd/proxy-checker/main.go b/cmd/proxy-checker/main.go new file mode 100644 index 0000000..bfd1e14 --- /dev/null +++ b/cmd/proxy-checker/main.go @@ -0,0 +1,123 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/checker/bootstrap" + "proxy-pool/internal/config" +) + +const ( + configEnvironment = "PROXY_POOL_CONFIG" + controlPlaneAddressEnvironment = "PROXY_POOL_CONTROL_PLANE_ADDRESS" + checkerIDEnvironment = "PROXY_POOL_CHECKER_ID" + instanceIDEnvironment = "PROXY_POOL_CHECKER_INSTANCE_ID" + maxInFlightEnvironment = "PROXY_POOL_CHECKER_MAX_IN_FLIGHT" +) + +type environmentLookup func(string) string +type checkerRun func(context.Context, bootstrap.Options) error + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + os.Exit(execute(ctx, os.Args[1:], os.Getenv, bootstrap.Run, os.Stderr)) +} + +func execute(ctx context.Context, args []string, getenv environmentLookup, run checkerRun, stderr io.Writer) int { + flags := flag.NewFlagSet("proxy-checker", flag.ContinueOnError) + flags.SetOutput(stderr) + configPath := flags.String("config", "", "configuration file path") + controlPlaneAddress := flags.String("control-plane", "", "remote Controller control-plane address") + checkerID := flags.String("checker-id", "", "unique Checker identifier") + instanceID := flags.String("instance-id", "", "unique Checker process instance identifier") + maxInFlight := flags.Int("max-in-flight", 0, "maximum concurrent tasks") + levels := flags.String("levels", "basic,egress,target", "supported levels: basic,egress,target") + if err := flags.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return 0 + } + return 2 + } + if flags.NArg() != 0 { + _, _ = fmt.Fprintln(stderr, "proxy-checker: unexpected positional arguments") + return 2 + } + if getenv != nil { + setIfEmpty(configPath, getenv(configEnvironment)) + setIfEmpty(controlPlaneAddress, getenv(controlPlaneAddressEnvironment)) + setIfEmpty(checkerID, getenv(checkerIDEnvironment)) + setIfEmpty(instanceID, getenv(instanceIDEnvironment)) + if *maxInFlight == 0 { + if value, err := strconv.Atoi(getenv(maxInFlightEnvironment)); err == nil { + *maxInFlight = value + } + } + } + supportedLevels, err := parseLevels(*levels) + if ctx == nil || run == nil || err != nil || !validValue(*configPath) || !validValue(*controlPlaneAddress) || + !validValue(*checkerID) || !validValue(*instanceID) || *maxInFlight <= 0 { + _, _ = fmt.Fprintf(stderr, + "proxy-checker: -config, -control-plane, -checker-id, -instance-id and positive -max-in-flight are required; "+ + "environment fallbacks: %s, %s, %s, %s, %s\n", + configEnvironment, controlPlaneAddressEnvironment, checkerIDEnvironment, instanceIDEnvironment, maxInFlightEnvironment, + ) + return 2 + } + err = run(ctx, bootstrap.Options{ + ConfigPath: *configPath, Resolver: config.OSResolver{}, ControlPlaneAddress: *controlPlaneAddress, + CheckerID: *checkerID, InstanceID: *instanceID, MaxInFlight: *maxInFlight, SupportedLevels: supportedLevels, + }) + if err == nil || (errors.Is(err, context.Canceled) && ctx.Err() != nil) { + return 0 + } + _, _ = fmt.Fprintf(stderr, "proxy-checker: %v\n", err) + return 1 +} + +func parseLevels(value string) ([]controlplanev1.CheckLevel, error) { + levels := make([]controlplanev1.CheckLevel, 0, 3) + seen := make(map[controlplanev1.CheckLevel]struct{}) + for _, item := range strings.Split(value, ",") { + var level controlplanev1.CheckLevel + switch strings.ToLower(strings.TrimSpace(item)) { + case "basic": + level = controlplanev1.CheckLevel_CHECK_LEVEL_BASIC + case "egress": + level = controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS + case "target": + level = controlplanev1.CheckLevel_CHECK_LEVEL_TARGET + default: + return nil, errors.New("invalid checker level") + } + if _, duplicate := seen[level]; duplicate { + return nil, errors.New("duplicate checker level") + } + seen[level] = struct{}{} + levels = append(levels, level) + } + if len(levels) == 0 { + return nil, errors.New("empty checker level set") + } + return levels, nil +} + +func setIfEmpty(target *string, value string) { + if target != nil && *target == "" { + *target = value + } +} + +func validValue(value string) bool { + return value != "" && strings.TrimSpace(value) == value +} diff --git a/cmd/proxy-checker/main_test.go b/cmd/proxy-checker/main_test.go new file mode 100644 index 0000000..9097b08 --- /dev/null +++ b/cmd/proxy-checker/main_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "context" + "io" + "testing" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/checker/bootstrap" +) + +func TestExecuteUsesFlagsAndPassesCheckerIdentity(t *testing.T) { + var received bootstrap.Options + code := execute(context.Background(), []string{ + "-config", "config.yaml", "-control-plane", "127.0.0.1:8443", "-checker-id", "checker-a", "-instance-id", "instance-a", + "-max-in-flight", "2", "-levels", "basic,target", + }, nil, func(_ context.Context, options bootstrap.Options) error { + received = options + return nil + }, io.Discard) + if code != 0 || received.CheckerID != "checker-a" || received.MaxInFlight != 2 || len(received.SupportedLevels) != 2 || + received.SupportedLevels[1] != controlplanev1.CheckLevel_CHECK_LEVEL_TARGET { + t.Fatalf("execute() code=%d options=%+v", code, received) + } +} + +func TestExecuteRejectsInvalidLevelSet(t *testing.T) { + code := execute(context.Background(), []string{ + "-config", "config.yaml", "-control-plane", "127.0.0.1:8443", "-checker-id", "checker-a", "-instance-id", "instance-a", + "-max-in-flight", "2", "-levels", "basic,basic", + }, nil, func(context.Context, bootstrap.Options) error { return nil }, io.Discard) + if code != 2 { + t.Fatalf("execute(invalid levels) = %d, want 2", code) + } +} diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 1207d76..a33c83e 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -172,6 +172,12 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY，并更新 Redis 活动池， 任务的同一事实可重放，由活动池摘要幂等处理。当前生产配置尚未装配 Redis 共享 broker， 因此无共享 broker 的服务会以 `Unavailable` 拒绝任务流，而不下发无租约任务。 +`proxy-checker` 使用固定大小 worker-pool 执行每个 pull 批次，任务数不超过该请求的 +`max_in_flight`；每次尝试都受 `deadline` 和 `timeout` 的较小值约束，失败可在同一 +deadline 内最多执行到 `max_attempts`。BASIC 针对 HTTP/HTTPS Proxy 验证到 Proxy 的 +请求/认证握手；TARGET 通过 Proxy 请求指定目标并将非成功状态作为事实。SOCKS5 与 +EGRESS 的专用出口语义仍待后续探测器扩展。 + ## 7. 兼容与演进 - Proto 字段号一旦发布不得复用。 @@ -205,3 +211,15 @@ Gateway 不复用 `controlPlane.listen` 作为客户端地址。`listen` 是 Con 客户端证书、私钥和 Controller CA 发起 TLS 1.3 连接；证书必须符合 Controller 的 SPIFFE Worker 身份校验。 `disabled` 仅接受回环控制面地址，供本地 fixture 使用。 + +## 10. Checker 启动参数 + +- `-control-plane` / `PROXY_POOL_CONTROL_PLANE_ADDRESS`：Controller 的可拨号地址。 +- `-checker-id` / `PROXY_POOL_CHECKER_ID`：唯一逻辑 Checker。 +- `-instance-id` / `PROXY_POOL_CHECKER_INSTANCE_ID`：唯一进程实例。 +- `-max-in-flight` / `PROXY_POOL_CHECKER_MAX_IN_FLIGHT`：本进程任务上限。 +- `-levels`：逗号分隔的 `basic,egress,target` 能力集合。 + +当 `controlPlane.tls.mode=mtls` 时，Checker 使用 `controlPlane.checkerTLS` 的独立 +客户端证书、私钥和 Controller CA 建立 TLS 1.3 连接；证书必须符合 Controller 的 +SPIFFE Checker 身份校验。`disabled` 仅接受回环控制面地址。 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index eb1f9ad..40d0345 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -171,6 +171,10 @@ controlPlane: certFile: /run/secrets/gateway-cert.pem keyFile: /run/secrets/gateway-key.pem serverCAFile: /run/secrets/controller-ca.pem + checkerTLS: + certFile: /run/secrets/checker-cert.pem + keyFile: /run/secrets/checker-key.pem + serverCAFile: /run/secrets/controller-ca.pem ``` - `protocolVersion` 当前固定为 `1`。 @@ -182,6 +186,9 @@ controlPlane: `environment`。 - `gatewayTLS` 是 Gateway 的客户端证书、私钥和 Controller CA，与 `tls` 的服务端 证书和 Worker CA 分离。三项可以同时省略（未运行 Gateway），配置任一项时必须完整提供。 +- `checkerTLS` 与 `gatewayTLS` 有相同字段和完整性校验，但必须使用独立的 Checker + 证书。Controller 分别验证 `.../worker/` 与 + `.../checker/` SPIFFE URI，不能跨角色复用证书。 Gateway 连接 Controller 时使用独立启动参数而非 `controlPlane.listen`。至少设置 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、`PROXY_POOL_CLUSTER_ID`、 @@ -190,6 +197,12 @@ Gateway 连接 Controller 时使用独立启动参数而非 `controlPlane.listen 模板保持 `controlPlane.enabled: false`，环境 Overlay 挂载 mTLS 证书并启用后才可启动 Gateway。 +Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-plane`、 +`-checker-id`、`-instance-id` 和 `-max-in-flight` 可由对应的 +`PROXY_POOL_*` 环境变量提供。mTLS 模式下该命令读取 `checkerTLS`，明文 fixture +模式只接受回环 Controller 地址。Checker 只从 gRPC 领取任务并批量上报事实，不读取 +Redis/PostgreSQL；生产 Redis 共享任务队列尚在后续实施范围。 + `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway 在本地维护容量为 `65536` 的非阻塞 Outcome 队列，默认微批上限为 `512`，实际取二者中 较小值；该队列与其序列确认状态仅存在于 Gateway 进程内。Controller 的 Redis 状态只保存 diff --git a/docs/design/project-structure.md b/docs/design/project-structure.md index c58a0c8..bddb1c1 100644 --- a/docs/design/project-structure.md +++ b/docs/design/project-structure.md @@ -46,8 +46,9 @@ Controller 是首版模块化单体。Provider、Pool、Routing 和 Extraction ### proxy-checker -Checker 只产生 Observation。最终状态迁移由 Controller 的确定性 reducer -完成，避免多个检查实例同时写 Proxy 状态。 +Checker 只产生 Observation。它从认证 gRPC 流领取有界任务，用固定 worker-pool 在任务 +deadline 内执行 HTTP/HTTPS BASIC/TARGET 探测并微批上报；最终状态迁移仍由 Controller 的 +确定性 reducer 完成，避免多个检查实例同时写 Proxy 状态。Checker 不访问 Redis 或 PostgreSQL。 ### proxy-loadgen diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index e17e6af..d10ca09 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -282,7 +282,9 @@ Checker Observation 上报 RPC 已复用既有控制面监听接入 Controller 每批上限、配置阈值解析、代理归属查询与 Reducer 均已闭环。`StreamCheckTasks` 现已实现为 有界 pull，并通过通用任务 broker 契约完成能力协商、同 Checker 并发窗口、租约到期回收、 领取者栅栏和完成后重放；任务凭据仅由认证流在执行期下发。Redis 共享 due-index/租约持久化、 -生产 broker、Checker 独立进程与探测器尚未实现，因此本任务保持未完成。 +生产 broker 已有契约但 Redis 实现尚未完成；`proxy-checker` 独立进程、固定大小 worker-pool、 +任务期重试/微批上报和 HTTP/HTTPS BASIC/TARGET 探测器已完成并有测试。SOCKS5、EGRESS +专用出口探测、Redis due-index/跨副本租约和部署运行态仍未实现，因此本任务保持未完成。 ## Task 12: Machine-readable Contracts diff --git a/internal/checker/bootstrap/bootstrap.go b/internal/checker/bootstrap/bootstrap.go new file mode 100644 index 0000000..78e7203 --- /dev/null +++ b/internal/checker/bootstrap/bootstrap.go @@ -0,0 +1,120 @@ +// Package bootstrap assembles the standalone proxy-checker process. Checker +// execution is intentionally isolated from Gateway request handling and from +// all Redis/PostgreSQL access. +package bootstrap + +import ( + "bytes" + "context" + "errors" + "fmt" + "reflect" + "strings" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/checker/controlplane" + "proxy-pool/internal/checker/probe" + "proxy-pool/internal/config" + "proxy-pool/internal/controlplane/clienttransport" + "proxy-pool/internal/domain/workerruntime" +) + +var ( + ErrInvalidOptions = errors.New("invalid checker bootstrap options") + ErrStartup = errors.New("checker startup failed") +) + +type Options struct { + ConfigPath string + Resolver config.Resolver + ControlPlaneAddress string + CheckerID string + InstanceID string + MaxInFlight int + SupportedLevels []controlplanev1.CheckLevel + GRPCTransport credentials.TransportCredentials +} + +func Run(ctx context.Context, options Options) error { + if err := validateOptions(ctx, options); err != nil { + return err + } + configuration, err := loadConfiguration(ctx, options.ConfigPath, options.Resolver) + if err != nil { + return fmt.Errorf("%w: load configuration: %w", ErrStartup, err) + } + if !configuration.ControlPlane.Enabled { + return errors.Join(ErrInvalidOptions, errors.New("controlPlane must be enabled")) + } + transport, err := clienttransport.New( + configuration.ControlPlane, options.ControlPlaneAddress, configuration.ControlPlane.CheckerTLS, options.GRPCTransport, "checkerTLS", + ) + if err != nil { + return fmt.Errorf("%w: %w", ErrInvalidOptions, err) + } + connection, err := grpc.NewClient(options.ControlPlaneAddress, + grpc.WithTransportCredentials(transport), + grpc.WithDefaultCallOptions( + grpc.MaxCallRecvMsgSize(configuration.ControlPlane.MaxMessageBytes), + grpc.MaxCallSendMsgSize(configuration.ControlPlane.MaxMessageBytes), + ), + ) + if err != nil { + return fmt.Errorf("%w: dial control plane: %w", ErrStartup, err) + } + defer connection.Close() + client := controlplane.NewGeneratedClient(controlplanev1.NewCheckerControlPlaneClient(connection)) + runner, err := controlplane.NewRunner(client, probe.NewExecutor(), controlplane.Options{ + CheckerID: options.CheckerID, InstanceID: options.InstanceID, MaxInFlight: options.MaxInFlight, + SupportedLevels: append([]controlplanev1.CheckLevel(nil), options.SupportedLevels...), + ReportBatchSize: options.MaxInFlight, + }) + if err != nil { + return fmt.Errorf("%w: build checker runner: %w", ErrStartup, err) + } + return runner.Run(ctx) +} + +func validateOptions(ctx context.Context, options Options) error { + if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" || + nilInterface(options.Resolver) || !clienttransport.ValidDialAddress(options.ControlPlaneAddress) || + !workerruntime.ValidIdentifier(options.CheckerID) || !workerruntime.ValidIdentifier(options.InstanceID) || + options.MaxInFlight <= 0 || len(options.SupportedLevels) == 0 { + return ErrInvalidOptions + } + return nil +} + +func loadConfiguration(ctx context.Context, path string, resolver config.Resolver) (*config.Config, error) { + if ctx == nil || strings.TrimSpace(path) != path || path == "" || nilInterface(resolver) { + return nil, ErrInvalidOptions + } + if err := ctx.Err(); err != nil { + return nil, err + } + content, err := resolver.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read configuration %q: %w", path, err) + } + configuration, err := config.LoadResolved(bytes.NewReader(content), resolver) + if err != nil { + return nil, fmt.Errorf("decode configuration %q: %w", path, err) + } + return configuration, nil +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/checker/controlplane/generated_client.go b/internal/checker/controlplane/generated_client.go new file mode 100644 index 0000000..06472aa --- /dev/null +++ b/internal/checker/controlplane/generated_client.go @@ -0,0 +1,40 @@ +package controlplane + +import ( + "context" + + controlplanev1 "proxy-pool/gen/controlplane/v1" +) + +type GeneratedClient struct { + client controlplanev1.CheckerControlPlaneClient +} + +func NewGeneratedClient(client controlplanev1.CheckerControlPlaneClient) *GeneratedClient { + if client == nil { + return nil + } + return &GeneratedClient{client: client} +} + +func (client *GeneratedClient) StreamCheckTasks( + ctx context.Context, + request *controlplanev1.StreamCheckTasksRequest, +) (TaskStream, error) { + if client == nil || client.client == nil { + return nil, ErrInvalidRunner + } + return client.client.StreamCheckTasks(ctx, request) +} + +func (client *GeneratedClient) ReportObservations( + ctx context.Context, + batch *controlplanev1.ObservationBatch, +) (*controlplanev1.ReportObservationsResponse, error) { + if client == nil || client.client == nil { + return nil, ErrInvalidRunner + } + return client.client.ReportObservations(ctx, batch) +} + +var _ Client = (*GeneratedClient)(nil) diff --git a/internal/checker/controlplane/runner.go b/internal/checker/controlplane/runner.go new file mode 100644 index 0000000..eb35afe --- /dev/null +++ b/internal/checker/controlplane/runner.go @@ -0,0 +1,254 @@ +// Package controlplane runs bounded Checker task pulls and fact reporting. +// It never accesses Controller storage; task lease validation stays server-side. +package controlplane + +import ( + "context" + "errors" + "io" + "sync" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/checker/probe" + "proxy-pool/internal/domain/workerruntime" + + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +var ( + ErrInvalidRunner = errors.New("invalid checker runner") + ErrTaskBatchTooLarge = errors.New("checker task batch exceeds configured concurrency") + ErrObservationsRejected = errors.New("checker observations were not accepted") +) + +const ( + defaultPollInterval = 500 * time.Millisecond + defaultRetryDelay = 20 * time.Millisecond + maximumInFlight = 4_096 +) + +type TaskStream interface { + Recv() (*controlplanev1.CheckTask, error) +} + +type Client interface { + StreamCheckTasks(context.Context, *controlplanev1.StreamCheckTasksRequest) (TaskStream, error) + ReportObservations(context.Context, *controlplanev1.ObservationBatch) (*controlplanev1.ReportObservationsResponse, error) +} + +type Executor interface { + Execute(context.Context, *controlplanev1.CheckTask) probe.Result +} + +type Options struct { + CheckerID string + InstanceID string + MaxInFlight int + SupportedLevels []controlplanev1.CheckLevel + ReportBatchSize int + PollInterval time.Duration + RetryDelay time.Duration + Now func() time.Time +} + +// Runner consumes at most one server-bounded pull per RunOnce. Run adds a +// bounded polling delay, avoiding a hot loop while no tasks are available. +type Runner struct { + client Client + executor Executor + options Options +} + +func NewRunner(client Client, executor Executor, options Options) (*Runner, error) { + if client == nil || executor == nil || !workerruntime.ValidIdentifier(options.CheckerID) || + !workerruntime.ValidIdentifier(options.InstanceID) || options.MaxInFlight <= 0 || options.MaxInFlight > maximumInFlight || + len(options.SupportedLevels) == 0 || options.Now == nil { + return nil, ErrInvalidRunner + } + if options.ReportBatchSize == 0 { + options.ReportBatchSize = options.MaxInFlight + } + if options.PollInterval == 0 { + options.PollInterval = defaultPollInterval + } + if options.RetryDelay == 0 { + options.RetryDelay = defaultRetryDelay + } + if options.ReportBatchSize <= 0 || options.ReportBatchSize > options.MaxInFlight || options.PollInterval <= 0 || options.RetryDelay < 0 || + !validLevels(options.SupportedLevels) { + return nil, ErrInvalidRunner + } + return &Runner{client: client, executor: executor, options: options}, nil +} + +func (runner *Runner) Run(ctx context.Context) error { + if ctx == nil || runner == nil { + return ErrInvalidRunner + } + for { + err := runner.RunOnce(ctx) + if err != nil && ctx.Err() != nil { + return ctx.Err() + } + wait := runner.options.PollInterval + if err != nil { + wait = runner.options.RetryDelay + } + if err := waitFor(ctx, wait); err != nil { + return err + } + } +} + +func (runner *Runner) RunOnce(ctx context.Context) error { + if ctx == nil || runner == nil || runner.client == nil || runner.executor == nil || runner.options.Now == nil { + return ErrInvalidRunner + } + if err := ctx.Err(); err != nil { + return err + } + stream, err := runner.client.StreamCheckTasks(ctx, &controlplanev1.StreamCheckTasksRequest{ + CheckerId: runner.options.CheckerID, InstanceId: runner.options.InstanceID, + MaxInFlight: uint32(runner.options.MaxInFlight), SupportedLevels: append([]controlplanev1.CheckLevel(nil), runner.options.SupportedLevels...), + }) + if err != nil { + return err + } + tasks, err := collectTasks(stream, runner.options.MaxInFlight) + if err != nil || len(tasks) == 0 { + return err + } + observations := runner.executeTasks(ctx, tasks) + for start := 0; start < len(observations); start += runner.options.ReportBatchSize { + end := start + runner.options.ReportBatchSize + if end > len(observations) { + end = len(observations) + } + response, reportErr := runner.client.ReportObservations(ctx, &controlplanev1.ObservationBatch{ + CheckerId: runner.options.CheckerID, Observations: observations[start:end], + }) + if reportErr != nil { + return reportErr + } + if response == nil || int(response.GetAccepted()) != end-start || response.GetRejected() != 0 { + return ErrObservationsRejected + } + } + return nil +} + +func collectTasks(stream TaskStream, maximum int) ([]*controlplanev1.CheckTask, error) { + if stream == nil || maximum <= 0 { + return nil, ErrInvalidRunner + } + tasks := make([]*controlplanev1.CheckTask, 0, maximum) + for { + task, err := stream.Recv() + if errors.Is(err, io.EOF) { + return tasks, nil + } + if err != nil { + return nil, err + } + if task == nil { + return nil, ErrInvalidRunner + } + tasks = append(tasks, proto.Clone(task).(*controlplanev1.CheckTask)) + if len(tasks) > maximum { + return nil, ErrTaskBatchTooLarge + } + } +} + +func (runner *Runner) executeTasks(ctx context.Context, tasks []*controlplanev1.CheckTask) []*controlplanev1.HealthObservation { + results := make([]*controlplanev1.HealthObservation, len(tasks)) + jobs := make(chan int) + var group sync.WaitGroup + workers := runner.options.MaxInFlight + if workers > len(tasks) { + workers = len(tasks) + } + for worker := 0; worker < workers; worker++ { + group.Add(1) + go func() { + defer group.Done() + for index := range jobs { + results[index] = runner.executeTask(ctx, tasks[index]) + } + }() + } + for index := range tasks { + jobs <- index + } + close(jobs) + group.Wait() + return results +} + +func (runner *Runner) executeTask(ctx context.Context, task *controlplanev1.CheckTask) *controlplanev1.HealthObservation { + result := probe.Result{FailureClass: probe.FailureInvalidTask} + if task != nil && task.GetAttempt() > 0 && task.GetMaxAttempts() >= task.GetAttempt() { + for attempt := task.GetAttempt(); attempt <= task.GetMaxAttempts(); attempt++ { + attemptTask := proto.Clone(task).(*controlplanev1.CheckTask) + attemptTask.Attempt = attempt + result = runner.executor.Execute(ctx, attemptTask) + if result.Success || ctx.Err() != nil || attempt == task.GetMaxAttempts() { + break + } + if waitFor(ctx, runner.options.RetryDelay) != nil { + break + } + } + } + if result.Latency < 0 { + result = probe.Result{FailureClass: probe.FailureInvalidTask} + } + if result.Success { + result.FailureClass = "" + } else if result.FailureClass == "" { + result.FailureClass = probe.FailureProxyRequest + } + observation := &controlplanev1.HealthObservation{ + TaskId: task.GetTaskId(), ProxyId: task.GetProxyId(), Level: task.GetLevel(), Success: result.Success, + FailureClass: result.FailureClass, Latency: durationpb.New(result.Latency), + ObservedAt: timestamppb.New(runner.options.Now().UTC()), + } + if task.GetLevel() == controlplanev1.CheckLevel_CHECK_LEVEL_TARGET { + observation.RoutingName = task.GetRoutingName() + observation.TargetUrl = task.GetTargetUrl() + } + return observation +} + +func validLevels(levels []controlplanev1.CheckLevel) bool { + seen := make(map[controlplanev1.CheckLevel]struct{}, len(levels)) + for _, level := range levels { + switch level { + case controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS, controlplanev1.CheckLevel_CHECK_LEVEL_TARGET: + default: + return false + } + if _, duplicate := seen[level]; duplicate { + return false + } + seen[level] = struct{}{} + } + return true +} + +func waitFor(ctx context.Context, duration time.Duration) error { + if duration <= 0 { + return ctx.Err() + } + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/internal/checker/controlplane/runner_test.go b/internal/checker/controlplane/runner_test.go new file mode 100644 index 0000000..5b80fd0 --- /dev/null +++ b/internal/checker/controlplane/runner_test.go @@ -0,0 +1,98 @@ +package controlplane + +import ( + "context" + "io" + "sync" + "testing" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/checker/probe" + + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestRunnerExecutesBoundedTaskBatchAndReportsFacts(t *testing.T) { + now := time.Date(2026, 8, 1, 13, 0, 0, 0, time.UTC) + client := &clientStub{stream: &taskStreamStub{tasks: []*controlplanev1.CheckTask{ + checkerTask("task-a", now), checkerTask("task-b", now), + }}} + executor := &executorStub{} + runner, err := NewRunner(client, executor, Options{ + CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 2, + SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_BASIC}, + ReportBatchSize: 2, RetryDelay: time.Millisecond, Now: func() time.Time { return now }, + }) + if err != nil { + t.Fatalf("NewRunner(): %v", err) + } + if err := runner.RunOnce(context.Background()); err != nil { + t.Fatalf("RunOnce(): %v", err) + } + if len(client.batches) != 1 || len(client.batches[0].GetObservations()) != 2 || len(executor.tasks) != 3 { + t.Fatalf("batches=%+v executions=%d", client.batches, len(executor.tasks)) + } + first, second := client.batches[0].GetObservations()[0], client.batches[0].GetObservations()[1] + if first.GetTaskId() != "task-a" || !first.GetSuccess() || first.GetFailureClass() != "" || + second.GetTaskId() != "task-b" || !second.GetSuccess() || first.GetObservedAt().AsTime() != now { + t.Fatalf("observations = %+v", client.batches[0].GetObservations()) + } +} + +func checkerTask(id string, now time.Time) *controlplanev1.CheckTask { + return &controlplanev1.CheckTask{ + TaskId: id, ProxyId: id + "-proxy", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, + Host: "proxy.example", Port: 8080, Level: controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, + Timeout: durationpb.New(time.Second), Attempt: 1, MaxAttempts: 2, Deadline: timestamppb.New(now.Add(time.Minute)), + } +} + +type clientStub struct { + stream TaskStream + batches []*controlplanev1.ObservationBatch +} + +func (stub *clientStub) StreamCheckTasks(context.Context, *controlplanev1.StreamCheckTasksRequest) (TaskStream, error) { + return stub.stream, nil +} + +func (stub *clientStub) ReportObservations(_ context.Context, batch *controlplanev1.ObservationBatch) (*controlplanev1.ReportObservationsResponse, error) { + stub.batches = append(stub.batches, batch) + return &controlplanev1.ReportObservationsResponse{Accepted: uint32(len(batch.GetObservations()))}, nil +} + +type taskStreamStub struct { + tasks []*controlplanev1.CheckTask + next int +} + +func (stub *taskStreamStub) Recv() (*controlplanev1.CheckTask, error) { + if stub.next >= len(stub.tasks) { + return nil, io.EOF + } + value := stub.tasks[stub.next] + stub.next++ + return value, nil +} + +type executorStub struct { + mu sync.Mutex + calls map[string]int + tasks []*controlplanev1.CheckTask +} + +func (stub *executorStub) Execute(_ context.Context, task *controlplanev1.CheckTask) probe.Result { + stub.mu.Lock() + defer stub.mu.Unlock() + stub.tasks = append(stub.tasks, task) + if stub.calls == nil { + stub.calls = make(map[string]int) + } + stub.calls[task.GetTaskId()]++ + if task.GetTaskId() == "task-a" && stub.calls[task.GetTaskId()] == 1 { + return probe.Result{FailureClass: probe.FailureProxyRequest, Latency: time.Millisecond} + } + return probe.Result{Success: true, Latency: 2 * time.Millisecond} +} diff --git a/internal/checker/probe/probe.go b/internal/checker/probe/probe.go new file mode 100644 index 0000000..908d9be --- /dev/null +++ b/internal/checker/probe/probe.go @@ -0,0 +1,156 @@ +// Package probe executes one bounded Checker task and returns only a health +// fact. It has no Controller, Redis, or PostgreSQL dependency. +package probe + +import ( + "context" + "errors" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" +) + +const ( + FailureInvalidTask = "INVALID_TASK" + FailureDeadlineExceeded = "DEADLINE_EXCEEDED" + FailureProxyRequest = "PROXY_REQUEST" + FailureProxyAuth = "PROXY_AUTH" + FailureTargetHTTPStatus = "TARGET_HTTP_STATUS" + FailureUnsupportedProxy = "UNSUPPORTED_PROXY_PROTOCOL" + basicHandshakeProbeURL = "http://example.invalid/" +) + +var errUnsupportedProtocol = errors.New("unsupported proxy protocol") + +type Result struct { + Success bool + FailureClass string + Latency time.Duration +} + +// Executor constructs a short-lived HTTP transport for each task. That keeps +// credentials task-scoped and avoids keeping stale short-TTL proxy sessions +// alive after a task completes. +type Executor struct{} + +func NewExecutor() *Executor { return &Executor{} } + +// Execute reports protocol/authorization reachability for BASIC and requires +// a successful target response for TARGET. Every path returns one fact so a +// Checker can report failures without treating ordinary probe failures as +// control-plane errors. +func (executor *Executor) Execute(ctx context.Context, task *controlplanev1.CheckTask) Result { + started := time.Now() + if ctx == nil || executor == nil { + return Result{FailureClass: FailureInvalidTask} + } + deadline, target, proxyURL, err := prepare(task, started.UTC()) + if err != nil { + if errors.Is(err, errUnsupportedProtocol) { + return Result{FailureClass: FailureUnsupportedProxy, Latency: time.Since(started)} + } + return Result{FailureClass: FailureInvalidTask, Latency: time.Since(started)} + } + probeContext, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + transport := &http.Transport{ + Proxy: http.ProxyURL(proxyURL), + DialContext: (&net.Dialer{}).DialContext, + ForceAttemptHTTP2: false, + TLSHandshakeTimeout: minDuration(time.Until(deadline), 10*time.Second), + ResponseHeaderTimeout: minDuration(time.Until(deadline), 10*time.Second), + } + defer transport.CloseIdleConnections() + client := &http.Client{ + Transport: transport, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + } + request, err := http.NewRequestWithContext(probeContext, http.MethodGet, target, nil) + if err != nil { + return Result{FailureClass: FailureInvalidTask, Latency: time.Since(started)} + } + response, err := client.Do(request) + latency := time.Since(started) + if err != nil { + if errors.Is(probeContext.Err(), context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) { + return Result{FailureClass: FailureDeadlineExceeded, Latency: latency} + } + return Result{FailureClass: FailureProxyRequest, Latency: latency} + } + _ = response.Body.Close() + if response.StatusCode == http.StatusProxyAuthRequired { + return Result{FailureClass: FailureProxyAuth, Latency: latency} + } + if task.GetLevel() == controlplanev1.CheckLevel_CHECK_LEVEL_BASIC { + return Result{Success: true, Latency: latency} + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusBadRequest { + return Result{FailureClass: FailureTargetHTTPStatus, Latency: latency} + } + return Result{Success: true, Latency: latency} +} + +func prepare(task *controlplanev1.CheckTask, now time.Time) (time.Time, string, *url.URL, error) { + if task == nil || task.GetTaskId() == "" || task.GetProxyId() == "" || task.GetHost() == "" || task.GetPort() == 0 || + task.GetTimeout() == nil || task.GetTimeout().CheckValid() != nil || task.GetTimeout().AsDuration() <= 0 || + task.GetDeadline() == nil || task.GetDeadline().CheckValid() != nil || task.GetDeadline().AsTime().UTC().Compare(now) <= 0 || + task.GetAttempt() == 0 || task.GetMaxAttempts() == 0 || task.GetAttempt() > task.GetMaxAttempts() || + (task.GetSecretRef() == "") != (task.GetCredentialVersion() == "") { + return time.Time{}, "", nil, errors.New("invalid checker task") + } + proxyScheme := "" + switch task.GetProtocol() { + case controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP: + proxyScheme = "http" + case controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTPS: + proxyScheme = "https" + default: + return time.Time{}, "", nil, errUnsupportedProtocol + } + deadline := task.GetDeadline().AsTime().UTC() + timeoutDeadline := now.Add(task.GetTimeout().AsDuration()) + if timeoutDeadline.Before(deadline) { + deadline = timeoutDeadline + } + if !deadline.After(now) { + return time.Time{}, "", nil, errors.New("expired checker task") + } + proxyURL := &url.URL{Scheme: proxyScheme, Host: net.JoinHostPort(task.GetHost(), strconv.FormatUint(uint64(task.GetPort()), 10))} + if task.GetUsername() != "" || task.GetPassword() != "" { + proxyURL.User = url.UserPassword(task.GetUsername(), task.GetPassword()) + } + target := basicHandshakeProbeURL + switch task.GetLevel() { + case controlplanev1.CheckLevel_CHECK_LEVEL_BASIC: + if task.GetRoutingName() != "" || task.GetTargetUrl() != "" { + return time.Time{}, "", nil, errors.New("invalid basic task") + } + case controlplanev1.CheckLevel_CHECK_LEVEL_TARGET: + if task.GetRoutingName() == "" { + return time.Time{}, "", nil, errors.New("target routing is required") + } + target = task.GetTargetUrl() + case controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS: + target = task.GetTargetUrl() + default: + return time.Time{}, "", nil, errors.New("unsupported task level") + } + parsedTarget, err := url.Parse(target) + if err != nil || parsedTarget.Scheme == "" || parsedTarget.Host == "" || parsedTarget.User != nil || parsedTarget.Fragment != "" || + (parsedTarget.Scheme != "http" && parsedTarget.Scheme != "https") || strings.TrimSpace(task.GetHost()) != task.GetHost() { + return time.Time{}, "", nil, errors.New("invalid probe target") + } + return deadline, parsedTarget.String(), proxyURL, nil +} + +func minDuration(left, right time.Duration) time.Duration { + if left <= 0 || left < right { + return left + } + return right +} diff --git a/internal/checker/probe/probe_test.go b/internal/checker/probe/probe_test.go new file mode 100644 index 0000000..2c79ca5 --- /dev/null +++ b/internal/checker/probe/probe_test.go @@ -0,0 +1,85 @@ +package probe + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestExecutorBasicConfirmsProxyHandshakeEvenWhenProbeTargetFails(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.String() != "http://example.invalid/" { + t.Errorf("proxy request URL = %q", request.URL) + } + if value := request.Header.Get("Proxy-Authorization"); value != "Basic dXNlcjpzZWNyZXQ=" { + t.Errorf("Proxy-Authorization = %q", value) + } + response.WriteHeader(http.StatusBadGateway) + })) + defer proxy.Close() + task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_BASIC) + result := NewExecutor().Execute(context.Background(), task) + if !result.Success || result.FailureClass != "" || result.Latency <= 0 { + t.Fatalf("Execute(BASIC) = %+v", result) + } +} + +func splitProxyAddress(t *testing.T, address string) (string, uint32) { + t.Helper() + host, rawPort, err := net.SplitHostPort(address[len("http://"):]) + if err != nil { + t.Fatalf("net.SplitHostPort(%q): %v", address, err) + } + port, err := strconv.ParseUint(rawPort, 10, 16) + if err != nil { + t.Fatalf("strconv.ParseUint(%q): %v", rawPort, err) + } + return host, uint32(port) +} + +func TestExecutorTargetReportsHTTPFailureAsFact(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.WriteHeader(http.StatusServiceUnavailable) + })) + defer proxy.Close() + task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_TARGET) + task.RoutingName = "route-a" + task.TargetUrl = "http://target.example/check" + result := NewExecutor().Execute(context.Background(), task) + if result.Success || result.FailureClass != FailureTargetHTTPStatus || result.Latency <= 0 { + t.Fatalf("Execute(TARGET) = %+v", result) + } +} + +func TestExecutorReportsUnsupportedProtocolAsFact(t *testing.T) { + now := time.Now().UTC() + result := NewExecutor().Execute(context.Background(), &controlplanev1.CheckTask{ + TaskId: "task-a", ProxyId: "proxy-a", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_SOCKS5, + Host: "proxy.example", Port: 1080, Level: controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, + Timeout: durationpb.New(time.Second), Attempt: 1, MaxAttempts: 1, Deadline: timestamppb.New(now.Add(time.Second)), + }) + if result.Success || result.FailureClass != FailureUnsupportedProxy { + t.Fatalf("Execute(SOCKS5) = %+v", result) + } +} + +func validTask(t *testing.T, proxyAddress string, level controlplanev1.CheckLevel) *controlplanev1.CheckTask { + t.Helper() + host, port := splitProxyAddress(t, proxyAddress) + now := time.Now().UTC() + return &controlplanev1.CheckTask{ + TaskId: "task-a", ProxyId: "proxy-a", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, + Host: host, Port: port, Level: level, Username: "user", Password: "secret", + Timeout: durationpb.New(time.Second), Attempt: 1, MaxAttempts: 1, + Deadline: timestamppb.New(now.Add(time.Second)), + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 2e4d57b..73234df 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -139,7 +139,8 @@ type ControlPlane struct { MaxRuntimeCounters int `yaml:"maxRuntimeCounters"` MaxConcurrentStreams uint32 `yaml:"maxConcurrentStreams"` TLS ControlPlaneTLS `yaml:"tls"` - GatewayTLS GatewayTLS `yaml:"gatewayTLS"` + GatewayTLS ClientTLS `yaml:"gatewayTLS"` + CheckerTLS ClientTLS `yaml:"checkerTLS"` } type ControlPlaneTLS struct { @@ -151,14 +152,19 @@ type ControlPlaneTLS struct { Environment string `yaml:"environment"` } -// GatewayTLS holds client-only mTLS material. It remains separate from the -// Controller's server certificate and CA configuration. -type GatewayTLS struct { +// ClientTLS holds client-only mTLS material. Each control-plane client type +// owns a distinct certificate so its SPIFFE identity cannot be confused with +// another process role. +type ClientTLS struct { CertFile string `yaml:"certFile"` KeyFile string `yaml:"keyFile"` ServerCAFile string `yaml:"serverCAFile"` } +// GatewayTLS remains as a source-compatible alias for callers that construct +// Gateway control-plane configuration in Go. +type GatewayTLS = ClientTLS + type Storage struct { PostgresURL string `yaml:"postgresURL"` RedisURL string `yaml:"redisURL"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index def7cdc..569e93e 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -462,6 +462,14 @@ func TestValidateControlPlane(t *testing.T) { }, want: "gatewayTLS", }, + { + name: "partial checker tls", + mutate: func(cfg *Config) { + cfg.ControlPlane = validMTLSControlPlane() + cfg.ControlPlane.CheckerTLS.KeyFile = "/run/secrets/checker-key.pem" + }, + want: "checkerTLS", + }, } for _, test := range tests { diff --git a/internal/config/validate.go b/internal/config/validate.go index 78aad50..dbb8808 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -114,7 +114,10 @@ func Validate(cfg *Config) error { } func validateControlPlane(item ControlPlane) error { - if err := validateGatewayTLS(item.GatewayTLS); err != nil { + if err := validateClientTLS("gatewayTLS", item.GatewayTLS); err != nil { + return err + } + if err := validateClientTLS("checkerTLS", item.CheckerTLS); err != nil { return err } if !item.Enabled { @@ -169,12 +172,16 @@ func validateControlPlane(item ControlPlane) error { } func validateGatewayTLS(item GatewayTLS) error { + return validateClientTLS("gatewayTLS", item) +} + +func validateClientTLS(name string, item ClientTLS) error { configured := item.CertFile != "" || item.KeyFile != "" || item.ServerCAFile != "" if !configured { return nil } if item.CertFile == "" || item.KeyFile == "" || item.ServerCAFile == "" { - return fmt.Errorf("validate controlPlane gatewayTLS: certFile, keyFile and serverCAFile are required together") + return fmt.Errorf("validate controlPlane %s: certFile, keyFile and serverCAFile are required together", name) } return nil } diff --git a/internal/controlplane/clienttransport/transport.go b/internal/controlplane/clienttransport/transport.go new file mode 100644 index 0000000..f603828 --- /dev/null +++ b/internal/controlplane/clienttransport/transport.go @@ -0,0 +1,95 @@ +// Package clienttransport builds the authenticated client side of the shared +// Controller gRPC endpoint for Gateway and Checker processes. +package clienttransport + +import ( + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "os" + "strconv" + "strings" + + "proxy-pool/internal/config" + + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +var ErrInvalidTransport = errors.New("invalid control-plane client transport") + +// New creates a TLS 1.3 client transport for the supplied client role. The +// caller supplies the role-specific certificate bundle, while the Controller +// validates its SPIFFE role/ID on the peer certificate. +func New( + configuration config.ControlPlane, + address string, + clientTLS config.ClientTLS, + override credentials.TransportCredentials, + role string, +) (credentials.TransportCredentials, error) { + if override != nil { + return override, nil + } + if !ValidDialAddress(address) || strings.TrimSpace(role) != role || role == "" { + return nil, ErrInvalidTransport + } + switch configuration.TLS.Mode { + case "disabled": + if !loopbackAddress(address) { + return nil, fmt.Errorf("%w: plaintext control-plane target must be loopback", ErrInvalidTransport) + } + return insecure.NewCredentials(), nil + case "mtls": + if clientTLS.CertFile == "" || clientTLS.KeyFile == "" || clientTLS.ServerCAFile == "" { + return nil, fmt.Errorf("%w: controlPlane.%s is required for mtls", ErrInvalidTransport, role) + } + certificate, err := tls.LoadX509KeyPair(clientTLS.CertFile, clientTLS.KeyFile) + if err != nil { + return nil, fmt.Errorf("load %s control-plane certificate: %w", role, err) + } + caPEM, err := os.ReadFile(clientTLS.ServerCAFile) + if err != nil { + return nil, fmt.Errorf("read %s control-plane CA: %w", role, err) + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(caPEM) { + return nil, fmt.Errorf("parse %s control-plane CA", role) + } + host, _, _ := net.SplitHostPort(address) + return credentials.NewTLS(&tls.Config{ + MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, RootCAs: roots, + ServerName: strings.Trim(host, "[]"), + }), nil + default: + return nil, fmt.Errorf("%w: unsupported control-plane tls mode", ErrInvalidTransport) + } +} + +func ValidDialAddress(address string) bool { + host, port, err := net.SplitHostPort(address) + if err != nil || host == "" { + return false + } + value, err := strconv.ParseUint(port, 10, 16) + if err != nil || value == 0 { + return false + } + parsed := net.ParseIP(strings.Trim(host, "[]")) + return parsed == nil || !parsed.IsUnspecified() +} + +func loopbackAddress(address string) bool { + host, _, err := net.SplitHostPort(address) + if err != nil { + return false + } + host = strings.Trim(host, "[]") + if strings.EqualFold(host, "localhost") { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go index 5ebe4f0..63a070a 100644 --- a/internal/gateway/bootstrap/bootstrap.go +++ b/internal/gateway/bootstrap/bootstrap.go @@ -5,24 +5,20 @@ package bootstrap import ( "bytes" "context" - "crypto/tls" - "crypto/x509" "errors" "fmt" "net" - "os" "reflect" - "strconv" "strings" "time" "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc" "google.golang.org/grpc/credentials" - "google.golang.org/grpc/credentials/insecure" controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/config" + "proxy-pool/internal/controlplane/clienttransport" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/gateway/controlplane" @@ -98,7 +94,7 @@ func Run(ctx context.Context, options Options) error { func validateOptions(ctx context.Context, options Options) error { if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" || nilInterface(options.Resolver) || !workerruntime.ValidIdentifier(options.ClusterID) || !workerruntime.ValidIdentifier(options.WorkerID) || - !workerruntime.ValidIdentifier(options.InstanceID) || !workerruntime.ValidIdentifier(options.Zone) || !validDialAddress(options.ControlPlaneAddress) { + !workerruntime.ValidIdentifier(options.InstanceID) || !workerruntime.ValidIdentifier(options.Zone) || !clienttransport.ValidDialAddress(options.ControlPlaneAddress) { return ErrInvalidOptions } if options.GatewayListener == nil && options.MetricsListener != nil { @@ -406,65 +402,11 @@ func controlPlaneTransport( address string, override credentials.TransportCredentials, ) (credentials.TransportCredentials, error) { - if override != nil { - return override, nil - } - switch configuration.TLS.Mode { - case "disabled": - if !loopbackAddress(address) { - return nil, fmt.Errorf("%w: plaintext control plane target must be loopback", ErrInvalidOptions) - } - return insecure.NewCredentials(), nil - case "mtls": - if configuration.GatewayTLS.CertFile == "" || configuration.GatewayTLS.KeyFile == "" || configuration.GatewayTLS.ServerCAFile == "" { - return nil, fmt.Errorf("%w: controlPlane.gatewayTLS is required for mtls", ErrInvalidOptions) - } - certificate, err := tls.LoadX509KeyPair(configuration.GatewayTLS.CertFile, configuration.GatewayTLS.KeyFile) - if err != nil { - return nil, fmt.Errorf("load gateway control plane certificate: %w", err) - } - caPEM, err := os.ReadFile(configuration.GatewayTLS.ServerCAFile) - if err != nil { - return nil, fmt.Errorf("read gateway control plane CA: %w", err) - } - roots := x509.NewCertPool() - if !roots.AppendCertsFromPEM(caPEM) { - return nil, errors.New("parse gateway control plane CA") - } - host, _, _ := net.SplitHostPort(address) - return credentials.NewTLS(&tls.Config{ - MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, RootCAs: roots, - ServerName: strings.Trim(host, "[]"), - }), nil - default: - return nil, fmt.Errorf("%w: unsupported control plane tls mode", ErrInvalidOptions) - } -} - -func validDialAddress(address string) bool { - host, port, err := net.SplitHostPort(address) - if err != nil || host == "" { - return false - } - value, err := strconv.ParseUint(port, 10, 16) - if err != nil || value == 0 { - return false - } - parsed := net.ParseIP(strings.Trim(host, "[]")) - return parsed == nil || !parsed.IsUnspecified() -} - -func loopbackAddress(address string) bool { - host, _, err := net.SplitHostPort(address) + transport, err := clienttransport.New(configuration, address, configuration.GatewayTLS, override, "gatewayTLS") if err != nil { - return false + return nil, fmt.Errorf("%w: %v", ErrInvalidOptions, err) } - host = strings.Trim(host, "[]") - if strings.EqualFold(host, "localhost") { - return true - } - ip := net.ParseIP(host) - return ip != nil && ip.IsLoopback() + return transport, nil } func nilInterface(value any) bool {