Add scheduled task of reloading proxy Acls from dynamoDB table.

cmd/inkfish/main.go

@@ -35,6 +35,7 @@ func main() {
 	insecureTestMode := flag.Bool("insecure-test-mode", false, "test mode (does not block)")
 	drainTime := flag.Int64("drain-time", 30, "shutdown drain deadline (seconds)")
 	connectPorts := flag.String("connect-ports", "443", "comma delimited list of valid CONNECT ports")
+	configDdb := flag.String("ddb-config", "", "name of dynamodb table of proxy rules")
 
 	flag.Parse()
 
@@ -51,10 +52,18 @@ func main() {
 	if err != nil {
 		log.Fatal("error loading CA config: ", err)
 	}
-	err = proxy.LoadConfigFromDirectory(*configDir)
-	if err != nil {
-		log.Fatal("config error: ", err)
-	}
+
+	go func() {
+		for {
+			log.Println("Load proxy configurations")
+			err = proxy.LoadConfig(*configDir, *configDdb)
+			if err != nil {
+				log.Fatal("config error: ", err)
+			}
+			time.Sleep(60 * time.Second)
+		}
+	}()
+
 	// Testmode
 	if *insecureTestMode {
 		log.Println("WARNING: PROXY IS IN TEST MODE, REQUESTS WILL NOT BE BLOCKED")

config.go

@@ -3,8 +3,14 @@ package inkfish
 import (
 	"crypto/sha256"
 	"crypto/subtle"
+	"encoding/base64"
 	"encoding/hex"
 	"fmt"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
+	"github.com/aws/aws-sdk-go/service/dynamodb/expression"
 	"github.com/pkg/errors"
 	"io/ioutil"
 	"log"
@@ -34,6 +40,11 @@ type UserEntry struct {
 	PasswordHash string
 }
 
+type ConfigItem struct {
+	ConfigName string
+	ConfigBody string
+}
+
 func listContainsString(haystack []string, needle string) bool {
 	// Return true iff needle is present in haystack
 	for _, s := range haystack {
@@ -277,13 +288,43 @@ func loadUsersFromFile(data []byte) ([]UserEntry, error) {
 	return result, nil
 }
 
-func (proxy *Inkfish) LoadConfigFromDirectory(configDir string) error {
+func (proxy *Inkfish) LoadConfig(configDir string, configDdb string) error {
+	acls := []Acl{}
+	passwd := []UserEntry{}
+	errstrings := []string{}
+	localAcls, localPasswd, err := LoadConfigFromDirectory(configDir)
+	if err != nil {
+		errstrings = append(errstrings, err.Error())
+	} else {
+		acls = append(acls, localAcls...)
+		passwd = append(passwd, localPasswd...)
+	}
+	if configDdb != "" {
+		ddbAcls, ddbPasswd, err := LoadConfigFromDynamoDB(configDdb)
+		if err != nil {
+			errstrings = append(errstrings, err.Error())
+		} else {
+			acls = append(acls, ddbAcls...)
+			passwd = append(passwd, ddbPasswd...)
+		}
+	}
+	if len(errstrings) > 0 {
+		msg := fmt.Sprintf("Load proxy configuration reported error(s):\n - %s", strings.Join(errstrings, "\n - "))
+		return errors.Errorf(msg)
+	}
+	proxy.ReplaceConfig(acls, passwd)
+	return nil
+}
+
+func LoadConfigFromDirectory(configDir string) ([]Acl, []UserEntry, error) {
 	// Load ACLs and passwd entries from a directory
 	files, err := ioutil.ReadDir(configDir)
 	if err != nil {
 		msg := fmt.Sprintf("failed to list config dir: %v", configDir)
-		return errors.Wrap(err, msg)
+		return nil, nil, errors.Wrap(err, msg)
 	}
+	acls := []Acl{}
+	passwd := []UserEntry{}
 	for _, fi := range files {
 		ext := filepath.Ext(fi.Name())
 		if ext != ".conf" && ext != ".passwd" {
@@ -307,23 +348,118 @@ func (proxy *Inkfish) LoadConfigFromDirectory(configDir string) error {
 		fullpath := filepath.Join(configDir, fi.Name())
 		data, err := ioutil.ReadFile(fullpath)
 		if err != nil {
-			return errors.Wrapf(err, "failed read config file: %v", fullpath)
+			return nil, nil, errors.Wrapf(err, "failed read config file: %v", fullpath)
 		}
 		if ext == ".conf" {
 			acl, err := loadAclFromFile(data)
 			if err != nil {
-				return errors.Wrapf(err, "error in acl file: %v", fullpath)
+				return nil, nil, errors.Wrapf(err, "error in acl file: %v", fullpath)
 			}
-			proxy.Acls = append(proxy.Acls, *acl)
-			log.Println("loaded config file", fullpath)
+			acls = append(acls, *acl)
+			log.Println("loaded config file:", fullpath)
 		} else if ext == ".passwd" {
 			userRecords, err := loadUsersFromFile(data)
 			if err != nil {
-				return errors.Wrapf(err, "error in passwd file: %v", fullpath)
+				return nil, nil, errors.Wrapf(err, "error in passwd file: %v", fullpath)
 			}
-			proxy.Passwd = append(proxy.Passwd, userRecords...)
+			passwd = append(passwd, userRecords...)
 			log.Println("loaded passwd file:", fullpath)
 		}
 	}
-	return nil
+	return acls, passwd, nil
+}
+
+func (proxy *Inkfish) ReplaceConfig(acls []Acl, passwd []UserEntry) {
+	proxy.configMutex.Lock()
+	defer proxy.configMutex.Unlock()
+
+	proxy.Acls = acls
+	proxy.Passwd = passwd
+}
+
+func LoadConfigFromDynamoDB(configDdb string) ([]Acl, []UserEntry, error) {
+	sess, err := session.NewSession()
+	svc := dynamodb.New(sess)
+	ddbTables, err := ListDDBProxyTables(sess, configDdb)
+	if err != nil {
+		msg := fmt.Sprintf("failed to list dynamodb proxy rules tables")
+		return nil, nil, errors.Wrap(err, msg)
+	}
+	proj := expression.NamesList(
+		expression.Name("ConfigName"),
+		expression.Name("ConfigBody"),
+	)
+	expr, err := expression.NewBuilder().WithProjection(proj).Build()
+	if err != nil {
+		panic(err)
+	}
+	acls := []Acl{}
+	passwd := []UserEntry{}
+	for _, t := range ddbTables {
+		params := &dynamodb.ScanInput{
+			ConsistentRead:            aws.Bool(true),
+			ExpressionAttributeNames:  expr.Names(),
+			ExpressionAttributeValues: expr.Values(),
+			FilterExpression:          expr.Filter(),
+			ProjectionExpression:      expr.Projection(),
+			TableName:                 aws.String(t),
+		}
+		result, err := svc.Scan(params)
+		if err != nil {
+			msg := fmt.Sprintf("failed to scan dynamodb table: %v", t)
+			return nil, nil, errors.Wrap(err, msg)
+		}
+		for _, i := range result.Items {
+			item := ConfigItem{}
+			err = dynamodbattribute.UnmarshalMap(i, &item)
+			if err != nil {
+				msg := "Got error unmarshalling"
+				return nil, nil, errors.Wrap(err, msg)
+			}
+			data, err := base64.StdEncoding.DecodeString(item.ConfigBody)
+			if err != nil {
+				msg := "Got error decoding config"
+				return nil, nil, errors.Wrap(err, msg)
+			}
+			if strings.HasSuffix(item.ConfigName, ".conf") {
+				acl, err := loadAclFromFile(data)
+				if err != nil {
+					return nil, nil, errors.Wrapf(err, "error in acl [%v] from dynamodb table [%v]", item.ConfigName, t)
+				}
+				acls = append(acls, *acl)
+				log.Printf("loaded acl [%v] from dynamodb table [%v]", item.ConfigName, t)
+			} else if strings.HasSuffix(item.ConfigName, ".passwd") {
+				userRecords, err := loadUsersFromFile(data)
+				if err != nil {
+					return nil, nil, errors.Wrapf(err, "error in passwd [%v] from dynamodb table [%v]", item.ConfigName, t)
+				}
+				passwd = append(passwd, userRecords...)
+				log.Printf("loaded passwd [%v] from dynamodb table [%v]", item.ConfigName, t)
+			}
+		}
+	}
+	return acls, passwd, nil
+}
+
+func ListDDBProxyTables(sess *session.Session, tableNameRegex string) ([]string, error) {
+	svc := dynamodb.New(sess)
+	input := &dynamodb.ListTablesInput{}
+	tableNames := []string{}
+	for {
+		result, err := svc.ListTables(input)
+		if err != nil {
+			return nil, err
+		}
+		validID := regexp.MustCompile(tableNameRegex)
+		for _, n := range result.TableNames {
+			if validID.MatchString(*n) {
+				tableNames = append(tableNames, *n)
+			}
+		}
+		input.ExclusiveStartTableName = result.LastEvaluatedTableName
+		if result.LastEvaluatedTableName == nil {
+			break
+		}
+	}
+	return tableNames, nil
 }

config_test.go

@@ -230,9 +230,9 @@ func TestAclConfigWithMissingPortInBypass(t *testing.T) {
 	assert.Equal(t, "missing port in bypass at line: 5", err.Error())
 }
 
-func TestLoadConfig(t *testing.T) {
+func TestLoadConfigFromLocalDirOnly(t *testing.T) {
 	proxy := NewInkfish(NewCertSigner(&StubCA))
-	err := proxy.LoadConfigFromDirectory("testdata/unit_test_config")
+	err := proxy.LoadConfig("testdata/unit_test_config", "")
 	assert.NotNil(t, proxy.Acls)
 	assert.Nil(t, err)
 
@@ -241,9 +241,26 @@ func TestLoadConfig(t *testing.T) {
 
 func TestLoadConfigWithSymlink(t *testing.T) {
 	proxy := NewInkfish(NewCertSigner(&StubCA))
-	err := proxy.LoadConfigFromDirectory("testdata/symlink_test_config")
+	err := proxy.LoadConfig("testdata/symlink_test_config", "")
 	assert.NotNil(t, proxy.Acls)
 	assert.Nil(t, err)
 
 	assert.Equal(t, 2, len(proxy.Acls))
 }
+
+func TestLoadConfigFromLocalDirFails(t *testing.T) {
+	proxy := NewInkfish(NewCertSigner(&StubCA))
+	err := proxy.LoadConfig("dir-does-not-exist/nofile", "")
+	assert.Nil(t, proxy.Acls)
+	assert.NotNil(t, err)
+
+	assert.Equal(t, 0, len(proxy.Acls))
+}
+
+func TestLoadEmptyConfig(t *testing.T) {
+	proxy := NewInkfish(NewCertSigner(&StubCA))
+	err := proxy.LoadConfig(".", "")
+	assert.Nil(t, err)
+
+	assert.Equal(t, 0, len(proxy.Acls))
+}

go.sum

@@ -50,7 +50,6 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -77,7 +76,6 @@ github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
@@ -99,6 +97,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82 h1:ywK/j/KkyTHcdyYSZNXGjMwgmDSfjglYZ3vStQ/gSCU=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

inkfish.go

@@ -22,6 +22,7 @@ import (
 	"net/http/httputil"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -40,6 +41,7 @@ type Metrics struct {
 type Inkfish struct {
 	Acls   []Acl
 	Passwd []UserEntry
+	configMutex sync.RWMutex
 
 	// Maintains an ip -> tag map, for access control based on instance metadata
 	MetadataProvider MetadataProvider
@@ -181,6 +183,9 @@ func (proxy *Inkfish) sendAccessDenied(w http.ResponseWriter, detail string) {
 }
 
 func (proxy *Inkfish) filterConnect(w http.ResponseWriter, req *http.Request) (action ConnectAction) {
+	proxy.configMutex.RLock()
+	defer proxy.configMutex.RUnlock()
+
 	host := req.Host
 
 	// Handle a CONNECT request
@@ -269,6 +274,9 @@ func (proxy *Inkfish) filterConnect(w http.ResponseWriter, req *http.Request) (a
 }
 
 func (proxy *Inkfish) filterRequest(connectReq *http.Request, scheme string, w http.ResponseWriter, req *http.Request) bool {
+	proxy.configMutex.RLock()
+	defer proxy.configMutex.RUnlock()
+
 	var user string
 	var err error
 	if scheme == "https" {