open-vault/vault/core.go
Calvin Leung Huang bb54e9c131 Backend plugin system (#2874)
* Add backend plugin changes

* Fix totp backend plugin tests

* Fix logical/plugin InvalidateKey test

* Fix plugin catalog CRUD test, fix NoopBackend

* Clean up commented code block

* Fix system backend mount test

* Set plugin_name to omitempty, fix handleMountTable config parsing

* Clean up comments, keep shim connections alive until cleanup

* Include pluginClient, disallow LookupPlugin call from within a plugin

* Add wrapper around backendPluginClient for proper cleanup

* Add logger shim tests

* Add logger, storage, and system shim tests

* Use pointer receivers for system view shim

* Use plugin name if no path is provided on mount

* Enable plugins for auth backends

* Add backend type attribute, move builtin/plugin/package

* Fix merge conflict

* Fix missing plugin name in mount config

* Add integration tests on enabling auth backend plugins

* Remove dependency cycle on mock-plugin

* Add passthrough backend plugin, use logical.BackendType to determine lease generation

* Remove vault package dependency on passthrough package

* Add basic impl test for passthrough plugin

* Incorporate feedback; set b.backend after shims creation on backendPluginServer

* Fix totp plugin test

* Add plugin backends docs

* Fix tests

* Fix builtin/plugin tests

* Remove flatten from PluginRunner fields

* Move mock plugin to logical/plugin, remove totp and passthrough plugins

* Move pluginMap into newPluginClient

* Do not create storage RPC connection on HandleRequest and HandleExistenceCheck

* Change shim logger's Fatal to no-op

* Change BackendType to uint32, match UX backend types

* Change framework.Backend Setup signature

* Add Setup func to logical.Backend interface

* Move OptionallyEnableMlock call into plugin.Serve, update docs and comments

* Remove commented var in plugin package

* RegisterLicense on logical.Backend interface (#3017)

* Add RegisterLicense to logical.Backend interface

* Update RegisterLicense to use callback func on framework.Backend

* Refactor framework.Backend.RegisterLicense

* plugin: Prevent plugin.SystemViewClient.ResponseWrapData from getting JWTs

* plugin: Revert BackendType to remove TypePassthrough and related references

* Fix typo in plugin backends docs
2017-07-20 13:28:40 -04:00

1853 lines
52 KiB
Go

package vault
import (
"crypto"
"crypto/ecdsa"
"crypto/subtle"
"crypto/x509"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"path/filepath"
"sync"
"time"
"github.com/armon/go-metrics"
log "github.com/mgutz/logxi/v1"
"golang.org/x/net/context"
"google.golang.org/grpc"
"github.com/hashicorp/errwrap"
"github.com/hashicorp/go-multierror"
"github.com/hashicorp/go-uuid"
"github.com/hashicorp/vault/audit"
"github.com/hashicorp/vault/helper/consts"
"github.com/hashicorp/vault/helper/errutil"
"github.com/hashicorp/vault/helper/jsonutil"
"github.com/hashicorp/vault/helper/logformat"
"github.com/hashicorp/vault/helper/mlock"
"github.com/hashicorp/vault/logical"
"github.com/hashicorp/vault/physical"
"github.com/hashicorp/vault/shamir"
cache "github.com/patrickmn/go-cache"
)
const (
// coreLockPath is the path used to acquire a coordinating lock
// for a highly-available deploy.
coreLockPath = "core/lock"
// The poison pill is used as a check during certain scenarios to indicate
// to standby nodes that they should seal
poisonPillPath = "core/poison-pill"
// coreLeaderPrefix is the prefix used for the UUID that contains
// the currently elected leader.
coreLeaderPrefix = "core/leader/"
// lockRetryInterval is the interval we re-attempt to acquire the
// HA lock if an error is encountered
lockRetryInterval = 10 * time.Second
// leaderCheckInterval is how often a standby checks for a new leader
leaderCheckInterval = 2500 * time.Millisecond
// keyRotateCheckInterval is how often a standby checks for a key
// rotation taking place.
keyRotateCheckInterval = 30 * time.Second
// keyRotateGracePeriod is how long we allow an upgrade path
// for standby instances before we delete the upgrade keys
keyRotateGracePeriod = 2 * time.Minute
// leaderPrefixCleanDelay is how long to wait between deletions
// of orphaned leader keys, to prevent slamming the backend.
leaderPrefixCleanDelay = 200 * time.Millisecond
// coreKeyringCanaryPath is used as a canary to indicate to replicated
// clusters that they need to perform a rekey operation synchronously; this
// isn't keyring-canary to avoid ignoring it when ignoring core/keyring
coreKeyringCanaryPath = "core/canary-keyring"
)
var (
// ErrAlreadyInit is returned if the core is already
// initialized. This prevents a re-initialization.
ErrAlreadyInit = errors.New("Vault is already initialized")
// ErrNotInit is returned if a non-initialized barrier
// is attempted to be unsealed.
ErrNotInit = errors.New("Vault is not initialized")
// ErrInternalError is returned when we don't want to leak
// any information about an internal error
ErrInternalError = errors.New("internal error")
// ErrHANotEnabled is returned if the operation only makes sense
// in an HA setting
ErrHANotEnabled = errors.New("Vault is not configured for highly-available mode")
// manualStepDownSleepPeriod is how long to sleep after a user-initiated
// step down of the active node, to prevent instantly regrabbing the lock.
// It's var not const so that tests can manipulate it.
manualStepDownSleepPeriod = 10 * time.Second
// Functions only in the Enterprise version
enterprisePostUnseal = enterprisePostUnsealImpl
enterprisePreSeal = enterprisePreSealImpl
startReplication = startReplicationImpl
stopReplication = stopReplicationImpl
LastRemoteWAL = lastRemoteWALImpl
)
// ReloadFunc are functions that are called when a reload is requested.
type ReloadFunc func(map[string]interface{}) error
// NonFatalError is an error that can be returned during NewCore that should be
// displayed but not cause a program exit
type NonFatalError struct {
Err error
}
func (e *NonFatalError) WrappedErrors() []error {
return []error{e.Err}
}
func (e *NonFatalError) Error() string {
return e.Err.Error()
}
// ErrInvalidKey is returned if there is a user-based error with a provided
// unseal key. This will be shown to the user, so should not contain
// information that is sensitive.
type ErrInvalidKey struct {
Reason string
}
func (e *ErrInvalidKey) Error() string {
return fmt.Sprintf("invalid key: %v", e.Reason)
}
type activeAdvertisement struct {
RedirectAddr string `json:"redirect_addr"`
ClusterAddr string `json:"cluster_addr,omitempty"`
ClusterCert []byte `json:"cluster_cert,omitempty"`
ClusterKeyParams *clusterKeyParams `json:"cluster_key_params,omitempty"`
}
type unlockInformation struct {
Parts [][]byte
Nonce string
}
// Core is used as the central manager of Vault activity. It is the primary point of
// interface for API handlers and is responsible for managing the logical and physical
// backends, router, security barrier, and audit trails.
type Core struct {
// N.B.: This is used to populate a dev token down replication, as
// otherwise, after replication is started, a dev would have to go through
// the generate-root process simply to talk to the new follower cluster.
devToken string
// HABackend may be available depending on the physical backend
ha physical.HABackend
// redirectAddr is the address we advertise as leader if held
redirectAddr string
// clusterAddr is the address we use for clustering
clusterAddr string
// physical backend is the un-trusted backend with durable data
physical physical.Backend
// Our Seal, for seal configuration information
seal Seal
// barrier is the security barrier wrapping the physical backend
barrier SecurityBarrier
// router is responsible for managing the mount points for logical backends.
router *Router
// logicalBackends is the mapping of backends to use for this core
logicalBackends map[string]logical.Factory
// credentialBackends is the mapping of backends to use for this core
credentialBackends map[string]logical.Factory
// auditBackends is the mapping of backends to use for this core
auditBackends map[string]audit.Factory
// stateLock protects mutable state
stateLock sync.RWMutex
sealed bool
standby bool
standbyDoneCh chan struct{}
standbyStopCh chan struct{}
manualStepDownCh chan struct{}
// unlockInfo has the keys provided to Unseal until the threshold number of parts is available, as well as the operation nonce
unlockInfo *unlockInformation
// generateRootProgress holds the shares until we reach enough
// to verify the master key
generateRootConfig *GenerateRootConfig
generateRootProgress [][]byte
generateRootLock sync.Mutex
// These variables holds the config and shares we have until we reach
// enough to verify the appropriate master key. Note that the same lock is
// used; this isn't time-critical so this shouldn't be a problem.
barrierRekeyConfig *SealConfig
barrierRekeyProgress [][]byte
recoveryRekeyConfig *SealConfig
recoveryRekeyProgress [][]byte
rekeyLock sync.RWMutex
// mounts is loaded after unseal since it is a protected
// configuration
mounts *MountTable
// mountsLock is used to ensure that the mounts table does not
// change underneath a calling function
mountsLock sync.RWMutex
// auth is loaded after unseal since it is a protected
// configuration
auth *MountTable
// authLock is used to ensure that the auth table does not
// change underneath a calling function
authLock sync.RWMutex
// audit is loaded after unseal since it is a protected
// configuration
audit *MountTable
// auditLock is used to ensure that the audit table does not
// change underneath a calling function
auditLock sync.RWMutex
// auditBroker is used to ingest the audit events and fan
// out into the configured audit backends
auditBroker *AuditBroker
// auditedHeaders is used to configure which http headers
// can be output in the audit logs
auditedHeaders *AuditedHeadersConfig
// systemBarrierView is the barrier view for the system backend
systemBarrierView *BarrierView
// expiration manager is used for managing LeaseIDs,
// renewal, expiration and revocation
expiration *ExpirationManager
// rollback manager is used to run rollbacks periodically
rollback *RollbackManager
// policy store is used to manage named ACL policies
policyStore *PolicyStore
// token store is used to manage authentication tokens
tokenStore *TokenStore
// metricsCh is used to stop the metrics streaming
metricsCh chan struct{}
// metricsMutex is used to prevent a race condition between
// metrics emission and sealing leading to a nil pointer
metricsMutex sync.Mutex
defaultLeaseTTL time.Duration
maxLeaseTTL time.Duration
logger log.Logger
// cachingDisabled indicates whether caches are disabled
cachingDisabled bool
// reloadFuncs is a map containing reload functions
reloadFuncs map[string][]ReloadFunc
// reloadFuncsLock controls access to the funcs
reloadFuncsLock sync.RWMutex
// wrappingJWTKey is the key used for generating JWTs containing response
// wrapping information
wrappingJWTKey *ecdsa.PrivateKey
//
// Cluster information
//
// Name
clusterName string
// Used to modify cluster parameters
clusterParamsLock sync.RWMutex
// The private key stored in the barrier used for establishing
// mutually-authenticated connections between Vault cluster members
localClusterPrivateKey crypto.Signer
// The local cluster cert
localClusterCert []byte
// The parsed form of the local cluster cert
localClusterParsedCert *x509.Certificate
// The TCP addresses we should use for clustering
clusterListenerAddrs []*net.TCPAddr
// The handler to use for request forwarding
clusterHandler http.Handler
// Tracks whether cluster listeners are running, e.g. it's safe to send a
// shutdown down the channel
clusterListenersRunning bool
// Shutdown channel for the cluster listeners
clusterListenerShutdownCh chan struct{}
// Shutdown success channel. We need this to be done serially to ensure
// that binds are removed before they might be reinstated.
clusterListenerShutdownSuccessCh chan struct{}
// Write lock used to ensure that we don't have multiple connections adjust
// this value at the same time
requestForwardingConnectionLock sync.RWMutex
// Most recent leader UUID. Used to avoid repeatedly JSON parsing the same
// values.
clusterLeaderUUID string
// Most recent leader redirect addr
clusterLeaderRedirectAddr string
// Lock for the cluster leader values
clusterLeaderParamsLock sync.RWMutex
// Info on cluster members
clusterPeerClusterAddrsCache *cache.Cache
// The grpc Server that handles server RPC calls
rpcServer *grpc.Server
// The context for the client
rpcClientConnContext context.Context
// The function for canceling the client connection
rpcClientConnCancelFunc context.CancelFunc
// The grpc ClientConn for RPC calls
rpcClientConn *grpc.ClientConn
// The grpc forwarding client
rpcForwardingClient *forwardingClient
// CORS Information
corsConfig *CORSConfig
// replicationState keeps the current replication state cached for quick
// lookup
replicationState consts.ReplicationState
// uiEnabled indicates whether Vault Web UI is enabled or not
uiEnabled bool
// pluginDirectory is the location vault will look for plugin binaries
pluginDirectory string
// pluginCatalog is used to manage plugin configurations
pluginCatalog *PluginCatalog
enableMlock bool
}
// CoreConfig is used to parameterize a core
type CoreConfig struct {
DevToken string `json:"dev_token" structs:"dev_token" mapstructure:"dev_token"`
LogicalBackends map[string]logical.Factory `json:"logical_backends" structs:"logical_backends" mapstructure:"logical_backends"`
CredentialBackends map[string]logical.Factory `json:"credential_backends" structs:"credential_backends" mapstructure:"credential_backends"`
AuditBackends map[string]audit.Factory `json:"audit_backends" structs:"audit_backends" mapstructure:"audit_backends"`
Physical physical.Backend `json:"physical" structs:"physical" mapstructure:"physical"`
// May be nil, which disables HA operations
HAPhysical physical.HABackend `json:"ha_physical" structs:"ha_physical" mapstructure:"ha_physical"`
Seal Seal `json:"seal" structs:"seal" mapstructure:"seal"`
Logger log.Logger `json:"logger" structs:"logger" mapstructure:"logger"`
// Disables the LRU cache on the physical backend
DisableCache bool `json:"disable_cache" structs:"disable_cache" mapstructure:"disable_cache"`
// Disables mlock syscall
DisableMlock bool `json:"disable_mlock" structs:"disable_mlock" mapstructure:"disable_mlock"`
// Custom cache size for the LRU cache on the physical backend, or zero for default
CacheSize int `json:"cache_size" structs:"cache_size" mapstructure:"cache_size"`
// Set as the leader address for HA
RedirectAddr string `json:"redirect_addr" structs:"redirect_addr" mapstructure:"redirect_addr"`
// Set as the cluster address for HA
ClusterAddr string `json:"cluster_addr" structs:"cluster_addr" mapstructure:"cluster_addr"`
DefaultLeaseTTL time.Duration `json:"default_lease_ttl" structs:"default_lease_ttl" mapstructure:"default_lease_ttl"`
MaxLeaseTTL time.Duration `json:"max_lease_ttl" structs:"max_lease_ttl" mapstructure:"max_lease_ttl"`
ClusterName string `json:"cluster_name" structs:"cluster_name" mapstructure:"cluster_name"`
EnableUI bool `json:"ui" structs:"ui" mapstructure:"ui"`
PluginDirectory string `json:"plugin_directory" structs:"plugin_directory" mapstructure:"plugin_directory"`
ReloadFuncs *map[string][]ReloadFunc
ReloadFuncsLock *sync.RWMutex
}
// NewCore is used to construct a new core
func NewCore(conf *CoreConfig) (*Core, error) {
if conf.HAPhysical != nil && conf.HAPhysical.HAEnabled() {
if conf.RedirectAddr == "" {
return nil, fmt.Errorf("missing redirect address")
}
}
if conf.DefaultLeaseTTL == 0 {
conf.DefaultLeaseTTL = defaultLeaseTTL
}
if conf.MaxLeaseTTL == 0 {
conf.MaxLeaseTTL = maxLeaseTTL
}
if conf.DefaultLeaseTTL > conf.MaxLeaseTTL {
return nil, fmt.Errorf("cannot have DefaultLeaseTTL larger than MaxLeaseTTL")
}
// Validate the advertise addr if its given to us
if conf.RedirectAddr != "" {
u, err := url.Parse(conf.RedirectAddr)
if err != nil {
return nil, fmt.Errorf("redirect address is not valid url: %s", err)
}
if u.Scheme == "" {
return nil, fmt.Errorf("redirect address must include scheme (ex. 'http')")
}
}
// Make a default logger if not provided
if conf.Logger == nil {
conf.Logger = logformat.NewVaultLogger(log.LevelTrace)
}
// Setup the core
c := &Core{
devToken: conf.DevToken,
physical: conf.Physical,
redirectAddr: conf.RedirectAddr,
clusterAddr: conf.ClusterAddr,
seal: conf.Seal,
router: NewRouter(),
sealed: true,
standby: true,
logger: conf.Logger,
defaultLeaseTTL: conf.DefaultLeaseTTL,
maxLeaseTTL: conf.MaxLeaseTTL,
cachingDisabled: conf.DisableCache,
clusterName: conf.ClusterName,
clusterListenerShutdownCh: make(chan struct{}),
clusterListenerShutdownSuccessCh: make(chan struct{}),
clusterPeerClusterAddrsCache: cache.New(3*heartbeatInterval, time.Second),
enableMlock: !conf.DisableMlock,
}
// Load CORS config and provide core
c.corsConfig = &CORSConfig{core: c}
// Wrap the physical backend in a cache layer if enabled and not already wrapped
if _, isCache := conf.Physical.(*physical.Cache); !conf.DisableCache && !isCache {
c.physical = physical.NewCache(conf.Physical, conf.CacheSize, conf.Logger)
}
if !conf.DisableMlock {
// Ensure our memory usage is locked into physical RAM
if err := mlock.LockMemory(); err != nil {
return nil, fmt.Errorf(
"Failed to lock memory: %v\n\n"+
"This usually means that the mlock syscall is not available.\n"+
"Vault uses mlock to prevent memory from being swapped to\n"+
"disk. This requires root privileges as well as a machine\n"+
"that supports mlock. Please enable mlock on your system or\n"+
"disable Vault from using it. To disable Vault from using it,\n"+
"set the `disable_mlock` configuration option in your configuration\n"+
"file.",
err)
}
}
var err error
if conf.PluginDirectory != "" {
c.pluginDirectory, err = filepath.Abs(conf.PluginDirectory)
if err != nil {
return nil, fmt.Errorf("core setup failed, could not verify plugin directory: %v", err)
}
}
// Construct a new AES-GCM barrier
c.barrier, err = NewAESGCMBarrier(c.physical)
if err != nil {
return nil, fmt.Errorf("barrier setup failed: %v", err)
}
if conf.HAPhysical != nil && conf.HAPhysical.HAEnabled() {
c.ha = conf.HAPhysical
}
// We create the funcs here, then populate the given config with it so that
// the caller can share state
conf.ReloadFuncsLock = &c.reloadFuncsLock
c.reloadFuncsLock.Lock()
c.reloadFuncs = make(map[string][]ReloadFunc)
c.reloadFuncsLock.Unlock()
conf.ReloadFuncs = &c.reloadFuncs
// Setup the backends
logicalBackends := make(map[string]logical.Factory)
for k, f := range conf.LogicalBackends {
logicalBackends[k] = f
}
_, ok := logicalBackends["generic"]
if !ok {
logicalBackends["generic"] = PassthroughBackendFactory
}
logicalBackends["cubbyhole"] = CubbyholeBackendFactory
logicalBackends["system"] = func(config *logical.BackendConfig) (logical.Backend, error) {
b := NewSystemBackend(c)
if err := b.Setup(config); err != nil {
return nil, err
}
return b, nil
}
c.logicalBackends = logicalBackends
credentialBackends := make(map[string]logical.Factory)
for k, f := range conf.CredentialBackends {
credentialBackends[k] = f
}
credentialBackends["token"] = func(config *logical.BackendConfig) (logical.Backend, error) {
return NewTokenStore(c, config)
}
c.credentialBackends = credentialBackends
auditBackends := make(map[string]audit.Factory)
for k, f := range conf.AuditBackends {
auditBackends[k] = f
}
c.auditBackends = auditBackends
if c.seal == nil {
c.seal = &DefaultSeal{}
}
c.seal.SetCore(c)
// Attempt unsealing with stored keys; if there are no stored keys this
// returns nil, otherwise returns nil or an error
storedKeyErr := c.UnsealWithStoredKeys()
return c, storedKeyErr
}
// Shutdown is invoked when the Vault instance is about to be terminated. It
// should not be accessible as part of an API call as it will cause an availability
// problem. It is only used to gracefully quit in the case of HA so that failover
// happens as quickly as possible.
func (c *Core) Shutdown() error {
c.stateLock.Lock()
defer c.stateLock.Unlock()
if c.sealed {
return nil
}
// Seal the Vault, causes a leader stepdown
return c.sealInternal()
}
// CORSConfig returns the current CORS configuration
func (c *Core) CORSConfig() *CORSConfig {
return c.corsConfig
}
// LookupToken returns the properties of the token from the token store. This
// is particularly useful to fetch the accessor of the client token and get it
// populated in the logical request along with the client token. The accessor
// of the client token can get audit logged.
func (c *Core) LookupToken(token string) (*TokenEntry, error) {
if token == "" {
return nil, fmt.Errorf("missing client token")
}
c.stateLock.RLock()
defer c.stateLock.RUnlock()
if c.sealed {
return nil, consts.ErrSealed
}
if c.standby {
return nil, consts.ErrStandby
}
// Many tests don't have a token store running
if c.tokenStore == nil {
return nil, nil
}
return c.tokenStore.Lookup(token)
}
func (c *Core) fetchACLandTokenEntry(req *logical.Request) (*ACL, *TokenEntry, error) {
defer metrics.MeasureSince([]string{"core", "fetch_acl_and_token"}, time.Now())
// Ensure there is a client token
if req.ClientToken == "" {
return nil, nil, fmt.Errorf("missing client token")
}
if c.tokenStore == nil {
c.logger.Error("core: token store is unavailable")
return nil, nil, ErrInternalError
}
// Resolve the token policy
te, err := c.tokenStore.Lookup(req.ClientToken)
if err != nil {
c.logger.Error("core: failed to lookup token", "error", err)
return nil, nil, ErrInternalError
}
// Ensure the token is valid
if te == nil {
return nil, nil, logical.ErrPermissionDenied
}
// Construct the corresponding ACL object
acl, err := c.policyStore.ACL(te.Policies...)
if err != nil {
c.logger.Error("core: failed to construct ACL", "error", err)
return nil, nil, ErrInternalError
}
return acl, te, nil
}
func (c *Core) checkToken(req *logical.Request) (*logical.Auth, *TokenEntry, error) {
defer metrics.MeasureSince([]string{"core", "check_token"}, time.Now())
acl, te, err := c.fetchACLandTokenEntry(req)
if err != nil {
return nil, te, err
}
// Check if this is a root protected path
rootPath := c.router.RootPath(req.Path)
// When we receive a write of either type, rather than require clients to
// PUT/POST and trust the operation, we ask the backend to give us the real
// skinny -- if the backend implements an existence check, it can tell us
// whether a particular resource exists. Then we can mark it as an update
// or creation as appropriate.
if req.Operation == logical.CreateOperation || req.Operation == logical.UpdateOperation {
checkExists, resourceExists, err := c.router.RouteExistenceCheck(req)
switch err {
case logical.ErrUnsupportedPath:
// fail later via bad path to avoid confusing items in the log
checkExists = false
case nil:
// Continue on
default:
c.logger.Error("core: failed to run existence check", "error", err)
if _, ok := err.(errutil.UserError); ok {
return nil, nil, err
} else {
return nil, nil, ErrInternalError
}
}
switch {
case checkExists == false:
// No existence check, so always treate it as an update operation, which is how it is pre 0.5
req.Operation = logical.UpdateOperation
case resourceExists == true:
// It exists, so force an update operation
req.Operation = logical.UpdateOperation
case resourceExists == false:
// It doesn't exist, force a create operation
req.Operation = logical.CreateOperation
default:
panic("unreachable code")
}
}
// Create the auth response
auth := &logical.Auth{
ClientToken: req.ClientToken,
Accessor: req.ClientTokenAccessor,
Policies: te.Policies,
Metadata: te.Meta,
DisplayName: te.DisplayName,
}
// Check the standard non-root ACLs. Return the token entry if it's not
// allowed so we can decrement the use count.
allowed, rootPrivs := acl.AllowOperation(req)
if !allowed {
// Return auth for audit logging even if not allowed
return auth, te, logical.ErrPermissionDenied
}
if rootPath && !rootPrivs {
// Return auth for audit logging even if not allowed
return auth, te, logical.ErrPermissionDenied
}
return auth, te, nil
}
// Sealed checks if the Vault is current sealed
func (c *Core) Sealed() (bool, error) {
c.stateLock.RLock()
defer c.stateLock.RUnlock()
return c.sealed, nil
}
// Standby checks if the Vault is in standby mode
func (c *Core) Standby() (bool, error) {
c.stateLock.RLock()
defer c.stateLock.RUnlock()
return c.standby, nil
}
// Leader is used to get the current active leader
func (c *Core) Leader() (isLeader bool, leaderAddr string, err error) {
c.stateLock.RLock()
defer c.stateLock.RUnlock()
// Check if sealed
if c.sealed {
return false, "", consts.ErrSealed
}
// Check if HA enabled
if c.ha == nil {
return false, "", ErrHANotEnabled
}
// Check if we are the leader
if !c.standby {
return true, c.redirectAddr, nil
}
// Initialize a lock
lock, err := c.ha.LockWith(coreLockPath, "read")
if err != nil {
return false, "", err
}
// Read the value
held, leaderUUID, err := lock.Value()
if err != nil {
return false, "", err
}
if !held {
return false, "", nil
}
c.clusterLeaderParamsLock.RLock()
localLeaderUUID := c.clusterLeaderUUID
localRedirAddr := c.clusterLeaderRedirectAddr
c.clusterLeaderParamsLock.RUnlock()
// If the leader hasn't changed, return the cached value; nothing changes
// mid-leadership, and the barrier caches anyways
if leaderUUID == localLeaderUUID && localRedirAddr != "" {
return false, localRedirAddr, nil
}
c.logger.Trace("core: found new active node information, refreshing")
c.clusterLeaderParamsLock.Lock()
defer c.clusterLeaderParamsLock.Unlock()
// Validate base conditions again
if leaderUUID == c.clusterLeaderUUID && c.clusterLeaderRedirectAddr != "" {
return false, localRedirAddr, nil
}
key := coreLeaderPrefix + leaderUUID
entry, err := c.barrier.Get(key)
if err != nil {
return false, "", err
}
if entry == nil {
return false, "", nil
}
var oldAdv bool
var adv activeAdvertisement
err = jsonutil.DecodeJSON(entry.Value, &adv)
if err != nil {
// Fall back to pre-struct handling
adv.RedirectAddr = string(entry.Value)
c.logger.Trace("core: parsed redirect addr for new active node", "redirect_addr", adv.RedirectAddr)
oldAdv = true
}
if !oldAdv {
c.logger.Trace("core: parsing information for new active node", "active_cluster_addr", adv.ClusterAddr, "active_redirect_addr", adv.RedirectAddr)
// Ensure we are using current values
err = c.loadLocalClusterTLS(adv)
if err != nil {
return false, "", err
}
// This will ensure that we both have a connection at the ready and that
// the address is the current known value
err = c.refreshRequestForwardingConnection(adv.ClusterAddr)
if err != nil {
return false, "", err
}
}
// Don't set these until everything has been parsed successfully or we'll
// never try again
c.clusterLeaderRedirectAddr = adv.RedirectAddr
c.clusterLeaderUUID = leaderUUID
return false, adv.RedirectAddr, nil
}
// SecretProgress returns the number of keys provided so far
func (c *Core) SecretProgress() (int, string) {
c.stateLock.RLock()
defer c.stateLock.RUnlock()
switch c.unlockInfo {
case nil:
return 0, ""
default:
return len(c.unlockInfo.Parts), c.unlockInfo.Nonce
}
}
// ResetUnsealProcess removes the current unlock parts from memory, to reset
// the unsealing process
func (c *Core) ResetUnsealProcess() {
c.stateLock.Lock()
defer c.stateLock.Unlock()
if !c.sealed {
return
}
c.unlockInfo = nil
}
// Unseal is used to provide one of the key parts to unseal the Vault.
//
// They key given as a parameter will automatically be zerod after
// this method is done with it. If you want to keep the key around, a copy
// should be made.
func (c *Core) Unseal(key []byte) (bool, error) {
defer metrics.MeasureSince([]string{"core", "unseal"}, time.Now())
// Verify the key length
min, max := c.barrier.KeyLength()
max += shamir.ShareOverhead
if len(key) < min {
return false, &ErrInvalidKey{fmt.Sprintf("key is shorter than minimum %d bytes", min)}
}
if len(key) > max {
return false, &ErrInvalidKey{fmt.Sprintf("key is longer than maximum %d bytes", max)}
}
// Get the seal configuration
config, err := c.seal.BarrierConfig()
if err != nil {
return false, err
}
// Ensure the barrier is initialized
if config == nil {
return false, ErrNotInit
}
c.stateLock.Lock()
defer c.stateLock.Unlock()
// Check if already unsealed
if !c.sealed {
return true, nil
}
masterKey, err := c.unsealPart(config, key)
if err != nil {
return false, err
}
if masterKey != nil {
return c.unsealInternal(masterKey)
}
return false, nil
}
func (c *Core) unsealPart(config *SealConfig, key []byte) ([]byte, error) {
// Check if we already have this piece
if c.unlockInfo != nil {
for _, existing := range c.unlockInfo.Parts {
if subtle.ConstantTimeCompare(existing, key) == 1 {
return nil, nil
}
}
} else {
uuid, err := uuid.GenerateUUID()
if err != nil {
return nil, err
}
c.unlockInfo = &unlockInformation{
Nonce: uuid,
}
}
// Store this key
c.unlockInfo.Parts = append(c.unlockInfo.Parts, key)
// Check if we don't have enough keys to unlock
if len(c.unlockInfo.Parts) < config.SecretThreshold {
if c.logger.IsDebug() {
c.logger.Debug("core: cannot unseal, not enough keys", "keys", len(c.unlockInfo.Parts), "threshold", config.SecretThreshold, "nonce", c.unlockInfo.Nonce)
}
return nil, nil
}
// Best-effort memzero of unlock parts once we're done with them
defer func() {
for i, _ := range c.unlockInfo.Parts {
memzero(c.unlockInfo.Parts[i])
}
c.unlockInfo = nil
}()
// Recover the master key
var masterKey []byte
var err error
if config.SecretThreshold == 1 {
masterKey = make([]byte, len(c.unlockInfo.Parts[0]))
copy(masterKey, c.unlockInfo.Parts[0])
} else {
masterKey, err = shamir.Combine(c.unlockInfo.Parts)
if err != nil {
return nil, fmt.Errorf("failed to compute master key: %v", err)
}
}
return masterKey, nil
}
// This must be called with the state write lock held
func (c *Core) unsealInternal(masterKey []byte) (bool, error) {
defer memzero(masterKey)
// Attempt to unlock
if err := c.barrier.Unseal(masterKey); err != nil {
return false, err
}
if c.logger.IsInfo() {
c.logger.Info("core: vault is unsealed")
}
// Do post-unseal setup if HA is not enabled
if c.ha == nil {
// We still need to set up cluster info even if it's not part of a
// cluster right now. This also populates the cached cluster object.
if err := c.setupCluster(); err != nil {
c.logger.Error("core: cluster setup failed", "error", err)
c.barrier.Seal()
c.logger.Warn("core: vault is sealed")
return false, err
}
if err := c.postUnseal(); err != nil {
c.logger.Error("core: post-unseal setup failed", "error", err)
c.barrier.Seal()
c.logger.Warn("core: vault is sealed")
return false, err
}
c.standby = false
} else {
// Go to standby mode, wait until we are active to unseal
c.standbyDoneCh = make(chan struct{})
c.standbyStopCh = make(chan struct{})
c.manualStepDownCh = make(chan struct{})
go c.runStandby(c.standbyDoneCh, c.standbyStopCh, c.manualStepDownCh)
}
// Success!
c.sealed = false
if c.ha != nil {
sd, ok := c.ha.(physical.ServiceDiscovery)
if ok {
if err := sd.NotifySealedStateChange(); err != nil {
if c.logger.IsWarn() {
c.logger.Warn("core: failed to notify unsealed status", "error", err)
}
}
}
}
return true, nil
}
// SealWithRequest takes in a logical.Request, acquires the lock, and passes
// through to sealInternal
func (c *Core) SealWithRequest(req *logical.Request) error {
defer metrics.MeasureSince([]string{"core", "seal-with-request"}, time.Now())
c.stateLock.Lock()
defer c.stateLock.Unlock()
if c.sealed {
return nil
}
return c.sealInitCommon(req)
}
// Seal takes in a token and creates a logical.Request, acquires the lock, and
// passes through to sealInternal
func (c *Core) Seal(token string) error {
defer metrics.MeasureSince([]string{"core", "seal"}, time.Now())
c.stateLock.Lock()
defer c.stateLock.Unlock()
if c.sealed {
return nil
}
req := &logical.Request{
Operation: logical.UpdateOperation,
Path: "sys/seal",
ClientToken: token,
}
return c.sealInitCommon(req)
}
// sealInitCommon is common logic for Seal and SealWithRequest and is used to
// re-seal the Vault. This requires the Vault to be unsealed again to perform
// any further operations.
func (c *Core) sealInitCommon(req *logical.Request) (retErr error) {
defer metrics.MeasureSince([]string{"core", "seal-internal"}, time.Now())
if req == nil {
retErr = multierror.Append(retErr, errors.New("nil request to seal"))
return retErr
}
// Validate the token is a root token
acl, te, err := c.fetchACLandTokenEntry(req)
if err != nil {
// Since there is no token store in standby nodes, sealing cannot
// be done. Ideally, the request has to be forwarded to leader node
// for validation and the operation should be performed. But for now,
// just returning with an error and recommending a vault restart, which
// essentially does the same thing.
if c.standby {
c.logger.Error("core: vault cannot seal when in standby mode; please restart instead")
retErr = multierror.Append(retErr, errors.New("vault cannot seal when in standby mode; please restart instead"))
return retErr
}
retErr = multierror.Append(retErr, err)
return retErr
}
// Audit-log the request before going any further
auth := &logical.Auth{
ClientToken: req.ClientToken,
Policies: te.Policies,
Metadata: te.Meta,
DisplayName: te.DisplayName,
}
if err := c.auditBroker.LogRequest(auth, req, c.auditedHeaders, nil); err != nil {
c.logger.Error("core: failed to audit request", "request_path", req.Path, "error", err)
retErr = multierror.Append(retErr, errors.New("failed to audit request, cannot continue"))
return retErr
}
// Attempt to use the token (decrement num_uses)
// On error bail out; if the token has been revoked, bail out too
if te != nil {
te, err = c.tokenStore.UseToken(te)
if err != nil {
c.logger.Error("core: failed to use token", "error", err)
retErr = multierror.Append(retErr, ErrInternalError)
return retErr
}
if te == nil {
// Token is no longer valid
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
if te.NumUses == -1 {
// Token needs to be revoked
defer func(id string) {
err = c.tokenStore.Revoke(id)
if err != nil {
c.logger.Error("core: token needed revocation after seal but failed to revoke", "error", err)
retErr = multierror.Append(retErr, ErrInternalError)
}
}(te.ID)
}
}
// Verify that this operation is allowed
allowed, rootPrivs := acl.AllowOperation(req)
if !allowed {
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
// We always require root privileges for this operation
if !rootPrivs {
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
//Seal the Vault
err = c.sealInternal()
if err != nil {
retErr = multierror.Append(retErr, err)
}
return retErr
}
// StepDown is used to step down from leadership
func (c *Core) StepDown(req *logical.Request) (retErr error) {
defer metrics.MeasureSince([]string{"core", "step_down"}, time.Now())
if req == nil {
retErr = multierror.Append(retErr, errors.New("nil request to step-down"))
return retErr
}
c.stateLock.Lock()
defer c.stateLock.Unlock()
if c.sealed {
return nil
}
if c.ha == nil || c.standby {
return nil
}
acl, te, err := c.fetchACLandTokenEntry(req)
if err != nil {
retErr = multierror.Append(retErr, err)
return retErr
}
// Audit-log the request before going any further
auth := &logical.Auth{
ClientToken: req.ClientToken,
Policies: te.Policies,
Metadata: te.Meta,
DisplayName: te.DisplayName,
}
if err := c.auditBroker.LogRequest(auth, req, c.auditedHeaders, nil); err != nil {
c.logger.Error("core: failed to audit request", "request_path", req.Path, "error", err)
retErr = multierror.Append(retErr, errors.New("failed to audit request, cannot continue"))
return retErr
}
// Attempt to use the token (decrement num_uses)
if te != nil {
te, err = c.tokenStore.UseToken(te)
if err != nil {
c.logger.Error("core: failed to use token", "error", err)
retErr = multierror.Append(retErr, ErrInternalError)
return retErr
}
if te == nil {
// Token has been revoked
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
if te.NumUses == -1 {
// Token needs to be revoked
defer func(id string) {
err = c.tokenStore.Revoke(id)
if err != nil {
c.logger.Error("core: token needed revocation after step-down but failed to revoke", "error", err)
retErr = multierror.Append(retErr, ErrInternalError)
}
}(te.ID)
}
}
// Verify that this operation is allowed
allowed, rootPrivs := acl.AllowOperation(req)
if !allowed {
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
// We always require root privileges for this operation
if !rootPrivs {
retErr = multierror.Append(retErr, logical.ErrPermissionDenied)
return retErr
}
select {
case c.manualStepDownCh <- struct{}{}:
default:
c.logger.Warn("core: manual step-down operation already queued")
}
return retErr
}
// sealInternal is an internal method used to seal the vault. It does not do
// any authorization checking. The stateLock must be held prior to calling.
func (c *Core) sealInternal() error {
// Enable that we are sealed to prevent furthur transactions
c.sealed = true
c.logger.Debug("core: marked as sealed")
// Clear forwarding clients
c.requestForwardingConnectionLock.Lock()
c.clearForwardingClients()
c.requestForwardingConnectionLock.Unlock()
// Do pre-seal teardown if HA is not enabled
if c.ha == nil {
// Even in a non-HA context we key off of this for some things
c.standby = true
if err := c.preSeal(); err != nil {
c.logger.Error("core: pre-seal teardown failed", "error", err)
return fmt.Errorf("internal error")
}
} else {
// Signal the standby goroutine to shutdown, wait for completion
close(c.standbyStopCh)
// Release the lock while we wait to avoid deadlocking
c.stateLock.Unlock()
<-c.standbyDoneCh
c.stateLock.Lock()
}
c.logger.Debug("core: sealing barrier")
if err := c.barrier.Seal(); err != nil {
c.logger.Error("core: error sealing barrier", "error", err)
return err
}
if c.ha != nil {
sd, ok := c.ha.(physical.ServiceDiscovery)
if ok {
if err := sd.NotifySealedStateChange(); err != nil {
if c.logger.IsWarn() {
c.logger.Warn("core: failed to notify sealed status", "error", err)
}
}
}
}
c.logger.Info("core: vault is sealed")
return nil
}
// postUnseal is invoked after the barrier is unsealed, but before
// allowing any user operations. This allows us to setup any state that
// requires the Vault to be unsealed such as mount tables, logical backends,
// credential stores, etc.
func (c *Core) postUnseal() (retErr error) {
defer metrics.MeasureSince([]string{"core", "post_unseal"}, time.Now())
defer func() {
if retErr != nil {
c.preSeal()
}
}()
c.logger.Info("core: post-unseal setup starting")
// Clear forwarding clients; we're active
c.requestForwardingConnectionLock.Lock()
c.clearForwardingClients()
c.requestForwardingConnectionLock.Unlock()
// Purge the backend if supported
if purgable, ok := c.physical.(physical.Purgable); ok {
purgable.Purge()
}
// Purge these for safety in case of a rekey
c.seal.SetBarrierConfig(nil)
if c.seal.RecoveryKeySupported() {
c.seal.SetRecoveryConfig(nil)
}
if err := enterprisePostUnseal(c); err != nil {
return err
}
if err := c.ensureWrappingKey(); err != nil {
return err
}
if err := c.loadMounts(); err != nil {
return err
}
if err := c.setupMounts(); err != nil {
return err
}
if err := c.startRollback(); err != nil {
return err
}
if err := c.setupPolicyStore(); err != nil {
return err
}
if err := c.loadCORSConfig(); err != nil {
return err
}
if err := c.loadCredentials(); err != nil {
return err
}
if err := c.setupCredentials(); err != nil {
return err
}
if err := c.setupExpiration(); err != nil {
return err
}
if err := c.loadAudits(); err != nil {
return err
}
if err := c.setupAudits(); err != nil {
return err
}
if err := c.setupAuditedHeadersConfig(); err != nil {
return err
}
if err := c.setupPluginCatalog(); err != nil {
return err
}
if c.ha != nil {
if err := c.startClusterListener(); err != nil {
return err
}
}
c.metricsCh = make(chan struct{})
go c.emitMetrics(c.metricsCh)
c.logger.Info("core: post-unseal setup complete")
return nil
}
// preSeal is invoked before the barrier is sealed, allowing
// for any state teardown required.
func (c *Core) preSeal() error {
defer metrics.MeasureSince([]string{"core", "pre_seal"}, time.Now())
c.logger.Info("core: pre-seal teardown starting")
// Clear any rekey progress
c.barrierRekeyConfig = nil
c.barrierRekeyProgress = nil
c.recoveryRekeyConfig = nil
c.recoveryRekeyProgress = nil
if c.metricsCh != nil {
close(c.metricsCh)
c.metricsCh = nil
}
var result error
c.stopClusterListener()
if err := c.teardownAudits(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error tearing down audits: {{err}}", err))
}
if err := c.stopExpiration(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error stopping expiration: {{err}}", err))
}
if err := c.teardownCredentials(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error tearing down credentials: {{err}}", err))
}
if err := c.teardownPolicyStore(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error tearing down policy store: {{err}}", err))
}
if err := c.stopRollback(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error stopping rollback: {{err}}", err))
}
if err := c.unloadMounts(); err != nil {
result = multierror.Append(result, errwrap.Wrapf("error unloading mounts: {{err}}", err))
}
if err := enterprisePreSeal(c); err != nil {
result = multierror.Append(result, err)
}
// Purge the backend if supported
if purgable, ok := c.physical.(physical.Purgable); ok {
purgable.Purge()
}
c.logger.Info("core: pre-seal teardown complete")
return result
}
func enterprisePostUnsealImpl(c *Core) error {
return nil
}
func enterprisePreSealImpl(c *Core) error {
return nil
}
func startReplicationImpl(c *Core) error {
return nil
}
func stopReplicationImpl(c *Core) error {
return nil
}
// runStandby is a long running routine that is used when an HA backend
// is enabled. It waits until we are leader and switches this Vault to
// active.
func (c *Core) runStandby(doneCh, stopCh, manualStepDownCh chan struct{}) {
defer close(doneCh)
defer close(manualStepDownCh)
c.logger.Info("core: entering standby mode")
// Monitor for key rotation
keyRotateDone := make(chan struct{})
keyRotateStop := make(chan struct{})
go c.periodicCheckKeyUpgrade(keyRotateDone, keyRotateStop)
// Monitor for new leadership
checkLeaderDone := make(chan struct{})
checkLeaderStop := make(chan struct{})
go c.periodicLeaderRefresh(checkLeaderDone, checkLeaderStop)
defer func() {
close(keyRotateStop)
<-keyRotateDone
close(checkLeaderStop)
<-checkLeaderDone
}()
for {
// Check for a shutdown
select {
case <-stopCh:
return
default:
}
// Create a lock
uuid, err := uuid.GenerateUUID()
if err != nil {
c.logger.Error("core: failed to generate uuid", "error", err)
return
}
lock, err := c.ha.LockWith(coreLockPath, uuid)
if err != nil {
c.logger.Error("core: failed to create lock", "error", err)
return
}
// Attempt the acquisition
leaderLostCh := c.acquireLock(lock, stopCh)
// Bail if we are being shutdown
if leaderLostCh == nil {
return
}
c.logger.Info("core: acquired lock, enabling active operation")
// This is used later to log a metrics event; this can be helpful to
// detect flapping
activeTime := time.Now()
// Grab the lock as we need it for cluster setup, which needs to happen
// before advertising;
c.stateLock.Lock()
// This block is used to wipe barrier/seal state and verify that
// everything is sane. If we have no sanity in the barrier, we actually
// seal, as there's little we can do.
{
c.seal.SetBarrierConfig(nil)
if c.seal.RecoveryKeySupported() {
c.seal.SetRecoveryConfig(nil)
}
if err := c.performKeyUpgrades(); err != nil {
// We call this in a goroutine so that we can give up the
// statelock and have this shut us down; sealInternal has a
// workflow where it watches for the stopCh to close so we want
// to return from here
go c.Shutdown()
c.logger.Error("core: error performing key upgrades", "error", err)
c.stateLock.Unlock()
lock.Unlock()
metrics.MeasureSince([]string{"core", "leadership_setup_failed"}, activeTime)
return
}
}
// Clear previous local cluster cert info so we generate new. Since the
// UUID will have changed, standbys will know to look for new info
c.clusterParamsLock.Lock()
c.localClusterCert = nil
c.localClusterParsedCert = nil
c.localClusterPrivateKey = nil
c.clusterParamsLock.Unlock()
if err := c.setupCluster(); err != nil {
c.stateLock.Unlock()
c.logger.Error("core: cluster setup failed", "error", err)
lock.Unlock()
metrics.MeasureSince([]string{"core", "leadership_setup_failed"}, activeTime)
continue
}
// Advertise as leader
if err := c.advertiseLeader(uuid, leaderLostCh); err != nil {
c.stateLock.Unlock()
c.logger.Error("core: leader advertisement setup failed", "error", err)
lock.Unlock()
metrics.MeasureSince([]string{"core", "leadership_setup_failed"}, activeTime)
continue
}
// Attempt the post-unseal process
err = c.postUnseal()
if err == nil {
c.standby = false
}
c.stateLock.Unlock()
// Handle a failure to unseal
if err != nil {
c.logger.Error("core: post-unseal setup failed", "error", err)
lock.Unlock()
metrics.MeasureSince([]string{"core", "leadership_setup_failed"}, activeTime)
continue
}
// Monitor a loss of leadership
var manualStepDown bool
select {
case <-leaderLostCh:
c.logger.Warn("core: leadership lost, stopping active operation")
case <-stopCh:
c.logger.Warn("core: stopping active operation")
case <-manualStepDownCh:
c.logger.Warn("core: stepping down from active operation to standby")
manualStepDown = true
}
metrics.MeasureSince([]string{"core", "leadership_lost"}, activeTime)
// Clear ourself as leader
if err := c.clearLeader(uuid); err != nil {
c.logger.Error("core: clearing leader advertisement failed", "error", err)
}
// Attempt the pre-seal process
c.stateLock.Lock()
c.standby = true
preSealErr := c.preSeal()
c.stateLock.Unlock()
// Give up leadership
lock.Unlock()
// Check for a failure to prepare to seal
if preSealErr != nil {
c.logger.Error("core: pre-seal teardown failed", "error", err)
}
// If we've merely stepped down, we could instantly grab the lock
// again. Give the other nodes a chance.
if manualStepDown {
time.Sleep(manualStepDownSleepPeriod)
}
}
}
// This checks the leader periodically to ensure that we switch RPC to a new
// leader pretty quickly. There is logic in Leader() already to not make this
// onerous and avoid more traffic than needed, so we just call that and ignore
// the result.
func (c *Core) periodicLeaderRefresh(doneCh, stopCh chan struct{}) {
defer close(doneCh)
for {
select {
case <-time.After(leaderCheckInterval):
c.Leader()
case <-stopCh:
return
}
}
}
// periodicCheckKeyUpgrade is used to watch for key rotation events as a standby
func (c *Core) periodicCheckKeyUpgrade(doneCh, stopCh chan struct{}) {
defer close(doneCh)
for {
select {
case <-time.After(keyRotateCheckInterval):
// Only check if we are a standby
c.stateLock.RLock()
standby := c.standby
c.stateLock.RUnlock()
if !standby {
continue
}
// Check for a poison pill. If we can read it, it means we have stale
// keys (e.g. from replication being activated) and we need to seal to
// be unsealed again.
entry, _ := c.barrier.Get(poisonPillPath)
if entry != nil && len(entry.Value) > 0 {
c.logger.Warn("core: encryption keys have changed out from underneath us (possibly due to replication enabling), must be unsealed again")
go c.Shutdown()
continue
}
if err := c.checkKeyUpgrades(); err != nil {
c.logger.Error("core: key rotation periodic upgrade check failed", "error", err)
}
case <-stopCh:
return
}
}
}
// checkKeyUpgrades is used to check if there have been any key rotations
// and if there is a chain of upgrades available
func (c *Core) checkKeyUpgrades() error {
for {
// Check for an upgrade
didUpgrade, newTerm, err := c.barrier.CheckUpgrade()
if err != nil {
return err
}
// Nothing to do if no upgrade
if !didUpgrade {
break
}
if c.logger.IsInfo() {
c.logger.Info("core: upgraded to new key term", "term", newTerm)
}
}
return nil
}
// scheduleUpgradeCleanup is used to ensure that all the upgrade paths
// are cleaned up in a timely manner if a leader failover takes place
func (c *Core) scheduleUpgradeCleanup() error {
// List the upgrades
upgrades, err := c.barrier.List(keyringUpgradePrefix)
if err != nil {
return fmt.Errorf("failed to list upgrades: %v", err)
}
// Nothing to do if no upgrades
if len(upgrades) == 0 {
return nil
}
// Schedule cleanup for all of them
time.AfterFunc(keyRotateGracePeriod, func() {
sealed, err := c.barrier.Sealed()
if err != nil {
c.logger.Warn("core: failed to check barrier status at upgrade cleanup time")
return
}
if sealed {
c.logger.Warn("core: barrier sealed at upgrade cleanup time")
return
}
for _, upgrade := range upgrades {
path := fmt.Sprintf("%s%s", keyringUpgradePrefix, upgrade)
if err := c.barrier.Delete(path); err != nil {
c.logger.Error("core: failed to cleanup upgrade", "path", path, "error", err)
}
}
})
return nil
}
func (c *Core) performKeyUpgrades() error {
if err := c.checkKeyUpgrades(); err != nil {
return errwrap.Wrapf("error checking for key upgrades: {{err}}", err)
}
if err := c.barrier.ReloadMasterKey(); err != nil {
return errwrap.Wrapf("error reloading master key: {{err}}", err)
}
if err := c.barrier.ReloadKeyring(); err != nil {
return errwrap.Wrapf("error reloading keyring: {{err}}", err)
}
if err := c.scheduleUpgradeCleanup(); err != nil {
return errwrap.Wrapf("error scheduling upgrade cleanup: {{err}}", err)
}
return nil
}
// acquireLock blocks until the lock is acquired, returning the leaderLostCh
func (c *Core) acquireLock(lock physical.Lock, stopCh <-chan struct{}) <-chan struct{} {
for {
// Attempt lock acquisition
leaderLostCh, err := lock.Lock(stopCh)
if err == nil {
return leaderLostCh
}
// Retry the acquisition
c.logger.Error("core: failed to acquire lock", "error", err)
select {
case <-time.After(lockRetryInterval):
case <-stopCh:
return nil
}
}
}
// advertiseLeader is used to advertise the current node as leader
func (c *Core) advertiseLeader(uuid string, leaderLostCh <-chan struct{}) error {
go c.cleanLeaderPrefix(uuid, leaderLostCh)
var key *ecdsa.PrivateKey
switch c.localClusterPrivateKey.(type) {
case *ecdsa.PrivateKey:
key = c.localClusterPrivateKey.(*ecdsa.PrivateKey)
default:
c.logger.Error("core: unknown cluster private key type", "key_type", fmt.Sprintf("%T", c.localClusterPrivateKey))
return fmt.Errorf("unknown cluster private key type %T", c.localClusterPrivateKey)
}
keyParams := &clusterKeyParams{
Type: corePrivateKeyTypeP521,
X: key.X,
Y: key.Y,
D: key.D,
}
adv := &activeAdvertisement{
RedirectAddr: c.redirectAddr,
ClusterAddr: c.clusterAddr,
ClusterCert: c.localClusterCert,
ClusterKeyParams: keyParams,
}
val, err := jsonutil.EncodeJSON(adv)
if err != nil {
return err
}
ent := &Entry{
Key: coreLeaderPrefix + uuid,
Value: val,
}
err = c.barrier.Put(ent)
if err != nil {
return err
}
sd, ok := c.ha.(physical.ServiceDiscovery)
if ok {
if err := sd.NotifyActiveStateChange(); err != nil {
if c.logger.IsWarn() {
c.logger.Warn("core: failed to notify active status", "error", err)
}
}
}
return nil
}
func (c *Core) cleanLeaderPrefix(uuid string, leaderLostCh <-chan struct{}) {
keys, err := c.barrier.List(coreLeaderPrefix)
if err != nil {
c.logger.Error("core: failed to list entries in core/leader", "error", err)
return
}
for len(keys) > 0 {
select {
case <-time.After(leaderPrefixCleanDelay):
if keys[0] != uuid {
c.barrier.Delete(coreLeaderPrefix + keys[0])
}
keys = keys[1:]
case <-leaderLostCh:
return
}
}
}
// clearLeader is used to clear our leadership entry
func (c *Core) clearLeader(uuid string) error {
key := coreLeaderPrefix + uuid
err := c.barrier.Delete(key)
// Advertise ourselves as a standby
sd, ok := c.ha.(physical.ServiceDiscovery)
if ok {
if err := sd.NotifyActiveStateChange(); err != nil {
if c.logger.IsWarn() {
c.logger.Warn("core: failed to notify standby status", "error", err)
}
}
}
return err
}
// emitMetrics is used to periodically expose metrics while runnig
func (c *Core) emitMetrics(stopCh chan struct{}) {
for {
select {
case <-time.After(time.Second):
c.metricsMutex.Lock()
if c.expiration != nil {
c.expiration.emitMetrics()
}
c.metricsMutex.Unlock()
case <-stopCh:
return
}
}
}
func (c *Core) ReplicationState() consts.ReplicationState {
var state consts.ReplicationState
c.clusterParamsLock.RLock()
state = c.replicationState
c.clusterParamsLock.RUnlock()
return state
}
func (c *Core) SealAccess() *SealAccess {
sa := &SealAccess{}
sa.SetSeal(c.seal)
return sa
}
func (c *Core) Logger() log.Logger {
return c.logger
}
func (c *Core) BarrierKeyLength() (min, max int) {
min, max = c.barrier.KeyLength()
max += shamir.ShareOverhead
return
}
func (c *Core) AuditedHeadersConfig() *AuditedHeadersConfig {
return c.auditedHeaders
}
func lastRemoteWALImpl(c *Core) uint64 {
return 0
}