relay: add static-IP egress relay service (TLS 1.3 + yamux)

cmd/relay.go

@@ -0,0 +1,193 @@
+package cmd
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"net/netip"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/spf13/cobra"
+
+	"github.com/modal-labs/vprox/lib"
+	"github.com/modal-labs/vprox/lib/relay"
+)
+
+var RelayCmd = &cobra.Command{
+	Use:   "relay",
+	Short: "Start the static-IP egress relay for macOS hosts (TLS 1.3 + yamux)",
+	Long: `Start the egress relay. A macOS host's egress proxy opens one TLS 1.3
+session per VM, authenticates with VPROX_PASSWORD, and multiplexes that VM's
+TCP and UDP flows over it. The relay dials each destination from the static
+IP the session was granted, after checking it against the destination policy.`,
+	RunE: runRelay,
+}
+
+var relayCmdArgs struct {
+	listen        string
+	allowSources  []string
+	denyCIDRs     []string
+	denyPorts     []uint
+	tlsCert       string
+	tlsKey        string
+	metricsAddr   string
+	maxSessions   int
+	maxPerSource  int
+	maxStreams    int
+	maxUDP        int
+	streamWindow  uint32
+	acceptBacklog int
+	dialTimeout   time.Duration
+	tcpIdle       time.Duration
+	udpIdle       time.Duration
+	drainTimeout  time.Duration
+}
+
+func init() {
+	f := RelayCmd.Flags()
+	def := relay.DefaultLimits()
+	f.StringVar(&relayCmdArgs.listen, "listen", "0.0.0.0:9443",
+		"address to accept relay sessions on")
+	f.StringArrayVar(&relayCmdArgs.allowSources, "allow-source", nil,
+		"source CIDR allowed to open sessions (repeatable, required)")
+	f.StringArrayVar(&relayCmdArgs.denyCIDRs, "deny-cidr", nil,
+		"destination CIDR to refuse in addition to the built-in list (repeatable)")
+	f.UintSliceVar(&relayCmdArgs.denyPorts, "deny-port", nil,
+		"destination port to refuse (repeatable)")
+	f.StringVar(&relayCmdArgs.tlsCert, "tls-cert", "", "TLS certificate PEM (default: embedded)")
+	f.StringVar(&relayCmdArgs.tlsKey, "tls-key", "", "TLS key PEM (default: embedded)")
+	f.StringVar(&relayCmdArgs.metricsAddr, "metrics-addr", "127.0.0.1:9444",
+		"address for the Prometheus /metrics endpoint (empty disables)")
+	f.IntVar(&relayCmdArgs.maxSessions, "max-sessions", def.MaxSessions, "max concurrent sessions")
+	f.IntVar(&relayCmdArgs.maxPerSource, "max-sessions-per-source", def.MaxSessionsPerSource,
+		"max concurrent sessions per source IP")
+	f.IntVar(&relayCmdArgs.maxStreams, "max-streams-per-session", def.MaxStreamsPerSession,
+		"max concurrent streams (TCP+UDP) per session")
+	f.IntVar(&relayCmdArgs.maxUDP, "max-udp-per-session", def.MaxUDPPerSession,
+		"max concurrent UDP flows per session")
+	f.Uint32Var(&relayCmdArgs.streamWindow, "stream-window", def.StreamWindow,
+		"yamux per-stream receive window in bytes")
+	f.IntVar(&relayCmdArgs.acceptBacklog, "accept-backlog", def.AcceptBacklog,
+		"yamux per-session stream accept backlog")
+	f.DurationVar(&relayCmdArgs.dialTimeout, "dial-timeout", def.DialTimeout, "destination dial timeout")
+	f.DurationVar(&relayCmdArgs.tcpIdle, "tcp-idle", def.TCPIdle, "idle timeout for TCP streams")
+	f.DurationVar(&relayCmdArgs.udpIdle, "udp-idle", def.UDPIdle, "idle timeout for UDP flows")
+	f.DurationVar(&relayCmdArgs.drainTimeout, "drain-timeout", def.DrainTimeout,
+		"how long open streams may finish after SIGTERM before they are cut")
+}
+
+func runRelay(cmd *cobra.Command, args []string) error {
+	var cert tls.Certificate
+	var err error
+	if relayCmdArgs.tlsCert != "" || relayCmdArgs.tlsKey != "" {
+		if relayCmdArgs.tlsCert == "" || relayCmdArgs.tlsKey == "" {
+			return errors.New("--tls-cert and --tls-key must be set together")
+		}
+		cert, err = tls.LoadX509KeyPair(relayCmdArgs.tlsCert, relayCmdArgs.tlsKey)
+	} else {
+		cert, err = lib.LoadServerTLS()
+	}
+	if err != nil {
+		return err
+	}
+	if len(relayCmdArgs.allowSources) == 0 {
+		return errors.New("missing required flag: --allow-source")
+	}
+	allow, err := parsePrefixes(relayCmdArgs.allowSources)
+	if err != nil {
+		return fmt.Errorf("--allow-source: %v", err)
+	}
+	deny, err := parsePrefixes(relayCmdArgs.denyCIDRs)
+	if err != nil {
+		return fmt.Errorf("--deny-cidr: %v", err)
+	}
+	password, err := lib.GetVproxPassword()
+	if err != nil {
+		return err
+	}
+
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(collectors.NewGoCollector(), collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
+	denyPorts := make([]uint16, 0, len(relayCmdArgs.denyPorts))
+	for _, p := range relayCmdArgs.denyPorts {
+		if p == 0 || p > 65535 {
+			return fmt.Errorf("--deny-port: invalid port %d", p)
+		}
+		denyPorts = append(denyPorts, uint16(p))
+	}
+	policy := relay.NewPolicy(relay.PolicyConfig{
+		ExtraDenyPrefixes: deny,
+		DenyPorts:         denyPorts,
+	})
+	srv, err := relay.NewServer(relay.Config{
+		TLS:          &tls.Config{Certificates: []tls.Certificate{cert}},
+		Password:     password,
+		AllowSources: allow,
+		Policy:       policy,
+		Metrics:      relay.NewMetrics(reg),
+		Limits: relay.Limits{
+			MaxSessions:          relayCmdArgs.maxSessions,
+			MaxSessionsPerSource: relayCmdArgs.maxPerSource,
+			MaxStreamsPerSession: relayCmdArgs.maxStreams,
+			MaxUDPPerSession:     relayCmdArgs.maxUDP,
+			StreamWindow:         relayCmdArgs.streamWindow,
+			AcceptBacklog:        relayCmdArgs.acceptBacklog,
+			DialTimeout:          relayCmdArgs.dialTimeout,
+			TCPIdle:              relayCmdArgs.tcpIdle,
+			UDPIdle:              relayCmdArgs.udpIdle,
+			DrainTimeout:         relayCmdArgs.drainTimeout,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	if relayCmdArgs.metricsAddr != "" {
+		mux := http.NewServeMux()
+		mux.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
+		ms := &http.Server{Addr: relayCmdArgs.metricsAddr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
+		go func() {
+			if err := ms.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+				log.Printf("relay: metrics server: %v", err)
+			}
+		}()
+		defer func() {
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+			defer cancel()
+			_ = ms.Shutdown(shutdownCtx)
+		}()
+	}
+
+	log.Printf("relay: allow sources %v; deny %s", allow, policy)
+
+	ln, err := net.Listen("tcp", relayCmdArgs.listen)
+	if err != nil {
+		return fmt.Errorf("listen %s: %v", relayCmdArgs.listen, err)
+	}
+	log.Printf("relay: listening on %s", ln.Addr())
+	return srv.Serve(ctx, ln)
+}
+
+func parsePrefixes(strs []string) ([]netip.Prefix, error) {
+	out := make([]netip.Prefix, 0, len(strs))
+	for _, s := range strs {
+		p, err := netip.ParsePrefix(s)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, p.Masked())
+	}
+	return out, nil
+}

docs/RELAY.md

@@ -0,0 +1,97 @@
+# Egress relay (`vprox relay`)
+
+The relay is the vprox side of the macOS static-IP userspace dataplane. A
+macOS host runs a root `egress-proxy` daemon that receives VM traffic via pf
+`rdr`, opens **one TLS 1.3 connection per VM** to the relay, and multiplexes
+that VM's TCP and UDP flows over it with yamux. The relay dials every
+destination from the static IP the session was granted, so the VM's egress
+address is the static IP without any tunnel or `route-to`.
+
+```
+VM -> pf rdr -> egress-proxy (mac, root) -> TLS1.3/yamux -> vprox relay -> dst
+                                                            bind(static_ip)
+```
+
+## Running
+
+```
+VPROX_PASSWORD=... vprox relay \
+  --listen 0.0.0.0:9443 \
+  --allow-source 205.234.200.0/24 \
+  --deny-cidr 10.100.0.0/16 --deny-cidr 10.0.0.0/24 \
+  --metrics-addr 127.0.0.1:9444
+```
+
+* `--allow-source` is required; connections from other sources are dropped
+  before the TLS handshake. Use the same CIDRs as the `/connect-ipip` UFW rule.
+* The TLS certificate defaults to the embedded vprox certificate. The host
+  proxy does not verify it against a CA; instead both ends prove knowledge of
+  `VPROX_PASSWORD` over the TLS channel (HMAC-SHA256 of the TLS exporter
+  value `EXPORTER-blacksmith-relay`, client and server roles). The password
+  never crosses the wire and a proof cannot be replayed on another
+  connection; a relay without the password is refused by the host proxy.
+* SIGTERM drains: the listener closes, every session gets a yamux GoAway so
+  the host proxy reconnects, and open streams get `--drain-timeout` to finish.
+
+## Destination policy
+
+The relay dials as a local process, so the FORWARD-chain rules that guard the
+IPIP path do not apply. Every stream is checked before dialling; refusals are
+reported to the host proxy with `StreamPolicyDenied` and counted in
+`vprox_relay_streams_refused_total{status="policy_denied"}`.
+
+Always denied: `0.0.0.0/8`, RFC1918 (`10/8`, `172.16/12`, `192.168/16`),
+`100.64.0.0/10`, `127.0.0.0/8`, `169.254.0.0/16`, `224.0.0.0/4`,
+`240.0.0.0/4`, IPv6, port 0, and every address currently assigned to one of
+the relay host's interfaces (refreshed every 10s, so new static IPs and
+tunnel addresses are covered without a restart). `--deny-cidr` and
+`--deny-port` add to that list; pass the WireGuard block and the internal
+network CIDR.
+
+## Limits
+
+| flag | default | refusal |
+|---|---|---|
+| `--max-sessions` | 4096 | `HelloSessionLimit` |
+| `--max-sessions-per-source` | 64 | `HelloSessionLimit` |
+| `--max-streams-per-session` | 4096 | `StreamLimit` |
+| `--max-udp-per-session` | 1024 | `StreamUDPLimit` |
+| `--stream-window` | 256 KiB | (yamux receive window) |
+| `--dial-timeout` | 5s | `StreamDialFailed` |
+| `--tcp-idle` | 6h | stream closed |
+| `--udp-idle` | 60s | stream closed |
+| `--drain-timeout` | 15s | streams cut on shutdown |
+
+The hello must arrive within 10s of the TCP accept.
+
+## Metrics (`--metrics-addr`, default `127.0.0.1:9444`)
+
+`vprox_relay_sessions_open`, `vprox_relay_sessions_total`,
+`vprox_relay_hello_refused_total{status}`, `vprox_relay_streams_open{proto}`,
+`vprox_relay_streams_total{proto}`, `vprox_relay_streams_refused_total{status}`,
+`vprox_relay_bytes_total{dir,static_ip}`, `vprox_relay_dial_seconds{proto}`,
+`vprox_relay_source_rejected_total`, `vprox_relay_draining`.
+
+## Wire protocol v1 (`lib/relayproto`)
+
+All integers big-endian. The same package, with identical hex fixtures,
+lives in FastActions/fa as `agent/egressproxy/relayproto`; the fixtures are
+the compatibility contract between the two modules.
+
+```
+client hello   "BRLY" | u8 version=1 | u16 len | JSON {"auth","vm_id","static_ip"}
+server reply   u8 status | u16 len | JSON {"egress_ip"} or {"error"}
+   status: 0 ok, 1 unauthorized, 2 bad_static_ip, 3 session_limit,
+           4 protocol_error, 5 unavailable
+-- yamux from here; each stream: --
+stream header  u8 proto (6=TCP, 17=UDP) | 4B IPv4 dst | u16 dst port
+stream reply   u8 status | u32 dial duration (us)
+   status: 0 ok, 1 dial_failed, 2 policy_denied, 3 limit, 4 udp_limit,
+           5 bad_header, 6 unavailable
+TCP payload    raw bytes; yamux FIN = half-close in that direction
+UDP payload    repeated: u16 len | datagram (len <= 65535, 0 allowed)
+```
+
+The hello JSON is capped at 4096 bytes. The static IP in the hello must be
+an address of the relay host or the session is refused with `bad_static_ip`;
+the reply's `egress_ip` is the address the relay binds outbound sockets to.

go.mod

@@ -6,33 +6,44 @@ require (
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.12
 	github.com/coreos/go-iptables v0.8.0
 	github.com/fatih/color v1.17.0
+	github.com/hashicorp/yamux v0.1.2
 	github.com/prometheus-community/pro-bing v0.4.1
+	github.com/prometheus/client_golang v1.23.2
 	github.com/spf13/cobra v1.8.1
-	github.com/stretchr/testify v1.9.0
+	github.com/stretchr/testify v1.11.1
 	github.com/vishvananda/netlink v1.3.0
 	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230429144221-925a1e7659e6
 )
 
 require (
 	github.com/aws/aws-sdk-go-v2 v1.30.4 // indirect
 	github.com/aws/smithy-go v1.20.4 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/native v1.1.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mdlayher/genetlink v1.3.2 // indirect
 	github.com/mdlayher/netlink v1.7.2 // indirect
 	github.com/mdlayher/socket v0.5.1 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.66.1 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/vishvananda/netns v0.0.4 // indirect
-	golang.org/x/crypto v0.27.0 // indirect
-	golang.org/x/net v0.29.0 // indirect
-	golang.org/x/sync v0.8.0 // indirect
-	golang.org/x/sys v0.25.0 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	golang.org/x/crypto v0.41.0 // indirect
+	golang.org/x/net v0.43.0 // indirect
+	golang.org/x/sync v0.13.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
 	golang.zx2c4.com/wireguard v0.0.0-20231211153847-12269c276173 // indirect
+	google.golang.org/protobuf v1.36.8 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )