Unix socket to mutual TLS port connection (#54)

Author: michelvocks

cmd/gaia/main.go

@@ -155,7 +155,7 @@ func main() {
 	pS := &plugin.Plugin{}
 
 	// Initialize scheduler
-	scheduler := scheduler.NewScheduler(store, pS)
+	scheduler := scheduler.NewScheduler(store, pS, cert)
 	err = scheduler.Init()
 	if err != nil {
 		gaia.Cfg.Logger.Error("cannot initialize scheduler:", "error", err.Error())

plugin/plugin.go

@@ -9,16 +9,22 @@ import (
 
 	"github.com/gaia-pipeline/gaia"
 	"github.com/gaia-pipeline/gaia/scheduler"
+	"github.com/gaia-pipeline/gaia/security"
 	"github.com/gaia-pipeline/protobuf"
 	plugin "github.com/hashicorp/go-plugin"
 )
 
 const (
 	pluginMapKey = "Plugin"
+
+	// env variable key names for TLS cert path
+	serverCertEnv = "GAIA_PLUGIN_CERT"
+	serverKeyEnv  = "GAIA_PLUGIN_KEY"
+	rootCACertEnv = "GAIA_PLUGIN_CA_CERT"
 )
 
 var handshake = plugin.HandshakeConfig{
-	ProtocolVersion: 1,
+	ProtocolVersion: 2,
 	MagicCookieKey:  "GAIA_PLUGIN",
 	// This cookie should never be changed again
 	MagicCookieValue: "FdXjW27mN6XuG2zDBP4LixXUwDAGCEkidxwqBGYpUhxiWHzctATYZvpz4ZJdALmh",
@@ -42,12 +48,21 @@ type Plugin struct {
 
 	// Writer used to write logs from execution to file
 	writer *bufio.Writer
+
+	// CA instance used to handle certificates
+	ca security.CAAPI
+
+	// Created certificates path for pipeline run
+	certPath       string
+	keyPath        string
+	serverCertPath string
+	serverKeyPath  string
 }
 
 // NewPlugin creates a new instance of Plugin.
 // One Plugin instance represents one connection to a plugin.
-func (p *Plugin) NewPlugin() scheduler.Plugin {
-	return &Plugin{}
+func (p *Plugin) NewPlugin(ca security.CAAPI) scheduler.Plugin {
+	return &Plugin{ca: ca}
 }
 
 // Connect prepares the log path, starts the plugin, initiates the
@@ -75,13 +90,40 @@ func (p *Plugin) Connect(command *exec.Cmd, logPath *string) error {
 	// Create new writer
 	p.writer = bufio.NewWriter(p.logFile)
 
+	// Create and sign a new pair of certificates for the server
+	var err error
+	p.serverCertPath, p.serverKeyPath, err = p.ca.CreateSignedCert()
+	if err != nil {
+		return err
+	}
+
+	// Expose path of server certificates as well as public CA cert.
+	// This allows the plugin to grab the certificates.
+	caCert, _ := p.ca.GetCACertPath()
+	command.Env = append(command.Env, serverCertEnv+"="+p.serverCertPath)
+	command.Env = append(command.Env, serverKeyEnv+"="+p.serverKeyPath)
+	command.Env = append(command.Env, rootCACertEnv+"="+caCert)
+
+	// Create and sign a new pair of certificates for the client
+	p.certPath, p.keyPath, err = p.ca.CreateSignedCert()
+	if err != nil {
+		return err
+	}
+
+	// Generate TLS config
+	tlsConfig, err := p.ca.GenerateTLSConfig(p.certPath, p.keyPath)
+	if err != nil {
+		return err
+	}
+
 	// Get new client
 	p.client = plugin.NewClient(&plugin.ClientConfig{
 		HandshakeConfig:  handshake,
 		Plugins:          pluginMap,
 		Cmd:              command,
 		AllowedProtocols: []plugin.Protocol{plugin.ProtocolGRPC},
 		Stderr:           p.writer,
+		TLSConfig:        tlsConfig,
 	})
 
 	// Connect via gRPC
@@ -176,5 +218,9 @@ func (p *Plugin) Close() {
 
 		// Close log file
 		p.logFile.Close()
+
+		// Cleanup certificates
+		p.ca.CleanupCerts(p.certPath, p.keyPath)
+		p.ca.CleanupCerts(p.serverCertPath, p.serverKeyPath)
 	}()
 }

scheduler/scheduler.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/gaia-pipeline/gaia"
+	"github.com/gaia-pipeline/gaia/security"
 	"github.com/gaia-pipeline/gaia/store"
 	uuid "github.com/satori/go.uuid"
 )
@@ -33,7 +34,7 @@ var (
 // during scheduling and execution.
 type Plugin interface {
 	// NewPlugin creates a new instance of plugin
-	NewPlugin() Plugin
+	NewPlugin(ca security.CAAPI) Plugin
 
 	// Connect initializes the connection with the execution command
 	// and the log path wbere the logs should be stored.
@@ -60,15 +61,19 @@ type Scheduler struct {
 
 	// pluginSystem is the used plugin system.
 	pluginSystem Plugin
+
+	// ca is the instance of the CA used to handle certs.
+	ca security.CAAPI
 }
 
 // NewScheduler creates a new instance of Scheduler.
-func NewScheduler(store *store.Store, pS Plugin) *Scheduler {
+func NewScheduler(store *store.Store, pS Plugin, ca security.CAAPI) *Scheduler {
 	// Create new scheduler
 	s := &Scheduler{
 		scheduledRuns: make(chan gaia.PipelineRun, schedulerBufferLimit),
 		storeService:  store,
 		pluginSystem:  pS,
+		ca:            ca,
 	}
 
 	return s
@@ -165,7 +170,7 @@ func (s *Scheduler) prepareAndExec(r gaia.PipelineRun) {
 	}
 
 	// Create new plugin instance
-	pS := s.pluginSystem.NewPlugin()
+	pS := s.pluginSystem.NewPlugin(s.ca)
 
 	// Connect to plugin(pipeline)
 	path = filepath.Join(path, gaia.LogsFileName)
@@ -368,7 +373,7 @@ func (s *Scheduler) getPipelineJobs(p *gaia.Pipeline) ([]gaia.Job, error) {
 	}
 
 	// Create new Plugin instance
-	pS := s.pluginSystem.NewPlugin()
+	pS := s.pluginSystem.NewPlugin(s.ca)
 
 	// Connect to plugin(pipeline)
 	if err := pS.Connect(c, nil); err != nil {

scheduler/scheduler_test.go

@@ -1,26 +1,35 @@
 package scheduler
 
 import (
+	"crypto/tls"
 	"hash/fnv"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"testing"
 
 	"github.com/gaia-pipeline/gaia"
+	"github.com/gaia-pipeline/gaia/security"
 	"github.com/gaia-pipeline/gaia/store"
 	hclog "github.com/hashicorp/go-hclog"
 	uuid "github.com/satori/go.uuid"
 )
 
 type PluginFake struct{}
 
-func (p *PluginFake) NewPlugin() Plugin                            { return &PluginFake{} }
+func (p *PluginFake) NewPlugin(ca security.CAAPI) Plugin           { return &PluginFake{} }
 func (p *PluginFake) Connect(cmd *exec.Cmd, logPath *string) error { return nil }
 func (p *PluginFake) Execute(j *gaia.Job) error                    { return nil }
 func (p *PluginFake) GetJobs() ([]gaia.Job, error)                 { return prepareJobs(), nil }
 func (p *PluginFake) Close()                                       {}
 
+type CAFake struct{}
+
+func (c *CAFake) CreateSignedCert() (string, string, error)                       { return "", "", nil }
+func (c *CAFake) GenerateTLSConfig(certPath, keyPath string) (*tls.Config, error) { return nil, nil }
+func (c *CAFake) CleanupCerts(crt, key string) error                              { return nil }
+func (c *CAFake) GetCACertPath() (string, string)                                 { return "", "" }
+
 func TestInit(t *testing.T) {
 	gaia.Cfg = &gaia.Config{}
 	storeInstance := store.NewStore()
@@ -36,7 +45,9 @@ func TestInit(t *testing.T) {
 	if err := storeInstance.Init(); err != nil {
 		t.Fatal(err)
 	}
-	s := NewScheduler(storeInstance, &PluginFake{})
+	var ca security.CAAPI
+	ca = &CAFake{}
+	s := NewScheduler(storeInstance, &PluginFake{}, ca)
 	err := s.Init()
 	if err != nil {
 		t.Fatal(err)
@@ -64,7 +75,9 @@ func TestPrepareAndExec(t *testing.T) {
 	}
 	p, r := prepareTestData()
 	storeInstance.PipelinePut(&p)
-	s := NewScheduler(storeInstance, &PluginFake{})
+	var ca security.CAAPI
+	ca = &CAFake{}
+	s := NewScheduler(storeInstance, &PluginFake{}, ca)
 	s.prepareAndExec(r)
 
 	// Iterate jobs
@@ -98,7 +111,9 @@ func TestSchedulePipeline(t *testing.T) {
 	}
 	p, _ := prepareTestData()
 	storeInstance.PipelinePut(&p)
-	s := NewScheduler(storeInstance, &PluginFake{})
+	var ca security.CAAPI
+	ca = &CAFake{}
+	s := NewScheduler(storeInstance, &PluginFake{}, ca)
 	err := s.Init()
 	if err != nil {
 		t.Fatal(err)
@@ -131,7 +146,9 @@ func TestSchedule(t *testing.T) {
 	}
 	p, _ := prepareTestData()
 	storeInstance.PipelinePut(&p)
-	s := NewScheduler(storeInstance, &PluginFake{})
+	var ca security.CAAPI
+	ca = &CAFake{}
+	s := NewScheduler(storeInstance, &PluginFake{}, ca)
 	_, err := s.SchedulePipeline(&p)
 	if err != nil {
 		t.Fatal(err)
@@ -166,7 +183,9 @@ func TestSetPipelineJobs(t *testing.T) {
 	}
 	p, _ := prepareTestData()
 	p.Jobs = nil
-	s := NewScheduler(storeInstance, &PluginFake{})
+	var ca security.CAAPI
+	ca = &CAFake{}
+	s := NewScheduler(storeInstance, &PluginFake{}, ca)
 	err := s.SetPipelineJobs(&p)
 	if err != nil {
 		t.Fatal(err)