diff --git a/CHANGELOG.md b/CHANGELOG.md index 5232c3e..7c7ffd9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,47 +6,63 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] + ### Added -- Option to configure trusted_proxies for which the X-Forwarded-For and X-Real-IP headers are used as client IP (b062edca) + +### Changed + +### Removed + +### Fixed + +## [1.10.0] - 2026-07-xx + +### Added + +- Support for stream processing tools like kafka or nsq +- Option to configure trusted_proxies for which the X-Forwarded-For and X-Real-IP headers are used as client IP (b062edc) +- Add `excluded_logs` config option to skip specific logs and operators. Works both for classic and tiled logs. (170fbeb) - Added multiple unit tests for ensuring correct functionality -- Add `excluded_logs` config option to skip specific logs and operators. Works both for classic and tiled logs. ### Changed -- Updated weak cipher suites to stronger ones (939517cd) + +- Updated weak cipher suites to stronger ones (939517c) - Updated http client settings to prevent timeouts and other connectivity issues - Updated http server settings to allow for higher delays -- Minor code improvements and refactoring, mostly style related - Updated batch size from 100 to 256 (#97) -### Removed - ### Fixed -- Calculate correct starting tile index for static ct logs -- Use proper websocket close code (1008) instead of 1005, which wasn't sent to the client -- Respect ct_index path in config file for the `create-index` command -- Ensure all additionalLog entries are processed correctly -### Docs +- Prevent excessive tile fetching for static CT logs (#104) +- Calculate correct starting tile index for static ct logs (a2b4a9f) +- Use proper websocket close code (1008) instead of 1005, which wasn't sent to the client (76ff240) +- Respect ct_index path in config file for the `create-index` command (67838a9) +- Ensure all additionalLog entries are processed correctly (c8bd735) ## [1.9.0] - 2026-04-03 + ### Added + - Ability to store and resume processing of certs from where it left off after a restart - see sample config "recovery" (#49) - New CLI switch for creating an index file from a CT log (#49) -- Support for [Static CT](https://github.com/C2SP/C2SP/blob/main/static-ct-api.md) logs +- Support for [Static CT](https://github.com/C2SP/C2SP/blob/main/static-ct-api.md) logs - Check for retired CT logs and prevent them from being watched / stop watching them (#77) - Accept websocket connections from all origins - Option to disable the default logs provided by Google - see sample config "disable_default_logs" - Use of cobra for CLI argument parsing. New commands for displaying version and creating an index file -- Override confiuration options via environment variables (e.g. `CERTSTREAM_WEBSERVER_LISTEN_PORT=1234` to change the listen port) +- Override configuration options via environment variables (e.g. `CERTSTREAM_WEBSERVER_LISTEN_PORT=1234` to change the listen port) - New field in json `source.timestamp` representing the timestamp of when the cert was processed by the CT log (#98) ### Changed + - **Breaking:** The configuration file for the docker container is now read from the /app/config/ directory (b9e5e6) ### Removed + - Deleted non-functional Dodo log from sample config (#78) ### Fixed + - Properly remove stopped ct log workers (#74) - Added missing fields certificatePolicies and ctlPoisonByte (#85) - Prevent race condition caused by simultaneous rw access to logmetrics (#91) @@ -54,140 +70,188 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Properly add new metrics for all newly found logs (#96) ## [1.8.2] - 2025-11-22 + ### Fixed + - Added missing fields certificatePolicies and ctlPoisonByte (#85) ## [1.8.1] - 2025-05-04 + ### Fixed + - No longer reject URLs with trailing slashes defined in the `additional_logs` config (#62) - When using `drop_old_logs` in the config, the server won't remove logs defined in `additional_logs` anymore (#64) ## [1.8.0] - 2025-05-03 + ### Security + - Close several CVEs in x/crypto and x/net dependencies (#59) ### Added + - New CLI tool for fetching certificates from a CT log (#47) - Ability to add custom CT logs to the config (#56) - Remove old CT logs as soon as they are removed from the Google CT Loglist (#60) - New configuration for buffer sizes (#58) ### Fixed + - Properly handle IPv6 addresses in config (#61) ## [1.7.1] - 2025-05-03 + ### Fixed + - Properly handle IPv6 addresses in config (#61) ## [1.7.0] - 2024-08-20 + ### Added + - Support for websocket compression - disabled by default (#40) - Support for non-browsers by implementing server initiated heartbeats (#39) - Start new ct-watchers as new ct logs become available (#42) - More logging to document currently watched logs (03d878e) - + ### Changed + - Changed log output to be better grepable (5c055cc) - Update ct log update interval to once per hour instead of once per 6 hours as previously (9b6e77d) ### Fixed + - Fixed a possible race condition when accessing metrics ## [1.6.0] - 2024-03-05 + ### Added + - New metric for skipped certs per client (#34) ## [1.5.2] - 2024-02-17 + ### Fixed + - Fixed an issue with ip whitelists for the websocket server (#33) ## [1.5.1] - 2024-01-18 + ### Fixed -- Fixed a rare issue where it was possible for the all_domains json property (or data property in case of the domains-only endpoint) to be null + +- Fixed a rare issue where it was possible for the all_domains json property (or data property in case of the domains-only endpoint) to be null ## [1.5.0] - 2023-12-21 + ### Added + - New `-version` switch to print version and exit afterwards - Print version on every run of the tool - Count and log number of skipped certificates per client ### Changed + - Update to chi/v5 - Update ct-watcher timeout from 5 to 30 seconds ### Fixed + - Prevent invalid subscription types to be used - Kill connection after broadcasthandler was stopped ## [1.4.0] - 2023-11-29 + ### Added + - Config option to use X-Forwarded-For or X-Real-IP header as client IP - Config option to whitelist client IPs for both websocket and metrics endpoints - Config option to enable system metrics (cpu, memory, etc.) ## [1.3.2] - 2023-11-28 + ### Fixed + - Memory leak related to clients disconnecting from the websocket not being handled properly ## [1.3.1] - 2023-09-18 + ### Changed + - Updated config.sample.yaml to run both certstream and prometheus metrics on same socket ### Docs + - Fixed wrong docker command in readme ## [1.3.0] - 2023-04-11 + ### Added + - Calculate and display Sha256 sum of certificate ### Changed + - Update dependencies - Better logging for CT log errors ### Fixed + - End execution after all workers stopped - Implement timeout for the http client - Keep ct watcher from crashing upon a connection reset from server ## [1.2.2] - 2023-01-10 + ### Added + - Two docker-compose files - Check for presence of .yml or .yaml files in the current directory ### Fixed + - Handle sudden disconnects of CT logs ### Docs -- Added [wiki entry for docker-compose](https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics) + +- Added [wiki entry for docker-compose](https://github.com/d-Rickyy-b/certstream-server-go/wiki/Collecting-and-Visualizing-Metrics) ## [1.2.1] - 2022-12-16 + ### Changed + - Updated ci pipeline to use new setup-go and checkout actions - Use correct package name `github.com/d-Rickyy-b/certstream-server-go` ## [1.2.0] - 2022-12-15 + ### Added + - Log x-Forwarded-For header for requests - More logging for certain error situations - Add operator to ct log cert count metrics ### Changed + - Updated certificate-transparency-go dependency to v1.1.4 - Code improvements, adhering to styleguide - Rename module to certstream-server-go - Use log_list.json instead of all_logs_list.json ## [1.1.0] - 2022-10-19 + Fix for missing loglist urls. ### Fixed + Fixed the connection issue due to the offline Google loglist urls. ## [1.0.0] - 2022-08-08 + Initial release! First stable version of certstream-server-go is published as v1.0.0 -[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.9.0...HEAD -[1.8.2]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.2...v1.9.0 +[unreleased]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.10.0...HEAD +[1.10.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.9.0...v1.10.0 +[1.9.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.2...v1.9.0 [1.8.2]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.1...v1.8.2 [1.8.1]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.8.0...v1.8.1 [1.8.0]: https://github.com/d-Rickyy-b/certstream-server-go/compare/v1.7.1...v1.8.0 diff --git a/cmd/certpicker/main.go b/cmd/certpicker/main.go index cc82bf5..7323fad 100644 --- a/cmd/certpicker/main.go +++ b/cmd/certpicker/main.go @@ -65,7 +65,7 @@ func main() { log.Fatalln("Error getting entry from CT log: ", getEntryErr) } - // Loop over entries and pars each one. + // Loop over entries and parse each one. for _, leafEntry := range entries.Entries { rawLogEntry, err := ct.RawLogEntryFromLeaf(certID, &leafEntry) if err != nil { diff --git a/config.sample.yaml b/config.sample.yaml index 74fe9a6..bed04c4 100644 --- a/config.sample.yaml +++ b/config.sample.yaml @@ -37,6 +37,35 @@ prometheus: whitelist: - "127.0.0.1/8" +# Configuration related to external stream processing tools go here. +stream_processing: + # List of stream processing configurations. + # Each configuration specifies a stream processing tool to use and its settings. + - name: "kafka-prod" + # Type describes the type of stream processing tool to use. + # Supported types: "kafka", "nqs" + type: "kafka" + # The stream type can be any of "full", "lite", "domains-only" - same as the certstream websocket urls + # Default is "full" + stream: "full" + # Enable or disable this stream processing configuration. + enabled: true + # The server address and port to connect to the stream processing tool + server_addr: "127.0.0.1" + server_port: 9092 + topic: "certstream" + # Available compression types depend on the stream processing tool. + # E.g. for Kafka, supported types are "none" / "", "gzip", "snappy", "lz4", "zstd" + compression: "" + + - name: "nqs-prod" + type: "nqs" + stream: "lite" + enabled: false + 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 @@ -66,9 +95,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 diff --git a/docker/docker-compose.metrics.yml b/docker/docker-compose.metrics.yml index 316a8ec..9c17a7f 100644 --- a/docker/docker-compose.metrics.yml +++ b/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: diff --git a/go.mod b/go.mod index 162d32c..1757d6f 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,8 @@ require ( github.com/google/certificate-transparency-go v1.3.3 github.com/google/trillian v1.7.3 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 + github.com/nsqio/go-nsq v1.1.0 + github.com/segmentio/kafka-go v0.4.51 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 golang.org/x/crypto v0.49.0 @@ -17,8 +19,11 @@ require ( github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect + github.com/golang/snappy v0.0.1 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/klauspost/compress v1.15.9 // indirect github.com/pelletier/go-toml/v2 v2.3.0 // indirect + github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect diff --git a/go.sum b/go.sum index 6fb7886..7b2c8cd 100644 --- a/go.sum +++ b/go.sum @@ -15,6 +15,8 @@ github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPE github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= 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 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/certificate-transparency-go v1.3.3 h1:hq/rSxztSkXN2tx/3jQqF6Xc0O565UQPdHrOWvZwybo= github.com/google/certificate-transparency-go v1.3.3/go.mod h1:iR17ZgSaXRzSa5qvjFl8TnVD5h8ky2JMVio+dzoKMgA= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -25,14 +27,20 @@ github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5T github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY= +github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= 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/pelletier/go-toml/v2 v2.3.0 h1:k59bC/lIZREW0/iVaQR8nDHxVq8OVlIzYCOJf421CaM= github.com/pelletier/go-toml/v2 v2.3.0/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= +github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0= +github.com/pierrec/lz4/v4 v4.1.15/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/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= @@ -40,6 +48,8 @@ github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/f github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4= github.com/sagikazarmark/locafero v0.12.0/go.mod h1:sZh36u/YSZ918v0Io+U9ogLYQJ9tLLBmM4eneO6WwsI= +github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef71Hno= +github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E= github.com/sergi/go-diff v1.4.0 h1:n/SP9D5ad1fORl+llWyN+D6qoUETXNZARKjyY2/KVCw= github.com/sergi/go-diff v1.4.0/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= @@ -61,6 +71,12 @@ github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G 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= +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= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= diff --git a/internal/backoff/backoff.go b/internal/backoff/backoff.go new file mode 100644 index 0000000..afbcb26 --- /dev/null +++ b/internal/backoff/backoff.go @@ -0,0 +1,32 @@ +package backoff + +import ( + "sync" + "time" +) + +// NewBackoff returns a function that implements exponential backoff for the given reset window. +func NewBackoff(resetWindow time.Duration) func(target func()) { + mu := sync.Mutex{} + var errorCount int + var errorLastTime time.Time + errorLogResetWindow := resetWindow + + return func(target func()) { + mu.Lock() + defer mu.Unlock() + now := time.Now() + + // Reset counter if enough time has passed since the last error + if now.Sub(errorLastTime) > errorLogResetWindow { + errorCount = 0 + } + + errorCount++ + errorLastTime = now + + if errorCount == 1 || (errorCount&(errorCount-1)) == 0 { + target() + } + } +} diff --git a/internal/backoff/backoff_test.go b/internal/backoff/backoff_test.go new file mode 100644 index 0000000..1db85e7 --- /dev/null +++ b/internal/backoff/backoff_test.go @@ -0,0 +1,18 @@ +package backoff + +import ( + "testing" + "time" +) + +func TestErrorBackoff(t *testing.T) { + backoffHandler := NewBackoff(1 * time.Second) + + for i := range 20 { + backoffHandler(func() { + test := i + 1 + t.Log("Hello World", test) + time.Sleep(1100 * time.Millisecond) + }) + } +} diff --git a/internal/broadcast/baseclient.go b/internal/broadcast/baseclient.go new file mode 100644 index 0000000..9b69e3e --- /dev/null +++ b/internal/broadcast/baseclient.go @@ -0,0 +1,48 @@ +package broadcast + +import "log" + +// BaseClient defines the basic structure for a client that can receive broadcast messages. +// Other client types can embed this struct to inherit its functionality. +type BaseClient struct { + broadcastChan chan []byte + stopChan chan struct{} + name string + subType SubscriptionType + skippedCerts uint64 +} + +// Close cleans up the client's resources by closing the stop and broadcast channels. +func (c *BaseClient) Close() { + close(c.stopChan) + close(c.broadcastChan) +} + +// Name returns the name of the client. +func (c *BaseClient) Name() string { + return c.name +} + +// SubType returns the subscription type of the client. +func (c *BaseClient) SubType() SubscriptionType { + return c.subType +} + +// SkippedCerts returns the number of certificates that were skipped due to the client's broadcast channel being full. +func (c *BaseClient) SkippedCerts() uint64 { + return c.skippedCerts +} + +// Write sends a message to the client's broadcast channel. +// If the channel is full, it increments the skippedCerts counter and logs a message. +func (c *BaseClient) Write(data []byte) { + select { + case c.broadcastChan <- data: + default: + // Default case is executed if the client's broadcast channel is full. + c.skippedCerts++ + if c.skippedCerts%1000 == 1 { + log.Printf("Not providing client '%s' with cert because client's buffer is full. The client can't keep up. Skipped certs: %d\n", c.name, c.skippedCerts) + } + } +} diff --git a/internal/broadcast/certprocessor.go b/internal/broadcast/certprocessor.go new file mode 100644 index 0000000..d10d47a --- /dev/null +++ b/internal/broadcast/certprocessor.go @@ -0,0 +1,26 @@ +package broadcast + +// CertProcessor defines the interface for processing certificate messages. +// Different implementations can be used for different types of clients, such as WebSocket clients, kafka, or other message queues for example. +type CertProcessor interface { + // Write sends a message to the client's broadcast channel. + Write(message []byte) + // Close closes the client's connection and cleans up resources. + Close() + + // To obtain client details, there are getter methods for several values + Name() string + SubType() SubscriptionType + SkippedCerts() uint64 +} + +const ( + // SubTypeFull represents full certificate updates. + SubTypeFull SubscriptionType = iota + // SubTypeLite represents certificate updates with less details. + SubTypeLite + // SubTypeDomain represents updates that only include domain information. + SubTypeDomain +) + +type SubscriptionType int diff --git a/internal/broadcast/dispatcher.go b/internal/broadcast/dispatcher.go new file mode 100644 index 0000000..b1a7fb6 --- /dev/null +++ b/internal/broadcast/dispatcher.go @@ -0,0 +1,150 @@ +package broadcast + +import ( + "log" + "sync" + + "github.com/d-Rickyy-b/certstream-server-go/internal/config" + "github.com/d-Rickyy-b/certstream-server-go/internal/metrics" + "github.com/d-Rickyy-b/certstream-server-go/internal/models" +) + +// ClientHandler dispatches certificate entries to registered clients. +var ClientHandler *Dispatcher + +type Dispatcher struct { + MessageQueue chan models.Entry + clients []CertProcessor + clientLock sync.RWMutex +} + +// NewDispatcher creates a new Dispatcher instance and assigns it to the ClientHandler variable. +func NewDispatcher() *Dispatcher { + d := &Dispatcher{} + d.MessageQueue = make(chan models.Entry, config.AppConfig.General.BufferSizes.Dispatcher) + ClientHandler = d + + metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"full\"}", d.ClientFullCount) + metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"lite\"}", d.ClientLiteCount) + metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"domain\"}", d.ClientDomainsCount) + return d +} + +// RegisterClient adds a client to the list of clients of the Dispatcher. +// The client will receive certificate broadcasts right after registration. +func (bm *Dispatcher) RegisterClient(c CertProcessor) { + // TODO: check if the client is already registered + bm.clientLock.Lock() + bm.clients = append(bm.clients, c) + log.Printf("Added new client. Clients: %d, Capacity: %d\n", len(bm.clients), cap(bm.clients)) + metrics.Prometheus.RegisterClient(c.Name(), func() float64 { return float64(c.SkippedCerts()) }) + bm.clientLock.Unlock() +} + +// UnregisterClient removes a client from the list of clients of the Dispatcher. +// The client will no longer receive certificate broadcasts right after unregistering. +func (bm *Dispatcher) UnregisterClient(clientName string) { + bm.clientLock.Lock() + + for i, client := range bm.clients { + if clientName == client.Name() { + // Copy the last element of the slice to the position of the removed element + // Then remove the last element by re-slicing + bm.clients[i] = bm.clients[len(bm.clients)-1] + bm.clients[len(bm.clients)-1] = nil + bm.clients = bm.clients[:len(bm.clients)-1] + + // Close the broadcast channel of the client, otherwise this leads to a memory leak + client.Close() + + metrics.Prometheus.UnregisterClient(client.Name()) + + break + } + } + + bm.clientLock.Unlock() +} + +// ClientFullCount returns the current number of clients connected to the service on the `full` endpoint. +func (bm *Dispatcher) ClientFullCount() (count int64) { + return bm.clientCountByType(SubTypeFull) +} + +// ClientLiteCount returns the current number of clients connected to the service on the `lite` endpoint. +func (bm *Dispatcher) ClientLiteCount() (count int64) { + return bm.clientCountByType(SubTypeLite) +} + +// ClientDomainsCount returns the current number of clients connected to the service on the `domains-only` endpoint. +func (bm *Dispatcher) ClientDomainsCount() (count int64) { + return bm.clientCountByType(SubTypeDomain) +} + +// clientCountByType returns the current number of clients connected to the service on the endpoint matching +// the specified SubscriptionType. +func (bm *Dispatcher) clientCountByType(subType SubscriptionType) (count int64) { + bm.clientLock.RLock() + defer bm.clientLock.RUnlock() + + for _, c := range bm.clients { + if c.SubType() == subType { + count++ + } + } + + return count +} + +// GetSkippedCerts returns a map of client names to the number of skipped certificates for each client. +func (bm *Dispatcher) GetSkippedCerts() map[string]uint64 { + bm.clientLock.RLock() + defer bm.clientLock.RUnlock() + + skippedCerts := make(map[string]uint64, len(bm.clients)) + for _, c := range bm.clients { + skippedCerts[c.Name()] = c.SkippedCerts() + } + + return skippedCerts +} + +// broadcaster is run in a goroutine and handles the dispatching of certs to clients. +func (bm *Dispatcher) broadcaster() { + for { + var data []byte + + // Take entry out of broadcast channel and generate JSON representations for the entry. + entry := <-bm.MessageQueue + dataLite := entry.JSONLite() + dataFull := entry.JSON() + dataDomain := entry.JSONDomains() + + bm.clientLock.RLock() + + for _, c := range bm.clients { + switch c.SubType() { + case SubTypeLite: + data = dataLite + case SubTypeFull: + data = dataFull + case SubTypeDomain: + data = dataDomain + default: + // This should never happen, but if it does, we log it and skip the client. + log.Printf("Unknown subscription type '%d' for client '%s'. Skipping this client!\n", c.SubType(), c.Name()) + continue + } + + c.Write(data) + } + + bm.clientLock.RUnlock() + } +} + +// Start starts the broadcaster goroutine. +func (bm *Dispatcher) Start() { + go bm.broadcaster() + log.Println("Dispatcher started. Listening for certificate entries...") +} diff --git a/internal/broadcast/kafkaclient.go b/internal/broadcast/kafkaclient.go new file mode 100644 index 0000000..e0e7ada --- /dev/null +++ b/internal/broadcast/kafkaclient.go @@ -0,0 +1,221 @@ +package broadcast + +import ( + "context" + "errors" + "log" + "net" + "time" + + "github.com/d-Rickyy-b/certstream-server-go/internal/backoff" + "github.com/segmentio/kafka-go" +) + +var ( + kafkaMaxBatchSize = 50 + kafkaMaxBatchWait = 1 * time.Second + kafkaConnTimeout = 5 * time.Second +) + +// KafkaClient connects to a Kafka server in order to provide it with certificates. +type KafkaClient struct { + conn *kafka.Conn // Kafka connection + addr string + topic string + compression kafka.Compression + isConnected bool + BaseClient +} + +// NewKafkaClient creates a new Kafka client that immediately connects to the configured Kafka server. +func NewKafkaClient(subType SubscriptionType, addr, name, topic, compression string, certBufferSize int) *KafkaClient { + // Connect to the Kafka server + ctx, cancel := context.WithTimeout(context.Background(), kafkaConnTimeout) + defer cancel() + conn, err := kafka.DialLeader(ctx, "tcp", addr, topic, 0) + if err != nil { + log.Println("failed to connect to kafka:", err) + } + // TODO implement explicit topic creation + + kc := &KafkaClient{ + conn: conn, + addr: addr, + topic: topic, + BaseClient: BaseClient{ + broadcastChan: make(chan []byte, certBufferSize), + stopChan: make(chan struct{}), + name: name, + subType: subType, + }, + } + + var kafkaCompression kafka.Compression + switch compression { + case "gzip": + kafkaCompression = kafka.Gzip + case "snappy": + kafkaCompression = kafka.Snappy + case "lz4": + kafkaCompression = kafka.Lz4 + case "none", "": + default: + log.Println("invalid compression type:", compression) + } + + kc.compression = kafkaCompression + + go kc.broadcastHandler() + go kc.reconnectHandler() + + return kc +} + +// reconnectHandler is a background job that attempts to reconnect to the Kafka server if the connection is lost. +func (c *KafkaClient) reconnectHandler() { + for { + select { + case <-c.stopChan: + log.Println("Stopping reconnectHandler for kafka producer:", c.addr) + c.conn.Close() + + return + default: + if c.isConnected { + // If already connected or no connection exists, skip reconnection + time.Sleep(5 * time.Second) + continue + } + + // Attempt to connect to the Kafka server + ctx, cancel := context.WithTimeout(context.Background(), kafkaConnTimeout) + defer cancel() + conn, err := kafka.DialLeader(ctx, "tcp", c.addr, c.topic, 0) + if err != nil { + log.Printf("Reconnect failed: %v. Retrying in 5s...", err) + time.Sleep(5 * time.Second) + + continue + } + // Close old connection if exists + if c.conn != nil { + _ = c.conn.Close() + } + + c.conn = conn + c.isConnected = true + log.Println("Reconnected to Kafka at", c.addr) + } + } +} + +// Each client has a broadcastHandler that runs in the background and sends out the broadcast messages to the client. +func (c *KafkaClient) broadcastHandler() { + defer func() { + log.Println("Closing broadcast handler for kafka producer:", c.addr) + if c.conn != nil { + if err := c.conn.Close(); err != nil { + log.Println("failed to close conn:", err) + return + } + } + + ClientHandler.UnregisterClient(c.name) + }() + + backoffHandler := backoff.NewBackoff(60 * time.Second) + batch := make([]kafka.Message, 0, kafkaMaxBatchSize) + t := time.NewTimer(kafkaMaxBatchWait) + + for { + select { + case <-c.stopChan: + return + case message, ok := <-c.broadcastChan: + if !ok { + log.Println("broadcastChan closed for kafkaClient:", c.addr) + return + } + + // Drop messages if not connected + if !c.isConnected { + time.Sleep(5 * time.Second) + continue + } + + msg := kafka.Message{Value: message} + batch = append(batch, msg) + + // Write batch if it reaches max size + if len(batch) >= kafkaMaxBatchSize { + err := c.writeBatch(batch) + if err != nil { + // Without using a backoff strategy, the errors would massively spam the log + backoffHandler(func() { + var netErr *net.OpError + if errors.As(err, &netErr) { + c.isConnected = false + } + + log.Printf("Error writing messages to kafka: %v", err) + }) + } + + batch = batch[:0] + t.Reset(kafkaMaxBatchWait) + } + case <-t.C: + // If batch size has not reached kafkaMaxBatchSize, write the batch after kafkaMaxBatchWait + if len(batch) == 0 { + continue + } + + err := c.writeBatch(batch) + if err != nil { + // Without using a backoff strategy, the errors would massively spam the log + backoffHandler(func() { + var netErr *net.OpError + if errors.As(err, &netErr) { + c.isConnected = false + } + + log.Printf("Error writing messages to kafka: %v", err) + }) + } + + batch = batch[:0] + } + } +} + +// writeBatch writes a batch of messages to Kafka, handling exponential backoff and connection state +func (c *KafkaClient) writeBatch(batch []kafka.Message) error { + if len(batch) == 0 { + return nil + } + + if c.conn == nil || !c.isConnected { + return errors.New("no connection to kafka") + } + + _ = c.conn.SetWriteDeadline(time.Now().Add(kafkaConnTimeout)) + + _, err := c.conn.WriteCompressedMessages( + c.compression.Codec(), + batch..., + ) + if err != nil { + // Treat kafka errors specially + var kafkaErr kafka.Error + if errors.As(err, &kafkaErr) { + if errors.Is(kafkaErr, kafka.MessageSizeTooLarge) { + log.Printf("Message size is too large for kafka broker '%s' - reducing batch size to %d", c.addr, kafkaMaxBatchSize/2) + kafkaMaxBatchSize = kafkaMaxBatchSize / 2 + // TODO: currently there is no retry mechanism implemented. We should try to resend the current batch with the reduced batch size. + } + } + + return err + } + return nil +} diff --git a/internal/broadcast/nsqclient.go b/internal/broadcast/nsqclient.go new file mode 100644 index 0000000..f8e042e --- /dev/null +++ b/internal/broadcast/nsqclient.go @@ -0,0 +1,173 @@ +package broadcast + +import ( + "log" + "time" + + "github.com/nsqio/go-nsq" +) + +const ( + nsqMaxBatchSize = 50 + nsqMaxBatchWait = 1 * time.Second +) + +// NSQClient connects to a NSQ server in order to provide it with certificates. +type NSQClient struct { + conn *nsq.Producer // nsq connection + addr string + topic string + isConnected bool + BaseClient +} + +// nullLogger is a logger that does not output anything. +// It is used to silence the NSQ logger. +type nullLogger struct{} + +func (l nullLogger) Output(calldepth int, s string) error { + // Do nothing, effectively silencing the logger + return nil +} + +// NewNSQClient creates a new NSQ client that immediately connects to the configured NSQ server +func NewNSQClient(subType SubscriptionType, addr, name, topic string, certBufferSize int) *NSQClient { + log.Println("Initializing NSQ client...") + + // Instantiate a producer. + conf := nsq.NewConfig() + + conn, err := nsq.NewProducer(addr, conf) + if err != nil { + log.Println(err) + } + + log.Println("Connected to NSQ server at", addr) + + // Silence log output from NSQ + conn.SetLogger(nullLogger{}, nsq.LogLevelError) + + nsqc := &NSQClient{ + conn: conn, + addr: addr, + topic: topic, + BaseClient: BaseClient{ + broadcastChan: make(chan []byte, certBufferSize), + stopChan: make(chan struct{}), + name: name, + subType: subType, + }, + } + nsqc.isConnected = true + + go nsqc.broadcastHandler() + go nsqc.reconnectHandler() + + return nsqc +} + +// reconnectHandler is a background job that attempts to reconnect to the NSQ server if the connection is lost. +func (c *NSQClient) reconnectHandler() { + for { + select { + case <-c.stopChan: + log.Println("Stopping reconnectHandler for nsq producer:", c.addr) + c.conn.Stop() + + return + default: + if c.isConnected { + // If already connected or no connection exists, skip reconnection + time.Sleep(5 * time.Second) + continue + } + + // Attempt to connect to the NSQ server + err := c.conn.Ping() + if err != nil { + log.Printf("Reconnect to NSQ server failed: '%v'. Retrying in 5s...", err) + time.Sleep(5 * time.Second) + + continue + } + + c.isConnected = true + log.Println("Reconnected to NSQ server at", c.addr) + } + } +} + +// Each client has a broadcastHandler that runs in the background and sends out the broadcast messages to the client. +func (c *NSQClient) broadcastHandler() { + defer func() { + log.Println("Closing broadcast handler for nsq producer:", c.addr) + if c.conn != nil { + // Gracefully stop the producer when appropriate (e.g. before shutting down the service) + c.conn.Stop() + } + + ClientHandler.UnregisterClient(c.name) + }() + + batch := make([][]byte, 0, nsqMaxBatchSize) + t := time.NewTicker(nsqMaxBatchWait) + + for { + select { + case <-c.stopChan: + return + case message, ok := <-c.broadcastChan: + if !ok { + log.Println("broadcastChan closed for nsqClient:", c.addr) + return + } + + // Drop messages if not connected + if !c.isConnected { + time.Sleep(5 * time.Second) + continue + } + + batch = append(batch, message) + + // Write batch if it reaches max size + if len(batch) >= nsqMaxBatchSize { + err := c.writeBatch(batch) + if err != nil { + log.Println("Failed to write messages to NSQ:", err) + c.isConnected = false + } + batch = batch[:0] + t.Reset(kafkaMaxBatchWait) + } + case <-t.C: + // If batch size has not reached nsqMaxBatchSize, write the batch after nsqMaxBatchWait + if len(batch) == 0 { + continue + } + + err := c.writeBatch(batch) + if err != nil { + log.Println("Failed to write messages to NSQ:", err) + c.isConnected = false + } + batch = batch[:0] + } + } +} + +func (c *NSQClient) writeBatch(batch [][]byte) error { + if len(batch) == 0 { + return nil + } + + // Synchronously publish a batch of messages to the specified topic. + err := c.conn.MultiPublish(c.topic, batch) + if err != nil { + log.Println("Failed to write messages to NSQ:", err) + c.isConnected = false + return err + } + + return nil +} diff --git a/internal/web/client.go b/internal/broadcast/websocketclient.go similarity index 79% rename from internal/web/client.go rename to internal/broadcast/websocketclient.go index 6626c2a..29b913e 100644 --- a/internal/web/client.go +++ b/internal/broadcast/websocketclient.go @@ -1,4 +1,4 @@ -package web +package broadcast import ( "fmt" @@ -13,41 +13,35 @@ import ( const idChars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" -const ( - SubTypeFull SubscriptionType = iota - SubTypeLite - SubTypeDomain -) - -type SubscriptionType int - -// client represents a single client's connection to the server. -type client struct { - clientData - - id string - conn *websocket.Conn - broadcastChan chan []byte - subType SubscriptionType - skippedCerts uint64 -} - -type clientData struct { +// WebsocketClient represents a single WebSocket client's connection to the server. +type WebsocketClient struct { + conn *websocket.Conn userAgent string - connectionIP string - connectionPort string + hostIP string + hostPort string realIPFromHeader string + + *BaseClient } -// newClient creates a new client struct that holds information about a connected client. -func newClient(conn *websocket.Conn, subType SubscriptionType, data clientData, certBufferSize int) *client { - return &client{ - clientData: data, - id: generateClientID(), - conn: conn, - broadcastChan: make(chan []byte, certBufferSize), - subType: subType, +// NewWebsocketClient creates a new WebSocket client from the given connection. +func NewWebsocketClient(conn *websocket.Conn, subType SubscriptionType, name, userAgent, hostIP, hostPort, realIPFromHeader string, certBufferSize int) *WebsocketClient { + c := &WebsocketClient{ + conn: conn, + userAgent: userAgent, + hostIP: hostIP, + hostPort: hostPort, + realIPFromHeader: realIPFromHeader, + BaseClient: &BaseClient{ + broadcastChan: make(chan []byte, certBufferSize), + name: name, + subType: subType, + }, } + go c.broadcastHandler() + go c.listenWebsocket() + + return c } // generateClientID generates a random 8-char identifier for the client. @@ -62,7 +56,7 @@ func generateClientID() string { } // Each client has a broadcastHandler that runs in the background and sends out the broadcast messages to the client. -func (c *client) broadcastHandler() { +func (c *WebsocketClient) broadcastHandler() { writeWait := 60 * time.Second pingTicker := time.NewTicker(30 * time.Second) @@ -109,10 +103,10 @@ func (c *client) broadcastHandler() { // listenWebsocket is running in the background on a goroutine and listens for messages from the client. // It responds to ping messages with a pong message. It closes the connection if the client sends // a close message or no ping is received within 65 seconds. -func (c *client) listenWebsocket() { +func (c *WebsocketClient) listenWebsocket() { defer func() { _ = c.conn.Close() - ClientHandler.unregisterClient(c) + ClientHandler.UnregisterClient(c.name) }() readWait := 65 * time.Second @@ -184,19 +178,14 @@ func sanitizeInput(s string) string { } // Name returns the name/identifier for this client. -func (c *client) Name() string { - var clientName string +func (c *WebsocketClient) Name() string { + clientName := fmt.Sprintf("[%s] - ", c.name) - clientName = fmt.Sprintf("[%s] - ", c.id) - - connIP := sanitizeInput(c.connectionIP) - connPort := sanitizeInput(c.connectionPort) realIP := sanitizeInput(c.realIPFromHeader) - - socket := net.JoinHostPort(connIP, connPort) + socket := net.JoinHostPort(c.hostIP, c.hostPort) // If the realIP is set and if it differs from the connection IP, return both the connection IP and the real IP. - if realIP != "" && realIP != connIP { + if realIP != "" && realIP != c.hostIP { clientName += fmt.Sprintf("%s (via %s)", socket, realIP) } else { clientName += socket diff --git a/internal/web/client_test.go b/internal/broadcast/websocketclient_test.go similarity index 92% rename from internal/web/client_test.go rename to internal/broadcast/websocketclient_test.go index b1e2b30..cc39dd3 100644 --- a/internal/web/client_test.go +++ b/internal/broadcast/websocketclient_test.go @@ -1,19 +1,17 @@ -package web +package broadcast import ( "fmt" "testing" ) -func newTestClient(id, connectionIP, connectionPort, realIP, userAgent string) *client { - c := &client{ - clientData: clientData{ - connectionIP: connectionIP, - connectionPort: connectionPort, - realIPFromHeader: realIP, - userAgent: userAgent, - }, - id: id, +func newTestClient(name, connectionIP, connectionPort, realIP, userAgent string) *WebsocketClient { + c := &WebsocketClient{ + hostIP: connectionIP, + hostPort: connectionPort, + realIPFromHeader: realIP, + userAgent: userAgent, + BaseClient: &BaseClient{name: name}, } return c } diff --git a/internal/certificatetransparency/ct-watcher.go b/internal/certificatetransparency/ct-watcher.go index b4eb458..6a271eb 100644 --- a/internal/certificatetransparency/ct-watcher.go +++ b/internal/certificatetransparency/ct-watcher.go @@ -13,6 +13,7 @@ import ( "sync/atomic" "time" + "github.com/d-Rickyy-b/certstream-server-go/internal/broadcast" "github.com/d-Rickyy-b/certstream-server-go/internal/config" "github.com/d-Rickyy-b/certstream-server-go/internal/metrics" "github.com/d-Rickyy-b/certstream-server-go/internal/models" @@ -553,7 +554,7 @@ func certHandler(entryChan chan models.Entry) { } // Run JSON encoding in the background and send the result to the clients. - web.ClientHandler.Broadcast <- entry + broadcast.ClientHandler.MessageQueue <- entry // Update metrics url := entry.Data.Source.NormalizedURL diff --git a/internal/certstream/certstream.go b/internal/certstream/certstream.go index 3bf4065..01cda11 100644 --- a/internal/certstream/certstream.go +++ b/internal/certstream/certstream.go @@ -7,10 +7,13 @@ package certstream import ( "fmt" "log" + "net" "os" "os/signal" + "strconv" "syscall" + "github.com/d-Rickyy-b/certstream-server-go/internal/broadcast" "github.com/d-Rickyy-b/certstream-server-go/internal/certificatetransparency" "github.com/d-Rickyy-b/certstream-server-go/internal/config" "github.com/d-Rickyy-b/certstream-server-go/internal/metrics" @@ -32,15 +35,20 @@ func NewRawCertstream(config config.Config) *Certstream { } // NewCertstreamServer creates a new Certstream server from a config struct. -func NewCertstreamServer(config config.Config) (*Certstream, error) { - cs := NewRawCertstream(config) +func NewCertstreamServer(cfg config.Config) (*Certstream, error) { + cs := NewRawCertstream(cfg) + // Start the broadcast dispatcher + broadcast.NewDispatcher() + broadcast.ClientHandler.Start() + + // TODO: add support do disable websocket Server // Initialize the webserver used for the websocket server webserver := web.NewWebsocketServer( - config.Webserver.ListenAddr, - config.Webserver.ListenPort, - config.Webserver.CertPath, - config.Webserver.CertKeyPath, + cfg.Webserver.ListenAddr, + cfg.Webserver.ListenPort, + cfg.Webserver.CertPath, + cfg.Webserver.CertKeyPath, ) cs.webserver = webserver cs.watcher = certificatetransparency.NewWatcher() @@ -48,6 +56,53 @@ func NewCertstreamServer(config config.Config) (*Certstream, error) { // Setup metrics server cs.setupMetrics(webserver) + // Initialize the stream processors if configured and enabled. + for _, streamProcessor := range cfg.StreamProcessing { + if !*streamProcessor.Enabled { + continue + } + + addr := net.JoinHostPort(streamProcessor.ServerAddr, strconv.Itoa(streamProcessor.ServerPort)) + log.Printf("Initializing stream processor: %s at %s\n", streamProcessor.Name, addr) + + var subscriptionType broadcast.SubscriptionType + switch streamProcessor.Stream { + case config.StreamTypeFull: + subscriptionType = broadcast.SubTypeFull + case config.StreamTypeLite: + subscriptionType = broadcast.SubTypeLite + case config.StreamTypeDomainsOnly: + subscriptionType = broadcast.SubTypeDomain + } + + switch streamProcessor.Type { + case "nsq": + log.Println("Initializing NSQ client...") + nc := broadcast.NewNSQClient( + subscriptionType, + addr, + streamProcessor.Name, + streamProcessor.Topic, + cfg.General.BufferSizes.Websocket, + ) + broadcast.ClientHandler.RegisterClient(nc) + case "kafka": + log.Println("Initializing Kafka client...") + kc := broadcast.NewKafkaClient( + subscriptionType, + addr, + streamProcessor.Name, + streamProcessor.Topic, + string(streamProcessor.Compression), + cfg.General.BufferSizes.Websocket, + ) + + broadcast.ClientHandler.RegisterClient(kc) + default: + log.Printf("Unknown stream processor type '%s' for %s. Skipping...\n", streamProcessor.Type, streamProcessor.Name) + } + } + return cs, nil } diff --git a/internal/config/buffersizes.go b/internal/config/buffersizes.go new file mode 100644 index 0000000..d826394 --- /dev/null +++ b/internal/config/buffersizes.go @@ -0,0 +1,16 @@ +package config + +type BufferSizes struct { + Websocket int `mapstructure:"websocket"` + CTLog int `mapstructure:"ctlog"` + BroadcastManager int `mapstructure:"broadcastmanager"` + Dispatcher int `mapstructure:"dispatcher"` +} + +func (b *BufferSizes) Valid() bool { + if b.Websocket <= 0 || b.CTLog <= 0 || b.BroadcastManager <= 0 || b.Dispatcher <= 0 { + return false + } + + return true +} diff --git a/internal/config/config.go b/internal/config/config.go index 02eefea..30a3505 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,7 +4,6 @@ import ( "errors" "fmt" "log" - "net" "regexp" "strings" @@ -12,64 +11,20 @@ import ( ) var ( + // AppConfig holds the parsed configuration. AppConfig Config Version = "1.9.0" ErrInvalidConfig = errors.New("invalid configuration") + URLPathRegex = regexp.MustCompile(`^(/[a-zA-Z0-9\-._]+)+$`) + URLRegex = regexp.MustCompile(`^https?://[a-zA-Z0-9\-._]+(:[0-9]+)?(/[a-zA-Z0-9\-._]+)*/?$`) ) -type ServerConfig struct { - ListenAddr string `mapstructure:"listen_addr"` - ListenPort int `mapstructure:"listen_port"` - CertPath string `mapstructure:"cert_path"` - CertKeyPath string `mapstructure:"cert_key_path"` - RealIP bool `mapstructure:"real_ip"` - TrustedProxies []string `mapstructure:"trusted_proxies"` - Whitelist []string `mapstructure:"whitelist"` -} - -type LogConfig struct { - Operator string `mapstructure:"operator"` - URL string `mapstructure:"url"` - Description string `mapstructure:"description"` -} - -type BufferSizes struct { - Websocket int `mapstructure:"websocket"` - CTLog int `mapstructure:"ctlog"` - BroadcastManager int `mapstructure:"broadcastmanager"` -} - type Config struct { - Webserver struct { - ServerConfig `mapstructure:",squash"` - - FullURL string `mapstructure:"full_url"` - LiteURL string `mapstructure:"lite_url"` - DomainsOnlyURL string `mapstructure:"domains_only_url"` - CompressionEnabled bool `mapstructure:"compression_enabled"` - } - Prometheus struct { - ServerConfig `mapstructure:",squash"` - - Enabled bool `mapstructure:"enabled"` - MetricsURL string `mapstructure:"metrics_url"` - ExposeSystemMetrics bool `mapstructure:"expose_system_metrics"` - } - General struct { - // DisableDefaultLogs indicates whether the default logs used in Google Chrome and provided by Google should be disabled. - DisableDefaultLogs bool `mapstructure:"disable_default_logs"` - // AdditionalLogs contains additional logs provided by the user that can be used in addition to the default logs. - AdditionalLogs []LogConfig `mapstructure:"additional_logs"` - AdditionalTiledLogs []LogConfig `mapstructure:"additional_tiled_logs"` - ExcludedLogs []LogConfig `mapstructure:"excluded_logs"` - BufferSizes BufferSizes `mapstructure:"buffer_sizes"` - DropOldLogs *bool `mapstructure:"drop_old_logs"` - Recovery struct { - Enabled bool `mapstructure:"enabled"` - CTIndexFile string `mapstructure:"ct_index_file"` - } `mapstructure:"recovery"` - } + Webserver Webserver + Prometheus Prometheus + StreamProcessing []StreamProcessor `mapstructure:"stream_processing"` + General General } // ReadConfig reads the configuration using Viper and returns a filled Config struct. @@ -173,176 +128,24 @@ func loadConfigFromViper(v *viper.Viper) (Config, error) { // validateConfig validates the config values and sets defaults for missing values. func validateConfig(config *Config) bool { // Still matches invalid IP addresses but good enough for detecting completely wrong formats - URLPathRegex := regexp.MustCompile(`^(/[a-zA-Z0-9\-._]+)+$`) - URLRegex := regexp.MustCompile(`^https?://[a-zA-Z0-9\-._]+(:[0-9]+)?(/[a-zA-Z0-9\-._]+)*/?$`) // Check webserver config - if config.Webserver.ListenAddr == "" || net.ParseIP(config.Webserver.ListenAddr) == nil { - log.Fatalln("Webhook listen IP is not a valid IP: ", config.Webserver.ListenAddr) + if !config.Webserver.Valid() { return false } - if config.Webserver.ListenPort == 0 { - log.Fatalln("Webhook listen port is not set") + if !config.Prometheus.Valid() { return false } - if config.Webserver.FullURL == "" || !URLPathRegex.MatchString(config.Webserver.FullURL) { - log.Println("Webhook full URL is not set or does not match pattern '/...'") - - config.Webserver.FullURL = "/full-stream" - } - - if config.Webserver.LiteURL == "" || !URLPathRegex.MatchString(config.Webserver.FullURL) { - log.Println("Webhook lite URL is not set or does not match pattern '/...'") - - config.Webserver.LiteURL = "/" - } - - if config.Webserver.DomainsOnlyURL == "" || !URLPathRegex.MatchString(config.Webserver.DomainsOnlyURL) { - log.Println("Webhook domains only URL is not set or does not match pattern '/...'") - - config.Webserver.FullURL = "/domains-only" - } - - if config.Webserver.FullURL == config.Webserver.LiteURL { - log.Fatalln("Webhook full URL is the same as lite URL - please fix the config!") - } - - if config.Webserver.DomainsOnlyURL == "" { - config.Webserver.FullURL = "/domains-only" - } - - for _, ip := range config.Webserver.TrustedProxies { - if net.ParseIP(ip) != nil { - continue - } - - _, _, err := net.ParseCIDR(ip) - if err != nil { - log.Fatalln("Invalid IP/CIDR in webserver trusted_proxies: ", ip) + for _, processor := range config.StreamProcessing { + if !processor.Valid() { return false } } - //nolint:nestif - if config.Prometheus.Enabled { - if config.Prometheus.ListenAddr == "" || net.ParseIP(config.Prometheus.ListenAddr) == nil { - log.Fatalln("Metrics export IP is not a valid IP") - return false - } - - if config.Prometheus.ListenPort == 0 { - log.Fatalln("Metrics export port is not set") - return false - } - - if config.Prometheus.Whitelist == nil { - config.Prometheus.Whitelist = []string{} - } - - // Check if IPs in whitelist match pattern - for _, ip := range config.Prometheus.Whitelist { - if net.ParseIP(ip) != nil { - continue - } - - // Provided entry is not an IP, check if it's a CIDR range - _, _, err := net.ParseCIDR(ip) - if err != nil { - log.Fatalln("Invalid IP in metrics whitelist: ", ip) - return false - } - } - - for _, ip := range config.Prometheus.TrustedProxies { - if net.ParseIP(ip) != nil { - continue - } - - _, _, err := net.ParseCIDR(ip) - if err != nil { - log.Fatalln("Invalid IP/CIDR in prometheus trusted_proxies: ", ip) - return false - } - } - } - - var validLogs, validTiledLogs, validExcludedLogs []LogConfig - - if len(config.General.AdditionalLogs) > 0 { - for _, ctLog := range config.General.AdditionalLogs { - if !URLRegex.MatchString(ctLog.URL) { - log.Println("Ignoring invalid additional log URL: ", ctLog.URL) - continue - } - - validLogs = append(validLogs, ctLog) - } - } - - if len(config.General.AdditionalTiledLogs) > 0 { - for _, ctLog := range config.General.AdditionalTiledLogs { - if !URLRegex.MatchString(ctLog.URL) { - log.Println("Ignoring invalid additional log URL: ", ctLog.URL) - continue - } - - validTiledLogs = append(validTiledLogs, ctLog) - } - } - - if len(config.General.ExcludedLogs) > 0 { - for _, excludedLog := range config.General.ExcludedLogs { - excludedLog.Operator = strings.TrimSpace(excludedLog.Operator) - excludedLog.URL = strings.TrimSpace(excludedLog.URL) - - if excludedLog.Operator == "" && excludedLog.URL == "" { - log.Println("Ignoring empty excluded_logs entry. Set operator and/or url.") - continue - } - - if excludedLog.URL != "" && !URLRegex.MatchString(excludedLog.URL) { - log.Println("Ignoring invalid excluded log URL: ", excludedLog.URL) - continue - } - - validExcludedLogs = append(validExcludedLogs, excludedLog) - } - } - - config.General.AdditionalLogs = validLogs - config.General.AdditionalTiledLogs = validTiledLogs - config.General.ExcludedLogs = validExcludedLogs - - if len(config.General.AdditionalLogs) == 0 && len(config.General.AdditionalTiledLogs) == 0 && config.General.DisableDefaultLogs { - log.Fatalln("Default logs are disabled, but no additional logs are configured. Please add at least one log to the config or enable default logs.") - } - - if config.General.BufferSizes.Websocket <= 0 { - config.General.BufferSizes.Websocket = 300 - } - - if config.General.BufferSizes.CTLog <= 0 { - config.General.BufferSizes.CTLog = 1000 - } - - if config.General.BufferSizes.BroadcastManager <= 0 { - config.General.BufferSizes.BroadcastManager = 10000 - } - - // If the cleanup flag is not set, default to true - if config.General.DropOldLogs == nil { - log.Println("drop_old_logs is not set, defaulting to true") - - defaultCleanup := true - config.General.DropOldLogs = &defaultCleanup - } - - if config.General.Recovery.Enabled && config.General.Recovery.CTIndexFile == "" { - log.Println("Recovery enabled but no index file specified. Defaulting to ./ct_index.json") - - config.General.Recovery.CTIndexFile = "./ct_index.json" + if !config.General.Valid() { + return false } return true diff --git a/internal/config/general.go b/internal/config/general.go new file mode 100644 index 0000000..4a244e9 --- /dev/null +++ b/internal/config/general.go @@ -0,0 +1,115 @@ +package config + +import ( + "log" + "strings" +) + +type LogConfig struct { + Operator string `mapstructure:"operator"` + URL string `mapstructure:"url"` + Description string `mapstructure:"description"` +} + +type General struct { + // DisableDefaultLogs indicates whether the default logs used in Google Chrome and provided by Google should be disabled. + DisableDefaultLogs bool `mapstructure:"disable_default_logs"` + // AdditionalLogs contains additional logs provided by the user that can be used in addition to the default logs. + AdditionalLogs []LogConfig `mapstructure:"additional_logs"` + AdditionalTiledLogs []LogConfig `mapstructure:"additional_tiled_logs"` + ExcludedLogs []LogConfig `mapstructure:"excluded_logs"` + // BufferSizes contains the buffer sizes for the different components of the server. They usually don't need any adjustments. + BufferSizes BufferSizes `mapstructure:"buffer_sizes"` + // DropOldLogs indicates whether downloading CT-Logs should start at the latest index (true) or should from the beginning (false). + DropOldLogs *bool `mapstructure:"drop_old_logs"` + Recovery struct { + Enabled bool `mapstructure:"enabled"` + CTIndexFile string `mapstructure:"ct_index_file"` + } `mapstructure:"recovery"` +} + +func (g *General) Valid() bool { + var validLogs, validTiledLogs, validExcludedLogs []LogConfig + + if len(g.AdditionalLogs) > 0 { + for _, ctLog := range g.AdditionalLogs { + if !URLRegex.MatchString(ctLog.URL) { + log.Println("Ignoring invalid additional log URL: ", ctLog.URL) + continue + } + + validLogs = append(validLogs, ctLog) + } + } + + if len(g.AdditionalTiledLogs) > 0 { + for _, ctLog := range g.AdditionalTiledLogs { + if !URLRegex.MatchString(ctLog.URL) { + log.Println("Ignoring invalid additional log URL: ", ctLog.URL) + continue + } + + validTiledLogs = append(validTiledLogs, ctLog) + } + } + + if len(g.ExcludedLogs) > 0 { + for _, excludedLog := range g.ExcludedLogs { + excludedLog.Operator = strings.TrimSpace(excludedLog.Operator) + excludedLog.URL = strings.TrimSpace(excludedLog.URL) + + if excludedLog.Operator == "" && excludedLog.URL == "" { + log.Println("Ignoring empty excluded_logs entry. Set operator and/or url.") + continue + } + + if excludedLog.URL != "" && !URLRegex.MatchString(excludedLog.URL) { + log.Println("Ignoring invalid excluded log URL: ", excludedLog.URL) + continue + } + + validExcludedLogs = append(validExcludedLogs, excludedLog) + } + } + + g.AdditionalLogs = validLogs + g.AdditionalTiledLogs = validTiledLogs + g.ExcludedLogs = validExcludedLogs + + if len(g.AdditionalLogs) == 0 && len(g.AdditionalTiledLogs) == 0 && g.DisableDefaultLogs { + log.Fatalln("Default logs are disabled, but no additional logs are configured. Please add at least one log to the config or enable default logs.") + } + + if g.BufferSizes.Websocket <= 0 { + g.BufferSizes.Websocket = 300 + } + + if g.BufferSizes.CTLog <= 0 { + g.BufferSizes.CTLog = 1000 + } + + // For backward compatibility, copy value from deprecated BroadcastManager field + if g.BufferSizes.BroadcastManager != 0 { + g.BufferSizes.Dispatcher = g.BufferSizes.BroadcastManager + } + + if g.BufferSizes.Dispatcher <= 0 { + g.BufferSizes.Dispatcher = 10000 + } + + // If the cleanup flag is not set, default to true + if g.DropOldLogs == nil { + log.Println("drop_old_logs is not set, defaulting to true") + + defaultCleanup := true + g.DropOldLogs = &defaultCleanup + } + + if g.Recovery.Enabled && g.Recovery.CTIndexFile == "" { + log.Println("Recovery enabled but no index file specified. Defaulting to ./ct_index.json") + + g.Recovery.CTIndexFile = "./ct_index.json" + } + + return true +} diff --git a/internal/config/prometheus.go b/internal/config/prometheus.go new file mode 100644 index 0000000..cfbcf58 --- /dev/null +++ b/internal/config/prometheus.go @@ -0,0 +1,62 @@ +package config + +import ( + "log" + "net" +) + +type Prometheus struct { + ServerConfig `mapstructure:",squash"` + + Enabled bool `mapstructure:"enabled"` + MetricsURL string `mapstructure:"metrics_url"` + ExposeSystemMetrics bool `mapstructure:"expose_system_metrics"` +} + +func (p *Prometheus) Valid() bool { + if !p.Enabled { + return true + } + + if p.ListenAddr == "" || net.ParseIP(p.ListenAddr) == nil { + log.Fatalln("Metrics export IP is not a valid IP") + return false + } + + if p.ListenPort == 0 { + log.Fatalln("Metrics export port is not set") + return false + } + + if p.Whitelist == nil { + p.Whitelist = []string{} + } + + // Check if IPs in whitelist match pattern + for _, ip := range p.Whitelist { + if net.ParseIP(ip) != nil { + continue + } + + // Provided entry is not an IP, check if it's a CIDR range + _, _, err := net.ParseCIDR(ip) + if err != nil { + log.Fatalln("Invalid IP in metrics whitelist: ", ip) + return false + } + } + + for _, ip := range p.TrustedProxies { + if net.ParseIP(ip) != nil { + continue + } + + _, _, err := net.ParseCIDR(ip) + if err != nil { + log.Fatalln("Invalid IP/CIDR in prometheus trusted_proxies: ", ip) + return false + } + } + + return true +} diff --git a/internal/config/serverconfig.go b/internal/config/serverconfig.go new file mode 100644 index 0000000..a246d4a --- /dev/null +++ b/internal/config/serverconfig.go @@ -0,0 +1,11 @@ +package config + +type ServerConfig struct { + ListenAddr string `mapstructure:"listen_addr"` + ListenPort int `mapstructure:"listen_port"` + CertPath string `mapstructure:"cert_path"` + CertKeyPath string `mapstructure:"cert_key_path"` + RealIP bool `mapstructure:"real_ip"` + TrustedProxies []string `mapstructure:"trusted_proxies"` + Whitelist []string `mapstructure:"whitelist"` +} diff --git a/internal/config/streamprocessing.go b/internal/config/streamprocessing.go new file mode 100644 index 0000000..7dd8478 --- /dev/null +++ b/internal/config/streamprocessing.go @@ -0,0 +1,133 @@ +package config + +import ( + "log" + "net" +) + +// StreamProcessorType represents the type of stream processing tool to use. +// Supported types are "kafka" and "nqs". +type StreamProcessorType string + +const ( + StreamProcessorTypeKafka StreamProcessorType = "kafka" + StreamProcessorTypeNQS StreamProcessorType = "nqs" +) + +// StreamType represents the type of stream to process. +// Supported types are "full", "lite", and "domains-only". +type StreamType string + +const ( + StreamTypeFull StreamType = "full" + StreamTypeLite StreamType = "lite" + StreamTypeDomainsOnly StreamType = "domains-only" +) + +// Compression represents the compression type for stream processing. +type Compression string + +const ( + CompressionNone Compression = "none" + CompressionGzip Compression = "gzip" + CompressionSnappy Compression = "snappy" + CompressionZstd Compression = "zstd" + CompressionLz4 Compression = "lz4" + CompressionDeflate Compression = "deflate" +) + +func (c Compression) setDefaults() { + if c == "" { + c = CompressionNone + } +} + +// Valid returns true if the compression type is valid. +func (c Compression) Valid() bool { + c.setDefaults() + + switch c { + case CompressionNone, CompressionGzip, CompressionSnappy, CompressionZstd, CompressionLz4: + return true + default: + return false + } +} + +// SupportedBy returns true if the compression type is supported by the stream processing tool. +func (c Compression) SupportedBy(t StreamProcessorType) bool { + if c == CompressionNone { + return true + } + + switch t { + case StreamProcessorTypeKafka: + return c == CompressionNone || c == CompressionGzip || c == CompressionSnappy || c == CompressionZstd || c == CompressionLz4 + case StreamProcessorTypeNQS: + return c == CompressionNone || c == CompressionDeflate || c == CompressionSnappy + default: + return false + } +} + +type StreamProcessor struct { + Name string `mapstructure:"name"` + Type StreamProcessorType `mapstructure:"type"` + Stream StreamType `mapstructure:"stream"` + Enabled *bool `mapstructure:"enabled"` + ServerAddr string `mapstructure:"server_addr"` + ServerPort int `mapstructure:"server_port"` + Topic string `mapstructure:"topic"` + Compression Compression `mapstructure:"compression"` +} + +func (s *StreamProcessor) setDefaults() { + if s.Enabled == nil { + enabled := true + s.Enabled = &enabled + } + if s.Compression == "" { + s.Compression = CompressionNone + } +} + +func (s *StreamProcessor) Valid() bool { + s.setDefaults() + + ip := net.ParseIP(s.ServerAddr) + if ip == nil { + log.Fatalln("Invalid IP for stream processor:", s.ServerAddr) + return false + } + + if s.ServerPort <= 0 { + log.Fatalln("Invalid server port for stream processor:", s.ServerPort) + return false + } + + switch s.Type { + case StreamProcessorTypeKafka, StreamProcessorTypeNQS: + default: + log.Fatalf("Invalid stream processor type '%s' for name '%s'\n", s.Type, s.Name) + return false + } + + if s.Topic == "" { + log.Println("Found stream processing config with empty topic - using \"certstream\" as topic for name", s.Name) + s.Topic = "certstream" + } + + // Check that the compression type is generally valid + if !s.Compression.Valid() { + log.Fatalf("Invalid compression '%s' for stream processor '%s'\n", s.Compression, s.Name) + return false + } + + // Check that the compression type is supported by the stream processing tool + if !s.Compression.SupportedBy(s.Type) { + log.Fatalf("Compression '%s' is not supported by stream processor type '%s' for name '%s'\n", s.Compression, s.Type, s.Name) + return false + } + + return true +} diff --git a/internal/config/webserver.go b/internal/config/webserver.go new file mode 100644 index 0000000..fb846d9 --- /dev/null +++ b/internal/config/webserver.go @@ -0,0 +1,69 @@ +package config + +import ( + "log" + "net" +) + +type Webserver struct { + ServerConfig `mapstructure:",squash"` + + FullURL string `mapstructure:"full_url"` + LiteURL string `mapstructure:"lite_url"` + DomainsOnlyURL string `mapstructure:"domains_only_url"` + CompressionEnabled bool `mapstructure:"compression_enabled"` +} + +func (w *Webserver) Valid() bool { + // Still matches invalid IP addresses but good enough for detecting completely wrong formats + + if w.ListenAddr == "" || net.ParseIP(w.ListenAddr) == nil { + log.Fatalln("Webhook listen IP is not a valid IP: ", w.ListenAddr) + return false + } + + if w.ListenPort == 0 { + log.Fatalln("Webhook listen port is not set") + return false + } + + if w.FullURL == "" || !URLPathRegex.MatchString(w.FullURL) { + log.Println("Webhook full URL is not set or does not match pattern '/...'") + + w.FullURL = "/full-stream" + } + + if w.LiteURL == "" || !URLPathRegex.MatchString(w.FullURL) { + log.Println("Webhook lite URL is not set or does not match pattern '/...'") + + w.LiteURL = "/" + } + + if w.DomainsOnlyURL == "" || !URLPathRegex.MatchString(w.DomainsOnlyURL) { + log.Println("Webhook domains only URL is not set or does not match pattern '/...'") + + w.FullURL = "/domains-only" + } + + if w.FullURL == w.LiteURL { + log.Fatalln("Webhook full URL is the same as lite URL - please fix the config!") + } + + if w.DomainsOnlyURL == "" { + w.FullURL = "/domains-only" + } + + for _, ip := range w.TrustedProxies { + if net.ParseIP(ip) != nil { + continue + } + + _, _, err := net.ParseCIDR(ip) + if err != nil { + log.Fatalln("Invalid IP/CIDR in webserver trusted_proxies: ", ip) + return false + } + } + + return true +} diff --git a/internal/metrics/logmetrics.go b/internal/metrics/logmetrics.go index 4f45102..fb10415 100644 --- a/internal/metrics/logmetrics.go +++ b/internal/metrics/logmetrics.go @@ -327,10 +327,12 @@ func GetProcessedPrecerts() int64 { return ProcessedPrecerts } +// GetCertMetrics returns a copy of the internal metrics map. func GetCertMetrics() CTMetrics { return Metrics.GetCTMetrics() } +// GetLogOperators returns a map of operator names to a list of CT logs. func GetLogOperators() map[string][]string { return Metrics.OperatorLogMapping() } diff --git a/internal/metrics/prometheus.go b/internal/metrics/prometheus.go index caca426..a6a9ac4 100644 --- a/internal/metrics/prometheus.go +++ b/internal/metrics/prometheus.go @@ -64,35 +64,22 @@ func (pm *PrometheusExporter) RegisterGaugeMetricInt(label string, callback func } // RegisterClient registers a new gauge metric for the client with the given name. -func (pm *PrometheusExporter) RegisterClient(id, connIP, connPort, realIP, useragent string, skippedCertsCallback func() float64) { +func (pm *PrometheusExporter) RegisterClient(id string, skippedCertsCallback func() float64) { argMap := make(map[string]string) argMap["id"] = id - argMap["conn_ip"] = connIP - argMap["conn_port"] = connPort - argMap["real_ip"] = realIP - argMap["useragent"] = useragent label := createMetric("certstreamservergo_skipped_certs", argMap) - // label := fmt.Sprintf("certstreamservergo_skipped_certs{id=\"%s\",conn_ip=\"%s\",conn_port=\"%s\",real_ip=\"%s\",useragent=\"%s\"}", - // id, connIP, connPort, realIP, useragent) metrics.GetOrCreateGauge(label, skippedCertsCallback) } // UnregisterClient unregisters the metric for the client with the given name. -func (pm *PrometheusExporter) UnregisterClient(id, connIP, connPort, realIP, useragent string) { +func (pm *PrometheusExporter) UnregisterClient(id string) { argMap := make(map[string]string) argMap["id"] = id - argMap["conn_ip"] = connIP - argMap["conn_port"] = connPort - argMap["real_ip"] = realIP - argMap["useragent"] = useragent label := createMetric("certstreamservergo_skipped_certs", argMap) - // label := fmt.Sprintf("certstreamservergo_skipped_certs{id=\"%s\",conn_ip=\"%s\",conn_port=\"%s\",real_ip=\"%s\",useragent=\"%s\"}", - // id, connIP, connPort, realIP, useragent) - ok := metrics.UnregisterMetric(label) if !ok { log.Printf("failed to unregister metric '%s'", label) diff --git a/internal/web/broadcastmanager.go b/internal/web/broadcastmanager.go deleted file mode 100644 index e2e4350..0000000 --- a/internal/web/broadcastmanager.go +++ /dev/null @@ -1,133 +0,0 @@ -package web - -import ( - "log" - "sync" - - "github.com/d-Rickyy-b/certstream-server-go/internal/metrics" - "github.com/d-Rickyy-b/certstream-server-go/internal/models" -) - -type BroadcastManager struct { - Broadcast chan models.Entry - clients []*client - clientLock sync.RWMutex -} - -func NewBroadcastManager() *BroadcastManager { - bm := &BroadcastManager{} - metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"full\"}", bm.ClientFullCount) - metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"lite\"}", bm.ClientLiteCount) - metrics.Prometheus.RegisterGaugeMetricInt("certstreamservergo_clients_total{type=\"domain\"}", bm.ClientDomainsCount) - - return bm -} - -// registerClient adds a client to the list of clients of the BroadcastManager. -// The client will receive certificate broadcasts right after registration. -func (bm *BroadcastManager) registerClient(c *client) { - bm.clientLock.Lock() - bm.clients = append(bm.clients, c) - log.Printf("Clients: %d, Capacity: %d\n", len(bm.clients), cap(bm.clients)) - metrics.Prometheus.RegisterClient(c.id, c.connectionIP, c.connectionPort, c.realIPFromHeader, c.userAgent, func() float64 { return float64(c.skippedCerts) }) - bm.clientLock.Unlock() -} - -// unregisterClient removes a client from the list of clients of the BroadcastManager. -// The client will no longer receive certificate broadcasts right after unregistering. -func (bm *BroadcastManager) unregisterClient(c *client) { - bm.clientLock.Lock() - log.Println("Unregistering client:", c.conn.RemoteAddr()) - - // Close the broadcast channel of the client, otherwise this leads to a memory leak - close(c.broadcastChan) - metrics.Prometheus.UnregisterClient(c.id, c.connectionIP, c.connectionPort, c.realIPFromHeader, c.userAgent) - - // Remove client from internal client list - for i, storedClient := range bm.clients { - if c != storedClient { - continue - } - - // Copy the last element of the slice to the position of the removed element - // Then remove the last element by re-slicing - bm.clients[i] = bm.clients[len(bm.clients)-1] - bm.clients[len(bm.clients)-1] = nil - bm.clients = bm.clients[:len(bm.clients)-1] - - break - } - - bm.clientLock.Unlock() -} - -// ClientFullCount returns the current number of clients connected to the service on the `full` endpoint. -func (bm *BroadcastManager) ClientFullCount() (count int64) { - return bm.clientCountByType(SubTypeFull) -} - -// ClientLiteCount returns the current number of clients connected to the service on the `lite` endpoint. -func (bm *BroadcastManager) ClientLiteCount() (count int64) { - return bm.clientCountByType(SubTypeLite) -} - -// ClientDomainsCount returns the current number of clients connected to the service on the `domains-only` endpoint. -func (bm *BroadcastManager) ClientDomainsCount() (count int64) { - return bm.clientCountByType(SubTypeDomain) -} - -// clientCountByType returns the current number of clients connected to the service on the endpoint matching -// the specified SubscriptionType. -func (bm *BroadcastManager) clientCountByType(subType SubscriptionType) (count int64) { - bm.clientLock.RLock() - defer bm.clientLock.RUnlock() - - for _, c := range bm.clients { - if c.subType == subType { - count++ - } - } - - return count -} - -// broadcaster is run in a goroutine and handles the dispatching of entries to clients. -func (bm *BroadcastManager) broadcaster() { - for { - var data []byte - - // Take entry out of broadcast channel and generate JSON representations for the entry. - entry := <-bm.Broadcast - dataLite := entry.JSONLite() - dataFull := entry.JSON() - dataDomain := entry.JSONDomains() - - bm.clientLock.RLock() - - for _, c := range bm.clients { - switch c.subType { - case SubTypeLite: - data = dataLite - case SubTypeFull: - data = dataFull - case SubTypeDomain: - data = dataDomain - default: - log.Printf("Unknown subscription type '%d'. Skipping client %s\n", c.subType, c.Name()) - continue - } - - select { - case c.broadcastChan <- data: - default: - // Default case is executed if the client's broadcast channel is full. - c.skippedCerts++ - if c.skippedCerts%1000 == 1 { - log.Printf("Client can't keep up. Skipped certs: %d | Client: %s\n", c.skippedCerts, c.Name()) - } - } - } - - bm.clientLock.RUnlock() - } -} diff --git a/internal/web/examplecert.go b/internal/web/examplecert.go index e841323..05576f3 100644 --- a/internal/web/examplecert.go +++ b/internal/web/examplecert.go @@ -29,7 +29,7 @@ func exampleDomains(w http.ResponseWriter, _ *http.Request) { w.Write(exampleCert.JSONDomains()) //nolint:errcheck } -// SetExampleCert sets one certificate as the example Cert that is returned by the example endpoints. +// SetExampleCert sets the example certificate to be used in the example endpoints. func SetExampleCert(cert models.Entry) { exampleCert = cert } diff --git a/internal/web/server.go b/internal/web/server.go index f3e1d80..72d0c8e 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -12,19 +12,15 @@ import ( "strings" "time" - "github.com/go-chi/chi/v5" - "github.com/go-chi/chi/v5/middleware" - + "github.com/d-Rickyy-b/certstream-server-go/internal/broadcast" "github.com/d-Rickyy-b/certstream-server-go/internal/config" - "github.com/d-Rickyy-b/certstream-server-go/internal/models" + "github.com/go-chi/chi/v5" + "github.com/go-chi/chi/v5/middleware" "github.com/gorilla/websocket" ) -var ( - ClientHandler = NewBroadcastManager() - upgrader websocket.Upgrader -) +var upgrader websocket.Upgrader type contextKey int @@ -228,7 +224,7 @@ func initFullWebsocket(w http.ResponseWriter, r *http.Request) { return } - setupClient(connection, SubTypeFull, r) + setupClient(connection, broadcast.SubTypeFull, r.RemoteAddr, r) } // initLiteWebsocket is called when a client connects to the / endpoint. @@ -240,7 +236,7 @@ func initLiteWebsocket(w http.ResponseWriter, r *http.Request) { return } - setupClient(connection, SubTypeLite, r) + setupClient(connection, broadcast.SubTypeLite, r.RemoteAddr, r) } // initDomainWebsocket is called when a client connects to the /domains-only endpoint. @@ -252,7 +248,7 @@ func initDomainWebsocket(w http.ResponseWriter, r *http.Request) { return } - setupClient(connection, SubTypeDomain, r) + setupClient(connection, broadcast.SubTypeDomain, r.RemoteAddr, r) } // upgradeConnection upgrades the connection to a websocket and returns the connection. @@ -283,7 +279,7 @@ func upgradeConnection(w http.ResponseWriter, r *http.Request) (*websocket.Conn, } // setupClient initializes a client struct and starts the broadcastHandler and websocket listener. -func setupClient(connection *websocket.Conn, subType SubscriptionType, r *http.Request) { +func setupClient(connection *websocket.Conn, subscriptionType broadcast.SubscriptionType, name string, r *http.Request) { // Extract data from request origConnAddr, _ := r.Context().Value(origConnAddrKey).(string) @@ -298,18 +294,8 @@ func setupClient(connection *websocket.Conn, subType SubscriptionType, r *http.R realIPFromHeader = r.RemoteAddr } - data := clientData{ - userAgent: r.Header.Get("User-Agent"), - connectionIP: hostIP, - connectionPort: hostPort, - realIPFromHeader: realIPFromHeader, - } - - c := newClient(connection, subType, data, config.AppConfig.General.BufferSizes.Websocket) - go c.broadcastHandler() - go c.listenWebsocket() - - ClientHandler.registerClient(c) + c := broadcast.NewWebsocketClient(connection, subscriptionType, name, r.Header.Get("User-Agent"), hostIP, hostPort, realIPFromHeader, config.AppConfig.General.BufferSizes.Websocket) + broadcast.ClientHandler.RegisterClient(c) } // setupWebsocketRoutes configures all the routes necessary for the websocket webserver. @@ -385,8 +371,7 @@ func NewMetricsServer(networkIf string, port int, certPath, keyPath string) *Ser } // NewWebsocketServer starts a new webserver and initialized it with the necessary routes. -// It also starts the broadcaster in ClientHandler as a background job and takes care of -// setting up websocket.Upgrader. +// It also takes care of setting up websocket.Upgrader. func NewWebsocketServer(networkIf string, port int, certPath, keyPath string) *Server { websocketServer := &Server{ networkIf: networkIf, @@ -417,9 +402,6 @@ func NewWebsocketServer(networkIf string, port int, certPath, keyPath string) *S setupWebsocketRoutes(websocketServer.routes) websocketServer.initServer() - ClientHandler.Broadcast = make(chan models.Entry, config.AppConfig.General.BufferSizes.BroadcastManager) - go ClientHandler.broadcaster() - return websocketServer } diff --git a/taskfile.yml b/taskfile.yml index 3fb3fc3..9140c8f 100644 --- a/taskfile.yml +++ b/taskfile.yml @@ -15,6 +15,11 @@ tasks: cmds: - go test -v ./... + run: + desc: Run the server binary. + cmds: + - go run ./cmd/certstream-server-go + build: desc: Build the server binary for your current OS and architecture. cmds: