constellation/coordinator/storewrapper/storewrapper.go

package storewrapper

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"net/netip"
	"strconv"
	"strings"

	"github.com/edgelesssys/constellation/coordinator/peer"
	"github.com/edgelesssys/constellation/coordinator/state"
	"github.com/edgelesssys/constellation/coordinator/store"
	kubeadm "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1beta3"
)

// variables which will be used as a store-prefix start with prefix[...].
// variables which will be used as a store-key start with key[...].
const (
	keyHighestAvailableCoordinatorIP = "highestAvailableCoordinatorIP"
	keyHighestAvailableNodeIP        = "highestAvailableNodeIP"
	keyKubernetesJoinCommand         = "kubeJoin"
	keyPeersResourceVersion          = "peersResourceVersion"
	keyMasterSecret                  = "masterSecret"
	keyKubeConfig                    = "kubeConfig"
	keyClusterID                     = "clusterID"
	keyVPNPubKey                     = "vpnKey"
	keyKEKID                         = "kekID"
	prefixFreeCoordinatorIPs         = "freeCoordinatorVPNIPs"
	prefixAdminLocation              = "externalAdminsData"
	prefixPeerLocation               = "peerPrefix"
	prefixFreeNodeIPs                = "freeNodeVPNIPs"
)

var (
	coordinatorIPRangeStart = netip.AddrFrom4([4]byte{10, 118, 0, 1})
	coordinatorIPRangeEnd   = netip.AddrFrom4([4]byte{10, 118, 0, 10})
	nodeIPRangeStart        = netip.AddrFrom4([4]byte{10, 118, 0, 11})
	nodeIPRangeEnd          = netip.AddrFrom4([4]byte{10, 118, 255, 254})
)

// StoreWrapper is a wrapper for the store interface.
type StoreWrapper struct {
	Store interface {
		Get(string) ([]byte, error)
		Put(string, []byte) error
		Delete(string) error
		Iterator(string) (store.Iterator, error)
	}
}

// GetState returns the state from store.
func (s StoreWrapper) GetState() (state.State, error) {
	rawState, err := s.Store.Get("state")
	if err != nil {
		return 0, err
	}

	currState, err := strconv.Atoi(string(rawState))
	if err != nil {
		return 0, err
	}

	return state.State(currState), nil
}

// PutState saves the state to store.
func (s StoreWrapper) PutState(currState state.State) error {
	rawState := []byte(strconv.Itoa(int(currState)))
	return s.Store.Put("state", rawState)
}

// GetVPNKey returns the VPN pubKey from Store.
func (s StoreWrapper) GetVPNKey() ([]byte, error) {
	return s.Store.Get(keyVPNPubKey)
}

// PutVPNKey saves the VPN pubKey to store.
func (s StoreWrapper) PutVPNKey(key []byte) error {
	return s.Store.Put(keyVPNPubKey, key)
}

// PutPeer puts a single peer in the store, with a unique key derived form the VPNIP.
func (s StoreWrapper) PutPeer(peer peer.Peer) error {
	if len(peer.VPNIP) == 0 {
		return errors.New("unique ID of peer not set")
	}
	jsonPeer, err := json.Marshal(peer)
	if err != nil {
		return err
	}
	return s.Store.Put(prefixPeerLocation+peer.VPNIP, jsonPeer)
}

// RemovePeer removes a peer from the store.
func (s StoreWrapper) RemovePeer(peer peer.Peer) error {
	return s.Store.Delete(prefixPeerLocation + peer.VPNIP)
}

// GetPeer returns a peer requested by the given VPN IP address.
func (s StoreWrapper) GetPeer(vpnIP string) (peer.Peer, error) {
	bytePeer, err := s.Store.Get(prefixPeerLocation + vpnIP)
	if err != nil {
		return peer.Peer{}, err
	}
	var peer peer.Peer
	err = json.Unmarshal(bytePeer, &peer)
	return peer, err
}

// GetPeers returns all peers in the store.
func (s StoreWrapper) GetPeers() ([]peer.Peer, error) {
	return s.getPeersByPrefix(prefixPeerLocation)
}

