Accept named-pipe agent connections, carry the remote peer through the agent handshake, and treat pipe disconnects as probes

This commit is contained in:
Viktor Liu
2026-09-23 14:03:59 +02:00
parent b9264d2824
commit c26b17653a
10 changed files with 245 additions and 57 deletions
+83 -23
View File
@@ -11,6 +11,8 @@ import (
"fmt"
"io"
"net"
"net/netip"
"slices"
"syscall"
"time"
@@ -43,6 +45,10 @@ const (
// agentHandshakeTimeout bounds the whole exchange. Both ends are local
// processes, so this only has to cover scheduling, never a network.
agentHandshakeTimeout = 5 * time.Second
// maxAgentPeerAddrLen is the longest peer address the grant can carry; its
// length travels as one byte.
maxAgentPeerAddrLen = 255
)
// Domain separation, so a tag one side produces can never be replayed as the
@@ -62,13 +68,38 @@ func agentMAC(token, label []byte, parts ...[]byte) []byte {
return mac.Sum(nil)
}
// agentGrant is what the daemon authenticates to the agent for one proxied
// connection: whether it is view-only, and the address of the remote peer the
// daemon accepted, which the agent otherwise only sees as the local socket.
type agentGrant struct {
viewOnly bool
peerAddr string
}
// encode renders the grant as the bytes the daemon's tag covers: the view-only
// byte, a one-byte address length, then the address.
func (g agentGrant) encode() ([]byte, error) {
if len(g.peerAddr) > maxAgentPeerAddrLen {
return nil, fmt.Errorf("peer address of %d bytes exceeds %d", len(g.peerAddr), maxAgentPeerAddrLen)
}
b := make([]byte, 0, 2+len(g.peerAddr))
b = append(b, viewOnlyByte(g.viewOnly)...)
b = append(b, byte(len(g.peerAddr)))
return append(b, g.peerAddr...), nil
}
// agentClientHandshake runs the daemon's half against a freshly dialled agent
// connection: read the agent's challenge, answer it, then challenge the agent
// back and check its answer before any session bytes are proxied.
//
// viewOnly travels inside the daemon's tag, so an impostor cannot flip a
// read-only session into a controlling one by rewriting the byte in flight.
func agentClientHandshake(conn net.Conn, token []byte, viewOnly bool) error {
// The grant travels inside the daemon's tag, so an impostor cannot flip a
// read-only session into a controlling one, or change the reported peer, by
// rewriting bytes in flight.
func agentClientHandshake(conn net.Conn, token []byte, grant agentGrant) error {
payload, err := grant.encode()
if err != nil {
return err
}
if err := conn.SetDeadline(time.Now().Add(agentHandshakeTimeout)); err != nil {
return fmt.Errorf("set handshake deadline: %w", err)
}
@@ -88,11 +119,10 @@ func agentClientHandshake(conn net.Conn, token []byte, viewOnly bool) error {
return fmt.Errorf("read random: %w", err)
}
flag := viewOnlyByte(viewOnly)
reply := make([]byte, 0, agentMACLen+agentNonceLen+1)
reply = append(reply, agentMAC(token, agentDaemonLabel, agentNonce, flag)...)
reply := make([]byte, 0, agentMACLen+agentNonceLen+len(payload))
reply = append(reply, agentMAC(token, agentDaemonLabel, agentNonce, payload)...)
reply = append(reply, daemonNonce...)
reply = append(reply, flag...)
reply = append(reply, payload...)
if _, err := conn.Write(reply); err != nil {
return fmt.Errorf("send handshake response: %w", err)
}
@@ -109,10 +139,11 @@ func agentClientHandshake(conn net.Conn, token []byte, viewOnly bool) error {
}
// agentServerHandshake runs the agent's half against an accepted connection,
// returning the view-only flag the daemon authenticated.
func agentServerHandshake(conn net.Conn, token []byte) (bool, error) {
// returning the grant the daemon authenticated.
func agentServerHandshake(conn net.Conn, token []byte) (agentGrant, error) {
var none agentGrant
if err := conn.SetDeadline(time.Now().Add(agentHandshakeTimeout)); err != nil {
return false, fmt.Errorf("set handshake deadline: %w", err)
return none, fmt.Errorf("set handshake deadline: %w", err)
}
defer func() {
if err := conn.SetDeadline(time.Time{}); err != nil {
@@ -122,29 +153,33 @@ func agentServerHandshake(conn net.Conn, token []byte) (bool, error) {
agentNonce := make([]byte, agentNonceLen)
if _, err := rand.Read(agentNonce); err != nil {
return false, fmt.Errorf("read random: %w", err)
return none, fmt.Errorf("read random: %w", err)
}
if _, err := conn.Write(agentNonce); err != nil {
return false, fmt.Errorf("send challenge: %w", err)
return none, fmt.Errorf("send challenge: %w", err)
}
buf := make([]byte, agentMACLen+agentNonceLen+1)
if _, err := io.ReadFull(conn, buf); err != nil {
return false, fmt.Errorf("read daemon response: %w", err)
head := make([]byte, agentMACLen+agentNonceLen+2)
if _, err := io.ReadFull(conn, head); err != nil {
return none, fmt.Errorf("read daemon response: %w", err)
}
daemonTag := buf[:agentMACLen]
daemonNonce := buf[agentMACLen : agentMACLen+agentNonceLen]
flag := buf[agentMACLen+agentNonceLen:]
daemonTag := head[:agentMACLen]
daemonNonce := head[agentMACLen : agentMACLen+agentNonceLen]
addr := make([]byte, head[len(head)-1])
if _, err := io.ReadFull(conn, addr); err != nil {
return none, fmt.Errorf("read daemon response: %w", err)
}
payload := slices.Concat(head[agentMACLen+agentNonceLen:], addr)
want := agentMAC(token, agentDaemonLabel, agentNonce, flag)
want := agentMAC(token, agentDaemonLabel, agentNonce, payload)
if subtle.ConstantTimeCompare(daemonTag, want) != 1 {
return false, fmt.Errorf("caller did not prove it holds the session token")
return none, fmt.Errorf("caller did not prove it holds the session token")
}
if _, err := conn.Write(agentMAC(token, agentAgentLabel, daemonNonce)); err != nil {
return false, fmt.Errorf("send response: %w", err)
return none, fmt.Errorf("send response: %w", err)
}
return flag[0] != 0, nil
return agentGrant{viewOnly: payload[0] != 0, peerAddr: string(addr)}, nil
}
// isProbeDisconnect reports whether err is a peer that connected and left
@@ -169,7 +204,7 @@ func isProbeDisconnect(err error) bool {
case errors.Is(err, syscall.EPIPE), errors.Is(err, syscall.ECONNRESET):
return true
default:
return false
return isPipeDisconnect(err)
}
}
@@ -180,3 +215,28 @@ func viewOnlyByte(viewOnly bool) []byte {
}
return []byte{0}
}
// peerAddrConn reports the remote peer the daemon authenticated in the grant
// as the connection's remote address, in place of the local socket the agent
// actually accepted on.
type peerAddrConn struct {
net.Conn
remote net.Addr
}
func (c *peerAddrConn) RemoteAddr() net.Addr { return c.remote }
// withGrantPeer wraps conn so RemoteAddr returns the grant's peer address. A
// grant without a parseable address leaves conn as it is.
func withGrantPeer(conn net.Conn, grant agentGrant) net.Conn {
if grant.peerAddr == "" {
return conn
}
ap, err := netip.ParseAddrPort(grant.peerAddr)
if err != nil {
log.Debugf("agent grant peer address %q: %v", grant.peerAddr, err)
return conn
}
ap = netip.AddrPortFrom(ap.Addr().Unmap(), ap.Port())
return &peerAddrConn{Conn: conn, remote: net.TCPAddrFromAddrPort(ap)}
}
@@ -0,0 +1,9 @@
//go:build !windows && !js && !ios && !android
package server
// isPipeDisconnect is Windows-only: other platforms have no named pipes, and
// their socket disconnects are covered by isProbeDisconnect.
func isPipeDisconnect(error) bool {
return false
}
+58 -13
View File
@@ -14,7 +14,14 @@ import (
// runHandshake drives both halves over an in-memory pipe and returns what each
// side concluded.
func runHandshake(t *testing.T, daemonToken, agentToken []byte, viewOnly bool) (daemonErr error, gotViewOnly bool, agentErr error) {
func runHandshake(t *testing.T, daemonToken, agentToken []byte, grant agentGrant) (daemonErr error, got agentGrant, agentErr error) {
t.Helper()
return runHandshakeOver(t, daemonToken, agentToken, grant, func(c net.Conn) net.Conn { return c })
}
// runHandshakeOver is runHandshake with the daemon's end of the connection
// wrapped, so a test can interfere with what the daemon sends.
func runHandshakeOver(t *testing.T, daemonToken, agentToken []byte, grant agentGrant, wrap func(net.Conn) net.Conn) (daemonErr error, got agentGrant, agentErr error) {
t.Helper()
daemonSide, agentSide := net.Pipe()
@@ -24,43 +31,70 @@ func runHandshake(t *testing.T, daemonToken, agentToken []byte, viewOnly bool) (
})
type agentResult struct {
viewOnly bool
err error
grant agentGrant
err error
}
agentDone := make(chan agentResult, 1)
go func() {
v, err := agentServerHandshake(agentSide, agentToken)
g, err := agentServerHandshake(agentSide, agentToken)
if err != nil {
// What the agent's caller does on rejection, so the daemon sees the
// close rather than waiting out its own deadline.
_ = agentSide.Close()
}
agentDone <- agentResult{v, err}
agentDone <- agentResult{g, err}
}()
daemonErr = agentClientHandshake(daemonSide, daemonToken, viewOnly)
daemonErr = agentClientHandshake(wrap(daemonSide), daemonToken, grant)
res := <-agentDone
return daemonErr, res.viewOnly, res.err
return daemonErr, res.grant, res.err
}
func TestAgentHandshake_MatchingTokens(t *testing.T) {
token := bytes.Repeat([]byte{0xA5}, agentTokenLen)
for _, viewOnly := range []bool{false, true} {
dErr, gotViewOnly, aErr := runHandshake(t, token, token, viewOnly)
for _, grant := range []agentGrant{
{viewOnly: false, peerAddr: "100.64.0.7:51234"},
{viewOnly: true, peerAddr: "[fd00:1234::2]:5900"},
{viewOnly: false, peerAddr: ""},
} {
dErr, got, aErr := runHandshake(t, token, token, grant)
require.NoError(t, dErr)
require.NoError(t, aErr)
assert.Equal(t, viewOnly, gotViewOnly, "the agent must see the flag the daemon authenticated")
assert.Equal(t, grant, got, "the agent must see the grant the daemon authenticated")
}
}
// The peer address is covered by the daemon's tag, so rewriting it in flight
// fails the handshake instead of misattributing the session.
func TestAgentHandshake_TamperedPeerAddrIsRefused(t *testing.T) {
token := bytes.Repeat([]byte{0x5A}, agentTokenLen)
grant := agentGrant{peerAddr: "100.64.0.7:51234"}
_, _, aErr := runHandshakeOver(t, token, token, grant, func(c net.Conn) net.Conn {
return &rewriteConn{Conn: c, from: []byte("100.64.0.7"), to: []byte("100.64.0.9")}
})
require.Error(t, aErr)
assert.Contains(t, aErr.Error(), "did not prove it holds the session token")
}
func TestAgentHandshake_OversizedPeerAddrIsRefused(t *testing.T) {
token := bytes.Repeat([]byte{0x6B}, agentTokenLen)
daemonSide, agentSide := net.Pipe()
defer daemonSide.Close()
defer agentSide.Close()
err := agentClientHandshake(daemonSide, token, agentGrant{peerAddr: string(bytes.Repeat([]byte{'a'}, maxAgentPeerAddrLen+1))})
require.Error(t, err, "an address longer than the length byte can carry must not be truncated silently")
}
// The point of the exchange: an impostor listening on the socket without the
// token cannot complete it, and the daemon refuses before proxying anything.
func TestAgentHandshake_ImpostorAgentIsRefused(t *testing.T) {
daemonToken := bytes.Repeat([]byte{0x01}, agentTokenLen)
impostorToken := bytes.Repeat([]byte{0x02}, agentTokenLen)
dErr, _, aErr := runHandshake(t, daemonToken, impostorToken, false)
dErr, _, aErr := runHandshake(t, daemonToken, impostorToken, agentGrant{})
require.Error(t, aErr, "the impostor cannot verify the daemon's tag")
require.Error(t, dErr, "the daemon must not proceed against an unproven peer")
}
@@ -71,7 +105,7 @@ func TestAgentHandshake_ImpostorDaemonIsRefused(t *testing.T) {
agentToken := bytes.Repeat([]byte{0x03}, agentTokenLen)
impostorToken := bytes.Repeat([]byte{0x04}, agentTokenLen)
_, _, aErr := runHandshake(t, impostorToken, agentToken, false)
_, _, aErr := runHandshake(t, impostorToken, agentToken, agentGrant{})
require.Error(t, aErr)
assert.Contains(t, aErr.Error(), "did not prove it holds the session token")
}
@@ -96,7 +130,7 @@ func TestAgentHandshake_TokenNeverSent(t *testing.T) {
_, _ = agentServerHandshake(&teeConn{Conn: agentSide, mu: &mu, read: &wire, written: &wire}, token)
}()
require.NoError(t, agentClientHandshake(daemonSide, token, false))
require.NoError(t, agentClientHandshake(daemonSide, token, agentGrant{}))
<-done
mu.Lock()
defer mu.Unlock()
@@ -151,3 +185,14 @@ func (c *teeConn) Write(b []byte) (int, error) {
}
return n, err
}
// rewriteConn replaces the first occurrence of from with to in what is written,
// standing in for something on the socket altering the daemon's bytes.
type rewriteConn struct {
net.Conn
from, to []byte
}
func (c *rewriteConn) Write(b []byte) (int, error) {
return c.Conn.Write(bytes.Replace(b, c.from, c.to, 1))
}
@@ -0,0 +1,17 @@
//go:build windows
package server
import (
"errors"
"golang.org/x/sys/windows"
)
// isPipeDisconnect reports the named-pipe errors a client that closed its end
// produces: the pipe is being closed, has ended, or was never connected.
func isPipeDisconnect(err error) bool {
return errors.Is(err, windows.ERROR_NO_DATA) ||
errors.Is(err, windows.ERROR_BROKEN_PIPE) ||
errors.Is(err, windows.ERROR_PIPE_NOT_CONNECTED)
}
+2 -1
View File
@@ -176,7 +176,8 @@ func proxyToAgent(ctx context.Context, client net.Conn, socketPath, authToken st
return fmt.Errorf("agent peer validation failed: %w", err)
}
if err := agentClientHandshake(agentConn, tokenBytes, viewOnly); err != nil {
grant := agentGrant{viewOnly: viewOnly, peerAddr: client.RemoteAddr().String()}
if err := agentClientHandshake(agentConn, tokenBytes, grant); err != nil {
_ = agentConn.Close()
return fmt.Errorf("agent handshake: %w", err)
}
@@ -64,6 +64,43 @@ func TestAgentPipePeerPID(t *testing.T) {
assert.Error(t, validateAgentPeer(conn, 0), "unknown PID must fail closed")
}
// The agent's server runs its source check on the accepted pipe connection's
// remote address, so that address must count as local IPC.
func TestAgentPipeRemoteAddrIsAllowedSource(t *testing.T) {
path, err := newAgentPipePath(0)
require.NoError(t, err)
ln, err := listenAgentPipe(path, testPipeSDDL)
if errors.Is(err, windows.ERROR_ACCESS_DENIED) {
t.Skip("creating a pipe under ProtectedPrefix needs administrator rights")
}
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
accepted := make(chan net.Conn, 1)
go func() {
c, err := ln.Accept()
if err != nil {
close(accepted)
return
}
accepted <- c
}()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
client, err := dialAgent(ctx, path)
require.NoError(t, err)
defer client.Close()
server, ok := <-accepted
require.True(t, ok, "accept must succeed")
defer server.Close()
srv := New(Config{Capturer: &testCapturer{}, Injector: &StubInputInjector{}})
assert.True(t, srv.isAllowedSource(server.RemoteAddr()),
"pipe remote %T must pass the source check", server.RemoteAddr())
}
// A second listener on the same name must fail, so an agent never serves a
// pipe name some other process created first.
func TestAgentPipeRefusesExistingName(t *testing.T) {
+7 -7
View File
@@ -224,13 +224,13 @@ func (s *Server) maybeRunNoiseHandshake(conn net.Conn, magic [4]byte, headerMode
}
// verifyAgentToken runs the agent's half of the mutual challenge-response with
// the daemon when a token is configured, and reports the view-only flag the
// daemon authenticated. Returns (ok, viewOnly). ok=false closes the connection.
func (s *Server) verifyAgentToken(conn net.Conn, connLog *log.Entry) (bool, bool) {
// the daemon when a token is configured, and reports the grant the daemon
// authenticated. Returns (ok, grant). ok=false closes the connection.
func (s *Server) verifyAgentToken(conn net.Conn, connLog *log.Entry) (bool, agentGrant) {
if len(s.agentToken) == 0 {
return true, false
return true, agentGrant{}
}
viewOnly, err := agentServerHandshake(conn, s.agentToken)
grant, err := agentServerHandshake(conn, s.agentToken)
if err != nil {
if isProbeDisconnect(err) {
connLog.Tracef("agent auth: %v", err)
@@ -238,9 +238,9 @@ func (s *Server) verifyAgentToken(conn net.Conn, connLog *log.Entry) (bool, bool
connLog.Warnf("agent auth: %v", err)
}
conn.Close()
return false, false
return false, agentGrant{}
}
return true, viewOnly
return true, grant
}
// authorizeSession runs the Noise_IK handshake when auth is enabled.
+19 -5
View File
@@ -1065,9 +1065,9 @@ func (s *Server) validateCapturer(capturer ScreenCapturer) error {
// and from the local WireGuard IP (prevents local privilege escalation).
// Matches the SSH server's connectionValidator logic.
func (s *Server) isAllowedSource(addr net.Addr) bool {
// Unix-socket remotes (the agent path) are local IPC, gated by the
// token, not by overlay membership.
if _, ok := addr.(*net.UnixAddr); ok {
// Unix-socket and named-pipe remotes (the agent path) are local IPC,
// gated by the token, not by overlay membership.
if isLocalIPCAddr(addr) {
return true
}
tcpAddr, ok := addr.(*net.TCPAddr)
@@ -1115,7 +1115,7 @@ func (s *Server) handleConnection(conn net.Conn) {
_ = conn.Close()
return
}
ok, agentViewOnly := s.verifyAgentToken(conn, connLog)
ok, grant := s.verifyAgentToken(conn, connLog)
if !ok {
// Reported there already, at a level that tells a liveness probe apart
// from a bad token. The daemon dials the agent socket to wait for it to
@@ -1124,6 +1124,10 @@ func (s *Server) handleConnection(conn net.Conn) {
connLog.Debug("VNC connection rejected: agent token check failed")
return
}
// Behind the daemon the accepted address is only the local socket; the
// remote peer is the one the daemon vouched for in the grant.
conn = withGrantPeer(conn, grant)
connLog = s.log.WithField("remote", conn.RemoteAddr().String())
header, err := s.readConnectionHeader(conn)
if err != nil {
connLog.Infof("VNC connection rejected: header read failed: %v", err)
@@ -1205,7 +1209,7 @@ func (s *Server) handleConnection(conn net.Conn) {
serverW: w,
serverH: h,
log: connLog,
viewOnly: decision.ViewOnly || agentViewOnly,
viewOnly: decision.ViewOnly || grant.viewOnly,
}
sess.serve()
connLog.Infof("VNC connection closed (%dms)", time.Since(start).Milliseconds())
@@ -1366,3 +1370,13 @@ func acceptRetryable(err error) bool {
return errors.Is(err, errno)
})
}
// isLocalIPCAddr reports whether addr belongs to a local IPC transport: a
// Unix-domain socket, or a Windows named pipe (go-winio names its network
// "pipe").
func isLocalIPCAddr(addr net.Addr) bool {
if _, ok := addr.(*net.UnixAddr); ok {
return true
}
return addr != nil && addr.Network() == "pipe"
}
+9 -2
View File
@@ -295,7 +295,7 @@ func TestAgentToken_MismatchClosesConnection(t *testing.T) {
require.NoError(t, conn.SetDeadline(time.Now().Add(10*time.Second)))
// Answer the agent's challenge with a tag derived from the wrong token.
if err := agentClientHandshake(conn, bytes.Repeat([]byte{0xff}, agentTokenLen), false); err != nil {
if err := agentClientHandshake(conn, bytes.Repeat([]byte{0xff}, agentTokenLen), agentGrant{}); err != nil {
// Expected: the server rejects and closes. The read below confirms it
// never reached the greeting.
_ = err
@@ -335,7 +335,8 @@ func TestAgentToken_MatchAllowsHandshake(t *testing.T) {
defer conn.Close()
require.NoError(t, conn.SetDeadline(time.Now().Add(10*time.Second)))
require.NoError(t, agentClientHandshake(conn, token, false))
const peer = "100.64.0.7:51234"
require.NoError(t, agentClientHandshake(conn, token, agentGrant{peerAddr: peer}))
// Re-armed because the handshake clears it on the way out.
require.NoError(t, conn.SetDeadline(time.Now().Add(10*time.Second)))
@@ -350,6 +351,12 @@ func TestAgentToken_MatchAllowsHandshake(t *testing.T) {
_, err = io.ReadFull(conn, version[:])
require.NoError(t, err, "server must keep the connection open after a valid agent token")
assert.Equal(t, "RFB 003.008\n", string(version[:]))
// The session is registered before the greeting goes out, and must carry
// the peer the daemon vouched for rather than the socket it came in on.
sessions := srv.ActiveSessions()
require.Len(t, sessions, 1, "one session must be active")
assert.Equal(t, peer, sessions[0].RemoteAddress, "the session must report the grant's peer")
}
func TestSessionMode_RejectedWhenNoVMGR(t *testing.T) {
+4 -6
View File
@@ -183,8 +183,6 @@ type fbRequest struct {
incremental bool
}
func (s *session) addr() string { return s.conn.RemoteAddr().String() }
// lockWrite takes writeMu and arms the write deadline for the writes that
// follow, returning the unlock. Every server-to-client write goes through it,
// so no write can outlive writeDeadline.
@@ -205,10 +203,10 @@ func (s *session) serve() {
s.encodeCh = make(chan fbRequest, 1)
if err := s.handshake(); err != nil {
s.log.Warnf("handshake with %s: %v", s.addr(), err)
s.log.Warnf("RFB handshake: %v", err)
return
}
s.log.Infof("client connected: %s", s.addr())
s.log.Info("client connected")
// View-only clients can't move the pointer, so default to compositing
// the host cursor into the framebuffer. The client can still send
@@ -246,9 +244,9 @@ func (s *session) serve() {
// messageLoop only ever returns an error, so the interesting question is
// which one: a clean client disconnect is io.EOF and not worth a warning.
if err := s.messageLoop(); !errors.Is(err, io.EOF) {
s.log.Warnf("client %s disconnected: %v", s.addr(), err)
s.log.Warnf("client disconnected: %v", err)
} else {
s.log.Infof("client disconnected: %s", s.addr())
s.log.Info("client disconnected")
}
}