feat(connectrpc): add ConnectRPC transport adapter

Add a connectrpc/ module (package connectrpcsec) that adapts the
transport-agnostic security core to the ConnectRPC framework, mirroring
the gRPC adapter.

ConnectRPC has a single connect.Interceptor interface covering unary and
streaming RPCs, so the adapter exposes two interceptors instead of the
four gRPC-style constructors:

  - NewAuthenticationInterceptor runs the Engine against the request
    headers and enriches the context; client-side calls pass through.
  - NewAuthorizationInterceptor enforces an AccessDecisionManager.

It also ships a Carrier over http.Header, an ErrorMapper translating
security sentinels to connect.Code (Unauthenticated / PermissionDenied /
InvalidArgument), and OTel spans connectrpcsec.Authenticate /
connectrpcsec.Authorize.

Module tests pass with -race at 100% coverage; golangci-lint is clean.

Author: euskadi31

connectrpc/authorize.go

@@ -0,0 +1,97 @@
+// Copyright 2026 Hyperscale. All rights reserved.
+// Use of this source code is governed by a MIT
+// license that can be found in the LICENSE file.
+
+package connectrpcsec
+
+import (
+	"context"
+
+	"connectrpc.com/connect"
+	"github.com/hyperscale-stack/security"
+	"go.opentelemetry.io/otel"
+)
+
+// AuthorizationInterceptor is a [connect.Interceptor] that enforces a
+// [security.AccessDecisionManager] against the request's
+// [security.Authentication].
+//
+// Install it AFTER [NewAuthenticationInterceptor] in the
+// connect.WithInterceptors(...) list so the context already carries an
+// authentication: the first interceptor of the list is the outermost, so
+// connect.WithInterceptors(authn, authz) runs authn (which enriches the
+// context) before authz.
+//
+// On grant the handler runs; on deny the configured [ErrorMapper] translates
+// the decision (typically connect.CodePermissionDenied).
+type AuthorizationInterceptor struct {
+	adm   security.AccessDecisionManager
+	attrs []security.Attribute
+	cfg   *config
+}
+
+// NewAuthorizationInterceptor builds a [connect.Interceptor] that enforces adm
+// against attrs for every inbound unary and streaming RPC.
+func NewAuthorizationInterceptor(
+	adm security.AccessDecisionManager,
+	attrs []security.Attribute,
+	opts ...Option,
+) *AuthorizationInterceptor {
+	return &AuthorizationInterceptor{adm: adm, attrs: attrs, cfg: buildConfig(opts...)}
+}
+
+// Compile-time check.
+var _ connect.Interceptor = (*AuthorizationInterceptor)(nil)
+
+// WrapUnary implements [connect.Interceptor]. Outbound client calls are passed
+// through untouched; inbound handler calls are authorized.
+func (i *AuthorizationInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
+		if req.Spec().IsClient {
+			return next(ctx, req) //nolint:wrapcheck // pass-through: the client error is the terminal value
+		}
+
+		if err := decide(ctx, i.adm, i.attrs); err != nil {
+			return nil, i.cfg.errorMapper.Map(ctx, err)
+		}
+
+		return next(ctx, req) //nolint:wrapcheck // the handler / connect error is the terminal wire value
+	}
+}
+
+// WrapStreamingHandler implements [connect.Interceptor]. It runs the access
+// decision before the handler runs.
+func (i *AuthorizationInterceptor) WrapStreamingHandler(
+	next connect.StreamingHandlerFunc,
+) connect.StreamingHandlerFunc {
+	return func(ctx context.Context, conn connect.StreamingHandlerConn) error {
+		if err := decide(ctx, i.adm, i.attrs); err != nil {
+			return i.cfg.errorMapper.Map(ctx, err)
+		}
+
+		return next(ctx, conn) //nolint:wrapcheck // the handler error is the terminal wire value
+	}
+}
+
+// WrapStreamingClient implements [connect.Interceptor] as a pass-through; the
+// access decision is server-side only.
+func (i *AuthorizationInterceptor) WrapStreamingClient(
+	next connect.StreamingClientFunc,
+) connect.StreamingClientFunc {
+	return next
+}
+
+// decide pulls the Authentication from ctx and runs the ADM, wrapping the call
+// in a "connectrpcsec.Authorize" span.
+func decide(ctx context.Context, adm security.AccessDecisionManager, attrs []security.Attribute) error {
+	ctx, span := otel.Tracer(tracerName).Start(ctx, "connectrpcsec.Authorize")
+	defer span.End()
+
+	auth, _ := security.FromContext(ctx)
+
+	if err := adm.Decide(ctx, auth, attrs); err != nil {
+		return err //nolint:wrapcheck // security.* sentinels pass through to the ErrorMapper
+	}
+
+	return nil
+}

connectrpc/authorize_test.go

