Compare commits

...

9 Commits

12 changed files with 222 additions and 39 deletions
+1 -1
View File
@@ -11,7 +11,7 @@ help: ## display this help
.PHONY: run
run: build ## run application
@LOG_LEVEL=debug BLOCK_SCHEMA_INTROSPECTION=false JWT_ROLE_RATE_LIMIT=false JWT_ROLE_CLAIM_PATH="Hasura.x-hasura-default-role" JWT_USER_CLAIM_PATH="Hasura.x-hasura-user-id" HOST_GRAPHQL=https://hasura8.lan/v1/graphql ./graphql-proxy
@LOG_LEVEL=debug BLOCK_SCHEMA_INTROSPECTION=false CACHE_TTL=10 JWT_ROLE_RATE_LIMIT=false JWT_ROLE_CLAIM_PATH="Hasura.x-hasura-default-role" JWT_USER_CLAIM_PATH="Hasura.x-hasura-user-id" HOST_GRAPHQL=https://hasura8.lan/ ./graphql-proxy
.PHONY: build
build: ## build the binary
+9 -2
View File
@@ -14,7 +14,7 @@ I wanted to monitor the queries and responses of our graphql endpoint. Still, we
### Endpoints
* `:8080/v1/graphql` - the graphql endpoint
* `:8080/*` - the graphql passthrough endpoint
* `:9393/metrics` - the prometheus metrics endpoint
* `:8080/healthz` - the healthcheck endpoint
@@ -30,6 +30,7 @@ I wanted to monitor the queries and responses of our graphql endpoint. Still, we
| security | Blocking schema introspection |
| security | Rate limiting queries based on user role |
| security | Blocking mutations in read-only mode |
| security | Allow access only to listed URLs |
### Configuration
@@ -38,7 +39,7 @@ I wanted to monitor the queries and responses of our graphql endpoint. Still, we
|---------------------------|------------------------------------------|----------------------------|
| `MONITORING_PORT` | The port to expose the metrics endpoint | `9393` |
| `PORT_GRAPHQL` | The port to expose the graphql endpoint | `8080` |
| `HOST_GRAPHQL` | The host to proxy the graphql endpoint | `http://localhost/v1/graphql` |
| `HOST_GRAPHQL` | The host to proxy the graphql endpoint | `http://localhost/` |
| `JWT_USER_CLAIM_PATH` | Path to the user claim in the JWT token | `` |
| `JWT_ROLE_CLAIM_PATH` | Path to the role claim in the JWT token | `` |
| `JWT_ROLE_FROM_HEADER` | Header name to extract the role from | `` |
@@ -49,6 +50,7 @@ I wanted to monitor the queries and responses of our graphql endpoint. Still, we
| `BLOCK_SCHEMA_INTROSPECTION`| Blocks the schema introspection | `false` |
| `ENABLE_ACCESS_LOG` | Enable the access log | `false` |
| `READ_ONLY_MODE` | Enable the read only mode | `false` |
| `ALLOWED_URLS` | Allow access only to certain URLs | `/v1/graphql,/v1/version` |
### Caching
@@ -101,6 +103,11 @@ If rate limit has been reached - the proxy will return `429 Too Many Requests` e
You can enable the read-only mode by setting the `READ_ONLY_MODE` environment variable to `true` - which will block all the `mutation` queries.
### Allowing access to listed URLs
You can allow access only to certain URLs by setting the `ALLOWED_URLS` environment variable to a comma-separated list of URLs. If enabled - other URLs will return `403 Forbidden` error and request will **not** reach the proxied service.
### Monitoring endpoint
Example metrics produced by the proxy:
+5 -12
View File
@@ -4,9 +4,9 @@ import (
"fmt"
"time"
"github.com/akyoto/cache"
fiber "github.com/gofiber/fiber/v2"
"github.com/gookit/goutil/strutil"
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
)
func calculateHash(c *fiber.Ctx) string {
@@ -14,20 +14,13 @@ func calculateHash(c *fiber.Ctx) string {
}
func enableCache() {
var err error
cfg.Cache.CacheClient = cache.New(time.Duration(cfg.Cache.CacheTTL) * time.Second * 2)
if err != nil {
cfg.Logger.Critical("Can't create cache client", map[string]interface{}{"error": err.Error()})
panic(err)
}
cfg.Cache.CacheClient = libpack_cache.New(time.Duration(cfg.Cache.CacheTTL) * time.Second * 100)
}
func cacheLookup(hash string) []byte {
if cfg.Cache.CacheClient != nil {
obj, found := cfg.Cache.CacheClient.Get(hash)
if found {
return obj.([]byte)
}
obj, found := cfg.Cache.CacheClient.Get(hash)
if found {
return obj
}
return nil
}
+112
View File
@@ -0,0 +1,112 @@
package libpack_cache
import (
"sync"
"time"
)
type CacheEntry struct {
Value []byte
ExpiresAt time.Time
}
type Cache struct {
sync.RWMutex
entries sync.Map
globalTTL time.Duration
bytePool sync.Pool
}
func New(globalTTL time.Duration) *Cache {
cache := &Cache{
globalTTL: globalTTL,
}
// Initialize the byte pool.
cache.bytePool.New = func() interface{} {
return make([]byte, 0)
}
// Start the cache cleanup.
go cache.cleanupRoutine(globalTTL)
return cache
}
func (c *Cache) cleanupRoutine(globalTTL time.Duration) {
ticker := time.NewTicker(globalTTL / 2)
defer ticker.Stop()
for range ticker.C {
c.CleanExpiredEntries()
}
}
func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
c.Lock()
defer c.Unlock()
expiresAt := time.Now().Add(ttl)
// Get a byte slice from the pool and ensure it's properly sized.
b := c.bytePool.Get().([]byte)
if cap(b) < len(value) {
b = make([]byte, len(value))
} else {
b = b[:len(value)]
}
copy(b, value)
entry := CacheEntry{
Value: b,
ExpiresAt: expiresAt,
}
c.entries.Store(key, entry)
}
func (c *Cache) Get(key string) ([]byte, bool) {
c.RLock()
defer c.RUnlock()
entry, ok := c.entries.Load(key)
if !ok || entry.(CacheEntry).ExpiresAt.Before(time.Now()) {
return nil, false
}
// Copy the value from the byte slice.
value := make([]byte, len(entry.(CacheEntry).Value))
copy(value, entry.(CacheEntry).Value)
return value, true
}
func (c *Cache) Delete(key string) {
c.Lock()
defer c.Unlock()
entry, ok := c.entries.Load(key)
if !ok {
return
}
// Return the byte slice to the pool.
c.bytePool.Put(entry.(CacheEntry).Value)
// Delete the entry from the cache.
c.entries.Delete(key)
}
func (c *Cache) CleanExpiredEntries() {
now := time.Now()
c.entries.Range(func(key, value interface{}) bool {
entry := value.(CacheEntry)
if entry.ExpiresAt.Before(now) {
// Return the byte slice to the pool.
c.bytePool.Put(entry.Value)
// Delete the entry from the cache.
c.entries.Delete(key)
}
// Return true to continue iterating over the map.
return true
})
}
+2 -2
View File
@@ -27,7 +27,7 @@ func (suite *Tests) Test_cacheLookup() {
{
name: "test_existent",
args: args{
hash: "00000000000000000000000000000000000001",
hash: "00000000000000000000000000000000001337",
},
want: []byte("it's fine."),
addCache: struct {
@@ -40,7 +40,7 @@ func (suite *Tests) Test_cacheLookup() {
for _, tt := range tests {
suite.T().Run(tt.name, func(t *testing.T) {
if tt.addCache.data != nil {
cfg.Cache.CacheClient.Set(tt.args.hash, tt.addCache.data, time.Duration(1)*time.Second)
cfg.Cache.CacheClient.Set(tt.args.hash, tt.addCache.data, time.Duration(90*time.Second))
}
got := cacheLookup(tt.args.hash)
assert.Equal(tt.want, got, "Unexpected cache lookup result")
+4 -4
View File
@@ -4,20 +4,21 @@ go 1.21
require (
github.com/VictoriaMetrics/metrics v1.24.0
github.com/akyoto/cache v1.0.6
github.com/buger/jsonparser v1.1.1
github.com/gofiber/fiber/v2 v2.49.2
github.com/gookit/goutil v0.6.12
github.com/gookit/goutil v0.6.13
github.com/graphql-go/graphql v0.8.1
github.com/json-iterator/go v1.1.12
github.com/lukaszraczylo/ask v0.0.0-20230927103145-2ff1123b4415
github.com/lukaszraczylo/go-ratecounter v0.1.8
github.com/lukaszraczylo/go-simple-graphql v1.1.31
github.com/lukaszraczylo/go-simple-graphql v1.1.32
github.com/rs/zerolog v1.31.0
github.com/stretchr/testify v1.8.4
github.com/valyala/fasthttp v1.50.0
)
require (
github.com/akyoto/cache v1.0.6 // indirect
github.com/andybalholm/brotli v1.0.5 // indirect
github.com/avast/retry-go/v4 v4.5.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
@@ -34,7 +35,6 @@ require (
github.com/rivo/uniseg v0.4.4 // indirect
github.com/rogpeppe/go-internal v1.11.0 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fasthttp v1.50.0 // indirect
github.com/valyala/fastrand v1.1.0 // indirect
github.com/valyala/histogram v1.2.0 // indirect
github.com/valyala/tcplisten v1.0.0 // indirect
+4 -4
View File
@@ -21,8 +21,8 @@ github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gookit/color v1.5.4 h1:FZmqs7XOyGgCAxmWyPslpiok1k05wmY3SJTytgvYFs0=
github.com/gookit/color v1.5.4/go.mod h1:pZJOeOS8DM43rXbp4AZo1n9zCU2qjpcRko0b6/QJi9w=
github.com/gookit/goutil v0.6.12 h1:73vPUcTtVGXbhSzBOFcnSB1aJl7Jq9np3RAE50yIDZc=
github.com/gookit/goutil v0.6.12/go.mod h1:g6krlFib8xSe3G1h02IETowOtrUGpAmetT8IevDpvpM=
github.com/gookit/goutil v0.6.13 h1:ttg7yMda6Q9fkE4P+YTwozd2wH1Le0CQldTAtOFBr7o=
github.com/gookit/goutil v0.6.13/go.mod h1:YyDBddefmjS+mU2PDPgCcjVzTDM5WgExiDv5ZA/b8I8=
github.com/graphql-go/graphql v0.8.1 h1:p7/Ou/WpmulocJeEx7wjQy611rtXGQaAcXGqanuMMgc=
github.com/graphql-go/graphql v0.8.1/go.mod h1:nKiHzRM0qopJEwCITUuIsxk9PlVlwIiiI8pnJEhordQ=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
@@ -40,8 +40,8 @@ github.com/lukaszraczylo/ask v0.0.0-20230927103145-2ff1123b4415 h1:lvI8Wlbg4PxkR
github.com/lukaszraczylo/ask v0.0.0-20230927103145-2ff1123b4415/go.mod h1:M+UVdyqZs++xtEPrascaVmZdOMhCnxjZ2SgH+xHpR0c=
github.com/lukaszraczylo/go-ratecounter v0.1.8 h1:ZYm6Wkn58ZAlFWRmC7PaD4oAYHWcu8/0MUDWGe3PnJQ=
github.com/lukaszraczylo/go-ratecounter v0.1.8/go.mod h1:TqXEOCtFJStk1i0tkipprv1kiDHGon1MVUisjSTBSKM=
github.com/lukaszraczylo/go-simple-graphql v1.1.31 h1:UA3f8M1cV+XnO8UZlAqveW0qF/2NN512eB/gRqe+BHs=
github.com/lukaszraczylo/go-simple-graphql v1.1.31/go.mod h1:MyftQ8jTdtkYImPXJpHoxz6+E53Ydv+7q9+Jr+eT8WU=
github.com/lukaszraczylo/go-simple-graphql v1.1.32 h1:udiz2hnLNO2b/hc9Z0xcjYt+HcFbChQQuIY4HZmky80=
github.com/lukaszraczylo/go-simple-graphql v1.1.32/go.mod h1:c1nN/qtjsvpqpkBFsnBCYCDLdLMuWzy/zxeJzTjm5qg=
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+5 -3
View File
@@ -36,12 +36,13 @@ var retrospection_queries = []string{
// Saving the introspection queries as a map O(1) operation instead of O(n) for a slice.
var retrospectionQuerySet = make(map[string]struct{}, len(retrospection_queries))
func parseGraphQLQuery(c *fiber.Ctx) (operationType, operationName string, cacheRequest bool, cache_time int, should_block bool) {
func parseGraphQLQuery(c *fiber.Ctx) (operationType, operationName string, cacheRequest bool, cache_time int, should_block bool, should_ignore bool) {
should_ignore = true
m := make(map[string]interface{})
err := json.Unmarshal(c.Body(), &m)
if err != nil {
cfg.Logger.Error("Can't unmarshal the request", map[string]interface{}{"error": err.Error(), "body": string(c.Body())})
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
cfg.Logger.Debug("Can't unmarshal the request", map[string]interface{}{"error": err.Error(), "body": string(c.Body())})
cfg.Monitoring.Increment(libpack_monitoring.MetricsSkipped, nil)
return
}
// get the query
@@ -59,6 +60,7 @@ func parseGraphQLQuery(c *fiber.Ctx) (operationType, operationName string, cache
return
}
should_ignore = false
operationName = "undefined"
for _, d := range p.Definitions {
if oper, ok := d.(*ast.OperationDefinition); ok {
+11 -1
View File
@@ -1,6 +1,8 @@
package main
import (
"strings"
"github.com/gookit/goutil/envutil"
graphql "github.com/lukaszraczylo/go-simple-graphql"
libpack_config "github.com/lukaszraczylo/graphql-monitoring-proxy/config"
@@ -20,7 +22,7 @@ func parseConfig() {
var c config
c.Server.PortGraphQL = envutil.GetInt("PORT_GRAPHQL", 8080)
c.Server.PortMonitoring = envutil.GetInt("MONITORING_PORT", 9393)
c.Server.HostGraphQL = envutil.Getenv("HOST_GRAPHQL", "http://localhost/v1/graphql")
c.Server.HostGraphQL = envutil.Getenv("HOST_GRAPHQL", "http://localhost/")
c.Client.JWTUserClaimPath = envutil.Getenv("JWT_USER_CLAIM_PATH", "")
c.Client.JWTRoleClaimPath = envutil.Getenv("JWT_ROLE_CLAIM_PATH", "")
c.Client.RoleFromHeader = envutil.Getenv("ROLE_FROM_HEADER", "")
@@ -33,6 +35,14 @@ func parseConfig() {
c.Client.GQLClient.SetEndpoint(c.Server.HostGraphQL)
c.Server.AccessLog = envutil.GetBool("ENABLE_ACCESS_LOG", false)
c.Server.ReadOnlyMode = envutil.GetBool("READ_ONLY_MODE", false)
c.Server.AllowURLs = func() []string {
urls := envutil.Getenv("ALLOWED_URLS", "")
if urls == "" {
return nil
}
return strings.Split(urls, ",")
}()
c.Client.FastProxyClient = createFasthttpClient()
cfg = &c
enableCache() // takes close to no resources, but can be used with dynamic query cache
loadRatelimitConfig()
+35 -5
View File
@@ -2,26 +2,56 @@ package main
import (
"crypto/tls"
"fmt"
"time"
fiber "github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/proxy"
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
"github.com/valyala/fasthttp"
)
func createFasthttpClient() *fasthttp.Client {
return &fasthttp.Client{
Name: "graphql_proxy",
NoDefaultUserAgentHeader: true,
TLSConfig: &tls.Config{
InsecureSkipVerify: true,
},
MaxConnsPerHost: 100,
MaxIdleConnDuration: 2 * time.Minute,
ReadTimeout: time.Second * 10,
WriteTimeout: time.Second * 10,
DisableHeaderNamesNormalizing: true,
}
}
func proxyTheRequest(c *fiber.Ctx) error {
if !checkAllowedURLs(c) {
cfg.Logger.Error("Request blocked", map[string]interface{}{"path": c.Path()})
cfg.Monitoring.Increment(libpack_monitoring.MetricsSkipped, nil)
c.Status(403).SendString("Request blocked - not allowed URL")
return nil
}
c.Request().Header.Add("X-Real-IP", c.IP())
c.Request().Header.Add("X-Forwarded-For", c.IP())
c.Request().Header.Add(fiber.HeaderXForwardedFor, string(c.Request().Header.Peek("X-Forwarded-For")))
proxy.WithTlsConfig(&tls.Config{
InsecureSkipVerify: true,
})
proxy.WithClient(cfg.Client.FastProxyClient)
err := proxy.DoRedirects(c, cfg.Server.HostGraphQL, 3)
cfg.Logger.Debug("Proxying the request", map[string]interface{}{"path": c.Path(), "body": string(c.Request().Body())})
err := proxy.DoRedirects(c, cfg.Server.HostGraphQL+c.Path(), 3)
if err != nil {
cfg.Logger.Error("Can't proxy the request", map[string]interface{}{"error": err.Error()})
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
return err
}
cfg.Logger.Debug("Received proxied response", map[string]interface{}{"path": c.Path(), "response_body": string(c.Response().Body()), "response_code": c.Response().StatusCode()})
if c.Response().StatusCode() != 200 {
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
return fmt.Errorf("Received non-200 response from the GraphQL server: %d", c.Response().StatusCode())
}
c.Response().Header.Del(fiber.HeaderServer)
return nil
+28 -3
View File
@@ -21,7 +21,8 @@ func StartHTTPProxy() {
AllowOrigins: "*",
}))
server.Post("/v1/graphql", processGraphQLRequest)
server.Post("/*", processGraphQLRequest)
server.Get("/*", proxyTheRequest)
server.Get("/healthz", healthCheck)
err := server.Listen(fmt.Sprintf(":%d", cfg.Server.PortGraphQL))
@@ -30,6 +31,18 @@ func StartHTTPProxy() {
}
}
func checkAllowedURLs(c *fiber.Ctx) bool {
if len(cfg.Server.AllowURLs) == 0 {
return true
}
for _, allowedURL := range cfg.Server.AllowURLs {
if c.Path() == allowedURL {
return true
}
}
return false
}
func healthCheck(c *fiber.Ctx) error {
// query := `{ __typename }`
// _, err := cfg.Client.GQLClient.Query(query, nil, nil)
@@ -70,11 +83,16 @@ func processGraphQLRequest(c *fiber.Ctx) error {
}
}
opType, opName, cacheFromQuery, cache_time, shouldBlock := parseGraphQLQuery(c)
opType, opName, cacheFromQuery, cache_time, shouldBlock, should_ignore := parseGraphQLQuery(c)
if shouldBlock {
return nil
}
if should_ignore {
cfg.Logger.Debug("Request passed as-is - not a GraphQL")
return proxyTheRequest(c)
}
if cache_time > 0 {
cfg.Logger.Debug("Cache time set via query", map[string]interface{}{"cache_time": cache_time})
cache_time = cfg.Cache.CacheTTL
@@ -109,7 +127,13 @@ func processGraphQLRequest(c *fiber.Ctx) error {
// Additional helper function to avoid code repetition
func proxyAndCacheTheRequest(c *fiber.Ctx, queryCacheHash string, cache_time int) {
proxyTheRequest(c)
err := proxyTheRequest(c)
if err != nil {
cfg.Logger.Error("Can't proxy the request", map[string]interface{}{"error": err.Error()})
cfg.Monitoring.Increment(libpack_monitoring.MetricsFailed, nil)
c.Status(500).SendString("Can't proxy the request - try again later")
return
}
cfg.Cache.CacheClient.Set(queryCacheHash, c.Response().Body(), time.Duration(cache_time)*time.Second)
c.Send(c.Response().Body())
}
@@ -125,6 +149,7 @@ func logAndMonitorRequest(c *fiber.Ctx, userID, opType, opName string, wasCached
if cfg.Server.AccessLog {
cfg.Logger.Info("Request processed", map[string]interface{}{
"ip": c.IP(),
"fwd-ip": string(c.Request().Header.Peek("X-Forwarded-For")),
"user_id": userID,
"op_type": opType,
"op_name": opName,
+6 -2
View File
@@ -1,10 +1,11 @@
package main
import (
"github.com/akyoto/cache"
graphql "github.com/lukaszraczylo/go-simple-graphql"
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
libpack_logging "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
"github.com/valyala/fasthttp"
)
// config is a struct that holds the configuration of the application.
@@ -19,6 +20,7 @@ type config struct {
HostGraphQL string
AccessLog bool
ReadOnlyMode bool
AllowURLs []string
}
Client struct {
@@ -27,12 +29,14 @@ type config struct {
RoleRateLimit bool
RoleFromHeader string
GQLClient *graphql.BaseClient
FastProxyClient *fasthttp.Client
proxy string
}
Cache struct {
CacheEnable bool
CacheTTL int
CacheClient *cache.Cache
CacheClient *libpack_cache.Cache
}
Security struct {