Maven/internal/netaddr/netaddr.go

// Package netaddr parses a daemon seam address and dials or binds it.
//
// Every seam between Maven's daemons used to be a unix socket with the
// network hardcoded at the call site — two dials in internal/ipc, one listen,
// and the same pair again in internal/worker. That is correct for co-located
// daemons and it is the reason a module cannot live on another host. This
// package moves the choice into the address string so a deploy picks the
// transport, not a recompile:
//
//	/run/maven/stt.sock                   unix   (the default, unchanged)
//	unix:///run/maven/stt.sock            unix   (explicit, same thing)
//	tcp://workstation:9310?token=hunter2  tcp
//
// A scheme-less address is unix and behaves exactly as it did before this
// package existed: same 0700 parent dir, same 0600 socket, same bytes on the
// wire with no handshake in front of them.
//
// Over TCP the filesystem permission that authenticated the unix socket is
// gone, and what crosses this seam is audio of the owner speaking and the
// text of his turns. So a TCP seam carries a shared token, checked before the
// first protocol frame is read. Wireguard is supported underneath and is not
// required.
package netaddr

import (
	"crypto/subtle"
	"errors"
	"fmt"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"golang.org/x/sys/unix"
)

// ErrUnauthorized — the peer presented a token the listener does not accept,
// or presented none when one is required.
var ErrUnauthorized = errors.New("netaddr: unauthorized")

// Addr is a parsed seam endpoint.
type Addr struct {
	// Network is "unix" or "tcp".
	Network string
	// Address is the socket path (unix) or host:port (tcp).
	Address string
	// Token is the shared secret for a tcp seam. Empty for unix, where the
	// filesystem does the same job.
	Token string
}

// String renders the address for logs and errors. The token is never included.
func (a Addr) String() string {
	if a.Network == "unix" {
		return a.Address
	}
	return a.Network + "://" + a.Address
}

// IsUnix reports whether this seam is a unix socket, and so is local, is
// authenticated by file permissions, and needs no handshake.
func (a Addr) IsUnix() bool { return a.Network == "unix" }

// Parse reads a seam address. Anything without a "scheme://" prefix is a unix
// socket path, which keeps every existing config and every default working
// untouched.
func Parse(s string) (Addr, error) {
	if !strings.Contains(s, "://") {
		return Addr{Network: "unix", Address: s}, nil
	}
	u, err := url.Parse(s)
	if err != nil {
		return Addr{}, fmt.Errorf("netaddr: parse %q: %w", s, err)
	}
	switch u.Scheme {
	case "unix":
		return Addr{Network: "unix", Address: u.Path}, nil
	case "tcp":
		if u.Host == "" {
			return Addr{}, fmt.Errorf("netaddr: %q has no host:port", s)
		}
		return Addr{Network: "tcp", Address: u.Host, Token: u.Query().Get("token")}, nil
	default:
		return Addr{}, fmt.Errorf("netaddr: unsupported scheme %q", u.Scheme)
	}
}

// MustParse is Parse for a literal known good at compile time. It panics on a
// bad address, so use it in tests and constants, never on config input.
func MustParse(s string) Addr {
	a, err := Parse(s)
	if err != nil {
		panic(err)
	}
	return a
}

// handshakeTimeout bounds the token exchange. A peer that cannot write one
// short line in this long is not going to serve a turn either.
const handshakeTimeout = 5 * time.Second

// greeting prefixes the token line. Versioned so a later mTLS seam can be
// told apart from this one on the wire.
const greeting = "MAVEN1 "

// Dial connects to a. On a tcp seam it sends the token and waits for the
// listener to accept it, so a returned conn is already authorized and the
// caller can write its first protocol frame.
func Dial(a Addr) (net.Conn, error) {
	return DialTimeout(a, 0)
}

// DialTimeout is Dial with a bound on the connect. Zero means the operating
// system default. The token exchange gets its own timeout either way.
func DialTimeout(a Addr, timeout time.Duration) (net.Conn, error) {
	var c net.Conn
	var err error
	if timeout > 0 {
		c, err = net.DialTimeout(a.Network, a.Address, timeout)
	} else {
		c, err = net.Dial(a.Network, a.Address)
	}
	if err != nil {
		return nil, err
	}
	if a.IsUnix() {
		return c, nil
	}
	if err := clientHandshake(c, a.Token); err != nil {
		_ = c.Close()
		return nil, err
	}
	return c, nil
}

func clientHandshake(c net.Conn, token string) error {
	_ = c.SetDeadline(time.Now().Add(handshakeTimeout))
	defer c.SetDeadline(time.Time{})
	if _, err := c.Write([]byte(greeting + token + "\n")); err != nil {
		return fmt.Errorf("netaddr: send token: %w", err)
	}
	var reply [3]byte
	if _, err := readFull(c, reply[:]); err != nil {
		return fmt.Errorf("%w: %v", ErrUnauthorized, err)
	}
	if string(reply[:]) != "ok\n" {
		return ErrUnauthorized
	}
	return nil
}

