diff --git a/Makefile b/Makefile index bc85743..170c1d5 100644 --- a/Makefile +++ b/Makefile @@ -80,3 +80,7 @@ test-dev: test-dev-infra: python3 -m unittest discover -s dev -p 'test_*.py' python3 dev/local.py verify-infra + +.PHONY: test-dev-rotation +test-dev-rotation: + python3 dev/rotation.py diff --git a/backend/notifications/store/mongo.go b/backend/notifications/store/mongo.go index 70bffd3..81b7c65 100644 --- a/backend/notifications/store/mongo.go +++ b/backend/notifications/store/mongo.go @@ -9,11 +9,11 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/repository" "git.sendico.io/tech/sxchnge/shared/database/repository/builder" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/template" merrors "git.sendico.io/tech/sxchnge/shared/errors" mlogger "git.sendico.io/tech/sxchnge/shared/logging" "go.mongodb.org/mongo-driver/v2/bson" - "go.mongodb.org/mongo-driver/v2/mongo" "go.uber.org/zap" ) @@ -21,7 +21,7 @@ type Store struct { template.DBImp[*model.Notification] } -func New(logger mlogger.Logger, db *mongo.Database) (*Store, error) { +func New(logger mlogger.Logger, db source.Source) (*Store, error) { s := &Store{*template.Create[*model.Notification](mlogger.Named(logger, "store"), "notifications", db)} ttl := int32(0) for _, index := range []*ri.Definition{ diff --git a/backend/product/customers/internal/customer_test.go b/backend/product/customers/internal/customer_test.go index bc5eec4..1607a8b 100644 --- a/backend/product/customers/internal/customer_test.go +++ b/backend/product/customers/internal/customer_test.go @@ -2,6 +2,10 @@ package accountserviceimp import ( "context" + "net/url" + "testing" + "time" + "git.sendico.io/tech/sxchnge/product/auth/config" customermodel "git.sendico.io/tech/sxchnge/product/customers/model" sessionmodel "git.sendico.io/tech/sxchnge/product/sessions/model" @@ -15,9 +19,6 @@ import ( "github.com/testcontainers/testcontainers-go/modules/mongodb" "go.mongodb.org/mongo-driver/v2/bson" "go.uber.org/zap" - "net/url" - "testing" - "time" ) func customer(login string) *customermodel.Account { @@ -86,7 +87,10 @@ func TestCustomerLifecycleWithoutTenancy(t *testing.T) { deps, err := edgemongo.ResolveDependencies(factory) require.NoError(t, err) - collections, err := deps.Database.ListCollectionNames(ctx, bson.D{}) + raw, rawContext, release, err := deps.Database.Acquire(ctx) + require.NoError(t, err) + defer release() + collections, err := raw.ListCollectionNames(rawContext, bson.D{}) require.NoError(t, err) for _, forbidden := range []string{"organizations", "roles", "policies", "policy_assignments", "invitations", "accounts"} { require.NotContains(t, collections, forbidden) @@ -116,7 +120,7 @@ func TestCustomerLifecycleWithoutTenancy(t *testing.T) { require.ErrorIs(t, err, outerFailure) _, err = svc.accountDB.GetByEmail(ctx, "outer@example.test") require.ErrorIs(t, err, merrors.ErrNoData) - count, err := deps.Database.Collection(config.VerificationTokensCollection).CountDocuments(ctx, bson.D{}) + count, err := raw.Collection(config.VerificationTokensCollection).CountDocuments(rawContext, bson.D{}) require.NoError(t, err) require.EqualValues(t, 2, count) } diff --git a/backend/product/customers/store/mongo/db.go b/backend/product/customers/store/mongo/db.go index 58dd464..58aed01 100644 --- a/backend/product/customers/store/mongo/db.go +++ b/backend/product/customers/store/mongo/db.go @@ -1,14 +1,14 @@ package accountdb import ( - customermodel "git.sendico.io/tech/sxchnge/product/customers/model" "strings" + customermodel "git.sendico.io/tech/sxchnge/product/customers/model" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/template" "git.sendico.io/tech/sxchnge/shared/logging" "git.sendico.io/tech/sxchnge/shared/service" - "go.mongodb.org/mongo-driver/v2/mongo" "go.uber.org/zap" ) @@ -16,11 +16,11 @@ type AccountDB struct { template.DBImp[*customermodel.Account] } -func Create(logger mlogger.Logger, db *mongo.Database) (*AccountDB, error) { +func Create(logger mlogger.Logger, db source.Source) (*AccountDB, error) { return CreateWithCollection(logger, db, mservice.Accounts) } -func CreateWithCollection(logger mlogger.Logger, db *mongo.Database, collection string) (*AccountDB, error) { +func CreateWithCollection(logger mlogger.Logger, db source.Source, collection string) (*AccountDB, error) { if normalized := strings.TrimSpace(collection); normalized != "" { collection = normalized } else { diff --git a/backend/product/sessions/store/db.go b/backend/product/sessions/store/db.go index 8b9549a..444068a 100644 --- a/backend/product/sessions/store/db.go +++ b/backend/product/sessions/store/db.go @@ -1,15 +1,15 @@ package refreshtokens import ( - sessionmodel "git.sendico.io/tech/sxchnge/product/sessions/model" "strings" + sessionmodel "git.sendico.io/tech/sxchnge/product/sessions/model" "git.sendico.io/tech/sxchnge/shared/database/repository" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/template" "git.sendico.io/tech/sxchnge/shared/logging" "git.sendico.io/tech/sxchnge/shared/service" - "go.mongodb.org/mongo-driver/v2/mongo" "go.uber.org/zap" ) @@ -24,7 +24,7 @@ type RefreshTokenDB struct { collection Collections } -func Create(logger mlogger.Logger, db *mongo.Database) (*RefreshTokenDB, error) { +func Create(logger mlogger.Logger, db source.Source) (*RefreshTokenDB, error) { return CreateWithCollections(logger, db, Collections{}) } @@ -38,11 +38,11 @@ func (c Collections) WithDefaults() Collections { return c } -func CreateWithCollections(logger mlogger.Logger, db *mongo.Database, collections Collections) (*RefreshTokenDB, error) { +func CreateWithCollections(logger mlogger.Logger, db source.Source, collections Collections) (*RefreshTokenDB, error) { collections = collections.WithDefaults() p := &RefreshTokenDB{ DBImp: *template.Create[*sessionmodel.RefreshToken](logger, mservice.Type(collections.RefreshTokens), db), - clients: repository.CreateMongoRepository(db, collections.Clients), + clients: repository.CreateRepository(db, collections.Clients), collection: collections, } diff --git a/backend/product/sessions/store/refreshtokens_test.go b/backend/product/sessions/store/refreshtokens_test.go index a3b707e..24b12bf 100644 --- a/backend/product/sessions/store/refreshtokens_test.go +++ b/backend/product/sessions/store/refreshtokens_test.go @@ -8,11 +8,12 @@ import ( "crypto/sha1" "errors" "fmt" - sessionmodel "git.sendico.io/tech/sxchnge/product/sessions/model" "testing" "time" + sessionmodel "git.sendico.io/tech/sxchnge/product/sessions/model" "git.sendico.io/tech/sxchnge/product/sessions/store" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/errors" factory "git.sendico.io/tech/sxchnge/shared/logging/factory" "github.com/stretchr/testify/assert" @@ -55,7 +56,7 @@ func setupTestDBWithMongo(t *testing.T) (*refreshtokens.RefreshTokenDB, *mongo.D database := client.Database(fmt.Sprintf("test_refresh_tokens_%x", sha1.Sum([]byte(t.Name())))) logger := factory.NewLogger(true) - db, err := refreshtokens.Create(logger, database) + db, err := refreshtokens.Create(logger, source.Fixed(database)) require.NoError(t, err, "failed to create refresh tokens db") cleanup := func() { diff --git a/backend/product/verification/store/db.go b/backend/product/verification/store/db.go index e495739..a163daf 100644 --- a/backend/product/verification/store/db.go +++ b/backend/product/verification/store/db.go @@ -1,16 +1,16 @@ package verification import ( - verificationmodel "git.sendico.io/tech/sxchnge/product/verification/model" "strings" + verificationmodel "git.sendico.io/tech/sxchnge/product/verification/model" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/template" "git.sendico.io/tech/sxchnge/shared/database/transaction" "git.sendico.io/tech/sxchnge/shared/logging" "git.sendico.io/tech/sxchnge/shared/ratelimit/store" "git.sendico.io/tech/sxchnge/shared/service" - "go.mongodb.org/mongo-driver/v2/mongo" "go.uber.org/zap" ) @@ -29,7 +29,7 @@ const verificationContextIdempotencyIndexName = "uniq_verification_context_idemp func Create( logger mlogger.Logger, - db *mongo.Database, + db source.Source, tf transaction.Factory, ) (*verificationDB, error) { return CreateWithCollection(logger, db, tf, "") @@ -37,7 +37,7 @@ func Create( func CreateWithCollection( logger mlogger.Logger, - db *mongo.Database, + db source.Source, tf transaction.Factory, collection string, ) (*verificationDB, error) { diff --git a/backend/shared/database/connection.go b/backend/shared/database/connection.go index 52092fd..2be6c7f 100644 --- a/backend/shared/database/connection.go +++ b/backend/shared/database/connection.go @@ -4,56 +4,38 @@ import ( "context" mongoimpl "git.sendico.io/tech/sxchnge/shared/database/internal/mongo" - "git.sendico.io/tech/sxchnge/shared/errors" - "git.sendico.io/tech/sxchnge/shared/logging" - "go.mongodb.org/mongo-driver/v2/mongo" + "git.sendico.io/tech/sxchnge/shared/database/source" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" "go.mongodb.org/mongo-driver/v2/mongo/readpref" ) -// Connection represents a low-level database connection lifecycle. type Connection interface { - Disconnect(ctx context.Context) error - Ping(ctx context.Context) error + Disconnect(context.Context) error + Ping(context.Context) error } - -// MongoConnection provides direct access to the underlying mongo client. type MongoConnection struct { - client *mongo.Client - database string -} - -func (c *MongoConnection) Client() *mongo.Client { - return c.client -} - -func (c *MongoConnection) Database() *mongo.Database { - return c.client.Database(c.database) -} - -func (c *MongoConnection) Disconnect(ctx context.Context) error { - return c.client.Disconnect(ctx) + database source.Source + close func(context.Context) error } +func (c *MongoConnection) Database() source.Source { return c.database } +func (c *MongoConnection) Disconnect(ctx context.Context) error { return c.close(ctx) } func (c *MongoConnection) Ping(ctx context.Context) error { - return c.client.Ping(ctx, readpref.Primary()) + database, ctx, release, err := c.database.Acquire(ctx) + if err != nil { + return err + } + defer release() + return database.Client().Ping(ctx, readpref.Primary()) } - -// ConnectMongo returns a low-level MongoDB connection without constructing repositories. func ConnectMongo(logger mlogger.Logger, config *Config) (*MongoConnection, error) { - if config == nil { - return nil, merrors.InvalidArgument("database configuration is nil", "config") + if config == nil || config.Driver != Mongo { + return nil, merrors.InvalidArgument("MongoDB configuration is required") } - if config.Driver != Mongo { - return nil, merrors.InvalidArgument("unsupported database driver: "+string(config.Driver), "config.driver") - } - - client, _, settings, err := mongoimpl.ConnectClient(logger, config.Settings) + database, close, err := mongoimpl.OpenSource(logger, config.Settings) if err != nil { return nil, err } - - return &MongoConnection{ - client: client, - database: settings.Database, - }, nil + return &MongoConnection{database: database, close: close}, nil } diff --git a/backend/shared/database/dependencies/dependencies.go b/backend/shared/database/dependencies/dependencies.go index e8c4dc2..49013aa 100644 --- a/backend/shared/database/dependencies/dependencies.go +++ b/backend/shared/database/dependencies/dependencies.go @@ -2,19 +2,19 @@ package mongo import ( "git.sendico.io/tech/sxchnge/shared/database" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" "git.sendico.io/tech/sxchnge/shared/errors" - "go.mongodb.org/mongo-driver/v2/mongo" ) // Dependencies resolves local DB/transaction primitives for edge-local store wiring only. type Dependencies struct { - Database *mongo.Database + Database source.Source Transactions transaction.Factory } type DatabaseProvider interface { - MongoDatabase() *mongo.Database + MongoDatabase() source.Source } func ResolveDependencies(dbf db.Factory) (*Dependencies, error) { diff --git a/backend/shared/database/internal/mongo/db.go b/backend/shared/database/internal/mongo/db.go index 1cbd093..170173b 100644 --- a/backend/shared/database/internal/mongo/db.go +++ b/backend/shared/database/internal/mongo/db.go @@ -3,15 +3,18 @@ package mongo import ( "context" "fmt" - sharedmodel "git.sendico.io/tech/sxchnge/shared/model" "net" "os" "strings" + "time" mutil "git.sendico.io/tech/sxchnge/shared/config" "git.sendico.io/tech/sxchnge/shared/database/internal/transaction" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" + dbvault "git.sendico.io/tech/sxchnge/shared/database/vault" "git.sendico.io/tech/sxchnge/shared/logging" + sharedmodel "git.sendico.io/tech/sxchnge/shared/model" "github.com/go-viper/mapstructure/v2" "go.mongodb.org/mongo-driver/v2/mongo" "go.mongodb.org/mongo-driver/v2/mongo/readpref" @@ -20,28 +23,29 @@ import ( // Config represents configuration type Config struct { - Port *string `mapstructure:"port"` - PortEnv *string `mapstructure:"port_env"` - User *string `mapstructure:"user"` - UserEnv *string `mapstructure:"user_env"` - PasswordEnv string `mapstructure:"password_env"` - Database *string `mapstructure:"database"` - DatabaseEnv *string `mapstructure:"database_env"` - Host *string `mapstructure:"host"` - HostEnv *string `mapstructure:"host_env"` - Hosts []string `mapstructure:"hosts,omitempty"` - HostsEnvPrefix *string `mapstructure:"hosts_env_prefix,omitempty"` - HostsEnvPrefixEnv *string `mapstructure:"hosts_env_prefix_env,omitempty"` - PortsEnvPrefix *string `mapstructure:"ports_env_prefix,omitempty"` - PortsEnvPrefixEnv *string `mapstructure:"ports_env_prefix_env,omitempty"` - URI *string `mapstructure:"uri,omitempty"` - URIEnv *string `mapstructure:"uri_env,omitempty"` - AuthSource *string `mapstructure:"auth_source,omitempty"` - AuthSourceEnv *string `mapstructure:"auth_source_env,omitempty"` - AuthMechanism *string `mapstructure:"auth_mechanism,omitempty"` - AuthMechanismEnv *string `mapstructure:"auth_mechanism_env,omitempty"` - ReplicaSet *string `mapstructure:"replica_set,omitempty"` - ReplicaSetEnv *string `mapstructure:"replica_set_env,omitempty"` + Vault *dbvault.Config `mapstructure:"vault"` + Port *string `mapstructure:"port"` + PortEnv *string `mapstructure:"port_env"` + User *string `mapstructure:"user"` + UserEnv *string `mapstructure:"user_env"` + PasswordEnv string `mapstructure:"password_env"` + Database *string `mapstructure:"database"` + DatabaseEnv *string `mapstructure:"database_env"` + Host *string `mapstructure:"host"` + HostEnv *string `mapstructure:"host_env"` + Hosts []string `mapstructure:"hosts,omitempty"` + HostsEnvPrefix *string `mapstructure:"hosts_env_prefix,omitempty"` + HostsEnvPrefixEnv *string `mapstructure:"hosts_env_prefix_env,omitempty"` + PortsEnvPrefix *string `mapstructure:"ports_env_prefix,omitempty"` + PortsEnvPrefixEnv *string `mapstructure:"ports_env_prefix_env,omitempty"` + URI *string `mapstructure:"uri,omitempty"` + URIEnv *string `mapstructure:"uri_env,omitempty"` + AuthSource *string `mapstructure:"auth_source,omitempty"` + AuthSourceEnv *string `mapstructure:"auth_source_env,omitempty"` + AuthMechanism *string `mapstructure:"auth_mechanism,omitempty"` + AuthMechanismEnv *string `mapstructure:"auth_mechanism_env,omitempty"` + ReplicaSet *string `mapstructure:"replica_set,omitempty"` + ReplicaSetEnv *string `mapstructure:"replica_set_env,omitempty"` } type DBSettings struct { @@ -119,37 +123,42 @@ func dialMongo(logger mlogger.Logger, dbSettings *DBSettings) (*mongo.Client, er return client, nil } -func ConnectClient(logger mlogger.Logger, settings sharedmodel.SettingsT) (*mongo.Client, *Config, *DBSettings, error) { - config, dbSettings, err := decodeConfig(logger, settings) - if err != nil { - return nil, nil, nil, err - } - - client, err := dialMongo(logger, dbSettings) - if err != nil { - return nil, nil, nil, err - } - - return client, config, dbSettings, nil -} - // DB represents the structure of the database type DB struct { - logger mlogger.Logger - config *DBSettings - client *mongo.Client -} - -func (db *DB) db() *mongo.Database { - return db.client.Database(db.config.Database) -} - -func (db *DB) MongoDatabase() *mongo.Database { - return db.db() + logger mlogger.Logger + database source.Source + close func(context.Context) error } +func (db *DB) MongoDatabase() source.Source { return db.database } func (db *DB) TransactionFactory() transaction.Factory { - return transactionimp.CreateFactory(db.client) + return transactionimp.CreateSourceFactory(db.database) +} + +// OpenSource owns either a fixed connection or a rotating Vault-backed source. +func OpenSource(logger mlogger.Logger, settings sharedmodel.SettingsT) (source.Source, func(context.Context) error, error) { + config, dbSettings, err := decodeConfig(logger, settings) + if err != nil { + return nil, nil, err + } + if config.Vault != nil { + issuer, err := dbvault.New(*config.Vault, dbSettings.URI) + if err != nil { + return nil, nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + manager, err := source.New(ctx, logger.Named("credentials"), dbSettings.Database, issuer) + if err != nil { + return nil, nil, err + } + return manager, manager.Close, nil + } + client, err := dialMongo(logger, dbSettings) + if err != nil { + return nil, nil, err + } + return source.Fixed(client.Database(dbSettings.Database)), client.Disconnect, nil } func collectReplicaHosts(configuredHosts []string, replicaSet, defaultPort, hostPrefix, portPrefix string) []string { @@ -217,24 +226,17 @@ func collectReplicaHosts(configuredHosts []string, replicaSet, defaultPort, host } func (db *DB) CloseConnection() { - if err := db.client.Disconnect(context.Background()); err != nil { - db.logger.Warn("Failed to close connection", zap.Error(err)) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := db.close(ctx); err != nil { + db.logger.Warn("Failed to drain database connection", zap.Error(err)) } - db.logger.Info("Database connection closed") } -// NewConnection creates a new database connection func NewConnection(logger mlogger.Logger, settings sharedmodel.SettingsT) (*DB, error) { - client, _, dbSettings, err := ConnectClient(logger, settings) + database, close, err := OpenSource(logger, settings) if err != nil { return nil, err } - - db := &DB{ - logger: logger.Named("db"), - config: dbSettings, - client: client, - } - - return db, nil + return &DB{logger: logger, database: database, close: close}, nil } diff --git a/backend/shared/database/internal/repository/index.go b/backend/shared/database/internal/repository/index.go index 0c4a39c..6e59ed9 100644 --- a/backend/shared/database/internal/repository/index.go +++ b/backend/shared/database/internal/repository/index.go @@ -12,7 +12,14 @@ import ( ) func (r *MongoRepository) CreateIndex(def *ri.Definition) error { - if r.collection == nil { + ctx := context.Background() + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + if collection == nil { return merrors.NoData("data collection is not set") } if len(def.Keys) == 0 { @@ -49,28 +56,40 @@ func (r *MongoRepository) CreateIndex(def *ri.Definition) error { opts.SetSparse(def.Sparse) } - _, err := r.collection.Indexes().CreateOne( - context.Background(), + _, err = collection.Indexes().CreateOne( + ctx, mongo.IndexModel{Keys: keys, Options: opts}, ) return err } func (r *MongoRepository) DropIndex(ctx context.Context, name string) error { - if r.collection == nil { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + if collection == nil { return merrors.NoData("data collection is not set") } if name == "" { return merrors.InvalidArgument("index name is required", "index.name") } - return r.collection.Indexes().DropOne(ctx, name) + return collection.Indexes().DropOne(ctx, name) } func (r *MongoRepository) ListIndexDefinitions(ctx context.Context) ([]*ri.Definition, error) { - if r.collection == nil { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return nil, err + } + defer release() + + if collection == nil { return nil, merrors.NoData("data collection is not set") } - cursor, err := r.collection.Indexes().List(ctx) + cursor, err := collection.Indexes().List(ctx) if err != nil { return nil, err } diff --git a/backend/shared/database/internal/repository/repository.go b/backend/shared/database/internal/repository/repository.go index f11e249..c937364 100644 --- a/backend/shared/database/internal/repository/repository.go +++ b/backend/shared/database/internal/repository/repository.go @@ -8,6 +8,7 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/record" "git.sendico.io/tech/sxchnge/shared/database/repository/builder" rd "git.sendico.io/tech/sxchnge/shared/database/repository/decoder" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" "git.sendico.io/tech/sxchnge/shared/errors" "go.mongodb.org/mongo-driver/v2/bson" @@ -17,7 +18,7 @@ import ( type MongoRepository struct { collectionName string - collection *mongo.Collection + source source.Source } func idFilter(id bson.ObjectID) bson.D { @@ -27,10 +28,23 @@ func idFilter(id bson.ObjectID) bson.D { } func NewMongoRepository(db *mongo.Database, collection string) *MongoRepository { - return &MongoRepository{ - collectionName: collection, - collection: db.Collection(collection), + return NewRepository(source.Fixed(db), collection) +} + +func NewRepository(database source.Source, collection string) *MongoRepository { + return &MongoRepository{collectionName: collection, source: database} +} + +func (r *MongoRepository) acquire(ctx context.Context) (*mongo.Collection, context.Context, func(), error) { + if r.source == nil { + return nil, ctx, nil, merrors.FailedPrecondition("database source is not configured") } + + database, scoped, release, err := r.source.Acquire(ctx) + if err != nil { + return nil, ctx, nil, err + } + return database.Collection(r.collectionName), scoped, release, nil } func (r *MongoRepository) Collection() string { @@ -48,8 +62,14 @@ func prepareForWrite(obj storable.Storable) bson.ObjectID { } func (r *MongoRepository) Insert(ctx context.Context, obj storable.Storable, getFilter builder.Query) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + prepareForWrite(obj) - _, err := r.collection.InsertOne(ctx, obj) + _, err = collection.InsertOne(ctx, obj) if mongo.IsDuplicateKeyError(err) { if getFilter != nil { if err = r.FindOneByFilter(ctx, getFilter, obj); err != nil { @@ -62,6 +82,12 @@ func (r *MongoRepository) Insert(ctx context.Context, obj storable.Storable, get } func (r *MongoRepository) InsertMany(ctx context.Context, objects []storable.Storable) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + if len(objects) == 0 { return nil } @@ -72,12 +98,18 @@ func (r *MongoRepository) InsertMany(ctx context.Context, objects []storable.Sto docs[i] = obj } - _, err := r.collection.InsertMany(ctx, docs) + _, err = collection.InsertMany(ctx, docs) return err } func (r *MongoRepository) findOneByFilterImp(ctx context.Context, filter bson.D, errMessage string, result storable.Storable) error { - err := r.collection.FindOne(ctx, filter).Decode(result) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + err = collection.FindOne(ctx, filter).Decode(result) if errors.Is(err, mongo.ErrNoDocuments) { return merrors.NoData(errMessage) } @@ -94,7 +126,13 @@ func (r *MongoRepository) Get(ctx context.Context, id bson.ObjectID, result stor type QueryFunc func(ctx context.Context, collection *mongo.Collection) (*mongo.Cursor, error) func (r *MongoRepository) executeQuery(ctx context.Context, queryFunc QueryFunc, decoder rd.DecodingFunc) error { - cursor, err := queryFunc(ctx, r.collection) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + cursor, err := queryFunc(ctx, collection) if errors.Is(err, mongo.ErrNoDocuments) { return merrors.NoData("no_items_in_array") } @@ -111,7 +149,7 @@ func (r *MongoRepository) executeQuery(ctx context.Context, queryFunc QueryFunc, } } - return nil + return cursor.Err() } func (r *MongoRepository) Aggregate(ctx context.Context, pipeline builder.Pipeline, decoder rd.DecodingFunc) error { @@ -129,8 +167,14 @@ func (r *MongoRepository) FindManyByFilter(ctx context.Context, query builder.Qu } func (r *MongoRepository) FindAllByFilter(ctx context.Context, query builder.Query, result any) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + if query == nil { - cursor, err := r.collection.Find(ctx, bson.D{}) + cursor, err := collection.Find(ctx, bson.D{}) if err != nil { return err } @@ -139,7 +183,7 @@ func (r *MongoRepository) FindAllByFilter(ctx context.Context, query builder.Que }() return cursor.All(ctx, result) } - cursor, err := r.collection.Find(ctx, query.BuildQuery(), query.BuildOptions()) + cursor, err := collection.Find(ctx, query.BuildQuery(), query.BuildOptions()) if err != nil { return err } @@ -154,18 +198,30 @@ func (r *MongoRepository) FindOneByFilter(ctx context.Context, query builder.Que } func (r *MongoRepository) Update(ctx context.Context, obj storable.Storable) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + obj.Update() - return r.collection.FindOneAndReplace(ctx, idFilter(*obj.GetID()), obj).Err() + return collection.FindOneAndReplace(ctx, idFilter(*obj.GetID()), obj).Err() } func (r *MongoRepository) ReplaceOneByFilter(ctx context.Context, query builder.Query, obj storable.Storable) (bool, error) { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return false, err + } + defer release() + if query == nil { return false, merrors.InvalidArgument("filter is required", "filter") } if obj == nil { return false, merrors.InvalidArgument("object is required", "obj") } - result, err := r.collection.ReplaceOne(ctx, query.BuildQuery(), obj) + result, err := collection.ReplaceOne(ctx, query.BuildQuery(), obj) if mongo.IsDuplicateKeyError(err) { return false, merrors.DataConflict("duplicate_key") } @@ -176,21 +232,39 @@ func (r *MongoRepository) ReplaceOneByFilter(ctx context.Context, query builder. } func (r *MongoRepository) Upsert(ctx context.Context, obj storable.Storable) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + id := prepareForWrite(obj) - _, err := r.collection.ReplaceOne(ctx, idFilter(id), obj, options.Replace().SetUpsert(true)) + _, err = collection.ReplaceOne(ctx, idFilter(id), obj, options.Replace().SetUpsert(true)) return err } func (r *MongoRepository) Patch(ctx context.Context, id bson.ObjectID, patch builder.Patch) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + if id.IsZero() { return merrors.InvalidArgument("zero id provided while patching", "id") } - _, err := r.collection.UpdateByID(ctx, id, patch.Build()) + _, err = collection.UpdateByID(ctx, id, patch.Build()) return err } func (r *MongoRepository) PatchMany(ctx context.Context, query builder.Query, patch builder.Patch) (int, error) { - result, err := r.collection.UpdateMany(ctx, query.BuildQuery(), patch.Build()) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return 0, err + } + defer release() + + result, err := collection.UpdateMany(ctx, query.BuildQuery(), patch.Build()) if err != nil { return 0, err } @@ -198,10 +272,16 @@ func (r *MongoRepository) PatchMany(ctx context.Context, query builder.Query, pa } func (r *MongoRepository) ListIDs(ctx context.Context, query builder.Query) ([]bson.ObjectID, error) { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return nil, err + } + defer release() + filter := query.BuildQuery() findOptions := query.BuildOptions().SetProjection(bson.M{storable.IDField: 1}) - cursor, err := r.collection.Find(ctx, filter, findOptions) + cursor, err := collection.Find(ctx, filter, findOptions) if err != nil { return nil, err } @@ -227,24 +307,45 @@ func (r *MongoRepository) ListIDs(ctx context.Context, query builder.Query) ([]b } func (r *MongoRepository) Delete(ctx context.Context, id bson.ObjectID) error { - _, err := r.collection.DeleteOne(ctx, idFilter(id)) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + _, err = collection.DeleteOne(ctx, idFilter(id)) return err } func (r *MongoRepository) DeleteMany(ctx context.Context, query builder.Query) error { - _, err := r.collection.DeleteMany(ctx, query.BuildQuery()) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + + _, err = collection.DeleteMany(ctx, query.BuildQuery()) return err } func (r *MongoRepository) Name() string { - return r.collection.Name() + return r.collectionName } func (r *MongoRepository) WithTransaction(ctx context.Context, cb transaction.Callback) (any, error) { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return nil, err + } + defer release() + if cb == nil { return nil, merrors.InvalidArgument("transaction callback is required", "callback") } - session, err := r.collection.Database().Client().StartSession() + if transaction.ActiveMongoTransaction(ctx) { + return cb(ctx) + } + session, err := collection.Database().Client().StartSession() if err != nil { return nil, err } diff --git a/backend/shared/database/internal/repository/repository_filter.go b/backend/shared/database/internal/repository/repository_filter.go index 381bb1e..420faf9 100644 --- a/backend/shared/database/internal/repository/repository_filter.go +++ b/backend/shared/database/internal/repository/repository_filter.go @@ -12,8 +12,14 @@ import ( ) func (r *MongoRepository) FindFirstByFilter(ctx context.Context, query builder.Query, result storable.Storable) error { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return err + } + defer release() + opts := query.BuildOptions().SetLimit(1) - cursor, err := r.collection.Find(ctx, query.BuildQuery(), opts) + cursor, err := collection.Find(ctx, query.BuildQuery(), opts) if err != nil { return err } @@ -30,10 +36,22 @@ func (r *MongoRepository) FindFirstByFilter(ctx context.Context, query builder.Q } func (r *MongoRepository) CountByFilter(ctx context.Context, query builder.Query) (int64, error) { - return r.collection.CountDocuments(ctx, query.BuildQuery()) + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return 0, err + } + defer release() + + return collection.CountDocuments(ctx, query.BuildQuery()) } func (r *MongoRepository) PatchOneByFilter(ctx context.Context, query builder.Query, patch builder.Patch, result storable.Storable) (bool, error) { + collection, ctx, release, err := r.acquire(ctx) + if err != nil { + return false, err + } + defer release() + if query == nil { return false, merrors.InvalidArgument("filter is required", "filter") } @@ -43,7 +61,7 @@ func (r *MongoRepository) PatchOneByFilter(ctx context.Context, query builder.Qu if result == nil { return false, merrors.InvalidArgument("result is required", "result") } - err := r.collection.FindOneAndUpdate( + err = collection.FindOneAndUpdate( ctx, query.BuildQuery(), patch.Build(), diff --git a/backend/shared/database/internal/transaction/factory.go b/backend/shared/database/internal/transaction/factory.go index 344b312..f84c736 100644 --- a/backend/shared/database/internal/transaction/factory.go +++ b/backend/shared/database/internal/transaction/factory.go @@ -1,18 +1,15 @@ package transactionimp import ( + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" - "go.mongodb.org/mongo-driver/v2/mongo" ) -type MongoTransactionFactory struct { - client *mongo.Client -} +type MongoTransactionFactory struct{ database source.Source } -func (mtf *MongoTransactionFactory) CreateTransaction() transaction.Transaction { - return Create(mtf.client) +func (f *MongoTransactionFactory) CreateTransaction() transaction.Transaction { + return &MongoTransaction{database: f.database} } - -func CreateFactory(client *mongo.Client) transaction.Factory { - return &MongoTransactionFactory{client: client} +func CreateSourceFactory(database source.Source) transaction.Factory { + return &MongoTransactionFactory{database: database} } diff --git a/backend/shared/database/internal/transaction/transaction.go b/backend/shared/database/internal/transaction/transaction.go index 458d6c3..b31ff10 100644 --- a/backend/shared/database/internal/transaction/transaction.go +++ b/backend/shared/database/internal/transaction/transaction.go @@ -3,32 +3,29 @@ package transactionimp import ( "context" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" - "go.mongodb.org/mongo-driver/v2/mongo" + merrors "git.sendico.io/tech/sxchnge/shared/errors" ) -type MongoTransaction struct { - client *mongo.Client -} +type MongoTransaction struct{ database source.Source } -func (mt *MongoTransaction) Execute(ctx context.Context, cb transaction.Callback) (any, error) { - // Reuse an outer customer workflow transaction when a token store joins it. +func (t *MongoTransaction) Execute(ctx context.Context, cb transaction.Callback) (any, error) { + if cb == nil { + return nil, merrors.InvalidArgument("transaction callback is required") + } + database, ctx, release, err := t.database.Acquire(ctx) + if err != nil { + return nil, err + } + defer release() if transaction.ActiveMongoTransaction(ctx) { return cb(ctx) } - session, err := mt.client.StartSession() + session, err := database.Client().StartSession() if err != nil { return nil, err } defer session.EndSession(ctx) - - callback := func(c context.Context) (any, error) { - return cb(c) - } - - return session.WithTransaction(ctx, callback) -} - -func Create(client *mongo.Client) *MongoTransaction { - return &MongoTransaction{client: client} + return session.WithTransaction(ctx, cb) } diff --git a/backend/shared/database/repository/repository.go b/backend/shared/database/repository/repository.go index f53bc95..2a34c32 100644 --- a/backend/shared/database/repository/repository.go +++ b/backend/shared/database/repository/repository.go @@ -8,6 +8,7 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/repository/builder" rd "git.sendico.io/tech/sxchnge/shared/database/repository/decoder" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/database/transaction" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" @@ -53,6 +54,11 @@ type Repository interface { Collection() string } +func CreateRepository(db source.Source, collection string) Repository { + return repositoryimp.NewRepository(db, collection) +} + +// CreateMongoRepository adapts an externally owned fixed database. func CreateMongoRepository(db *mongo.Database, collection string) Repository { - return repositoryimp.NewMongoRepository(db, collection) + return CreateRepository(source.Fixed(db), collection) } diff --git a/backend/shared/database/rotation_test.go b/backend/shared/database/rotation_test.go new file mode 100644 index 0000000..433e498 --- /dev/null +++ b/backend/shared/database/rotation_test.go @@ -0,0 +1,275 @@ +package db_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + db "git.sendico.io/tech/sxchnge/shared/database" + dependencies "git.sendico.io/tech/sxchnge/shared/database/dependencies" + storable "git.sendico.io/tech/sxchnge/shared/database/record" + "git.sendico.io/tech/sxchnge/shared/database/repository" + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "git.sendico.io/tech/sxchnge/shared/model" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go/modules/mongodb" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + "go.uber.org/zap" +) + +type rotationRecord struct { + storable.Base `bson:",inline"` + Value string `bson:"value"` +} + +func (*rotationRecord) Collection() string { return "rotation" } + +type leaseServer struct { + admin *mongo.Database + issued atomic.Int64 + revoked sync.Map + outage atomic.Bool + badPassword atomic.Bool +} + +func (s *leaseServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if s.outage.Load() { + w.WriteHeader(503) + return + } + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/v1/auth/approle/login": + // A unique token owns each credential in this fake Vault HTTP boundary. + token := fmt.Sprintf("test-token-%d", s.issued.Load()+1) + _ = json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{"client_token": token, "lease_duration": 60}}) + case "/v1/database/creds/bff": + n := s.issued.Add(1) + user := fmt.Sprintf("lease%d", n) + err := s.admin.RunCommand(r.Context(), bson.D{{Key: "createUser", Value: user}, {Key: "pwd", Value: "test-password"}, {Key: "roles", Value: bson.A{bson.M{"role": "readWrite", "db": "rotation"}}}}).Err() + if err != nil { + w.WriteHeader(500) + return + } + password := "test-password" + if s.badPassword.Load() { + password = "wrong-password" + } + _ = json.NewEncoder(w).Encode(map[string]any{"lease_id": user, "lease_duration": 8, "data": map[string]string{"username": user, "password": password}}) + case "/v1/auth/token/revoke-self": + user := "lease" + strings.TrimPrefix(r.Header.Get("X-Vault-Token"), "test-token-") + _ = s.admin.RunCommand(r.Context(), bson.D{{Key: "dropUser", Value: user}}).Err() + s.revoked.Store(user, true) + w.WriteHeader(204) + default: + w.WriteHeader(404) + } +} + +func TestRotationPreservesRequestsCursorsAndTransactions(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + container, err := mongodb.Run(ctx, "mongo:latest", mongodb.WithReplicaSet("rs"), mongodb.WithUsername("root"), mongodb.WithPassword("test-admin-password")) + require.NoError(t, err) + t.Cleanup(func() { _ = container.Terminate(context.Background()) }) + raw, err := container.ConnectionString(ctx) + require.NoError(t, err) + uri, err := url.Parse(raw) + require.NoError(t, err) + q := uri.Query() + q.Set("directConnection", "true") + q.Set("authSource", "admin") + uri.RawQuery = q.Encode() + admin, err := mongo.Connect(options.Client().ApplyURI(uri.String())) + require.NoError(t, err) + defer admin.Disconnect(context.Background()) + fake := &leaseServer{admin: admin.Database("admin")} + vault := httptest.NewServer(fake) + defer vault.Close() + directory := t.TempDir() + role := filepath.Join(directory, "role") + secret := filepath.Join(directory, "secret") + require.NoError(t, os.WriteFile(role, []byte("test-role"), 0600)) + require.NoError(t, os.WriteFile(secret, []byte("test-secret"), 0600)) + uri.User = nil + uri.Path = "/rotation" + factory, err := db.NewConnection(zap.NewNop(), &db.Config{Driver: db.Mongo, Settings: model.SettingsT{"database": "rotation", "uri": uri.String(), "vault": map[string]any{"address": vault.URL, "role": "bff", "role_id_file": role, "secret_id_file": secret}}}) + require.NoError(t, err) + defer factory.CloseConnection() + deps, err := dependencies.ResolveDependencies(factory) + require.NoError(t, err) + repo := repository.CreateRepository(deps.Database, "rotation") + require.NoError(t, repo.Insert(ctx, &rotationRecord{Value: "cursor-fixture"}, nil)) + + var failures, requests atomic.Int64 + httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if _, e := repo.CountByFilter(r.Context(), repository.Query()); e != nil { + failures.Add(1) + w.WriteHeader(503) + return + } + requests.Add(1) + w.WriteHeader(200) + })) + defer httpServer.Close() + loadCtx, stopLoad := context.WithCancel(ctx) + var load sync.WaitGroup + defer func() { stopLoad(); load.Wait() }() + for range 4 { + load.Add(1) + go func() { + defer load.Done() + for loadCtx.Err() == nil { + res, e := http.Get(httpServer.URL) + if e != nil { + failures.Add(1) + } else { + res.Body.Close() + if res.StatusCode != 200 { + failures.Add(1) + } + } + time.Sleep(5 * time.Millisecond) + } + }() + } + + // A cursor and a transaction must keep generation 1 alive through publication. + first, scope, release, err := deps.Database.Acquire(ctx) + require.NoError(t, err) + firstClient := first.Client() + release() + _, _, _, err = deps.Database.Acquire(scope) + require.Error(t, err, "escaped operation scope must be rejected") + txEntered := make(chan struct{}) + finishTx := make(chan struct{}) + txDone := make(chan error, 1) + var entered sync.Once + go func() { + _, e := factory.TransactionFactory().CreateTransaction().Execute(ctx, func(tx context.Context) (any, error) { + if e := repo.Insert(tx, &rotationRecord{Value: "transaction-before"}, nil); e != nil { + return nil, e + } + entered.Do(func() { close(txEntered) }) + select { + case <-finishTx: + case <-tx.Done(): + return nil, tx.Err() + } + // Nested transaction must share the same session/generation. + return repo.WithTransaction(tx, func(nested context.Context) (any, error) { + return nil, repo.Insert(nested, &rotationRecord{Value: "transaction-after"}, nil) + }) + }) + txDone <- e + }() + select { + case <-txEntered: + case err := <-txDone: + t.Fatalf("transaction did not enter: %v", err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + cursorEntered := make(chan struct{}) + finishCursor := make(chan struct{}) + cursorDone := make(chan error, 1) + go func() { + var once sync.Once + cursorDone <- repo.FindManyByFilter(ctx, repository.Query(), func(d *mongo.Cursor) error { + once.Do(func() { close(cursorEntered) }) + select { + case <-finishCursor: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + }() + select { + case <-cursorEntered: + case err := <-cursorDone: + t.Fatalf("cursor did not enter: %v", err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + require.Eventually(t, func() bool { + database, _, free, e := deps.Database.Acquire(ctx) + if e != nil { + return false + } + defer free() + return database.Client() != firstClient + }, 6*time.Second, 20*time.Millisecond) + _, revoked := fake.revoked.Load("lease1") + require.False(t, revoked, "in-flight work lost its credentials") + close(finishTx) + require.NoError(t, <-txDone) + _, revoked = fake.revoked.Load("lease1") + require.False(t, revoked, "cursor must retain old pool") + close(finishCursor) + require.NoError(t, <-cursorDone) + require.Eventually(t, func() bool { _, ok := fake.revoked.Load("lease1"); return ok }, 2*time.Second, 20*time.Millisecond) + count, err := repo.CountByFilter(ctx, repository.Filter("value", "transaction-after")) + require.NoError(t, err) + require.EqualValues(t, 1, count) + + // Reject a candidate with invalid credentials while retaining the valid pool. + fake.badPassword.Store(true) + before := fake.issued.Load() + require.Eventually(t, func() bool { return fake.issued.Load() > before }, 5*time.Second, 20*time.Millisecond) + require.Eventually(t, func() bool { _, ok := fake.revoked.Load(fmt.Sprintf("lease%d", before+1)); return ok }, time.Second, 10*time.Millisecond) + require.NoError(t, repo.Insert(ctx, &rotationRecord{Value: "during-bad-refresh"}, nil)) + fake.badPassword.Store(false) + fake.outage.Store(true) + require.NoError(t, repo.Insert(ctx, &rotationRecord{Value: "during-vault-outage"}, nil)) + time.Sleep(300 * time.Millisecond) + fake.outage.Store(false) + require.Eventually(t, func() bool { return fake.issued.Load() > before+1 }, 3*time.Second, 20*time.Millisecond) + stopLoad() + load.Wait() + require.Greater(t, requests.Load(), int64(100)) + require.Zero(t, failures.Load(), "HTTP requests failed during rotation") + + // If Vault stays unavailable beyond the lease, fail closed without restarting. + fake.outage.Store(true) + require.Eventually(t, func() bool { _, e := repo.CountByFilter(ctx, repository.Query()); return e != nil }, 10*time.Second, 50*time.Millisecond) + fake.outage.Store(false) + require.Eventually(t, func() bool { _, e := repo.CountByFilter(ctx, repository.Query()); return e == nil }, 6*time.Second, 20*time.Millisecond) + // Transaction errors still roll back all writes after generation changes. + _, err = repo.WithTransaction(ctx, func(tx context.Context) (any, error) { + if e := repo.Insert(tx, &rotationRecord{Value: "rollback"}, nil); e != nil { + return nil, e + } + return nil, merrors.Internal("injected rollback") + }) + require.Error(t, err) + count, err = repo.CountByFilter(ctx, repository.Filter("value", "rollback")) + require.NoError(t, err) + require.Zero(t, count) + held, heldContext, releaseHeld, err := deps.Database.Acquire(ctx) + require.NoError(t, err) + manager := deps.Database.(interface{ Close(context.Context) error }) + short, stop := context.WithTimeout(ctx, 30*time.Millisecond) + require.ErrorIs(t, manager.Close(short), context.DeadlineExceeded) + stop() + _, _, _, err = deps.Database.Acquire(ctx) + require.Error(t, err) + require.NoError(t, held.Client().Ping(heldContext, nil), "shutdown must drain, not kill held work") + releaseHeld() + closed, stop := context.WithTimeout(ctx, 5*time.Second) + defer stop() + require.NoError(t, manager.Close(closed)) + +} diff --git a/backend/shared/database/source/manager.go b/backend/shared/database/source/manager.go new file mode 100644 index 0000000..0c35d53 --- /dev/null +++ b/backend/shared/database/source/manager.go @@ -0,0 +1,253 @@ +package source + +import ( + "context" + "sync" + "sync/atomic" + "time" + + merrors "git.sendico.io/tech/sxchnge/shared/errors" + mlogger "git.sendico.io/tech/sxchnge/shared/logging" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + "go.mongodb.org/mongo-driver/v2/mongo/readpref" + "go.uber.org/zap" +) + +// Lease owns a credential and its revocation capability. Neither may be logged. +type Lease struct { + URI string + ExpiresAt time.Time + Revoke func(context.Context) error +} +type Issuer interface { + Issue(context.Context) (Lease, error) +} + +type generation struct { + client *mongo.Client + lease Lease + refs int + retired bool + drained chan struct{} +} +type pin struct { + manager *Manager + generation *generation + active atomic.Bool +} +type pinKey struct{} + +// Manager publishes a verified replacement before retiring the previous pool. +// Every cursor, operation and transaction retains its originating generation. +type Manager struct { + mu sync.Mutex + current *generation + closed bool + issuer Issuer + database string + logger mlogger.Logger + cancel context.CancelFunc + workerDone chan struct{} + draining sync.WaitGroup +} + +func New(ctx context.Context, logger mlogger.Logger, database string, issuer Issuer) (*Manager, error) { + if database == "" || issuer == nil { + return nil, merrors.InvalidArgument("database and credential issuer are required") + } + life, cancel := context.WithCancel(context.Background()) + m := &Manager{issuer: issuer, database: database, logger: logger, cancel: cancel, workerDone: make(chan struct{})} + g, err := m.prepare(ctx) + if err != nil { + cancel() + return nil, err + } + m.current = g + go m.run(life) + return m, nil +} + +func (m *Manager) Acquire(ctx context.Context) (*mongo.Database, context.Context, func(), error) { + if err := ctx.Err(); err != nil { + return nil, ctx, nil, err + } + if p, ok := ctx.Value(pinKey{}).(*pin); ok && p.manager == m { + if !p.active.Load() { + return nil, ctx, nil, merrors.FailedPrecondition("database operation scope has ended") + } + return p.generation.client.Database(m.database), ctx, func() {}, nil + } + m.mu.Lock() + g := m.current + if m.closed || g == nil || !time.Now().Before(g.lease.ExpiresAt) { + m.mu.Unlock() + return nil, ctx, nil, merrors.ServiceUnavailable("no valid database credential is available") + } + if session := mongo.SessionFromContext(ctx); session != nil && session.Client() != g.client { + m.mu.Unlock() + return nil, ctx, nil, merrors.FailedPrecondition("transaction has no matching database operation scope") + } + g.refs++ + m.mu.Unlock() + p := &pin{manager: m, generation: g} + p.active.Store(true) + bounded, cancel := context.WithDeadline(ctx, g.lease.ExpiresAt) + scoped := context.WithValue(bounded, pinKey{}, p) + release := func() { + if !p.active.CompareAndSwap(true, false) { + return + } + cancel() + m.mu.Lock() + g.refs-- + if g.retired && g.refs == 0 { + close(g.drained) + } + m.mu.Unlock() + } + return g.client.Database(m.database), scoped, release, nil +} + +func (m *Manager) prepare(ctx context.Context) (*generation, error) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + lease, err := m.issuer.Issue(ctx) + if err != nil { + return nil, err + } + var client *mongo.Client + success := false + defer func() { + if !success { + if client != nil { + cleanup, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = client.Disconnect(cleanup) + cancel() + } + m.draining.Add(1) + go func() { defer m.draining.Done(); m.revoke(lease) }() + } + }() + if lease.Revoke == nil || time.Until(lease.ExpiresAt) < 2*time.Second { + return nil, merrors.FailedPrecondition("database credential lifetime is too short") + } + client, err = mongo.Connect(options.Client().ApplyURI(lease.URI).SetServerSelectionTimeout(5 * time.Second)) + if err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "create replacement database pool") + } + if err = client.Ping(ctx, readpref.Primary()); err != nil { + return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "validate replacement database pool") + } + if time.Until(lease.ExpiresAt) < time.Second { + return nil, merrors.FailedPrecondition("replacement database credential expired during validation") + } + success = true + return &generation{client: client, lease: lease, drained: make(chan struct{})}, nil +} + +func (m *Manager) retireLocked(g *generation) { + g.retired = true + if g.refs == 0 { + close(g.drained) + } + m.draining.Add(1) + go func() { + defer m.draining.Done() + <-g.drained + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := g.client.Disconnect(ctx); err != nil { + m.logger.Warn("Retired database pool close failed") + } + m.revoke(g.lease) + m.logger.Info("Retired database pool drained") + }() +} + +func (m *Manager) revoke(lease Lease) { + if lease.Revoke == nil { + return + } + // Expiration is the backstop when Vault is unavailable during bounded cleanup. + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := lease.Revoke(ctx); err != nil { + m.logger.Warn("Database credential cleanup failed; Vault expiry will revoke it") + } + +} + +func (m *Manager) run(ctx context.Context) { + defer close(m.workerDone) + m.mu.Lock() + wait := time.Until(m.current.lease.ExpiresAt) / 2 + m.mu.Unlock() + backoff := 250 * time.Millisecond + failureReported, expiryReported := false, false + for { + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + next, err := m.prepare(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + m.mu.Lock() + remaining := time.Until(m.current.lease.ExpiresAt) + m.mu.Unlock() + if !failureReported { + m.logger.Warn("Database credential refresh failed; retaining current pool", zap.Duration("valid_for", max(remaining, 0))) + failureReported = true + } + if remaining <= 0 && !expiryReported { + m.logger.Error("Database credentials expired; rejecting new operations until refresh succeeds") + expiryReported = true + } + wait = backoff + if remaining > 0 { + wait = max(100*time.Millisecond, min(wait, remaining/4)) + } + backoff = min(5*time.Second, backoff*2) + + continue + } + m.mu.Lock() + if m.closed { + m.retireLocked(next) + m.mu.Unlock() + return + } + old := m.current + m.current = next + m.retireLocked(old) + m.mu.Unlock() + failureReported, expiryReported = false, false + backoff = 250 * time.Millisecond + m.logger.Info("Database pool replaced without restarting service") + wait = time.Until(next.lease.ExpiresAt) / 2 + } +} + +func (m *Manager) Close(ctx context.Context) error { + m.mu.Lock() + if !m.closed { + m.closed = true + m.cancel() + m.retireLocked(m.current) + } + m.mu.Unlock() + done := make(chan struct{}) + go func() { <-m.workerDone; m.draining.Wait(); close(done) }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/backend/shared/database/source/source.go b/backend/shared/database/source/source.go new file mode 100644 index 0000000..b372796 --- /dev/null +++ b/backend/shared/database/source/source.go @@ -0,0 +1,30 @@ +// Package source owns Mongo connection generations and operation lifetimes. +package source + +import ( + "context" + + merrors "git.sendico.io/tech/sxchnge/shared/errors" + "go.mongodb.org/mongo-driver/v2/mongo" +) + +// Source pins one database/client until release. The returned context must be +// passed to nested operations so transactions never cross client generations. +type Source interface { + Acquire(context.Context) (*mongo.Database, context.Context, func(), error) +} + +type fixed struct{ database *mongo.Database } + +// Fixed adapts an externally owned database, e.g. a test fixture. It never closes it. +func Fixed(database *mongo.Database) Source { return &fixed{database: database} } + +func (s *fixed) Acquire(ctx context.Context) (*mongo.Database, context.Context, func(), error) { + if s.database == nil { + return nil, ctx, nil, merrors.FailedPrecondition("database is not configured") + } + if session := mongo.SessionFromContext(ctx); session != nil && session.Client() != s.database.Client() { + return nil, ctx, nil, merrors.FailedPrecondition("transaction belongs to another database client") + } + return s.database, ctx, func() {}, nil +} diff --git a/backend/shared/database/template/template.go b/backend/shared/database/template/template.go index 93b02cc..4dd10bd 100644 --- a/backend/shared/database/template/template.go +++ b/backend/shared/database/template/template.go @@ -6,12 +6,12 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/record" "git.sendico.io/tech/sxchnge/shared/database/repository" "git.sendico.io/tech/sxchnge/shared/database/repository/builder" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/errors" "git.sendico.io/tech/sxchnge/shared/logging" "git.sendico.io/tech/sxchnge/shared/logging/fields" "git.sendico.io/tech/sxchnge/shared/service" "go.mongodb.org/mongo-driver/v2/bson" - "go.mongodb.org/mongo-driver/v2/mongo" ) type CascadeDeleterT = func(ctx context.Context, objectRef bson.ObjectID) error @@ -88,11 +88,11 @@ func (db *DBImp[T]) DeleteCascade(ctx context.Context, objectRef bson.ObjectID) func Create[T storable.Storable]( logger mlogger.Logger, collection mservice.Type, - db *mongo.Database, + db source.Source, ) *DBImp[T] { res := &DBImp[T]{ Logger: logger.Named(collection), - Repository: repository.CreateMongoRepository(db, collection), + Repository: repository.CreateRepository(db, collection), } res.cdeleter = res.deleteCascadeStub return res diff --git a/backend/shared/database/vault/issuer.go b/backend/shared/database/vault/issuer.go new file mode 100644 index 0000000..73e8d88 --- /dev/null +++ b/backend/shared/database/vault/issuer.go @@ -0,0 +1,140 @@ +// Package vault issues independently owned MongoDB leases through Vault AppRole. +package vault + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/url" + "os" + "regexp" + "strings" + "time" + + "git.sendico.io/tech/sxchnge/shared/database/source" + merrors "git.sendico.io/tech/sxchnge/shared/errors" +) + +type Config struct { + Address string `mapstructure:"address"` + DatabaseRole string `mapstructure:"role"` + RoleIDFile string `mapstructure:"role_id_file"` + SecretIDFile string `mapstructure:"secret_id_file"` +} +type Issuer struct { + config Config + uri *url.URL + http *http.Client +} + +func New(config Config, uri string) (*Issuer, error) { + endpoint, err := url.Parse(config.Address) + if err != nil || endpoint.Host == "" || (endpoint.Scheme != "http" && endpoint.Scheme != "https") || endpoint.User != nil || endpoint.RawQuery != "" || endpoint.Fragment != "" { + return nil, merrors.InvalidArgument("invalid Vault address") + } + if !regexp.MustCompile(`^[a-zA-Z0-9_-]+$`).MatchString(config.DatabaseRole) || config.RoleIDFile == "" || config.SecretIDFile == "" { + return nil, merrors.InvalidArgument("Vault database role and AppRole files are required") + } + database, err := url.Parse(uri) + if err != nil || database.Scheme != "mongodb" || database.Host == "" || database.User != nil { + return nil, merrors.InvalidArgument("Vault requires a MongoDB URI without embedded credentials") + } + return &Issuer{config: config, uri: database, http: &http.Client{Timeout: 10 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}}, nil +} + +type response struct { + Auth struct { + Token string `json:"client_token"` + TTL int `json:"lease_duration"` + } `json:"auth"` + TTL int `json:"lease_duration"` + LeaseID string `json:"lease_id"` + Data struct { + Username string `json:"username"` + Password string `json:"password"` + } `json:"data"` +} + +func (i *Issuer) request(ctx context.Context, path, token string, body any) (response, error) { + var out response + var input io.Reader + method := http.MethodGet + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return out, merrors.InternalWrap(err, "encode Vault request") + } + input = bytes.NewReader(raw) + method = http.MethodPost + } + req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(i.config.Address, "/")+"/v1/"+path, input) + if err != nil { + return out, merrors.OpaqueWrap(merrors.ErrInvalidArg, err, "create Vault request") + } + req.Header.Set("Content-Type", "application/json") + if token != "" { + req.Header.Set("X-Vault-Token", token) + } + res, err := i.http.Do(req) + if err != nil { + return out, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "Vault request failed") + } + defer res.Body.Close() + if res.StatusCode < 200 || res.StatusCode >= 300 { + return out, merrors.ServiceUnavailable("Vault request was rejected") + } + if res.StatusCode == http.StatusNoContent { + return out, nil + } + if err = json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(&out); err != nil { + return out, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "invalid Vault response") + } + return out, nil +} + +func (i *Issuer) Issue(ctx context.Context) (source.Lease, error) { + var lease source.Lease + role, err := os.ReadFile(i.config.RoleIDFile) + if err != nil { + return lease, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "read Vault AppRole identity") + } + secret, err := os.ReadFile(i.config.SecretIDFile) + if err != nil { + return lease, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "read Vault AppRole credential") + } + started := time.Now() + login, err := i.request(ctx, "auth/approle/login", "", map[string]string{"role_id": strings.TrimSpace(string(role)), "secret_id": strings.TrimSpace(string(secret))}) + if err != nil { + return lease, err + } + if login.Auth.Token == "" { + return lease, merrors.FailedPrecondition("Vault returned no application identity") + } + // One token owns one database lease. Revoking it never touches Agent's token + // or the replacement generation, and requires only the default self policy. + revoke := func(ctx context.Context) error { + _, err := i.request(ctx, "auth/token/revoke-self", login.Auth.Token, map[string]any{}) + return err + } + if login.Auth.TTL <= 0 { + cleanup, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = revoke(cleanup) + return lease, merrors.FailedPrecondition("Vault returned no bounded identity lifetime") + } + credential, err := i.request(ctx, "database/creds/"+i.config.DatabaseRole, login.Auth.Token, nil) + if err != nil || credential.Data.Username == "" || credential.Data.Password == "" || credential.TTL <= 0 || credential.LeaseID == "" { + cleanup, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = revoke(cleanup) + if err != nil { + return lease, err + } + return lease, merrors.FailedPrecondition("Vault returned incomplete database credentials") + } + uri := *i.uri + uri.User = url.UserPassword(credential.Data.Username, credential.Data.Password) + return source.Lease{URI: uri.String(), ExpiresAt: started.Add(time.Duration(min(login.Auth.TTL, credential.TTL)) * time.Second), Revoke: revoke}, nil +} diff --git a/backend/shared/database/vault/issuer_test.go b/backend/shared/database/vault/issuer_test.go new file mode 100644 index 0000000..3e08815 --- /dev/null +++ b/backend/shared/database/vault/issuer_test.go @@ -0,0 +1,79 @@ +package vault + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestIssuerLifetimeEncodingAndCleanup(t *testing.T) { + for _, test := range []struct { + name string + identityTTL, databaseTTL int + reject bool + wantError bool + }{ + {"identity limits lease", 4, 20, false, false}, + {"database limits lease", 20, 4, false, false}, + {"invalid identity", 0, 20, false, true}, + {"invalid database lifetime", 20, 0, false, true}, + {"rejected issuance", 20, 20, true, true}, + } { + t.Run(test.name, func(t *testing.T) { + var revoked atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/auth/approle/login": + _ = json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{"client_token": "private-token", "lease_duration": test.identityTTL}}) + case "/v1/database/creds/bff": + if test.reject { + http.Error(w, "private-password", 403) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"lease_id": "lease", "lease_duration": test.databaseTTL, "data": map[string]string{"username": "user@host", "password": "p@ss:/?#"}}) + case "/v1/auth/token/revoke-self": + if r.Header.Get("X-Vault-Token") == "private-token" { + revoked.Store(true) + } + w.WriteHeader(204) + default: + w.WriteHeader(404) + } + })) + defer server.Close() + directory := t.TempDir() + identity, secret := filepath.Join(directory, "identity"), filepath.Join(directory, "secret") + require.NoError(t, os.WriteFile(identity, []byte("identity"), 0600)) + require.NoError(t, os.WriteFile(secret, []byte("secret"), 0600)) + issuer, err := New(Config{Address: server.URL, DatabaseRole: "bff", RoleIDFile: identity, SecretIDFile: secret}, "mongodb://localhost/db?authSource=admin") + require.NoError(t, err) + started := time.Now() + lease, err := issuer.Issue(context.Background()) + if test.wantError { + require.Error(t, err) + require.NotContains(t, err.Error(), "private-") + require.True(t, revoked.Load()) + return + } + require.NoError(t, err) + require.WithinDuration(t, started.Add(4*time.Second), lease.ExpiresAt, time.Second) + parsed, err := url.Parse(lease.URI) + require.NoError(t, err) + require.Equal(t, "user@host", parsed.User.Username()) + password, _ := parsed.User.Password() + require.Equal(t, "p@ss:/?#", password) + require.False(t, revoked.Load()) + require.NoError(t, lease.Revoke(context.Background())) + require.True(t, revoked.Load()) + }) + } +} diff --git a/backend/shared/ratelimit/store/db.go b/backend/shared/ratelimit/store/db.go index 7069e54..a45cd87 100644 --- a/backend/shared/ratelimit/store/db.go +++ b/backend/shared/ratelimit/store/db.go @@ -11,9 +11,9 @@ import ( "git.sendico.io/tech/sxchnge/shared/database/record" "git.sendico.io/tech/sxchnge/shared/database/repository" ri "git.sendico.io/tech/sxchnge/shared/database/repository/index" + "git.sendico.io/tech/sxchnge/shared/database/source" "git.sendico.io/tech/sxchnge/shared/errors" "git.sendico.io/tech/sxchnge/shared/logging" - "go.mongodb.org/mongo-driver/v2/mongo" "go.uber.org/zap" ) @@ -55,7 +55,7 @@ type DB struct { logger mlogger.Logger } -func Create(logger mlogger.Logger, database *mongo.Database, collection string) (*DB, error) { +func Create(logger mlogger.Logger, database source.Source, collection string) (*DB, error) { if database == nil { return nil, merrors.InvalidArgument("database is required", "database") } @@ -64,7 +64,7 @@ func Create(logger mlogger.Logger, database *mongo.Database, collection string) collection = defaultCollection } db := &DB{ - repo: repository.CreateMongoRepository(database, collection), + repo: repository.CreateRepository(database, collection), logger: logger.Named("rate_limits"), } if err := db.ensureIndexes(); err != nil { diff --git a/dev/agents.py b/dev/agents.py index 70e71fe..5fecdf7 100644 --- a/dev/agents.py +++ b/dev/agents.py @@ -14,10 +14,6 @@ def config(service): if service == 'nats': env['NATS_TOKEN'] = kv('nats', 'token') else: - database = 'sxchnge' if service == 'bff' else 'sxchnge_notifications' - env['SXCHNGE_MONGO_URI'] = ('{{ with secret "database/creds/' + service + '" }}' - 'mongodb://{{ .Data.username | urlquery }}:{{ .Data.password | urlquery }}' - '@mongo:27017/' + database + '?authSource=admin&replicaSet=rs0{{ end }}') env['TLS_CA'] = kv('tls', 'ca') if service == 'bff': env['SXCHNGE_JWT_SECRET'] = kv('bff', 'jwt') diff --git a/dev/config/bff.yml b/dev/config/bff.yml index c2b493c..ca914db 100644 --- a/dev/config/bff.yml +++ b/dev/config/bff.yml @@ -6,7 +6,12 @@ http_server: database: driver: mongodb settings: - uri_env: SXCHNGE_MONGO_URI + uri: mongodb://mongo:27017/sxchnge?authSource=admin&replicaSet=rs0 + vault: + address: http://vault:8200 + role: bff + role_id_file: /etc/agent/role-id + secret_id_file: /etc/agent/secret-id database: sxchnge api: middleware: diff --git a/dev/config/notifications.yml b/dev/config/notifications.yml index 687274a..a5a3882 100644 --- a/dev/config/notifications.yml +++ b/dev/config/notifications.yml @@ -4,7 +4,12 @@ tls_key: "/run/sxchnge/server.key" database: driver: mongodb settings: - uri_env: SXCHNGE_MONGO_URI + uri: mongodb://mongo:27017/sxchnge_notifications?authSource=admin&replicaSet=rs0 + vault: + address: http://vault:8200 + role: notifications + role_id_file: /etc/agent/role-id + secret_id_file: /etc/agent/secret-id database: sxchnge_notifications messaging: driver: NATS diff --git a/dev/rotation.py b/dev/rotation.py new file mode 100644 index 0000000..8a37c46 --- /dev/null +++ b/dev/rotation.py @@ -0,0 +1,92 @@ +"""Exercise real Vault database rotation under BFF HTTP transaction traffic. + +Run after make build/bootstrap. No Resend credential or live email is needed. +Temporarily shortens the BFF database role TTL, restores it on every exit, and +preserves whether BFF/frontend were running before this check. +""" +import concurrent.futures +import json +import ssl +import threading +import time +import urllib.error +import urllib.request +import uuid + +from bootstrap import compose +from vault import STATE, Vault + + +def verify_rotation(): + vault = Vault() + vault.existing() + role = 'database/roles/bff' + original = vault.call(role)['data'] + running = {json.loads(line)['Service'] for line in compose('ps', '--format', 'json', capture=True).splitlines()} + stop = threading.Event() + successes = [] + failures = [] + lock = threading.Lock() + context = ssl.create_default_context(cafile=str(STATE / 'ca.pem')) + + def requests(): + while not stop.is_set(): + # Exactly 43 characters invokes the activation transaction and token + # lookup, then returns the expected invalid-token response. No user + # or notification is created and the test never sends an email. + token = uuid.uuid4().hex + uuid.uuid4().hex[:11] + request = urllib.request.Request('https://localhost:5126/api/v1/accounts/verification/confirm', + data=json.dumps({'token': token}).encode(), headers={ + 'Content-Type': 'application/json', 'Origin': 'https://localhost:5126', + 'X-Auth-Transport': 'cookie', 'X-CSRF-Protection': '1'}) + try: + with urllib.request.urlopen(request, context=context, timeout=5) as response: + code = response.status + body = response.read() + except urllib.error.HTTPError as error: + code = error.code + body = error.read() + except Exception: + code = 0 + body = b'' + with lock: + (successes if code == 400 and b'Verification is invalid or already used' in body else failures).append(code) + stop.wait(0.03) + + try: + vault.call(role, dict(original, default_ttl='12s', max_ttl='24s')) + compose('up', '-d', '--no-deps', '--force-recreate', '--wait', '--wait-timeout', '90', 'bff', 'frontend') + process = compose('exec', '-T', 'bff', 'pidof', 'service', capture=True).strip() + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + workers = [pool.submit(requests) for _ in range(4)] + try: + deadline = time.monotonic() + 40 + while time.monotonic() < deadline: + time.sleep(1) + current = compose('exec', '-T', 'bff', 'pidof', 'service', capture=True).strip() + if current != process: + raise RuntimeError('BFF process restarted during credential rotation') + logs = compose('logs', '--no-color', 'bff', capture=True) + if logs.count('Database pool replaced without restarting service') >= 3: + break + else: + raise RuntimeError('Three database pool rotations were not observed') + finally: + stop.set() + for worker in workers: + worker.result() + if failures or len(successes) < 100: + raise RuntimeError(f'Rotation traffic check: {len(successes)} expected responses, {len(failures)} failures') + print(f'PASS: three real Vault rotations, unchanged BFF PID, {len(successes)} transaction-backed HTTP requests, zero failures') + finally: + stop.set() + vault.call(role, original) + newly_started = [name for name in ['frontend', 'bff'] if name not in running] + if newly_started: + compose('stop', *newly_started) + compose('rm', '-f', *newly_started) + # Existing services stay online; their next refresh picks up normal TTLs. + + +if __name__ == '__main__': + verify_rotation() diff --git a/dev/test_local.py b/dev/test_local.py index 4af9e93..3a60194 100644 --- a/dev/test_local.py +++ b/dev/test_local.py @@ -4,6 +4,7 @@ import tempfile import unittest from unittest.mock import patch +import agents import bootstrap import images from vault import Vault @@ -33,6 +34,14 @@ class Releases(unittest.TestCase): self.assertEqual(target.read_text(), 'previous build') +class AgentContract(unittest.TestCase): + def test_database_lease_does_not_trigger_process_restart(self): + for service in ['bff', 'notifications']: + config = agents.config(service) + self.assertNotIn('database/creds/', config) + self.assertNotIn('SXCHNGE_MONGO_URI', config) + + class Bootstrap(unittest.TestCase): def test_missing_resend_credentials_stops_before_product_start(self): with tempfile.TemporaryDirectory() as temporary: diff --git a/docs/implementation.md b/docs/implementation.md index 68d3004..e2deefa 100644 --- a/docs/implementation.md +++ b/docs/implementation.md @@ -601,8 +601,9 @@ commands, current-stable backend test images and repository container policy. committed. Root `AGENTS.md` now records the stable-container policy. - Bootstrap configures Vault KV/PKI/AppRole and `mongodb-database-plugin`, rotates its Mongo manager credential and issues separate database roles for the BFF - and notifications. Vault Agent renews credentials and restarts child processes - on changes. Existing shared database/messaging paths remain intact. + and notifications. The shared database wrapper replaces credential pools in + process and drains transactions/cursors before revoking retired credentials. + Agent handles non-database secrets. Shared messaging paths remain intact. - Resend is the default. The user explicitly chose to provision its API key and verified sender manually after bootstrap. `make up` leaves healthy infrastructure running and intentionally exits nonzero before product startup while those @@ -622,8 +623,8 @@ The initial image build exposed missing protobuf headers; installing Additional live checks passed: a down/up cycle preserved a Mongo marker and Vault secrets and automatically unsealed Vault; changing/restoring the NATS token -restarted the NATS child; a temporary 10-second/20-second BFF database role proved -replacement credentials restart a healthy BFF (normal TTLs restored afterward). +restarted the NATS child; the initial restart-based database credential lifecycle was superseded by +the uninterrupted pool rotation implementation described below. Temporary BFF/frontend containers served the web bundle and proxied BFF health successfully over HTTPS verified with the generated CA; those test containers were removed. The notifications image rendered its Agent configuration and exited @@ -636,3 +637,42 @@ NATS remain running. No optional security scan or merge was performed. See the [local development guide](local-development.md) for exact provisioning fields, certificate trust, restart behavior, local bootstrap-material handling and verification commands. + +### Uninterrupted Mongo credential rotation + +BFF and notifications now acquire database operation scopes from the shared +wrapper. A replacement is issued at half the effective token/lease lifetime, +validated before publication, and independently revoked after draining. Whole +transactions and cursors retain their original pool. Failed refreshes retry while +the current credential remains valid. Expiry rejects new operations until recovery; +there is no scheduled service restart. Non-database Agent template changes still +restart their child process. This is planned-rotation continuity, not infrastructure +high availability or a guarantee beyond Vault credential expiry. + +The repository and store constructor contract now uses `source.Source` rather +than a concrete Mongo database. Static test integrations use `source.Fixed`. +Application code continues through shared repositories and transaction factories. +The new `make test-dev-rotation` command proves real Vault plugin replacements +under BFF transaction traffic and checks process identity. It restores shortened +role TTLs and removes temporary product containers afterward. + +Validation of the revision: `make build`, `make test`, `make lint-backend`, +`make test-dev-infra` (six setup tests and live checks), and +`make test-dev-rotation` passed. The live rotation check observed three replacements +with an unchanged BFF PID and 1,155 expected invalid-token HTTP responses, zero +failures. Product integration-tag packages also compiled. The race-enabled +rotation suite covers active transactions/cursors, rejected credentials, transient +and extended outages, automatic recovery, rollback and shutdown; issuer tests +cover effective lifetime, credential escaping and failed-issuance cleanup. + +An additional `go test -race -tags integration ./database/internal/repository/... +./database/vault/...` run in `backend/shared` found two existing test issues: the +partial-index test expects nested BSON maps instead of ordered documents, and the +patch timestamp assertion can require a strict increase within the same Mongo +millisecond. The BSON mismatch reproduces against `origin/main`; five isolated baseline +repetitions of the timestamp test pass, consistent with its timing-dependent +assertion. Neither assertion nor the timestamp update logic changes in this task; +the supplementary failures are recorded and are not suppressed. The required +normal test and lint commands pass. No optional security scan or email delivery +was performed. Resend remains intentionally unconfigured; infrastructure stays up +and product startup awaits manual `kv/email` provisioning. diff --git a/docs/local-development.md b/docs/local-development.md index ff108ca..b4b1937 100644 --- a/docs/local-development.md +++ b/docs/local-development.md @@ -108,16 +108,39 @@ and are retained in `kv/bootstrap/mongo` for local provisioning. The BFF role has `readWrite` only on `sxchnge`; notifications has `readWrite` only on `sxchnge_notifications`. Mongo URIs come from dynamic Vault credentials, not static application users. Leases default to one hour with a 24-hour maximum. -Vault Agent renews leases, obtains replacement credentials, and restarts its child -process when credentials change. Apps retain their existing shared database and -messaging connection wrappers. No new direct application database path is added. +The shared database wrapper obtains credentials directly from Vault using the +service's AppRole. At half the effective lifetime (the shorter of the database +lease and owning token TTL), it authenticates independently and prepares a new +Mongo pool. Only a successful primary ping permits publication. New operations +use that pool; active operations, cursors and complete transactions, including +nested stores and commit retries, retain the old pool until they finish. The old +pool then closes and its dedicated token is revoked. Mongo credential replacement +never restarts BFF or notifications. Each generation has a separate token so +cleanup cannot revoke the replacement or Vault Agent's identity. + +Refresh failures keep the current valid pool and retry with bounded backoff. +Operations are bounded by credential expiry. If Vault remains unavailable beyond +that deadline, new database operations fail closed; refresh continues and access +recovers automatically without restarting the service. Administrative revocation, +operations exceeding their lease and infrastructure failures cannot be made +uninterrupted by pool rotation. Shutdown rejects new scopes and drains existing +work. Failed cleanup relies on Vault expiration as a backstop. + +Repositories and transaction factories share `source.Source` rather than retaining +concrete Mongo collections. Low-level consumers must acquire a scope, use its +returned context and release it after all work completes; never retain its client +or context beyond release. `source.Fixed` adapts externally owned test/static +connections. Database credentials are no longer Agent environment templates; +Agent continues to fetch non-database secrets and may restart children when those +secrets change. Each agent authenticates using its own AppRole with a renewable periodic token. Local SecretIDs are reusable for container restarts; each container mounts only its own identity directory under `.local/dev/agents/`. Revoking an AppRole SecretID requires removing its matching local `secret-id` file and rerunning `make up` to issue a replacement. Revoking an active token/lease terminates its credentials; -Agent's retries and the container restart policy recover when Vault is reachable. +The database wrapper retries with a fresh identity; Agent independently retries +its non-database secrets when Vault is reachable. The notification gRPC connection uses a Vault-issued certificate for `notifications` and the BFF validates the Vault CA and hostname. Frontend HTTPS @@ -147,6 +170,7 @@ make logs make down # preserves named volumes and local bootstrap material make up # idempotent provisioning and readiness checks make test-dev-infra # works before Resend provisioning +make test-dev-rotation # real Vault rotations under BFF HTTP traffic make test-dev # requires all six services running ``` @@ -154,6 +178,15 @@ The infrastructure check exercises AppRole access restrictions, real Vault MongoDB credential issuance, per-database access, lease renewal and user deletion on revocation. The complete check also verifies service health and frontend/API HTTPS with the generated CA. These checks never send a real email. +`test-dev-rotation` temporarily sets a 12-second BFF lease, deploys the built BFF +and frontend, and requires three pool replacements with unchanged process ID and +zero unexpected HTTP responses. Its invalid-token requests exercise Mongo-backed +transactions without sending email. The check restores the role TTL and removes +services it started; it recreates BFF/frontend at the beginning, so run it as a +development validation command, not against active work you need to preserve. +Race-enabled backend tests additionally hold a transaction and cursor across a +replacement and exercise failed candidate validation, outage recovery, rollback, +expiry rejection and graceful shutdown. Backend regression tests use fresh `mongo:latest` and `nats:latest` official stable images, pulled before `make test-backend`. Normal validation remains