Add Kafka log consumer with dynamic authentication and configuration

- Implement dynamic Kafka authentication mechanism (PLAIN/SCRAM)
- Add environment variable-based Kafka broker and authentication configuration
- Modify log consumer service to support flexible Kafka connection setup
- Update Elasticsearch store initialization with new index template
- Extend WAF service types to support listener name and hosts

api/router.go

@@ -7,11 +7,13 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/rs/zerolog/log"
 	"gitlab.com/tensorsecurity-rd/waf-console/internal/config"
+
+	//"gitlab.com/tensorsecurity-rd/waf-console/internal/middleware"
 	"gitlab.com/tensorsecurity-rd/waf-console/internal/utils"
 	"gorm.io/gorm"
 )
 
-func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager, gatewayUrl string) *gin.Engine {
+func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager, gatewayUrl string, ssoUrl string) *gin.Engine {
 	var engine *gin.Engine
 
 	if !config.Conf.Debug {
@@ -19,17 +21,19 @@ func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager, g
 		log.Info().Msg("release mode")
 		engine = ReleaseRouter()
 		engine.Use(
+		// middleware.AuthMiddleware(ssoUrl),
 		// middleware.RequestCostHandler(),
 		// middleware.CustomLogger(),
 		// middleware.CustomRecovery(),
 		// middleware.CorsHandler(),
+		// middleware.CorsHandler(),
 		)
 	} else {
 		// 开发调试模式
 		log.Info().Msg("debug mode")
 		engine = gin.New()
 		engine.Use(
-			// middleware.RequestCostHandler(),
+			// middleware.AuthMiddleware(ssoUrl),
 			gin.Logger(),
 			// middleware.CustomRecovery(),
 			// middleware.CorsHandler(),

build/Dockerfile

@@ -3,6 +3,7 @@ FROM ubuntu:22.04
 ARG MIRROR_SOURCE=mirrors.aliyun.com
 
 ADD dist/waf-console /
+COPY config/waf_detection_index_template.json /
 
 
 

cmd/app/cmd.go

@@ -4,10 +4,13 @@ import (
 	"encoding/base64"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
 	"gitlab.com/tensorsecurity-rd/waf-console/api"
+	"gitlab.com/tensorsecurity-rd/waf-console/internal/service"
+	es "gitlab.com/tensorsecurity-rd/waf-console/internal/store"
 	"gitlab.com/tensorsecurity-rd/waf-console/internal/utils"
 	"gitlab.com/tensorsecurity-rd/waf-console/pkg/generated/clientset/versioned"
 	"gorm.io/driver/mysql"
@@ -77,7 +80,19 @@ func NewRootCommand() *cobra.Command {
 				clusterClientManager.AddClient(regionConfig.RegionCode, client)
 			}
 
-			e := api.SetRouters(db, clusterClientManager, config.GatewayUrl)
+			esClient, err := es.CreateEsClientFromEnv()
+			if err != nil {
+				panic(err)
+			}
+			e := api.SetRouters(db, clusterClientManager, config.GatewayUrl, config.SSOUrl)
+			esStore := es.NewESStore(es.Config{
+				ESBatchSize:   100,
+				ESConcurrency: 10,
+				ESTimeout:     10 * time.Second,
+			}, esClient)
+			esStore.Init()
+			logConsumerService := service.NewLogConsumerService(db, esStore)
+			go logConsumerService.Consume()
 			return e.Run(":8080")
 		},
 	}

cmd/app/config.go

@@ -20,6 +20,7 @@ type Config struct {
 	RegionConfigs []RegionConfig `json:"region_configs"`
 	Debug         bool           `json:"debug"`
 	GatewayUrl    string         `json:"gateway_url"`
+	SSOUrl        string         `json:"sso_url"`
 }
 
 type DBConfig struct {

config/waf_detection_index_template.json

+{
+    "template": {
+      "settings": {
+        "index": {
+          "lifecycle": {
+            "name": "iml-event-signal",
+            "rollover_alias": "waf-detections"
+          },
+          "search": {
+            "slowlog": {
+              "threshold": {
+                "fetch": {
+                  "warn": "1s",
+                  "trace": "200ms",
+                  "debug": "500ms",
+                  "info": "800ms"
+                },
+                "query": {
+                  "warn": "3s",
+                  "trace": "500ms",
+                  "debug": "1s",
+                  "info": "2s"
+                }
+              }
+            }
+          },
+          "refresh_interval": "10s",
+          "indexing": {
+            "slowlog": {
+              "threshold": {
+                "index": {
+                  "warn": "2s",
+                  "trace": "500ms",
+                  "debug": "800ms",
+                  "info": "1s"
+                }
+              }
+            }
+          },
+          "number_of_shards": "3",
+          "translog": {
+            "flush_threshold_size": "1024mb",
+            "sync_interval": "60s",
+            "durability": "async"
+          },
+          "merge": {
+            "scheduler": {
+              "max_thread_count": "1"
+            }
+          },
+          "sort": {
+            "field": [
+              "id.digit",
+              "attack_time"
+            ],
+            "order": [
+              "desc",
+              "desc"
+            ]
+          },
+          "analysis": {
+            "analyzer": {
+              "ngramTokenizerAnalyzer": {
+                "filter": [
+                  "lowercase"
+                ],
+                "type": "custom",
+                "tokenizer": "ngram_tokenizer"
+              },
+              "ngramTokenizerAnalyzerCustomSymbolPunctuation": {
+                "filter": [
+                  "lowercase"
+                ],
+                "type": "custom",
+                "tokenizer": "ngram_tokenizer_custom_symbol_punctuation"
+              }
+            },
+            "tokenizer": {
+              "ngram_tokenizer": {
+                "token_chars": [
+                  "letter",
+                  "digit"
+                ],
+                "min_gram": "1",
+                "type": "ngram",
+                "max_gram": "1"
+              },
+              "ngram_tokenizer_custom_symbol_punctuation": {
+                "token_chars": [
+                  "letter",
+                  "digit",
+                  "symbol",
+                  "punctuation"
+                ],
+                "min_gram": "1",
+                "type": "ngram",
+                "max_gram": "1"
+              }
+            }
+          },
+          "number_of_replicas": "1"
+        }
+      },
+      "mappings": {
+        "dynamic": false,
+        "_source": {
+          "enabled": true,
+          "includes": [],
+          "excludes": []
+        },
+        "_routing": {
+          "required": false
+        },
+        "dynamic_templates": [],
+        "properties": {
+          "id": {
+            "eager_global_ordinals": false,
+            "index_phrases": false,
+            "fielddata": false,
+            "norms": true,
+            "analyzer": "ngramTokenizerAnalyzer",
+            "index": true,
+            "store": false,
+            "type": "text",
+            "fields": {
+              "digit": {
+                "coerce": true,
+                "index": true,
+                "ignore_malformed": false,
+                "store": false,
+                "type": "long",
+                "doc_values": true
+              },
+              "keyword": {
+                "eager_global_ordinals": false,
+                "norms": false,
+                "ignore_above": 32,
+                "index": true,
+                "store": false,
+                "type": "keyword",
+                "split_queries_on_whitespace": false,
+                "index_options": "docs",
+                "doc_values": true
+              }
+            },
+            "index_options": "positions"
+          },
+          "attacked_url": {
+            "eager_global_ordinals": false,
+            "index_phrases": false,
+            "fielddata": false,
+            "norms": true,
+            "analyzer": "ngramTokenizerAnalyzerCustomSymbolPunctuation",
+            "index": true,
+            "store": false,
+            "type": "text",
+            "fields": {
+              "keyword": {
+                "eager_global_ordinals": false,
+                "norms": false,
+                "ignore_above": 32,
+                "index": true,
+                "store": false,
+                "type": "keyword",
+                "split_queries_on_whitespace": false,
+                "index_options": "docs",
+                "doc_values": true
+              }
+            },
+            "index_options": "positions"
+          },
+          "attack_ip": {
+            "eager_global_ordinals": false,
+            "index_phrases": false,
+            "fielddata": false,
+            "norms": true,
+            "analyzer": "ngramTokenizerAnalyzerCustomSymbolPunctuation",
+            "index": true,
+            "store": false,
+            "type": "text",
+            "fields": {
+              "keyword": {
+                "eager_global_ordinals": false,
+                "norms": false,
+                "ignore_above": 32,
+                "index": true,
+                "store": false,
+                "type": "keyword",
+                "split_queries_on_whitespace": false,
+                "index_options": "docs",
+                "doc_values": true
+              },
+              "ip": {
+                "type": "ip"
+              }
+            },
+            "index_options": "positions"
+          },
+          "attacked_app": {
+            "eager_global_ordinals": false,
+            "index_phrases": false,
+            "fielddata": false,
+            "norms": true,
+            "analyzer": "ngramTokenizerAnalyzerCustomSymbolPunctuation",
+            "index": true,
+            "store": false,
+            "type": "text",
+            "fields": {
+              "keyword": {
+                "eager_global_ordinals": false,
+                "norms": false,
+                "ignore_above": 32,
+                "index": true,
+                "store": false,
+                "type": "keyword",
+                "split_queries_on_whitespace": false,
+                "index_options": "docs",
+                "doc_values": true
+              }
+            },
+            "index_options": "positions"
+          },
+          "attack_type": {
+            "type": "keyword"
+          },
+          "cluster_key": {
+            "type": "keyword"
+          },
+          "attack_time": {
+            "coerce": true,
+            "index": true,
+            "ignore_malformed": false,
+            "store": false,
+            "type": "long",
+            "doc_values": true
+          },
+          "action": {
+            "type": "keyword"
+          },
+          "created_at": {
+            "type": "long"
+          },
+          "service_id": {
+            "type": "long"
+          }
+        }
+      }
+    },
+    "index_patterns": [
+      "waf-detections-*"
+    ],
+    "composed_of": []
+  }
\ No newline at end of file

go.mod

@@ -4,6 +4,9 @@ go 1.20
 
 require (
 	github.com/gin-gonic/gin v1.10.0
+	github.com/olivere/elastic/v7 v7.0.32
+	github.com/segmentio/kafka-go v0.4.47
+	github.com/sony/sonyflake v1.0.0
 	gorm.io/driver/mysql v1.5.0
 	gorm.io/gorm v1.25.12
 	k8s.io/apimachinery v0.27.2
@@ -40,8 +43,8 @@ require (
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
-	gopkg.in/yaml.v2 v2.4.0
-	gopkg.in/yaml.v3 v3.0.1
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/api v0.27.2 // indirect
 	k8s.io/gengo v0.0.0-20210813121822-485abfe95c7c // indirect
 	k8s.io/klog/v2 v2.90.1 // indirect
@@ -55,10 +58,10 @@ require (
 require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/klauspost/compress v1.15.9 // indirect
-	github.com/olivere/elastic/v7 v7.0.32 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
-	github.com/segmentio/kafka-go v0.4.40 // indirect
-	github.com/sony/sonyflake v1.0.0 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
 )
 
 require (
@@ -72,7 +75,7 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.22.0 // indirect
 	github.com/goccy/go-json v0.10.3 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect

go.sum

@@ -23,6 +23,7 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/deckarep/golang-set v1.7.1 h1:SCQV0S6gTtp6itiFrTqI+pfmJ4LN85S1YzhDf9rTHJQ=
 github.com/deckarep/golang-set v1.7.1/go.mod h1:93vsz/8Wt4joVM7c2AVqh+YRMiUSc14yDtF28KmMOgQ=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
@@ -33,6 +34,7 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/evanphx/json-patch v5.6.0+incompatible h1:jBYDEEiFBPxA0v50tFdvOzQQTCvpL6mnFh5mB2/l16U=
 github.com/evanphx/json-patch v5.6.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
+github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/gabriel-vasile/mimetype v1.4.5 h1:J7wGKdGu33ocBOhGy0z653k/lFKLFDPJMG8Gql0kxn4=
 github.com/gabriel-vasile/mimetype v1.4.5/go.mod h1:ibHel+/kbxn9x2407k1izTA1S81ku1z/DlgOW2QE0M4=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -184,8 +186,8 @@ github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.29.1 h1:cO+d60CHkknCbvzEWxP0S9K6KqyTjrCNUy1LdQLCGPc=
 github.com/rs/zerolog v1.29.1/go.mod h1:Le6ESbR7hc+DP6Lt1THiV8CQSdkkNrd3R0XbEgp3ZBU=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
-github.com/segmentio/kafka-go v0.4.40 h1:sszW7c0/uyv7+VcTW5trx2ZC7kMWDTxuR/6Zn8U1bm8=
-github.com/segmentio/kafka-go v0.4.40/go.mod h1:naFEZc5MQKdeL3W6NkZIAn48Y6AazqjRFDhnXeg3h94=
+github.com/segmentio/kafka-go v0.4.47 h1:IqziR4pA3vrZq7YdRxaT3w1/5fvIH5qpCwstUanQQB0=
+github.com/segmentio/kafka-go v0.4.47/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/sony/sonyflake v1.0.0 h1:MpU6Ro7tfXwgn2l5eluf9xQvQJDROTBImNCfRXn/YeM=
 github.com/sony/sonyflake v1.0.0/go.mod h1:Jv3cfhf/UFtolOTTRd3q4Nl6ENqM+KfyZ5PseKfZGF4=
 github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
@@ -213,8 +215,11 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
 github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -226,6 +231,7 @@ golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
 golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -235,6 +241,7 @@ golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
 golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -252,6 +259,9 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE=
 golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -263,6 +273,7 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -279,10 +290,15 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
 golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
 golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -291,6 +307,9 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
 golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
@@ -306,6 +325,7 @@ golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

internal/middleware/auth.go

+package middleware
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"gitlab.com/tensorsecurity-rd/waf-console/internal/utils"
+)
+
+const (
+	// Cookie name to check for the auth token
+	AuthCookieName = "auth_token"
+)
+
+// AuthMiddleware validates the auth cookie with SSO service
+func AuthMiddleware(ssoUrl string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		// Get auth cookie
+		cookies := c.Request.Cookies()
+		if len(cookies) == 0 {
+			utils.AssembleResponse(c, nil, utils.ErrUnauthorized)
+			c.Abort()
+			return
+		}
+
+		// Create request to SSO service
+		req, err := http.NewRequest(http.MethodPost, ssoUrl, nil)
+		if err != nil {
+			utils.AssembleResponse(c, nil, utils.ErrInternalServer)
+			c.Abort()
+			return
+		}
+
+		// Add auth cookie to request
+		for _, cookie := range cookies {
+			req.AddCookie(cookie)
+		}
+
+		// Make request to SSO service
+		client := &http.Client{}
+		resp, err := client.Do(req)
+		if err != nil {
+			utils.AssembleResponse(c, nil, utils.ErrInternalServer)
+			c.Abort()
+			return
+		}
+		defer resp.Body.Close()
+
+		// Read response body
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			utils.AssembleResponse(c, nil, utils.ErrInternalServer)
+			c.Abort()
+			return
+		}
+
+		// Parse SSO response
+		var ssoResp SSOResponse
+		if err := json.Unmarshal(body, &ssoResp); err != nil {
+			utils.AssembleResponse(c, nil, utils.ErrInternalServer)
+			c.Abort()
+			return
+		}
+
+		// Check if authentication was successful
+		if ssoResp.Code != "OK" {
+			utils.AssembleResponse(c, nil, utils.ErrUnauthorized)
+			c.Abort()
+			return
+		}
+
+		// Store user info in context for later use
+		c.Set("userInfo", ssoResp.Data)
+
+		// Authentication successful, continue
+		c.Next()
+	}
+}

internal/middleware/types.go

+package middleware
+
+// SSOResponse represents the response structure from SSO service
+type SSOResponse struct {
+	Code    string   `json:"code"`
+	Message string   `json:"message"`
+	Data    UserInfo `json:"data"`
+}
+
+// UserInfo represents the user information in SSO response
+type UserInfo struct {
+	UserID          string      `json:"userID"`
+	UserName        string      `json:"userName"`
+	Email           string      `json:"email"`
+	Mobile          string      `json:"mobile"`
+	AccountID       string      `json:"accountID"`
+	IsEnableConsole int         `json:"isEnableConsole"`
+	IsEnableProgram int         `json:"isEnableProgram"`
+	IsRoot          int         `json:"isRoot"`
+	Type            string      `json:"type"`
+	OtherInfos      interface{} `json:"otherInfos"`
+}

internal/service/log_consumer.go

@@ -3,9 +3,16 @@ package service
 import (
 	"context"
 	"encoding/json"
+	"errors"
+	"os"
+	"strings"
+	"time"
 
 	"github.com/rs/zerolog/log"
 	"github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+	"github.com/segmentio/kafka-go/sasl/scram"
 	"gitlab.com/tensorsecurity-rd/waf-console/internal/model"
 	"gitlab.com/tensorsecurity-rd/waf-console/internal/utils/id"
 
@@ -22,13 +29,85 @@ const (
 	EsIndexWafDetectionsAlias = "waf-detections"
 )
 
+var scramAlgo = map[string]scram.Algorithm{
+	"SHA512": scram.SHA512,
+	"SHA256": scram.SHA256,
+}
+
+var ErrMissingBrokers = errors.New("no KAFKA_BROKERS env variable set")
+var ErrMissingAuthInfo = errors.New("no username or passwrod set when auth is enabled")
+var ErrUnsupportedAuthMethod = errors.New("unsupported auth method")
+
+const (
+	EnvKafkaAuthMethod   = "KAFKA_AUTH_METHOD"
+	EnvKafkaAuthUsername = "KAFKA_AUTH_USERNAME"
+	EnvKafkaAuthPassword = "KAFKA_AUTH_PASSWORD"
+	EnvKafkaScramAlgo    = "KAFKA_SCRAM_ALGO"
+	EnvKafkaBrokers      = "KAFKA_BROKERS"
+	KafkaAuthPlain       = "plain"
+	KafkaAuthScram       = "scram"
+)
+
+func getSASLMechanismByEnv() (sasl.Mechanism, bool, error) {
+	authMethod := os.Getenv(EnvKafkaAuthMethod)
+	username := os.Getenv(EnvKafkaAuthUsername)
+	password := os.Getenv(EnvKafkaAuthPassword)
+	var mechanism sasl.Mechanism
+	switch authMethod {
+	case KafkaAuthPlain:
+		if username == "" || password == "" {
+			return nil, true, ErrMissingAuthInfo
+		}
+		mechanism = &plain.Mechanism{
+			Username: username,
+			Password: password,
+		}
+	case KafkaAuthScram:
+		if username == "" || password == "" {
+			return nil, true, ErrMissingAuthInfo
+		}
+		algoKey := os.Getenv(EnvKafkaScramAlgo)
+		algo := scram.SHA512
+		algoTmp, exist := scramAlgo[algoKey]
+		if exist {
+			algo = algoTmp
+		}
+		var err error
+		mechanism, err = scram.Mechanism(algo, username, password)
+		if err != nil {
+			return nil, true, err
+		}
+	}
+	return mechanism, mechanism != nil, nil
+}
+
 type LogConsumerService struct {
 	consumer *kafka.Reader
 	db       *gorm.DB
 	esStore  *esStore.ESStore
 }
 
-func NewLogConsumerService(consumer *kafka.Reader, db *gorm.DB, esStore *esStore.ESStore) *LogConsumerService {
+func NewLogConsumerService(db *gorm.DB, esStore *esStore.ESStore) *LogConsumerService {
+	mechanism, _, err := getSASLMechanismByEnv()
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to get SASL mechanism")
+	}
+	brokers := strings.Split(os.Getenv(EnvKafkaBrokers), ",")
+	if len(brokers) == 0 {
+		log.Fatal().Msg("no KAFKA_BROKERS env variable set")
+	}
+
+	consumer := kafka.NewReader(kafka.ReaderConfig{
+		Brokers: brokers,
+		GroupID: KafkaGroupIvanWafDetectionsPalace,
+		Topic:   KafkaTopicIvanWafDetections,
+		Dialer: &kafka.Dialer{
+			SASLMechanism: mechanism,
+			Timeout:       10 * time.Second,
+			DualStack:     true,
+		},
+	})
+
 	return &LogConsumerService{
 		consumer: consumer,
 		db:       db,

internal/service/types.go

@@ -189,6 +189,7 @@ type CreateWafReq struct {
 	Port         uint32   `json:"port"`
 	Host         []string `json:"host"`
 	Mode         WafMode  `json:"mode"`
+	ListenerName string   `json:"listener_name"`
 }
 
 type DeleteWafReq struct {
@@ -223,6 +224,7 @@ type EnableListenerWafReq struct {
 	Hosts        []string `json:"hosts"`
 	Port         int      `json:"port"`
 	Mode         WafMode  `json:"mode"`
+	ListenerName string   `json:"listener_name"`
 }
 
 type EnableGatewayWafReq struct {
@@ -232,7 +234,7 @@ type EnableGatewayWafReq struct {
 }
 
 type ListenerWaf struct {
-	// Hosts        []string `json:"hosts"`
+	Hosts []string `json:"hosts"`
 	// Port         int      `json:"port"`
 	HostsAndPort string `json:"hosts_and_port"`
 	Name         string `json:"name"`

internal/service/waf.go

@@ -202,9 +202,10 @@ func (s *wafService) CreateWaf(ctx context.Context, req *CreateWafReq) (*WafServ
 			ServiceName: req.GatewayName,
 			Port:        req.Port,
 			Workload: v1alpha1.WorkloadRef{
-				Kind:      "Deployment",
-				Name:      req.GatewayName,
+				Kind:       req.GatewayName,
+				Name:       req.ListenerName,
 				Namespace:  req.Namespace,
+				ClusterKey: req.RegionCode,
 			},
 			Uri: &v1alpha1.StringMatch{
 				Prefix: "/",
@@ -437,6 +438,7 @@ func (s *wafService) EnableListenerWaf(ctx context.Context, req *EnableListenerW
 			Port:         uint32(req.Port),
 			Host:         req.Hosts,
 			Mode:         req.Mode,
+			ListenerName: req.ListenerName,
 		})
 		if err != nil {
 			return err
@@ -656,7 +658,7 @@ func (s *wafService) EnableListenerWafs(ctx context.Context, req *EnableListener
 	currentPortSet := sets.NewInt(portList...)
 
 	desiredPortSet := sets.NewInt()
-	wafMap := map[int][]string{}
+	wafMap := map[int]ListenerWaf{}
 	for _, listener := range req.Listeners {
 		// get port from listener.HostsAndPort, like hosts1@127.0.0.1@abc.com-8080
 		index := strings.LastIndex(listener.HostsAndPort, "-")
@@ -672,7 +674,11 @@ func (s *wafService) EnableListenerWafs(ctx context.Context, req *EnableListener
 		log.Info().Msgf("listener: %v", listener.Name)
 
 		hosts := strings.Split(listener.HostsAndPort[:index], "@")
-		wafMap[portInt] = hosts
+		wafMap[portInt] = ListenerWaf{
+			Hosts:        hosts,
+			HostsAndPort: listener.HostsAndPort,
+			Name:         listener.Name,
+		}
 	}
 
 	// enable WAF for ports that are in the desired port set but not in the current port set
@@ -697,9 +703,10 @@ func (s *wafService) EnableListenerWafs(ctx context.Context, req *EnableListener
 				RegionCode:  req.RegionCode,
 			},
 			Port:         port,
-			Hosts:  wafMap[port],
+			Hosts:        wafMap[port].Hosts,
 			Enable:       true,
 			Mode:         mode,
+			ListenerName: wafMap[port].Name,
 		})
 		if err != nil {
 			return fmt.Errorf("failed to enable listener WAF: %v", err)

internal/store/es.go

@@ -26,20 +26,20 @@ type Config struct {
 
 func CreateEsClientFromEnv() (*es.Client, error) {
 	var opts []es.ClientOptionFunc
-	esURL := os.Getenv("ES_URL")
+	esURL := os.Getenv("ELASTIC_URL")
 	if esURL == "" {
 		return nil, fmt.Errorf("ES_URL is not set")
 	}
 	opts = append(opts, es.SetURL(esURL))
 
-	esUsername := os.Getenv("ES_USERNAME")
+	esUsername := os.Getenv("ELASTIC_USERNAME")
 
-	esPassword := os.Getenv("ES_PASSWORD")
+	esPassword := os.Getenv("ELASTIC_PASSWORD")
 	if esUsername != "" && esPassword != "" {
 		opts = append(opts, es.SetBasicAuth(esUsername, esPassword))
 	}
 
-	sniff := os.Getenv("ES_SNIFF")
+	sniff := os.Getenv("ELASTIC_SNIFF")
 	if sniff == "" {
 		sniff = "false"
 	}
@@ -84,7 +84,7 @@ func NewESStore(config Config, client *es.Client) *ESStore {
 // Init 初始化ES
 func (e *ESStore) Init() {
 	ctx := context.Background()
-	err := e.createIndex(ctx, "waf_detections_index", "test.json", "waf-detections-000001", "waf-detections")
+	err := e.createIndex(ctx, "waf_detections_index", "waf_detection_index_template.json", "waf-detections-000001", "waf-detections")
 	if err != nil {
 		log.Error().Err(err).Msg("create index test failed")
 	}

internal/utils/errors.go

+package utils
+
+import "errors"
+
+var (
+	ErrUnauthorized   = errors.New("unauthorized")
+	ErrInternalServer = errors.New("internal server error")
+)