@@ -0,0 +1,135 @@
+// Copyright 2026 Hyperscale. All rights reserved.
+// Use of this source code is governed by a MIT
+// license that can be found in the LICENSE file.
+
+package connectrpcsec_test
+
+import (
+	"context"
+	"testing"
+
+	"connectrpc.com/connect"
+	"github.com/hyperscale-stack/security"
+	connectrpcsec "github.com/hyperscale-stack/security/connectrpc"
+	"github.com/hyperscale-stack/security/voter"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// adminADM grants only when the principal holds the ADMIN role.
+func adminADM() security.AccessDecisionManager {
+	return security.NewAffirmativeDecisionManager(voter.HasRole("ADMIN"))
+}
+
+var adminAttrs = []security.Attribute{security.Role("ADMIN")}
+
+// chainUnary composes the authentication and authorization interceptors the
+// same way connect.WithInterceptors(authn, authz) would: authn is the
+// outermost, so it enriches the context before authz reads it.
+func chainUnary(
+	authn *connectrpcsec.AuthenticationInterceptor,
+	authz *connectrpcsec.AuthorizationInterceptor,
+	handler connect.UnaryFunc,
+) connect.UnaryFunc {
+	return authn.WrapUnary(authz.WrapUnary(handler))
+}
+
+func TestUnaryAuthorizeGrantsWhenRolePresent(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingUnary{}
+	wrapped := chainUnary(
+		connectrpcsec.NewAuthenticationInterceptor(newEngine("ROLE_ADMIN")),
+		connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs),
+		spy.fn,
+	)
+
+	resp, err := wrapped(context.Background(), unaryReq("letmein"))
+	require.NoError(t, err)
+	assert.NotNil(t, resp)
+	assert.True(t, spy.called)
+}
+
+func TestUnaryAuthorizeDeniesWhenRoleMissing(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingUnary{}
+	wrapped := chainUnary(
+		connectrpcsec.NewAuthenticationInterceptor(newEngine()),
+		connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs),
+		spy.fn,
+	)
+
+	_, err := wrapped(context.Background(), unaryReq("letmein"))
+	require.Error(t, err)
+	assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
+	assert.False(t, spy.called)
+}
+
+func TestUnaryAuthorizeDeniesAnonymous(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingUnary{}
+	// No authentication in the context: the voter denies the anonymous caller.
+	wrapped := connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs).WrapUnary(spy.fn)
+
+	_, err := wrapped(context.Background(), unaryReq("letmein"))
+	require.Error(t, err)
+	assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
+	assert.False(t, spy.called)
+}
+
+func TestUnaryAuthorizeSkipsClientCall(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingUnary{}
+	wrapped := connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs).WrapUnary(spy.fn)
+
+	_, err := wrapped(context.Background(), clientRequest{connect.NewRequest(&struct{}{})})
+	require.NoError(t, err)
+	assert.True(t, spy.called)
+}
+
+func TestStreamAuthorizeGrantsWhenRolePresent(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingStream{}
+	authn := connectrpcsec.NewAuthenticationInterceptor(newEngine("ROLE_ADMIN"))
+	authz := connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs)
+	wrapped := authn.WrapStreamingHandler(authz.WrapStreamingHandler(spy.fn))
+
+	err := wrapped(context.Background(), newStreamConn(bearerHeader("letmein")))
+	require.NoError(t, err)
+	assert.True(t, spy.called)
+}
+
+func TestStreamAuthorizeDeniesWhenRoleMissing(t *testing.T) {
+	t.Parallel()
+
+	spy := &recordingStream{}
+	authn := connectrpcsec.NewAuthenticationInterceptor(newEngine())
+	authz := connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs)
+	wrapped := authn.WrapStreamingHandler(authz.WrapStreamingHandler(spy.fn))
+
+	err := wrapped(context.Background(), newStreamConn(bearerHeader("letmein")))
+	require.Error(t, err)
+	assert.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err))
+	assert.False(t, spy.called)
+}
+
+func TestStreamAuthorizeIsPassThroughForClient(t *testing.T) {
+	t.Parallel()
+
+	called := false
+
+	next := func(_ context.Context, _ connect.Spec) connect.StreamingClientConn {
+		called = true
+
+		return nil
+	}
+
+	wrapped := connectrpcsec.NewAuthorizationInterceptor(adminADM(), adminAttrs).WrapStreamingClient(next)
+	_ = wrapped(context.Background(), connect.Spec{})
+
+	assert.True(t, called)
+}

connectrpc/carrier.go

@@ -0,0 +1,69 @@
+// Copyright 2026 Hyperscale. All rights reserved.
+// Use of this source code is governed by a MIT
+// license that can be found in the LICENSE file.
+
+package connectrpcsec
+
+import (
+	"net/http"
+
+	"github.com/hyperscale-stack/security"
+)
+
+// Carrier adapts a ConnectRPC request to [security.Carrier].
+//
+// Reads consult the request header. ConnectRPC speaks net/http, so the keys
+// follow http.Header semantics (case-insensitive, canonicalised via
+// textproto.CanonicalMIMEHeaderKey); the conventional "Authorization"
+// spelling works directly with no manual case folding.
+//
+// Writes accumulate in a private staged header that the interceptor flushes
+// onto the live response header once one is available — immediately for a
+// streaming handler (conn.ResponseHeader() is live) and after the handler
+// runs for a unary call (the response Header()). This lets an ErrorMapper or
+// an extractor attach, e.g., a diagnostic header alongside the response.
+//
+// Carrier is NOT safe for concurrent use; one instance per RPC.
+type Carrier struct {
+	in  http.Header
+	out http.Header
+}
+
+// NewCarrier builds a Carrier from a request header. When h is nil (a unit
+// test, a non-Connect caller) the read side is simply empty.
+func NewCarrier(h http.Header) *Carrier {
+	if h == nil {
+		h = http.Header{}
+	}
+
+	return &Carrier{in: h, out: http.Header{}}
+}
+
+// Get implements [security.Carrier]. Returns the first value for key.
+func (c *Carrier) Get(key string) string {
+	return c.in.Get(key)
+}
+
+// Values implements [security.Carrier].
+func (c *Carrier) Values(key string) []string {
+	return c.in.Values(key)
+}
+
+// Set implements [security.Carrier]. The value is staged in the response
+// header; the interceptor flushes it onto the live response.
+func (c *Carrier) Set(key, value string) {
+	c.out.Set(key, value)
+}
+
+// Add implements [security.Carrier].
+func (c *Carrier) Add(key, value string) {
+	c.out.Add(key, value)
+}
+
+// ResponseHeader returns the staged response header. The interceptor calls it
+// after the engine / handler run and, when non-empty, copies it onto the live
+// response header.
+func (c *Carrier) ResponseHeader() http.Header { return c.out }
+
+// Compile-time check.
+var _ security.Carrier = (*Carrier)(nil)

connectrpc/carrier_test.go

@@ -0,0 +1,56 @@
+// Copyright 2026 Hyperscale. All rights reserved.
+// Use of this source code is governed by a MIT
+// license that can be found in the LICENSE file.
+
+package connectrpcsec_test
+
+import (
+	"net/http"
+	"testing"
+
+	connectrpcsec "github.com/hyperscale-stack/security/connectrpc"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCarrierReads(t *testing.T) {
+	t.Parallel()
+
+	hdr := http.Header{"Authorization": {"Bearer one", "Bearer two"}}
+	carrier := connectrpcsec.NewCarrier(hdr)
+
+	// Get returns the first value; lookups are case-insensitive.
+	assert.Equal(t, "Bearer one", carrier.Get("Authorization"))
+	assert.Equal(t, "Bearer one", carrier.Get("authorization"))
+
+	// Values returns every value.
+	assert.Equal(t, []string{"Bearer one", "Bearer two"}, carrier.Values("authorization"))
+
+	// Absent keys yield the zero values.
+	assert.Empty(t, carrier.Get("X-Absent"))
+	assert.Nil(t, carrier.Values("X-Absent"))
+}
+
+func TestCarrierWritesResponseHeader(t *testing.T) {
+	t.Parallel()
+
+	carrier := connectrpcsec.NewCarrier(http.Header{})
+
+	carrier.Set("X-Trace", "first")
+	carrier.Set("X-Trace", "second") // Set replaces.
+	carrier.Add("X-Trace", "third")  // Add appends.
+
+	assert.Equal(t, []string{"second", "third"}, carrier.ResponseHeader().Values("X-Trace"))
+}
+
+func TestCarrierNilHeader(t *testing.T) {
+	t.Parallel()
+
+	carrier := connectrpcsec.NewCarrier(nil)
+
+	assert.Empty(t, carrier.Get("Authorization"))
+	assert.Nil(t, carrier.Values("Authorization"))
+
+	// Writes still work against the staged header.
+	carrier.Set("X-Trace", "value")
+	assert.Equal(t, "value", carrier.ResponseHeader().Get("X-Trace"))
+}

connectrpc/doc.go

@@ -0,0 +1,24 @@
+// Copyright 2026 Hyperscale. All rights reserved.
+// Use of this source code is governed by a MIT
+// license that can be found in the LICENSE file.
+
+// Package connectrpcsec is the ConnectRPC transport adapter for the security
+// core.
+//
+// It exposes connect.Interceptor values that hand the request headers (the
+// Carrier) to the core Engine and map security errors to the appropriate
+// Connect error codes (connect.CodeUnauthenticated, connect.CodePermissionDenied,
+// …).
+//
+// ConnectRPC has a single Interceptor interface covering both unary and
+// streaming RPCs, installed once via connect.WithInterceptors. The adapter
+// therefore exposes two interceptors instead of the four gRPC-style
+// constructors: NewAuthenticationInterceptor authenticates every inbound RPC
+// and NewAuthorizationInterceptor enforces an access decision manager.
+//
+// Allowed dependencies:
+//   - github.com/hyperscale-stack/security (core)
+//   - connectrpc.com/connect
+//   - go.opentelemetry.io/otel
+//   - stdlib only
+package connectrpcsec