mirrors/ctrl
mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
Code Activity

moved signatures into signatures.go

Browse source
This commit is contained in:
postmannen 2022-02-07 04:23:13 +01:00
parent 426b98941e
commit b3630f62c4
5 changed files with 215 additions and 202 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
process.go
View file

@ -114,6 +114,7 @@ func newProcess(ctx context.Context, metrics *metrics, natsConn *nats.Conn, proc
ctx: ctx,
ctxCancel: cancel,
startup: newStartup(metrics, signatures),
signatures: signatures,
}
return proc
@ -514,7 +515,7 @@ func (p process) messageSubscriberHandler(natsConn *nats.Conn, thisNode string,
out := []byte{}
var err error
if p.verifySignature(message) {
if p.signatures.verifySignature(message) {
// Call the method handler for the specified method.
out, err = mh.handler(p, message, thisNode)
@ -535,7 +536,7 @@ func (p process) messageSubscriberHandler(natsConn *nats.Conn, thisNode string,
p.processes.errorKernel.errSend(p, message, er)
}
if p.verifySignature(message) {
if p.signatures.verifySignature(message) {
_, err := mf.handler(p, message, thisNode)
@ -552,22 +553,6 @@ func (p process) messageSubscriberHandler(natsConn *nats.Conn, thisNode string,
}
}
// verifySignature
func (p process) verifySignature(m Message) bool {
if p.configuration.AllowEmptySignature {
fmt.Printf(" * verifySignature: AllowEmptySignature set to TRUE\n")
return true
}
// Verify if the signature matches.
argsStringified := argsToString(m.MethodArgs)
ok := ed25519.Verify(p.processes.SignPublicKey, []byte(argsStringified), m.ArgSignature)
fmt.Printf(" * verifySignature, result: %v, fromNode: %v, method: %v, signature: %s\n", ok, m.FromNode, m.Method, m.ArgSignature)
return ok
}
// argsToString takes args in the format of []string and returns a string.
func argsToString(args []string) string {
return strings.Join(args, " ")
@ -641,7 +626,7 @@ func (p process) publishMessages(natsConn *nats.Conn) {
func (p process) addMethodArgSignature(m Message) []byte {
argsString := argsToString(m.MethodArgs)
sign := ed25519.Sign(p.processes.SignPrivateKey, []byte(argsString))
sign := ed25519.Sign(p.signatures.SignPrivateKey, []byte(argsString))
return sign
}

144
processes.go
View file

@ -2,13 +2,8 @@ package steward
import (
"context"
"crypto/ed25519"
"encoding/base64"
"fmt"
"io/ioutil"
"log"
"os"
"path/filepath"
"sync"
"time"
@ -38,18 +33,6 @@ type processes struct {
// configuration
configuration *Configuration
// Full path to the signing keys folder
SignKeyFolder string
// Full path to private signing key.
SignKeyPrivateKeyPath string
// Full path to public signing key.
SignKeyPublicKeyPath string
// private key for ed25519 signing.
SignPrivateKey []byte
// public key for ed25519 signing.
SignPublicKey []byte
// Signatures
Signatures *signatures
}
@ -62,6 +45,7 @@ func newProcesses(ctx context.Context, metrics *metrics, tui *tui, errorKernel *
tui: tui,
errorKernel: errorKernel,
configuration: configuration,
Signatures: signatures,
}
// Prepare the parent context for the subscribers.
@ -77,135 +61,11 @@ func newProcesses(ctx context.Context, metrics *metrics, tui *tui, errorKernel *
p.metrics = metrics
// Set the signing key paths.
p.SignKeyFolder = filepath.Join(p.configuration.ConfigFolder, "signing")
p.SignKeyPrivateKeyPath = filepath.Join(p.SignKeyFolder, "private.key")
p.SignKeyPublicKeyPath = filepath.Join(p.SignKeyFolder, "public.key")
return &p
}
// ----------------------
// loadSigningKeys will try to load the ed25519 signing keys. If the
// files are not found new keys will be generated and written to disk.
func (p *processes) loadSigningKeys(initProc process) error {
// Check if folder structure exist, if not create it.
if _, err := os.Stat(p.SignKeyFolder); os.IsNotExist(err) {
err := os.MkdirAll(p.SignKeyFolder, 0700)
if err != nil {
er := fmt.Errorf("error: failed to create directory for signing keys : %v", err)
return er
}
}
// Check if there already are any keys in the etc folder.
foundKey := false
if _, err := os.Stat(p.SignKeyPublicKeyPath); !os.IsNotExist(err) {
foundKey = true
}
if _, err := os.Stat(p.SignKeyPrivateKeyPath); !os.IsNotExist(err) {
foundKey = true
}
// If no keys where found generete a new pair, load them into the
// processes struct fields, and write them to disk.
if !foundKey {
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
er := fmt.Errorf("error: failed to generate ed25519 keys for signing: %v", err)
return er
}
pubB64string := base64.RawStdEncoding.EncodeToString(pub)
privB64string := base64.RawStdEncoding.EncodeToString(priv)
// Write public key to file.
err = p.writeSigningKey(p.SignKeyPublicKeyPath, pubB64string)
if err != nil {
return err
}
// Write private key to file.
err = p.writeSigningKey(p.SignKeyPrivateKeyPath, privB64string)
if err != nil {
return err
}
// Also store the keys in the processes structure so we can
// reference them from there when we need them.
p.SignPublicKey = pub
p.SignPrivateKey = priv
er := fmt.Errorf("info: no signing keys found, generating new keys")
p.errorKernel.errSend(initProc, Message{}, er)
// We got the new generated keys now, so we can return.
return nil
}
// Key files found, load them into the processes struct fields.
pubKey, _, err := p.readKeyFile(p.SignKeyPublicKeyPath)
if err != nil {
return err
}
p.SignPublicKey = pubKey
privKey, _, err := p.readKeyFile(p.SignKeyPrivateKeyPath)
if err != nil {
return err
}
p.SignPublicKey = pubKey
p.SignPrivateKey = privKey
return nil
}
// readKeyFile will take the path of a key file as input, read the base64
// encoded data, decode the data. It will return the raw data as []byte,
// the base64 encoded data, and any eventual error.
func (p *processes) readKeyFile(keyFile string) (ed2519key []byte, b64Key []byte, err error) {
fh, err := os.Open(keyFile)
if err != nil {
er := fmt.Errorf("error: failed to open key file: %v", err)
return nil, nil, er
}
defer fh.Close()
b, err := ioutil.ReadAll(fh)
if err != nil {
er := fmt.Errorf("error: failed to read key file: %v", err)
return nil, nil, er
}
key, err := base64.RawStdEncoding.DecodeString(string(b))
if err != nil {
er := fmt.Errorf("error: failed to base64 decode key data: %v", err)
return nil, nil, er
}
return key, b, nil
}
// writeSigningKey will write the base64 encoded signing key to file.
func (p *processes) writeSigningKey(realPath string, keyB64 string) error {
fh, err := os.OpenFile(realPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
er := fmt.Errorf("error: failed to open key file for writing: %v", err)
return er
}
defer fh.Close()
_, err = fh.Write([]byte(keyB64))
if err != nil {
er := fmt.Errorf("error: failed to write key to file: %v", err)
return er
}
return nil
}
// ----------------------
type procsMap struct {
@ -351,7 +211,7 @@ type startup struct {
}
func newStartup(metrics *metrics, signatures *signatures) *startup {
s := startup{metrics: metrics}
s := startup{metrics: metrics, Signatures: signatures}
return &s
}

69
server.go
View file

@ -60,17 +60,17 @@ type server struct {
// processInitial is the initial process that all other processes are tied to.
processInitial process
// Signatures holds all the signatures,
// signatures holds all the signatures,
// and the public keys
Signatures *signatures
signatures *signatures
}
// newServer will prepare and return a server type
func NewServer(c *Configuration, version string) (*server, error) {
func NewServer(configuration *Configuration, version string) (*server, error) {
// Set up the main background context.
ctx, cancel := context.WithCancel(context.Background())
metrics := newMetrics(c.PromHostAndPort)
metrics := newMetrics(configuration.PromHostAndPort)
// Start the error kernel that will do all the error handling
// that is not done within a process.
@ -78,14 +78,14 @@ func NewServer(c *Configuration, version string) (*server, error) {
var opt nats.Option
if c.RootCAPath != "" {
opt = nats.RootCAs(c.RootCAPath)
if configuration.RootCAPath != "" {
opt = nats.RootCAs(configuration.RootCAPath)
}
if c.NkeySeedFile != "" {
if configuration.NkeySeedFile != "" {
var err error
opt, err = nats.NkeyOptionFromSeed(c.NkeySeedFile)
opt, err = nats.NkeyOptionFromSeed(configuration.NkeySeedFile)
if err != nil {
cancel()
return nil, fmt.Errorf("error: failed to read nkey seed file: %v", err)
@ -98,16 +98,16 @@ func NewServer(c *Configuration, version string) (*server, error) {
for {
var err error
// Setting MaxReconnects to -1 which equals unlimited.
conn, err = nats.Connect(c.BrokerAddress,
conn, err = nats.Connect(configuration.BrokerAddress,
opt,
nats.MaxReconnects(-1),
nats.ReconnectJitter(time.Duration(c.NatsReconnectJitter)*time.Millisecond, time.Duration(c.NatsReconnectJitterTLS)*time.Second),
nats.Timeout(time.Second*time.Duration(c.NatsConnOptTimeout)),
nats.ReconnectJitter(time.Duration(configuration.NatsReconnectJitter)*time.Millisecond, time.Duration(configuration.NatsReconnectJitterTLS)*time.Second),
nats.Timeout(time.Second*time.Duration(configuration.NatsConnOptTimeout)),
)
// If no servers where available, we loop and retry until succesful.
if err != nil {
log.Printf("error: could not connect, waiting %v seconds, and retrying: %v\n", c.NatsConnectRetryInterval, err)
time.Sleep(time.Duration(time.Second * time.Duration(c.NatsConnectRetryInterval)))
log.Printf("error: could not connect, waiting %v seconds, and retrying: %v\n", configuration.NatsConnectRetryInterval, err)
time.Sleep(time.Duration(time.Second * time.Duration(configuration.NatsConnectRetryInterval)))
continue
}
@ -121,8 +121,8 @@ func NewServer(c *Configuration, version string) (*server, error) {
var err error
// Open the steward socket file, and start the listener if enabled.
if c.EnableSocket {
stewardSocket, err = createSocket(c.SocketFolder, "steward.sock")
if configuration.EnableSocket {
stewardSocket, err = createSocket(configuration.SocketFolder, "steward.sock")
if err != nil {
cancel()
return nil, err
@ -131,8 +131,8 @@ func NewServer(c *Configuration, version string) (*server, error) {
// Create the tui client structure if enabled.
var tuiClient *tui
if c.EnableTUI {
tuiClient, err = newTui(Node(c.NodeName))
if configuration.EnableTUI {
tuiClient, err = newTui(Node(configuration.NodeName))
if err != nil {
cancel()
return nil, err
@ -140,36 +140,37 @@ func NewServer(c *Configuration, version string) (*server, error) {
}
signatures := newSignatures()
signatures := newSignatures(configuration, errorKernel)
fmt.Printf(" * DEBUG: newServer: signatures contains: %+v\n", signatures)
s := &server{
ctx: ctx,
cancel: cancel,
configuration: c,
nodeName: c.NodeName,
configuration: configuration,
nodeName: configuration.NodeName,
natsConn: conn,
StewardSocket: stewardSocket,
processes: newProcesses(ctx, metrics, tuiClient, errorKernel, c, signatures),
processes: newProcesses(ctx, metrics, tuiClient, errorKernel, configuration, signatures),
ringBufferBulkInCh: make(chan []subjectAndMessage),
metrics: metrics,
version: version,
tui: tuiClient,
errorKernel: errorKernel,
Signatures: signatures,
signatures: signatures,
}
// Create the default data folder for where subscribers should
// write it's data, check if data folder exist, and create it if needed.
if _, err := os.Stat(c.SubscribersDataFolder); os.IsNotExist(err) {
if c.SubscribersDataFolder == "" {
return nil, fmt.Errorf("error: subscribersDataFolder value is empty, you need to provide the config or the flag value at startup %v: %v", c.SubscribersDataFolder, err)
if _, err := os.Stat(configuration.SubscribersDataFolder); os.IsNotExist(err) {
if configuration.SubscribersDataFolder == "" {
return nil, fmt.Errorf("error: subscribersDataFolder value is empty, you need to provide the config or the flag value at startup %v: %v", configuration.SubscribersDataFolder, err)
}
err := os.Mkdir(c.SubscribersDataFolder, 0700)
err := os.Mkdir(configuration.SubscribersDataFolder, 0700)
if err != nil {
return nil, fmt.Errorf("error: failed to create data folder directory %v: %v", c.SubscribersDataFolder, err)
return nil, fmt.Errorf("error: failed to create data folder directory %v: %v", configuration.SubscribersDataFolder, err)
}
log.Printf("info: Creating subscribers data folder at %v\n", c.SubscribersDataFolder)
log.Printf("info: Creating subscribers data folder at %v\n", configuration.SubscribersDataFolder)
}
return s, nil
@ -253,18 +254,10 @@ func (s *server) Start() {
//
// NB: The context of the initial process are set in processes.Start.
sub := newSubject(REQInitial, s.nodeName)
s.processInitial = newProcess(context.TODO(), s.metrics, s.natsConn, s.processes, s.ringBufferBulkInCh, s.configuration, sub, s.errorKernel.errorCh, "", nil, s.Signatures)
s.processInitial = newProcess(context.TODO(), s.metrics, s.natsConn, s.processes, s.ringBufferBulkInCh, s.configuration, sub, s.errorKernel.errorCh, "", nil, s.signatures)
// Start all wanted subscriber processes.
s.processes.Start(s.processInitial)
// We need the initial process to be able to send error messages so
// we have to load the signing keys here.
err := s.processes.loadSigningKeys(s.processInitial)
if err != nil {
log.Printf("%v\n", err)
os.Exit(1)
}
time.Sleep(time.Second * 1)
s.processes.printProcessesMap()
@ -486,7 +479,7 @@ func (s *server) routeMessagesToProcess(dbFileName string) {
// log.Printf("info: processNewMessages: did not find that specific subject, starting new process for subject: %v\n", subjName)
sub := newSubject(sam.Subject.Method, sam.Subject.ToNode)
proc := newProcess(s.ctx, s.metrics, s.natsConn, s.processes, s.ringBufferBulkInCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, nil, s.Signatures)
proc := newProcess(s.ctx, s.metrics, s.natsConn, s.processes, s.ringBufferBulkInCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, nil, s.signatures)
proc.spawnWorker(s.processes, s.natsConn)
// log.Printf("info: processNewMessages: new process started, subject: %v, processID: %v\n", subjName, proc.processID)

180
signatures.go
View file

@ -1,6 +1,15 @@
package steward
import "sync"
import (
"crypto/ed25519"
"encoding/base64"
"fmt"
"io/ioutil"
"log"
"os"
"path/filepath"
"sync"
)
type signature string
@ -11,12 +20,177 @@ type signatures struct {
// allowed is a map for holding all the allowed signatures.
allowed map[signature]struct{}
mu sync.Mutex
// Full path to the signing keys folder
SignKeyFolder string
// Full path to private signing key.
SignKeyPrivateKeyPath string
// Full path to public signing key.
SignKeyPublicKeyPath string
// private key for ed25519 signing.
SignPrivateKey []byte
// public key for ed25519 signing.
SignPublicKey []byte
configuration *Configuration
errorKernel *errorKernel
}
func newSignatures() *signatures {
func newSignatures(configuration *Configuration, errorKernel *errorKernel) *signatures {
s := signatures{
allowed: make(map[signature]struct{}),
allowed: make(map[signature]struct{}),
configuration: configuration,
errorKernel: errorKernel,
}
// Set the signing key paths.
s.SignKeyFolder = filepath.Join(configuration.ConfigFolder, "signing")
s.SignKeyPrivateKeyPath = filepath.Join(s.SignKeyFolder, "private.key")
s.SignKeyPublicKeyPath = filepath.Join(s.SignKeyFolder, "public.key")
err := s.loadSigningKeys()
if err != nil {
log.Printf("%v\n", err)
os.Exit(1)
}
return &s
}
// loadSigningKeys will try to load the ed25519 signing keys. If the
// files are not found new keys will be generated and written to disk.
func (s *signatures) loadSigningKeys() error {
// Check if folder structure exist, if not create it.
if _, err := os.Stat(s.SignKeyFolder); os.IsNotExist(err) {
err := os.MkdirAll(s.SignKeyFolder, 0700)
if err != nil {
er := fmt.Errorf("error: failed to create directory for signing keys : %v", err)
return er
}
}
// Check if there already are any keys in the etc folder.
foundKey := false
if _, err := os.Stat(s.SignKeyPublicKeyPath); !os.IsNotExist(err) {
foundKey = true
}
if _, err := os.Stat(s.SignKeyPrivateKeyPath); !os.IsNotExist(err) {
foundKey = true
}
// If no keys where found generete a new pair, load them into the
// processes struct fields, and write them to disk.
if !foundKey {
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
er := fmt.Errorf("error: failed to generate ed25519 keys for signing: %v", err)
return er
}
pubB64string := base64.RawStdEncoding.EncodeToString(pub)
privB64string := base64.RawStdEncoding.EncodeToString(priv)
// Write public key to file.
err = s.writeSigningKey(s.SignKeyPublicKeyPath, pubB64string)
if err != nil {
return err
}
// Write private key to file.
err = s.writeSigningKey(s.SignKeyPrivateKeyPath, privB64string)
if err != nil {
return err
}
// Also store the keys in the processes structure so we can
// reference them from there when we need them.
s.SignPublicKey = pub
s.SignPrivateKey = priv
er := fmt.Errorf("info: no signing keys found, generating new keys")
log.Printf("%v\n", er)
// We got the new generated keys now, so we can return.
return nil
}
// Key files found, load them into the processes struct fields.
pubKey, _, err := s.readKeyFile(s.SignKeyPublicKeyPath)
if err != nil {
return err
}
s.SignPublicKey = pubKey
privKey, _, err := s.readKeyFile(s.SignKeyPrivateKeyPath)
if err != nil {
return err
}
s.SignPublicKey = pubKey
s.SignPrivateKey = privKey
return nil
}
// writeSigningKey will write the base64 encoded signing key to file.
func (s *signatures) writeSigningKey(realPath string, keyB64 string) error {
fh, err := os.OpenFile(realPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
er := fmt.Errorf("error: failed to open key file for writing: %v", err)
return er
}
defer fh.Close()
_, err = fh.Write([]byte(keyB64))
if err != nil {
er := fmt.Errorf("error: failed to write key to file: %v", err)
return er
}
return nil
}
// readKeyFile will take the path of a key file as input, read the base64
// encoded data, decode the data. It will return the raw data as []byte,
// the base64 encoded data, and any eventual error.
func (s *signatures) readKeyFile(keyFile string) (ed2519key []byte, b64Key []byte, err error) {
fh, err := os.Open(keyFile)
if err != nil {
er := fmt.Errorf("error: failed to open key file: %v", err)
return nil, nil, er
}
defer fh.Close()
b, err := ioutil.ReadAll(fh)
if err != nil {
er := fmt.Errorf("error: failed to read key file: %v", err)
return nil, nil, er
}
key, err := base64.RawStdEncoding.DecodeString(string(b))
if err != nil {
er := fmt.Errorf("error: failed to base64 decode key data: %v", err)
return nil, nil, er
}
return key, b, nil
}
// verifySignature
func (s *signatures) verifySignature(m Message) bool {
fmt.Printf(" * DEBUG: verifySignature, method: %v ,s contains: %v\n", m.Method, s)
if s.configuration.AllowEmptySignature {
fmt.Printf(" * DEBUG: verifySignature: AllowEmptySignature set to TRUE\n")
return true
}
// Verify if the signature matches.
argsStringified := argsToString(m.MethodArgs)
ok := ed25519.Verify(s.SignPublicKey, []byte(argsStringified), m.ArgSignature)
fmt.Printf(" * DEBUG: verifySignature, result: %v, fromNode: %v, method: %v, signature: %s\n", ok, m.FromNode, m.Method, m.ArgSignature)
return ok
}

1
steward_test.go
View file

@ -55,6 +55,7 @@ func TestStewardServer(t *testing.T) {
DefaultMessageRetries: 1,
DefaultMessageTimeout: 3,
EnableSocket: true,
// AllowEmptySignature: true,
StartSubREQCliCommand: true,
StartSubREQCliCommandCont: true,