diff --git a/cmd/eipc-server/main.go b/cmd/eipc-server/main.go index e5c9a91..707e0ce 100644 --- a/cmd/eipc-server/main.go +++ b/cmd/eipc-server/main.go @@ -153,8 +153,16 @@ func main() { if err := codec.Unmarshal(msg.Payload, &hb); err != nil { return nil, err } - healthSvc.RecordHeartbeat(hb.Service, hb.Status) - log.Printf("[HEARTBEAT] service=%s status=%s", hb.Service, hb.Status) + // Record under the envelope Source, never the payload's Service + // field: the payload is sender-controlled, so trusting it would + // let any peer forge health for an arbitrary service. The + // envelope Source is bound to the authenticated peer by + // DispatchFor before this handler runs. + if hb.Service != "" && hb.Service != msg.Source { + log.Printf("[HEARTBEAT] ignoring forged service=%q from source=%s", hb.Service, msg.Source) + } + healthSvc.RecordHeartbeat(msg.Source, hb.Status) + log.Printf("[HEARTBEAT] service=%s status=%s", msg.Source, hb.Status) return nil, nil }) @@ -543,8 +551,12 @@ func handleConnection( log.Printf("[AUDIT] failed: %v", err) } - // Set peer capabilities on the endpoint for validation - endpoint.SetPeerCapabilities(peer.Capabilities) + // Bind the authenticated peer identity to the endpoint: every later + // message on this connection must carry this peer's Source and + // session token (Receive rejects forgeries), and only its granted + // capabilities (ValidateCapability). This also sets the peer's + // capabilities, replacing the old SetPeerCapabilities call. + endpoint.BindPeer(peer.BoundPeer()) type authResult struct { Status string `json:"status"` @@ -632,7 +644,9 @@ func handleConnection( continue } - resp, err := router.Dispatch(msg) + // Dispatch bound to the authenticated peer: a forged Source or + // session token never reaches a handler. + resp, err := router.DispatchFor(msg, endpoint.BoundPeer()) if err != nil { log.Printf("[DISPATCH] error: %v", err) errPayload, err := codec.Marshal(core.AckEvent{ diff --git a/core/endpoint.go b/core/endpoint.go index a0aa4f5..529ac13 100644 --- a/core/endpoint.go +++ b/core/endpoint.go @@ -44,6 +44,13 @@ func NewClientEndpoint(conn transport.Connection, codec protocol.Codec, hmacKey func (e *ClientEndpoint) Send(msg Message) error { seq := e.sequence.Add(1) + // Stamp the endpoint's session token when the message carries none, + // so the server can bind the message to this authenticated session. + // An explicitly set SessionID is left untouched. + if msg.SessionID == "" { + msg.SessionID = e.sessionID + } + hdr := protocol.Header{ ServiceID: msg.Source, SessionID: msg.SessionID, @@ -118,6 +125,8 @@ type ServerEndpoint struct { sendMu sync.Mutex seq atomic.Uint64 peerCapabilities []string + peerMu sync.RWMutex + peer *AuthenticatedPeer } // SetPeerCapabilities sets the authenticated peer's capability list. @@ -129,6 +138,48 @@ func (e *ServerEndpoint) SetPeerCapabilities(caps []string) { e.peerCapabilities = cpy } +// BindPeer binds the authenticated peer to this connection. Once bound, +// Receive rejects any message whose Source or SessionID does not match +// the peer, so a peer cannot impersonate another service on this +// connection. Binding also sets the peer's capabilities for +// ValidateCapability. Pass nil to clear the binding. +func (e *ServerEndpoint) BindPeer(peer *AuthenticatedPeer) { + e.peerMu.Lock() + if peer == nil { + e.peer = nil + } else { + cpy := *peer + caps := make([]string, len(peer.Capabilities)) + copy(caps, peer.Capabilities) + cpy.Capabilities = caps + e.peer = &cpy + } + e.peerMu.Unlock() + e.SetPeerCapabilities(peerCapabilitiesOf(peer)) +} + +func peerCapabilitiesOf(peer *AuthenticatedPeer) []string { + if peer == nil { + return nil + } + return peer.Capabilities +} + +// BoundPeer returns a copy of the peer bound to this connection, or nil +// when no peer is bound. +func (e *ServerEndpoint) BoundPeer() *AuthenticatedPeer { + e.peerMu.RLock() + defer e.peerMu.RUnlock() + if e.peer == nil { + return nil + } + cpy := *e.peer + caps := make([]string, len(e.peer.Capabilities)) + copy(caps, e.peer.Capabilities) + cpy.Capabilities = caps + return &cpy +} + // ValidateCapability checks if msgCap is in the peer's granted capabilities. func (e *ServerEndpoint) ValidateCapability(msgCap string) error { if msgCap == "" { @@ -232,7 +283,7 @@ func (e *ServerEndpoint) Receive() (Message, error) { ts = t } - return Message{ + msg := Message{ Version: frame.Version, Type: msgTypeFromByte(frame.MsgType), Source: hdr.ServiceID, @@ -242,7 +293,20 @@ func (e *ServerEndpoint) Receive() (Message, error) { Priority: Priority(hdr.Priority), Capability: hdr.Capability, Payload: frame.Payload, - }, nil + } + + // Identity binding (PR-EIPC-2): once a peer is bound to this + // connection, only messages genuinely from that peer are accepted. + // A forged Source or a session token from another session is + // rejected here, before dispatch. Capability is deliberately not + // checked here: the server loop enforces it with an error ack and + // keeps the connection up, while an identity forgery drops it. + // Unbound endpoints (e.g. during the auth handshake) are unaffected. + if err := e.BoundPeer().ValidateIdentity(msg); err != nil { + return Message{}, err + } + + return msg, nil } func (e *ServerEndpoint) Close() error { diff --git a/core/endpoint_identity_test.go b/core/endpoint_identity_test.go new file mode 100644 index 0000000..72442c8 --- /dev/null +++ b/core/endpoint_identity_test.go @@ -0,0 +1,174 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package core + +import ( + "errors" + "testing" + + "github.com/embeddedos-org/eipc/protocol" +) + +// TestServerEndpointBindPeer_AcceptsGenuineMessage proves a bound endpoint +// accepts a message whose Source and SessionID match the peer, including +// the client-side session stamping (ClientEndpoint stamps its session +// token when the message carries none). +func TestServerEndpointBindPeer_AcceptsGenuineMessage(t *testing.T) { + clientConn, serverConn := setupTCPPair(t) + defer clientConn.Close() + defer serverConn.Close() + + hmacKey := []byte("test-secret-key-for-hmac-256bit!") + c := protocol.DefaultCodec() + + clientEP := NewClientEndpoint(clientConn, c, hmacKey, "tok-abc123") + serverEP := NewServerEndpoint(serverConn, c, hmacKey) + serverEP.BindPeer(&AuthenticatedPeer{ + ServiceID: "ebot.client", + SessionToken: "tok-abc123", + Capabilities: []string{"ai:chat"}, + }) + + msgCh := make(chan Message, 1) + errCh := make(chan error, 1) + go func() { + received, err := serverEP.Receive() + if err != nil { + errCh <- err + return + } + msgCh <- received + }() + + // No SessionID set on the message: the client endpoint stamps it. + msg := NewMessage(TypeChat, "ebot.client", []byte(`{"prompt":"hi"}`)) + if err := clientEP.Send(msg); err != nil { + t.Fatalf("Send failed: %v", err) + } + + select { + case received := <-msgCh: + if received.Source != "ebot.client" { + t.Errorf("source mismatch: got %q", received.Source) + } + if received.SessionID != "tok-abc123" { + t.Errorf("session was not stamped/bound: got %q", received.SessionID) + } + case err := <-errCh: + t.Fatalf("Receive rejected a genuine message: %v", err) + } +} + +// TestServerEndpointBindPeer_RejectsForgedSource proves a bound endpoint +// drops a message that claims another service's identity, even when the +// session token is correct. +func TestServerEndpointBindPeer_RejectsForgedSource(t *testing.T) { + clientConn, serverConn := setupTCPPair(t) + defer clientConn.Close() + defer serverConn.Close() + + hmacKey := []byte("test-secret-key-for-hmac-256bit!") + c := protocol.DefaultCodec() + + clientEP := NewClientEndpoint(clientConn, c, hmacKey, "tok-abc123") + serverEP := NewServerEndpoint(serverConn, c, hmacKey) + serverEP.BindPeer(&AuthenticatedPeer{ + ServiceID: "ebot.client", + SessionToken: "tok-abc123", + }) + + errCh := make(chan error, 1) + go func() { + _, err := serverEP.Receive() + errCh <- err + }() + + msg := NewMessage(TypeChat, "eipc-server", []byte(`{}`)) + msg.SessionID = "tok-abc123" + if err := clientEP.Send(msg); err != nil { + t.Fatalf("Send failed: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for forged source, got %v", err) + } +} + +// TestServerEndpointBindPeer_RejectsWrongSession proves a bound endpoint +// drops a message from the right service carrying another session's token. +func TestServerEndpointBindPeer_RejectsWrongSession(t *testing.T) { + clientConn, serverConn := setupTCPPair(t) + defer clientConn.Close() + defer serverConn.Close() + + hmacKey := []byte("test-secret-key-for-hmac-256bit!") + c := protocol.DefaultCodec() + + clientEP := NewClientEndpoint(clientConn, c, hmacKey, "tok-other-session") + serverEP := NewServerEndpoint(serverConn, c, hmacKey) + serverEP.BindPeer(&AuthenticatedPeer{ + ServiceID: "ebot.client", + SessionToken: "tok-abc123", + }) + + errCh := make(chan error, 1) + go func() { + _, err := serverEP.Receive() + errCh <- err + }() + + msg := NewMessage(TypeChat, "ebot.client", []byte(`{}`)) + msg.SessionID = "tok-other-session" + if err := clientEP.Send(msg); err != nil { + t.Fatalf("Send failed: %v", err) + } + + if err := <-errCh; !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for wrong session, got %v", err) + } +} + +// TestServerEndpointBindPeer_ClearsOnNil proves BindPeer(nil) removes the +// binding and restores the previous unbound behavior. +func TestServerEndpointBindPeer_ClearsOnNil(t *testing.T) { + clientConn, serverConn := setupTCPPair(t) + defer clientConn.Close() + defer serverConn.Close() + + hmacKey := []byte("test-secret-key-for-hmac-256bit!") + c := protocol.DefaultCodec() + + serverEP := NewServerEndpoint(serverConn, c, hmacKey) + serverEP.BindPeer(&AuthenticatedPeer{ServiceID: "ebot.client", SessionToken: "tok-abc123"}) + if serverEP.BoundPeer() == nil { + t.Fatal("expected a bound peer") + } + serverEP.BindPeer(nil) + if serverEP.BoundPeer() != nil { + t.Fatal("expected the binding to be cleared") + } + + msgCh := make(chan Message, 1) + errCh := make(chan error, 1) + go func() { + received, err := serverEP.Receive() + if err != nil { + errCh <- err + return + } + msgCh <- received + }() + + clientEP := NewClientEndpoint(clientConn, c, hmacKey, "tok-abc123") + msg := NewMessage(TypeChat, "anyone", []byte(`{}`)) + if err := clientEP.Send(msg); err != nil { + t.Fatalf("Send failed: %v", err) + } + + select { + case <-msgCh: + case err := <-errCh: + t.Fatalf("unbound endpoint rejected a message: %v", err) + } +} diff --git a/core/identity.go b/core/identity.go new file mode 100644 index 0000000..c462b5b --- /dev/null +++ b/core/identity.go @@ -0,0 +1,73 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package core + +import "fmt" + +// AuthenticatedPeer is the core-level view of a peer that has completed +// challenge-response authentication (see security/auth.Authenticator). +// It is the identity that dispatch binds a message to: a message is only +// dispatched for the peer it claims to come from when its envelope +// (Source + SessionID) matches this peer and its Capability is granted. +// +// The type lives in core rather than security/auth so the router, broker, +// health service and endpoints can enforce the binding without an import +// cycle (security/auth already imports core). +type AuthenticatedPeer struct { + // ServiceID is the authenticated service identity (e.g. "ebot.client"). + ServiceID string + // SessionToken is the session token issued at authentication time. + // Senders carry it as Message.SessionID. + SessionToken string + // Capabilities granted to this peer at authentication time. + Capabilities []string +} + +// ValidateIdentity checks the envelope identity of msg against the +// authenticated peer. A nil peer means "no peer bound" (unsecured/local +// path) and passes every message; callers that require authentication +// must not pass nil. +// +// When a peer is bound the check is fail-closed: +// - msg.Source must be present and equal the peer's ServiceID, so a +// peer cannot impersonate another service by forging Source; +// - msg.SessionID must be present and equal the peer's SessionToken, +// so an envelope from another session is rejected. +func (p *AuthenticatedPeer) ValidateIdentity(msg Message) error { + if p == nil { + return nil + } + if msg.Source == "" { + return fmt.Errorf("%w: message has no source", ErrAuth) + } + if msg.Source != p.ServiceID { + return fmt.Errorf("%w: source %q is not the authenticated peer %q", ErrAuth, msg.Source, p.ServiceID) + } + if msg.SessionID == "" { + return fmt.Errorf("%w: message from %q carries no session", ErrAuth, msg.Source) + } + if msg.SessionID != p.SessionToken { + return fmt.Errorf("%w: session mismatch for %q", ErrAuth, msg.Source) + } + return nil +} + +// ValidateMessage is ValidateIdentity plus capability validation: when +// msg.Capability is set it must be one of the peer's granted +// capabilities. An empty Capability passes (the message claims no +// capability-gated action). +func (p *AuthenticatedPeer) ValidateMessage(msg Message) error { + if err := p.ValidateIdentity(msg); err != nil { + return err + } + if p == nil || msg.Capability == "" { + return nil + } + for _, c := range p.Capabilities { + if c == msg.Capability { + return nil + } + } + return fmt.Errorf("%w: peer %q lacks capability %q", ErrCapability, p.ServiceID, msg.Capability) +} diff --git a/core/identity_test.go b/core/identity_test.go new file mode 100644 index 0000000..6971cbd --- /dev/null +++ b/core/identity_test.go @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package core + +import ( + "errors" + "testing" +) + +func testPeer() *AuthenticatedPeer { + return &AuthenticatedPeer{ + ServiceID: "ebot.client", + SessionToken: "tok-abc123", + Capabilities: []string{"ai:chat", "ui:control"}, + } +} + +func testMessage() Message { + return Message{ + Type: TypeChat, + Source: "ebot.client", + SessionID: "tok-abc123", + Capability: "ai:chat", + } +} + +func TestAuthenticatedPeerValidateMessage_OK(t *testing.T) { + if err := testPeer().ValidateMessage(testMessage()); err != nil { + t.Fatalf("valid message rejected: %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_EmptyCapabilityOK(t *testing.T) { + msg := testMessage() + msg.Capability = "" + if err := testPeer().ValidateMessage(msg); err != nil { + t.Fatalf("message without capability rejected: %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_ForgedSource(t *testing.T) { + msg := testMessage() + msg.Source = "eipc-server" // impersonating another service + if err := testPeer().ValidateMessage(msg); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for forged source, got %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_MissingSource(t *testing.T) { + msg := testMessage() + msg.Source = "" + if err := testPeer().ValidateMessage(msg); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for missing source, got %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_WrongSession(t *testing.T) { + msg := testMessage() + msg.SessionID = "tok-stolen-from-another-session" + if err := testPeer().ValidateMessage(msg); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for wrong session, got %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_MissingSession(t *testing.T) { + msg := testMessage() + msg.SessionID = "" + if err := testPeer().ValidateMessage(msg); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for missing session, got %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_UngrantedCapability(t *testing.T) { + msg := testMessage() + msg.Capability = "device:write" // not granted to ebot.client + if err := testPeer().ValidateMessage(msg); !errors.Is(err, ErrCapability) { + t.Fatalf("expected ErrCapability for ungranted capability, got %v", err) + } +} + +func TestAuthenticatedPeerValidateMessage_NilPeerAllows(t *testing.T) { + var peer *AuthenticatedPeer + msg := testMessage() + msg.Source = "anyone" + msg.SessionID = "whatever" + if err := peer.ValidateMessage(msg); err != nil { + t.Fatalf("nil peer should pass every message, got %v", err) + } +} + +func TestAuthenticatedPeerValidateIdentity_IgnoresCapability(t *testing.T) { + // Identity binding is source + session only; capability is a + // separate dispatch-time check. + msg := testMessage() + msg.Capability = "device:write" + if err := testPeer().ValidateIdentity(msg); err != nil { + t.Fatalf("ValidateIdentity should ignore capability, got %v", err) + } +} diff --git a/core/router.go b/core/router.go index dabdbd4..b4607c1 100644 --- a/core/router.go +++ b/core/router.go @@ -33,6 +33,10 @@ func (r *Router) Handle(msgType MessageType, handler HandlerFunc) { } // Dispatch routes a single message to its registered handler. +// +// Dispatch performs no identity check: any message reaches its handler. +// That is the unsecured/local path. Secured servers must use DispatchFor +// with the peer bound to the connection instead. func (r *Router) Dispatch(msg Message) (*Message, error) { r.mu.RLock() handler, ok := r.handlers[msg.Type] @@ -44,8 +48,32 @@ func (r *Router) Dispatch(msg Message) (*Message, error) { return handler(msg) } +// DispatchFor validates msg against the authenticated peer before +// dispatching: a forged Source or SessionID, or an ungranted Capability, +// never reaches a handler. A nil peer behaves exactly like Dispatch +// (unsecured/local path). +func (r *Router) DispatchFor(msg Message, peer *AuthenticatedPeer) (*Message, error) { + if err := peer.ValidateMessage(msg); err != nil { + return nil, err + } + return r.Dispatch(msg) +} + // DispatchBatch processes a batch of messages in priority order (P0 first). func (r *Router) DispatchBatch(messages []Message) []DispatchResult { + return r.dispatchBatch(messages, r.Dispatch) +} + +// DispatchBatchFor is DispatchBatch with per-message identity binding: +// each message is validated against the authenticated peer before its +// handler runs. A nil peer behaves exactly like DispatchBatch. +func (r *Router) DispatchBatchFor(messages []Message, peer *AuthenticatedPeer) []DispatchResult { + return r.dispatchBatch(messages, func(msg Message) (*Message, error) { + return r.DispatchFor(msg, peer) + }) +} + +func (r *Router) dispatchBatch(messages []Message, dispatch func(Message) (*Message, error)) []DispatchResult { pq := make(priorityQueue, len(messages)) for i, msg := range messages { pq[i] = &priorityItem{msg: msg, index: i} @@ -55,7 +83,7 @@ func (r *Router) DispatchBatch(messages []Message) []DispatchResult { results := make([]DispatchResult, len(messages)) for pq.Len() > 0 { item := heap.Pop(&pq).(*priorityItem) - resp, err := r.Dispatch(item.msg) + resp, err := dispatch(item.msg) results[item.index] = DispatchResult{Response: resp, Err: err} } return results diff --git a/core/router_identity_test.go b/core/router_identity_test.go new file mode 100644 index 0000000..fdb4c0e --- /dev/null +++ b/core/router_identity_test.go @@ -0,0 +1,128 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package core + +import ( + "errors" + "testing" +) + +func TestRouterDispatchFor_OK(t *testing.T) { + r := NewRouter() + called := false + r.Handle(TypeChat, func(msg Message) (*Message, error) { + called = true + return nil, nil + }) + + peer := testPeer() + msg := testMessage() + if _, err := r.DispatchFor(msg, peer); err != nil { + t.Fatalf("DispatchFor rejected a valid message: %v", err) + } + if !called { + t.Error("handler was not called for a valid message") + } +} + +func TestRouterDispatchFor_ForgedSourceNeverReachesHandler(t *testing.T) { + r := NewRouter() + called := false + r.Handle(TypeChat, func(msg Message) (*Message, error) { + called = true + return nil, nil + }) + + msg := testMessage() + msg.Source = "eipc-server" + if _, err := r.DispatchFor(msg, testPeer()); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for forged source, got %v", err) + } + if called { + t.Error("handler must not run for a forged message") + } +} + +func TestRouterDispatchFor_WrongSession(t *testing.T) { + r := NewRouter() + called := false + r.Handle(TypeChat, func(msg Message) (*Message, error) { + called = true + return nil, nil + }) + + msg := testMessage() + msg.SessionID = "tok-other-session" + if _, err := r.DispatchFor(msg, testPeer()); !errors.Is(err, ErrAuth) { + t.Fatalf("expected ErrAuth for wrong session, got %v", err) + } + if called { + t.Error("handler must not run for a wrong-session message") + } +} + +func TestRouterDispatchFor_UngrantedCapability(t *testing.T) { + r := NewRouter() + called := false + r.Handle(TypeChat, func(msg Message) (*Message, error) { + called = true + return nil, nil + }) + + msg := testMessage() + msg.Capability = "device:write" + if _, err := r.DispatchFor(msg, testPeer()); !errors.Is(err, ErrCapability) { + t.Fatalf("expected ErrCapability for ungranted capability, got %v", err) + } + if called { + t.Error("handler must not run for an ungranted capability") + } +} + +func TestRouterDispatchFor_NilPeerBehavesLikeDispatch(t *testing.T) { + r := NewRouter() + called := false + r.Handle(TypeChat, func(msg Message) (*Message, error) { + called = true + return nil, nil + }) + + msg := testMessage() + msg.Source = "unbound" + msg.SessionID = "" + if _, err := r.DispatchFor(msg, nil); err != nil { + t.Fatalf("nil peer should dispatch like Dispatch, got %v", err) + } + if !called { + t.Error("handler was not called for nil-peer dispatch") + } +} + +func TestRouterDispatchBatchFor_MixedBatch(t *testing.T) { + r := NewRouter() + var seen []string + r.Handle(TypeChat, func(msg Message) (*Message, error) { + seen = append(seen, msg.Source) + return nil, nil + }) + + peer := testPeer() + good := testMessage() + forged := testMessage() + forged.Source = "eipc-server" + + results := r.DispatchBatchFor([]Message{forged, good}, peer) + if len(results) != 2 { + t.Fatalf("expected 2 results, got %d", len(results)) + } + if !errors.Is(results[0].Err, ErrAuth) { + t.Fatalf("expected ErrAuth for forged message, got %v", results[0].Err) + } + if results[1].Err != nil { + t.Fatalf("expected valid message to dispatch, got %v", results[1].Err) + } + if len(seen) != 1 || seen[0] != "ebot.client" { + t.Errorf("expected only the valid message dispatched, got %v", seen) + } +} diff --git a/docs/security-model.md b/docs/security-model.md index ee4932a..e91762c 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -118,6 +118,33 @@ integrity must be moved to keyed endpoints. This change covers the Go core only; the C SDK server (`sdk/c/src/eipc_server.c`) still verifies HMAC only "if present" and is a recorded follow-up. +### Identity binding in dispatch + +Authentication used to end at the handshake: after challenge-response +succeeded, nothing checked that later messages on the connection +actually came from the authenticated peer. A peer authenticated as +`ebot.client` could send a message with `Source: eipc-server` (or any +other service) and it would be dispatched, and the heartbeat handler +recorded liveness under the sender-controlled payload `Service` field, +so any peer could forge health for an arbitrary service ID. + +After authentication the server now binds the peer to the connection +(`ServerEndpoint.BindPeer`): every later message must carry the peer's +`Source` and session token as `SessionID`, or `Receive` rejects it with +`ErrAuth` and the connection is dropped. Dispatch goes through +`Router.DispatchFor` (plus `Broker.RouteFor` / `FanoutFor` and +`health.Service.RecordHeartbeatFor` for direct users), which additionally +validates `Capability` against the peer's granted capabilities. The +heartbeat handler records liveness under the envelope `Source`, never +the payload's `Service` field. The shipped client already stamps its +session token as `SessionID` on every post-auth message, so in-repo +peers are unaffected. + +The unbound path is unchanged by design: `Dispatch`, `Route`, +`Fanout`, and `RecordHeartbeat` perform no identity check and remain +the unsecured/local API. Callers that need authentication must use the +`*For` variants or bind the peer — the plain methods do not fail closed. + --- ## 3. Replay Protection diff --git a/security/auth/identity.go b/security/auth/identity.go index a9da571..d81c630 100644 --- a/security/auth/identity.go +++ b/security/auth/identity.go @@ -169,6 +169,22 @@ func (a *Authenticator) Authenticate(serviceID string) (*PeerIdentity, error) { return peer, nil } +// BoundPeer returns the core-level view of this authenticated peer for +// dispatch binding (see core.AuthenticatedPeer). Returns nil for a nil +// identity. +func (p *PeerIdentity) BoundPeer() *core.AuthenticatedPeer { + if p == nil { + return nil + } + caps := make([]string, len(p.Capabilities)) + copy(caps, p.Capabilities) + return &core.AuthenticatedPeer{ + ServiceID: p.ServiceID, + SessionToken: p.SessionToken, + Capabilities: caps, + } +} + // ValidateSession checks whether a session token is valid and not expired. func (a *Authenticator) ValidateSession(token string) (*PeerIdentity, error) { a.mu.RLock() diff --git a/security/auth/identity_binding_test.go b/security/auth/identity_binding_test.go new file mode 100644 index 0000000..1eab648 --- /dev/null +++ b/security/auth/identity_binding_test.go @@ -0,0 +1,59 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package auth + +import ( + "testing" + + "github.com/embeddedos-org/eipc/core" +) + +func TestPeerIdentityBoundPeer_RoundTrip(t *testing.T) { + authenticator := NewAuthenticator([]byte("test-key-32-bytes-long-enough!!"), map[string][]string{ + "ebot.client": {"ai:chat"}, + }) + + peer, err := authenticator.Authenticate("ebot.client") + if err != nil { + t.Fatalf("Authenticate: %v", err) + } + + bound := peer.BoundPeer() + if bound == nil { + t.Fatal("expected a bound peer") + } + if bound.ServiceID != "ebot.client" { + t.Errorf("ServiceID mismatch: got %q", bound.ServiceID) + } + if bound.SessionToken != peer.SessionToken { + t.Error("SessionToken was not carried over") + } + if len(bound.Capabilities) != 1 || bound.Capabilities[0] != "ai:chat" { + t.Errorf("Capabilities mismatch: got %v", bound.Capabilities) + } + + // The bound peer validates a genuine message from this identity. + msg := core.Message{ + Type: core.TypeChat, + Source: "ebot.client", + SessionID: peer.SessionToken, + Capability: "ai:chat", + } + if err := bound.ValidateMessage(msg); err != nil { + t.Fatalf("genuine message rejected: %v", err) + } + + // ... and rejects a message impersonating another service. + msg.Source = "eipc-server" + if err := bound.ValidateMessage(msg); err == nil { + t.Fatal("expected rejection for forged source") + } +} + +func TestPeerIdentityBoundPeer_NilIdentity(t *testing.T) { + var peer *PeerIdentity + if peer.BoundPeer() != nil { + t.Fatal("expected nil for nil identity") + } +} diff --git a/services/broker/broker.go b/services/broker/broker.go index 6150cf9..a449fa3 100644 --- a/services/broker/broker.go +++ b/services/broker/broker.go @@ -122,6 +122,17 @@ func (b *Broker) Route(msg core.Message) []RouteResult { return results } +// RouteFor validates msg against the authenticated peer before routing: +// a forged Source or SessionID, or an ungranted Capability, is rejected +// and nothing is delivered. A nil peer behaves exactly like Route +// (unsecured/local path). +func (b *Broker) RouteFor(msg core.Message, peer *core.AuthenticatedPeer) ([]RouteResult, error) { + if err := peer.ValidateMessage(msg); err != nil { + return nil, err + } + return b.Route(msg), nil +} + func (b *Broker) Fanout(msg core.Message) []RouteResult { b.mu.RLock() subs := make([]*Subscriber, 0, len(b.subscribers)) @@ -137,6 +148,16 @@ func (b *Broker) Fanout(msg core.Message) []RouteResult { return results } +// FanoutFor is Fanout with the same identity binding: the message is +// only fanned out to subscribers when it genuinely comes from the +// authenticated peer. A nil peer behaves exactly like Fanout. +func (b *Broker) FanoutFor(msg core.Message, peer *core.AuthenticatedPeer) ([]RouteResult, error) { + if err := peer.ValidateMessage(msg); err != nil { + return nil, err + } + return b.Fanout(msg), nil +} + func (b *Broker) Subscribers() []string { b.mu.RLock() defer b.mu.RUnlock() diff --git a/services/broker/broker_identity_test.go b/services/broker/broker_identity_test.go new file mode 100644 index 0000000..3e8a337 --- /dev/null +++ b/services/broker/broker_identity_test.go @@ -0,0 +1,123 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package broker + +import ( + "errors" + "testing" + + "github.com/embeddedos-org/eipc/core" +) + +func testPeer() *core.AuthenticatedPeer { + return &core.AuthenticatedPeer{ + ServiceID: "eni.min", + SessionToken: "tok-abc123", + Capabilities: []string{"ui:control"}, + } +} + +func TestBrokerRouteFor_OK(t *testing.T) { + b := NewBroker(nil, nil) + ep := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "eai.agent", Endpoint: ep, Priority: core.PriorityP1}) + b.AddRoute(core.TypeIntent, "eai.agent") + + msg := core.NewMessage(core.TypeIntent, "eni.min", []byte(`{}`)) + msg.SessionID = "tok-abc123" + msg.Capability = "ui:control" + results, err := b.RouteFor(msg, testPeer()) + if err != nil { + t.Fatalf("RouteFor rejected a valid message: %v", err) + } + if len(results) != 1 || len(ep.sent) != 1 { + t.Fatalf("expected 1 delivery, got %d results / %d sent", len(results), len(ep.sent)) + } +} + +func TestBrokerRouteFor_ForgedSourceDeliversNothing(t *testing.T) { + b := NewBroker(nil, nil) + ep := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "eai.agent", Endpoint: ep, Priority: core.PriorityP1}) + b.AddRoute(core.TypeIntent, "eai.agent") + + msg := core.NewMessage(core.TypeIntent, "eipc-server", []byte(`{}`)) + msg.SessionID = "tok-abc123" + _, err := b.RouteFor(msg, testPeer()) + if !errors.Is(err, core.ErrAuth) { + t.Fatalf("expected ErrAuth for forged source, got %v", err) + } + if len(ep.sent) != 0 { + t.Fatalf("forged message must not be delivered, got %d sent", len(ep.sent)) + } +} + +func TestBrokerRouteFor_WrongSessionDeliversNothing(t *testing.T) { + b := NewBroker(nil, nil) + ep := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "eai.agent", Endpoint: ep, Priority: core.PriorityP1}) + b.AddRoute(core.TypeIntent, "eai.agent") + + msg := core.NewMessage(core.TypeIntent, "eni.min", []byte(`{}`)) + msg.SessionID = "tok-other-session" + _, err := b.RouteFor(msg, testPeer()) + if !errors.Is(err, core.ErrAuth) { + t.Fatalf("expected ErrAuth for wrong session, got %v", err) + } + if len(ep.sent) != 0 { + t.Fatalf("wrong-session message must not be delivered, got %d sent", len(ep.sent)) + } +} + +func TestBrokerRouteFor_NilPeerBehavesLikeRoute(t *testing.T) { + b := NewBroker(nil, nil) + ep := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "eai.agent", Endpoint: ep, Priority: core.PriorityP1}) + b.AddRoute(core.TypeIntent, "eai.agent") + + msg := core.NewMessage(core.TypeIntent, "unbound", []byte(`{}`)) + results, err := b.RouteFor(msg, nil) + if err != nil { + t.Fatalf("nil peer should route like Route, got %v", err) + } + if len(results) != 1 || len(ep.sent) != 1 { + t.Fatalf("expected 1 delivery, got %d results / %d sent", len(results), len(ep.sent)) + } +} + +func TestBrokerFanoutFor_ForgedDeliversNothing(t *testing.T) { + b := NewBroker(nil, nil) + ep1 := &mockEndpoint{} + ep2 := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "svc1", Endpoint: ep1}) + _ = b.Subscribe(&Subscriber{ServiceID: "svc2", Endpoint: ep2}) + + msg := core.NewMessage(core.TypeHeartbeat, "eipc-server", []byte(`{}`)) + msg.SessionID = "tok-abc123" + _, err := b.FanoutFor(msg, testPeer()) + if !errors.Is(err, core.ErrAuth) { + t.Fatalf("expected ErrAuth for forged fanout, got %v", err) + } + if len(ep1.sent) != 0 || len(ep2.sent) != 0 { + t.Fatal("forged fanout must reach no subscriber") + } +} + +func TestBrokerFanoutFor_OK(t *testing.T) { + b := NewBroker(nil, nil) + ep1 := &mockEndpoint{} + ep2 := &mockEndpoint{} + _ = b.Subscribe(&Subscriber{ServiceID: "svc1", Endpoint: ep1}) + _ = b.Subscribe(&Subscriber{ServiceID: "svc2", Endpoint: ep2}) + + msg := core.NewMessage(core.TypeHeartbeat, "eni.min", []byte(`{}`)) + msg.SessionID = "tok-abc123" + results, err := b.FanoutFor(msg, testPeer()) + if err != nil { + t.Fatalf("FanoutFor rejected a valid message: %v", err) + } + if len(results) != 2 || len(ep1.sent) != 1 || len(ep2.sent) != 1 { + t.Fatalf("expected fanout to 2 subscribers, got %d results", len(results)) + } +} diff --git a/services/health/health.go b/services/health/health.go index 6fb4755..5030360 100644 --- a/services/health/health.go +++ b/services/health/health.go @@ -4,8 +4,11 @@ package health import ( + "fmt" "sync" "time" + + "github.com/embeddedos-org/eipc/core" ) // PeerStatus tracks the liveness state of a peer. @@ -40,6 +43,11 @@ func NewService(interval, timeout time.Duration) *Service { } // RecordHeartbeat updates the liveness timestamp for a peer. +// +// RecordHeartbeat performs no identity check: any caller can record +// health for any service ID. That is the unsecured/local path. Secured +// servers must use RecordHeartbeatFor with the peer bound to the +// connection instead. func (s *Service) RecordHeartbeat(serviceID, status string) { s.mu.Lock() defer s.mu.Unlock() @@ -51,6 +59,18 @@ func (s *Service) RecordHeartbeat(serviceID, status string) { } } +// RecordHeartbeatFor records a heartbeat only when serviceID is the +// authenticated peer's own identity: a peer can no longer report health +// for an arbitrary service ID. A nil peer behaves exactly like +// RecordHeartbeat (unsecured/local path). +func (s *Service) RecordHeartbeatFor(serviceID, status string, peer *core.AuthenticatedPeer) error { + if peer != nil && serviceID != peer.ServiceID { + return fmt.Errorf("%w: heartbeat for %q is not from authenticated peer %q", core.ErrAuth, serviceID, peer.ServiceID) + } + s.RecordHeartbeat(serviceID, status) + return nil +} + // IsAlive returns whether a peer has been heard from within the timeout window. func (s *Service) IsAlive(serviceID string) bool { s.mu.RLock() diff --git a/services/health/health_identity_test.go b/services/health/health_identity_test.go new file mode 100644 index 0000000..6f2c969 --- /dev/null +++ b/services/health/health_identity_test.go @@ -0,0 +1,54 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 EoS Project + +package health + +import ( + "errors" + "testing" + "time" + + "github.com/embeddedos-org/eipc/core" +) + +func TestRecordHeartbeatFor_OK(t *testing.T) { + svc := NewService(5*time.Second, 15*time.Second) + peer := &core.AuthenticatedPeer{ServiceID: "eni.min", SessionToken: "tok-abc123"} + + if err := svc.RecordHeartbeatFor("eni.min", "ready", peer); err != nil { + t.Fatalf("own heartbeat rejected: %v", err) + } + if !svc.IsAlive("eni.min") { + t.Error("peer should be alive after its own heartbeat") + } +} + +func TestRecordHeartbeatFor_ForgedServiceRejected(t *testing.T) { + svc := NewService(5*time.Second, 15*time.Second) + peer := &core.AuthenticatedPeer{ServiceID: "eni.min", SessionToken: "tok-abc123"} + + // A peer authenticated as eni.min reports health for eai.agent. + if err := svc.RecordHeartbeatFor("eai.agent", "ready", peer); !errors.Is(err, core.ErrAuth) { + t.Fatalf("expected ErrAuth for forged heartbeat, got %v", err) + } + if svc.IsAlive("eai.agent") { + t.Error("forged heartbeat must not mark the victim alive") + } + if len(svc.AllPeers()) != 0 { + t.Fatalf("forged heartbeat must record nothing, got %d peers", len(svc.AllPeers())) + } + if svc.IsAlive("eni.min") { + t.Error("forged heartbeat must not mark the sender alive either") + } +} + +func TestRecordHeartbeatFor_NilPeerBehavesLikeRecord(t *testing.T) { + svc := NewService(5*time.Second, 15*time.Second) + + if err := svc.RecordHeartbeatFor("eni.min", "ready", nil); err != nil { + t.Fatalf("nil peer should record like RecordHeartbeat, got %v", err) + } + if !svc.IsAlive("eni.min") { + t.Error("peer should be alive after nil-peer heartbeat") + } +}