allow list for proxy

.gitignore

@@ -2,4 +2,5 @@
 .dgate
 cov.out
 go.work.sum
-.env
+.env
+coverage.txt

README.md

@@ -21,7 +21,12 @@ http://dgate.io/docs/getting-started
 
 ```bash
 # requires go 1.22+
+
+# install dgate-server
 go install github.com/dgate-io/dgate-api/cmd/dgate-server@latest
+
+# install dgate-cli
+go install github.com/dgate-io/dgate-api/cmd/dgate-cli@latest
 ```
 
 ## Application Architecture
@@ -33,15 +38,3 @@ DGate Server is proxy and admin server bundled into one. the admin server is res
 ### DGate CLI (dgate-cli)
 
 DGate CLI is a command-line interface that can be used to interact with the DGate Server. It can be used to deploy modules, manage the state of the cluster, and more.
-
-#### Proxy Modules
-
-- Fetch Upstream Module (`fetchUpstream`) - executed before the request is sent to the upstream server. This module is used to decided which upstream server to send the current request to. (Essentially a custom load balancer module)
-
-- Request Modifier Module (`requestModifier`) - executed before the request is sent to the upstream server. This module is used to modify the request before it is sent to the upstream server.
-
-- Response Modifier Module (`responseModifier`) - executed after the response is received from the upstream server. This module is used to modify the response before it is sent to the client.
-
-- Error Handler Module (`errorHandler`) - executed when an error occurs when sending a request to the upstream server. This module is used to modify the response before it is sent to the client.
-
-- Request Handler Module (`requestHandler`) - executed when a request is received from the client. This module is used to handle arbitrary requests, instead of using an upstream service.

config.dgate.yaml

@@ -1,6 +1,6 @@
 version: v1
 debug: true
-log_level: ${LOG_LEVEL:-info}
+log_level: ${LOG_LEVEL:-debug}
 disable_default_namespace: true
 tags: [debug, local, test]
 storage:

internal/config/config.go

