CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into plugin-hostcatalogs

Browse Source
This commit is contained in:
Jeff Mitchell committed 2021-10-27 21:57:30 -04:00
2 parents 86f2535aaf 1782f6a4e4
commit b1a72bd445
69 files changed
+1213 -277
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/servers/controller/auth/auth.go
+2 -1
View File
@@ -18,6 +18,7 @@ import (
"github.com/hashicorp/boundary/internal/observability/event"
"github.com/hashicorp/boundary/internal/perms"
"github.com/hashicorp/boundary/internal/requests"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/boundary/internal/servers/controller/common"
"github.com/hashicorp/boundary/internal/servers/controller/handlers"
"github.com/hashicorp/boundary/internal/types/action"
@@ -391,7 +392,7 @@ func (v *verifier) decryptToken(ctx context.Context) {
v.requestInfo.TokenFormat = uint32(AuthTokenTypeUnknown)
return
}
if err := repo.AddRecoveryNonce(v.ctx, info.Nonce); err != nil {
if err := repo.AddNonce(v.ctx, info.Nonce, servers.NoncePurposeRecovery); err != nil {
event.WriteError(ctx, op, err, event.WithInfoMsg("decrypt recovery token: error adding nonce to database (possible replay attack)"))
v.requestInfo.TokenFormat = uint32(AuthTokenTypeUnknown)
return
internal/servers/controller/controller.go
+8 -7
View File
@@ -32,7 +32,6 @@ import (
external_host_plugins "github.com/hashicorp/boundary/sdk/plugins/host"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/go-secure-stdlib/mlock"
"github.com/patrickmn/go-cache"
ua "go.uber.org/atomic"
"google.golang.org/grpc"
)
@@ -45,10 +44,10 @@ type Controller struct {
baseCancel context.CancelFunc
started *ua.Bool
tickerWg sync.WaitGroup
schedulerWg sync.WaitGroup
tickerWg *sync.WaitGroup
schedulerWg *sync.WaitGroup
workerAuthCache *cache.Cache
workerAuthCache *sync.Map
// Used for testing and tracking worker health
workerStatusUpdateTimes *sync.Map
@@ -82,6 +81,9 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
conf: conf,
logger: conf.Logger.Named("controller"),
started: ua.NewBool(false),
tickerWg: new(sync.WaitGroup),
schedulerWg: new(sync.WaitGroup),
workerAuthCache: new(sync.Map),
workerStatusUpdateTimes: new(sync.Map),
}
@@ -209,7 +211,6 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
c.SessionRepoFn = func() (*session.Repository, error) {
return session.NewRepository(dbase, dbase, c.kms)
}
c.workerAuthCache = cache.New(0, 0)
return c, nil
}
@@ -224,7 +225,7 @@ func (c *Controller) Start() error {
if err := c.registerJobs(); err != nil {
return fmt.Errorf("error registering jobs: %w", err)
}
if err := c.scheduler.Start(c.baseContext, &c.schedulerWg); err != nil {
if err := c.scheduler.Start(c.baseContext, c.schedulerWg); err != nil {
return fmt.Errorf("error starting scheduler: %w", err)
}
if err := c.startListeners(c.baseContext); err != nil {
@@ -238,7 +239,7 @@ func (c *Controller) Start() error {
}()
go func() {
defer c.tickerWg.Done()
c.startRecoveryNonceCleanupTicking(c.baseContext)
c.startNonceCleanupTicking(c.baseContext)
}()
go func() {
defer c.tickerWg.Done()
internal/servers/controller/gateway.go
+19 -3
View File
@@ -11,6 +11,7 @@ import (
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/observability/event"
"github.com/hashicorp/boundary/internal/servers/controller/common"
"github.com/hashicorp/boundary/internal/servers/controller/handlers"
"google.golang.org/grpc"
@@ -49,19 +50,34 @@ func newGatewayMux() *runtime.ServeMux {
)
}
func newGatewayServer(ctx context.Context, iamRepoFn common.IamRepoFactory, authTokenRepoFn common.AuthTokenRepoFactory, serversRepoFn common.ServersRepoFactory, kms *kms.Kms) (*grpc.Server, string, error) {
func newGatewayServer(
ctx context.Context,
iamRepoFn common.IamRepoFactory,
authTokenRepoFn common.AuthTokenRepoFactory,
serversRepoFn common.ServersRepoFactory,
kms *kms.Kms,
eventer *event.Eventer,
) (*grpc.Server, string, error) {
const op = "controller.newGatewayServer"
ticket, err := db.NewPrivateId("gwticket")
if err != nil {
return nil, "", errors.Wrap(ctx, err, op, errors.WithMsg("unable to generate gateway ticket"))
}
requestCtxInterceptor, err := requestCtxInterceptor(ctx, iamRepoFn, authTokenRepoFn, serversRepoFn, kms, ticket)
requestCtxInterceptor, err := requestCtxInterceptor(ctx, iamRepoFn, authTokenRepoFn, serversRepoFn, kms, ticket, eventer)
if err != nil {
return nil, "", err
}
return grpc.NewServer(
grpc.MaxRecvMsgSize(math.MaxInt32),
grpc.MaxSendMsgSize(math.MaxInt32),
grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(requestCtxInterceptor, errorInterceptor(ctx), statusCodeInterceptor(ctx))),
grpc.UnaryInterceptor(
grpc_middleware.ChainUnaryServer(
requestCtxInterceptor, // populated requestInfo from headers into the request ctx
auditRequestInterceptor(ctx), // before we get started, audit the request
errorInterceptor(ctx), // convert domain and api errors into headers for the http proxy
statusCodeInterceptor(ctx), // convert grpc codes into http status codes for the http proxy (can modify the resp)
auditResponseInterceptor(ctx), // as we finish, audit the response
),
),
), ticket, nil
}
internal/servers/controller/handler.go
+12
View File
@@ -288,6 +288,16 @@ func wrapHandlerWithCommonFuncs(h http.Handler, c *Controller, props HandlerProp
requestInfo.PublicId, requestInfo.EncryptedToken, requestInfo.TokenFormat = auth.GetTokenFromRequest(ctx, c.kms, r)
if info, ok := event.RequestInfoFromContext(ctx); ok {
// piggyback some eventing fields with the auth info proto message
requestInfo.EventId = info.EventId
requestInfo.TraceId = info.Id
} else {
w.WriteHeader(http.StatusInternalServerError)
event.WriteError(ctx, op, errors.New("unable to read event request info from context"))
return
}
// Serialize the request info to send it across the wire to the
// grpc-gateway via an http header
requestInfo.Ticket = c.gatewayTicket // allows the grpc-gateway to verify the request info came from it's in-memory companion http proxy
@@ -302,6 +312,8 @@ func wrapHandlerWithCommonFuncs(h http.Handler, c *Controller, props HandlerProp
// See: https://pkg.go.dev/github.com/grpc-ecosystem/grpc-gateway/runtime#DefaultHeaderMatcher
r.Header.Set("Grpc-Metadata-"+requestInfoMdKey, base58.FastBase58Encoding(marshalledRequestInfo))
// Set the context back on the request
r = r.WithContext(ctx)
h.ServeHTTP(w, r)
})
}
internal/servers/controller/handlers/credentialstores/credentialstore_service_test.go
+9
View File
@@ -1,6 +1,7 @@
package credentialstores
import (
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
@@ -52,6 +53,8 @@ func TestList(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
rw := db.New(conn)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
iamRepo := iam.TestRepo(t, conn, wrapper)
iamRepoFn := func() (*iam.Repository, error) {
@@ -156,6 +159,8 @@ func TestCreate(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
rw := db.New(conn)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
iamRepo := iam.TestRepo(t, conn, wrapper)
iamRepoFn := func() (*iam.Repository, error) {
@@ -655,6 +660,8 @@ func TestDelete(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
rw := db.New(conn)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
iamRepo := iam.TestRepo(t, conn, wrapper)
iamRepoFn := func() (*iam.Repository, error) {
@@ -714,6 +721,8 @@ func TestUpdate(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
rw := db.New(conn)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
iamRepo := iam.TestRepo(t, conn, wrapper)
iamRepoFn := func() (*iam.Repository, error) {
internal/servers/controller/handlers/outgoing_response_filter.go
-7
View File
@@ -11,7 +11,6 @@ import (
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
"github.com/hashicorp/boundary/internal/observability/event"
authtokenpb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/authtokens"
"google.golang.org/grpc"
@@ -65,12 +64,6 @@ func OutgoingResponseFilter(ctx context.Context, w http.ResponseWriter, m proto.
}
}
// add the outgoing resp proto. See: common.flushGatedEvents is where the
// event.Response.StatusCode is added.
if err := event.WriteAudit(ctx, op, event.WithResponse(&event.Response{Details: m})); err != nil {
return fmt.Errorf("%s: unable to write audit event: %w", op, err)
}
switch m := m.(type) {
case *pbs.AuthenticateResponse:
if m.GetAttributes() == nil || m.GetAttributes().GetFields() == nil {
internal/servers/controller/handlers/targets/target_service_test.go
+10 -2
View File
@@ -581,11 +581,13 @@ func TestUpdate(t *testing.T) {
tar, err := tcp.New(proj.GetPublicId(), target.WithName("default"), target.WithDescription("default"))
tar.DefaultPort = 2
require.NoError(t, err)
gtar, _, _, err := repo.CreateTarget(context.Background(), tar, target.WithHostSources([]string{hs[0].GetPublicId(), hs[1].GetPublicId()}))
gtar, _, _, err := repo.CreateTarget(context.Background(), tar)
require.NoError(t, err)
gtar, _, _, err = repo.AddTargetHostSources(context.Background(), gtar.GetPublicId(), gtar.GetVersion(), []string{hs[0].GetPublicId(), hs[1].GetPublicId()})
require.NoError(t, err)
tar = gtar.(*tcp.Target)
var version uint32 = 1
var version uint32 = gtar.GetVersion()
resetTarget := func() {
version++
@@ -2567,6 +2569,9 @@ func TestAuthorizeSession(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
repoFn := func() (*target.Repository, error) {
return target.NewRepository(rw, rw, kms)
}
@@ -2799,6 +2804,9 @@ func TestAuthorizeSession_Errors(t *testing.T) {
kms := kms.TestKms(t, conn, wrapper)
sche := scheduler.TestScheduler(t, conn, wrapper)
err := vault.RegisterJobs(context.Background(), sche, rw, rw, kms)
require.NoError(t, err)
repoFn := func() (*target.Repository, error) {
return target.NewRepository(rw, rw, kms)
}
internal/servers/controller/handlers/verifiers.go
+1 -5
View File
@@ -195,9 +195,5 @@ func ValidNameDescription(in string) bool {
return !unicode.IsPrint(c)
})
if idx != -1 {
return false
}
return true
return idx == -1
}
internal/servers/controller/intercepting_listener.go
+1 -1
View File
@@ -70,7 +70,7 @@ func (m *interceptingListener) Accept() (net.Conn, error) {
}
return nil, fmt.Errorf("error reading nonce from worker, expected %d bytes, got %d", 20, read)
}
workerInfoRaw, found := m.c.workerAuthCache.Get(string(nonce))
workerInfoRaw, found := m.c.workerAuthCache.Load(string(nonce))
if !found {
if err := conn.Close(); err != nil {
event.WriteError(ctx, op, err, event.WithInfoMsg("error closing worker connection"))
internal/servers/controller/interceptor.go
+63
View File
@@ -11,6 +11,7 @@ import (
authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
pberrors "github.com/hashicorp/boundary/internal/gen/errors"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/observability/event"
"github.com/hashicorp/boundary/internal/requests"
"github.com/hashicorp/boundary/internal/servers/controller/auth"
"github.com/hashicorp/boundary/internal/servers/controller/common"
@@ -48,6 +49,7 @@ func requestCtxInterceptor(
serversRepoFn common.ServersRepoFactory,
kms *kms.Kms,
ticket string,
eventer *event.Eventer,
) (grpc.UnaryServerInterceptor, error) {
const op = "controller.requestCtxInterceptor"
if iamRepoFn == nil {
@@ -65,6 +67,9 @@ func requestCtxInterceptor(
if ticket == "" {
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing ticket")
}
if eventer == nil {
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing eventer")
}
// Authorization unary interceptor function to handle authorize per RPC call
return func(interceptorCtx context.Context,
req interface{},
@@ -112,6 +117,23 @@ func requestCtxInterceptor(
Method: requestInfo.Method,
})
// This event request info is required by downstream handlers
info := &event.RequestInfo{
EventId: requestInfo.EventId,
Id: requestInfo.TraceId,
PublicId: requestInfo.PublicId,
Method: requestInfo.Method,
Path: requestInfo.Path,
}
interceptorCtx, err = event.NewRequestInfoContext(interceptorCtx, info)
if err != nil {
return nil, errors.Wrap(interceptorCtx, err, op, errors.WithCode(errors.Internal), errors.WithMsg("unable to create context with request info"))
}
interceptorCtx, err = event.NewEventerContext(interceptorCtx, eventer)
if err != nil {
return nil, errors.Wrap(interceptorCtx, err, op, errors.WithCode(errors.Internal), errors.WithMsg("unable to create context with eventer"))
}
// Calls the handler
h, err := handler(interceptorCtx, req)
@@ -205,3 +227,44 @@ func isNil(i interface{}) bool {
}
return false
}
func auditRequestInterceptor(
_ context.Context,
) grpc.UnaryServerInterceptor {
const op = "controller.auditRequestInterceptor"
return func(interceptorCtx context.Context,
req interface{},
_ *grpc.UnaryServerInfo,
handler grpc.UnaryHandler) (interface{}, error) {
if msg, ok := req.(proto.Message); ok {
if err := event.WriteAudit(interceptorCtx, op, event.WithRequest(&event.Request{Details: msg})); err != nil {
return req, status.Errorf(codes.Internal, "unable to write request msg audit: %s", err)
}
}
return handler(interceptorCtx, req)
}
}
func auditResponseInterceptor(
_ context.Context,
) grpc.UnaryServerInterceptor {
const op = "controller.auditRequestInterceptor"
return func(interceptorCtx context.Context,
req interface{},
_ *grpc.UnaryServerInfo,
handler grpc.UnaryHandler) (interface{}, error) {
// call the handler...
resp, err := handler(interceptorCtx, req)
if msg, ok := resp.(proto.Message); ok {
if err := event.WriteAudit(interceptorCtx, op, event.WithResponse(&event.Response{Details: msg})); err != nil {
return req, status.Errorf(codes.Internal, "unable to write response msg audit: %s", err)
}
}
return resp, err
}
}
internal/servers/controller/interceptor_test.go
+42 -3
View File
@@ -6,6 +6,7 @@ import (
"net/http"
"net/http/httptest"
"strconv"
"sync"
"testing"
"github.com/hashicorp/boundary/internal/authtoken"
@@ -17,10 +18,12 @@ import (
"github.com/hashicorp/boundary/internal/gen/testing/interceptor"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/observability/event"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/boundary/internal/servers/controller/auth"
"github.com/hashicorp/boundary/internal/servers/controller/common"
"github.com/hashicorp/boundary/internal/servers/controller/handlers"
"github.com/hashicorp/go-hclog"
"github.com/mr-tron/base58"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -59,8 +62,10 @@ func Test_requestCtxInterceptor(t *testing.T) {
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", tokValue))
// Add values for authn/authz checking
requestInfo := authpb.RequestInfo{
Path: req.URL.Path,
Method: req.Method,
Path: req.URL.Path,
Method: req.Method,
EventId: "test-event-id",
TraceId: "test-trace-id",
}
requestInfo.PublicId, requestInfo.EncryptedToken, requestInfo.TokenFormat = auth.GetTokenFromRequest(context.TODO(), kmsCache, req)
requestInfo.Ticket = gwTicket // allows the grpc-gateway to verify the request info came from it's in-memory companion http proxy
@@ -82,6 +87,15 @@ func Test_requestCtxInterceptor(t *testing.T) {
return ctx, nil
}
c := event.TestEventerConfig(t, "Test_requestCtxInterceptor", event.TestWithAuditSink(t), event.TestWithObservationSink(t))
testLock := &sync.Mutex{}
testLogger := hclog.New(&hclog.LoggerOptions{
Mutex: testLock,
Name: "test",
})
testEventer, err := event.NewEventer(testLogger, testLock, "Test_requestCtxInterceptor", c.EventerConfig)
require.NoError(t, err)
tests := []struct {
name string
requestCtx context.Context
@@ -89,6 +103,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn common.AuthTokenRepoFactory
serversRepoFn common.ServersRepoFactory
kms *kms.Kms
eventer *event.Eventer
ticket string
wantFactoryErr bool
wantFactoryErrMatch *errors.Template
@@ -103,6 +118,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing iam repo",
@@ -113,6 +129,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
iamRepoFn: iamRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing auth token repo",
@@ -123,6 +140,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
iamRepoFn: iamRepoFn,
authTokenRepoFn: atRepoFn,
kms: kmsCache,
eventer: testEventer,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing servers repo function",
@@ -133,6 +151,17 @@ func Test_requestCtxInterceptor(t *testing.T) {
iamRepoFn: iamRepoFn,
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
eventer: testEventer,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing kms",
},
{
name: "missing-eventer",
requestCtx: newReqCtx(validGatewayTicket),
iamRepoFn: iamRepoFn,
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing kms",
@@ -144,6 +173,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
wantFactoryErr: true,
wantFactoryErrMatch: errors.T(errors.InvalidParameter),
wantFactoryErrContains: "missing ticket",
@@ -155,6 +185,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -170,6 +201,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -185,6 +217,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -200,6 +233,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -215,6 +249,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -230,6 +265,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -242,6 +278,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: "validGatewayTicket",
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -254,6 +291,7 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: "validGatewayTicket",
wantRequestErr: true,
wantRequestErrMatch: errors.T(errors.Internal),
@@ -266,13 +304,14 @@ func Test_requestCtxInterceptor(t *testing.T) {
authTokenRepoFn: atRepoFn,
serversRepoFn: serversRepoFn,
kms: kmsCache,
eventer: testEventer,
ticket: validGatewayTicket,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
interceptor, err := requestCtxInterceptor(factoryCtx, tt.iamRepoFn, tt.authTokenRepoFn, tt.serversRepoFn, tt.kms, tt.ticket)
interceptor, err := requestCtxInterceptor(factoryCtx, tt.iamRepoFn, tt.authTokenRepoFn, tt.serversRepoFn, tt.kms, tt.ticket, tt.eventer)
if tt.wantFactoryErr {
require.Error(err)
assert.Nil(interceptor)
internal/servers/controller/listeners.go
+1 -1
View File
@@ -26,7 +26,7 @@ func (c *Controller) startListeners(ctx context.Context) error {
configureForAPI := func(ln *base.ServerListener) error {
var err error
if c.gatewayServer, c.gatewayTicket, err = newGatewayServer(ctx, c.IamRepoFn, c.AuthTokenRepoFn, c.ServersRepoFn, c.kms); err != nil {
if c.gatewayServer, c.gatewayTicket, err = newGatewayServer(ctx, c.IamRepoFn, c.AuthTokenRepoFn, c.ServersRepoFn, c.kms, c.conf.Eventer); err != nil {
return err
}
c.gatewayMux = newGatewayMux()
internal/servers/controller/tickers.go
+4 -4
View File
@@ -17,7 +17,7 @@ const (
)
// This is exported so it can be tweaked in tests
var RecoveryNonceCleanupInterval = 2 * time.Minute
var NonceCleanupInterval = 2 * time.Minute
func (c *Controller) startStatusTicking(cancelCtx context.Context) {
const op = "controller.(Controller).startStatusTicking"
@@ -51,8 +51,8 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) {
}
}
func (c *Controller) startRecoveryNonceCleanupTicking(cancelCtx context.Context) {
const op = "controller.(Controller).startRecoveryNonceCleanupTicking"
func (c *Controller) startNonceCleanupTicking(cancelCtx context.Context) {
const op = "controller.(Controller).startNonceCleanupTicking"
timer := time.NewTimer(0)
for {
select {
@@ -72,7 +72,7 @@ func (c *Controller) startRecoveryNonceCleanupTicking(cancelCtx context.Context)
event.WriteSysEvent(cancelCtx, op, "recovery nonce cleanup successful", "nonces_cleaned", nonceCount)
}
}
timer.Reset(RecoveryNonceCleanupInterval)
timer.Reset(NonceCleanupInterval)
}
}
}
internal/servers/controller/worker_tls_config.go
+35 -3
View File
@@ -7,10 +7,12 @@ import (
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net"
"strings"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/servers"
wrapping "github.com/hashicorp/go-kms-wrapping"
"google.golang.org/protobuf/proto"
)
@@ -20,16 +22,30 @@ type workerAuthEntry struct {
conn net.Conn
}
// validateWorkerTls is called by the Go TLS stack with client info. It calls
// v1WorkerAuthConfig to validate the encryption and ensure the nonce has not
// been replayed, then stores connection information into the auth cache, which
// is used by the intercepting listener to log information.
func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config, error) {
for _, p := range hello.SupportedProtos {
switch {
case strings.HasPrefix(p, "v1workerauth-"):
tlsConf, workerInfo, err := c.v1WorkerAuthConfig(hello.SupportedProtos)
if err == nil {
// Set the info we need to prevent replays
c.workerAuthCache.Set(workerInfo.ConnectionNonce, &workerAuthEntry{
// Store info that can be retrieved in the intercepting
// listener.
//
// NOTE: This is not a load-or-store because we need to ensure
// replays can't happen across controllers, hence storing in the
// database in v1WorkerAuthConfig. Although this could serve as
// a secondary check, a load-or-store here keeps the test from
// ensuring that it is rejected at the database level via the
// nonce unique constraint, which is by far the more important
// thing to validate (although there are DB-level tests for that
// too).
c.workerAuthCache.Store(workerInfo.ConnectionNonce, &workerAuthEntry{
WorkerAuthInfo: workerInfo,
}, 0)
})
}
return tlsConf, err
}
@@ -37,6 +53,13 @@ func (c Controller) validateWorkerTls(hello *tls.ClientHelloInfo) (*tls.Config,
return nil, nil
}
// v1WorkerAuthConfig:
//
// * Reads the information from ALPN, combining across protos
// * Combines them to form the encrypted auth information
// * Validates it by decrypting against the configured KMS
// * Ensures that the nonce is unique to prevent replay attacks
// * Returns the shared TLS configuration that is used to establish the connection
func (c Controller) v1WorkerAuthConfig(protos []string) (*tls.Config, *base.WorkerAuthInfo, error) {
var firstMatchProto string
var encString string
@@ -69,6 +92,15 @@ func (c Controller) v1WorkerAuthConfig(protos []string) (*tls.Config, *base.Work
return nil, nil, err
}
// Check for replays
serversRepo, err := c.ServersRepoFn()
if err != nil {
return nil, nil, fmt.Errorf("unable to fetch servers repo: %w", err)
}
if err := serversRepo.AddNonce(c.baseContext, info.ConnectionNonce, servers.NoncePurposeWorkerAuth); err != nil {
return nil, nil, fmt.Errorf("unable to add connection nonce to database: %w", err)
}
rootCAs := x509.NewCertPool()
if ok := rootCAs.AppendCertsFromPEM(info.CertPEM); !ok {
return nil, info, errors.New("unable to add ca cert to cert pool")