diff --git a/internal/checker/controlplane/runner_test.go b/internal/checker/controlplane/runner_test.go index 5b80fd0..2b91ba9 100644 --- a/internal/checker/controlplane/runner_test.go +++ b/internal/checker/controlplane/runner_test.go @@ -41,6 +41,29 @@ func TestRunnerExecutesBoundedTaskBatchAndReportsFacts(t *testing.T) { } } +func TestRunnerOmitsEgressProbeURLFromGlobalObservation(t *testing.T) { + now := time.Date(2026, 8, 1, 13, 30, 0, 0, time.UTC) + task := checkerTask("task-egress", now) + task.Level = controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS + task.TargetUrl = "https://egress.example/identity" + client := &clientStub{stream: &taskStreamStub{tasks: []*controlplanev1.CheckTask{task}}} + runner, err := NewRunner(client, &executorStub{}, Options{ + CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1, + SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS}, + ReportBatchSize: 1, 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) + } + observation := client.batches[0].GetObservations()[0] + if observation.GetLevel() != controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS || observation.GetTargetUrl() != "" || observation.GetRoutingName() != "" { + t.Fatalf("Observation = %+v, want global EGRESS fact without target profile", observation) + } +} + func checkerTask(id string, now time.Time) *controlplanev1.CheckTask { return &controlplanev1.CheckTask{ TaskId: id, ProxyId: id + "-proxy", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, diff --git a/internal/checker/probe/probe.go b/internal/checker/probe/probe.go index 908d9be..0d4c254 100644 --- a/internal/checker/probe/probe.go +++ b/internal/checker/probe/probe.go @@ -136,6 +136,9 @@ func prepare(task *controlplanev1.CheckTask, now time.Time) (time.Time, string, } target = task.GetTargetUrl() case controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS: + if task.GetRoutingName() != "" || task.GetTargetUrl() == "" { + return time.Time{}, "", nil, errors.New("invalid egress task") + } target = task.GetTargetUrl() default: return time.Time{}, "", nil, errors.New("unsupported task level") diff --git a/internal/checker/probe/probe_test.go b/internal/checker/probe/probe_test.go index 2c79ca5..0e8d216 100644 --- a/internal/checker/probe/probe_test.go +++ b/internal/checker/probe/probe_test.go @@ -60,6 +60,20 @@ func TestExecutorTargetReportsHTTPFailureAsFact(t *testing.T) { } } +func TestExecutorRejectsEgressTaskWithRoutingProfile(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + response.WriteHeader(http.StatusOK) + })) + defer proxy.Close() + task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS) + task.RoutingName = "route-a" + task.TargetUrl = "https://egress.example/identity" + result := NewExecutor().Execute(context.Background(), task) + if result.Success || result.FailureClass != FailureInvalidTask { + t.Fatalf("Execute(EGRESS with routing) = %+v", result) + } +} + func TestExecutorReportsUnsupportedProtocolAsFact(t *testing.T) { now := time.Now().UTC() result := NewExecutor().Execute(context.Background(), &controlplanev1.CheckTask{ diff --git a/internal/controller/health/grpc_handler.go b/internal/controller/health/grpc_handler.go index c3952f8..c079366 100644 --- a/internal/controller/health/grpc_handler.go +++ b/internal/controller/health/grpc_handler.go @@ -239,13 +239,23 @@ func wireCheckTask(task LeasedTask, now time.Time) (*controlplanev1.CheckTask, e if !ok { return nil, ErrInvalidLeasedTask } - if task.Level == healthDomain.LevelTarget { + switch task.Level { + case healthDomain.LevelBasic: + if task.RoutingName != "" || task.TargetURL != "" { + return nil, ErrInvalidLeasedTask + } + case healthDomain.LevelEgress: + targetURL, err := healthDomain.NormalizeEgressTarget(task.TargetURL) + if task.RoutingName != "" || err != nil || targetURL != task.TargetURL { + return nil, ErrInvalidLeasedTask + } + case healthDomain.LevelTarget: if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ RoutingName: task.RoutingName, TargetURL: task.TargetURL, }); err != nil { return nil, ErrInvalidLeasedTask } - } else if task.RoutingName != "" || task.TargetURL != "" { + default: return nil, ErrInvalidLeasedTask } return &controlplanev1.CheckTask{ diff --git a/internal/controller/health/grpc_handler_test.go b/internal/controller/health/grpc_handler_test.go index 5f2f6f4..f4caa9f 100644 --- a/internal/controller/health/grpc_handler_test.go +++ b/internal/controller/health/grpc_handler_test.go @@ -173,6 +173,60 @@ func TestGRPCHandlerStreamsLeasedTasksAndFencesReportedFacts(t *testing.T) { } } +func TestGRPCHandlerStreamsEgressProbeURLAndAcceptsGlobalFact(t *testing.T) { + now := time.Date(2026, 8, 1, 12, 30, 0, 0, time.UTC) + broker, err := NewMemoryTaskBroker(MemoryTaskBrokerOptions{ + LeaseTTL: time.Minute, Now: func() time.Time { return now }, + Material: TaskMaterialResolverFunc(func(context.Context, Candidate) (TaskMaterial, error) { + return TaskMaterial{Protocol: proxyDomain.SchemeHTTP, Host: "proxy.example", Port: 8080}, nil + }), + }) + if err != nil { + t.Fatalf("NewMemoryTaskBroker(): %v", err) + } + const egressURL = "https://egress.example/identity" + if offered, offerErr := broker.Offer(context.Background(), []PlannedTask{{ + Candidate: Candidate{ProxyID: "proxy-a", State: proxyDomain.StateAvailable, Level: healthDomain.LevelEgress, TargetURL: egressURL, DueAt: now}, + Deadline: now.Add(10 * time.Second), Attempts: 1, + }}); offerErr != nil || offered != 1 { + t.Fatalf("Offer() = (%d, %v)", offered, offerErr) + } + global := &recordingGlobalStore{} + reducer, err := NewReducer(global, &recordingTargetStore{}, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) + if err != nil { + t.Fatalf("NewReducer(): %v", err) + } + handler, err := NewGRPCHandler(reducer, &recordingCheckerIdentity{}, GRPCHandlerOptions{ + MaxObservationsPerBatch: 1, MaxTasksPerClaim: 1, TaskBroker: broker, Now: func() time.Time { return now }, + }) + if err != nil { + t.Fatalf("NewGRPCHandler(): %v", err) + } + client, closeClient := newCheckerGRPCClient(t, handler) + defer closeClient() + + stream, err := client.StreamCheckTasks(context.Background(), &controlplanev1.StreamCheckTasksRequest{ + CheckerId: "checker-a", InstanceId: "instance-a", MaxInFlight: 1, + SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS}, + }) + if err != nil { + t.Fatalf("StreamCheckTasks(): %v", err) + } + task, err := stream.Recv() + if err != nil || task.GetLevel() != controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS || task.GetTargetUrl() != egressURL || task.GetRoutingName() != "" { + t.Fatalf("Recv() = (%+v, %v)", task, err) + } + observation := grpcHealthObservation(task.GetTaskId(), controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS, now.Add(time.Second)) + observation.LeaseToken = task.GetLeaseToken() + response, err := client.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ + CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{observation}, + }) + if err != nil || response.GetAccepted() != 1 || len(global.commands) != 1 || + global.commands[0].Observation.TargetURL != "" || global.commands[0].Observation.RoutingName != "" { + t.Fatalf("ReportObservations() = (%+v, %v), commands=%+v", response, err, global.commands) + } +} + func newCheckerGRPCClient(t *testing.T, handler controlplanev1.CheckerControlPlaneServer) (controlplanev1.CheckerControlPlaneClient, func()) { t.Helper() listener := bufconn.Listen(1 << 20) diff --git a/internal/controller/health/scheduler.go b/internal/controller/health/scheduler.go index 5a33c74..4038dcc 100644 --- a/internal/controller/health/scheduler.go +++ b/internal/controller/health/scheduler.go @@ -142,10 +142,15 @@ func (planner *Planner) validateCandidate(candidate Candidate) (Priority, bool, return 0, false, ErrInvalidScheduleRequest } switch candidate.Level { - case healthDomain.LevelBasic, healthDomain.LevelEgress: + case healthDomain.LevelBasic: if candidate.RoutingName != "" || candidate.TargetURL != "" { return 0, false, ErrInvalidScheduleRequest } + case healthDomain.LevelEgress: + targetURL, err := healthDomain.NormalizeEgressTarget(candidate.TargetURL) + if candidate.RoutingName != "" || err != nil || targetURL != candidate.TargetURL { + return 0, false, ErrInvalidScheduleRequest + } case healthDomain.LevelTarget: if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL, diff --git a/internal/controller/health/scheduler_test.go b/internal/controller/health/scheduler_test.go index bb36afe..b79d98b 100644 --- a/internal/controller/health/scheduler_test.go +++ b/internal/controller/health/scheduler_test.go @@ -19,7 +19,7 @@ func TestPlannerPrioritizesAndBoundsOverdueCandidates(t *testing.T) { now := time.Date(2026, 7, 31, 17, 0, 0, 0, time.UTC) plans, err := planner.Plan(now, 1, 10, []Candidate{ {ProxyID: "available", State: proxyDomain.StateAvailable, Level: healthDomain.LevelBasic, DueAt: now.Add(-time.Second)}, - {ProxyID: "suspect", State: proxyDomain.StateSuspect, Level: healthDomain.LevelEgress, DueAt: now.Add(-2 * time.Second)}, + {ProxyID: "suspect", State: proxyDomain.StateSuspect, Level: healthDomain.LevelEgress, TargetURL: "https://egress.example/identity", DueAt: now.Add(-2 * time.Second)}, {ProxyID: "fetched", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(-3 * time.Second)}, {ProxyID: "future", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now.Add(time.Second)}, }) diff --git a/internal/controller/health/task_broker.go b/internal/controller/health/task_broker.go index 30d5e07..5279218 100644 --- a/internal/controller/health/task_broker.go +++ b/internal/controller/health/task_broker.go @@ -352,10 +352,15 @@ func validatePlannedTask(task PlannedTask, now time.Time) error { return ErrInvalidLeasedTask } switch task.Candidate.Level { - case healthDomain.LevelBasic, healthDomain.LevelEgress: + case healthDomain.LevelBasic: if task.Candidate.RoutingName != "" || task.Candidate.TargetURL != "" { return ErrInvalidLeasedTask } + case healthDomain.LevelEgress: + targetURL, err := healthDomain.NormalizeEgressTarget(task.Candidate.TargetURL) + if task.Candidate.RoutingName != "" || err != nil || targetURL != task.Candidate.TargetURL { + return ErrInvalidLeasedTask + } case healthDomain.LevelTarget: if _, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ RoutingName: task.Candidate.RoutingName, TargetURL: task.Candidate.TargetURL, @@ -403,8 +408,17 @@ func newLeaseToken() (string, error) { } func taskMatchesObservation(task PlannedTask, observation healthDomain.Observation) bool { - return task.Candidate.ProxyID == observation.ProxyID && task.Candidate.Level == observation.Level && - task.Candidate.RoutingName == observation.RoutingName && task.Candidate.TargetURL == observation.TargetURL + if task.Candidate.ProxyID != observation.ProxyID || task.Candidate.Level != observation.Level { + return false + } + switch task.Candidate.Level { + case healthDomain.LevelBasic, healthDomain.LevelEgress: + return observation.RoutingName == "" && observation.TargetURL == "" + case healthDomain.LevelTarget: + return task.Candidate.RoutingName == observation.RoutingName && task.Candidate.TargetURL == observation.TargetURL + default: + return false + } } func validTaskIdentifier(value string) bool { diff --git a/internal/domain/health/health.go b/internal/domain/health/health.go index b2f42a6..ee0ac34 100644 --- a/internal/domain/health/health.go +++ b/internal/domain/health/health.go @@ -145,6 +145,22 @@ func NormalizeTargetProfile(value TargetProfile) (TargetProfile, error) { return value, nil } +// NormalizeEgressTarget canonicalizes the HTTP endpoint used to observe a +// proxy's public egress identity. The endpoint is task execution material, +// never part of the resulting global health fact. +func NormalizeEgressTarget(value string) (string, error) { + value = strings.TrimSpace(value) + if value == "" { + return "", ErrInvalidObservation + } + parsed, err := url.Parse(value) + if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" || + (parsed.Scheme != "http" && parsed.Scheme != "https") { + return "", ErrInvalidObservation + } + return parsed.String(), nil +} + func (value TargetProfile) Key() string { return value.RoutingName + "\x00" + value.TargetURL } diff --git a/internal/domain/health/health_test.go b/internal/domain/health/health_test.go index bc3f8a1..837dafe 100644 --- a/internal/domain/health/health_test.go +++ b/internal/domain/health/health_test.go @@ -86,6 +86,16 @@ func TestNormalizeObservationRejectsCrossLevelFields(t *testing.T) { } } +func TestNormalizeEgressTargetAcceptsOnlyHTTPOrHTTPSURL(t *testing.T) { + target, err := NormalizeEgressTarget("https://egress.example/identity?format=json") + if err != nil || target != "https://egress.example/identity?format=json" { + t.Fatalf("NormalizeEgressTarget(valid) = (%q, %v)", target, err) + } + if _, err := NormalizeEgressTarget("ftp://egress.example/identity"); !errors.Is(err, ErrInvalidObservation) { + t.Fatalf("NormalizeEgressTarget(ftp) error = %v, want ErrInvalidObservation", err) + } +} + func globalObservation(taskID string, success bool, observedAt time.Time) Observation { observation := Observation{TaskID: taskID, ProxyID: "proxy-a", Level: LevelBasic, Success: success, ObservedAt: observedAt} if !success {