@@ -46,6 +46,7 @@ type (
 		DisableXForwardedHeaders bool                     `koanf:"disable_x_forwarded_headers"`
 		StrictMode               bool                     `koanf:"strict_mode"`
 		XForwardedForDepth       int                      `koanf:"x_forwarded_for_depth"`
+		AllowList                []string                 `koanf:"allow_list"`
 
 		// WARN: debug use only
 		InitResources *DGateResources `koanf:"init_resources"`

internal/config/loader.go

@@ -12,6 +12,7 @@
 	"strings"
 
 	"github.com/dgate-io/dgate-api/pkg/util"
+	"github.com/dgate-io/dgate-api/pkg/util/iplist"
 	"github.com/hashicorp/raft"
 	kjson "github.com/knadh/koanf/parsers/json"
 	ktoml "github.com/knadh/koanf/parsers/toml"
@@ -158,6 +159,15 @@
 	kDefault(k, "proxy.enable_http2", false)
 	kDefault(k, "proxy.console_log_level", k.Get("log_level"))
 
+	if k.Exists("proxy.allow_list") {
+		var ips []string = k.Get("proxy.allow_list").([]string)
+		ipList := iplist.NewIPList()
+		err = ipList.AddAll(ips)
+		if err != nil {
+			return nil, errors.New("proxy.allow_list error: " + err.Error())
+		}
+	}
+
 	if k.Get("proxy.enable_h2c") == true &&
 		k.Get("proxy.enable_http2") == false {
 		return nil, errors.New("proxy: enable_h2c is true but enable_http2 is false")

internal/proxy/change_log.go

@@ -14,7 +14,7 @@
 )
 
 // processChangeLog - processes a change log and applies the change to the proxy state
-func (ps *ProxyState) processChangeLog(cl *spec.ChangeLog, reload, store bool) (err error) {
+func (ps *ProxyState) processChangeLog(cl *spec.ChangeLog, reload, store bool) (restartNeeded bool, err error) {
 	if reload {
 		defer func(start time.Time) {
 			if err != nil {
@@ -68,12 +68,6 @@
 				} else if !ps.changeHash.CompareAndSwap(oldHash, newHash) {
 					goto hash_retry
 				}
-			} else {
-				go ps.restartState(func(err error) {
-					if err != nil {
-						ps.Stop()
-					}
-				})
 			}
 		}()
 		if cl.Cmd.Resource() == spec.Documents {
@@ -99,24 +93,25 @@
 
 	// apply state changes to the proxy
 	if reload {
+		ps.logger.Debug("Reloading change log", zap.String("id", cl.ID))
 		overrideReload := cl.Cmd.IsNoop() || ps.pendingChanges
 		if overrideReload || cl.Cmd.Resource().IsRelatedTo(spec.Routes) {
 			if err := ps.storeCachedDocuments(); err != nil {
 				ps.logger.Error("error storing cached documents", zap.Error(err))
-				return err
+				return false, err
 			}
 			ps.logger.Debug("Reloading change log", zap.String("id", cl.ID))
 			if err = ps.reconfigureState(cl); err != nil {
 				ps.logger.Error("Error registering change log", zap.Error(err))
-				return
+				return false, err
 			}
 			ps.pendingChanges = false
 		}
 	} else if !cl.Cmd.IsNoop() {
 		ps.pendingChanges = true
 	}
 
-	return nil
+	return restartNeeded, nil
 }
 
 func decode[T any](input any) (T, error) {
@@ -353,7 +348,8 @@
 		if cl.Cmd.Resource() == spec.Documents {
 			continue
 		}
-		if err = ps.processChangeLog(cl, false, false); err != nil {
+		_, err = ps.processChangeLog(cl, false, false)
+		if err != nil {
 			ps.logger.Error("error processing change log",
 				zap.Bool("skip", ps.debugMode),
 				zap.Error(err),
@@ -371,12 +367,15 @@
 		if err = ps.reconfigureState(cl); err != nil {
 			return err
 		}
-	} else if err = ps.processChangeLog(cl, true, false); err != nil {
-		return err
+	} else {
+		_, err = ps.processChangeLog(cl, true, false)
+		if err != nil {
+			return err
+		}
 	}
 
 	// DISABLED: compaction of change logs needs to have better testing
-	if len(logs) < 0 {
+	if false {
 		removed, err := ps.compactChangeLogs(logs)
 		if err != nil {
 			ps.logger.Error("failed to compact state change logs", zap.Error(err))

internal/proxy/dynamic_proxy.go

@@ -7,12 +7,14 @@
 	"math"
 	"net/http"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/dgate-io/dgate-api/internal/router"
 	"github.com/dgate-io/dgate-api/pkg/modules/extractors"
 	"github.com/dgate-io/dgate-api/pkg/spec"
 	"github.com/dgate-io/dgate-api/pkg/typescript"
+	"github.com/dgate-io/dgate-api/pkg/util/iplist"
 	"github.com/dgate-io/dgate-api/pkg/util/tree/avl"
 	"github.com/dop251/goja"
 	"go.uber.org/zap"
@@ -391,11 +393,33 @@
 	}
 }
 
-func (ps *ProxyState) HandleRoute(requestCtxProvider *RequestContextProvider, pattern string) http.HandlerFunc {
+func (ps *ProxyState) HandleRoute(ctxProvider *RequestContextProvider, pattern string) http.HandlerFunc {
+	ipList := iplist.NewIPList()
+	if len(ps.config.ProxyConfig.AllowList) > 0 {
+		for _, address := range ps.config.ProxyConfig.AllowList {
+			if strings.Contains(address, "/") {
+				if err := ipList.AddCIDRString(address); err != nil {
+					panic(fmt.Errorf("invalid cidr address in proxy.allow_list: %s", address))
+				}
+			} else {
+				if err := ipList.AddIPString(address); err != nil {
+					panic(fmt.Errorf("invalid ip address in proxy.allow_list: %s", address))
+				}
+			}
+		}
+	}
 	return func(w http.ResponseWriter, r *http.Request) {
-		// ctx, cancel := context.WithCancel(requestCtxPrdovider.ctx)
-		// defer cancel()
-		ps.ProxyHandler(ps, requestCtxProvider.
-			CreateRequestContext(requestCtxProvider.ctx, w, r, pattern))
+		if ipList.Len() > 0 {
+			allowed, err := ipList.Contains(r.RemoteAddr)
+			if err != nil {
+				ps.logger.Error("Error checking ")
+			}
+
+			if !allowed {
+				http.Error(w, "Forbidden", http.StatusForbidden)
+			}
+		}
+		reqContext := ctxProvider.CreateRequestContext(w, r, pattern)
+		ps.ProxyHandler(ps, reqContext)
 	}
 }

internal/proxy/proxy_handler.go

@@ -81,7 +81,7 @@
 }
 
 func handleServiceProxy(ps *ProxyState, reqCtx *RequestContext, modExt ModuleExtractor) {
-	var host string
+	var upstreamUrlString string
 	if fetchUpstreamUrl, ok := modExt.FetchUpstreamUrlFunc(); ok {
 		fetchUpstreamStart := time.Now()
 		hostUrl, err := fetchUpstreamUrl(modExt.ModuleContext())
@@ -98,17 +98,17 @@
 			util.WriteStatusCodeError(reqCtx.rw, http.StatusInternalServerError)
 			return
 		}
-		host = hostUrl.String()
+		upstreamUrlString = hostUrl.String()
 	} else {
-		if reqCtx.route.Service.URLs == nil || len(reqCtx.route.Service.URLs) == 0 {
+		if len(reqCtx.route.Service.URLs) == 0 {
 			ps.logger.Error("Error getting service urls",
 				zap.String("service", reqCtx.route.Service.Name),
 				zap.String("namespace", reqCtx.route.Namespace.Name),
 			)
 			util.WriteStatusCodeError(reqCtx.rw, http.StatusInternalServerError)
 			return
 		}
-		host = reqCtx.route.Service.URLs[0].String()
+		upstreamUrlString = reqCtx.route.Service.URLs[0].String()
 	}
 
 	if reqCtx.route.Service.HideDGateHeaders {
@@ -122,10 +122,10 @@
 
 		// downstream headers
 		if ps.debugMode {
-			reqCtx.rw.Header().Set("X-Upstream-URL", host)
+			reqCtx.rw.Header().Set("X-Upstream-URL", upstreamUrlString)
 		}
 	}
-	upstreamUrl, err := url.Parse(host)
+	upstreamUrl, err := url.Parse(upstreamUrlString)
 	if err != nil {
 		ps.logger.Error("Error parsing upstream url",
 			zap.String("error", err.Error()),

internal/proxy/proxy_handler_test.go

@@ -1,7 +1,6 @@
 package proxy_test
 
 import (
-	"context"
 	"errors"
 	"io"
 	"net/http"
@@ -57,8 +56,7 @@ func TestProxyHandler_ReverseProxy(t *testing.T) {
 		wr.SetWriteFallThrough()
 		wr.On("Header").Return(http.Header{})
 		wr.On("Write", mock.Anything).Return(0, nil).Maybe()
-		reqCtx := reqCtxProvider.CreateRequestContext(
-			context.Background(), wr, req, "/")
+		reqCtx := reqCtxProvider.CreateRequestContext(wr, req, "/")
 
 		modExt := NewMockModuleExtractor()
 		modExt.ConfigureDefaultMock(req, wr, ps, rt)
@@ -78,7 +76,7 @@ func TestProxyHandler_ReverseProxy(t *testing.T) {
 		modPool.AssertExpectations(t)
 		modExt.AssertExpectations(t)
 		rpBuilder.AssertExpectations(t)
-		// rpe.AssertExpectations(t)
+		rpe.AssertExpectations(t)
 	}
 }
 
@@ -129,8 +127,7 @@ func TestProxyHandler_ProxyHandler(t *testing.T) {
 		modPool.On("Return", modExt).Return().Once()
 		reqCtxProvider.UpdateModulePool(modPool)
 
-		reqCtx := reqCtxProvider.CreateRequestContext(
-			context.Background(), wr, req, "/")
+		reqCtx := reqCtxProvider.CreateRequestContext(wr, req, "/")
 		ps.ProxyHandler(ps, reqCtx)
 
 		wr.AssertExpectations(t)
@@ -181,8 +178,7 @@ func TestProxyHandler_ProxyHandlerError(t *testing.T) {
 		modPool.On("Return", modExt).Return().Once()
 		reqCtxProvider := proxy.NewRequestContextProvider(rt, ps)
 		reqCtxProvider.UpdateModulePool(modPool)
-		reqCtx := reqCtxProvider.CreateRequestContext(
-			context.Background(), wr, req, "/")
+		reqCtx := reqCtxProvider.CreateRequestContext(wr, req, "/")
 		ps.ProxyHandler(ps, reqCtx)
 
 		wr.AssertExpectations(t)