Implement server TLS support

Changes.md

@@ -2,8 +2,11 @@
 
 Changelog: Faktory || [Faktory Enterprise](https://github.com/contribsys/faktory/blob/master/Ent-Changes.md)
 
-## HEAD
+## 1.9.0
 
+- Implement native TLS support [#469]
+  Put `public.cert.pem` and `private.key.pem` in your config directory
+  and Faktory will automatically enable TLS on port 7419 and 7420.
 - Unwrap and display ActiveJob class names [#460, ibrahima]
 - Add Go client API so batches can push jobs in bulk [#437, tylerian]
 

client/faktory.go

@@ -2,5 +2,5 @@ package client
 
 var (
 	Name    = "Faktory"
-	Version = "1.8.1"
+	Version = "1.9.0"
 )

server/server.go

@@ -5,11 +5,14 @@ import (
 	"context"
 	"crypto/sha256"
 	"crypto/subtle"
+	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"math/rand"
 	"net"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -34,6 +37,10 @@ type Server struct {
 	Stats      *RuntimeStats
 	Subsystems []Subsystem
 
+	TLSPublicCert string
+	TLSPrivateKey string
+
+	tlsConfig  *tls.Config
 	listener   net.Listener
 	store      storage.Store
 	manager    manager.Manager
@@ -44,6 +51,41 @@ type Server struct {
 	closed     bool
 }
 
+func (s *Server) useTLS() error {
+	privateKey := filepath.Join(s.Options.ConfigDirectory, "private.key.pem")
+	publicCert := filepath.Join(s.Options.ConfigDirectory, "public.cert.pem")
+
+	if _, err := os.Stat(privateKey); errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if _, err := os.Stat(publicCert); errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	x, err := os.Open(privateKey)
+	if err != nil {
+		return fmt.Errorf("Unable to open private key: %w", err)
+	}
+	defer x.Close()
+	y, err := os.Open(publicCert)
+	if err != nil {
+		return fmt.Errorf("Unable to open public cert: %w", err)
+	}
+	defer y.Close()
+	cert, err := tls.LoadX509KeyPair(publicCert, privateKey)
+	if err != nil {
+		return fmt.Errorf("Unable to initialize TLS certificate: %w", err)
+	}
+	util.Infof("TLS activated with %s", publicCert)
+
+	s.TLSPublicCert = publicCert
+	s.TLSPrivateKey = privateKey
+	s.tlsConfig = &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		MinVersion:   tls.VersionTLS12,
+	}
+	return nil
+}
+
 func NewServer(opts *ServerOptions) (*Server, error) {
 	if opts.Binding == "" {
 		opts.Binding = "localhost:7419"
@@ -95,7 +137,17 @@ func (s *Server) Boot() error {
 		return fmt.Errorf("cannot open redis database: %w", err)
 	}
 
-	listener, err := net.Listen("tcp", s.Options.Binding)
+	err = s.useTLS()
+	if err != nil {
+		return err
+	}
+
+	var listener net.Listener
+	if s.tlsConfig != nil {
+		listener, err = tls.Listen("tcp", s.Options.Binding, s.tlsConfig)
+	} else {
+		listener, err = net.Listen("tcp", s.Options.Binding)
+	}
 	if err != nil {
 		store.Close()
 		return fmt.Errorf("cannot listen on %s: %w", s.Options.Binding, err)
@@ -221,19 +273,32 @@ func startConnection(conn net.Conn, s *Server) *Connection {
 
 	line, err := buf.ReadString('\n')
 	if err != nil {
+		defer conn.Close()
 		// TCP probes on the socket will close connection
 		// immediately and lead to EOF. Don't flood logs with them.
-		if err != io.EOF {
-			util.Error("Bad connection", err)
+		if errors.Is(err, io.EOF) {
+			return nil
 		}
-		conn.Close()
+
+		opErr, ok := err.(net.Error)
+		if ok && opErr.Timeout() {
+			// util.Debugf("Error establishing connection with client %v", err)
+			return nil
+		}
+		_, ok = err.(tls.RecordHeaderError)
+		if ok {
+			// util.Debugf("Client not using TLS %v", err)
+			return nil
+
+		}
+		util.Infof("Bad connection %T: %v", err, err)
 		return nil
 	}
 
 	valid := strings.HasPrefix(line, "HELLO {")
 	if !valid {
-		util.Infof("Invalid preamble: %s", line)
-		util.Info("Need a valid HELLO")
+		util.Debugf("Invalid preamble: %s", line)
+		util.Debug("Need a valid HELLO, is client using TLS?")
 		conn.Close()
 		return nil
 	}

server/server_test.go

@@ -31,7 +31,7 @@ func runServer(binding string, runner func()) {
 		Binding:          binding,
 		StorageDirectory: dir,
 		RedisSock:        sock,
-		ConfigDirectory:  os.ExpandEnv("$HOME/.faktory"),
+		ConfigDirectory:  os.ExpandEnv("test/.faktory"),
 	}
 	s, err := NewServer(opts)
 	if err != nil {

test/cfg/private.key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDNLCm+Mz3aqpxg
+Y1lwiZjKBpWfT6W8ZJCRS75P5RAIAH3pOqO7g3ZmhQprA9KbtuKqjuDrdLnnkEC3
+jmE/7bLMFAVVbF+N1dj+1CtMsdH5T8aweC62XuaeHIc2MMvXiSX3QcAMUjdaXWW4
+wCNQhRJwbvE1XWcgzuoPJneeJdvCQLbW+c2B0vTI8k+lexz9iuVReWBumaFXubeE
+vy4BeITGcWPnTsRZ7KJlPUwkfHxWxStatwI+gjCXv76tOIM+6cWh0QHbUqnMpiOv
+0G4SkHunnI9DGxRv7jZBrQYCWqxO/G/bFceIoWCmTaGna+sMqprdH2SBHYDGIotn
+s+sekDrXAgMBAAECggEABQjTuJe9S51AH7zUnnmmJKu8bLC7PIRWaFYKiIW/VGyJ
+VRMXr5IZC+71ZdGDINBeB8PG1R4dVFyh7zz9bRxKlwE/0/0uhTNBvdvVmxjWupGo
+CHboXGVABDyrgPctBy3Z9IIOQL8lURozdw7ULOAqQJlclMm2FKZHrAay0PrECdeA
+oKoxoRldlqrjSxyaRAt13vrVbaVB3cvtnGlIeSlRfSx8m04u0U1uWGlBlb8xhgOR
+65j6+bKNAPaE2qxOuxyrIu5DbKnyjfky3Mf1FApmrpjmu3XBwrZuhnPFNzLeMx0w
+YsYUX19Sbqp1YpisK9ok5OQ0AiA+DP8SD70y7OicQQKBgQDqntooG9LRqIcMs6cw
+JijOXTQilwnYjRAfpRnkkp8Fu33gsmXa7Oa1j9rVcBnpLxBCTFR2iFmOxpm7f9JX
+XnNC+zVgHYgxftX1L9yX+nkuSi/bWSWRhEJDcCOSG1vTV2x1aoYb8RqUyKIpF+h1
+SES1bE08cDW19T07Wmdxz6W6kQKBgQDf3luahCbvGQqmR53LEyJHNuQc2FgPdAKv
+kiL4bf3F68vQPuMuvsPim1UyyxmG4rhA1BrnUZq2nERZ+Cv7MXLcy51Q1Lq9K/93
+WUbOxYsT+3NeQRNEJVpfhZZ0ALUZmfMNp5aemrH//1yBEuk4TLikPbn3AQ114wjo
+vwfqgfPC5wKBgAlxp3ph1FSYIgeC28H2Z0IXQlf6GG1dED8V2Dr5I+mJKIH47Fyp
+mIfKaJaa0pAuUss4Y1X6GxDCMcH4XTEjHiSeFAHwbmD/qAEckhaUAHi6h76ekgKP
+flNUmjnxW/rf0//N7+QECnver5hT3AmMhSeAWoOKSL8wReyzsOJF53fhAoGBANLj
+W/XGMgsg0uhrJJlN8AeYDPGjV+lOxszv5GOU8fAFvZzx8P9zE4KgA3Vy4Bwx7ZKc
+fK+WLyGBOd5rK7tZDLQ0V4DytOtJzEF453wXmXl8cWTD9stGSMkdResHU5LHdLBT
+RE8quS3IODMbRnoTxAhsYYfvBOgdtKHUezeNrbzXAoGAZai4rNxmuPKsZS2uuT45
+6edkS0OW9n8y6NHso8chxBsGdXuz45bbWdndW3Xs+4BTaahUOJ59ILvN+s4ZEG1z
+6GALGp1Jg4dAvMCks6tvH2mPTK69Zjp8NZSuiNpJbhL1mHtmCchhZPEKW8f+BKLx
+3W139Ps4mqU+H6oo45iOS1Y=
+-----END PRIVATE KEY-----

test/cfg/public.cert.pem

@@ -0,0 +1,25 @@
+-----BEGIN CERTIFICATE-----
+MIIEQzCCAyugAwIBAgIUOf/FXcVBElRQYGEmbBqVoepecV4wDQYJKoZIhvcNAQEL
+BQAwgbAxCzAJBgNVBAYTAlVTMQ8wDQYDVQQIDAZPcmVnb24xETAPBgNVBAcMCFBv
+cnRsYW5kMSAwHgYDVQQKDBdDb250cmlidXRlZCBTeXN0ZW1zIExMQzEQMA4GA1UE
+CwwHRmFrdG9yeTEkMCIGA1UEAwwbZmFrdG9yeS10ZXN0LmNvbnRyaWJzeXMuY29t
+MSMwIQYJKoZIhvcNAQkBFhRhZG1pbkBjb250cmlic3lzLmNvbTAeFw0yNDAyMTky
+MDE2MzNaFw0yNTAyMTgyMDE2MzNaMIGwMQswCQYDVQQGEwJVUzEPMA0GA1UECAwG
+T3JlZ29uMREwDwYDVQQHDAhQb3J0bGFuZDEgMB4GA1UECgwXQ29udHJpYnV0ZWQg
+U3lzdGVtcyBMTEMxEDAOBgNVBAsMB0Zha3RvcnkxJDAiBgNVBAMMG2Zha3Rvcnkt
+dGVzdC5jb250cmlic3lzLmNvbTEjMCEGCSqGSIb3DQEJARYUYWRtaW5AY29udHJp
+YnN5cy5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDNLCm+Mz3a
+qpxgY1lwiZjKBpWfT6W8ZJCRS75P5RAIAH3pOqO7g3ZmhQprA9KbtuKqjuDrdLnn
+kEC3jmE/7bLMFAVVbF+N1dj+1CtMsdH5T8aweC62XuaeHIc2MMvXiSX3QcAMUjda
+XWW4wCNQhRJwbvE1XWcgzuoPJneeJdvCQLbW+c2B0vTI8k+lexz9iuVReWBumaFX
+ubeEvy4BeITGcWPnTsRZ7KJlPUwkfHxWxStatwI+gjCXv76tOIM+6cWh0QHbUqnM
+piOv0G4SkHunnI9DGxRv7jZBrQYCWqxO/G/bFceIoWCmTaGna+sMqprdH2SBHYDG
+Iotns+sekDrXAgMBAAGjUzBRMB0GA1UdDgQWBBTydABM8WHFVsPGVpf5JXKqXuR3
+qjAfBgNVHSMEGDAWgBTydABM8WHFVsPGVpf5JXKqXuR3qjAPBgNVHRMBAf8EBTAD
+AQH/MA0GCSqGSIb3DQEBCwUAA4IBAQBoofu4cyxCdZgq15+A4Zj1WyHX3DlSK+N0
+FUphoukwLLMyeo/WiTPkL76CLcP/0nu2GwbBmlNsGkBSJPhcxwrZXm3tGZ1fchrD
+b6T60hYWskXju+D+LjoG8MXzImcF1FboCyVTePK+2+cy5Lmm1IwFaa+TeqyLeQnN
+uM8YLDn+bx7/G+uHkZoCQCTb3iXl5gz197aVOKLfMC/8FIAO9lQspUHEOEzKPB5e
+dkVQo4uwi9+8doXVVHfJmK1je/6/LNWaNKOeCkXRx5DICMdpsV+/4GPpR+YrTbBc
+W4Q23OOsGsBNHVVcOC8fePHwPPfdEZqttwqo9yZUaizqpQCKRQ7M
+-----END CERTIFICATE-----

test/go_system_test.go

@@ -2,6 +2,7 @@ package tester
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"log"
 	"math/rand"
@@ -29,7 +30,7 @@ func TestSystem(t *testing.T) {
 	dir := "/tmp/system.db"
 	defer os.RemoveAll(dir)
 
-	opts.ConfigDirectory = "."
+	opts.ConfigDirectory = "./cfg"
 	opts.StorageDirectory = dir
 	s, stopper, err := cli.BuildServer(&opts)
 	if stopper != nil {
@@ -81,7 +82,16 @@ func TestSystem(t *testing.T) {
 
 func pushAndPop(t *testing.T, count int) {
 	time.Sleep(300 * time.Millisecond)
-	cl, err := client.Dial(client.DefaultServer(), "123456")
+	s := &client.Server{
+		Network:  "tcp+tls",
+		Address:  "localhost:7419",
+		Username: "",
+		Password: "",
+		Timeout:  1 * time.Second,
+		TLS: &tls.Config{InsecureSkipVerify: true, // nolint:gosec
+			MinVersion: tls.VersionTLS12},
+	}
+	cl, err := client.Dial(s, "123456")
 	if err != nil {
 		handleError(err)
 		return