feat(secureservice): audit admission decisions

net/peerservice/peerservice.go

@@ -12,6 +12,7 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/pool"
 	"github.com/anyproto/any-sync/net/rpc/server"
+	"github.com/anyproto/any-sync/net/secureservice"
 	"github.com/anyproto/any-sync/net/transport"
 	"github.com/anyproto/any-sync/net/transport/quic"
 	"github.com/anyproto/any-sync/net/transport/webtransport"
@@ -51,6 +52,8 @@ type peerService struct {
 	server       server.DRPCServer
 	preferQuic   bool
 	mu           sync.RWMutex
+
+	admissionTokenProvider secureservice.AdmissionTokenProvider
 }
 
 func (p *peerService) Init(a *app.App) (err error) {
@@ -69,6 +72,11 @@ func (p *peerService) Init(a *app.App) (err error) {
 	p.nodeConf = a.MustComponent(nodeconf.CName).(nodeconf.NodeConf)
 	p.pool = a.MustComponent(pool.CName).(pool.Pool)
 	p.server = a.MustComponent(server.CName).(server.DRPCServer)
+	if provider, providerErr := app.GetComponent[secureservice.AdmissionTokenProvider](a); providerErr == nil {
+		p.admissionTokenProvider = provider
+	} else if !errors.Is(providerErr, app.ErrComponentNotFound) {
+		return providerErr
+	}
 	p.peerAddrs = map[string][]string{}
 	return nil
 }
@@ -118,6 +126,9 @@ func (p *peerService) Dial(ctx context.Context, peerId string) (pr peer.Peer, er
 
 	// Pass expected peerId in context for transports that need it (e.g. WebTransport)
 	ctx = peer.CtxWithExpectedPeerId(ctx, peerId)
+	if ctx, err = p.withOutboundAdmissionToken(ctx, peerId); err != nil {
+		return nil, err
+	}
 
 	var mc transport.MultiConn
 	log.DebugCtx(ctx, "dial", zap.String("peerId", peerId), zap.Strings("addrs", addrs))
@@ -141,6 +152,20 @@ func (p *peerService) Dial(ctx context.Context, peerId string) (pr peer.Peer, er
 	return peer.NewPeer(mc, p.server)
 }
 
+func (p *peerService) withOutboundAdmissionToken(ctx context.Context, peerId string) (context.Context, error) {
+	if p.admissionTokenProvider == nil || secureservice.CtxOutboundAdmissionToken(ctx) != "" {
+		return ctx, nil
+	}
+	token, err := p.admissionTokenProvider.OutboundAdmissionToken(ctx, peerId)
+	if err != nil {
+		return nil, err
+	}
+	if token == "" {
+		return ctx, nil
+	}
+	return secureservice.CtxWithOutboundAdmissionToken(ctx, token), nil
+}
+
 func (p *peerService) dialScheme(ctx context.Context, sch string, addrs []string) (mc transport.MultiConn, err error) {
 	var tr transport.Transport
 	switch sch {

net/peerservice/peerservice_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/pool"
 	"github.com/anyproto/any-sync/net/rpc/rpctest"
+	"github.com/anyproto/any-sync/net/secureservice"
 	"github.com/anyproto/any-sync/net/transport"
 	"github.com/anyproto/any-sync/net/transport/mock_transport"
 	"github.com/anyproto/any-sync/net/transport/quic"
@@ -159,6 +160,67 @@ func TestPeerService_DialWebTransport(t *testing.T) {
 	})
 }
 
+func TestPeerService_DialAdmissionTokenProvider(t *testing.T) {
+	const peerId = "p1"
+	addrs := []string{"yamux://127.0.0.1:1111"}
+
+	t.Run("injects provider token", func(t *testing.T) {
+		provider := &testAdmissionTokenProvider{token: "provider-token"}
+		fx := newFixtureWithAdmissionTokenProvider(t, provider)
+		defer fx.finish(t)
+		fx.PreferQuic(false)
+
+		fx.nodeConf.EXPECT().PeerAddresses(peerId).Return(addrs, true)
+		fx.yamux.MockTransport.EXPECT().Dial(gomock.Any(), "127.0.0.1:1111").DoAndReturn(
+			func(dialCtx context.Context, addr string) (transport.MultiConn, error) {
+				assert.Equal(t, "provider-token", secureservice.CtxOutboundAdmissionToken(dialCtx))
+				expectedPeerId, err := peer.CtxExpectedPeerId(dialCtx)
+				require.NoError(t, err)
+				assert.Equal(t, peerId, expectedPeerId)
+				return fx.mockMC(peerId), nil
+			})
+
+		p, err := fx.Dial(ctx, peerId)
+		require.NoError(t, err)
+		assert.NotNil(t, p)
+		require.Len(t, provider.requests, 1)
+		assert.Equal(t, peerId, provider.requests[0].peerId)
+	})
+
+	t.Run("preserves existing token", func(t *testing.T) {
+		provider := &testAdmissionTokenProvider{token: "provider-token"}
+		fx := newFixtureWithAdmissionTokenProvider(t, provider)
+		defer fx.finish(t)
+		fx.PreferQuic(false)
+
+		fx.nodeConf.EXPECT().PeerAddresses(peerId).Return(addrs, true)
+		fx.yamux.MockTransport.EXPECT().Dial(gomock.Any(), "127.0.0.1:1111").DoAndReturn(
+			func(dialCtx context.Context, addr string) (transport.MultiConn, error) {
+				assert.Equal(t, "caller-token", secureservice.CtxOutboundAdmissionToken(dialCtx))
+				return fx.mockMC(peerId), nil
+			})
+
+		p, err := fx.Dial(secureservice.CtxWithOutboundAdmissionToken(ctx, "caller-token"), peerId)
+		require.NoError(t, err)
+		assert.NotNil(t, p)
+		assert.Empty(t, provider.requests)
+	})
+
+	t.Run("provider error fails dial", func(t *testing.T) {
+		providerErr := fmt.Errorf("provider failed")
+		provider := &testAdmissionTokenProvider{err: providerErr}
+		fx := newFixtureWithAdmissionTokenProvider(t, provider)
+		defer fx.finish(t)
+		fx.PreferQuic(false)
+
+		fx.nodeConf.EXPECT().PeerAddresses(peerId).Return(addrs, true)
+
+		p, err := fx.Dial(ctx, peerId)
+		assert.Equal(t, providerErr, err)
+		assert.Nil(t, p)
+	})
+}
+
 func TestPeerService_Accept(t *testing.T) {
 	fx := newFixture(t)
 	defer fx.finish(t)
@@ -177,6 +239,10 @@ type fixture struct {
 }
 
 func newFixture(t *testing.T) *fixture {
+	return newFixtureWithAdmissionTokenProvider(t, nil)
+}
+
+func newFixtureWithAdmissionTokenProvider(t *testing.T, admissionTokenProvider *testAdmissionTokenProvider) *fixture {
 	ctrl := gomock.NewController(t)
 	fx := &fixture{
 		PeerService: New(),
@@ -195,12 +261,35 @@ func newFixture(t *testing.T) *fixture {
 	fx.nodeConf.EXPECT().Run(gomock.Any())
 	fx.nodeConf.EXPECT().Close(gomock.Any())
 
+	if admissionTokenProvider != nil {
+		fx.a.Register(admissionTokenProvider)
+	}
 	fx.a.Register(fx.PeerService).Register(fx.quic).Register(fx.yamux).Register(fx.nodeConf).Register(pool.New()).Register(rpctest.NewTestServer())
 
 	require.NoError(t, fx.a.Start(ctx))
 	return fx
 }
 
+type testAdmissionTokenProvider struct {
+	token    string
+	err      error
+	requests []admissionTokenRequest
+}
+
+type admissionTokenRequest struct {
+	ctx    context.Context
+	peerId string
+}
+
+func (p *testAdmissionTokenProvider) Init(a *app.App) error { return nil }
+
+func (p *testAdmissionTokenProvider) Name() string { return "test.admissionTokenProvider" }
+
+func (p *testAdmissionTokenProvider) OutboundAdmissionToken(ctx context.Context, peerId string) (string, error) {
+	p.requests = append(p.requests, admissionTokenRequest{ctx: ctx, peerId: peerId})
+	return p.token, p.err
+}
+
 func (fx *fixture) mockMC(peerId string) *mock_transport.MockMultiConn {
 	mc := mock_transport.NewMockMultiConn(fx.ctrl)
 	cctx := peer.CtxWithPeerId(ctx, peerId)

net/secureservice/admission.go

@@ -0,0 +1,80 @@
+package secureservice
+
+import "context"
+
+const (
+	DefaultAdmissionIdentityClaim = "anytype_identity"
+	DefaultAdmissionNetworkClaim  = "network_id"
+	DefaultAdmissionSubjectClaim  = "sub"
+	DefaultAdmissionClockSkewSec  = 60
+)
+
+type AdmissionConfig struct {
+	Enabled        bool           `yaml:"enabled"`
+	Required       bool           `yaml:"required"`
+	Issuer         string         `yaml:"issuer"`
+	Audience       string         `yaml:"audience"`
+	JWKSURL        string         `yaml:"jwksUrl"`
+	RequiredClaims map[string]any `yaml:"requiredClaims"`
+	IdentityClaim  string         `yaml:"identityClaim"`
+	NetworkClaim   string         `yaml:"networkClaim"`
+	SubjectClaim   string         `yaml:"subjectClaim"`
+	ClockSkewSec   int            `yaml:"clockSkewSec"`
+}
+
+func (c AdmissionConfig) WithDefaults() AdmissionConfig {
+	if c.IdentityClaim == "" {
+		c.IdentityClaim = DefaultAdmissionIdentityClaim
+	}
+	if c.NetworkClaim == "" {
+		c.NetworkClaim = DefaultAdmissionNetworkClaim
+	}
+	if c.SubjectClaim == "" {
+		c.SubjectClaim = DefaultAdmissionSubjectClaim
+	}
+	if c.ClockSkewSec == 0 {
+		c.ClockSkewSec = DefaultAdmissionClockSkewSec
+	}
+	return c
+}
+
+type AdmissionRequest struct {
+	Token         string
+	Identity      []byte
+	NetworkID     string
+	PeerID        string
+	ClientVersion string
+}
+
+type AdmissionClaims struct {
+	Subject   string
+	Issuer    string
+	Audience  []string
+	NetworkID string
+	Identity  []byte
+	Claims    map[string]any
+}
+
+type AdmissionDecision struct {
+	Allowed bool
+	Reason  string
+	Claims  AdmissionClaims
+}
+
+type AdmissionVerifier interface {
+	VerifyAdmission(ctx context.Context, req AdmissionRequest) (AdmissionDecision, error)
+}
+
+// AdmissionTokenProvider supplies bearer tokens for outbound handshakes.
+type AdmissionTokenProvider interface {
+	OutboundAdmissionToken(ctx context.Context, peerId string) (string, error)
+}
+
+type NoopAdmissionVerifier struct{}
+
+func (NoopAdmissionVerifier) VerifyAdmission(ctx context.Context, req AdmissionRequest) (AdmissionDecision, error) {
+	return AdmissionDecision{
+		Allowed: true,
+		Reason:  "admission disabled",
+	}, nil
+}

net/secureservice/admission_checker.go

@@ -0,0 +1,84 @@
+package secureservice
+
+import (
+	"context"
+
+	"github.com/anyproto/any-sync/net/secureservice/handshake"
+	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
+	"go.uber.org/zap"
+)
+
+func withAdmissionVerifier(ctx context.Context, checker handshake.CredentialChecker, verifier AdmissionVerifier, required bool, networkID string) handshake.CredentialChecker {
+	if verifier == nil {
+		return checker
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return admissionVerifierCredentialChecker{
+		CredentialChecker: checker,
+		ctx:               ctx,
+		verifier:          verifier,
+		required:          required,
+		networkID:         networkID,
+	}
+}
+
+type admissionVerifierCredentialChecker struct {
+	handshake.CredentialChecker
+	ctx       context.Context
+	verifier  AdmissionVerifier
+	required  bool
+	networkID string
+}
+
+func (a admissionVerifierCredentialChecker) CheckCredential(remotePeerId string, cred *handshakeproto.Credentials) (handshake.Result, error) {
+	res, err := a.CredentialChecker.CheckCredential(remotePeerId, cred)
+	if err != nil {
+		return handshake.Result{}, err
+	}
+	if res.AdmissionToken == "" {
+		if a.required {
+			log.InfoCtx(a.ctx, "admission rejected: missing token", a.logFields(remotePeerId, res,
+				zap.Bool("hasToken", false),
+				zap.Bool("allowed", false),
+			)...)
+			return handshake.Result{}, handshake.ErrInvalidCredentials
+		}
+		log.DebugCtx(a.ctx, "admission token absent", a.logFields(remotePeerId, res,
+			zap.Bool("hasToken", false),
+		)...)
+		return res, nil
+	}
+	decision, err := a.verifier.VerifyAdmission(a.ctx, AdmissionRequest{
+		Token:         res.AdmissionToken,
+		Identity:      res.Identity,
+		NetworkID:     a.networkID,
+		PeerID:        remotePeerId,
+		ClientVersion: res.ClientVersion,
+	})
+	if err != nil || !decision.Allowed {
+		log.InfoCtx(a.ctx, "admission rejected", a.logFields(remotePeerId, res,
+			zap.Bool("hasToken", true),
+			zap.Bool("allowed", false),
+			zap.Bool("verifierError", err != nil),
+		)...)
+		return handshake.Result{}, handshake.ErrInvalidCredentials
+	}
+	log.DebugCtx(a.ctx, "admission accepted", a.logFields(remotePeerId, res,
+		zap.Bool("hasToken", true),
+		zap.Bool("allowed", true),
+	)...)
+	return res, nil
+}
+
+func (a admissionVerifierCredentialChecker) logFields(remotePeerId string, res handshake.Result, fields ...zap.Field) []zap.Field {
+	base := []zap.Field{
+		zap.String("peerId", remotePeerId),
+		zap.String("networkId", a.networkID),
+		zap.String("clientVersion", res.ClientVersion),
+		zap.Uint32("protoVersion", res.ProtoVersion),
+		zap.Bool("required", a.required),
+	}
+	return append(base, fields...)
+}