lixen/logwisp
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.4.3 refactor and minor improvements

Browse Source
This commit is contained in:
Lixen Wraith
2025-09-25 03:18:08 -04:00
parent 94b5f3111e
commit 5ea4dc8f5f
11 changed files with 346 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/internal/config/pipeline.go
View File

@ -123,9 +123,9 @@ func validateSource(pipelineName string, sourceIndex int, cfg *SourceConfig) err
} }
} }
// CHANGED: Validate SSL if present // Validate TLS if present
if ssl, ok := cfg.Options["ssl"].(map[string]any); ok { if tls, ok := cfg.Options["tls"].(map[string]any); ok {
if err := validateSSLOptions("HTTP source", pipelineName, sourceIndex, ssl); err != nil { if err := validateTLSOptions("HTTP source", pipelineName, sourceIndex, tls); err != nil {
return err return err
} }
} }
@ -145,9 +145,9 @@ func validateSource(pipelineName string, sourceIndex int, cfg *SourceConfig) err
} }
} }
// CHANGED: Validate SSL if present // Validate TLS if present
if ssl, ok := cfg.Options["ssl"].(map[string]any); ok { if tls, ok := cfg.Options["tls"].(map[string]any); ok {
if err := validateSSLOptions("TCP source", pipelineName, sourceIndex, ssl); err != nil { if err := validateTLSOptions("TCP source", pipelineName, sourceIndex, tls); err != nil {
return err return err
} }
} }
@ -211,9 +211,9 @@ func validateSink(pipelineName string, sinkIndex int, cfg *SinkConfig, allPorts
} }
} }
// Validate SSL if present // Validate TLS if present
if ssl, ok := cfg.Options["ssl"].(map[string]any); ok { if tls, ok := cfg.Options["tls"].(map[string]any); ok {
if err := validateSSLOptions("HTTP", pipelineName, sinkIndex, ssl); err != nil { if err := validateTLSOptions("HTTP", pipelineName, sinkIndex, tls); err != nil {
return err return err
} }
} }
@ -255,9 +255,9 @@ func validateSink(pipelineName string, sinkIndex int, cfg *SinkConfig, allPorts
} }
} }
// Validate SSL if present // Validate TLS if present
if ssl, ok := cfg.Options["ssl"].(map[string]any); ok { if tls, ok := cfg.Options["tls"].(map[string]any); ok {
if err := validateSSLOptions("TCP", pipelineName, sinkIndex, ssl); err != nil { if err := validateTLSOptions("TCP", pipelineName, sinkIndex, tls); err != nil {
return err return err
} }
} }

8
src/internal/config/server.go
View File

