diff --git a/api/runner/rpc.yml b/api/runner/rpc.yml index dac50aa20..97474895b 100644 --- a/api/runner/rpc.yml +++ b/api/runner/rpc.yml @@ -163,9 +163,9 @@ interfaces: index: 6 doc: | Report whether the coordinator has a workload identity issuer - configured, and its issuer URL. Distributed runners call this once at - startup to decide whether to mint workload identity tokens via the - coordinator. + configured, its issuer URL, and its internal WireGuard address. + Distributed runners call this once at startup to configure identity + tokens and registry pulls. results: - name: enabled type: bool @@ -173,6 +173,9 @@ interfaces: - name: issuer_url type: string doc: The issuer URL (iss claim anchor) when enabled + - name: coordinator_internal_ip + type: string + doc: Current coordinator bridge gateway reachable over WireGuard - name: IssueWorkloadToken index: 7 diff --git a/api/runner/runner_v1alpha/rpc.gen.go b/api/runner/runner_v1alpha/rpc.gen.go index 9f8961e5b..efef3153a 100644 --- a/api/runner/runner_v1alpha/rpc.gen.go +++ b/api/runner/runner_v1alpha/rpc.gen.go @@ -967,8 +967,9 @@ func (v *RunnerRegistrationWorkloadIssuerInfoArgs) UnmarshalJSON(data []byte) er } type runnerRegistrationWorkloadIssuerInfoResultsData struct { - Enabled *bool `cbor:"0,keyasint,omitempty" json:"enabled,omitempty"` - IssuerUrl *string `cbor:"1,keyasint,omitempty" json:"issuer_url,omitempty"` + Enabled *bool `cbor:"0,keyasint,omitempty" json:"enabled,omitempty"` + IssuerUrl *string `cbor:"1,keyasint,omitempty" json:"issuer_url,omitempty"` + CoordinatorInternalIp *string `cbor:"2,keyasint,omitempty" json:"coordinator_internal_ip,omitempty"` } type RunnerRegistrationWorkloadIssuerInfoResults struct { @@ -984,6 +985,10 @@ func (v *RunnerRegistrationWorkloadIssuerInfoResults) SetIssuerUrl(issuer_url st v.data.IssuerUrl = &issuer_url } +func (v *RunnerRegistrationWorkloadIssuerInfoResults) SetCoordinatorInternalIp(coordinator_internal_ip string) { + v.data.CoordinatorInternalIp = &coordinator_internal_ip +} + func (v *RunnerRegistrationWorkloadIssuerInfoResults) MarshalCBOR() ([]byte, error) { return cbor.Marshal(v.data) } @@ -2485,6 +2490,17 @@ func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) IssuerUrl() string { return *v.data.IssuerUrl } +func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) HasCoordinatorInternalIp() bool { + return v.data.CoordinatorInternalIp != nil +} + +func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) CoordinatorInternalIp() string { + if v.data.CoordinatorInternalIp == nil { + return "" + } + return *v.data.CoordinatorInternalIp +} + func (v RunnerRegistrationClient) WorkloadIssuerInfo(ctx context.Context) (*RunnerRegistrationClientWorkloadIssuerInfoResults, error) { args := RunnerRegistrationWorkloadIssuerInfoArgs{} diff --git a/components/coordinate/coordinate.go b/components/coordinate/coordinate.go index 351ee43bb..64e75cc7f 100644 --- a/components/coordinate/coordinate.go +++ b/components/coordinate/coordinate.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "net/netip" "os" "path/filepath" "time" @@ -27,14 +28,15 @@ type EtcdTLSConfig struct { } type CoordinatorConfig struct { - Address string `json:"address" yaml:"address"` - EtcdEndpoints []string `json:"etcd_endpoints" yaml:"etcd_endpoints"` - Prefix string `json:"prefix" yaml:"prefix"` - Resolver netresolve.Resolver `json:"resolver" yaml:"resolver"` - TempDir string `json:"temp_dir" yaml:"temp_dir"` - DataPath string `json:"data_path" yaml:"data_path"` - AdditionalNames []string `json:"additional_names" yaml:"additional_names"` - IPs *IPSet `json:"ips" yaml:"ips"` + Address string `json:"address" yaml:"address"` + CoordinatorInternalIP netip.Addr `json:"-" yaml:"-"` + EtcdEndpoints []string `json:"etcd_endpoints" yaml:"etcd_endpoints"` + Prefix string `json:"prefix" yaml:"prefix"` + Resolver netresolve.Resolver `json:"resolver" yaml:"resolver"` + TempDir string `json:"temp_dir" yaml:"temp_dir"` + DataPath string `json:"data_path" yaml:"data_path"` + AdditionalNames []string `json:"additional_names" yaml:"additional_names"` + IPs *IPSet `json:"ips" yaml:"ips"` // ACME certificate configuration AcmeEmail string `json:"acme_email" yaml:"acme_email"` diff --git a/components/coordinate/runner_endpoints.go b/components/coordinate/runner_endpoints.go index d83cf485f..d63eb2514 100644 --- a/components/coordinate/runner_endpoints.go +++ b/components/coordinate/runner_endpoints.go @@ -52,6 +52,7 @@ func (c *RunnerEndpoints) Start(context.Context) error { Authority: c.authority, EAC: c.eac, CoordinatorAddr: c.Address, + CoordinatorInternalIP: c.CoordinatorInternalIP, EtcdEndpoints: c.EtcdEndpoints, EtcdPrefix: c.Prefix, VictoriametricsAddress: c.VictoriametricsAddress, diff --git a/components/distributedrunner/boot_sandbox_host.go b/components/distributedrunner/boot_sandbox_host.go index 9380bc659..48cf43b29 100644 --- a/components/distributedrunner/boot_sandbox_host.go +++ b/components/distributedrunner/boot_sandbox_host.go @@ -92,7 +92,7 @@ func (b *sandboxHostBoot) start( EtcdEndpoints: append([]string(nil), b.inputs.etcdEndpoints...), EtcdPrefix: b.inputs.etcdPrefix, } - if err := b.prepareNetworkDeps(&dependencies); err != nil { + if err := b.prepareNetworkDeps(&dependencies, access.access.CoordinatorInternalIP()); err != nil { return nil, err } @@ -107,9 +107,15 @@ func (b *sandboxHostBoot) start( return b.value, nil } -func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps) error { +func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps, coordinatorInternalIP netip.Addr) error { resolver, hostMapper := netresolve.NewLocalResolver() deps.Resolver = resolver + if coordinatorInternalIP.Is4() { + if err := hostMapper.SetHost("cluster.local", coordinatorInternalIP); err != nil { + return fmt.Errorf("mapping cluster registry: %w", err) + } + b.inputs.log.Info("mapped cluster.local to coordinator WireGuard gateway", "addr", coordinatorInternalIP) + } coordinatorHost, coordinatorPort, splitErr := net.SplitHostPort(b.inputs.coordinator) if splitErr != nil { b.inputs.log.Warn("in-cluster API access disabled: coordinator address has no usable host and port", @@ -120,11 +126,16 @@ func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps) error { // Sandboxes reach the API on the coordinator rather than the local bridge // router. This must be an IP because sandbox DNS resolves app.miren names // and nothing else, so the coordinator hostname would not resolve there. - hostMapper.SetHost("cluster.local", coordinatorAddr) deps.ApiAddress = net.JoinHostPort(coordinatorAddr.String(), coordinatorPort) deps.CACert = []byte(b.inputs.caCert) - b.inputs.log.Info("mapped cluster.local to coordinator", "hostname", coordinatorHost, "addr", coordinatorAddr) b.inputs.log.Info("sandboxes will reach the cluster API at", "address", deps.ApiAddress) + if !coordinatorInternalIP.IsValid() { + // Older coordinators serve the registry on the same address as the API. + if err := hostMapper.SetHost("cluster.local", coordinatorAddr); err != nil { + return fmt.Errorf("mapping legacy cluster registry: %w", err) + } + b.inputs.log.Warn("coordinator did not advertise an internal address; using its API address for registry pulls", "addr", coordinatorAddr) + } } if b.inputs.clientCert == "" || b.inputs.clientKey == "" || b.inputs.caCert == "" { diff --git a/components/distributedrunner/boot_sandbox_host_test.go b/components/distributedrunner/boot_sandbox_host_test.go new file mode 100644 index 000000000..70d49b551 --- /dev/null +++ b/components/distributedrunner/boot_sandbox_host_test.go @@ -0,0 +1,35 @@ +//go:build linux + +package distributedrunner + +import ( + "net/netip" + "testing" + + "github.com/stretchr/testify/require" + "miren.dev/runtime/components/runner" +) + +func TestRegistryResolvesOverWireGuardNotPublicCoordinator(t *testing.T) { + boot := &sandboxHostBoot{inputs: sandboxHostBootInputs{ + log: testLogger(), coordinator: "198.51.100.9:8443", + }} + var deps runner.RunnerDeps + require.NoError(t, boot.prepareNetworkDeps(&deps, netip.MustParseAddr("10.8.42.1"))) + addr, err := deps.Resolver.LookupHost("cluster.local") + require.NoError(t, err) + require.Equal(t, netip.MustParseAddr("10.8.42.1"), addr) + require.Equal(t, "198.51.100.9:8443", deps.ApiAddress) +} + +func TestLegacyCoordinatorRegistryUsesAPIAddress(t *testing.T) { + boot := &sandboxHostBoot{inputs: sandboxHostBootInputs{ + log: testLogger(), coordinator: "198.51.100.9:8443", + }} + var deps runner.RunnerDeps + require.NoError(t, boot.prepareNetworkDeps(&deps, netip.Addr{})) + addr, err := deps.Resolver.LookupHost("cluster.local") + require.NoError(t, err) + require.Equal(t, netip.MustParseAddr("198.51.100.9"), addr) + require.Equal(t, "198.51.100.9:8443", deps.ApiAddress) +} diff --git a/components/runner/runner.go b/components/runner/runner.go index 863b21f05..139decfd2 100644 --- a/components/runner/runner.go +++ b/components/runner/runner.go @@ -237,8 +237,9 @@ func NewRunner(log *slog.Logger, deps RunnerDeps, cfg RunnerConfig) (*Runner, er // runner. It has no container or host-network responsibilities. type ClusterAccess struct { RunnerConfig - Log *slog.Logger - deps RunnerDeps + Log *slog.Logger + deps RunnerDeps + coordinatorInternalIP netip.Addr state *rpc.State eac *es.EntityAccessClient @@ -474,8 +475,8 @@ func (r *ClusterAccess) Start(ctx context.Context) (retErr error) { r.state = rs r.eac = es.NewEntityAccessClient(client) r.entityBase = entityserver.NewClient(r.Log, r.eac) - if err := r.setupRemoteWorkloadIssuer(ctx, rs); err != nil { - r.Log.Warn("failed to set up workload identity issuer", "error", err) + if err := r.setupRemoteCoordinatorInfo(ctx, rs); err != nil { + return fmt.Errorf("setting up coordinator registry and workload identity: %w", err) } if err := r.setupRemoteSecrets(rs); err != nil { return fmt.Errorf("setting up secret resolution: %w", err) @@ -601,6 +602,9 @@ func (r *ClusterAccess) WorkloadIssuer() workloadidentity.TokenIssuer { return r.deps.WorkloadIssuer } +// CoordinatorInternalIP is the coordinator's current WireGuard-routed bridge gateway. +func (r *ClusterAccess) CoordinatorInternalIP() netip.Addr { return r.coordinatorInternalIP } + // setupSqliteDisks connects to the coordinator's SQLite backup service so // sqlite-provider disks are replicated as they are written. // @@ -645,14 +649,10 @@ func (c sqliteDiskCloser) Close() error { return c.m.Close(ctx) } -// setupRemoteWorkloadIssuer wires a remote workload identity issuer for -// distributed runners. Runners do not hold the cluster signing key, so they -// mint tokens by calling the coordinator's RunnerRegistration service. When the -// coordinator reports no issuer is configured, token issuance stays disabled -// (deps.WorkloadIssuer remains nil). The coordinator's embedded runner -// (r.Config == nil) keeps the concrete issuer it was constructed with. -func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.State) error { - if r.Config == nil || r.deps.WorkloadIssuer != nil { +// setupRemoteCoordinatorInfo obtains the internal registry address and optional +// workload issuer from the coordinator. The embedded runner already has both. +func (r *ClusterAccess) setupRemoteCoordinatorInfo(ctx context.Context, rs *rpc.State) error { + if r.Config == nil { return nil } @@ -664,8 +664,8 @@ func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.S regClient := runner_v1alpha.NewRunnerRegistrationClient(client) // Retry transient failures: the entities connection was just established, so - // a failure here is usually a brief blip. Giving up immediately would leave - // the runner with no token issuance until it is restarted. + // a failure here is usually a brief blip. This result is required to set up + // registry routing, even when workload identity is disabled. var info *runner_v1alpha.RunnerRegistrationClientWorkloadIssuerInfoResults for attempt := 1; ; attempt++ { info, err = queryWorkloadIssuerInfo(ctx, regClient) @@ -683,7 +683,16 @@ func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.S case <-time.After(issuerInfoRetryDelay): } } + if info.HasCoordinatorInternalIp() { + r.coordinatorInternalIP, err = netip.ParseAddr(info.CoordinatorInternalIp()) + if err != nil || !r.coordinatorInternalIP.Is4() { + return fmt.Errorf("invalid coordinator internal IP %q", info.CoordinatorInternalIp()) + } + } + if r.deps.WorkloadIssuer != nil { + return nil + } if !info.Enabled() { r.Log.Info("coordinator has no workload identity issuer; sandbox tokens disabled") return nil diff --git a/components/server/boot_foundation.go b/components/server/boot_foundation.go index 5adc6cb95..f729977c7 100644 --- a/components/server/boot_foundation.go +++ b/components/server/boot_foundation.go @@ -25,17 +25,18 @@ type foundationBoot struct { output boot.Output[foundationBootOutput] } -func newFoundationBoot(config coordinate.CoordinatorConfig, ipDiscovery boot.Output[ipDiscoveryBootOutput], registration boot.Output[registrationBootOutput], identity boot.Output[workloadIdentityBootOutput], etcd boot.Output[etcdBootOutput], buildkit boot.Output[buildkitBootOutput], observability boot.Output[observabilityBootOutput]) *foundationBoot { +func newFoundationBoot(config coordinate.CoordinatorConfig, ipDiscovery boot.Output[ipDiscoveryBootOutput], registration boot.Output[registrationBootOutput], identity boot.Output[workloadIdentityBootOutput], etcd boot.Output[etcdBootOutput], buildkit boot.Output[buildkitBootOutput], registryHostMapping boot.Output[registryHostMappingBootOutput], observability boot.Output[observabilityBootOutput]) *foundationBoot { b := &foundationBoot{config: config} - b.component, b.output = boot.Provide6( - "cluster-foundation", ipDiscovery, registration, identity, etcd, buildkit, observability, + b.component, b.output = boot.Provide7( + "cluster-foundation", ipDiscovery, registration, identity, etcd, buildkit, registryHostMapping, observability, b.start, boot.WithStop(b.stop, componentStopTimeout), ) return b } -func (b *foundationBoot) start(ctx context.Context, ipDiscovery ipDiscoveryBootOutput, registration registrationBootOutput, identity workloadIdentityBootOutput, etcd etcdBootOutput, buildkit buildkitBootOutput, observability observabilityBootOutput) (foundationBootOutput, error) { +func (b *foundationBoot) start(ctx context.Context, ipDiscovery ipDiscoveryBootOutput, registration registrationBootOutput, identity workloadIdentityBootOutput, etcd etcdBootOutput, buildkit buildkitBootOutput, hostMapping registryHostMappingBootOutput, observability observabilityBootOutput) (foundationBootOutput, error) { config := b.config + config.CoordinatorInternalIP = hostMapping.registryIP config.IPs = ipDiscovery.ipSet config.CloudAuth = registration.cloudAuth config.WorkloadIssuer = identity.issuer diff --git a/components/server/boot_oci_registry.go b/components/server/boot_oci_registry.go index 50e25aed1..0d470b7a3 100644 --- a/components/server/boot_oci_registry.go +++ b/components/server/boot_oci_registry.go @@ -4,13 +4,13 @@ package server import ( "context" + "net" "miren.dev/runtime/components/ocireg" + "miren.dev/runtime/network" "miren.dev/runtime/pkg/boot" ) -const ociRegistryListenAddress = ":5000" - type ociRegistryBootInputs struct { dataPath string } @@ -26,26 +26,30 @@ func ociRegistryInputs(options StartOptions) ociRegistryBootInputs { return ociRegistryBootInputs{dataPath: options.Config.Server.GetDataPath()} } -func newOCIRegistryBoot(inputs ociRegistryBootInputs, identity boot.Output[workloadIdentityBootOutput], entityAccess boot.Output[entityAccessBootOutput], hostMapping *boot.Component, observability boot.Output[observabilityBootOutput]) *ociRegistryBoot { +func newOCIRegistryBoot(inputs ociRegistryBootInputs, identity boot.Output[workloadIdentityBootOutput], entityAccess boot.Output[entityAccessBootOutput], hostMapping boot.Output[registryHostMappingBootOutput], sandboxHost *boot.Component, observability boot.Output[observabilityBootOutput]) *ociRegistryBoot { b := &ociRegistryBoot{inputs: inputs} - b.component, b.output = boot.Provide3("oci-registry", identity, entityAccess, observability, b.start, - boot.DependsOn(hostMapping), + b.component, b.output = boot.Provide4("oci-registry", identity, entityAccess, hostMapping, observability, b.start, + boot.DependsOn(sandboxHost), boot.WithStop(b.stop, componentStopTimeout), ) return b } -func (b *ociRegistryBoot) start(ctx context.Context, identity workloadIdentityBootOutput, entityAccess entityAccessBootOutput, observability observabilityBootOutput) (struct{}, error) { +func (b *ociRegistryBoot) start(ctx context.Context, identity workloadIdentityBootOutput, entityAccess entityAccessBootOutput, hostMapping registryHostMappingBootOutput, observability observabilityBootOutput) (struct{}, error) { + if err := network.AllowRegistryFromWireGuard(); err != nil { + return struct{}{}, err + } + listenAddress := net.JoinHostPort(hostMapping.registryIP.String(), "5000") b.registry = ocireg.NewRegistry( b.inputs.dataPath, observability.log, entityAccess.client, identity.issuer, ) - if err := b.registry.Start(ctx, ociRegistryListenAddress); err != nil { + if err := b.registry.Start(ctx, listenAddress); err != nil { return struct{}{}, err } - observability.log.Info("OCI registry listening", "listen-address", ociRegistryListenAddress, "service-address", ocireg.Host) + observability.log.Info("OCI registry listening", "listen-address", listenAddress, "service-address", ocireg.Host) return struct{}{}, nil } diff --git a/components/server/startup.go b/components/server/startup.go index 910f1a44e..311522d2d 100644 --- a/components/server/startup.go +++ b/components/server/startup.go @@ -115,6 +115,7 @@ func newStartup(runtime *Runtime, options StartOptions) *startup { workloadIdentity.output, etcd.output, buildkit.output, + registryHostMapping.output, observability.output, ) appData := newAppDataBoot(foundation.output) @@ -160,7 +161,7 @@ func newStartup(runtime *Runtime, options StartOptions) *startup { serverInfo := newServerInfoBoot(instance, foundation.output) serverLifecycle := newServerLifecycleBoot(serverLifecycleInputsFrom(options), instance, foundation.output) cloudUplink := newCloudUplinkBoot(cloudControl.output, deploymentAttempts.output, ingress.output, serverLifecycle.output) - ociRegistry := newOCIRegistryBoot(ociRegistryInputs(options), workloadIdentity.output, entityAccess.output, registryHostMapping.component, observability.output) + ociRegistry := newOCIRegistryBoot(ociRegistryInputs(options), workloadIdentity.output, entityAccess.output, registryHostMapping.output, sandboxHost.component, observability.output) workAdmission := newWorkAdmissionBoot(applicationManagement.output, workloadControl.component, nodePresence.Component, buildkit.component, ociRegistry.component, registryHostMapping.component) buildSagaRecovery := newBuildSagaRecoveryBoot( buildSagaRecoveryInputs(options), diff --git a/docs/docs/distributed-runners.md b/docs/docs/distributed-runners.md index 592a8b5b2..f5b9217ca 100644 --- a/docs/docs/distributed-runners.md +++ b/docs/docs/distributed-runners.md @@ -199,6 +199,10 @@ Day-to-day fleet management happens through the `runner` subcommands. A quick to A typical maintenance window looks like: drain the node, do your work, then uncordon it (or remove it if it's not coming back). +:::warning[Upgrading to the internal-only registry] +When upgrading from a release that serves the registry on the coordinator's public address to one that serves it only over WireGuard, image pulls can briefly fail. Miren Cloud-managed upgrades update the coordinator first, then restart runners one at a time; each runner resumes pulling images after its upgrade. For manual upgrades, upgrade runner binaries first while the old coordinator still serves the registry, then upgrade the coordinator and restart the runners again so they learn its internal address. +::: + ## Things to know A few properties of distributed clusters are worth keeping in mind as you plan: diff --git a/docs/docs/firewall.md b/docs/docs/firewall.md index 27ce055a2..e38b85f9d 100644 --- a/docs/docs/firewall.md +++ b/docs/docs/firewall.md @@ -26,6 +26,10 @@ Host-level rules are managed for you. The only thing to configure yourself is yo Also open any `node_port` values your apps declare for TCP/UDP services, and the [inter-node ports](#between-nodes-distributed-runners) if you run distributed runners. +:::warning[Registry port 5000] +Starting with the release that routes registry pulls over WireGuard, the OCI registry listens only on the coordinator's Miren bridge gateway; you never need to open 5000/tcp on public interfaces. On older releases, it listens on all interfaces. If you still run one, block untrusted access to 5000/tcp with a host firewall (and a cloud security group, if applicable); allow only trusted runner addresses if you use distributed runners. +::: + ## How Miren Configures Firewall Rules :::info[Both iptables and nftables are required] @@ -50,7 +54,7 @@ The INPUT chain controls traffic destined for the host itself. Miren adds rules | Port | Protocol | Purpose | |------|----------|---------| | 53 | UDP/TCP | DNS resolution (containers query host DNS) | -| 5000 | TCP | Local container registry (buildkit pushes images here) | +| 5000 | TCP | Cluster-internal OCI registry (BuildKit pushes and runners pull images) | ## Rule Ordering @@ -89,6 +93,8 @@ If you've grown the cluster with [distributed runners](./distributed-runners.md) **The overlay port is the one people miss.** Sandboxes on different machines send traffic directly to each other, so 51820/udp has to be open between every pair of nodes, not just from each runner back to the coordinator. +Runners pull from the coordinator's internal bridge address over WireGuard. Keep 51820/udp available between nodes; you do not need to open 5000/tcp between their public interfaces. + **Telemetry doesn't need its own ports.** VictoriaMetrics and VictoriaLogs stay bound to loopback on the coordinator and are never reachable from a runner. A runner ships its metrics and logs over the coordinator API on 8443 like everything else it sends. ### Outbound Connectivity diff --git a/network/bridge.go b/network/bridge.go index 8b19cc1b7..3025df5bd 100644 --- a/network/bridge.go +++ b/network/bridge.go @@ -394,6 +394,28 @@ func enableBridgeInputRules(br netlink.Link, apiPort int) error { return nil } +// AllowRegistryFromWireGuard restricts the coordinator's bridge-bound registry +// to local traffic and overlay peers, even if a public interface receives a +// packet addressed to the bridge gateway (Linux accepts local IPs on any link). +func AllowRegistryFromWireGuard() error { + ipt, err := iptables.NewWithProtocol(iptables.ProtocolIPv4) + if err != nil { + return err + } + match := []string{"-d", "10.8.0.0/16", "-p", "tcp", "--dport", "5000"} + if err := ipt.InsertUnique("filter", "INPUT", 1, + append(append([]string{}, match...), "-j", "DROP")...); err != nil { + return err + } + for _, iface := range []string{"lo", "rt0", "flannel-wg"} { + if err := ipt.InsertUnique("filter", "INPUT", 1, + append(append([]string{"-i", iface}, match...), "-j", "ACCEPT")...); err != nil { + return err + } + } + return nil +} + func ConfigureGW(br netlink.Link, ec *EndpointConfig) error { for _, ac := range ec.Bridge.Addresses { gwIP := netipx.PrefixIPNet(ac) diff --git a/network/bridge_registry_test.go b/network/bridge_registry_test.go new file mode 100644 index 000000000..c0f07bd00 --- /dev/null +++ b/network/bridge_registry_test.go @@ -0,0 +1,66 @@ +//go:build linux + +package network + +import ( + "os" + "strings" + "testing" + + "github.com/coreos/go-iptables/iptables" + "github.com/stretchr/testify/require" +) + +func TestAllowRegistryFromWireGuardRules(t *testing.T) { + if os.Geteuid() != 0 { + t.Skip("requires root to inspect the INPUT chain") + } + ipt, err := iptables.NewWithProtocol(iptables.ProtocolIPv4) + require.NoError(t, err) + + match := []string{"-d", "10.8.0.0/16", "-p", "tcp", "--dport", "5000"} + legacy := []string{"-i", "rt0", "-p", "tcp", "--dport", "5000", "-j", "ACCEPT"} + rules := [][]string{legacy, append(append([]string{}, match...), "-j", "DROP")} + for _, iface := range []string{"lo", "rt0", "flannel-wg"} { + rules = append(rules, append(append([]string{"-i", iface}, match...), "-j", "ACCEPT")) + } + for _, rule := range rules { + present, err := ipt.Exists("filter", "INPUT", rule...) + require.NoError(t, err) + if !present { + rule := rule + t.Cleanup(func() { require.NoError(t, ipt.DeleteIfExists("filter", "INPUT", rule...)) }) + } + } + require.NoError(t, ipt.InsertUnique("filter", "INPUT", 1, legacy...)) + require.NoError(t, AllowRegistryFromWireGuard()) + + installed, err := ipt.List("filter", "INPUT") + require.NoError(t, err) + allowed := map[string]int{} + denied, oldAccept := -1, -1 + for i, rule := range installed { + if !strings.Contains(rule, "--dport 5000") { + continue + } + switch { + case strings.Contains(rule, "-d 10.8.0.0/16") && strings.Contains(rule, "-j DROP"): + denied = i + case strings.Contains(rule, "-d 10.8.0.0/16") && strings.Contains(rule, "-j ACCEPT"): + for _, iface := range []string{"lo", "rt0", "flannel-wg"} { + if strings.Contains(rule, "-i "+iface) { + allowed[iface] = i + } + } + case strings.Contains(rule, "-i rt0") && strings.Contains(rule, "-j ACCEPT"): + oldAccept = i + } + } + require.GreaterOrEqual(t, denied, 0) + for _, iface := range []string{"lo", "rt0", "flannel-wg"} { + index, ok := allowed[iface] + require.True(t, ok, "missing registry accept for %s", iface) + require.Less(t, index, denied, "%s must be allowed before the registry drop", iface) + } + require.Greater(t, oldAccept, denied, "registry drop must precede old bridge allow") +} diff --git a/servers/runner/registration.go b/servers/runner/registration.go index 063ddb96c..dd2b3e514 100644 --- a/servers/runner/registration.go +++ b/servers/runner/registration.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "net/netip" "slices" "strings" "time" @@ -36,12 +37,13 @@ const ( ) type RegistrationServerConfig struct { - Log *slog.Logger - Authority *caauth.Authority - EAC *entityserver_v1alpha.EntityAccessClient - CoordinatorAddr string - EtcdEndpoints []string - EtcdPrefix string + Log *slog.Logger + Authority *caauth.Authority + EAC *entityserver_v1alpha.EntityAccessClient + CoordinatorAddr string + CoordinatorInternalIP netip.Addr + EtcdEndpoints []string + EtcdPrefix string // Observability endpoints provided to runners at join time VictoriametricsAddress string @@ -960,12 +962,13 @@ func (s *RegistrationServer) DrainRunner(ctx context.Context, req *runner_v1alph return nil } -// WorkloadIssuerInfo reports whether the coordinator has a workload identity -// issuer configured and, if so, its issuer URL. Distributed runners call this -// once at startup to decide whether to mint workload identity tokens via the -// coordinator. +// WorkloadIssuerInfo reports the issuer and current coordinator internal address. +// Distributed runners read both at startup, including when identity is disabled. func (s *RegistrationServer) WorkloadIssuerInfo(ctx context.Context, req *runner_v1alpha.RunnerRegistrationWorkloadIssuerInfo) error { results := req.Results() + if s.CoordinatorInternalIP.IsValid() { + results.SetCoordinatorInternalIp(s.CoordinatorInternalIP.String()) + } if s.WorkloadIssuer == nil { results.SetEnabled(false) diff --git a/servers/runner/registration_test.go b/servers/runner/registration_test.go index d99f48a08..fc6660cfb 100644 --- a/servers/runner/registration_test.go +++ b/servers/runner/registration_test.go @@ -5,6 +5,7 @@ import ( "crypto/x509" "encoding/pem" "net" + "net/netip" "slices" "testing" "time" @@ -56,6 +57,32 @@ func newTestServer(t *testing.T) (*testEnv, func()) { return &testEnv{client: client, ec: es.Client, store: es.Store, server: regServer, ca: ca}, cleanup } +func TestWorkloadIssuerInfoPublishesCoordinatorInternalIPWithoutIssuer(t *testing.T) { + env, cleanup := newTestServer(t) + defer cleanup() + env.server.CoordinatorInternalIP = netip.MustParseAddr("10.8.42.1") + + result, err := env.client.WorkloadIssuerInfo(t.Context()) + if err != nil { + t.Fatal(err) + } + if result.Enabled() || result.CoordinatorInternalIp() != "10.8.42.1" { + t.Fatalf("unexpected coordinator internal address: enabled=%t ip=%q", result.Enabled(), result.CoordinatorInternalIp()) + } +} + +func TestWorkloadIssuerInfoInternalAddressJSONField(t *testing.T) { + var result runner_v1alpha.RunnerRegistrationWorkloadIssuerInfoResults + result.SetCoordinatorInternalIp("10.8.42.1") + data, err := result.MarshalJSON() + if err != nil { + t.Fatal(err) + } + if string(data) != `{"coordinator_internal_ip":"10.8.42.1"}` { + t.Fatalf("unexpected coordinator internal address field: %s", data) + } +} + // issueLeafCert issues a certificate from the given authority and returns the // parsed leaf certificate (the first PEM block; IssueCertificate appends the CA // cert after it).