fix tls and cached nearby check (#45)

* remove tls

* cleanup and a couple tls fixes

* more cleanup

---------

Co-authored-by: Luke Lombardi <[email protected]>

Author: luke-lombardi

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/moby/sys/mountinfo v0.7.2
 	github.com/redis/go-redis/v9 v9.5.1
 	github.com/shirou/gopsutil v2.21.11+incompatible
+	github.com/tj/assert v0.0.3
 	google.golang.org/grpc v1.71.1
 	google.golang.org/protobuf v1.36.4
 	gotest.tools v2.2.0+incompatible
@@ -30,6 +31,7 @@ require (
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
+	github.com/stretchr/testify v1.9.0 // indirect
 )
 
 require (

go.sum

@@ -12,6 +12,8 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/deckarep/golang-set/v2 v2.6.0 h1:XfcQbWM1LlMB8BsJ8N9vW5ehnnPVIw0je80NsVHagjM=
@@ -85,6 +87,7 @@ github.com/moby/sys/mountinfo v0.7.2/go.mod h1:1YOa8w8Ih7uW0wALDUgT1dTTSBrZ+HiBL
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
@@ -97,8 +100,12 @@ github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
 github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
 github.com/shirou/gopsutil v2.21.11+incompatible h1:lOGOyCG67a5dv2hq5Z1BLDUqqKp3HkbjPcz5j6XMS0U=
 github.com/shirou/gopsutil v2.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/tj/assert v0.0.3 h1:Df/BlaZ20mq6kuai7f5z2TvPFiwC3xaWJSDQNiIS3Rk=
+github.com/tj/assert v0.0.3/go.mod h1:Ne6X72Q+TB1AteidzQncjw9PabbMp4PBMZ1k+vd1Pvk=
 github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
@@ -135,6 +142,8 @@ google.golang.org/protobuf v1.36.4/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojt
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=

pkg/blobfs.go

@@ -187,7 +187,8 @@ func NewFileSystem(ctx context.Context, opts BlobFsSystemOpts) (*BlobFs, error)
 
 		err := coordinatorClient.SetFsNode(bfs.ctx, rootID, dirMeta)
 		if err != nil {
-			Logger.Fatalf("Unable to create blobfs root node dir metdata: %+v", err)
+			Logger.Errorf("Unable to create blobfs root node dir metdata: %+v", err)
+			return nil, err
 		}
 	}
 

pkg/client.go

@@ -2,6 +2,7 @@ package blobcache
 
 import (
 	"context"
+	"crypto/tls"
 	"io"
 	"sync"
 	"time"
@@ -10,6 +11,7 @@ import (
 	rendezvous "github.com/beam-cloud/rendezvous"
 	"github.com/hanwen/go-fuse/v2/fuse"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/metadata"
 )
@@ -133,16 +135,17 @@ func (c *BlobCacheClient) GetNearbyHosts() ([]*BlobCacheHost, error) {
 }
 
 func (c *BlobCacheClient) addHost(host *BlobCacheHost) error {
-	transportCredentials := grpc.WithTransportCredentials(insecure.NewCredentials())
-
-	// TODO: revisit TLS here for external addresses, for now since we're using the internal (private) addresses, it should only use TLS for the coordinator
-
 	addr := host.Addr
-
 	if host.PrivateAddr != "" {
 		addr = host.PrivateAddr
 	}
 
+	transportCredentials := grpc.WithTransportCredentials(insecure.NewCredentials())
+	if isTLSEnabled(addr) {
+		h2creds := credentials.NewTLS(&tls.Config{NextProtos: []string{"h2"}})
+		transportCredentials = grpc.WithTransportCredentials(h2creds)
+	}
+
 	var dialOpts = []grpc.DialOption{
 		transportCredentials,
 		grpc.WithContextDialer(DialWithTimeout),

pkg/config.default.yaml

@@ -39,5 +39,4 @@ global:
   grpcMessageSizeBytes: 1000000000
   debugMode: false
   prettyLogs: false
-  tlsEnabled: false
   sources: []

pkg/coordinator_remote.go

@@ -18,16 +18,14 @@ type CoordinatorClientRemote struct {
 }
 
 func NewCoordinatorClientRemote(cfg BlobCacheGlobalConfig, token string) (CoordinatorClient, error) {
-	transportCredentials := grpc.WithTransportCredentials(insecure.NewCredentials())
+	addr := cfg.CoordinatorHost
 
-	isTLS := cfg.TLSEnabled
-	if isTLS {
+	transportCredentials := grpc.WithTransportCredentials(insecure.NewCredentials())
+	if isTLSEnabled(addr) {
 		h2creds := credentials.NewTLS(&tls.Config{NextProtos: []string{"h2"}})
 		transportCredentials = grpc.WithTransportCredentials(h2creds)
 	}
 
-	addr := cfg.CoordinatorHost
-
 	var dialOpts = []grpc.DialOption{
 		transportCredentials,
 		grpc.WithContextDialer(DialWithTimeout),

pkg/discovery.go

@@ -2,12 +2,14 @@ package blobcache
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"sync"
 	"time"
 
 	proto "github.com/beam-cloud/blobcache-v2/proto"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
@@ -68,28 +70,27 @@ func (d *DiscoveryClient) discoverHosts(ctx context.Context) ([]*BlobCacheHost,
 	mu := sync.Mutex{}
 
 	for _, host := range hosts {
-		if host.PrivateAddr != "" {
-			// Don't try to get the state on peers we're already aware of
-			if d.hostMap.Get(host.HostId) != nil {
-				continue
-			}
+		// Don't try to get the state on peers we're already aware of
+		if d.hostMap.Get(host.HostId) != nil {
+			continue
+		}
 
-			wg.Add(1)
-			go func(addr string) {
-				defer wg.Done()
+		wg.Add(1)
+		go func(addr string) {
+			defer wg.Done()
 
-				hostState, err := d.GetHostState(ctx, host)
-				if err != nil {
-					return
-				}
+			hostState, err := d.GetHostState(ctx, host)
+			if err != nil {
+				return
+			}
 
-				mu.Lock()
-				filteredHosts = append(filteredHosts, hostState)
-				mu.Unlock()
+			mu.Lock()
+			filteredHosts = append(filteredHosts, hostState)
+			mu.Unlock()
+
+			Logger.Debugf("Added host with private address to map: %s", hostState.PrivateAddr)
+		}(host.Addr)
 
-				Logger.Debugf("Added host with private address to map: %s", hostState.PrivateAddr)
-			}(host.Addr)
-		}
 	}
 
 	wg.Wait()
@@ -98,7 +99,18 @@ func (d *DiscoveryClient) discoverHosts(ctx context.Context) ([]*BlobCacheHost,
 
 // GetHostState attempts to connect to the gRPC service and verifies its availability
 func (d *DiscoveryClient) GetHostState(ctx context.Context, host *BlobCacheHost) (*BlobCacheHost, error) {
+	addr := host.Addr
+
+	if host.PrivateAddr != "" {
+		addr = host.PrivateAddr
+	}
+
 	transportCredentials := grpc.WithTransportCredentials(insecure.NewCredentials())
+	if isTLSEnabled(addr) {
+		h2creds := credentials.NewTLS(&tls.Config{NextProtos: []string{"h2"}})
+		transportCredentials = grpc.WithTransportCredentials(h2creds)
+	}
+
 
 	var dialOpts = []grpc.DialOption{
 		transportCredentials,

pkg/network.go

@@ -30,6 +30,14 @@ func getDefaultInterface() (string, error) {
 	return "", fmt.Errorf("default route not found")
 }
 
+func isTLSEnabled(addr string) bool {
+	_, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return false
+	}
+	return port == "443"
+}
+
 func GetPublicIpAddr() (string, error) {
 	resp, err := http.Get("https://api.ipify.org?format=text")
 	if err != nil {

pkg/network_test.go

@@ -0,0 +1,14 @@
+package blobcache
+
+import (
+	"testing"
+
+	"github.com/tj/assert"
+)
+
+func Test_IsTLSEnabled(t *testing.T) {
+	assert.True(t, isTLSEnabled("localhost:443"))
+	assert.False(t, isTLSEnabled("localhost:2049"))
+	assert.True(t, isTLSEnabled("127.0.0.1:443"))
+	assert.False(t, isTLSEnabled("127.0.0.1:2049"))
+}

pkg/storage.go

@@ -162,7 +162,18 @@ func (cas *ContentAddressableStorage) Add(ctx context.Context, hash string, cont
 }
 
 func (cas *ContentAddressableStorage) Exists(hash string) bool {
-	_, exists := cas.cache.GetTTL(hash)
+	var exists bool = false
+
+	_, exists = cas.cache.GetTTL(hash)
+	if !exists {
+		exists, err := os.Stat(filepath.Join(cas.diskCacheDir, hash))
+		if err != nil {
+			return false
+		}
+
+		return exists.IsDir()
+	}
+
 	return exists
 }