Support for stream processing tools such as Kafka and NSQ

.github/goreleaser.yml

@@ -14,21 +14,21 @@ builds:
       - darwin
       - windows
     goarch:
-      - 386
+      - "386"
       - amd64
       - arm
       - arm64
     ignore:
       - goos: darwin
-        goarch: 386
+        goarch: "386"
       - goos: darwin
         goarch: arm
       - goos: windows
         goarch: arm
       - goos: windows
         goarch: arm64
       - goos: windows
-        goarch: 386
+        goarch: "386"
 checksum:
   name_template: '{{.ProjectName}}_{{.Version}}_checksums.txt'
 changelog:

.golangci.yml

@@ -1,3 +1,5 @@
+version: "2"
+
 # Options for analysis running.
 run:
   # The default concurrency value is the number of available CPU.
@@ -32,16 +34,16 @@ linters:
   # Default: false
   # fast: true
 
-  # Disable all linters.
-  disable-all: true
+  # Disable all linters for now.
+  default: none
 
   # Enable specific linter
   # https://golangci-lint.run/usage/linters/#enabled-by-default
   enable:
     - bodyclose
     - containedctx
     # - deadcode
-    - depguard
+    # - depguard
     - dogsled
     - dupl
     - dupword
@@ -50,23 +52,16 @@ linters:
     - errchkjson
     - errname
     - errorlint
-    - execinquery
     - exhaustive
-    - exportloopref
     - gochecknoinits
     - gocritic
     - godot
     # - godox
-    - gofmt
-    # - gofumpt
-    - goimports
-    # - golint
     # - gomnd
     - gomoddirectives
     - gomodguard
     - goprintffuncname
     - gosec
-    - gosimple
     - govet
     - grouper
     - importas
@@ -91,9 +86,6 @@ linters:
     - reassign
     - revive
     - staticcheck
-    - stylecheck
-    - tenv
-    - typecheck
     - unconvert
     - unparam
     - unused
@@ -102,43 +94,51 @@ linters:
     - wastedassign
     - whitespace
     - wrapcheck
-    - wsl
+    # - wsl_v5
+
+  exclusions:
+    rules:
+      # Exclude some linters from running on tests files.
+      - path: _test\.go
+        linters:
+          - nlreturn
 
-linters-settings:
-  nlreturn:
-    # Size of the block (including return statement that is still "OK") so no return split required.
-    # Default: 1
-    block-size: 2
-  wsl:
-    enforce-err-cuddling: true
-    allow-cuddle-declarations: true
-    allow-assign-and-call: true
-    allow-cuddle-with-calls:
-      - log.Println
-      - log.Printf
-      - RLock
-      - RUnlock
-      - Lock
-      - Unlock
-    allow-assign-and-anything: true
-  gofumpt:
-    extra-rules: true
-  lll:
-    line-length: 160
-  varnamelen:
-    ignore-decls:
-      - w io.Writer
-      - w io.WriteCloser
-      - w http.ResponseWriter
-      - r *http.Request
-      - r chi.Router
-      - r *chi.Mux
-      - i int
+  settings:
+    nlreturn:
+      # Size of the block (including return statement that is still "OK") so no return split required.
+      # Default: 1
+      block-size: 2
+    wsl:
+      force-err-cuddling: true
+      allow-cuddle-declarations: true
+      allow-assign-and-call: true
+      allow-cuddle-with-calls:
+        - log.Println
+        - log.Printf
+        - RLock
+        - RUnlock
+        - Lock
+        - Unlock
+      allow-assign-and-anything: true
+    lll:
+      line-length: 160
+    varnamelen:
+      ignore-decls:
+        - w io.Writer
+        - w io.WriteCloser
+        - w http.ResponseWriter
+        - r *http.Request
+        - r chi.Router
+        - r *chi.Mux
+        - i int
+
+formatters:
+  enable:
+    # - gofmt
+    # - gofumpt
+    - goimports
+    # - golint
 
-issues:
-  # List of regexps of issue texts to exclude.
-  exclude-rules:
-    # Exclude some linters from running on tests files.
-    - path: _test\.go
-      linters:
-        - nlreturn
+  settings:
+    gofumpt:
+      extra-rules: true

cmd/certpicker/main.go

@@ -46,7 +46,9 @@ func main() {
 	}
 
 	// Get entries from CT log
