diff --git a/internal/cli/broker_dial_darwin_test.go b/internal/cli/broker_dial_darwin_test.go new file mode 100644 index 0000000..bbd7597 --- /dev/null +++ b/internal/cli/broker_dial_darwin_test.go @@ -0,0 +1,12 @@ +//go:build darwin + +package cli + +import "syscall" + +// controlSockType is the control-channel socket type used by the broker dial +// test. darwin's AF_UNIX has no SOCK_SEQPACKET support, so the native test +// falls back to SOCK_DGRAM, which preserves datagram boundaries identically for +// the CLI dialer's Sendmsg/Recvmsg+SCM_RIGHTS path. Production runners are +// Linux-only (SOCK_SEQPACKET). +const controlSockType = syscall.SOCK_DGRAM diff --git a/internal/cli/broker_dial_linux_test.go b/internal/cli/broker_dial_linux_test.go new file mode 100644 index 0000000..779f094 --- /dev/null +++ b/internal/cli/broker_dial_linux_test.go @@ -0,0 +1,9 @@ +//go:build linux + +package cli + +import "syscall" + +// controlSockType is the control-channel socket type used by the broker dial +// test. On Linux (production) the runner uses SOCK_SEQPACKET. +const controlSockType = syscall.SOCK_SEQPACKET diff --git a/internal/cli/broker_dial_other.go b/internal/cli/broker_dial_other.go new file mode 100644 index 0000000..d516138 --- /dev/null +++ b/internal/cli/broker_dial_other.go @@ -0,0 +1,12 @@ +//go:build !unix + +package cli + +import ( + "errors" + "net/http" +) + +func newBrokerHTTPClient(int) *http.Client { return nil } + +var errBrokerUnsupported = errors.New("flashduty: broker mode is not supported on this platform") diff --git a/internal/cli/broker_dial_unix.go b/internal/cli/broker_dial_unix.go new file mode 100644 index 0000000..7070ed7 --- /dev/null +++ b/internal/cli/broker_dial_unix.go @@ -0,0 +1,86 @@ +//go:build unix + +package cli + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "sync" + "syscall" + "time" +) + +// errBrokerUnsupported is returned when broker mode is requested on a build that +// cannot provide it. On unix this is effectively unreachable (newBrokerHTTPClient +// never returns nil), but defaultNewClient references it on every platform. +var errBrokerUnsupported = errors.New("flashduty: broker mode is not supported on this platform") + +// brokerDialer owns the inherited control fd and serializes per-dial handshakes. +// Each Dial sends a 1-byte request datagram on the control channel and receives +// one dedicated SOCK_STREAM fd back via SCM_RIGHTS. +type brokerDialer struct { + mu sync.Mutex // serialize send+recv so concurrent dials don't cross fds + credFD int +} + +func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { + d.mu.Lock() + defer d.mu.Unlock() + + if err := syscall.Sendmsg(d.credFD, []byte{0x01}, nil, nil, 0); err != nil { + return nil, fmt.Errorf("broker handshake send: %w", err) + } + body := make([]byte, 1) + oob := make([]byte, syscall.CmsgSpace(4)) // room for exactly one fd + n, oobn, _, _, err := syscall.Recvmsg(d.credFD, body, oob, 0) + if err != nil { + return nil, fmt.Errorf("broker handshake recv: %w", err) + } + if n < 1 || body[0] != 0x01 { + return nil, fmt.Errorf("broker refused connection (code %v)", body[:n]) + } + scms, err := syscall.ParseSocketControlMessage(oob[:oobn]) + if err != nil { + return nil, fmt.Errorf("broker parse scm: %w", err) + } + if len(scms) == 0 { + return nil, fmt.Errorf("broker sent no fd") + } + fds, err := syscall.ParseUnixRights(&scms[0]) + if err != nil || len(fds) == 0 { + return nil, fmt.Errorf("broker parse rights: %w", err) + } + f := os.NewFile(uintptr(fds[0]), "broker-conn") + conn, err := net.FileConn(f) // dups + registers with the netpoller + _ = f.Close() + if err != nil { + return nil, fmt.Errorf("broker fileconn: %w", err) + } + return conn, nil +} + +// newBrokerHTTPClient builds an *http.Client whose Transport.DialContext routes +// every connection over the inherited control fd. Timeout matches the SDK's +// historical default (30s) so behavior is unchanged for non-streaming calls; +// streaming export relies on request context like before. +func newBrokerHTTPClient(credFD int) *http.Client { + d := &brokerDialer{credFD: credFD} + return &http.Client{ + Timeout: 30 * time.Second, + Transport: &http.Transport{ + DialContext: d.dial, + DisableCompression: false, + MaxIdleConns: 0, + // All dials target the same logical host (the broker sentinel base + // URL) over the one control fd, so a single idle keep-alive conn is + // enough for pagination loops; cap it so dispatched conns don't linger. + MaxIdleConnsPerHost: 1, + IdleConnTimeout: 90 * time.Second, + ResponseHeaderTimeout: 0, + }, + } +} diff --git a/internal/cli/broker_dial_unix_test.go b/internal/cli/broker_dial_unix_test.go new file mode 100644 index 0000000..6f5d13a --- /dev/null +++ b/internal/cli/broker_dial_unix_test.go @@ -0,0 +1,293 @@ +//go:build unix + +package cli + +import ( + "bufio" + "context" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "sync" + "syscall" + "testing" +) + +// fakeBroker mimics the runner: it owns the parent end of a SEQPACKET control +// socket, and for each 1-byte handshake it dispatches a dedicated STREAM conn +// (via SCM_RIGHTS) that proxies to the given upstream URL, overwriting app_key. +func fakeBroker(t *testing.T, parentFD int, upstream string, realKey string) (stop func()) { + t.Helper() + var ( + mu sync.Mutex + conns []net.Conn + ctlGone = make(chan struct{}) + ) + go func() { + defer close(ctlGone) + buf := make([]byte, 8) + for { + // Blocks until a handshake datagram arrives or stop() shuts down + // parentFD (recvmsg then returns EOF → clean exit, no leak). NOTE: a + // bare Close(parentFD) does NOT wake a blocked recvmsg on Linux (only on + // darwin/BSD), so stop() uses Shutdown — see the stop func below. + n, _, _, _, err := syscall.Recvmsg(parentFD, buf, nil, 0) + if err != nil || n == 0 { + return + } + // Create the dedicated STREAM pair; keep one end, send the other. + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + _ = syscall.Sendmsg(parentFD, []byte{0xFF}, nil, nil, 0) + continue + } + rights := syscall.UnixRights(pair[1]) + if serr := syscall.Sendmsg(parentFD, []byte{0x01}, rights, nil, 0); serr != nil { + _ = syscall.Close(pair[0]) + _ = syscall.Close(pair[1]) + return + } + _ = syscall.Close(pair[1]) + // Serve HTTP on pair[0], proxying to upstream with the real key. + myEnd := os.NewFile(uintptr(pair[0]), "broker-end") + conn, _ := net.FileConn(myEnd) + _ = myEnd.Close() + if conn == nil { + continue + } + mu.Lock() + conns = append(conns, conn) + mu.Unlock() + go serveProxyConn(conn, upstream, realKey) + } + }() + return func() { + // Shutdown (NOT a bare Close) wakes the control goroutine's blocked + // Recvmsg portably — on Linux, closing an fd does not interrupt a recvmsg + // blocked on it in another goroutine; shutdown returns EOF on both Linux + // and darwin. Join, then close. + _ = syscall.Shutdown(parentFD, syscall.SHUT_RDWR) + <-ctlGone + _ = syscall.Close(parentFD) + mu.Lock() + for _, c := range conns { + _ = c.Close() + } + mu.Unlock() + } +} + +func TestBrokerHTTPClient_DialAndRewrite(t *testing.T) { + // Upstream asserts the real key arrived (sentinel was overwritten). + gotKey := make(chan string, 16) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotKey <- r.URL.Query().Get("app_key") + _, _ = io.WriteString(w, `{"ok":true}`) + })) + defer upstream.Close() + + // Control channel: child end → CLI; parent end → fake broker. Production + // (Linux runner) uses SOCK_SEQPACKET, but darwin's AF_UNIX has no SEQPACKET + // support, so the native test uses SOCK_DGRAM — both preserve the datagram + // boundaries the 1-byte handshake relies on, and the CLI dialer's + // Sendmsg/Recvmsg+SCM_RIGHTS path is identical for either socket type. + pair, err := syscall.Socketpair(syscall.AF_UNIX, controlSockType, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + childFD, parentFD := pair[0], pair[1] + stop := fakeBroker(t, parentFD, upstream.URL, "REAL-KEY") // owns + closes parentFD + defer func() { _ = syscall.Close(childFD) }() + defer stop() + + client := newBrokerHTTPClient(childFD) + if client == nil { + t.Fatal("newBrokerHTTPClient returned nil") + } + defer client.CloseIdleConnections() // release dispatched keep-alive conns + // The CLI's base URL is an http placeholder; broker rewrites host. + req, _ := http.NewRequestWithContext(context.Background(), "GET", + "http://flashduty.broker.local/incident/channels?app_key=SENTINEL", nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("client.Do: %v", err) + } + _, _ = io.Copy(io.Discard, resp.Body) // drain so the conn is reusable + _ = resp.Body.Close() + if got := <-gotKey; got != "REAL-KEY" { + t.Fatalf("upstream saw app_key=%q, want REAL-KEY (sentinel not overwritten)", got) + } + + // Concurrency: 8 parallel requests each get their own dispatched conn. + var wg sync.WaitGroup + errs := make(chan error, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + r, _ := http.NewRequest("GET", "http://flashduty.broker.local/x?app_key=SENTINEL", nil) + rsp, e := client.Do(r) + if e != nil { + errs <- e + return + } + _, _ = io.Copy(io.Discard, rsp.Body) + _ = rsp.Body.Close() + }() + } + wg.Wait() + close(errs) + for e := range errs { + t.Fatalf("concurrent client.Do: %v", e) + } + for i := 0; i < 8; i++ { + if got := <-gotKey; got != "REAL-KEY" { + t.Fatalf("concurrent req saw app_key=%q", got) + } + } +} + +// TestDefaultNewClient_BrokerMode covers the defaultNewClient wiring: with +// FLASHDUTY_CRED_FD set it builds a client with no configured app_key (the +// broker supplies the real key), and a malformed fd value is a clean error. +func TestDefaultNewClient_BrokerMode(t *testing.T) { + // Hermetic config: empty HOME (no config file) + no env app key. + t.Setenv("HOME", t.TempDir()) + t.Setenv("FLASHDUTY_APP_KEY", "") + + // A real, open fd so newBrokerHTTPClient gets a usable control channel. + pair, err := syscall.Socketpair(syscall.AF_UNIX, controlSockType, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + defer func() { _ = syscall.Close(pair[0]) }() + defer func() { _ = syscall.Close(pair[1]) }() + + t.Setenv("FLASHDUTY_CRED_FD", strconv.Itoa(pair[0])) + client, err := defaultNewClient() + if err != nil { + t.Fatalf("broker mode with no app key must succeed, got: %v", err) + } + if client == nil { + t.Fatal("broker mode returned nil client") + } + + // A malformed fd value is rejected up front (not silently ignored). + t.Setenv("FLASHDUTY_CRED_FD", "not-a-number") + if _, err := defaultNewClient(); err == nil { + t.Fatal("invalid FLASHDUTY_CRED_FD must error") + } + + // Both a configured app key AND a control fd: broker mode wins. The client + // builds with the sentinel key (the broker overwrites it with the real + // per-person key), so the configured app key never reaches the wire. + t.Setenv("FLASHDUTY_APP_KEY", "ENV-KEY-SHOULD-NOT-BE-USED") + t.Setenv("FLASHDUTY_CRED_FD", strconv.Itoa(pair[0])) + if c, err := defaultNewClient(); err != nil || c == nil { + t.Fatalf("both app key + cred fd set: want broker client, got client=%v err=%v", c, err) + } +} + +// TestDefaultNewClient_RejectsStdioFD verifies the fd>=3 guard: fds 0/1/2 are +// stdio and can never be the runner-injected control end, so they are rejected +// rather than handshaking on stdin/stdout. +func TestDefaultNewClient_RejectsStdioFD(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("FLASHDUTY_APP_KEY", "") + for _, fd := range []string{"-1", "0", "1", "2"} { + t.Setenv("FLASHDUTY_CRED_FD", fd) + if _, err := defaultNewClient(); err == nil { + t.Fatalf("FLASHDUTY_CRED_FD=%q must be rejected (stdio/invalid)", fd) + } + } +} + +// TestBrokerHTTPClient_RefusedReturnsError verifies the dialer surfaces the +// broker's 0xFF refusal (e.g. the runner failed to mint a connection) as a real +// error instead of hanging or wrapping a nil conn. +func TestBrokerHTTPClient_RefusedReturnsError(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, controlSockType, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + childFD, parentFD := pair[0], pair[1] + defer func() { _ = syscall.Close(childFD) }() + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 8) + for { + n, _, _, _, rerr := syscall.Recvmsg(parentFD, buf, nil, 0) + if rerr != nil || n == 0 { + return + } + _ = syscall.Sendmsg(parentFD, []byte{0xFF}, nil, nil, 0) // always refuse + } + }() + + client := newBrokerHTTPClient(childFD) + req, _ := http.NewRequestWithContext(context.Background(), "GET", + "http://flashduty.broker.local/x?app_key=SENTINEL", nil) + if _, err := client.Do(req); err == nil { + t.Fatal("client.Do must fail when broker refuses with 0xFF") + } + // Shutdown (not bare Close) to wake the goroutine's blocked Recvmsg on Linux. + _ = syscall.Shutdown(parentFD, syscall.SHUT_RDWR) + <-done + _ = syscall.Close(parentFD) +} + +// serveProxyConn is a tiny test upstream-proxy used by fakeBroker; the real +// implementation lives in the runner, this mirrors it for the CLI test. +func serveProxyConn(conn net.Conn, upstream, realKey string) { + defer func() { _ = conn.Close() }() + br := newReadProxy(conn, upstream, realKey) + br.run() +} + +// readProxy is a minimal test-only HTTP proxy: it reads requests off conn, +// overwrites the app_key query param with realKey, forwards to upstream, and +// copies each response back. Serves sequential keep-alive requests until the +// connection closes. +type readProxy struct { + conn net.Conn + upstream *url.URL + realKey string +} + +func newReadProxy(conn net.Conn, upstream, realKey string) *readProxy { + u, _ := url.Parse(upstream) + return &readProxy{conn: conn, upstream: u, realKey: realKey} +} + +func (p *readProxy) run() { + br := bufio.NewReader(p.conn) + for { + req, err := http.ReadRequest(br) + if err != nil { + return + } + // Point the request at the real upstream and overwrite the sentinel. + req.URL.Scheme = p.upstream.Scheme + req.URL.Host = p.upstream.Host + req.Host = p.upstream.Host + q := req.URL.Query() + q.Set("app_key", p.realKey) + req.URL.RawQuery = q.Encode() + req.RequestURI = "" + resp, err := http.DefaultTransport.RoundTrip(req) + if err != nil { + return + } + if err := resp.Write(p.conn); err != nil { + _ = resp.Body.Close() + return + } + _ = resp.Body.Close() + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index eb97646..d3012fb 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "os" + "strconv" "strings" "github.com/flashcatcloud/go-flashduty" @@ -128,26 +129,44 @@ func newClient() (*flashduty.Client, error) { } // defaultNewClient creates a real go-flashduty client from resolved config + -// flag overrides. This is the typed SDK every command uses. +// flag overrides. In broker mode (FLASHDUTY_CRED_FD set — runner-injected), it +// routes all egress over the inherited control fd and sends a sentinel app_key +// the broker overwrites with the real per-person key. func defaultNewClient() (*flashduty.Client, error) { cfg, err := loadResolvedConfig() if err != nil { return nil, err } - if cfg.AppKey == "" { - return nil, fmt.Errorf("no app key configured. Run 'flashduty login' or set FLASHDUTY_APP_KEY") - } - opts := []flashduty.Option{ flashduty.WithUserAgent("flashduty-cli/" + versionStr), flashduty.WithLogger(&silentLogger{}), } + + appKey := cfg.AppKey + if fdStr := os.Getenv("FLASHDUTY_CRED_FD"); fdStr != "" { + fd, perr := strconv.Atoi(fdStr) + // fds 0/1/2 are stdio; the runner inherits the control end at fd 3+. A + // value below 3 means a misconfigured runner, not a real control fd — + // reject it rather than handshaking on stdin/stdout. + if perr != nil || fd < 3 { + return nil, fmt.Errorf("invalid FLASHDUTY_CRED_FD=%q", fdStr) + } + hc := newBrokerHTTPClient(fd) + if hc == nil { + return nil, errBrokerUnsupported + } + opts = append(opts, flashduty.WithHTTPClient(hc)) + appKey = "broker-sentinel" // non-empty: go-flashduty rejects ""; broker overwrites it + } else if appKey == "" { + return nil, fmt.Errorf("no app key configured. Run 'flashduty login' or set FLASHDUTY_APP_KEY") + } + if cfg.BaseURL != "" && cfg.BaseURL != config.DefaultBaseURL { opts = append(opts, flashduty.WithBaseURL(cfg.BaseURL)) } - return flashduty.NewClient(cfg.AppKey, opts...) + return flashduty.NewClient(appKey, opts...) } func loadResolvedConfig() (*config.Config, error) {