ipc, worker: dial and bind through netaddr (V-484)
Five hardcoded transports, three in internal/ipc and two in internal/worker, all now go through the seam address. The unix perms logic moved into netaddr, so the two copies of parentDir and the umask dance are gone. peerCaller already returned ok=false for a non-unix conn, so the SO_PEERCRED path degrades correctly on tcp with no change.
This commit is contained in:
+24
-13
@@ -8,13 +8,15 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/netaddr"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Client — the module side of the boundary. Wraps a unix-socket connection
|
// Client — the module side of the boundary. Wraps a connection to core and
|
||||||
// and satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is
|
// satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is
|
||||||
// agnostic to whether it's been wired in-process (tests / daemon-embedded)
|
// agnostic to whether it's been wired in-process (tests / daemon-embedded),
|
||||||
// or over this socket (full topology). The swappability is the seam auth
|
// over a local unix socket, or over tcp to another host. The swappability is
|
||||||
// will insert into without touching module code.
|
// the seam auth will insert into without touching module code.
|
||||||
//
|
//
|
||||||
// One Client ⇒ one conn ⇒ one concurrent request at a time. A module that
|
// One Client ⇒ one conn ⇒ one concurrent request at a time. A module that
|
||||||
// wants parallel requests opens one Client per goroutine; the store is the
|
// wants parallel requests opens one Client per goroutine; the store is the
|
||||||
@@ -22,7 +24,8 @@ import (
|
|||||||
// per-Client lock keeps frame interleaving impossible by construction.
|
// per-Client lock keeps frame interleaving impossible by construction.
|
||||||
type Client struct {
|
type Client struct {
|
||||||
conn net.Conn
|
conn net.Conn
|
||||||
path string // kept so a dropped conn can be re-dialed (core restart)
|
path string // the address as configured, kept for errors and logs
|
||||||
|
addr netaddr.Addr // parsed, so a dropped conn can be re-dialed (core restart)
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -81,14 +84,22 @@ var readOnlyMethods = map[Method]bool{
|
|||||||
MethodPing: true,
|
MethodPing: true,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Dial connects to a core socket at path and returns a Client. The module
|
// Dial connects to core at path and returns a Client. The module owns its
|
||||||
// owns its Client lifecycle; Close on shutdown.
|
// Client lifecycle; Close on shutdown.
|
||||||
|
//
|
||||||
|
// path is a netaddr seam address: a bare path is the unix socket it has
|
||||||
|
// always been, and "tcp://host:port?token=..." reaches a core on another
|
||||||
|
// host. See internal/netaddr.
|
||||||
func Dial(path string) (*Client, error) {
|
func Dial(path string) (*Client, error) {
|
||||||
c, err := net.Dial("unix", path)
|
addr, err := netaddr.Parse(path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("ipc: dial %s: %w", path, err)
|
return nil, err
|
||||||
}
|
}
|
||||||
return &Client{conn: c, path: path}, nil
|
c, err := netaddr.Dial(addr)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("ipc: dial %s: %w", addr, err)
|
||||||
|
}
|
||||||
|
return &Client{conn: c, path: path, addr: addr}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Client) Close() error {
|
func (c *Client) Close() error {
|
||||||
@@ -189,9 +200,9 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error {
|
|||||||
// re-dials clean. Caller holds c.mu.
|
// re-dials clean. Caller holds c.mu.
|
||||||
func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error {
|
func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error {
|
||||||
if c.conn == nil {
|
if c.conn == nil {
|
||||||
conn, err := net.Dial("unix", c.path)
|
conn, err := netaddr.Dial(c.addr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.path, err)
|
return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.addr, err)
|
||||||
}
|
}
|
||||||
c.conn = conn
|
c.conn = conn
|
||||||
}
|
}
|
||||||
|
|||||||
+20
-40
@@ -8,11 +8,11 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"net"
|
"net"
|
||||||
"os"
|
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/netaddr"
|
||||||
"github.com/kami/maven/internal/store"
|
"github.com/kami/maven/internal/store"
|
||||||
"golang.org/x/sys/unix"
|
"golang.org/x/sys/unix"
|
||||||
)
|
)
|
||||||
@@ -446,6 +446,7 @@ func mapErr(err error) error {
|
|||||||
type Server struct {
|
type Server struct {
|
||||||
api atomic.Value // stores CoreAPI
|
api atomic.Value // stores CoreAPI
|
||||||
path string
|
path string
|
||||||
|
addr netaddr.Addr
|
||||||
|
|
||||||
ln net.Listener
|
ln net.Listener
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
@@ -610,31 +611,29 @@ type CheckFunc func(ctx context.Context, m Method, params json.RawMessage) error
|
|||||||
// MethodAssertStepUp dispatch calls this instead of going through CoreAPI.
|
// MethodAssertStepUp dispatch calls this instead of going through CoreAPI.
|
||||||
type StepUpFunc func(ctx context.Context) error
|
type StepUpFunc func(ctx context.Context) error
|
||||||
|
|
||||||
// Listen creates a Server bound to path. path's parent dir must exist and be
|
// Listen creates a Server bound to path.
|
||||||
// 0700 (we chmod it if we own it); the socket file itself is created 0600 so
|
//
|
||||||
// only the same unix user can connect — the current "auth floor", same radius
|
// A bare path is a unix socket, unchanged: its parent dir is 0700 and the
|
||||||
// as wg at the network boundary. Removing a stale socket at path first lets
|
// socket file itself is 0600, so only the same unix user can connect — the
|
||||||
// the daemon restart cleanly.
|
// current "auth floor", same radius as wg at the network boundary. A stale
|
||||||
|
// socket is removed first so the daemon restarts cleanly.
|
||||||
|
//
|
||||||
|
// A "tcp://host:port?token=..." address binds a network listener instead, for
|
||||||
|
// a module that lives on another host. There is no filesystem there to be the
|
||||||
|
// auth floor, so netaddr checks the shared token before this package sees the
|
||||||
|
// connection and a token is mandatory. See internal/netaddr.
|
||||||
func Listen(path string, api CoreAPI) (*Server, error) {
|
func Listen(path string, api CoreAPI) (*Server, error) {
|
||||||
_ = os.Remove(path) // stale socket from a crashed daemon; ignore missing
|
addr, err := netaddr.Parse(path)
|
||||||
if err := os.MkdirAll(parentDir(path), 0o700); err != nil {
|
|
||||||
return nil, fmt.Errorf("ipc: 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 {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("ipc: listen %s: %w", path, err)
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := os.Chmod(path, 0o600); err != nil {
|
ln, err := netaddr.Listen(addr)
|
||||||
_ = ln.Close()
|
if err != nil {
|
||||||
_ = os.Remove(path)
|
return nil, err
|
||||||
return nil, fmt.Errorf("ipc: chmod socket: %w", err)
|
|
||||||
}
|
}
|
||||||
s := &Server{
|
s := &Server{
|
||||||
path: path,
|
path: path,
|
||||||
|
addr: addr,
|
||||||
ln: ln,
|
ln: ln,
|
||||||
done: make(chan struct{}),
|
done: make(chan struct{}),
|
||||||
}
|
}
|
||||||
@@ -1268,7 +1267,7 @@ func (s *Server) Close() error {
|
|||||||
// missing the seal costs every write since the last clean shutdown.
|
// missing the seal costs every write since the last clean shutdown.
|
||||||
log.Printf("ipc: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace)
|
log.Printf("ipc: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace)
|
||||||
}
|
}
|
||||||
_ = os.Remove(s.path)
|
netaddr.Cleanup(s.addr)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1338,25 +1337,6 @@ func (s *Server) Path() string { return s.path }
|
|||||||
// while the server is serving (dispatch loads api once per request via atomic).
|
// while the server is serving (dispatch loads api once per request via atomic).
|
||||||
func (s *Server) SetAPI(api CoreAPI) { s.api.Store(api) }
|
func (s *Server) SetAPI(api CoreAPI) { s.api.Store(api) }
|
||||||
|
|
||||||
func parentDir(p string) string {
|
|
||||||
if i := lastIndexByte(p, '/'); i >= 0 {
|
|
||||||
if i == 0 {
|
|
||||||
return "/"
|
|
||||||
}
|
|
||||||
return p[:i]
|
|
||||||
}
|
|
||||||
return "."
|
|
||||||
}
|
|
||||||
|
|
||||||
func lastIndexByte(s string, b byte) int {
|
|
||||||
for i := len(s) - 1; i >= 0; i-- {
|
|
||||||
if s[i] == b {
|
|
||||||
return i
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return -1
|
|
||||||
}
|
|
||||||
|
|
||||||
// peerCaller — read SO_PEERCRED off a unix conn to identify the connecting
|
// peerCaller — read SO_PEERCRED off a unix conn to identify the connecting
|
||||||
// process. Returns ok=false on a non-unix conn or a platform without
|
// process. Returns ok=false on a non-unix conn or a platform without
|
||||||
// SO_PEERCRED; the caller then proceeds without a Caller (the socket perms
|
// SO_PEERCRED; the caller then proceeds without a Caller (the socket perms
|
||||||
|
|||||||
@@ -24,6 +24,8 @@ import (
|
|||||||
"net"
|
"net"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/netaddr"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Client — one connection to one worker module. NOT goroutine-safe for
|
// Client — one connection to one worker module. NOT goroutine-safe for
|
||||||
@@ -38,15 +40,25 @@ type Client struct {
|
|||||||
dial func() (net.Conn, error)
|
dial func() (net.Conn, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Dial opens a Client to the worker socket at path. The first call lazily
|
// Dial opens a Client to the worker module at path. The first call lazily
|
||||||
// dials; subsequent calls reuse the conn (a fresh dial happens on next call
|
// dials; subsequent calls reuse the conn (a fresh dial happens on next call
|
||||||
// after a teardown). Lazy dial keeps a worker that's restarting from
|
// after a teardown). Lazy dial keeps a worker that's restarting from
|
||||||
// blocking core's startup; core attempts the dial on first use.
|
// blocking core's startup; core attempts the dial on first use.
|
||||||
|
//
|
||||||
|
// path is a netaddr seam address. A bare path is the unix socket it has
|
||||||
|
// always been; "tcp://workstation:9310?token=..." reaches a module on another
|
||||||
|
// host, which is how stt and tts move to the machine with the GPU and the
|
||||||
|
// microphone. A bad address surfaces on the first call, not here, because
|
||||||
|
// Dial does not fail — see internal/netaddr.
|
||||||
func Dial(path string) *Client {
|
func Dial(path string) *Client {
|
||||||
|
addr, err := netaddr.Parse(path)
|
||||||
return &Client{
|
return &Client{
|
||||||
path: path,
|
path: path,
|
||||||
dial: func() (net.Conn, error) {
|
dial: func() (net.Conn, error) {
|
||||||
return net.Dial("unix", path)
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return netaddr.Dial(addr)
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+22
-34
@@ -7,10 +7,11 @@
|
|||||||
// which module to dial; mixing the two is a config error caught cleanly by
|
// which module to dial; mixing the two is a config error caught cleanly by
|
||||||
// the wire, not a runtime goroutine panic). One Server per module process.
|
// the wire, not a runtime goroutine panic). One Server per module process.
|
||||||
//
|
//
|
||||||
// Socket perms mirror ipc.Server: dir 0700, socket 0600 ⇒ same unix user.
|
// The seam address decides the transport. On the default unix socket the
|
||||||
// The module has no key, so the floor is "same user"; the wg/mTLS layers
|
// perms mirror ipc.Server — dir 0700, socket 0600 ⇒ same unix user — and that
|
||||||
// are out of scope here (this socket never crosses the network radius —
|
// is the whole auth floor, because the seam never leaves the box. A tcp
|
||||||
// it's local-only, point-to-point between two processes on the box).
|
// address moves the module to another host and takes that floor away, so
|
||||||
|
// netaddr checks a shared token before the first frame. See internal/netaddr.
|
||||||
package worker
|
package worker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -18,11 +19,10 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
"os"
|
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
|
|
||||||
"golang.org/x/sys/unix"
|
"github.com/kami/maven/internal/netaddr"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Server — a worker module process's listener. Wires either a Transcriber,
|
// Server — a worker module process's listener. Wires either a Transcriber,
|
||||||
@@ -34,6 +34,7 @@ type Server struct {
|
|||||||
s Synthesizer
|
s Synthesizer
|
||||||
|
|
||||||
path string
|
path string
|
||||||
|
addr netaddr.Addr
|
||||||
ln net.Listener
|
ln net.Listener
|
||||||
|
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
@@ -61,25 +62,24 @@ func NewSynthesizerServer(path string, s Synthesizer) *Server {
|
|||||||
// two separate processes per the restart-free / fail-independent invariant).
|
// two separate processes per the restart-free / fail-independent invariant).
|
||||||
func (srv *Server) SetSynthesizer(s Synthesizer) { srv.s = s }
|
func (srv *Server) SetSynthesizer(s Synthesizer) { srv.s = s }
|
||||||
|
|
||||||
// Listen binds the unix socket with 0700 dir + 0600 socket perms (same floor
|
// Listen binds the seam address the Server was built with.
|
||||||
// as internal/ipc). A stale socket at path is removed first so the worker
|
//
|
||||||
// process restarts cleanly after a crash, no manual cleanup needed.
|
// A bare path is a unix socket with 0700 dir + 0600 socket perms, the same
|
||||||
|
// floor as internal/ipc, and a stale socket is removed first so the worker
|
||||||
|
// process restarts cleanly after a crash. A "tcp://host:port?token=..."
|
||||||
|
// address binds a network listener instead, so this module can run on the
|
||||||
|
// workstation while core stays on homesrv; the token is mandatory there,
|
||||||
|
// because there is no filesystem to be the auth floor. See internal/netaddr.
|
||||||
func (srv *Server) Listen() error {
|
func (srv *Server) Listen() error {
|
||||||
_ = os.Remove(srv.path)
|
addr, err := netaddr.Parse(srv.path)
|
||||||
if err := os.MkdirAll(parentDir(srv.path), 0o700); err != nil {
|
|
||||||
return fmt.Errorf("worker: mkdir socket dir: %w", err)
|
|
||||||
}
|
|
||||||
oldMask := unix.Umask(0o077)
|
|
||||||
ln, err := net.Listen("unix", srv.path)
|
|
||||||
unix.Umask(oldMask)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("worker: listen %s: %w", srv.path, err)
|
return err
|
||||||
}
|
}
|
||||||
if err := os.Chmod(srv.path, 0o600); err != nil {
|
ln, err := netaddr.Listen(addr)
|
||||||
_ = ln.Close()
|
if err != nil {
|
||||||
_ = os.Remove(srv.path)
|
return err
|
||||||
return fmt.Errorf("worker: chmod socket: %w", err)
|
|
||||||
}
|
}
|
||||||
|
srv.addr = addr
|
||||||
srv.ln = ln
|
srv.ln = ln
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -193,7 +193,7 @@ func (srv *Server) Close() error {
|
|||||||
}
|
}
|
||||||
err := srv.ln.Close()
|
err := srv.ln.Close()
|
||||||
srv.wg.Wait()
|
srv.wg.Wait()
|
||||||
_ = os.Remove(srv.path)
|
netaddr.Cleanup(srv.addr)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -214,15 +214,3 @@ func marshalResult(v any) json.RawMessage {
|
|||||||
b, _ := json.Marshal(v)
|
b, _ := json.Marshal(v)
|
||||||
return b
|
return b
|
||||||
}
|
}
|
||||||
|
|
||||||
func parentDir(p string) string {
|
|
||||||
for i := len(p) - 1; i >= 0; i-- {
|
|
||||||
if p[i] == '/' {
|
|
||||||
if i == 0 {
|
|
||||||
return "/"
|
|
||||||
}
|
|
||||||
return p[:i]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return "."
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user