Skip to content

rpc: add a MessageConn transport for RPC over any message pipe #979

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
evanphx wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

rpc: add a MessageConn transport for RPC over any message pipe #979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 6 additions & 10 deletions pkg/cloudauth/rpc_authenticator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"context"
"fmt"
"log/slog"
"net/http"
"strings"

"miren.dev/runtime/pkg/auth"
Expand Down Expand Up @@ -110,10 +109,10 @@ func NewRPCAuthenticator(ctx context.Context, config Config) (*RPCAuthenticator,
// Authenticate implements rpc.Authenticator.
// It tries JWT authentication first, then falls back to TLS client certificate.
// Authorization (RBAC) is handled separately via the Authorize method.
func (a *RPCAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*rpc.Identity, error) {
func (a *RPCAuthenticator) Authenticate(ctx context.Context, creds *rpc.Credentials) (*rpc.Identity, error) {
// Try JWT authentication first if Authorization header is present
if authHeader := r.Header.Get("Authorization"); authHeader != "" {
identity, err := a.authenticateJWT(ctx, authHeader)
if creds.Authorization != "" {
identity, err := a.authenticateJWT(ctx, creds.Authorization)
if err != nil {
return nil, err
}
Expand All @@ -123,12 +122,9 @@ func (a *RPCAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*
// Invalid JWT format, fall through to cert check
}

// Fall back to TLS client certificate. Only trust a cert that the TLS
// layer verified against the cluster CA (VerifiedChains non-empty); a
// presented-but-unverified cert must never yield a cert identity, which
// grants RBAC-bypassing privileges in Authorize.
if r.TLS != nil && len(r.TLS.VerifiedChains) > 0 && len(r.TLS.PeerCertificates) > 0 {
cert := r.TLS.PeerCertificates[0]
// Fall back to a TLS client certificate the TLS layer verified against the
// cluster CA.
if cert := creds.VerifiedPeerCertificate(); cert != nil {
return &rpc.Identity{
Subject: cert.Subject.CommonName,
Method: rpc.AuthMethodCert,
Expand Down
4 changes: 2 additions & 2 deletions pkg/cloudauth/rpc_authenticator_cert_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ func TestRPCAuthenticator_CertRequiresVerifiedChain(t *testing.T) {
req := newReq(t, &tls.ConnectionState{
PeerCertificates: []*x509.Certificate{cert},
})
id, err := auth.Authenticate(t.Context(), req)
id, err := auth.Authenticate(t.Context(), rpc.CredentialsFromRequest(req))
require.NoError(t, err)
require.Nil(t, id, "unverified client cert must not yield an identity")
})
Expand All @@ -51,7 +51,7 @@ func TestRPCAuthenticator_CertRequiresVerifiedChain(t *testing.T) {
PeerCertificates: []*x509.Certificate{cert},
VerifiedChains: [][]*x509.Certificate{{cert}},
})
id, err := auth.Authenticate(t.Context(), req)
id, err := auth.Authenticate(t.Context(), rpc.CredentialsFromRequest(req))
require.NoError(t, err)
require.NotNil(t, id)
require.Equal(t, "test-client", id.Subject)
Expand Down
21 changes: 9 additions & 12 deletions pkg/oidcauth/authenticator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"encoding/json"
"fmt"
"log/slog"
"net/http"
"strings"
"sync/atomic"

Expand Down Expand Up @@ -41,15 +40,12 @@ func (a *OIDCAuthenticator) SetEAC(eac *entityserver_v1alpha.EntityAccessClient)

// Authenticate checks if the request carries a valid OIDC bearer token that
// matches a configured oidc_binding entity.
func (a *OIDCAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*rpc.Identity, error) {
authHeader := r.Header.Get("Authorization")
if !strings.HasPrefix(authHeader, "Bearer ") {
func (a *OIDCAuthenticator) Authenticate(ctx context.Context, creds *rpc.Credentials) (*rpc.Identity, error) {
tokenString := creds.BearerToken()
if tokenString == "" {
return nil, nil
}

tokenString := strings.TrimPrefix(authHeader, "Bearer ")
tokenString = strings.TrimSpace(tokenString)

eac := a.eac.Load()
if eac == nil {
return nil, nil // Entity access client not yet initialized
Expand All @@ -72,11 +68,12 @@ func (a *OIDCAuthenticator) Authenticate(ctx context.Context, r *http.Request) (
return nil, nil // No bindings for this issuer
}

// Determine the audience to validate against.
// Use the hostname from the request.
audience := r.Host
if audience == "" && r.TLS != nil {
audience = r.TLS.ServerName
// Determine the audience to validate against: the address the caller
// addressed. A message transport is not addressed by name and so carries
// neither, in which case there is nothing to validate against.
audience := creds.Host
if audience == "" && creds.TLS != nil {
audience = creds.TLS.ServerName
}
if audience == "" {
return nil, nil
Expand Down
24 changes: 12 additions & 12 deletions pkg/oidcauth/authenticator_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ func TestOIDCAuthenticator_NoEAC(t *testing.T) {
req := httptest.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand All @@ -128,7 +128,7 @@ func TestOIDCAuthenticator_NoBearerToken(t *testing.T) {
auth := NewOIDCAuthenticator(testutils.TestLogger(t))

req := httptest.NewRequest("GET", "/", nil)
identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand All @@ -147,7 +147,7 @@ func TestOIDCAuthenticator_NonJWTBearer(t *testing.T) {
req := httptest.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Bearer not-a-jwt-token")

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -177,7 +177,7 @@ func TestOIDCAuthenticator_NoMatchingBindings(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -238,7 +238,7 @@ func TestOIDCAuthenticator_MatchingBinding(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -305,7 +305,7 @@ func TestOIDCAuthenticator_SubjectMismatch(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -364,7 +364,7 @@ func TestOIDCAuthenticator_ClaimConditionMismatch(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -418,7 +418,7 @@ func TestOIDCAuthenticator_ExpiredToken(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -496,7 +496,7 @@ func TestOIDCAuthenticator_MultipleBindings_FirstMatch(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -562,7 +562,7 @@ func TestOIDCAuthenticator_AudienceFromHost(t *testing.T) {
req.Host = "my-cluster.example.com"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -618,7 +618,7 @@ func TestOIDCAuthenticator_AudienceFromTLS(t *testing.T) {
req.TLS = &tls.ConnectionState{ServerName: "tls-server.example.com"}
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -673,7 +673,7 @@ func TestOIDCAuthenticator_WrongAudience(t *testing.T) {
req.Host = "test-host"
req.Header.Set("Authorization", "Bearer "+token)

identity, err := auth.Authenticate(ctx, req)
identity, err := auth.Authenticate(ctx, rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down
12 changes: 6 additions & 6 deletions pkg/oidcauth/composite.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package oidcauth
import (
"context"
"fmt"
"net/http"

"miren.dev/runtime/pkg/rpc"
)
Expand Down Expand Up @@ -55,9 +54,9 @@ func NewCompositeAuthenticator(primary rpc.Authenticator, oidc *OIDCAuthenticato
return &CompositeAuthenticator{primary: primary, oidc: oidc}
}

func (c *CompositeAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*rpc.Identity, error) {
func (c *CompositeAuthenticator) Authenticate(ctx context.Context, creds *rpc.Credentials) (*rpc.Identity, error) {
// Try primary authenticator first
identity, err := c.primary.Authenticate(ctx, r)
identity, err := c.primary.Authenticate(ctx, creds)
if identity != nil {
return identity, nil
}
Expand All @@ -68,7 +67,7 @@ func (c *CompositeAuthenticator) Authenticate(ctx context.Context, r *http.Reque
var oidcIdentity *rpc.Identity
var oidcErr error
if c.oidc != nil {
oidcIdentity, oidcErr = c.oidc.Authenticate(ctx, r)
oidcIdentity, oidcErr = c.oidc.Authenticate(ctx, creds)
if oidcIdentity != nil {
return oidcIdentity, nil
}
Expand Down Expand Up @@ -102,8 +101,9 @@ func (c *CompositeAuthorizer) Authorize(ctx context.Context, identity *rpc.Ident
// OIDC callers are restricted to the oidc-deploy role
return authorizeOIDC(resource, action)

case rpc.AuthMethodJWT, rpc.AuthMethodAnonymous, rpc.AuthMethodToken:
// Delegate to primary (cloud RBAC).
case rpc.AuthMethodJWT, rpc.AuthMethodAnonymous, rpc.AuthMethodToken, rpc.AuthMethodSigned:
// Delegate to primary (cloud RBAC). A signed identity names the key a
// capability was issued to, which carries no privilege of its own.
fallthrough
default:
if c.primary != nil {
Expand Down
11 changes: 5 additions & 6 deletions pkg/oidcauth/composite_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package oidcauth
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"testing"

Expand All @@ -17,7 +16,7 @@ type mockAuthenticator struct {
err error
}

func (m *mockAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*rpc.Identity, error) {
func (m *mockAuthenticator) Authenticate(ctx context.Context, creds *rpc.Credentials) (*rpc.Identity, error) {
return m.identity, m.err
}

Expand All @@ -41,7 +40,7 @@ func TestCompositeAuthenticator_PrimarySucceeds(t *testing.T) {
comp := NewCompositeAuthenticator(primary, oidcAuth)

req := httptest.NewRequest("GET", "/", nil)
identity, err := comp.Authenticate(context.Background(), req)
identity, err := comp.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand All @@ -63,7 +62,7 @@ func TestCompositeAuthenticator_PrimaryErrorOIDCNil(t *testing.T) {
comp := NewCompositeAuthenticator(primary, oidcAuth)

req := httptest.NewRequest("GET", "/", nil)
_, err := comp.Authenticate(context.Background(), req)
_, err := comp.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err == nil {
t.Fatal("expected error to propagate from primary")
}
Expand All @@ -85,7 +84,7 @@ func TestCompositeAuthenticator_PrimaryErrorOIDCSucceeds(t *testing.T) {
comp := &CompositeAuthenticator{primary: primary, oidc: oidcStub}

req := httptest.NewRequest("GET", "/", nil)
identity, err := comp.Authenticate(context.Background(), req)
identity, err := comp.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("expected no error when OIDC succeeds, got: %v", err)
}
Expand All @@ -110,7 +109,7 @@ func TestCompositeAuthenticator_FallbackToOIDC(t *testing.T) {
req := httptest.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Bearer some-token")

identity, err := comp.Authenticate(context.Background(), req)
identity, err := comp.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down
9 changes: 5 additions & 4 deletions pkg/outboard/outboard_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"miren.dev/runtime/pkg/rpc"
)

func TestConfigRoundTrip(t *testing.T) {
Expand Down Expand Up @@ -64,7 +65,7 @@ func TestTokenAuthenticatorValid(t *testing.T) {
req, _ := http.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Bearer secret-token")

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
assert.NoError(t, err)
assert.NotNil(t, identity)
assert.Equal(t, "outboard", identity.Subject)
Expand All @@ -76,7 +77,7 @@ func TestTokenAuthenticatorInvalid(t *testing.T) {
req, _ := http.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Bearer wrong-token")

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
assert.NoError(t, err)
assert.Nil(t, identity, "invalid token should return nil identity")
}
Expand All @@ -86,7 +87,7 @@ func TestTokenAuthenticatorMissing(t *testing.T) {

req, _ := http.NewRequest("GET", "/", nil)

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
assert.NoError(t, err)
assert.Nil(t, identity, "missing auth header should return nil identity")
}
Expand All @@ -97,7 +98,7 @@ func TestTokenAuthenticatorBadScheme(t *testing.T) {
req, _ := http.NewRequest("GET", "/", nil)
req.Header.Set("Authorization", "Basic dXNlcjpwYXNz")

identity, err := auth.Authenticate(context.Background(), req)
identity, err := auth.Authenticate(context.Background(), rpc.CredentialsFromRequest(req))
assert.NoError(t, err)
assert.Nil(t, identity, "non-bearer auth should return nil identity")
}
Expand Down
5 changes: 2 additions & 3 deletions pkg/outboard/token_auth.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package outboard
import (
"context"
"crypto/subtle"
"net/http"
"strings"

"miren.dev/runtime/pkg/rpc"
Expand All @@ -20,8 +19,8 @@ func NewTokenAuthenticator(token string) *TokenAuthenticator {
return &TokenAuthenticator{token: token}
}

func (t *TokenAuthenticator) Authenticate(ctx context.Context, r *http.Request) (*rpc.Identity, error) {
auth := r.Header.Get("Authorization")
func (t *TokenAuthenticator) Authenticate(ctx context.Context, creds *rpc.Credentials) (*rpc.Identity, error) {
auth := creds.Authorization
if auth == "" {
return nil, nil // No credentials
}
Expand Down
Loading
Loading