// IncrementPeersResourceVersion increments the version of the stored peers.
// Should be called in a transaction together with Add/Remove operation(s).
func (s StoreWrapper) IncrementPeersResourceVersion() error {
	val, err := s.GetPeersResourceVersion()
	var unsetErr *store.ValueUnsetError
	if errors.As(err, &unsetErr) {
		val = 0
	} else if err != nil {
		return err
	}
	return s.Store.Put(keyPeersResourceVersion, []byte(strconv.Itoa(val+1)))
}

// GetPeersResourceVersion returns the current version of the stored peers.
func (s StoreWrapper) GetPeersResourceVersion() (int, error) {
	raw, err := s.Store.Get(keyPeersResourceVersion)
	if err != nil {
		return 0, err
	}
	val, err := strconv.Atoi(string(raw))
	if err != nil {
		return 0, err
	}
	return val, nil
}

// UpdatePeers synchronizes the stored peers with the passed peers, returning added and removed peers.
func (s StoreWrapper) UpdatePeers(peers []peer.Peer) (added, removed []peer.Peer, err error) {
	// convert to map for easier lookup
	updatedPeers := make(map[string]peer.Peer)
	for _, p := range peers {
		updatedPeers[p.VPNIP] = p
	}

	it, err := s.Store.Iterator(prefixPeerLocation)
	if err != nil {
		return nil, nil, err
	}

	// collect peers that need to be added or removed
	for it.HasNext() {
		key, err := it.GetNext()
		if err != nil {
			return nil, nil, err
		}
		val, err := s.Store.Get(key)
		if err != nil {
			return nil, nil, err
		}
		var storedPeer peer.Peer
		if err := json.Unmarshal(val, &storedPeer); err != nil {
			return nil, nil, err
		}

		if updPeer, ok := updatedPeers[storedPeer.VPNIP]; ok {
			if updPeer.PublicEndpoint != storedPeer.PublicEndpoint || !bytes.Equal(updPeer.VPNPubKey, storedPeer.VPNPubKey) {
				// stored peer must be updated, so mark for addition AND removal
				added = append(added, updPeer)
				removed = append(removed, storedPeer)
			}
			delete(updatedPeers, updPeer.VPNIP)
		} else {
			// stored peer is not contained in the updated peers, so mark for removal
			removed = append(removed, storedPeer)
		}
	}

	// remaining updated peers were not in the store, so mark for addition
	for _, p := range updatedPeers {
		added = append(added, p)
	}

	// perform remove and add
	for _, p := range removed {
		if err := s.Store.Delete(prefixPeerLocation + p.VPNIP); err != nil {
			return nil, nil, err
		}
	}
	for _, p := range added {
		data, err := json.Marshal(p)
		if err != nil {
			return nil, nil, err
		}
		if err := s.Store.Put(prefixPeerLocation+p.VPNIP, data); err != nil {
			return nil, nil, err
		}
	}

	return added, removed, nil
}

// PutAdmin puts a single admin in the store, with a unique key derived form the VPNIP.
func (s StoreWrapper) PutAdmin(peer peer.AdminData) error {
	jsonPeer, err := json.Marshal(peer)
	if err != nil {
		return err
	}
	return s.Store.Put(prefixAdminLocation+peer.VPNIP, jsonPeer)
}

// GetAdmin gets a single admin from the store.
// TODO: extend if we want to have multiple admins.
func (s StoreWrapper) GetAdmin() (peer.AdminData, error) {
	iter, err := s.Store.Iterator(prefixAdminLocation)
	if err != nil {
		return peer.AdminData{}, err
	}
	key, err := iter.GetNext()
	if err != nil {
		return peer.AdminData{}, err
	}
	value, err := s.Store.Get(key)
	if err != nil {
		return peer.AdminData{}, err
	}
	var adminData peer.AdminData
	if err := json.Unmarshal(value, &adminData); err != nil {
		return peer.AdminData{}, err
	}
	return adminData, nil
}

func (s StoreWrapper) getPeersByPrefix(prefix string) ([]peer.Peer, error) {
	peerKeys, err := s.Store.Iterator(prefix)
	if err != nil {
		return nil, err
	}
	var peers []peer.Peer
	for peerKeys.HasNext() {
		storeKey, err := peerKeys.GetNext()
		if err != nil {
			return nil, err
		}
		marshalPeer, err := s.Store.Get(storeKey)
		if err != nil {
			return nil, err
		}
		var peer peer.Peer
		if err := json.Unmarshal(marshalPeer, &peer); err != nil {
			return nil, err
		}
		peers = append(peers, peer)

	}
	return peers, nil
}

// GetKubernetesJoinArgs returns the Kubernetes join command from store.
func (s StoreWrapper) GetKubernetesJoinArgs() (*kubeadm.BootstrapTokenDiscovery, error) {
	rawJoinCommand, err := s.Store.Get(keyKubernetesJoinCommand)
	if err != nil {
		return nil, err
	}
	joinCommand := kubeadm.BootstrapTokenDiscovery{}
	if err := json.Unmarshal(rawJoinCommand, &joinCommand); err != nil {
		return nil, err
	}
	return &joinCommand, nil
}

// PutKubernetesJoinArgs saves the Kubernetes join command to store.
func (s StoreWrapper) PutKubernetesJoinArgs(args *kubeadm.BootstrapTokenDiscovery) error {
	j, err := json.Marshal(args)
	if err != nil {
		return err
	}
	return s.Store.Put(keyKubernetesJoinCommand, j)
}

// GetKubernetesConfig returns the Kubernetes kubeconfig file to authenticate with the kubernetes API.
func (s StoreWrapper) GetKubernetesConfig() ([]byte, error) {
	return s.Store.Get(keyKubeConfig)
}

// PutKubernetesConfig saves the Kubernetes kubeconfig file command to store.
func (s StoreWrapper) PutKubernetesConfig(kubeConfig []byte) error {
	return s.Store.Put(keyKubeConfig, kubeConfig)
}

// GetMasterSecret returns the Constellation master secret from store.
func (s StoreWrapper) GetMasterSecret() ([]byte, error) {
	return s.Store.Get(keyMasterSecret)
}

// PutMasterSecret saves the Constellation master secret to store.
func (s StoreWrapper) PutMasterSecret(masterSecret []byte) error {
	return s.Store.Put(keyMasterSecret, masterSecret)
}

// GetKEKID returns the key encryption key ID from store.
func (s StoreWrapper) GetKEKID() (string, error) {
	kekID, err := s.Store.Get(keyKEKID)
	return string(kekID), err
}

// PutKEKID saves the key encryption key ID to store.
func (s StoreWrapper) PutKEKID(kekID string) error {
	return s.Store.Put(keyKEKID, []byte(kekID))
}

// GetClusterID returns the unique identifier of the cluster from store.
func (s StoreWrapper) GetClusterID() ([]byte, error) {
	return s.Store.Get(keyClusterID)
}

// PutClusterID saves the unique identifier of the cluster to store.
func (s StoreWrapper) PutClusterID(clusterID []byte) error {
	return s.Store.Put(keyClusterID, clusterID)
}

func (s StoreWrapper) InitializeStoreIPs() error {
	if err := s.PutNextNodeIP(nodeIPRangeStart); err != nil {
		return err
	}
	return s.PutNextCoordinatorIP(coordinatorIPRangeStart)
}

// PutNextCoordinatorIP puts the last used ip into the store.
func (s StoreWrapper) PutNextCoordinatorIP(ip netip.Addr) error {
	return s.Store.Put(keyHighestAvailableCoordinatorIP, ip.AsSlice())
}

// getNextCoordinatorIP generates addresses from a /16 subnet.
func (s StoreWrapper) getNextCoordinatorIP() (netip.Addr, error) {
	byteIP, err := s.Store.Get(keyHighestAvailableCoordinatorIP)
	if err != nil {
		return netip.Addr{}, errors.New("could not obtain IP from store")
	}
	ip, ok := netip.AddrFromSlice(byteIP)
	if !ok {
		return netip.Addr{}, fmt.Errorf("ip addr malformed %v", byteIP)
	}
	if !ip.IsValid() || ip.Compare(coordinatorIPRangeEnd) == 1 {
		return netip.Addr{}, errors.New("no ips left to assign")
	}
	nextIP := ip.Next()
	if err := s.PutNextCoordinatorIP(nextIP); err != nil {
		return netip.Addr{}, errors.New("could not put IP to store")
	}
	return ip, nil
}