-	c, _ := context.WithTimeout(context.Background(), 10*time.Second)
+	c, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
 	entries, getEntriesErr := jsonClient.GetRawEntries(c, certID, certID)
 	if getEntriesErr != nil {
 		log.Fatalln("Error getting entries from CT log: ", getEntriesErr)

config.sample.yaml

@@ -11,6 +11,11 @@ webserver:
   cert_path: ""
   cert_key_path: ""
   compression_enabled: false
+  # Use True-Client-IP, X-Real-IP or the X-Forwarded-For headers (in that order) to determine the real IP address of the client.
+  # If you are using a reverse proxy, you should set this to true.
+  real_ip: false
+  whitelist:
+    - "127.0.0.1/8"
 
 prometheus:
   enabled: true
@@ -22,6 +27,20 @@ prometheus:
   whitelist:
     - "127.0.0.1/8"
 
+# Configuration related to external stream processing tools go here.
+stream_processing:
+  - name: "kafka"
+    enabled: false
+    server_addr: "127.0.0.1"
+    server_port: 9092
+    topic: "certstream"
+
+  - name: "nqs"
+    enabled: true
+    server_addr: "127.0.0.1"
+    server_port: 9092
+    topic: "certstream"
+
 general:
   # DisableDefaultLogs indicates whether the default logs used in Google Chrome and provided by Google should be disabled.
   disable_default_logs: false
@@ -39,9 +58,8 @@ general:
     websocket: 300
     # Buffer for each CT log connection
     ctlog: 1000
-    # Combined buffer for the broadcast manager
-    broadcastmanager: 10000
-
+    # Combined buffer for the cert dispatcher
+    dispatcher: 10000
   # Google regularly updates the log list. If this option is set to true, the server will remove all logs no longer listed in the Google log list.
   # This option defaults to true. See https://github.com/d-Rickyy-b/certstream-server-go/issues/51
   drop_old_logs: true

docker/docker-compose.metrics.yml

@@ -1,6 +1,6 @@
 version: '2'
 
-# Make sure to create the sub directories "prometheus", "prometheus_data", "grafana", "grafana_data" and "certstream"
+# Make sure to create the subdirectories "prometheus", "prometheus_data", "grafana", "grafana_data" and "certstream"
 # and create the config files for all three services. For further details please refer to https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics
 
 networks:
@@ -30,7 +30,7 @@ services:
     ports:
       # Exposing Prometheus is NOT required, if you don't want to access it from outside the Docker network.
       # Using localhost enables you to use a reverse proxy (e.g. with basic auth) to access Prometheus in a more secure way.
-      - 127.0.0.1:9090:9090
+      - "127.0.0.1:9090:9090"
     networks:
       - monitoring
     extra_hosts:
@@ -44,7 +44,7 @@ services:
     depends_on:
       - prometheus
     ports:
-      - 127.0.0.1:8082:3000
+      - "127.0.0.1:8082:3000"
     volumes:
       - ./grafana_data:/var/lib/grafana
       - ./grafana/provisioning/:/etc/grafana/provisioning/
@@ -60,7 +60,7 @@ services:
     # Configure the service to run as specific user.
     # user: "1000:1000"
     ports:
-      - 127.0.0.1:8080:80
+      - "127.0.0.1:8080:80"
       # Don't forget to open the other port in case you run the Prometheus endpoint on another port than the websocket server.
       # - 127.0.0.1:8081:81
     volumes:

docker/docker-compose.yml

@@ -7,7 +7,7 @@ services:
     # Configure the service to run as specific user
     # user: "1000:1000"
     ports:
-      - 127.0.0.1:8080:80
+      - "127.0.0.1:8080:80"
       # Don't forget to open the other port in case you run the Prometheus endpoint on another port than the websocket server.
       # - 127.0.0.1:8081:81
     volumes:

go.mod

@@ -5,23 +5,27 @@ go 1.24.0
 toolchain go1.24.2
 
 require (
-	github.com/VictoriaMetrics/metrics v1.40.1
+	github.com/VictoriaMetrics/metrics v1.40.2
 	github.com/go-chi/chi/v5 v5.2.3
 	github.com/google/certificate-transparency-go v1.3.2
 	github.com/gorilla/websocket v1.5.3
+	github.com/nsqio/go-nsq v1.1.0
+	github.com/segmentio/kafka-go v0.4.49
 	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
 	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/golang/snappy v1.0.0 // indirect
 	github.com/google/trillian v1.7.2 // indirect
+	github.com/klauspost/compress v1.18.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/valyala/fastrand v1.1.0 // indirect
 	github.com/valyala/histogram v1.2.0 // indirect
-	golang.org/x/crypto v0.42.0 // indirect
-	golang.org/x/net v0.44.0 // indirect
-	golang.org/x/sys v0.36.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250922171735-9219d122eba9 // indirect
-	google.golang.org/grpc v1.75.1 // indirect
-	google.golang.org/protobuf v1.36.9 // indirect
+	golang.org/x/crypto v0.45.0 // indirect
+	golang.org/x/sys v0.38.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect
+	google.golang.org/grpc v1.77.0 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 )

go.sum

@@ -1,11 +1,16 @@
-github.com/VictoriaMetrics/metrics v1.40.1 h1:FrF5uJRpIVj9fayWcn8xgiI+FYsKGMslzPuOXjdeyR4=
-github.com/VictoriaMetrics/metrics v1.40.1/go.mod h1:XE4uudAAIRaJE614Tl5HMrtoEU6+GDZO4QTnNSsZRuA=
+github.com/VictoriaMetrics/metrics v1.40.2 h1:OVSjKcQEx6JAwGeu8/KQm9Su5qJ72TMEW4xYn5vw3Ac=
+github.com/VictoriaMetrics/metrics v1.40.2/go.mod h1:XE4uudAAIRaJE614Tl5HMrtoEU6+GDZO4QTnNSsZRuA=
+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/go-chi/chi/v5 v5.2.3 h1:WQIt9uxdsAbgIYgid+BpYc+liqQZGMHRaUwp0JUcvdE=
 github.com/go-chi/chi/v5 v5.2.3/go.mod h1:L2yAIGWB3H+phAw1NxKwWM+7eUH/lU8pOMm5hHcoops=
 github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
 github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs=
+github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/certificate-transparency-go v1.3.2 h1:9ahSNZF2o7SYMaKaXhAumVEzXB2QaayzII9C8rv7v+A=
 github.com/google/certificate-transparency-go v1.3.2/go.mod h1:H5FpMUaGa5Ab2+KCYsxg6sELw3Flkl7pGZzWdBoYLXs=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
@@ -14,28 +19,46 @@ github.com/google/trillian v1.7.2 h1:EPBxc4YWY4Ak8tcuhyFleY+zYlbCDCa4Sn24e1Ka8Js
 github.com/google/trillian v1.7.2/go.mod h1:mfQJW4qRH6/ilABtPYNBerVJAJ/upxHLX81zxNQw05s=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co=
+github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/nsqio/go-nsq v1.1.0 h1:PQg+xxiUjA7V+TLdXw7nVrJ5Jbl3sN86EhGCQj4+FYE=
+github.com/nsqio/go-nsq v1.1.0/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQTuDEY=
+github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
+github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+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/segmentio/kafka-go v0.4.49 h1:GJiNX1d/g+kG6ljyJEoi9++PUMdXGAxb7JGPiDCuNmk=
+github.com/segmentio/kafka-go v0.4.49/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
 github.com/sergi/go-diff v1.3.1 h1:xkr+Oxo4BOQKmkn/B9eMK0g5Kg/983T9DqqPHwYqD+8=
 github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NFbPK1I=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
 github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
 github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ=
 github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY=
-golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI=
-golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8=
-golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I=
-golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY=
-golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k=
-golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
-golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
-golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250922171735-9219d122eba9 h1:V1jCN2HBa8sySkR5vLcCSqJSTMv093Rw9EJefhQGP7M=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250922171735-9219d122eba9/go.mod h1:HSkG/KdJWusxU1F6CNrwNDjBMgisKxGnc5dAZfT0mjQ=
-google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI=
-google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
-google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
-google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+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=
+golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
+golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
+golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
+golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
+golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
+golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM=
+golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba h1:UKgtfRM7Yh93Sya0Fo8ZzhDP4qBckrrxEr2oF5UIVb8=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
+google.golang.org/grpc v1.77.0 h1:wVVY6/8cGA6vvffn+wWK5ToddbgdU3d8MNENr4evgXM=
+google.golang.org/grpc v1.77.0/go.mod h1:z0BY1iVj0q8E1uSQCjL9cppRj+gnZjzDnzV0dHhrNig=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=