@ -12,8 +12,8 @@ type TCPConfig struct {
Port int64 `toml:"port"` Port int64 `toml:"port"`
BufferSize int64 `toml:"buffer_size"` BufferSize int64 `toml:"buffer_size"`
// SSL/TLS Configuration // TLS Configuration
SSL *SSLConfig `toml:"ssl"` TLS *TLSConfig `toml:"tls"`
// Net limiting // Net limiting
NetLimit *NetLimitConfig `toml:"net_limit"` NetLimit *NetLimitConfig `toml:"net_limit"`
@ -31,8 +31,8 @@ type HTTPConfig struct {
StreamPath string `toml:"stream_path"` StreamPath string `toml:"stream_path"`
StatusPath string `toml:"status_path"` StatusPath string `toml:"status_path"`
// SSL/TLS Configuration // TLS Configuration
SSL *SSLConfig `toml:"ssl"` TLS *TLSConfig `toml:"tls"`
// Nate limiting // Nate limiting
NetLimit *NetLimitConfig `toml:"net_limit"` NetLimit *NetLimitConfig `toml:"net_limit"`

22
src/internal/config/ssl.go → src/internal/config/tls.go
View File

@ -1,4 +1,4 @@
// FILE: logwisp/src/internal/config/ssl.go // FILE: logwisp/src/internal/config/tls.go
package config package config
import ( import (
@ -6,7 +6,7 @@ import (
"os" "os"
) )
type SSLConfig struct { type TLSConfig struct {
Enabled bool `toml:"enabled"` Enabled bool `toml:"enabled"`
CertFile string `toml:"cert_file"` CertFile string `toml:"cert_file"`
KeyFile string `toml:"key_file"` KeyFile string `toml:"key_file"`
@ -30,13 +30,13 @@ type SSLConfig struct {
CipherSuites string `toml:"cipher_suites"` CipherSuites string `toml:"cipher_suites"`
} }
func validateSSLOptions(serverType, pipelineName string, sinkIndex int, ssl map[string]any) error { func validateTLSOptions(serverType, pipelineName string, sinkIndex int, tls map[string]any) error {
if enabled, ok := ssl["enabled"].(bool); ok && enabled { if enabled, ok := tls["enabled"].(bool); ok && enabled {
certFile, certOk := ssl["cert_file"].(string) certFile, certOk := tls["cert_file"].(string)
keyFile, keyOk := ssl["key_file"].(string) keyFile, keyOk := tls["key_file"].(string)
if !certOk || certFile == "" || !keyOk || keyFile == "" { if !certOk || certFile == "" || !keyOk || keyFile == "" {
return fmt.Errorf("pipeline '%s' sink[%d] %s: SSL enabled but cert/key files not specified", return fmt.Errorf("pipeline '%s' sink[%d] %s: TLS enabled but cert/key files not specified",
pipelineName, sinkIndex, serverType) pipelineName, sinkIndex, serverType)
} }
@ -50,8 +50,8 @@ func validateSSLOptions(serverType, pipelineName string, sinkIndex int, ssl map[
pipelineName, sinkIndex, serverType, err) pipelineName, sinkIndex, serverType, err)
} }
if clientAuth, ok := ssl["client_auth"].(bool); ok && clientAuth { if clientAuth, ok := tls["client_auth"].(bool); ok && clientAuth {
caFile, caOk := ssl["client_ca_file"].(string) caFile, caOk := tls["client_ca_file"].(string)
if !caOk || caFile == "" { if !caOk || caFile == "" {
return fmt.Errorf("pipeline '%s' sink[%d] %s: client auth enabled but CA file not specified", return fmt.Errorf("pipeline '%s' sink[%d] %s: client auth enabled but CA file not specified",
pipelineName, sinkIndex, serverType) pipelineName, sinkIndex, serverType)
@ -65,13 +65,13 @@ func validateSSLOptions(serverType, pipelineName string, sinkIndex int, ssl map[
// Validate TLS versions // Validate TLS versions
validVersions := map[string]bool{"TLS1.0": true, "TLS1.1": true, "TLS1.2": true, "TLS1.3": true} validVersions := map[string]bool{"TLS1.0": true, "TLS1.1": true, "TLS1.2": true, "TLS1.3": true}
if minVer, ok := ssl["min_version"].(string); ok && minVer != "" { if minVer, ok := tls["min_version"].(string); ok && minVer != "" {
if !validVersions[minVer] { if !validVersions[minVer] {
return fmt.Errorf("pipeline '%s' sink[%d] %s: invalid min TLS version: %s", return fmt.Errorf("pipeline '%s' sink[%d] %s: invalid min TLS version: %s",
pipelineName, sinkIndex, serverType, minVer) pipelineName, sinkIndex, serverType, minVer)
} }
} }
if maxVer, ok := ssl["max_version"].(string); ok && maxVer != "" { if maxVer, ok := tls["max_version"].(string); ok && maxVer != "" {
if !validVersions[maxVer] { if !validVersions[maxVer] {
return fmt.Errorf("pipeline '%s' sink[%d] %s: invalid max TLS version: %s", return fmt.Errorf("pipeline '%s' sink[%d] %s: invalid max TLS version: %s",
pipelineName, sinkIndex, serverType, maxVer) pipelineName, sinkIndex, serverType, maxVer)

40
src/internal/sink/http.go
View File

@ -63,7 +63,7 @@ type HTTPConfig struct {
StreamPath string StreamPath string
StatusPath string StatusPath string
Heartbeat *config.HeartbeatConfig Heartbeat *config.HeartbeatConfig
SSL *config.SSLConfig TLS *config.TLSConfig
NetLimit *config.NetLimitConfig NetLimit *config.NetLimitConfig
} }
@ -104,29 +104,29 @@ func NewHTTPSink(options map[string]any, logger *log.Logger, formatter format.Fo
} }
} }
// Extract SSL config // Extract TLS config
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
cfg.SSL = &config.SSLConfig{} cfg.TLS = &config.TLSConfig{}
cfg.SSL.Enabled, _ = ssl["enabled"].(bool) cfg.TLS.Enabled, _ = tc["enabled"].(bool)
if certFile, ok := ssl["cert_file"].(string); ok { if certFile, ok := tc["cert_file"].(string); ok {
cfg.SSL.CertFile = certFile cfg.TLS.CertFile = certFile
} }
if keyFile, ok := ssl["key_file"].(string); ok { if keyFile, ok := tc["key_file"].(string); ok {
cfg.SSL.KeyFile = keyFile cfg.TLS.KeyFile = keyFile
} }
cfg.SSL.ClientAuth, _ = ssl["client_auth"].(bool) cfg.TLS.ClientAuth, _ = tc["client_auth"].(bool)
if caFile, ok := ssl["client_ca_file"].(string); ok { if caFile, ok := tc["client_ca_file"].(string); ok {
cfg.SSL.ClientCAFile = caFile cfg.TLS.ClientCAFile = caFile
} }
cfg.SSL.VerifyClientCert, _ = ssl["verify_client_cert"].(bool) cfg.TLS.VerifyClientCert, _ = tc["verify_client_cert"].(bool)
if minVer, ok := ssl["min_version"].(string); ok { if minVer, ok := tc["min_version"].(string); ok {
cfg.SSL.MinVersion = minVer cfg.TLS.MinVersion = minVer
} }
if maxVer, ok := ssl["max_version"].(string); ok { if maxVer, ok := tc["max_version"].(string); ok {
cfg.SSL.MaxVersion = maxVer cfg.TLS.MaxVersion = maxVer
} }
if ciphers, ok := ssl["cipher_suites"].(string); ok { if ciphers, ok := tc["cipher_suites"].(string); ok {
cfg.SSL.CipherSuites = ciphers cfg.TLS.CipherSuites = ciphers
} }
} }
@ -231,7 +231,7 @@ func (h *HTTPSink) Start(ctx context.Context) error {
var err error var err error
if h.tlsManager != nil { if h.tlsManager != nil {
// HTTPS server // HTTPS server
err = h.server.ListenAndServeTLS(addr, h.config.SSL.CertFile, h.config.SSL.KeyFile) err = h.server.ListenAndServeTLS(addr, h.config.TLS.CertFile, h.config.TLS.KeyFile)
} else { } else {
// HTTP server // HTTP server
err = h.server.ListenAndServe(addr) err = h.server.ListenAndServe(addr)

18
src/internal/sink/http_client.go
View File

@ -138,25 +138,25 @@ func NewHTTPClientSink(options map[string]any, logger *log.Logger, formatter for
cfg.CAFile = caFile cfg.CAFile = caFile
} }
// Extract client certificate options from SSL config // Extract client certificate options from TLS config
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
if enabled, _ := ssl["enabled"].(bool); enabled { if enabled, _ := tc["enabled"].(bool); enabled {
// Extract client certificate files for mTLS // Extract client certificate files for mTLS
if certFile, ok := ssl["cert_file"].(string); ok && certFile != "" { if certFile, ok := tc["cert_file"].(string); ok && certFile != "" {
if keyFile, ok := ssl["key_file"].(string); ok && keyFile != "" { if keyFile, ok := tc["key_file"].(string); ok && keyFile != "" {
// These will be used below when configuring TLS // These will be used below when configuring TLS
cfg.CertFile = certFile // Need to add these fields to HTTPClientConfig cfg.CertFile = certFile // Need to add these fields to HTTPClientConfig
cfg.KeyFile = keyFile cfg.KeyFile = keyFile
} }
} }
// Extract CA file from ssl config if not already set // Extract CA file from TLS config if not already set
if cfg.CAFile == "" { if cfg.CAFile == "" {
if caFile, ok := ssl["ca_file"].(string); ok { if caFile, ok := tc["ca_file"].(string); ok {
cfg.CAFile = caFile cfg.CAFile = caFile
} }
} }
// Extract insecure skip verify from ssl config // Extract insecure skip verify from TLS config
if insecure, ok := ssl["insecure_skip_verify"].(bool); ok { if insecure, ok := tc["insecure_skip_verify"].(bool); ok {
cfg.InsecureSkipVerify = insecure cfg.InsecureSkipVerify = insecure
} }
} }

57
src/internal/sink/tcp.go
View File

@ -61,7 +61,7 @@ type TCPConfig struct {
Port int64 Port int64
BufferSize int64 BufferSize int64
Heartbeat *config.HeartbeatConfig Heartbeat *config.HeartbeatConfig
SSL *config.SSLConfig TLS *config.TLSConfig
NetLimit *config.NetLimitConfig NetLimit *config.NetLimitConfig
} }
@ -94,29 +94,29 @@ func NewTCPSink(options map[string]any, logger *log.Logger, formatter format.For
} }
} }
// Extract SSL config // Extract TLS config
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
cfg.SSL = &config.SSLConfig{} cfg.TLS = &config.TLSConfig{}
cfg.SSL.Enabled, _ = ssl["enabled"].(bool) cfg.TLS.Enabled, _ = tc["enabled"].(bool)
if certFile, ok := ssl["cert_file"].(string); ok { if certFile, ok := tc["cert_file"].(string); ok {
cfg.SSL.CertFile = certFile cfg.TLS.CertFile = certFile
} }
if keyFile, ok := ssl["key_file"].(string); ok { if keyFile, ok := tc["key_file"].(string); ok {
cfg.SSL.KeyFile = keyFile cfg.TLS.KeyFile = keyFile
} }
cfg.SSL.ClientAuth, _ = ssl["client_auth"].(bool) cfg.TLS.ClientAuth, _ = tc["client_auth"].(bool)
if caFile, ok := ssl["client_ca_file"].(string); ok { if caFile, ok := tc["client_ca_file"].(string); ok {
cfg.SSL.ClientCAFile = caFile cfg.TLS.ClientCAFile = caFile
} }
cfg.SSL.VerifyClientCert, _ = ssl["verify_client_cert"].(bool) cfg.TLS.VerifyClientCert, _ = tc["verify_client_cert"].(bool)
if minVer, ok := ssl["min_version"].(string); ok { if minVer, ok := tc["min_version"].(string); ok {
cfg.SSL.MinVersion = minVer cfg.TLS.MinVersion = minVer
} }
if maxVer, ok := ssl["max_version"].(string); ok { if maxVer, ok := tc["max_version"].(string); ok {
cfg.SSL.MaxVersion = maxVer cfg.TLS.MaxVersion = maxVer
} }
if ciphers, ok := ssl["cipher_suites"].(string); ok { if ciphers, ok := tc["cipher_suites"].(string); ok {
cfg.SSL.CipherSuites = ciphers cfg.TLS.CipherSuites = ciphers
} }
} }
@ -627,19 +627,6 @@ func (s *tcpServer) OnTraffic(c gnet.Conn) gnet.Action {
return gnet.Close return gnet.Close
} }
// // Check auth timeout
// if !client.authenticated && time.Now().After(client.authTimeout) {
// s.sink.logger.Warn("msg", "Authentication timeout",
// "component", "tcp_sink",
// "remote_addr", c.RemoteAddr().String())
// if client.tlsBridge != nil && client.tlsBridge.IsHandshakeDone() {
// client.tlsBridge.Write([]byte("AUTH TIMEOUT\n"))
// } else if client.tlsBridge == nil {
// c.AsyncWrite([]byte("AUTH TIMEOUT\n"), nil)
// }
// return gnet.Close
// }
// Read all available data // Read all available data
data, err := c.Next(-1) data, err := c.Next(-1)
if err != nil { if err != nil {
@ -801,9 +788,9 @@ func (t *TCPSink) SetAuthConfig(authCfg *config.AuthConfig) {
} }
t.authenticator = authenticator t.authenticator = authenticator
// Initialize TLS manager if SSL is configured // Initialize TLS manager if TLS is configured
if t.config.SSL != nil && t.config.SSL.Enabled { if t.config.TLS != nil && t.config.TLS.Enabled {
tlsManager, err := tls.NewManager(t.config.SSL, t.logger) tlsManager, err := tls.NewManager(t.config.TLS, t.logger)
if err != nil { if err != nil {
t.logger.Error("msg", "Failed to create TLS manager", t.logger.Error("msg", "Failed to create TLS manager",
"component", "tcp_sink", "component", "tcp_sink",

62
src/internal/sink/tcp_client.go
View File

@ -66,7 +66,7 @@ type TCPClientConfig struct {
ReconnectBackoff float64 ReconnectBackoff float64
// TLS config // TLS config
SSL *config.SSLConfig TLS *config.TLSConfig
} }
// NewTCPClientSink creates a new TCP client sink // NewTCPClientSink creates a new TCP client sink
@ -121,25 +121,25 @@ func NewTCPClientSink(options map[string]any, logger *log.Logger, formatter form
cfg.ReconnectBackoff = backoff cfg.ReconnectBackoff = backoff
} }
// Extract SSL config // Extract TLS config
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
cfg.SSL = &config.SSLConfig{} cfg.TLS = &config.TLSConfig{}
cfg.SSL.Enabled, _ = ssl["enabled"].(bool) cfg.TLS.Enabled, _ = tc["enabled"].(bool)
if certFile, ok := ssl["cert_file"].(string); ok { if certFile, ok := tc["cert_file"].(string); ok {
cfg.SSL.CertFile = certFile cfg.TLS.CertFile = certFile
} }
if keyFile, ok := ssl["key_file"].(string); ok { if keyFile, ok := tc["key_file"].(string); ok {
cfg.SSL.KeyFile = keyFile cfg.TLS.KeyFile = keyFile
} }
cfg.SSL.ClientAuth, _ = ssl["client_auth"].(bool) cfg.TLS.ClientAuth, _ = tc["client_auth"].(bool)
if caFile, ok := ssl["client_ca_file"].(string); ok { if caFile, ok := tc["client_ca_file"].(string); ok {
cfg.SSL.ClientCAFile = caFile cfg.TLS.ClientCAFile = caFile
} }
if insecure, ok := ssl["insecure_skip_verify"].(bool); ok { if insecure, ok := tc["insecure_skip_verify"].(bool); ok {
cfg.SSL.InsecureSkipVerify = insecure cfg.TLS.InsecureSkipVerify = insecure
} }
if caFile, ok := ssl["ca_file"].(string); ok { if caFile, ok := tc["ca_file"].(string); ok {
cfg.SSL.CAFile = caFile cfg.TLS.CAFile = caFile
} }
} }
@ -154,11 +154,11 @@ func NewTCPClientSink(options map[string]any, logger *log.Logger, formatter form
t.lastProcessed.Store(time.Time{}) t.lastProcessed.Store(time.Time{})
t.connectionUptime.Store(time.Duration(0)) t.connectionUptime.Store(time.Duration(0))
// Initialize TLS manager if SSL is configured // Initialize TLS manager if TLS is configured
if cfg.SSL != nil && cfg.SSL.Enabled { if cfg.TLS != nil && cfg.TLS.Enabled {
// Build custom TLS config for client // Build custom TLS config for client
t.tlsConfig = &tls.Config{ t.tlsConfig = &tls.Config{
InsecureSkipVerify: cfg.SSL.InsecureSkipVerify, InsecureSkipVerify: cfg.TLS.InsecureSkipVerify,
} }
// Extract server name from address for SNI // Extract server name from address for SNI
@ -169,36 +169,36 @@ func NewTCPClientSink(options map[string]any, logger *log.Logger, formatter form
t.tlsConfig.ServerName = host t.tlsConfig.ServerName = host
// Load custom CA for server verification // Load custom CA for server verification
if cfg.SSL.CAFile != "" { if cfg.TLS.CAFile != "" {
caCert, err := os.ReadFile(cfg.SSL.CAFile) caCert, err := os.ReadFile(cfg.TLS.CAFile)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to read CA file '%s': %w", cfg.SSL.CAFile, err) return nil, fmt.Errorf("failed to read CA file '%s': %w", cfg.TLS.CAFile, err)
} }
caCertPool := x509.NewCertPool() caCertPool := x509.NewCertPool()
if !caCertPool.AppendCertsFromPEM(caCert) { if !caCertPool.AppendCertsFromPEM(caCert) {
return nil, fmt.Errorf("failed to parse CA certificate from '%s'", cfg.SSL.CAFile) return nil, fmt.Errorf("failed to parse CA certificate from '%s'", cfg.TLS.CAFile)
} }
t.tlsConfig.RootCAs = caCertPool t.tlsConfig.RootCAs = caCertPool
logger.Debug("msg", "Custom CA loaded for server verification", logger.Debug("msg", "Custom CA loaded for server verification",
"component", "tcp_client_sink", "component", "tcp_client_sink",
"ca_file", cfg.SSL.CAFile) "ca_file", cfg.TLS.CAFile)
} }
// Load client certificate for mTLS // Load client certificate for mTLS
if cfg.SSL.CertFile != "" && cfg.SSL.KeyFile != "" { if cfg.TLS.CertFile != "" && cfg.TLS.KeyFile != "" {
cert, err := tls.LoadX509KeyPair(cfg.SSL.CertFile, cfg.SSL.KeyFile) cert, err := tls.LoadX509KeyPair(cfg.TLS.CertFile, cfg.TLS.KeyFile)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to load client certificate: %w", err) return nil, fmt.Errorf("failed to load client certificate: %w", err)
} }
t.tlsConfig.Certificates = []tls.Certificate{cert} t.tlsConfig.Certificates = []tls.Certificate{cert}
logger.Info("msg", "Client certificate loaded for mTLS", logger.Info("msg", "Client certificate loaded for mTLS",
"component", "tcp_client_sink", "component", "tcp_client_sink",
"cert_file", cfg.SSL.CertFile) "cert_file", cfg.TLS.CertFile)
} }
// Set minimum TLS version if configured // Set minimum TLS version if configured
if cfg.SSL.MinVersion != "" { if cfg.TLS.MinVersion != "" {
t.tlsConfig.MinVersion = parseTLSVersion(cfg.SSL.MinVersion, tls.VersionTLS12) t.tlsConfig.MinVersion = parseTLSVersion(cfg.TLS.MinVersion, tls.VersionTLS12)
} else { } else {
t.tlsConfig.MinVersion = tls.VersionTLS12 // Default minimum t.tlsConfig.MinVersion = tls.VersionTLS12 // Default minimum
} }
@ -207,8 +207,8 @@ func NewTCPClientSink(options map[string]any, logger *log.Logger, formatter form
"component", "tcp_client_sink", "component", "tcp_client_sink",
"address", cfg.Address, "address", cfg.Address,
"server_name", host, "server_name", host,
"insecure", cfg.SSL.InsecureSkipVerify, "insecure", cfg.TLS.InsecureSkipVerify,
"mtls", cfg.SSL.CertFile != "") "mtls", cfg.TLS.CertFile != "")
} }
return t, nil return t, nil
} }

64
src/internal/source/http.go
View File

@ -4,7 +4,6 @@ package source
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"logwisp/src/internal/tls"
"net" "net"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -13,6 +12,7 @@ import (
"logwisp/src/internal/config" "logwisp/src/internal/config"
"logwisp/src/internal/core" "logwisp/src/internal/core"
"logwisp/src/internal/limit" "logwisp/src/internal/limit"
"logwisp/src/internal/tls"
"github.com/lixenwraith/log" "github.com/lixenwraith/log"
"github.com/valyala/fasthttp" "github.com/valyala/fasthttp"
@ -33,7 +33,7 @@ type HTTPSource struct {
// CHANGED: Add TLS support // CHANGED: Add TLS support
tlsManager *tls.Manager tlsManager *tls.Manager
sslConfig *config.SSLConfig tlsConfig *config.TLSConfig
// Statistics // Statistics
totalEntries atomic.Uint64 totalEntries atomic.Uint64
@ -77,7 +77,7 @@ func NewHTTPSource(options map[string]any, logger *log.Logger) (*HTTPSource, err
Enabled: true, Enabled: true,
} }
if rps, ok := toFloat(rl["requests_per_second"]); ok { if rps, ok := rl["requests_per_second"].(float64); ok {
cfg.RequestsPerSecond = rps cfg.RequestsPerSecond = rps
} }
if burst, ok := rl["burst_size"].(int64); ok { if burst, ok := rl["burst_size"].(int64); ok {
@ -100,35 +100,35 @@ func NewHTTPSource(options map[string]any, logger *log.Logger) (*HTTPSource, err
} }
} }
// Extract SSL config after existing options // Extract TLS config after existing options
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
h.sslConfig = &config.SSLConfig{} h.tlsConfig = &config.TLSConfig{}
h.sslConfig.Enabled, _ = ssl["enabled"].(bool) h.tlsConfig.Enabled, _ = tc["enabled"].(bool)
if certFile, ok := ssl["cert_file"].(string); ok { if certFile, ok := tc["cert_file"].(string); ok {
h.sslConfig.CertFile = certFile h.tlsConfig.CertFile = certFile
} }
if keyFile, ok := ssl["key_file"].(string); ok { if keyFile, ok := tc["key_file"].(string); ok {
h.sslConfig.KeyFile = keyFile h.tlsConfig.KeyFile = keyFile
} }
h.sslConfig.ClientAuth, _ = ssl["client_auth"].(bool) h.tlsConfig.ClientAuth, _ = tc["client_auth"].(bool)
if caFile, ok := ssl["client_ca_file"].(string); ok { if caFile, ok := tc["client_ca_file"].(string); ok {
h.sslConfig.ClientCAFile = caFile h.tlsConfig.ClientCAFile = caFile
} }
h.sslConfig.VerifyClientCert, _ = ssl["verify_client_cert"].(bool) h.tlsConfig.VerifyClientCert, _ = tc["verify_client_cert"].(bool)
h.sslConfig.InsecureSkipVerify, _ = ssl["insecure_skip_verify"].(bool) h.tlsConfig.InsecureSkipVerify, _ = tc["insecure_skip_verify"].(bool)
if minVer, ok := ssl["min_version"].(string); ok { if minVer, ok := tc["min_version"].(string); ok {
h.sslConfig.MinVersion = minVer h.tlsConfig.MinVersion = minVer
} }
if maxVer, ok := ssl["max_version"].(string); ok { if maxVer, ok := tc["max_version"].(string); ok {
h.sslConfig.MaxVersion = maxVer h.tlsConfig.MaxVersion = maxVer
} }
if ciphers, ok := ssl["cipher_suites"].(string); ok { if ciphers, ok := tc["cipher_suites"].(string); ok {
h.sslConfig.CipherSuites = ciphers h.tlsConfig.CipherSuites = ciphers
} }
// Create TLS manager // Create TLS manager
if h.sslConfig.Enabled { if h.tlsConfig.Enabled {
tlsManager, err := tls.NewManager(h.sslConfig, logger) tlsManager, err := tls.NewManager(h.tlsConfig, logger)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to create TLS manager: %w", err) return nil, fmt.Errorf("failed to create TLS manager: %w", err)
} }
@ -173,7 +173,7 @@ func (h *HTTPSource) Start() error {
// Check for TLS manager and start the appropriate server type // Check for TLS manager and start the appropriate server type
if h.tlsManager != nil { if h.tlsManager != nil {
h.server.TLSConfig = h.tlsManager.GetHTTPConfig() h.server.TLSConfig = h.tlsManager.GetHTTPConfig()
err = h.server.ListenAndServeTLS(addr, h.sslConfig.CertFile, h.sslConfig.KeyFile) err = h.server.ListenAndServeTLS(addr, h.tlsConfig.CertFile, h.tlsConfig.KeyFile)
} else { } else {
err = h.server.ListenAndServe(addr) err = h.server.ListenAndServe(addr)
} }
@ -452,18 +452,4 @@ func splitLines(data []byte) [][]byte {
} }
return lines return lines
}
// Helper function for type conversion
func toFloat(v any) (float64, bool) {
switch val := v.(type) {
case float64:
return val, true
case int:
return float64(val), true
case int64:
return float64(val), true
default:
return 0, false
}
} }

32
src/internal/source/tcp.go
View File

@ -43,7 +43,7 @@ type TCPSource struct {
wg sync.WaitGroup wg sync.WaitGroup
netLimiter *limit.NetLimiter netLimiter *limit.NetLimiter
tlsManager *tls.Manager tlsManager *tls.Manager
sslConfig *config.SSLConfig tlsConfig *config.TLSConfig
logger *log.Logger logger *log.Logger
// Statistics // Statistics
@ -83,7 +83,7 @@ func NewTCPSource(options map[string]any, logger *log.Logger) (*TCPSource, error
Enabled: true, Enabled: true,
} }
if rps, ok := toFloat(rl["requests_per_second"]); ok { if rps, ok := rl["requests_per_second"].(float64); ok {
cfg.RequestsPerSecond = rps cfg.RequestsPerSecond = rps
} }
if burst, ok := rl["burst_size"].(int64); ok { if burst, ok := rl["burst_size"].(int64); ok {
@ -103,25 +103,25 @@ func NewTCPSource(options map[string]any, logger *log.Logger) (*TCPSource, error
} }
} }
// Extract SSL config and initialize TLS manager // Extract TLS config and initialize TLS manager
if ssl, ok := options["ssl"].(map[string]any); ok { if tc, ok := options["tls"].(map[string]any); ok {
t.sslConfig = &config.SSLConfig{} t.tlsConfig = &config.TLSConfig{}
t.sslConfig.Enabled, _ = ssl["enabled"].(bool) t.tlsConfig.Enabled, _ = tc["enabled"].(bool)
if certFile, ok := ssl["cert_file"].(string); ok { if certFile, ok := tc["cert_file"].(string); ok {
t.sslConfig.CertFile = certFile t.tlsConfig.CertFile = certFile
} }
if keyFile, ok := ssl["key_file"].(string); ok { if keyFile, ok := tc["key_file"].(string); ok {
t.sslConfig.KeyFile = keyFile t.tlsConfig.KeyFile = keyFile
} }
t.sslConfig.ClientAuth, _ = ssl["client_auth"].(bool) t.tlsConfig.ClientAuth, _ = tc["client_auth"].(bool)
if caFile, ok := ssl["client_ca_file"].(string); ok { if caFile, ok := tc["client_ca_file"].(string); ok {
t.sslConfig.ClientCAFile = caFile t.tlsConfig.ClientCAFile = caFile
} }
t.sslConfig.VerifyClientCert, _ = ssl["verify_client_cert"].(bool) t.tlsConfig.VerifyClientCert, _ = tc["verify_client_cert"].(bool)
// Create TLS manager if enabled // Create TLS manager if enabled
if t.sslConfig.Enabled { if t.tlsConfig.Enabled {
tlsManager, err := tls.NewManager(t.sslConfig, logger) tlsManager, err := tls.NewManager(t.tlsConfig, logger)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to create TLS manager: %w", err) return nil, fmt.Errorf("failed to create TLS manager: %w", err)
} }

240
src/internal/tls/generator.go
View File

@ -92,8 +92,7 @@ func (c *CertGeneratorCommand) Execute(args []string) error {
} }
} }
// Crate and manage private CA // Create and manage private CA
// TODO: Future implementation, not useful without implementation of generateServerCert, generateClientCert
func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits int, certFile, keyFile string) error { func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits int, certFile, keyFile string) error {
// Generate RSA key // Generate RSA key
priv, err := rsa.GenerateKey(rand.Reader, bits) priv, err := rsa.GenerateKey(rand.Reader, bits)
@ -102,8 +101,9 @@ func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits in
} }
// Create certificate template // Create certificate template
serialNumber, _ := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
template := x509.Certificate{ template := x509.Certificate{
SerialNumber: big.NewInt(1), SerialNumber: serialNumber,
Subject: pkix.Name{ Subject: pkix.Name{
Organization: []string{org}, Organization: []string{org},
Country: []string{country}, Country: []string{country},
@ -111,11 +111,9 @@ func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits in
}, },
NotBefore: time.Now(), NotBefore: time.Now(),
NotAfter: time.Now().AddDate(0, 0, days), NotAfter: time.Now().AddDate(0, 0, days),
KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign | x509.KeyUsageDigitalSignature, KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
BasicConstraintsValid: true, BasicConstraintsValid: true,
IsCA: true, IsCA: true,
MaxPathLen: 1,
} }
// Generate certificate // Generate certificate
@ -133,25 +131,12 @@ func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits in
} }
// Save certificate // Save certificate
certOut, err := os.Create(certFile) if err := saveCert(certFile, certDER); err != nil {
if err != nil { return err
return fmt.Errorf("failed to create cert file: %w", err)
} }
defer certOut.Close() if err := saveKey(keyFile, priv); err != nil {
return err
pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: certDER})
// Save private key
keyOut, err := os.OpenFile(keyFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
return fmt.Errorf("failed to create key file: %w", err)
} }
defer keyOut.Close()
pem.Encode(keyOut, &pem.Block{
Type: "RSA PRIVATE KEY",
Bytes: x509.MarshalPKCS1PrivateKey(priv),
})
fmt.Printf("✓ CA certificate generated:\n") fmt.Printf("✓ CA certificate generated:\n")
fmt.Printf(" Certificate: %s\n", certFile) fmt.Printf(" Certificate: %s\n", certFile)
@ -162,7 +147,6 @@ func (c *CertGeneratorCommand) generateCA(cn, org, country string, days, bits in
return nil return nil
} }
// Added parseHosts helper for IP/hostname parsing
func parseHosts(hostList string) ([]string, []net.IP) { func parseHosts(hostList string) ([]string, []net.IP) {
var dnsNames []string var dnsNames []string
var ipAddrs []net.IP var ipAddrs []net.IP
@ -196,11 +180,7 @@ func (c *CertGeneratorCommand) generateSelfSigned(cn, org, country, hosts string
dnsNames, ipAddrs := parseHosts(hosts) dnsNames, ipAddrs := parseHosts(hosts)
// 3. Create the certificate template // 3. Create the certificate template
serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128) serialNumber, _ := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
if err != nil {
return fmt.Errorf("failed to generate serial number: %w", err)
}
template := x509.Certificate{ template := x509.Certificate{
SerialNumber: serialNumber, SerialNumber: serialNumber,
@ -235,25 +215,14 @@ func (c *CertGeneratorCommand) generateSelfSigned(cn, org, country, hosts string
} }
// 6. Save the certificate with 0644 permissions // 6. Save the certificate with 0644 permissions
certOut, err := os.Create(certFile) if err := saveCert(certFile, certDER); err != nil {
if err != nil { return err
return fmt.Errorf("failed to create certificate file: %w", err)
} }
pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: certDER}) if err := saveKey(keyFile, priv); err != nil {
certOut.Close() return err
// 7. Save the private key with 0600 permissions
keyOut, err := os.OpenFile(keyFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
return fmt.Errorf("failed to create key file: %w", err)
} }
pem.Encode(keyOut, &pem.Block{
Type: "RSA PRIVATE KEY",
Bytes: x509.MarshalPKCS1PrivateKey(priv),
})
keyOut.Close()
// 8. Print summary // 7. Print summary
fmt.Printf("\n✓ Self-signed certificate generated:\n") fmt.Printf("\n✓ Self-signed certificate generated:\n")
fmt.Printf(" Certificate: %s\n", certFile) fmt.Printf(" Certificate: %s\n", certFile)
fmt.Printf(" Private Key: %s (mode 0600)\n", keyFile) fmt.Printf(" Private Key: %s (mode 0600)\n", keyFile)
@ -266,10 +235,187 @@ func (c *CertGeneratorCommand) generateSelfSigned(cn, org, country, hosts string
return nil return nil
} }
// Generate server cert with CA
func (c *CertGeneratorCommand) generateServerCert(cn, org, country, hosts, caFile, caKeyFile string, days, bits int, certFile, keyFile string) error { func (c *CertGeneratorCommand) generateServerCert(cn, org, country, hosts, caFile, caKeyFile string, days, bits int, certFile, keyFile string) error {
return fmt.Errorf("server certificate generation with CA is not implemented; use --self-signed instead") caCert, caKey, err := loadCA(caFile, caKeyFile)
if err != nil {
return err
}
priv, err := rsa.GenerateKey(rand.Reader, bits)
if err != nil {
return fmt.Errorf("failed to generate server private key: %w", err)
}
dnsNames, ipAddrs := parseHosts(hosts)
serialNumber, _ := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
certExpiry := time.Now().AddDate(0, 0, days)
if certExpiry.After(caCert.NotAfter) {
return fmt.Errorf("certificate validity period (%d days) exceeds CA expiry (%s)", days, caCert.NotAfter.Format(time.RFC3339))
}
template := x509.Certificate{
SerialNumber: serialNumber,
Subject: pkix.Name{
CommonName: cn,
Organization: []string{org},
Country: []string{country},
},
NotBefore: time.Now(),
NotAfter: certExpiry,
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
DNSNames: dnsNames,
IPAddresses: ipAddrs,
}
certDER, err := x509.CreateCertificate(rand.Reader, &template, caCert, &priv.PublicKey, caKey)
if err != nil {
return fmt.Errorf("failed to sign server certificate: %w", err)
}
if certFile == "" {
certFile = "server.crt"
}
if keyFile == "" {
keyFile = "server.key"
}
if err := saveCert(certFile, certDER); err != nil {
return err
}
if err := saveKey(keyFile, priv); err != nil {
return err
}
fmt.Printf("\n✓ Server certificate generated:\n")
fmt.Printf(" Certificate: %s\n", certFile)
fmt.Printf(" Private Key: %s (mode 0600)\n", keyFile)
fmt.Printf(" Signed by: CN=%s\n", caCert.Subject.CommonName)
if len(hosts) > 0 {
fmt.Printf(" Hosts (SANs): %s\n", hosts)
}
return nil
} }
// Generate client cert with CA
func (c *CertGeneratorCommand) generateClientCert(cn, org, country, caFile, caKeyFile string, days, bits int, certFile, keyFile string) error { func (c *CertGeneratorCommand) generateClientCert(cn, org, country, caFile, caKeyFile string, days, bits int, certFile, keyFile string) error {
return fmt.Errorf("client certificate generation with CA is not implemented; use --self-signed instead") caCert, caKey, err := loadCA(caFile, caKeyFile)
if err != nil {
return err
}
priv, err := rsa.GenerateKey(rand.Reader, bits)
if err != nil {
return fmt.Errorf("failed to generate client private key: %w", err)
}
serialNumber, _ := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
certExpiry := time.Now().AddDate(0, 0, days)
if certExpiry.After(caCert.NotAfter) {
return fmt.Errorf("certificate validity period (%d days) exceeds CA expiry (%s)", days, caCert.NotAfter.Format(time.RFC3339))
}
template := x509.Certificate{
SerialNumber: serialNumber,
Subject: pkix.Name{
CommonName: cn,
Organization: []string{org},
Country: []string{country},
},
NotBefore: time.Now(),
NotAfter: certExpiry,
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
}
certDER, err := x509.CreateCertificate(rand.Reader, &template, caCert, &priv.PublicKey, caKey)
if err != nil {
return fmt.Errorf("failed to sign client certificate: %w", err)
}
if certFile == "" {
certFile = "client.crt"
}
if keyFile == "" {
keyFile = "client.key"
}
if err := saveCert(certFile, certDER); err != nil {
return err
}
if err := saveKey(keyFile, priv); err != nil {
return err
}
fmt.Printf("\n✓ Client certificate generated:\n")
fmt.Printf(" Certificate: %s\n", certFile)
fmt.Printf(" Private Key: %s (mode 0600)\n", keyFile)
fmt.Printf(" Signed by: CN=%s\n", caCert.Subject.CommonName)
return nil
}
// Load cert with CA
func loadCA(caFile, caKeyFile string) (*x509.Certificate, *rsa.PrivateKey, error) {
if caFile == "" || caKeyFile == "" {
return nil, nil, fmt.Errorf("--ca-cert and --ca-key are required for signing")
}
caCertPEM, err := os.ReadFile(caFile)
if err != nil {
return nil, nil, fmt.Errorf("failed to read CA certificate: %w", err)
}
caCertBlock, _ := pem.Decode(caCertPEM)
if caCertBlock == nil {
return nil, nil, fmt.Errorf("failed to decode CA certificate PEM")
}
caCert, err := x509.ParseCertificate(caCertBlock.Bytes)
if err != nil {
return nil, nil, fmt.Errorf("failed to parse CA certificate: %w", err)
}
if !caCert.IsCA {
return nil, nil, fmt.Errorf("provided certificate is not a valid CA")
}
caKeyPEM, err := os.ReadFile(caKeyFile)
if err != nil {
return nil, nil, fmt.Errorf("failed to read CA key: %w", err)
}
caKeyBlock, _ := pem.Decode(caKeyPEM)
if caKeyBlock == nil {
return nil, nil, fmt.Errorf("failed to decode CA key PEM")
}
caKey, err := x509.ParsePKCS1PrivateKey(caKeyBlock.Bytes)
if err != nil {
return nil, nil, fmt.Errorf("failed to parse CA private key: %w", err)
}
// Verify key matches certificate
if caCert.PublicKey.(*rsa.PublicKey).N.Cmp(caKey.N) != 0 {
return nil, nil, fmt.Errorf("CA private key does not match CA certificate")
}
return caCert, caKey, nil
}
func saveCert(filename string, derBytes []byte) error {
certOut, err := os.Create(filename)
if err != nil {
return fmt.Errorf("failed to create cert file %s: %w", filename, err)
}
defer certOut.Close()
return pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes})
}
func saveKey(filename string, key *rsa.PrivateKey) error {
keyOut, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
return fmt.Errorf("failed to create key file %s: %w", filename, err)
}
defer keyOut.Close()
return pem.Encode(keyOut, &pem.Block{
Type: "RSA PRIVATE KEY",
Bytes: x509.MarshalPKCS1PrivateKey(key),
})
} }

7
src/internal/tls/manager.go
View File

@ -15,13 +15,13 @@ import (
// Manager handles TLS configuration for servers // Manager handles TLS configuration for servers
type Manager struct { type Manager struct {
config *config.SSLConfig config *config.TLSConfig
tlsConfig *tls.Config tlsConfig *tls.Config
logger *log.Logger logger *log.Logger
} }
// NewManager creates a TLS configuration from SSL config // NewManager creates a TLS configuration from TLS config
func NewManager(cfg *config.SSLConfig, logger *log.Logger) (*Manager, error) { func NewManager(cfg *config.TLSConfig, logger *log.Logger) (*Manager, error) {
if cfg == nil || !cfg.Enabled { if cfg == nil || !cfg.Enabled {
return nil, nil return nil, nil
} }
@ -83,7 +83,6 @@ func NewManager(cfg *config.SSLConfig, logger *log.Logger) (*Manager, error) {
} }
// Set secure defaults // Set secure defaults
m.tlsConfig.PreferServerCipherSuites = true
m.tlsConfig.SessionTicketsDisabled = false m.tlsConfig.SessionTicketsDisabled = false
m.tlsConfig.Renegotiation = tls.RenegotiateNever m.tlsConfig.Renegotiation = tls.RenegotiateNever