1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
ctrl/requests_keys.go
2024-11-27 08:43:44 +01:00

471 lines
15 KiB
Go

package ctrl
import (
"bytes"
"encoding/json"
"fmt"
)
// ---
// Handler to get the public ed25519 key from a node.
func methodPublicKey(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
outCh := make(chan []byte)
go func() {
// Normally we would do some logic here, where the result is passed to outCh when done,
// so we can split up the working logic, and f.ex. sending a reply logic.
// In this case this go func and the below select is not needed, but keeping it so the
// structure is the same as the other handlers.
select {
case <-ctx.Done():
case outCh <- proc.nodeAuth.SignPublicKey:
}
}()
select {
// case proc.toRingbufferCh <- []subjectAndMessage{sam}:
case <-ctx.Done():
case out := <-outCh:
// Prepare and queue for sending a new message with the output
// of the action executed.
newReplyMessage(proc, message, out)
}
}()
// Send back an ACK message.
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
}
// ----
// Handler to get all the public ed25519 keys from a central server.
func methodKeysRequestUpdate(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
// TODO:
// - Since this is implemented as a NACK message we could implement a
// metric thats shows the last time a node did a key request.
// - We could also implement a metrics on the receiver showing the last
// time a node had done an update.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
outCh := make(chan []byte)
go func() {
// Normally we would do some logic here, where the result is passed to outCh when done,
// so we can split up the working logic, and f.ex. sending a reply logic.
// In this case this go func and the below select is not needed, but keeping it so the
// structure is the same as the other handlers.
select {
case <-ctx.Done():
case outCh <- []byte{}:
}
}()
select {
case <-ctx.Done():
// case out := <-outCh:
case <-outCh:
// Using a func here to set the scope of the lock, and then be able to
// defer the unlock when leaving that scope.
func() {
proc.centralAuth.pki.nodesAcked.mu.Lock()
defer proc.centralAuth.pki.nodesAcked.mu.Unlock()
er := fmt.Errorf(" <---- methodKeysRequestUpdate: received hash from NODE=%v, HASH=%v", message.FromNode, message.Data)
proc.errorKernel.logDebug(er)
// Check if the received hash is the same as the one currently active,
if bytes.Equal(proc.centralAuth.pki.nodesAcked.keysAndHash.Hash[:], message.Data) {
er := fmt.Errorf("info: methodKeysRequestUpdate: node %v and central have equal keys, nothing to do, exiting key update handler", message.FromNode)
// proc.errorKernel.infoSend(proc, message, er)
proc.errorKernel.logDebug(er)
return
}
er = fmt.Errorf("info: methodKeysRequestUpdate: node %v and central had not equal keys, preparing to send new version of keys", message.FromNode)
proc.errorKernel.logDebug(er)
er = fmt.Errorf("info: methodKeysRequestUpdate: marshalling new keys and hash to send: map=%v, hash=%v", proc.centralAuth.pki.nodesAcked.keysAndHash.Keys, proc.centralAuth.pki.nodesAcked.keysAndHash.Hash)
proc.errorKernel.logDebug(er)
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodKeysRequestUpdate, failed to marshal keys map: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
er = fmt.Errorf("----> methodKeysRequestUpdate: SENDING KEYS TO NODE=%v", message.FromNode)
proc.errorKernel.logDebug(er)
newReplyMessage(proc, message, b)
}()
}
}()
// We're not sending an ACK message for this request type.
return nil, nil
}
// ----
// Handler to receive the public keys from a central server.
func methodKeysDeliverUpdate(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
outCh := make(chan []byte)
go func() {
// Normally we would do some logic here, where the result is passed to outCh when done,
// so we can split up the working logic, and f.ex. sending a reply logic.
// In this case this go func and the below select is not needed, but keeping it so the
// structure is the same as the other handlers.
select {
case <-ctx.Done():
case outCh <- []byte{}:
}
}()
select {
// case proc.toRingbufferCh <- []subjectAndMessage{sam}:
case <-ctx.Done():
case <-outCh:
proc.nodeAuth.publicKeys.mu.Lock()
var keysAndHash keysAndHash
err := json.Unmarshal(message.Data, &keysAndHash)
if err != nil {
er := fmt.Errorf("error: REQKeysDeliverUpdate : json unmarshal failed: %v, message: %v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
er := fmt.Errorf("<---- REQKeysDeliverUpdate: after unmarshal, nodeAuth keysAndhash contains: %+v", keysAndHash)
proc.errorKernel.logDebug(er)
// If the received map was empty we also want to delete all the locally stored keys,
// else we copy the marshaled keysAndHash we received from central into our map.
if len(keysAndHash.Keys) < 1 {
proc.nodeAuth.publicKeys.keysAndHash = newKeysAndHash()
} else {
proc.nodeAuth.publicKeys.keysAndHash = &keysAndHash
}
proc.nodeAuth.publicKeys.mu.Unlock()
if err != nil {
er := fmt.Errorf("error: REQKeysDeliverUpdate : json unmarshal failed: %v, message: %v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
// We need to also persist the hash on the receiving nodes. We can then load
// that key upon startup.
err = proc.nodeAuth.publicKeys.saveToFile()
if err != nil {
er := fmt.Errorf("error: REQKeysDeliverUpdate : save to file failed: %v, message: %v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
// Prepare and queue for sending a new message with the output
// of the action executed.
// newReplyMessage(proc, message, out)
}
}()
// Send back an ACK message.
// ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return nil, nil
}
// ----
// Handler to allow new public keys into the database on central auth.
// Nodes will send the public key in the REQHello messages. When they
// are recived on the central server they will be put into a temp key
// map, and we need to acknowledge them before they are moved into the
// main key map, and then allowed to be sent out to other nodes.
func methodKeysAllow(proc process, message Message, node string) ([]byte, error) {
// Get a context with the timeout specified in message.MethodTimeout.
ctx, _ := getContextForMethodTimeout(proc.ctx, message)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
outCh := make(chan []byte)
go func() {
// Normally we would do some logic here, where the result is passed to outCh when done,
// so we can split up the working logic, and f.ex. sending a reply logic.
// In this case this go func and the below select is not needed, but keeping it so the
// structure is the same as the other handlers.
select {
case <-ctx.Done():
case outCh <- []byte{}:
}
}()
select {
case <-ctx.Done():
case <-outCh:
proc.centralAuth.pki.nodeNotAckedPublicKeys.mu.Lock()
defer proc.centralAuth.pki.nodeNotAckedPublicKeys.mu.Unlock()
// Range over all the MethodArgs, where each element represents a node to allow,
// and move the node from the notAcked map to the allowed map.
for _, n := range message.MethodArgs {
key, ok := proc.centralAuth.pki.nodeNotAckedPublicKeys.KeyMap[Node(n)]
if ok {
func() {
proc.centralAuth.pki.nodesAcked.mu.Lock()
defer proc.centralAuth.pki.nodesAcked.mu.Unlock()
// Store/update the node and public key on the allowed pubKey map.
proc.centralAuth.pki.nodesAcked.keysAndHash.Keys[Node(n)] = key
}()
// Add key to persistent storage.
proc.centralAuth.pki.dbUpdatePublicKey(string(n), key)
// Delete the key from the NotAcked map
delete(proc.centralAuth.pki.nodeNotAckedPublicKeys.KeyMap, Node(n))
er := fmt.Errorf("info: REQKeysAllow : allowed new/updated public key for %v to allowed public key map", n)
proc.errorKernel.infoSend(proc, message, er)
}
}
// All new elements are now added, and we can create a new hash
// representing the current keys in the allowed map.
proc.centralAuth.updateHash(proc, message)
// If new keys were allowed into the main map, we should send out one
// single update to all the registered nodes to inform of an update.
// If a node is not reachable at the time the update is sent it is
// not a problem since the nodes will periodically check for updates.
//
// If there are errors we will return from the function, and send no
// updates.
err := pushKeys(proc, message, []Node{})
if err != nil {
proc.errorKernel.errSend(proc, message, err, logWarning)
return
}
}
}()
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
}
// pushKeys will try to push the the current keys and hashes to each node once.
// As input arguments it takes the current process, the current message, and a
// []Node in nodes.
// The nodes are used when a node or nodes have been deleted from the current
// nodesAcked map since it will contain the nodes that were deleted so we are
// also able to send an update to them as well.
func pushKeys(proc process, message Message, nodes []Node) error {
er := fmt.Errorf("info: beginning of pushKeys, nodes=%v", nodes)
var knh []byte
proc.errorKernel.logDebug(er)
err := func() error {
proc.centralAuth.pki.nodesAcked.mu.Lock()
defer proc.centralAuth.pki.nodesAcked.mu.Unlock()
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodKeysAllow, failed to marshal keys map: %v", err)
return er
}
copy(knh, b)
return nil
}()
if err != nil {
return err
}
// proc.centralAuth.pki.nodeNotAckedPublicKeys.mu.Lock()
// defer proc.centralAuth.pki.nodeNotAckedPublicKeys.mu.Unlock()
// For all nodes that is not ack'ed we try to send an update once.
for n := range proc.centralAuth.pki.nodeNotAckedPublicKeys.KeyMap {
er := fmt.Errorf("info: node to send REQKeysDeliverUpdate to:%v ", n)
proc.errorKernel.logDebug(er)
msg := Message{
ToNode: n,
Method: KeysDeliverUpdate,
ReplyMethod: None,
ACKTimeout: 0,
}
sam, err := newSubjectAndMessage(msg)
if err != nil {
// In theory the system should drop the message before it reaches here.
er := fmt.Errorf("error: newSubjectAndMessage : %v, message: %v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
proc.newMessagesCh <- sam
er = fmt.Errorf("----> methodKeysAllow: SENDING KEYS TO NODE=%v", message.FromNode)
proc.errorKernel.logDebug(er)
}
// Create the data payload of the current allowed keys.
b, err := json.Marshal(proc.centralAuth.pki.nodesAcked.keysAndHash)
if err != nil {
er := fmt.Errorf("error: methodKeysAllow, failed to marshal keys map: %v", err)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
proc.centralAuth.pki.nodesAcked.mu.Lock()
defer proc.centralAuth.pki.nodesAcked.mu.Unlock()
// Concatenate the current nodes in the keysAndHash map and the nodes
// we got from the function argument when this function was called.
nodeMap := make(map[Node]struct{})
for n := range proc.centralAuth.pki.nodesAcked.keysAndHash.Keys {
nodeMap[n] = struct{}{}
}
for _, n := range nodes {
nodeMap[n] = struct{}{}
}
// For all nodes that is ack'ed we try to send an update once.
for n := range nodeMap {
er := fmt.Errorf("info: node to send REQKeysDeliverUpdate to:%v ", n)
proc.errorKernel.logDebug(er)
msg := Message{
ToNode: n,
Method: KeysDeliverUpdate,
Data: b,
ReplyMethod: None,
ACKTimeout: 0,
}
sam, err := newSubjectAndMessage(msg)
if err != nil {
// In theory the system should drop the message before it reaches here.
er := fmt.Errorf("error: newSubjectAndMessage : %v, message: %v", err, message)
proc.errorKernel.errSend(proc, message, er, logWarning)
}
proc.newMessagesCh <- sam
er = fmt.Errorf("----> methodKeysAllow: sending keys update to node=%v", message.FromNode)
proc.errorKernel.logDebug(er)
}
return nil
}
func methodKeysDelete(proc process, message Message, node string) ([]byte, error) {
inf := fmt.Errorf("<--- methodKeysDelete received from: %v, containing: %v", message.FromNode, message.MethodArgs)
proc.errorKernel.logDebug(inf)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
// Get a context with the timeout specified in message.MethodTimeout.
ctx, cancel := getContextForMethodTimeout(proc.ctx, message)
outCh := make(chan []byte)
errCh := make(chan error)
proc.processes.wg.Add(1)
go func() {
defer proc.processes.wg.Done()
switch {
case len(message.MethodArgs) < 1:
errCh <- fmt.Errorf("error: methodAclGroupNodesDeleteNode: got <1 number methodArgs, want >0")
return
}
// HERE:
// We want to be able to define more nodes keys to delete.
// Loop over the methodArgs, and call the keyDelete function for each node,
// or rewrite the deleteKey to deleteKeys and it takes a []node as input
// so all keys can be deleted in 1 go, and we create 1 new hash, instead
// of doing it for each node delete.
proc.centralAuth.deletePublicKeys(proc, message, message.MethodArgs)
er := fmt.Errorf("info: Deleted public keys: %v", message.MethodArgs)
proc.errorKernel.logDebug(er)
// All new elements are now added, and we can create a new hash
// representing the current keys in the allowed map.
proc.centralAuth.updateHash(proc, message)
er = fmt.Errorf(" * DEBUG updated hash for public keys")
proc.errorKernel.logDebug(er)
var nodes []Node
for _, n := range message.MethodArgs {
nodes = append(nodes, Node(n))
}
err := pushKeys(proc, message, nodes)
if err != nil {
proc.errorKernel.errSend(proc, message, err, logWarning)
return
}
outString := fmt.Sprintf("deleted public keys for the nodes=%v\n", message.MethodArgs)
out := []byte(outString)
select {
case outCh <- out:
case <-ctx.Done():
return
}
}()
select {
case err := <-errCh:
proc.errorKernel.errSend(proc, message, err, logWarning)
case <-ctx.Done():
cancel()
er := fmt.Errorf("error: methodAclGroupNodesDeleteNode: method timed out: %v", message.MethodArgs)
proc.errorKernel.errSend(proc, message, er, logWarning)
case out := <-outCh:
newReplyMessage(proc, message, out)
}
}()
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
}