// PopNextFreeCoordinatorIP return the next free IP, these could be a old one from a removed peer
// or a newly generated IP.
func (s StoreWrapper) PopNextFreeCoordinatorIP() (netip.Addr, error) {
	vpnIP, err := s.getFreedVPNIP(prefixFreeCoordinatorIPs)
	var noElementsError *store.NoElementsLeftError
	if errors.As(err, &noElementsError) {
		return s.getNextCoordinatorIP()
	}
	if err != nil {
		return netip.Addr{}, err
	}
	return vpnIP, nil
}

// PutFreedCoordinatorVPNIP puts a already generated VPNIP (IP < highestAvailableCoordinatorIP ),
// which is currently unused, into the store.
// The IP is saved at a specific prefix and will be used with priority when we
// request a new Coordinator IP.
func (s StoreWrapper) PutFreedCoordinatorVPNIP(vpnIP string) error {
	return s.Store.Put(prefixFreeCoordinatorIPs+vpnIP, nil)
}

// PutNextNodeIP puts the last used ip into the store.
func (s StoreWrapper) PutNextNodeIP(ip netip.Addr) error {
	return s.Store.Put(keyHighestAvailableNodeIP, ip.AsSlice())
}

// getNextNodeIP generates addresses from a /16 subnet.
func (s StoreWrapper) getNextNodeIP() (netip.Addr, error) {
	byteIP, err := s.Store.Get(keyHighestAvailableNodeIP)
	if err != nil {
		return netip.Addr{}, errors.New("could not obtain IP from store")
	}
	ip, ok := netip.AddrFromSlice(byteIP)
	if !ok {
		return netip.Addr{}, fmt.Errorf("ip addr malformed %v", byteIP)
	}
	if !ip.IsValid() || ip.Compare(nodeIPRangeEnd) == 1 {
		return netip.Addr{}, errors.New("no ips left to assign")
	}
	nextIP := ip.Next()

	if err := s.PutNextNodeIP(nextIP); err != nil {
		return netip.Addr{}, errors.New("could not put IP to store")
	}
	return ip, nil
}

// PopNextFreeNodeIP return the next free IP, these could be a old one from a removed peer
// or a newly generated IP.
func (s StoreWrapper) PopNextFreeNodeIP() (netip.Addr, error) {
	vpnIP, err := s.getFreedVPNIP(prefixFreeNodeIPs)
	var noElementsError *store.NoElementsLeftError
	if errors.As(err, &noElementsError) {
		return s.getNextNodeIP()
	}
	if err != nil {
		return netip.Addr{}, err
	}
	return vpnIP, nil
}

// PutFreedNodeVPNIP puts a already generated VPNIP (IP < highestAvailableNodeIP ),
// which is currently unused, into the store.
// The IP is saved at a specific prefix and will be used with priority when we
// request a new Node IP.
func (s StoreWrapper) PutFreedNodeVPNIP(vpnIP string) error {
	return s.Store.Put(prefixFreeNodeIPs+vpnIP, nil)
}

// getFreedVPNIP reclaims a VPNIP from the store and removes it from there.
func (s StoreWrapper) getFreedVPNIP(prefix string) (netip.Addr, error) {
	iter, err := s.Store.Iterator(prefix)
	if err != nil {
		return netip.Addr{}, err
	}
	vpnIPWithPrefix, err := iter.GetNext()
	if err != nil {
		return netip.Addr{}, err
	}
	stringVPNIP := strings.TrimPrefix(vpnIPWithPrefix, prefix)
	vpnIP, err := netip.ParseAddr(stringVPNIP)
	if err != nil {
		return netip.Addr{}, fmt.Errorf("ip addr malformed %v, %w", stringVPNIP, err)
	}

	return vpnIP, s.Store.Delete(vpnIPWithPrefix)
}