// Listener wraps a net.Listener so Accept performs the token check for a tcp
// seam. A connection that fails the check is closed and never surfaces, so
// the protocol above this layer only ever sees authorized peers.
//
// A unix seam takes none of that machinery: Accept delegates straight to the
// wrapped listener, which is what it did before the token existed.
type Listener struct {
	net.Listener
	addr Addr

	start     sync.Once
	closeOnce sync.Once
	conns     chan net.Conn
	errc      chan error // buffered 1, re-armed so every Accept sees the error
	done      chan struct{}
}

// Accept returns the next authorized connection. Unauthorized peers are
// dropped and Accept keeps waiting: a bad token is a rejected stranger, not a
// reason to stop serving.
//
// Each tcp handshake runs in its own goroutine rather than inline here. A peer
// that connects and then says nothing holds its greeting open for
// handshakeTimeout, and inline that peer stalls every other connection for
// five seconds — one silent stranger was enough to freeze the seam.
func (l *Listener) Accept() (net.Conn, error) {
	if l.addr.IsUnix() {
		return l.Listener.Accept()
	}
	l.start.Do(func() { go l.acceptLoop() })
	select {
	case c := <-l.conns:
		return c, nil
	case err := <-l.errc:
		l.errc <- err
		return nil, err
	}
}

// acceptLoop takes connections off the wrapped listener and greets each one
// concurrently. It ends on the first listener error, which every later Accept
// then reports.
func (l *Listener) acceptLoop() {
	for {
		c, err := l.Listener.Accept()
		if err != nil {
			select {
			case l.errc <- err:
			case <-l.done:
			}
			return
		}
		go l.greet(c)
	}
}

func (l *Listener) greet(c net.Conn) {
	if err := serverHandshake(c, l.addr.Token); err != nil {
		_ = c.Close()
		return
	}
	select {
	case l.conns <- c:
	case <-l.done:
		_ = c.Close()
	}
}

// Close stops the listener and releases any connection still waiting to be
// handed to Accept.
func (l *Listener) Close() error {
	l.closeOnce.Do(func() { close(l.done) })
	return l.Listener.Close()
}

// Addr reports the parsed seam address this listener was built from.
func (l *Listener) SeamAddr() Addr { return l.addr }

func serverHandshake(c net.Conn, want string) error {
	_ = c.SetDeadline(time.Now().Add(handshakeTimeout))
	defer c.SetDeadline(time.Time{})
	// The line is bounded: greeting, token, newline. Read a byte at a time so
	// nothing of the first protocol frame is consumed when the token is short.
	line := make([]byte, 0, 128)
	var b [1]byte
	for {
		if _, err := readFull(c, b[:]); err != nil {
			return err
		}
		if b[0] == '\n' {
			break
		}
		line = append(line, b[0])
		if len(line) > 512 {
			return ErrUnauthorized
		}
	}
	got, ok := strings.CutPrefix(string(line), greeting)
	if !ok {
		return ErrUnauthorized
	}
	if subtle.ConstantTimeCompare([]byte(got), []byte(want)) != 1 {
		return ErrUnauthorized
	}
	if _, err := c.Write([]byte("ok\n")); err != nil {
		return err
	}
	return nil
}

func readFull(c net.Conn, p []byte) (int, error) {
	n := 0
	for n < len(p) {
		m, err := c.Read(p[n:])
		n += m
		if err != nil {
			return n, err
		}
	}
	return n, nil
}

// Listen binds a. A unix seam gets the perms it has always had: parent dir
// 0700, socket 0600, and any stale socket from a crashed daemon removed
// first. A tcp seam must carry a token, because there is no filesystem to
// stand in for one.
func Listen(a Addr) (*Listener, error) {
	if a.IsUnix() {
		ln, err := listenUnix(a.Address)
		if err != nil {
			return nil, err
		}
		return wrap(ln, a), nil
	}
	if a.Token == "" {
		return nil, fmt.Errorf("netaddr: listen %s: tcp seam requires a token", a)
	}
	ln, err := net.Listen("tcp", a.Address)
	if err != nil {
		return nil, fmt.Errorf("netaddr: listen %s: %w", a, err)
	}
	return wrap(ln, a), nil
}

func wrap(ln net.Listener, a Addr) *Listener {
	return &Listener{
		Listener: ln,
		addr:     a,
		conns:    make(chan net.Conn),
		errc:     make(chan error, 1),
		done:     make(chan struct{}),
	}
}

func listenUnix(path string) (net.Listener, error) {
	_ = os.Remove(path) // stale socket from a crashed daemon; ignore missing
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return nil, fmt.Errorf("netaddr: mkdir socket dir: %w", err)
	}
	// umask could widen the perms on socket creation; tighten then chmod to
	// be explicit. 0600 ⇒ read+write by owner only.
	oldMask := unix.Umask(0o077)
	ln, err := net.Listen("unix", path)
	unix.Umask(oldMask)
	if err != nil {
		return nil, fmt.Errorf("netaddr: listen %s: %w", path, err)
	}
	if err := os.Chmod(path, 0o600); err != nil {
		_ = ln.Close()
		_ = os.Remove(path)
		return nil, fmt.Errorf("netaddr: chmod socket: %w", err)
	}
	return ln, nil
}

// Cleanup removes the socket file behind a unix seam. It is a no-op for tcp.
func Cleanup(a Addr) {
	if a.IsUnix() && a.Address != "" {
		_ = os.Remove(a.Address)
	}
}