fix(run): hop route refusals name both addresses; the first-Init port guard probes the proposed ports (r129, r141)

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.

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)
+	})
+}

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")
+}