Jeff Mitchell
7364d84835
* Add mechanism for plugin to normalize attributes This introduces a new normalization call to the plugin RPC service. If implemented, the plugin will be given the attributes prior to saving them in a create/update call, and can perform any normalization needed, e.g. ensuring that a parameter passed in is represented in a canonical way. As an example, the CLI allows a user of the AWS plugin to pass in a string, but canonically it is a string slice, and should be returned that way as well. Not by chance, this is the exact behavior of the testing implementation of this function. This does not update any individual plugin, but rather sets the controller-side logic up for plugins to support this functionality.
801 lines
28 KiB
Go
801 lines
28 KiB
Go
package plugin
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/hashicorp/boundary/internal/db"
|
|
"github.com/hashicorp/boundary/internal/errors"
|
|
"github.com/hashicorp/boundary/internal/host"
|
|
"github.com/hashicorp/boundary/internal/kms"
|
|
"github.com/hashicorp/boundary/internal/libs/patchstruct"
|
|
"github.com/hashicorp/boundary/internal/observability/event"
|
|
"github.com/hashicorp/boundary/internal/oplog"
|
|
hostplugin "github.com/hashicorp/boundary/internal/plugin/host"
|
|
"github.com/hashicorp/boundary/internal/scheduler"
|
|
"github.com/hashicorp/boundary/internal/util"
|
|
pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostcatalogs"
|
|
pbset "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/hostsets"
|
|
plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
|
|
"github.com/mr-tron/base58"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/types/known/structpb"
|
|
"google.golang.org/protobuf/types/known/wrapperspb"
|
|
)
|
|
|
|
// normalizeCatalogAttributes allows a plugin to normalize attributes before
|
|
// they are saved
|
|
func normalizeCatalogAttributes(ctx context.Context, plgClient plgpb.HostPluginServiceClient, plgHc *pb.HostCatalog) error {
|
|
const op = "plugin.(Repository).normalizeCatalogAttributes"
|
|
switch {
|
|
case util.IsNil(plgClient):
|
|
return errors.New(ctx, errors.InvalidParameter, op, "plugin client is nil")
|
|
case plgHc == nil:
|
|
return errors.New(ctx, errors.InvalidParameter, op, "host catalog is nil")
|
|
case plgHc.GetAttributes() == nil:
|
|
return nil
|
|
}
|
|
|
|
ret, err := plgClient.NormalizeCatalogData(ctx, &plgpb.NormalizeCatalogDataRequest{
|
|
Attributes: plgHc.GetAttributes(),
|
|
})
|
|
switch {
|
|
case err == nil:
|
|
if ret.Attributes != nil {
|
|
plgHc.Attrs = &pb.HostCatalog_Attributes{
|
|
Attributes: ret.Attributes,
|
|
}
|
|
}
|
|
case status.Code(err) == codes.Unimplemented:
|
|
// Do nothing
|
|
default:
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("error asking plugin to normalize catalog data"))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// CreateCatalog inserts c into the repository and returns a new
|
|
// HostCatalog containing the catalog's PublicId. c must contain a valid
|
|
// ScopeID and PluginID. c must not contain a PublicId. The PublicId is
|
|
// generated and assigned by this method. opt is ignored.
|
|
//
|
|
// c.Secret, c.Name and c.Description are optional. If c.Name is set, it must be
|
|
// unique within c.ScopeID. If c.Secret is set, it will be stored encrypted but
|
|
// not included in the returned *HostCatalog.
|
|
//
|
|
// Both c.CreateTime and c.UpdateTime are ignored.
|
|
func (r *Repository) CreateCatalog(ctx context.Context, c *HostCatalog, _ ...Option) (*HostCatalog, *hostplugin.Plugin, error) {
|
|
const op = "plugin.(Repository).CreateCatalog"
|
|
if c == nil {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "nil HostCatalog")
|
|
}
|
|
if c.HostCatalog == nil {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "nil embedded HostCatalog")
|
|
}
|
|
if c.ScopeId == "" {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no scope id")
|
|
}
|
|
if c.PublicId != "" {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "public id not empty")
|
|
}
|
|
if c.PluginId == "" {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no plugin id")
|
|
}
|
|
if c.Attributes == nil {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "nil attributes")
|
|
}
|
|
c = c.clone()
|
|
id, err := newHostCatalogId(ctx)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
c.PublicId = id
|
|
|
|
// Use PatchBytes' functionality that does not add keys where the values
|
|
// are nil to the resulting struct since we do not want to store nil valued
|
|
// attributes.
|
|
c.Attributes, err = patchstruct.PatchBytes([]byte{}, c.Attributes)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
// If secrets were passed in, HMAC 'em
|
|
if c.Secrets != nil && len(c.Secrets.GetFields()) > 0 {
|
|
databaseWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeDatabase)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get database wrapper"))
|
|
}
|
|
if err := c.hmacSecrets(ctx, databaseWrapper); err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("error hmac'ing passed-in secrets"))
|
|
}
|
|
}
|
|
|
|
plgHc, err := toPluginCatalog(ctx, c)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
plgClient, ok := r.plugins[c.GetPluginId()]
|
|
if !ok || plgClient == nil {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("plugin %q not available", c.GetPluginId()))
|
|
}
|
|
|
|
if plgHc.GetAttributes() != nil {
|
|
if err := normalizeCatalogAttributes(ctx, plgClient, plgHc); err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
|
|
oplogWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeOplog)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
|
|
}
|
|
|
|
// If the call to the plugin succeeded, we do not want to call it again if
|
|
// the transaction failed and is being retried.
|
|
var pluginCalledSuccessfully bool
|
|
var plgResp *plgpb.OnCreateCatalogResponse
|
|
|
|
var newHostCatalog *HostCatalog
|
|
_, err = r.writer.DoTx(
|
|
ctx,
|
|
db.StdRetryCnt,
|
|
db.ExpBackoff{},
|
|
func(_ db.Reader, w db.Writer) error {
|
|
msgs := make([]*oplog.Message, 0, 3)
|
|
ticket, err := w.GetTicket(ctx, c)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get ticket"))
|
|
}
|
|
|
|
newHostCatalog = c.clone()
|
|
var cOplogMsg oplog.Message
|
|
if err := w.Create(ctx, newHostCatalog, db.NewOplogMsg(&cOplogMsg)); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
msgs = append(msgs, &cOplogMsg)
|
|
|
|
if !pluginCalledSuccessfully {
|
|
plgResp, err = plgClient.OnCreateCatalog(ctx, &plgpb.OnCreateCatalogRequest{Catalog: plgHc})
|
|
if err != nil {
|
|
if status.Code(err) != codes.Unimplemented {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
pluginCalledSuccessfully = true
|
|
}
|
|
|
|
if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
|
|
hcSecret, err := newHostCatalogSecret(ctx, id, plgResp.GetPersisted().GetSecrets())
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
dbWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeDatabase)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
|
|
}
|
|
if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if hcSecret != nil {
|
|
newSecret := hcSecret.clone()
|
|
var sOplogMsg oplog.Message
|
|
if err := w.Create(ctx, newSecret, db.NewOplogMsg(&sOplogMsg)); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
msgs = append(msgs, &sOplogMsg)
|
|
}
|
|
}
|
|
|
|
metadata := c.oplog(oplog.OpType_OP_TYPE_CREATE)
|
|
if err := w.WriteOplogEntryWith(ctx, oplogWrapper, ticket, metadata, msgs); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog"))
|
|
}
|
|
return nil
|
|
},
|
|
)
|
|
|
|
if err != nil {
|
|
if errors.IsUniqueError(err) {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("in scope: %s: name %s already exists", c.ScopeId, c.Name)))
|
|
}
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("in scope: %s", c.ScopeId)))
|
|
}
|
|
plg, err := r.getPlugin(ctx, newHostCatalog.GetPluginId())
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
return newHostCatalog, plg, nil
|
|
}
|
|
|
|
// UpdateCatalog updates the repository entry for c.PublicId with the
|
|
// values in c for the fields listed in fieldMask. It returns a new
|
|
// HostCatalog containing the updated values and a count of the number of
|
|
// records updated. c is not changed.
|
|
//
|
|
// c must contain a valid PublicId. c.Name, c.Description, and
|
|
// c.Attributes can be updated; if c.Secrets is present, its contents
|
|
// are sent to the plugin (along with any other changes, see below)
|
|
// before the update is sent to the database.
|
|
//
|
|
// An attribute of c will be set to NULL in the database if the
|
|
// attribute in c is the zero value and it is included in fieldMask.
|
|
// Note that this does not apply to c.Attributes - a null
|
|
// c.Attributes is a no-op for modifications. Rather, if fields need
|
|
// to be reset, its field in c.Attributes should individually set to
|
|
// null.
|
|
//
|
|
// Updates are sent to OnUpdateCatalog with a full copy of both the
|
|
// current catalog, and the state of the new catalog should it be
|
|
// updated, along with any secrets included in the new request. This
|
|
// request may alter the returned persisted state. Update of the
|
|
// record in the database is aborted if this call fails.
|
|
func (r *Repository) UpdateCatalog(ctx context.Context, c *HostCatalog, version uint32, fieldMask []string, _ ...Option) (*HostCatalog, *hostplugin.Plugin, int, error) {
|
|
const op = "plugin.(Repository).UpdateCatalog"
|
|
if c == nil {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "nil HostCatalog")
|
|
}
|
|
if c.HostCatalog == nil {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "nil embedded HostCatalog")
|
|
}
|
|
if c.PublicId == "" {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "no public id")
|
|
}
|
|
if c.ScopeId == "" {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "no scope id")
|
|
}
|
|
if len(fieldMask) == 0 {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.EmptyFieldMask, op, "empty field mask")
|
|
}
|
|
|
|
// Get the old catalog first. We patch the record first before
|
|
// sending it to the DB for updating so that we can run on
|
|
// OnUpdateCatalog. Note that the field masks are still used for
|
|
// updating.
|
|
currentCatalog, currentCatalogPersisted, err := r.getCatalog(ctx, c.PublicId)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("error looking up catalog with id %q", c.PublicId)))
|
|
}
|
|
|
|
if currentCatalog == nil {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.RecordNotFound, op, fmt.Sprintf("catalog with id %q not found", c.PublicId))
|
|
}
|
|
|
|
// Assert the version of the current catalog to make sure we're
|
|
// updating the correct one.
|
|
if currentCatalog.GetVersion() != version {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.VersionMismatch, op, fmt.Sprintf("catalog version mismatch, want=%d, got=%d", currentCatalog.GetVersion(), version))
|
|
}
|
|
|
|
// Clone the catalog so that we can set fields.
|
|
newCatalog := currentCatalog.clone()
|
|
var updateAttributes bool
|
|
var dbMask, nullFields []string
|
|
var alreadySetSecrets bool
|
|
for _, f := range fieldMask {
|
|
switch {
|
|
case strings.EqualFold("name", f) && c.Name == "":
|
|
nullFields = append(nullFields, "name")
|
|
newCatalog.Name = c.Name
|
|
case strings.EqualFold("name", f) && c.Name != "":
|
|
dbMask = append(dbMask, "name")
|
|
newCatalog.Name = c.Name
|
|
case strings.EqualFold("description", f) && c.Description == "":
|
|
nullFields = append(nullFields, "description")
|
|
newCatalog.Description = c.Description
|
|
case strings.EqualFold("description", f) && c.Description != "":
|
|
dbMask = append(dbMask, "description")
|
|
newCatalog.Description = c.Description
|
|
case strings.EqualFold("attributes", strings.Split(f, ".")[0]):
|
|
// Flag attributes for updating. While multiple masks may be
|
|
// sent, we only need to do this once.
|
|
updateAttributes = true
|
|
case strings.EqualFold("secrets", strings.Split(f, ".")[0]):
|
|
if alreadySetSecrets {
|
|
continue
|
|
}
|
|
alreadySetSecrets = true
|
|
// While in a similar format, secrets are passed along
|
|
// wholesale (for the time being). Don't append this mask
|
|
// field, as secrets do not have a database entry.
|
|
newCatalog.Secrets = c.Secrets
|
|
switch {
|
|
case newCatalog.Secrets == nil,
|
|
len(newCatalog.Secrets.GetFields()) == 0:
|
|
nullFields = append(nullFields, "SecretsHmac")
|
|
default:
|
|
// If secrets were passed in, HMAC 'em
|
|
databaseWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeDatabase)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get database wrapper"))
|
|
}
|
|
if err := newCatalog.hmacSecrets(ctx, databaseWrapper); err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("error hmac'ing passed-in secrets"))
|
|
}
|
|
dbMask = append(dbMask, "SecretsHmac")
|
|
}
|
|
default:
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidFieldMask, op, fmt.Sprintf("invalid field mask: %s", f))
|
|
}
|
|
}
|
|
|
|
var needSetSync, runSyncJob bool
|
|
if updateAttributes {
|
|
dbMask = append(dbMask, "attributes")
|
|
newCatalog.Attributes, err = patchstruct.PatchBytes(newCatalog.Attributes, c.Attributes)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("error in catalog attribute JSON"))
|
|
}
|
|
|
|
// Flag for host sets under this catalog to be synced.
|
|
needSetSync = true
|
|
}
|
|
|
|
// Get the plugin for the host catalog - this is to return it back
|
|
// after the update is complete. We don't actually do anything with
|
|
// the record otherwise. Fetch it here so that if there's an
|
|
// integrity error, we don't call the plugin.
|
|
plg, err := r.getPlugin(ctx, currentCatalog.GetPluginId())
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
// Get the plugin client.
|
|
plgClient, ok := r.plugins[currentCatalog.GetPluginId()]
|
|
if !ok || plgClient == nil {
|
|
return nil, nil, db.NoRowsAffected, errors.New(ctx, errors.Internal, op, fmt.Sprintf("plugin %q not available", currentCatalog.GetPluginId()))
|
|
}
|
|
|
|
// Convert the catalog values to API protobuf values, which is what
|
|
// we use for the plugin hook calls.
|
|
currPlgHc, err := toPluginCatalog(ctx, currentCatalog)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
newPlgHc, err := toPluginCatalog(ctx, newCatalog)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
if updateAttributes {
|
|
if newPlgHc.GetAttributes() != nil {
|
|
if err := normalizeCatalogAttributes(ctx, plgClient, newPlgHc); err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
if newCatalog.Attributes, err = proto.Marshal(newPlgHc.GetAttributes()); err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
}
|
|
|
|
dbWrapper, err := r.kms.GetWrapper(ctx, newCatalog.ScopeId, kms.KeyPurposeDatabase)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
|
|
}
|
|
// Get the oplog.
|
|
oplogWrapper, err := r.kms.GetWrapper(ctx, newCatalog.ScopeId, kms.KeyPurposeOplog)
|
|
if err != nil {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
|
|
}
|
|
|
|
var pluginCalledSuccessfully bool
|
|
var plgResp *plgpb.OnUpdateCatalogResponse
|
|
|
|
var recordUpdated bool
|
|
var returnedCatalog *HostCatalog
|
|
_, err = r.writer.DoTx(
|
|
ctx,
|
|
db.StdRetryCnt,
|
|
db.ExpBackoff{},
|
|
func(_ db.Reader, w db.Writer) error {
|
|
msgs := make([]*oplog.Message, 0, 3)
|
|
ticket, err := w.GetTicket(ctx, newCatalog)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get ticket"))
|
|
}
|
|
|
|
if len(dbMask) != 0 || len(nullFields) != 0 {
|
|
returnedCatalog = newCatalog.clone()
|
|
var cOplogMsg oplog.Message
|
|
catalogsUpdated, err := w.Update(
|
|
ctx,
|
|
returnedCatalog,
|
|
dbMask,
|
|
nullFields,
|
|
db.NewOplogMsg(&cOplogMsg),
|
|
db.WithVersion(&version),
|
|
)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if catalogsUpdated != 1 {
|
|
return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog to be deleted, got %d", catalogsUpdated))
|
|
}
|
|
msgs = append(msgs, &cOplogMsg)
|
|
recordUpdated = true
|
|
} else {
|
|
// Returned catalog needs to be the old copy, as no fields in the
|
|
// catalog itself are being updated (note: secrets may still be
|
|
// updated).
|
|
returnedCatalog = currentCatalog.clone()
|
|
}
|
|
|
|
if !pluginCalledSuccessfully {
|
|
plgResp, err = plgClient.OnUpdateCatalog(ctx, &plgpb.OnUpdateCatalogRequest{
|
|
CurrentCatalog: currPlgHc,
|
|
NewCatalog: newPlgHc,
|
|
Persisted: currentCatalogPersisted,
|
|
})
|
|
if err != nil {
|
|
if status.Code(err) != codes.Unimplemented {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
pluginCalledSuccessfully = true
|
|
}
|
|
var updatedPersisted bool
|
|
if plgResp != nil && plgResp.GetPersisted().GetSecrets() != nil {
|
|
if len(plgResp.GetPersisted().GetSecrets().GetFields()) == 0 {
|
|
// Flag the secret to be deleted.
|
|
hcSecret, err := newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
// We didn't set/encrypt the persisted data because there was
|
|
// none returned. Just delete the entry.
|
|
deletedSecret := hcSecret.clone()
|
|
var sOplogMsg oplog.Message
|
|
secretsDeleted, err := w.Delete(ctx, deletedSecret, db.NewOplogMsg(&sOplogMsg))
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if secretsDeleted != 1 {
|
|
return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog secret to be deleted, got %d", secretsDeleted))
|
|
}
|
|
updatedPersisted = true
|
|
msgs = append(msgs, &sOplogMsg)
|
|
} else {
|
|
hcSecret, err := newHostCatalogSecret(ctx, currentCatalog.GetPublicId(), plgResp.GetPersisted().GetSecrets())
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if err := hcSecret.encrypt(ctx, dbWrapper); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
// Update the secrets.
|
|
updatedSecret := hcSecret.clone()
|
|
var sOplogMsg oplog.Message
|
|
if err := w.Create(
|
|
ctx,
|
|
updatedSecret,
|
|
db.WithOnConflict(&db.OnConflict{
|
|
Target: db.Columns{"catalog_id"},
|
|
Action: db.SetColumns([]string{"secret", "key_id"}),
|
|
}),
|
|
db.NewOplogMsg(&sOplogMsg),
|
|
); err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
updatedPersisted = true
|
|
msgs = append(msgs, &sOplogMsg)
|
|
}
|
|
}
|
|
|
|
if !recordUpdated && updatedPersisted {
|
|
// we only updated secrets, so we need to increment the
|
|
// version of the host catalog manually.
|
|
returnedCatalog = newCatalog.clone()
|
|
returnedCatalog.Version = uint32(version) + 1
|
|
var cOplogMsg oplog.Message
|
|
catalogsUpdated, err := w.Update(
|
|
ctx,
|
|
returnedCatalog,
|
|
[]string{"version"},
|
|
[]string{},
|
|
db.NewOplogMsg(&cOplogMsg),
|
|
db.WithVersion(&version),
|
|
)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if catalogsUpdated != 1 {
|
|
return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 catalog to be updated, got %d", catalogsUpdated))
|
|
}
|
|
msgs = append(msgs, &cOplogMsg)
|
|
recordUpdated = true
|
|
}
|
|
|
|
if needSetSync {
|
|
// We also need to mark all host sets in this catalog to be
|
|
// synced as well.
|
|
setsForCatalog, _, err := r.getSets(ctx, "", returnedCatalog.PublicId)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get sets for host catalog"))
|
|
}
|
|
|
|
for _, set := range setsForCatalog {
|
|
newSet := set.clone()
|
|
newSet.NeedSync = true
|
|
var msg oplog.Message
|
|
n, err := w.Update(ctx, newSet, []string{"NeedSync"}, []string{}, db.NewOplogMsg(&msg))
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update host set"))
|
|
}
|
|
|
|
if n > 1 {
|
|
return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected no more than 1 host set to be updated while flagging host set id %q for synchronization, got %d", newSet.PublicId, n))
|
|
}
|
|
|
|
msgs = append(msgs, &msg)
|
|
runSyncJob = true
|
|
}
|
|
}
|
|
|
|
if len(msgs) != 0 {
|
|
metadata := newCatalog.oplog(oplog.OpType_OP_TYPE_UPDATE)
|
|
if err := w.WriteOplogEntryWith(ctx, oplogWrapper, ticket, metadata, msgs); err != nil {
|
|
return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog"))
|
|
}
|
|
}
|
|
|
|
return nil
|
|
},
|
|
)
|
|
|
|
if err != nil {
|
|
if errors.IsUniqueError(err) {
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("in %s: name %s already exists", newCatalog.PublicId, newCatalog.Name)))
|
|
}
|
|
return nil, nil, db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("in %s", newCatalog.PublicId)))
|
|
}
|
|
|
|
if runSyncJob {
|
|
_ = r.scheduler.UpdateJobNextRunInAtLeast(ctx, setSyncJobName, 0, scheduler.WithRunNow(true))
|
|
}
|
|
|
|
// Even if we didn't update any records, if we were able to find the record
|
|
// with the appropriate version, returning 1 row updated is consistent with
|
|
// static's update catalog behavior.
|
|
numUpdated := 1
|
|
|
|
return returnedCatalog, plg, numUpdated, nil
|
|
}
|
|
|
|
// LookupCatalog returns the HostCatalog for id. Returns nil, nil if no
|
|
// HostCatalog is found for id.
|
|
func (r *Repository) LookupCatalog(ctx context.Context, id string, _ ...Option) (*HostCatalog, *hostplugin.Plugin, error) {
|
|
const op = "plugin.(Repository).LookupCatalog"
|
|
if id == "" {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no public id")
|
|
}
|
|
c, _, err := r.getCatalog(ctx, id)
|
|
if errors.IsNotFoundError(err) {
|
|
return nil, nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("failed for: %s", id)))
|
|
}
|
|
plg, err := r.getPlugin(ctx, c.GetPluginId())
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
return c, plg, nil
|
|
}
|
|
|
|
// ListCatalogs returns a slice of HostCatalogs for the scope IDs. WithLimit is the only option supported.
|
|
func (r *Repository) ListCatalogs(ctx context.Context, scopeIds []string, opt ...host.Option) ([]*HostCatalog, []*hostplugin.Plugin, error) {
|
|
const op = "plugin.(Repository).ListCatalogs"
|
|
if len(scopeIds) == 0 {
|
|
return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no scope id")
|
|
}
|
|
opts, err := host.GetOpts(opt...)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
limit := r.defaultLimit
|
|
if opts.WithLimit != 0 {
|
|
// non-zero signals an override of the default limit for the repo.
|
|
limit = opts.WithLimit
|
|
}
|
|
var hostCatalogs []*HostCatalog
|
|
if err := r.reader.SearchWhere(ctx, &hostCatalogs, "scope_id in (?)", []interface{}{scopeIds}, db.WithLimit(limit)); err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
plgIds := make([]string, 0, len(hostCatalogs))
|
|
for _, c := range hostCatalogs {
|
|
plgIds = append(plgIds, c.PluginId)
|
|
}
|
|
var plgs []*hostplugin.Plugin
|
|
if err := r.reader.SearchWhere(ctx, &plgs, "public_id in (?)", []interface{}{plgIds}); err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
return hostCatalogs, plgs, nil
|
|
}
|
|
|
|
// DeleteCatalog deletes catalog for the provided id from the repository
|
|
// returning a count of the number of records deleted. All options are ignored.
|
|
func (r *Repository) DeleteCatalog(ctx context.Context, id string, _ ...Option) (int, error) {
|
|
const op = "plugin.(Repository).DeleteCatalog"
|
|
if id == "" {
|
|
return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "no public id")
|
|
}
|
|
|
|
c, p, err := r.getCatalog(ctx, id)
|
|
if err != nil && !errors.IsNotFoundError(err) {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
if c == nil {
|
|
return db.NoRowsAffected, nil
|
|
}
|
|
plgHc, err := toPluginCatalog(ctx, c)
|
|
if err != nil {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
sets, _, err := r.getSets(ctx, "", c.GetPublicId())
|
|
if err != nil {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
var plgSets []*pbset.HostSet
|
|
for _, s := range sets {
|
|
ps, err := toPluginSet(ctx, s)
|
|
if err != nil {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op)
|
|
}
|
|
plgSets = append(plgSets, ps)
|
|
}
|
|
|
|
plgClient, ok := r.plugins[c.GetPluginId()]
|
|
if !ok || plgClient == nil {
|
|
return 0, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("plugin %q not available", c.GetPluginId()))
|
|
}
|
|
_, err = plgClient.OnDeleteCatalog(ctx, &plgpb.OnDeleteCatalogRequest{
|
|
Catalog: plgHc,
|
|
Sets: plgSets,
|
|
Persisted: p,
|
|
})
|
|
if err != nil {
|
|
// Even if the plugin returns an error, we ignore it and proceed with
|
|
// deleting the catalog.
|
|
event.WriteError(ctx, op, err, event.WithInfoMsg("plugin deleting catalog", "host plugin id", c.GetPluginId()))
|
|
}
|
|
|
|
oplogWrapper, err := r.kms.GetWrapper(ctx, c.ScopeId, kms.KeyPurposeOplog)
|
|
if err != nil {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper"))
|
|
}
|
|
metadata := c.oplog(oplog.OpType_OP_TYPE_DELETE)
|
|
|
|
var rowsDeleted int
|
|
var deleteCatalog *HostCatalog
|
|
_, err = r.writer.DoTx(
|
|
ctx,
|
|
db.StdRetryCnt,
|
|
db.ExpBackoff{},
|
|
func(_ db.Reader, w db.Writer) error {
|
|
deleteCatalog = c.clone()
|
|
var err error
|
|
rowsDeleted, err = w.Delete(
|
|
ctx,
|
|
deleteCatalog,
|
|
db.WithOplog(oplogWrapper, metadata),
|
|
)
|
|
if err != nil {
|
|
return errors.Wrap(ctx, err, op)
|
|
}
|
|
if rowsDeleted > 1 {
|
|
return errors.New(ctx, errors.MultipleRecords, op, "more than 1 resource would have been deleted")
|
|
}
|
|
return nil
|
|
},
|
|
)
|
|
|
|
if err != nil {
|
|
return db.NoRowsAffected, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("delete failed for %s", c.PublicId)))
|
|
}
|
|
|
|
return rowsDeleted, nil
|
|
}
|
|
|
|
// getCatalog retrieves the *HostCatalog with the provided id. If it is not found or there
|
|
// is an problem getting it from the database an error is returned instead.
|
|
func (r *Repository) getCatalog(ctx context.Context, id string) (*HostCatalog, *plgpb.HostCatalogPersisted, error) {
|
|
const op = "plugin.(Repository).getCatalog"
|
|
ca := &catalogAgg{}
|
|
ca.PublicId = id
|
|
if err := r.reader.LookupByPublicId(ctx, ca); err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("failed for: %s", id)))
|
|
}
|
|
c, s := ca.toCatalogAndPersisted()
|
|
var p *plgpb.HostCatalogPersisted
|
|
if s != nil {
|
|
var err error
|
|
p, err = toPluginPersistedData(ctx, r.kms, c.GetScopeId(), s)
|
|
if err != nil {
|
|
return nil, nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
}
|
|
return c, p, nil
|
|
}
|
|
|
|
func (r *Repository) getPlugin(ctx context.Context, plgId string) (*hostplugin.Plugin, error) {
|
|
const op = "plugin.(Repository).getPlugin"
|
|
if plgId == "" {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "no plugin id")
|
|
}
|
|
plg := hostplugin.NewPlugin()
|
|
plg.PublicId = plgId
|
|
if err := r.reader.LookupByPublicId(ctx, plg); err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("unable to get host plugin with id %q", plgId)))
|
|
}
|
|
return plg, nil
|
|
}
|
|
|
|
// toPluginCatalog returns a host catalog, with it's secret if available, in the format expected
|
|
// by the host plugin system.
|
|
func toPluginCatalog(ctx context.Context, in *HostCatalog) (*pb.HostCatalog, error) {
|
|
const op = "plugin.toPluginCatalog"
|
|
if in == nil {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "nil storage plugin")
|
|
}
|
|
var name, description *wrapperspb.StringValue
|
|
if inName := in.GetName(); inName != "" {
|
|
name = wrapperspb.String(inName)
|
|
}
|
|
if inDescription := in.GetDescription(); inDescription != "" {
|
|
description = wrapperspb.String(inDescription)
|
|
}
|
|
|
|
hc := &pb.HostCatalog{
|
|
Id: in.GetPublicId(),
|
|
ScopeId: in.GetScopeId(),
|
|
Name: name,
|
|
Description: description,
|
|
}
|
|
if len(in.GetSecretsHmac()) > 0 {
|
|
hc.SecretsHmac = base58.Encode(in.GetSecretsHmac())
|
|
}
|
|
if in.GetAttributes() != nil {
|
|
attrs := &structpb.Struct{}
|
|
if err := proto.Unmarshal(in.GetAttributes(), attrs); err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to unmarshal attributes"))
|
|
}
|
|
hc.Attrs = &pb.HostCatalog_Attributes{
|
|
Attributes: attrs,
|
|
}
|
|
}
|
|
if in.Secrets != nil {
|
|
hc.Secrets = in.Secrets
|
|
}
|
|
return hc, nil
|
|
}
|
|
|
|
// toPluginCatalog converts a *HostCatalogSecret from storage to a
|
|
// *plgpb.HostCatalogPersisted expected by a plugin. Scope Id must be set.
|
|
func toPluginPersistedData(ctx context.Context, kmsCache *kms.Kms, scopeId string, cSecret *HostCatalogSecret) (*plgpb.HostCatalogPersisted, error) {
|
|
const op = "plugin.(Repository).getPersistedDataForCatalog"
|
|
if scopeId == "" {
|
|
return nil, errors.New(ctx, errors.InvalidParameter, op, "empty scope id")
|
|
}
|
|
if cSecret == nil {
|
|
return nil, nil
|
|
}
|
|
dbWrapper, err := kmsCache.GetWrapper(ctx, scopeId, kms.KeyPurposeDatabase)
|
|
if err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get db wrapper"))
|
|
}
|
|
if err := cSecret.decrypt(ctx, dbWrapper); err != nil {
|
|
return nil, errors.Wrap(ctx, err, op)
|
|
}
|
|
|
|
secrets := &structpb.Struct{}
|
|
if err := proto.Unmarshal(cSecret.GetSecret(), secrets); err != nil {
|
|
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unmarshaling secret"))
|
|
}
|
|
return &plgpb.HostCatalogPersisted{Secrets: secrets}, nil
|
|
}
|