diff --git a/docs/orchestration.md b/docs/orchestration.md index 5702d08b..b489e4f6 100644 --- a/docs/orchestration.md +++ b/docs/orchestration.md @@ -1230,3 +1230,33 @@ agent's process effects. This is an explicit restart boundary, not live route reconfiguration. Existing hop versions that do not declare `hop-routes` keep their existing configuration mode; a new hop requires both the CLI projection and the generated group. + +A routed module ordered after the hop — one that calls through it is — has +loaded but not initialized when the hop starts. The snapshot routes it at the +address its own Init will be proposed, derived from the workspace, the service +and the naming scope, and that Init is the check: the proposed address is +accepted; an Init that accepts another address, or drops the routed endpoint, +is refused naming the route, the address the snapshot holds and the one +offered. A restart derives the same address again, so the remedy is the +module's agent serving the proposed address or, for a module that does not +call through the hop, ordering it before the hop. + +Until that Init, the hop routes to a loopback port nothing holds yet, and a +local process that binds it first receives what the hop forwards until then. +The module's first Init ends that: before the agent's Init, the run dials +every native port the Init proposes on 127.0.0.1 and on ::1 and, if anything +answers, refuses the Init naming the port and its holder, so the run does not +come up with the hop routing to a squatter. Each loopback family is probed +because on macOS a squatter on 127.0.0.1:P and a service listening on the +wildcard :P both bind. The one listener that is not a squatter is the +service's own kept container: a Docker store with `keep-running` leaves its +container up for the next run to adopt, still publishing the service's port. +An address is let through only when the container listening there is the one +the service's agent will adopt — its name in this run's workspace and naming +scope, codefly-owned, labelled with this run's recovery scope, running, not +ephemeral, and publishing that port on that loopback family; any other holder, +including one on the other family beside it, is refused. Not covered: a +process that binds a loopback address on the port after that probe, and a +routed module that crashes or is withdrawn mid-run, whose port the hop keeps +routing to. `--temporary-ports` derives no +address, so a reference that needs one is refused before the hop starts. diff --git a/pkg/orchestration/adopted_container_ports.go b/pkg/orchestration/adopted_container_ports.go new file mode 100644 index 00000000..aa2b9d59 --- /dev/null +++ b/pkg/orchestration/adopted_container_ports.go @@ -0,0 +1,121 @@ +package orchestration + +import ( + "context" + "net" + "slices" + "strconv" + "strings" + "time" + + "github.com/codefly-dev/core/runners/dockerrun" + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" +) + +// The loopback and wildcard addresses of each family, as Docker and the +// socket table spell them. +const ( + loopbackIPv4 = "127.0.0.1" + loopbackIPv6 = "::1" + wildcardIPv4 = "0.0.0.0" + wildcardIPv6 = "::" +) + +// listOwnedContainers lists the codefly containers carrying one exact recovery +// scope. It is a variable so a test can stand in for the Docker engine. +var listOwnedContainers = func(ctx context.Context, scopeID string) ([]container.Summary, error) { + cli, err := dockerrun.NewClient() + if err != nil { + return nil, err + } + defer cli.Close() + listCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + return cli.ContainerList(listCtx, container.ListOptions{Filters: filters.NewArgs( + filters.Arg("label", dockerrun.LabelCodeflyOwner+"=true"), + filters.Arg("label", dockerrun.LabelCodeflyRecoveryScope+"="+scopeID), + )}) +} + +// adoptedContainerAddresses answers which loopback addresses the first-Init +// port guard must not treat as a squatter: the ones published by the +// container this service's agent will adopt. A Docker store with +// keep-running leaves its container up so the next run reattaches to it by +// name, and that container still publishes the service's deterministic port — +// the one this Init proposes. +// +// The container is identified the way the agent's runner reuses it: the name +// every Core service agent gives it (dockerrun.ContainerName over the +// service's unique name in the run's workspace and naming scope, as +// Base.UniqueWithWorkspace spells it), owned by codefly, +// labelled with this flow's exact recovery scope and, when it carries one, its +// namespace (the rule dockerrun applies before reusing a container), running, +// and not an ephemeral one, which is never reused across sessions. An address +// is excused only for the loopback family the container publishes on: a +// 127.0.0.1 or 0.0.0.0 publication covers 127.0.0.1, a ::1 or :: one covers +// ::1, so a squatter on the other family is still refused. Anything this +// cannot establish — no recovery identity, no Docker engine — excuses nothing. +func (runner *Runner) adoptedContainerAddresses(ctx context.Context) func(address string) bool { + scopeID, namespace, ok := strings.Cut(runner.containerRecoveryIdentity, ":") + if !ok || scopeID == "" || runner.instance == nil || runner.instance.Identity == nil || + runner.world == nil || runner.world.Workspace == nil || runner.world.Env == nil { + return func(string) bool { return false } + } + // The workspace and naming scope the agent's Load receives + // (RuntimeInstance.Load sends the run's workspace name, not the identity's). + identity, workspace := runner.instance.Identity, runner.world.Workspace.Name + unique := identity.UniqueWithWorkspace(workspace) + if scope := runner.world.Env.NamingScope; scope != "" { + unique = identity.UniqueWithWorkspaceAndScope(workspace, scope) + } + name := dockerrun.ContainerName(unique) + var containers []container.Summary + listed := false + return func(address string) bool { + if !listed { + listed = true + containers, _ = listOwnedContainers(ctx, scopeID) + } + return slices.ContainsFunc(containers, func(c container.Summary) bool { + return adoptableContainer(&c, name, scopeID, namespace) && publishesAddress(&c, address) + }) + } +} + +func adoptableContainer(c *container.Summary, name, scopeID, namespace string) bool { + labels := c.Labels + if labels[dockerrun.LabelCodeflyOwner] != "true" || labels[dockerrun.LabelCodeflyRecoveryScope] != scopeID || + labels[dockerrun.LabelCodeflyName] != name || labels[dockerrun.LabelCodeflyEphemeral] == "true" || c.State != "running" { + return false + } + if found := labels[dockerrun.LabelCodeflyRecoveryNamespace]; found != "" && found != namespace { + return false + } + return slices.Contains(c.Names, "/"+name) +} + +// publishesAddress reports whether c publishes the loopback host:port address +// on that address's own family. +func publishesAddress(c *container.Summary, address string) bool { + host, rawPort, err := net.SplitHostPort(address) + if err != nil { + return false + } + port, err := strconv.ParseUint(rawPort, 10, 16) + if err != nil { + return false + } + var covering []string + switch host { + case loopbackIPv4: + covering = []string{loopbackIPv4, wildcardIPv4} + case loopbackIPv6: + covering = []string{loopbackIPv6, wildcardIPv6} + default: + return false + } + return slices.ContainsFunc(c.Ports, func(p container.Port) bool { + return uint64(p.PublicPort) == port && p.Type == "tcp" && slices.Contains(covering, p.IP) + }) +} diff --git a/pkg/orchestration/adopted_container_ports_test.go b/pkg/orchestration/adopted_container_ports_test.go new file mode 100644 index 00000000..93768fef --- /dev/null +++ b/pkg/orchestration/adopted_container_ports_test.go @@ -0,0 +1,120 @@ +package orchestration + +import ( + "context" + "fmt" + "net" + "strings" + "testing" + + "github.com/codefly-dev/core/runners/dockerrun" + "github.com/docker/docker/api/types/container" + "github.com/stretchr/testify/require" +) + +const ( + testRecoveryScope = "1111111111111111111111111111111111111111111111111111111111111111" + testRecoveryNamespace = "2222222222222222222222222222222222222222222222222222222222222222" +) + +// keptContainer is the container a keep-running agent left up for this +// service, publishing its proposed port on 127.0.0.1 the way Core publishes it. +func keptContainer(name string, port uint16) container.Summary { + return container.Summary{ + Names: []string{"/" + name}, + State: "running", + Labels: map[string]string{ + dockerrun.LabelCodeflyOwner: "true", + dockerrun.LabelCodeflyName: name, + dockerrun.LabelCodeflyRecoveryScope: testRecoveryScope, + dockerrun.LabelCodeflyRecoveryNamespace: testRecoveryNamespace, + }, + Ports: []container.Port{{IP: "127.0.0.1", PrivatePort: 5432, PublicPort: port, Type: "tcp"}}, + } +} + +// A first Init whose proposed port is held by the service's own kept +// container — the one its agent adopts under keep-running — is not refused; +// any other holder of that port still is, on either loopback family. +func TestFirstInitAdoptsItsOwnKeptContainerAndRefusesAnyOtherHolder(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(*container.Summary) + ipv6 bool // a squatter on [::1] beside the kept container + refused bool + }{ + {name: "kept container", refused: false}, + {name: "foreign holder on ::1 beside it", ipv6: true, refused: true}, + {name: "another recovery scope", mutate: func(c *container.Summary) { c.Labels[dockerrun.LabelCodeflyRecoveryScope] = strings.Repeat("3", 64) }, refused: true}, + {name: "another namespace", mutate: func(c *container.Summary) { + c.Labels[dockerrun.LabelCodeflyRecoveryNamespace] = strings.Repeat("4", 64) + }, refused: true}, + {name: "another service's name", mutate: func(c *container.Summary) { + c.Names = []string{"/codefly-other"} + c.Labels[dockerrun.LabelCodeflyName] = "codefly-other" + }, refused: true}, + {name: "ephemeral", mutate: func(c *container.Summary) { c.Labels[dockerrun.LabelCodeflyEphemeral] = "true" }, refused: true}, + {name: "not running", mutate: func(c *container.Summary) { c.State = "exited" }, refused: true}, + {name: "published on another port", mutate: func(c *container.Summary) { c.Ports[0].PublicPort++ }, refused: true}, + } { + t.Run(tc.name, func(t *testing.T) { + // Docker's port forwarder, stood in for by a listener on the address + // the kept container publishes. + forwarder, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer forwarder.Close() + held := uint16(forwarder.Addr().(*net.TCPAddr).Port) + if tc.ipv6 { + squatter, listenErr := net.Listen("tcp", net.JoinHostPort("::1", fmt.Sprint(held))) + if listenErr != nil { + t.Skipf("no ::1 loopback here: %v", listenErr) + } + defer squatter.Close() + } + runner, world := gatewayRunner(t, startRealAgent(t, realAgentEcho)) + runner.containerRecoveryIdentity = testRecoveryScope + ":" + testRecoveryNamespace + world.LocalNetworkManager.WithPortOverrides(map[string]uint16{"web/gateway/rest": held, "web/gateway/grpc": freePort(t)}) + // A run's instance identity carries no workspace: the agent is told + // the run's workspace name at Load, and names its container with it. + runner.instance.Identity.Workspace = "" + kept := keptContainer(dockerrun.ContainerName(world.Workspace.Name+"-web-gateway"), held) + if tc.mutate != nil { + tc.mutate(&kept) + } + listed := "" + previous := listOwnedContainers + listOwnedContainers = func(_ context.Context, scopeID string) ([]container.Summary, error) { + listed = scopeID + return []container.Summary{kept}, nil + } + t.Cleanup(func() { listOwnedContainers = previous }) + + _, err = runner.Init(context.Background()) + if tc.refused { + require.ErrorContains(t, err, fmt.Sprintf("port %d (rest)", held)) + require.ErrorContains(t, err, "already in use") + return + } + require.NoError(t, err) + require.Equal(t, testRecoveryScope, listed, "only this flow's recovery scope is consulted") + }) + } +} + +// Without a recovery identity nothing can be attributed, so nothing is +// excused: the port guard refuses as before. +func TestAdoptedContainerAddressesExcuseNothingWithoutARecoveryIdentity(t *testing.T) { + runner := &Runner{} + require.False(t, runner.adoptedContainerAddresses(context.Background())("127.0.0.1:5432")) +} + +// A publication covers only its own loopback family. +func TestPublishesAddressIsPerLoopbackFamily(t *testing.T) { + c := container.Summary{Ports: []container.Port{{IP: "0.0.0.0", PublicPort: 5432, Type: "tcp"}}} + require.True(t, publishesAddress(&c, "127.0.0.1:5432")) + require.False(t, publishesAddress(&c, "[::1]:5432")) + c.Ports = []container.Port{{IP: "::", PublicPort: 5432, Type: "tcp"}} + require.True(t, publishesAddress(&c, "[::1]:5432")) + require.False(t, publishesAddress(&c, "127.0.0.1:5432")) + require.False(t, publishesAddress(&c, "localhost:5432"), "a name is never attributed") +} diff --git a/pkg/orchestration/hop_configurations.go b/pkg/orchestration/hop_configurations.go index bf6da4c7..281bed6f 100644 --- a/pkg/orchestration/hop_configurations.go +++ b/pkg/orchestration/hop_configurations.go @@ -270,6 +270,17 @@ func (flow *Flow) CheckRoutingSnapshot(ctx context.Context, workspace *resources // the routed service's Init the check (RecordNetworkMappings): the same // address is accepted, any other is refused naming the route. A service that // never loaded has no mapping at all and is refused here by name. +// +// Until that Init, the hop routes to a derived loopback port nothing holds +// yet, so a local process that binds it first receives what the hop forwards +// until then. That Init ends it: before calling the agent, the routed +// service's first Init dials the port it is proposed on 127.0.0.1 and ::1 and +// refuses, naming the holder, if anything answers +// (Runner.checkInitialPortAvailability), so the run never comes up with the +// hop routing to a squatter. --temporary-ports never derives an address at +// all. A process that binds a loopback address on the same port after that +// probe, or a routed service that crashes or is withdrawn mid-run, is not +// covered. func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.NetworkMapping) error { routes, err := routing.DecodeRouting(raw) if err != nil { @@ -294,12 +305,19 @@ func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.Network } key, hash := resources.EndpointDestination(selected.GetEndpoint()), resources.NetworkMappingHash(selected) if recorded, initialized := s.networkMappings[unique]; initialized && !carriesMapping(recorded, key, hash) { - return routingRestartRequired(fmt.Sprintf("hop route %s is not the address %s initialized with", key, unique)) + // One record means the hop resolved the address that first Init + // was proposed: a restart proposes it again. + if s.networkMappingRecords[unique] == 1 { + return proposedAddressRefused(key, mappingAddresses(selected), unique, addressesOf(recorded, key)) + } + return routingRestartRequired(fmt.Sprintf("hop route %s was resolved at %s, but %s initialized at %s while the routes were being resolved", + key, mappingAddresses(selected), unique, addressesOf(recorded, key))) } if previous, ok := s.routingMappings[key]; ok && previous.hash != hash { - return routingRestartRequired(fmt.Sprintf("hop route %s moved from the address this run's hop snapshot holds", key)) + return routingRestartRequired(fmt.Sprintf("hop route %s moved from %s, the address this run's hop snapshot holds, to %s", + key, mappingAddresses(previous.mapping), mappingAddresses(selected))) } - pending[key] = frozenRoute{owner: unique, hash: hash} + pending[key] = frozenRoute{owner: unique, hash: hash, mapping: selected} } if s.routingMappings == nil { s.routingMappings = map[string]frozenRoute{} @@ -310,24 +328,45 @@ func (s *StateManager) freezeHopAddresses(raw string, resolved []*basev0.Network return nil } -// frozenRoute is one address a hop snapshot holds: the service that serves it -// and the hash of the mapping the snapshot was built with. +// frozenRoute is one address a hop snapshot holds: the service that serves it, +// the mapping the snapshot was built with, and that mapping's hash. type frozenRoute struct { - owner string - hash string + owner string + hash string + mapping *basev0.NetworkMapping } // checkFrozenRoutes refuses mappings for unique that drop or move an address a -// hop snapshot holds. The caller holds s.mu. +// hop snapshot holds. The caller holds s.mu and has not recorded them yet. +// +// A service recording for the first time was frozen at the address its Init +// was proposed, and that address is derived from the workspace, the service +// and the naming scope: a restart derives it again and is refused the same +// way, so the refusal says what does help instead of asking for one. func (s *StateManager) checkFrozenRoutes(unique string, mappings []*basev0.NetworkMapping) error { + _, initialized := s.networkMappings[unique] for key, frozen := range s.routingMappings { - if frozen.owner == unique && !carriesMapping(mappings, key, frozen.hash) { - return routingRestartRequired(fmt.Sprintf("the hop snapshot routes %s at an address %s does not accept", key, unique)) + if frozen.owner != unique || carriesMapping(mappings, key, frozen.hash) { + continue + } + routed, offered := mappingAddresses(frozen.mapping), addressesOf(mappings, key) + if initialized { + return routingRestartRequired(fmt.Sprintf("hop route %s is routed at %s, but %s now offers %s", key, routed, unique, offered)) } + return proposedAddressRefused(key, routed, unique, offered) } return nil } +// proposedAddressRefused is the refusal of a first Init that accepted another +// address than the one it was proposed, which the hop routes. A restart +// proposes the same derived address again, so it is not offered as a remedy. +func proposedAddressRefused(key, routed, unique, offered string) error { + return fmt.Errorf("hop route %s is routed at %s, the address %s was proposed, but its Init accepted %s: "+ + "the proposed address is derived from the workspace, the service and the naming scope, so restarting this run derives it again and is refused the same way; "+ + "have %s's agent serve the proposed address, or, if %s does not call through the hop, order it before the hop", key, routed, unique, offered, unique, unique) +} + // carriesMapping reports whether mappings hold the endpoint key, and hold it // only at the mapping hashed as hash. func carriesMapping(mappings []*basev0.NetworkMapping, key, hash string) bool { @@ -344,6 +383,34 @@ func carriesMapping(mappings []*basev0.NetworkMapping, key, hash string) bool { return found } +// addressesOf spells where mappings reach the endpoint key. +func addressesOf(mappings []*basev0.NetworkMapping, key string) string { + var matching []*basev0.NetworkMapping + for _, mapping := range mappings { + if resources.EndpointDestination(mapping.GetEndpoint()) == key { + matching = append(matching, mapping) + } + } + if len(matching) == 0 { + return "no address for it" + } + return mappingAddresses(matching...) +} + +// mappingAddresses spells each instance of mappings as its access and address. +func mappingAddresses(mappings ...*basev0.NetworkMapping) string { + var parts []string + for _, mapping := range mappings { + for _, instance := range mapping.GetInstances() { + parts = append(parts, instance.GetAccess().GetKind()+" "+instance.GetAddress()) + } + } + if len(parts) == 0 { + return "no address" + } + return "[" + strings.Join(parts, ", ") + "]" +} + func (world *World) freezeDeliveredHopAddresses(configs []*basev0.Configuration, mappings []*basev0.NetworkMapping) error { if world.Mode != RunMode || world.SharedState == nil { return nil diff --git a/pkg/orchestration/hop_configurations_test.go b/pkg/orchestration/hop_configurations_test.go index 5f1e5d8b..738c8ec5 100644 --- a/pkg/orchestration/hop_configurations_test.go +++ b/pkg/orchestration/hop_configurations_test.go @@ -92,6 +92,7 @@ func TestGeneratedHopRoutesUseTheNormalGraphAndAddressResolver(t *testing.T) { moved := endpointMapping("platform", "first", "web", "http", "public", nativeInstance("http://moved.native"), containerInstance("http://first.cell")) err = world.SharedState.RecordNetworkMappings(ctx, first, []*basev0.NetworkMapping{moved}) require.ErrorContains(t, err, "stop and restart") + require.ErrorContains(t, err, "(hop route platform/first/web::http is routed at [native http://first.native, container http://first.cell], but platform/first now offers [native http://moved.native, container http://first.cell])") previous, found := world.SharedState.GetNetworkMappingsFromUnique("platform/first") require.True(t, found) require.NotContains(t, fmt.Sprint(previous), "moved.native", "rejected mapping never replaces accepted route") @@ -148,12 +149,17 @@ func TestHopRoutesFreezeAModuleThatInitializesAfterTheHop(t *testing.T) { // name, and nothing is recorded in its place. moved := endpointMapping("platform", "second", "web", "http", "public", nativeInstance("http://moved.native")) err = world.SharedState.RecordNetworkMappings(ctx, second, []*basev0.NetworkMapping{moved}) - require.ErrorContains(t, err, "stop and restart") - require.ErrorContains(t, err, "platform/second") + require.ErrorContains(t, err, "hop route platform/second/web::http is routed at "+mappingAddresses(proposed...)) + require.ErrorContains(t, err, "its Init accepted [native http://moved.native]") + // The address is derived, so a restart cannot help and is not asked for; + // ordering before the hop is offered only to a module that can be. + require.ErrorContains(t, err, "restarting this run derives it again") + require.ErrorContains(t, err, "have platform/second's agent serve the proposed address, or, if platform/second does not call through the hop, order it before the hop") + require.NotContains(t, err.Error(), "stop and restart") _, recorded := world.SharedState.GetNetworkMappingsFromUnique("platform/second") require.False(t, recorded, "a refused Init records nothing") // An Init that drops the routed endpoint is refused the same way. - require.ErrorContains(t, world.SharedState.RecordNetworkMappings(ctx, second, nil), "stop and restart") + require.ErrorContains(t, world.SharedState.RecordNetworkMappings(ctx, second, nil), "its Init accepted no address for it") // The address the hop froze is accepted. require.NoError(t, world.SharedState.RecordNetworkMappings(ctx, second, proposed)) // And the hop's next delivery reads the same snapshot. @@ -161,6 +167,83 @@ func TestHopRoutesFreezeAModuleThatInitializesAfterTheHop(t *testing.T) { require.NoError(t, err) } +// hopFenceState is a run's shared state over hopRouteWorkspace with +// platform/second loaded but not initialized, and one hop route to its web +// listener. +func hopFenceState(t *testing.T) (*StateManager, *resources.Service, string) { + t.Helper() + t.Setenv(resources.CodeflyHomeEnv, filepath.Join(t.TempDir(), "home")) + ctx := context.Background() + workspace := hopRouteWorkspace(t, "public") + state, err := NewStateManager(ctx, nil, nil, workspace) + require.NoError(t, err) + module, err := workspace.LoadModuleFromName(ctx, "platform") + require.NoError(t, err) + second, err := module.LoadServiceFromName(ctx, "second") + require.NoError(t, err) + raw, err := json.Marshal(routing.Routing{Schema: routing.RoutingSchema, Endpoints: []routing.RouteEndpoint{{ + Audience: "platform", Service: "second", Endpoint: "web", Transport: "http", Address: "http://derived.native", + Package: "owner/platform", Version: "1.0.0", Digest: "test", Methods: []string{"GET /second"}, + }}}) + require.NoError(t, err) + return state, second, string(raw) +} + +func secondAt(address string) *basev0.NetworkMapping { + return endpointMapping("platform", "second", "web", "http", "public", nativeInstance(address)) +} + +// The hop resolved second at the address it derived, and second's Init +// recorded another one before the freeze ran: the freeze refuses, naming both. +func TestHopFreezeRefusesARecordBetweenDerivationAndFreeze(t *testing.T) { + ctx := context.Background() + state, second, raw := hopFenceState(t) + require.NoError(t, state.RecordNetworkMappings(ctx, second, []*basev0.NetworkMapping{secondAt("http://moved.native")})) + err := state.freezeHopAddresses(raw, []*basev0.NetworkMapping{secondAt("http://derived.native")}) + // Its first record: the hop resolved the address that Init was proposed, + // and a restart proposes it again, so no restart is asked for. + require.ErrorContains(t, err, "hop route platform/second/web::http is routed at [native http://derived.native], the address platform/second was proposed, but its Init accepted [native http://moved.native]") + require.ErrorContains(t, err, "restarting this run derives it again") + require.NotContains(t, err.Error(), "stop and restart") + require.Empty(t, state.routingMappings, "a refused freeze holds nothing") + // A service that re-records inside the window (a reload) is a real change: + // a restart adopts its new address. + require.NoError(t, state.RecordNetworkMappings(ctx, second, []*basev0.NetworkMapping{secondAt("http://reloaded.native")})) + err = state.freezeHopAddresses(raw, []*basev0.NetworkMapping{secondAt("http://moved.native")}) + require.ErrorContains(t, err, "stop and restart") + require.ErrorContains(t, err, "hop route platform/second/web::http was resolved at [native http://moved.native], but platform/second initialized at [native http://reloaded.native]") +} + +// A frozen route is carried only when every mapping for its endpoint is the +// frozen one: a second, different mapping for the same endpoint is refused, +// whichever comes first. +func TestHopFrozenRouteRefusesADuplicateEndpointAtAnotherAddress(t *testing.T) { + ctx := context.Background() + derived, other := secondAt("http://derived.native"), secondAt("http://other.native") + key := resources.EndpointDestination(derived.GetEndpoint()) + require.True(t, carriesMapping([]*basev0.NetworkMapping{derived}, key, resources.NetworkMappingHash(derived))) + for _, offered := range [][]*basev0.NetworkMapping{{derived, other}, {other, derived}} { + require.False(t, carriesMapping(offered, key, resources.NetworkMappingHash(derived))) + state, second, raw := hopFenceState(t) + require.NoError(t, state.freezeHopAddresses(raw, []*basev0.NetworkMapping{derived})) + err := state.RecordNetworkMappings(ctx, second, offered) + require.ErrorContains(t, err, "its Init accepted [native http://") + _, recorded := state.GetNetworkMappingsFromUnique("platform/second") + require.False(t, recorded) + } +} + +// A later delivery that resolves a routed service, still not initialized, at +// another address than the snapshot froze is refused as a move. +func TestHopFreezeRefusesARouteThatMovedBeforeItsOwnerInitialized(t *testing.T) { + state, _, raw := hopFenceState(t) + require.NoError(t, state.freezeHopAddresses(raw, []*basev0.NetworkMapping{secondAt("http://derived.native")})) + require.NoError(t, state.freezeHopAddresses(raw, []*basev0.NetworkMapping{secondAt("http://derived.native")}), "the same address re-freezes") + err := state.freezeHopAddresses(raw, []*basev0.NetworkMapping{secondAt("http://other.native")}) + require.ErrorContains(t, err, "stop and restart") + require.ErrorContains(t, err, "hop route platform/second/web::http moved from [native http://derived.native], the address this run's hop snapshot holds, to [native http://other.native]") +} + func TestGeneratedHopRoutesRefusePrivatePublicationAndStaleArtifacts(t *testing.T) { t.Setenv(resources.CodeflyHomeEnv, filepath.Join(t.TempDir(), "home")) t.Run("private", func(t *testing.T) { diff --git a/pkg/orchestration/runner.go b/pkg/orchestration/runner.go index 97142b21..70b06411 100644 --- a/pkg/orchestration/runner.go +++ b/pkg/orchestration/runner.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "slices" + "strconv" "strings" "sync" "sync/atomic" @@ -268,18 +269,6 @@ func (runner *Runner) Init(ctx context.Context) (*OutputProperty, error) { return nil, err } - // Reject stale native listeners before calling the agent's Init hook. - // Infrastructure agents are allowed to bind their assigned endpoint during - // Init (for example Vault starts its Docker/Nix runtime there), so probing in - // Start is already too late and mistakes the service we just initialized for - // a stale process. - // - // A running service being hot-reloaded keeps its port and is marked started, - // so it must skip this first-start guard. - if err := runner.checkInitialPortAvailability(ctx); err != nil { - return nil, w.Wrapf(err, "cannot initialize %s", runner.instance.Unique()) - } - // Configuration reads can block on a stalled provider (e.g. a dependency // service that failed to export its config). Bound each read with a // timeout so a single bad provider doesn't stall the entire Init. @@ -290,6 +279,21 @@ func (runner *Runner) Init(ctx context.Context) (*OutputProperty, error) { if err != nil { return nil, err } + + // Reject stale native listeners before calling the agent's Init hook, on + // the addresses this Init proposes: on a first Init nothing has been + // accepted yet, and those are the ports the service is about to serve — + // and the ones a hop routing to it already holds. Infrastructure agents + // are allowed to bind their assigned endpoint during Init (for example + // Vault starts its Docker/Nix runtime there), so probing in Start is + // already too late and mistakes the service we just initialized for a + // stale process. + // + // A running service being hot-reloaded keeps its port and is marked started, + // so it must skip this first-start guard. + if err = runner.checkInitialPortAvailability(ctx, inputs.networkMappings); err != nil { + return nil, w.Wrapf(err, "cannot initialize %s", runner.instance.Unique()) + } dependenciesEndpoints := inputs.dependenciesEndpoints dependenciesNetworkMappings := inputs.dependenciesNetworkMappings conf := inputs.conf @@ -1103,16 +1107,30 @@ const portHolderLookupTimeout = 3 * time.Second type portCollision struct { port uint32 endpoint string + // address is the host:port the holder answered on, so the holder named is + // the one on that address — a squatter on [::1]:P beside another process + // on 127.0.0.1:P is the one reported. + address string } -// boundNativePorts returns " ()" descriptions for each of the -// runner's own native endpoint ports that another process is already listening -// on, naming the process that holds each one when it can be identified. A +// boundNativePorts returns " ()" descriptions for each native +// endpoint port in mappings that another process is already listening on, +// naming the process that holds each one when it can be identified. A // successful TCP dial is the honest signal that the port is held: unlike a // listen probe it doesn't trip on TIME_WAIT sockets that have no listener. -func (runner *Runner) boundNativePorts(ctx context.Context) []string { +// +// Each loopback family is dialled explicitly, besides the instance's own host: +// a listener on 127.0.0.1:P and one on [::1]:P are different sockets, and on +// macOS a service that later binds the wildcard :P coexists with a squatter on +// 127.0.0.1:P, which keeps receiving every dial to 127.0.0.1:P. Only a probe +// of both families before the service binds sees either squatter. +// +// adopted excuses an address whose listener is the service's own: the +// container its agent will adopt (Runner.adoptedContainerAddresses). nil +// excuses nothing. +func boundNativePorts(ctx context.Context, mappings []*basev0.NetworkMapping, adopted func(address string) bool) []string { var collisions []portCollision - for _, mapping := range runner.networkMappings { + for _, mapping := range mappings { // networkMappings arrive over the agent's Init gRPC response, so a nil // mapping or endpoint is possible at this boundary — skip rather than panic. if mapping == nil || mapping.Endpoint == nil { @@ -1122,26 +1140,51 @@ func (runner *Runner) boundNativePorts(ctx context.Context) []string { if instance == nil { continue } - conn, err := net.DialTimeout("tcp", instance.Host, 200*time.Millisecond) - if err != nil { - continue + for _, address := range nativeProbeAddresses(instance) { + if listening(address) && (adopted == nil || !adopted(address)) { + collisions = append(collisions, portCollision{port: instance.Port, endpoint: mapping.Endpoint.Name, address: address}) + break + } } - _ = conn.Close() - collisions = append(collisions, portCollision{port: instance.Port, endpoint: mapping.Endpoint.Name}) } if len(collisions) == 0 { return nil } lookupCtx, cancel := holderLookupContext(ctx) defer cancel() - listeners := listeningPorts(lookupCtx) + sockets := listeningSockets(lookupCtx) held := make([]string, 0, len(collisions)) for _, collision := range collisions { - held = append(held, fmt.Sprintf("%d (%s)%s", collision.port, collision.endpoint, portHolderSuffix(lookupCtx, listeners, collision.port))) + held = append(held, fmt.Sprintf("%d (%s)%s", collision.port, collision.endpoint, portHolderSuffix(lookupCtx, holderPID(sockets, collision.address, collision.port)))) } return held } +// nativeProbeAddresses is every address a native instance's port can be +// held at: the port on each loopback family, and the instance's own host when +// it is not a loopback name those two already cover. A loopback host is +// probed per family only, so a holder is always attributed to one family. +func nativeProbeAddresses(instance *basev0.NetworkInstance) []string { + var addresses []string + if host := instance.GetHost(); host != "" && !isLoopbackHostname(instance.GetHostname()) { + addresses = append(addresses, host) + } + if instance.GetPort() > 0 { + port := strconv.FormatUint(uint64(instance.GetPort()), 10) + addresses = append(addresses, net.JoinHostPort(loopbackIPv4, port), net.JoinHostPort(loopbackIPv6, port)) + } + return addresses +} + +func listening(address string) bool { + conn, err := net.DialTimeout("tcp", address, 200*time.Millisecond) + if err != nil { + return false + } + _ = conn.Close() + return true +} + // holderLookupContext derives the deadline the holder attribution runs under. // One deadline covers the whole attribution — the socket enumeration and every // executable lookup it feeds — so naming holders is bounded no matter how many @@ -1151,29 +1194,58 @@ func holderLookupContext(ctx context.Context) (context.Context, context.CancelFu return context.WithTimeout(ctx, portHolderLookupTimeout) } -// listeningPorts maps each locally LISTENing TCP port to the pid that holds it. -// Building the whole map once per probe keeps a stack-wide collision from -// re-enumerating the machine's sockets for every clashing endpoint. -func listeningPorts(ctx context.Context) map[uint32]int32 { +// listeningSocket is one locally LISTENing TCP socket and the pid holding it. +type listeningSocket struct { + ip string + port uint32 + pid int32 +} + +// listeningSockets enumerates the machine's listening TCP sockets once per +// probe, so a stack-wide collision does not re-enumerate them for every +// clashing endpoint. +func listeningSockets(ctx context.Context) []listeningSocket { connections, err := gopsnet.ConnectionsWithContext(ctx, "tcp") if err != nil { - return map[uint32]int32{} + return nil } - ports := make(map[uint32]int32, len(connections)) + var sockets []listeningSocket for _, connection := range connections { if connection.Status != "LISTEN" || connection.Pid <= 0 { continue } - // A port bound on several addresses (v4 + v6, or per-interface) yields - // one connection each; they belong to the same listener, so first wins. - if _, seen := ports[connection.Laddr.Port]; !seen { - ports[connection.Laddr.Port] = connection.Pid + sockets = append(sockets, listeningSocket{ip: connection.Laddr.IP, port: connection.Laddr.Port, pid: connection.Pid}) + } + return sockets +} + +// holderPID is the pid listening where address answered: a socket on that +// exact IP first, then a wildcard socket of its family, then any socket on the +// port. Zero when none is readable. +func holderPID(sockets []listeningSocket, address string, port uint32) int32 { + host, _, err := net.SplitHostPort(address) + if err != nil { + host = "" + } + wildcards := []string{wildcardIPv4, "*", ""} + if ip := net.ParseIP(host); ip != nil && ip.To4() == nil { + wildcards = []string{wildcardIPv6, "*", ""} + } + for _, matches := range []func(string) bool{ + func(ip string) bool { return host != "" && ip == host }, + func(ip string) bool { return slices.Contains(wildcards, ip) }, + func(string) bool { return true }, + } { + for _, socket := range sockets { + if socket.port == port && matches(socket.ip) { + return socket.pid + } } } - return ports + return 0 } -// portHolderSuffix names the process squatting on port, so an "already in use" +// portHolderSuffix names the process holding a port, so an "already in use" // failure says which pid to look at instead of leaving the user to run lsof. // // It reports the holder's executable PATH and never its command line: the path @@ -1185,9 +1257,8 @@ func listeningPorts(ctx context.Context) map[uint32]int32 { // It returns "" when the holder cannot be identified: socket ownership is not // always readable (a listener owned by another user, a lookup that ran past // the deadline), and a missing name must not cost us the port report itself. -func portHolderSuffix(ctx context.Context, listeners map[uint32]int32, port uint32) string { - pid, ok := listeners[port] - if !ok { +func portHolderSuffix(ctx context.Context, pid int32) string { + if pid <= 0 { return "" } proc, err := process.NewProcessWithContext(ctx, pid) @@ -1201,7 +1272,7 @@ func portHolderSuffix(ctx context.Context, listeners map[uint32]int32, port uint return fmt.Sprintf(" held by pid %d (%s)", pid, executable) } -func (runner *Runner) checkInitialPortAvailability(ctx context.Context) error { +func (runner *Runner) checkInitialPortAvailability(ctx context.Context, proposed []*basev0.NetworkMapping) error { // Only the very first Init can see a ghost: once this runner has started // the service, a bound port is its own — either still running (hot reload) // or still releasing after the Stop that precedes a restart. Testing @@ -1210,7 +1281,9 @@ func (runner *Runner) checkInitialPortAvailability(ctx context.Context) error { if runner.isStarted.Load() || runner.everStarted.Load() { return nil } - if held := runner.boundNativePorts(ctx); len(held) > 0 { + // A listener that is the service's own kept container is not a ghost: its + // agent adopts it (keep-running), and 'codefly clear' would delete it. + if held := boundNativePorts(ctx, proposed, runner.adoptedContainerAddresses(ctx)); len(held) > 0 { // The holder is whoever bound the port first — it may be a leftover // codefly run, but it may equally be the user's own server, so the // remedy is stated conditionally rather than asserting a provenance diff --git a/pkg/orchestration/runner_accepted_network_mappings_test.go b/pkg/orchestration/runner_accepted_network_mappings_test.go index fe7654ca..55ec1539 100644 --- a/pkg/orchestration/runner_accepted_network_mappings_test.go +++ b/pkg/orchestration/runner_accepted_network_mappings_test.go @@ -525,7 +525,8 @@ func TestInitRejectsAStalePortBeforeTheFirstStart(t *testing.T) { ctx := context.Background() runner, _ := gatewayRunner(t, startRealAgent(t, realAgentEcho)) - // The guard probes the addresses a previous Init accepted, so run one first. + // A second Init before the first Start: its proposal is the accepted set + // the first Init pinned, and the guard still refuses a ghost on it. _, err := runner.Init(ctx) require.NoError(t, err) @@ -536,3 +537,28 @@ func TestInitRejectsAStalePortBeforeTheFirstStart(t *testing.T) { _, err = runner.Init(ctx) require.ErrorContains(t, err, "already in use") } + +// A first Init probes the addresses it proposes, before the agent's Init: on a +// Runner built as a run builds it nothing has been accepted yet, so probing +// the accepted set probed nothing. Each loopback family is probed on its own, +// since a squatter on one of them coexists with a later wildcard bind on macOS. +func TestFirstInitRefusesAProposedPortAnotherProcessHolds(t *testing.T) { + for _, family := range []string{"127.0.0.1", "::1"} { + t.Run(family, func(t *testing.T) { + squatter, err := net.Listen("tcp", net.JoinHostPort(family, "0")) + if err != nil { + t.Skipf("no %s loopback here: %v", family, err) + } + defer squatter.Close() + held := squatter.Addr().(*net.TCPAddr).Port + runner, world := gatewayRunner(t, startRealAgent(t, realAgentEcho)) + world.LocalNetworkManager.WithPortOverrides(map[string]uint16{"web/gateway/rest": uint16(held), "web/gateway/grpc": freePort(t)}) + + _, err = runner.Init(context.Background()) + require.ErrorContains(t, err, fmt.Sprintf("port %d (rest)", held)) + require.ErrorContains(t, err, "already in use") + recorded, found := world.SharedState.GetNetworkMappingsFromUnique(runner.instance.Identity.Unique()) + require.False(t, found, "a refused first Init records nothing: %v", recorded) + }) + } +} diff --git a/pkg/orchestration/runner_test.go b/pkg/orchestration/runner_test.go index 87d64f53..5f9b0db0 100644 --- a/pkg/orchestration/runner_test.go +++ b/pkg/orchestration/runner_test.go @@ -76,12 +76,10 @@ func TestBoundNativePortsDetectsHeldPort(t *testing.T) { free := freeLn.Addr().(*net.TCPAddr).Port require.NoError(t, freeLn.Close()) - runner := &Runner{networkMappings: []*basev0.NetworkMapping{ + reported := boundNativePorts(context.Background(), []*basev0.NetworkMapping{ nativeMapping("grpc", uint16(held)), nativeMapping("free", uint16(free)), - }} - - reported := runner.boundNativePorts(context.Background()) + }, nil) require.Len(t, reported, 1) require.Contains(t, reported[0], fmt.Sprintf("%d (grpc)", held)) // The holder is this test process, so the report must name its pid — @@ -120,12 +118,12 @@ func TestBoundNativePortsDegradesWhenHolderLookupFails(t *testing.T) { defer ln.Close() held := ln.Addr().(*net.TCPAddr).Port - runner := &Runner{networkMappings: []*basev0.NetworkMapping{nativeMapping("grpc", uint16(held))}} + mappings := []*basev0.NetworkMapping{nativeMapping("grpc", uint16(held))} ctx, cancel := context.WithCancel(context.Background()) cancel() done := make(chan []string, 1) - go func() { done <- runner.boundNativePorts(ctx) }() + go func() { done <- boundNativePorts(ctx, mappings, nil) }() select { case reported := <-done: require.Len(t, reported, 1) @@ -141,7 +139,7 @@ func TestBoundNativePortsDegradesWhenHolderLookupFails(t *testing.T) { // codefly binary — and never by argv, which routinely carries secrets. func TestPortHolderSuffixNamesExecutableNotCommandLine(t *testing.T) { self := int32(os.Getpid()) - suffix := portHolderSuffix(context.Background(), map[uint32]int32{30093: self}, 30093) + suffix := portHolderSuffix(context.Background(), self) require.Contains(t, suffix, fmt.Sprintf("held by pid %d", self)) executable, err := os.Executable() @@ -158,16 +156,48 @@ func TestPortHolderSuffixNamesExecutableNotCommandLine(t *testing.T) { // TestPortHolderSuffixOmitsUnknownHolder keeps an unidentifiable listener from // costing us the port report itself: socket ownership is not always readable. func TestPortHolderSuffixOmitsUnknownHolder(t *testing.T) { - require.Empty(t, portHolderSuffix(context.Background(), map[uint32]int32{}, 30093)) + require.Empty(t, portHolderSuffix(context.Background(), 0)) +} + +// The holder named is the one on the address that answered: a squatter on +// [::1]:P beside another process on 127.0.0.1:P is the one reported. +func TestHolderPIDNamesTheListenerOnTheAddressThatAnswered(t *testing.T) { + sockets := []listeningSocket{{ip: "127.0.0.1", port: 5432, pid: 10}, {ip: "::1", port: 5432, pid: 20}, {ip: "*", port: 6000, pid: 30}} + require.Equal(t, int32(20), holderPID(sockets, "[::1]:5432", 5432)) + require.Equal(t, int32(10), holderPID(sockets, "127.0.0.1:5432", 5432)) + require.Equal(t, int32(30), holderPID(sockets, "[::1]:6000", 6000), "a wildcard listener holds both families") + require.Equal(t, int32(10), holderPID(sockets, "db.internal:5432", 5432), "an unattributable address falls back to the port") + require.Zero(t, holderPID(sockets, "127.0.0.1:7000", 7000)) } func TestBoundNativePortsToleratesNilMappings(t *testing.T) { - runner := &Runner{networkMappings: []*basev0.NetworkMapping{ + require.Empty(t, boundNativePorts(context.Background(), []*basev0.NetworkMapping{ nil, {Endpoint: nil}, {Endpoint: &basev0.Endpoint{Name: "grpc"}, Instances: nil}, - }} - require.Empty(t, runner.boundNativePorts(context.Background())) + }, nil)) +} + +// A port is held whichever loopback family the holder bound, even when the +// instance names the other one: each family is a separate socket, and on +// macOS a squatter on one coexists with the service's later wildcard bind. +func TestBoundNativePortsProbesBothLoopbackFamilies(t *testing.T) { + for _, tc := range []struct{ squatter, named string }{{"::1", "127.0.0.1"}, {"127.0.0.1", "::1"}} { + t.Run(tc.squatter, func(t *testing.T) { + ln, err := net.Listen("tcp", net.JoinHostPort(tc.squatter, "0")) + if err != nil { + t.Skipf("no %s loopback here: %v", tc.squatter, err) + } + defer ln.Close() + held := ln.Addr().(*net.TCPAddr).Port + instance := resources.NewNetworkInstance(tc.named, uint16(held)) + instance.Access = resources.NewNativeNetworkAccess() + reported := boundNativePorts(context.Background(), []*basev0.NetworkMapping{{ + Endpoint: &basev0.Endpoint{Name: "grpc"}, Instances: []*basev0.NetworkInstance{instance}, + }}, nil) + require.Len(t, reported, 1, "a holder on %s is missed when the instance names %s", tc.squatter, tc.named) + }) + } } func TestInitialPortGuardRejectsFirstInitAndSkipsRunningService(t *testing.T) { @@ -176,11 +206,10 @@ func TestInitialPortGuardRejectsFirstInitAndSkipsRunningService(t *testing.T) { defer ln.Close() held := ln.Addr().(*net.TCPAddr).Port - runner := &Runner{networkMappings: []*basev0.NetworkMapping{ - nativeMapping("http", uint16(held)), - }} + runner := &Runner{} + proposed := []*basev0.NetworkMapping{nativeMapping("http", uint16(held))} - err = runner.checkInitialPortAvailability(context.Background()) + err = runner.checkInitialPortAvailability(context.Background(), proposed) require.ErrorContains(t, err, fmt.Sprintf("port %d (http)", held)) require.ErrorContains(t, err, "already in use") // The holder here is this test process, not a leftover codefly run, so the @@ -191,7 +220,7 @@ func TestInitialPortGuardRejectsFirstInitAndSkipsRunningService(t *testing.T) { // the time Start is called. Once the service is marked running, reloads // must not classify its own port as stale. runner.isStarted.Store(true) - require.NoError(t, runner.checkInitialPortAvailability(context.Background())) + require.NoError(t, runner.checkInitialPortAvailability(context.Background(), proposed)) } func TestStatusDiagnosticPreservesAgentMessage(t *testing.T) { diff --git a/pkg/orchestration/state_manager.go b/pkg/orchestration/state_manager.go index 6130d240..36bf6762 100644 --- a/pkg/orchestration/state_manager.go +++ b/pkg/orchestration/state_manager.go @@ -35,6 +35,9 @@ type StateManager struct { networkMappings map[string][]*basev0.NetworkMapping // Generated hop snapshots retain these exact addresses until run restart. routingMappings map[string]frozenRoute + // How many times each service has recorded its mappings: one means the + // recorded set is its first Init's. + networkMappingRecords map[string]int } // SetDependencies rebinds the dependency graph. Used by Flow when @@ -301,5 +304,9 @@ func (s *StateManager) RecordNetworkMappings(ctx context.Context, service *resou return err } s.networkMappings[id.Unique()] = mappings + if s.networkMappingRecords == nil { + s.networkMappingRecords = map[string]int{} + } + s.networkMappingRecords[id.Unique()]++ return nil }