diff --git a/Makefile b/Makefile index 0a664c1..a77870b 100644 --- a/Makefile +++ b/Makefile @@ -21,7 +21,7 @@ test-e2e: .PHONY: build clean test run-node run-node2 run-node3 run-example deps tidy fmt vet lint clear-ports -VERSION := 0.51.2-beta +VERSION := 0.51.3-beta COMMIT ?= $(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) DATE ?= $(shell date -u +%Y-%m-%dT%H:%M:%SZ) LDFLAGS := -X 'main.version=$(VERSION)' -X 'main.commit=$(COMMIT)' -X 'main.date=$(DATE)' diff --git a/README.md b/README.md index ef538c1..8f447fa 100644 --- a/README.md +++ b/README.md @@ -185,20 +185,21 @@ sudo journalctl -u debros-node.service -f ```yaml node: id: "" + type: "bootstrap" listen_addresses: - "/ip4/0.0.0.0/tcp/4001" data_dir: "./data/bootstrap" max_connections: 100 database: - data_dir: "./data/db" + data_dir: "./data/bootstrap/rqlite" replication_factor: 3 shard_count: 16 max_database_size: 1073741824 backup_interval: 24h rqlite_port: 5001 rqlite_raft_port: 7001 - rqlite_join_address: "" # Bootstrap node does not join + rqlite_join_address: "" discovery: bootstrap_peers: [] @@ -223,20 +224,21 @@ logging: ```yaml node: id: "node2" + type: "node" listen_addresses: - "/ip4/0.0.0.0/tcp/4002" data_dir: "./data/node2" max_connections: 50 database: - data_dir: "./data/db" + data_dir: "./data/node2/rqlite" replication_factor: 3 shard_count: 16 max_database_size: 1073741824 backup_interval: 24h rqlite_port: 5002 rqlite_raft_port: 7002 - rqlite_join_address: "http://127.0.0.1:5001" + rqlite_join_address: "127.0.0.1:7001" discovery: bootstrap_peers: @@ -280,7 +282,7 @@ database: - backup_interval (duration) e.g., "24h". Default: 24h. - rqlite_port (int) RQLite HTTP API port. Default: 5001. - rqlite_raft_port (int) RQLite Raft port. Default: 7001. -- rqlite_join_address (string) HTTP address of an existing RQLite node to join. Empty for bootstrap. +- rqlite_join_address (string) Raft address of an existing RQLite node to join (host:port format). Empty for bootstrap. discovery: @@ -310,25 +312,25 @@ Example node.yaml ```yaml node: id: "node2" + type: "node" listen_addresses: - "/ip4/0.0.0.0/tcp/4002" data_dir: "./data/node2" max_connections: 50 - disable_anonrc: true database: - data_dir: "./data/db" + data_dir: "./data/node2/rqlite" replication_factor: 3 shard_count: 16 max_database_size: 1073741824 backup_interval: 24h - rqlite_port: 5001 - rqlite_raft_port: 7001 - rqlite_join_address: "http://127.0.0.1:5001" + rqlite_port: 5002 + rqlite_raft_port: 7002 + rqlite_join_address: "127.0.0.1:7001" discovery: bootstrap_peers: - - "" + - "/ip4/127.0.0.1/tcp/4001/p2p/" discovery_interval: 15s bootstrap_port: 4001 http_adv_address: "127.0.0.1" @@ -339,7 +341,6 @@ security: enable_tls: false private_key_file: "" certificate_file: "" - auth_enabled: false logging: level: "info" @@ -382,6 +383,64 @@ bootstrap_peers: - **Database endpoints**: Set in config or via `RQLITE_NODES` env var. - **Development mode**: Use `NETWORK_DEV_LOCAL=1` for localhost defaults. +### Configuration Validation + +DeBros Network performs strict validation of all configuration files at startup. This ensures invalid configurations are caught immediately rather than causing silent failures later. + +#### Validation Features + +- **Strict YAML Parsing:** Unknown configuration keys are rejected with helpful error messages +- **Format Validation:** Multiaddrs, ports, durations, and other formats are validated for correctness +- **Cross-Field Validation:** Configuration constraints (e.g., bootstrap nodes don't join clusters) are enforced +- **Aggregated Error Reporting:** All validation errors are reported together, not one-by-one + +#### Common Validation Errors + +**Missing or Invalid `node.type`** +``` +node.type: must be one of [bootstrap node]; got "invalid" +``` +Solution: Set `type: "bootstrap"` or `type: "node"` + +**Invalid Bootstrap Peer Format** +``` +discovery.bootstrap_peers[0]: invalid multiaddr; expected /ip{4,6}/.../tcp//p2p/ +discovery.bootstrap_peers[0]: missing /p2p/ component +``` +Solution: Use full multiaddr format: `/ip4/127.0.0.1/tcp/4001/p2p/12D3KooW...` + +**Port Conflicts** +``` +database.rqlite_raft_port: must differ from database.rqlite_port (5001) +``` +Solution: Use different ports for HTTP and Raft (e.g., 5001 and 7001) + +**RQLite Join Address Issues (Nodes)** +``` +database.rqlite_join_address: required for node type (non-bootstrap) +database.rqlite_join_address: invalid format; expected host:port +``` +Solution: Non-bootstrap nodes must specify where to join the cluster. Use Raft port: `127.0.0.1:7001` + +**Bootstrap Nodes Cannot Join** +``` +database.rqlite_join_address: must be empty for bootstrap type +``` +Solution: Bootstrap nodes should have `rqlite_join_address: ""` + +**Invalid Listen Addresses** +``` +node.listen_addresses[0]: invalid TCP port 99999; port must be between 1 and 65535 +``` +Solution: Use valid ports [1-65535], e.g., `/ip4/0.0.0.0/tcp/4001` + +**Unknown Configuration Keys** +``` +invalid config: yaml: unmarshal errors: + line 42: field migrations_path not found in type config.DatabaseConfig +``` +Solution: Remove unsupported keys. Supported keys are documented in the YAML Reference section above. + --- ## CLI Usage @@ -744,8 +803,6 @@ ws.send("Hello, network!"); ## Development - - ### Project Structure ``` @@ -923,195 +980,4 @@ if err := client.FindOneBy(ctx, &u, "users", map[string]any{"email": "alice@exam `Save` inserts if PK is zero, otherwise updates by PK. `Remove` deletes by PK. -```go -// Insert (ID is zero) -u := &User{Email: "bob@example.com", FirstName: "Bob"} -_ = client.Save(ctx, u) // INSERT; sets u.ID if autoincrement - -// Update (ID is non-zero) -u.FirstName = "Bobby" -_ = client.Save(ctx, u) // UPDATE ... WHERE id = ? - -// Remove -_ = client.Remove(ctx, u) // DELETE ... WHERE id = ? - -``` - -### transactions - -Run multiple operations atomically. If your function returns an error, the transaction is rolled back; otherwise it commits. - -```go -err := client.Tx(ctx, func(tx rqlite.Tx) error { - // Read inside the same transaction - var me User - if err := tx.Query(ctx, &me, "SELECT * FROM users WHERE id = ?", 1); err != nil { - return err - } - - // Write inside the same transaction - me.LastName = "Updated" - if err := tx.Save(ctx, &me); err != nil { - return err - } - - // Complex query via builder - var recent []User - if err := tx.CreateQueryBuilder("users"). - OrderBy("created_at DESC"). - Limit(5). - GetMany(ctx, &recent); err != nil { - return err - } - - return nil // commit -}) - -``` - -### Repositories (optional, generic) - -Strongly-typed convenience layer bound to a table: - -```go -repo := client.Repository[User]("users") - -var many []User -_ = repo.Find(ctx, &many, map[string]any{"last_name": "A"}, rqlite.WithOrderBy("created_at DESC"), rqlite.WithLimit(10)) - -var one User -_ = repo.FindOne(ctx, &one, map[string]any{"email": "alice@example.com"}) - -_ = repo.Save(ctx, &one) -_ = repo.Remove(ctx, &one) - -``` - -### Migrations - -Option A: From the node (after rqlite is ready) - -```go -ctx := context.Background() -dirs := []string{ - "network/migrations", // default - "path/to/your/app/migrations", // extra -} - -if err := rqliteManager.ApplyMigrationsDirs(ctx, dirs); err != nil { - logger.Fatal("apply migrations failed", zap.Error(err)) -} -``` - -Option B: Using the adapter sql.DB - -```go -ctx := context.Background() -db := adapter.GetSQLDB() -dirs := []string{"network/migrations", "app/migrations"} - -if err := rqlite.ApplyMigrationsDirs(ctx, db, dirs, logger); err != nil { - logger.Fatal("apply migrations failed", zap.Error(err)) -} -``` - ---- - -## Troubleshooting - -### Common Issues - -#### Bootstrap Connection Failed - -- **Symptoms:** `Failed to connect to bootstrap peer` -- **Solutions:** Check node is running, firewall settings, peer ID validity. - -#### Database Operations Timeout - -- **Symptoms:** `Query timeout` or `No RQLite connection available` -- **Solutions:** Ensure RQLite ports are open, leader election completed, cluster join config correct. - -#### Message Delivery Failures - -- **Symptoms:** Messages not received by subscribers -- **Solutions:** Verify topic names, active subscriptions, network connectivity. - -#### High Memory Usage - -- **Symptoms:** Memory usage grows continuously -- **Solutions:** Unsubscribe when done, monitor connection pool, review message retention. - -#### Authentication Issues - -- **Symptoms:** `Authentication failed`, `Invalid wallet signature`, `JWT token expired` -- **Solutions:** - - Check wallet signature format (65-byte r||s||v hex) - - Ensure nonce matches exactly during wallet verification - - Verify wallet address case-insensitivity - - Use refresh endpoint or re-authenticate for expired tokens - - Clear credential cache if multi-wallet conflicts occur: `rm -rf ~/.debros/credentials` - -#### Gateway Issues - -- **Symptoms:** `Gateway connection refused`, `CORS errors`, `WebSocket disconnections` -- **Solutions:** - - Verify gateway is running and accessible on configured port - - Check CORS configuration for web applications - - Ensure proper authentication headers for protected endpoints - - Verify namespace configuration and enforcement - -#### Database Migration Issues - -- **Symptoms:** `Migration failed`, `SQL syntax error`, `Version conflict` -- **Solutions:** - - Check SQL syntax in migration files - - Ensure proper statement termination - - Verify migration file naming and sequential order - - Review migration logs for transaction rollbacks - -### Debugging & Health Checks - -```bash -export LOG_LEVEL=debug -./bin/network-cli health -./bin/network-cli peers -./bin/network-cli query "SELECT 1" -./bin/network-cli pubsub publish test "hello" -./bin/network-cli pubsub subscribe test 10s - -# Gateway health checks -curl http://localhost:6001/health -curl http://localhost:6001/v1/status -``` - -### Service Logs - -```bash -# Node service logs -sudo journalctl -u debros-node.service --since "1 hour ago" - -# Gateway service logs (if running as service) -sudo journalctl -u debros-gateway.service --since "1 hour ago" - -# Application logs -tail -f ./logs/gateway.log -tail -f ./logs/node.log -``` - ---- - -## License - -Distributed under the MIT License. See [LICENSE](LICENSE) for details. - ---- - -## Further Reading - -- [DeBros Network Documentation](https://network.debros.io/docs/) -- [RQLite Documentation](https://github.com/rqlite/rqlite) -- [LibP2P Documentation](https://libp2p.io) - ---- - -_This README reflects the latest architecture, configuration, and operational practices for the DeBros Network. For questions or contributions, please open an issue or pull request._ +``` \ No newline at end of file diff --git a/cmd/gateway/config.go b/cmd/gateway/config.go index acfbdcc..a32619e 100644 --- a/cmd/gateway/config.go +++ b/cmd/gateway/config.go @@ -2,13 +2,14 @@ package main import ( "flag" + "fmt" "os" "strings" + "github.com/DeBrosOfficial/network/pkg/config" "github.com/DeBrosOfficial/network/pkg/gateway" "github.com/DeBrosOfficial/network/pkg/logging" "go.uber.org/zap" - "gopkg.in/yaml.v3" ) // For transition, alias main.GatewayConfig to pkg/gateway.Config @@ -59,30 +60,32 @@ func parseGatewayConfig(logger *logging.ColoredLogger) *gateway.Config { const path = "configs/gateway.yaml" if data, err := os.ReadFile(path); err == nil { var y yamlCfg - if err := yaml.Unmarshal(data, &y); err != nil { - logger.ComponentWarn(logging.ComponentGeneral, "failed to parse configs/gateway.yaml; ignoring", zap.Error(err)) - } else { - if v := strings.TrimSpace(y.ListenAddr); v != "" { - cfg.ListenAddr = v - } - if v := strings.TrimSpace(y.ClientNamespace); v != "" { - cfg.ClientNamespace = v - } - if v := strings.TrimSpace(y.RQLiteDSN); v != "" { - cfg.RQLiteDSN = v - } - if len(y.BootstrapPeers) > 0 { - var bp []string - for _, p := range y.BootstrapPeers { - p = strings.TrimSpace(p) - if p != "" { - bp = append(bp, p) - } - } - if len(bp) > 0 { - cfg.BootstrapPeers = bp + // Use strict YAML decoding to reject unknown fields + if err := config.DecodeStrict(strings.NewReader(string(data)), &y); err != nil { + logger.ComponentError(logging.ComponentGeneral, "failed to parse configs/gateway.yaml", zap.Error(err)) + fmt.Fprintf(os.Stderr, "Configuration load error: %v\n", err) + os.Exit(1) + } + if v := strings.TrimSpace(y.ListenAddr); v != "" { + cfg.ListenAddr = v + } + if v := strings.TrimSpace(y.ClientNamespace); v != "" { + cfg.ClientNamespace = v + } + if v := strings.TrimSpace(y.RQLiteDSN); v != "" { + cfg.RQLiteDSN = v + } + if len(y.BootstrapPeers) > 0 { + var bp []string + for _, p := range y.BootstrapPeers { + p = strings.TrimSpace(p) + if p != "" { + bp = append(bp, p) } } + if len(bp) > 0 { + cfg.BootstrapPeers = bp + } } } } @@ -135,6 +138,16 @@ func parseGatewayConfig(logger *logging.ColoredLogger) *gateway.Config { cfg.BootstrapPeers = bp } + // Validate configuration + if errs := cfg.ValidateConfig(); len(errs) > 0 { + fmt.Fprintf(os.Stderr, "\nGateway configuration errors (%d):\n", len(errs)) + for _, err := range errs { + fmt.Fprintf(os.Stderr, " - %s\n", err) + } + fmt.Fprintf(os.Stderr, "\nPlease fix the configuration and try again.\n") + os.Exit(1) + } + logger.ComponentInfo(logging.ComponentGeneral, "Loaded gateway configuration", zap.String("addr", cfg.ListenAddr), zap.String("namespace", cfg.ClientNamespace), diff --git a/cmd/node/main.go b/cmd/node/main.go index 5c9f30c..70646ef 100644 --- a/cmd/node/main.go +++ b/cmd/node/main.go @@ -4,8 +4,6 @@ import ( "context" "flag" "fmt" - "io/ioutil" - "log" "os" "os/signal" "path/filepath" @@ -15,7 +13,6 @@ import ( "github.com/DeBrosOfficial/network/pkg/logging" "github.com/DeBrosOfficial/network/pkg/node" "go.uber.org/zap" - "gopkg.in/yaml.v3" ) // setup_logger initializes a logger for the given component. @@ -24,16 +21,15 @@ func setup_logger(component logging.Component) (logger *logging.ColoredLogger) { logger, err = logging.NewColoredLogger(component, true) if err != nil { - log.Fatalf("Failed to create logger: %v", err) + fmt.Fprintf(os.Stderr, "Failed to create logger: %v\n", err) + os.Exit(1) } return logger } -// parse_and_return_network_flags it initializes all the network flags coming from the .yaml files -func parse_and_return_network_flags() (configPath *string, dataDir, nodeID *string, p2pPort, rqlHTTP, rqlRaft *int, rqlJoinAddr *string, advAddr *string, help *bool) { - logger := setup_logger(logging.ComponentNode) - +// parse_flags parses command-line flags and returns them. +func parse_flags() (configPath, dataDir, nodeID *string, p2pPort, rqlHTTP, rqlRaft *int, rqlJoinAddr, advAddr *string, help *bool) { configPath = flag.String("config", "", "Path to config YAML file (overrides defaults)") dataDir = flag.String("data", "", "Data directory (auto-detected if not provided)") nodeID = flag.String("id", "", "Node identifier (for running multiple local nodes)") @@ -45,54 +41,20 @@ func parse_and_return_network_flags() (configPath *string, dataDir, nodeID *stri help = flag.Bool("help", false, "Show help") flag.Parse() - logger.Info("Successfully parsed all flags and arguments.") - - if *configPath != "" { - cfg, err := LoadConfigFromYAML(*configPath) - if err != nil { - logger.Error("Failed to load config from YAML", zap.Error(err)) - os.Exit(1) - } - logger.ComponentInfo(logging.ComponentNode, "Configuration loaded from YAML file", zap.String("path", *configPath)) - - // Instead of returning flag values, return config values - // For ListenAddresses, extract port from multiaddr string if possible, else use default - var p2pPortVal int - if len(cfg.Node.ListenAddresses) > 0 { - // Try to parse port from multiaddr string - var port int - _, err := fmt.Sscanf(cfg.Node.ListenAddresses[0], "/ip4/0.0.0.0/tcp/%d", &port) - if err == nil { - p2pPortVal = port - } else { - p2pPortVal = 4001 - } - } else { - p2pPortVal = 4001 - } - return configPath, - &cfg.Node.DataDir, - &cfg.Node.ID, - &p2pPortVal, - &cfg.Database.RQLitePort, - &cfg.Database.RQLiteRaftPort, - &cfg.Database.RQLiteJoinAddress, - &cfg.Discovery.HttpAdvAddress, - help - } - return } -// LoadConfigFromYAML loads a config from a YAML file +// LoadConfigFromYAML loads a config from a YAML file using strict decoding. func LoadConfigFromYAML(path string) (*config.Config, error) { - data, err := ioutil.ReadFile(path) + file, err := os.Open(path) if err != nil { - return nil, fmt.Errorf("failed to read config file: %w", err) + return nil, fmt.Errorf("failed to open config file: %w", err) } + defer file.Close() + var cfg config.Config - if err := yaml.Unmarshal(data, &cfg); err != nil { - return nil, fmt.Errorf("failed to unmarshal YAML: %w", err) + if err := config.DecodeStrict(file, &cfg); err != nil { + return nil, err } return &cfg, nil } @@ -101,7 +63,7 @@ func LoadConfigFromYAML(path string) (*config.Config, error) { func check_if_should_open_help(help *bool) { if *help { flag.Usage() - return + os.Exit(0) } } @@ -115,7 +77,9 @@ func select_data_dir(dataDir *string, nodeID *string, hasConfigFile bool) { os.Exit(1) } - logger.Info("Successfully selected Data Directory of: %s", zap.String("dataDir", *dataDir)) + if *dataDir != "" { + logger.Info("Data directory selected: %s", zap.String("dataDir", *dataDir)) + } } // startNode starts the node with the given configuration and port @@ -125,10 +89,12 @@ func startNode(ctx context.Context, cfg *config.Config, port int) error { n, err := node.NewNode(cfg) if err != nil { logger.Error("failed to create node: %v", zap.Error(err)) + return err } if err := n.Start(ctx); err != nil { logger.Error("failed to start node: %v", zap.Error(err)) + return err } // Save the peer ID to a file for CLI access (especially useful for bootstrap) @@ -152,8 +118,8 @@ func startNode(ctx context.Context, cfg *config.Config, port int) error { return n.Stop() } -// load_args_into_config applies command line argument overrides to the config -func load_args_into_config(cfg *config.Config, p2pPort, rqlHTTP, rqlRaft *int, rqlJoinAddr *string, advAddr *string, dataDir *string) { +// apply_flag_overrides applies command line argument overrides to the config +func apply_flag_overrides(cfg *config.Config, p2pPort, rqlHTTP, rqlRaft *int, rqlJoinAddr *string, advAddr *string, dataDir *string) { logger := setup_logger(logging.ComponentNode) // Apply RQLite HTTP port override @@ -183,8 +149,8 @@ func load_args_into_config(cfg *config.Config, p2pPort, rqlHTTP, rqlRaft *int, r } if *advAddr != "" { - cfg.Discovery.HttpAdvAddress = fmt.Sprintf("%s:%d", *advAddr, *rqlHTTP) - cfg.Discovery.RaftAdvAddress = fmt.Sprintf("%s:%d", *advAddr, *rqlRaft) + cfg.Discovery.HttpAdvAddress = fmt.Sprintf("%s:%d", *advAddr, cfg.Database.RQLitePort) + cfg.Discovery.RaftAdvAddress = fmt.Sprintf("%s:%d", *advAddr, cfg.Database.RQLiteRaftPort) } if *dataDir != "" { @@ -192,23 +158,52 @@ func load_args_into_config(cfg *config.Config, p2pPort, rqlHTTP, rqlRaft *int, r } } +// printValidationErrors prints aggregated validation errors and exits. +func printValidationErrors(errs []error) { + fmt.Fprintf(os.Stderr, "\nConfiguration errors (%d):\n", len(errs)) + for _, err := range errs { + fmt.Fprintf(os.Stderr, " - %s\n", err) + } + fmt.Fprintf(os.Stderr, "\nPlease fix the configuration and try again.\n") + os.Exit(1) +} + func main() { logger := setup_logger(logging.ComponentNode) - configPath, dataDir, nodeID, p2pPort, rqlHTTP, rqlRaft, rqlJoinAddr, advAddr, help := parse_and_return_network_flags() + // Parse command-line flags + configPath, dataDir, nodeID, p2pPort, rqlHTTP, rqlRaft, rqlJoinAddr, advAddr, help := parse_flags() check_if_should_open_help(help) select_data_dir(dataDir, nodeID, *configPath != "") - // Load Node Configuration + // Load configuration var cfg *config.Config - cfg = config.DefaultConfig() - logger.ComponentInfo(logging.ComponentNode, "Default configuration loaded successfully") + if *configPath != "" { + // Load from YAML with strict decoding + var err error + cfg, err = LoadConfigFromYAML(*configPath) + if err != nil { + logger.Error("Failed to load config from YAML", zap.Error(err)) + fmt.Fprintf(os.Stderr, "Configuration load error: %v\n", err) + os.Exit(1) + } + logger.ComponentInfo(logging.ComponentNode, "Configuration loaded from YAML file", zap.String("path", *configPath)) + } else { + // Use default configuration + cfg = config.DefaultConfig() + logger.ComponentInfo(logging.ComponentNode, "Default configuration loaded successfully") + } - // Apply command line argument overrides - load_args_into_config(cfg, p2pPort, rqlHTTP, rqlRaft, rqlJoinAddr, advAddr, dataDir) + // Apply command-line flag overrides + apply_flag_overrides(cfg, p2pPort, rqlHTTP, rqlRaft, rqlJoinAddr, advAddr, dataDir) logger.ComponentInfo(logging.ComponentNode, "Command line arguments applied to configuration") + // Validate configuration + if errs := cfg.Validate(); len(errs) > 0 { + printValidationErrors(errs) + } + // LibP2P uses configurable port (default 4001); RQLite uses 5001 (HTTP) and 7001 (Raft) port := *p2pPort @@ -219,7 +214,7 @@ func main() { zap.Int("p2p_port", port), zap.Strings("bootstrap_peers", cfg.Discovery.BootstrapPeers), zap.String("rqlite_join_address", cfg.Database.RQLiteJoinAddress), - zap.String("data_directory", *dataDir)) + zap.String("data_directory", cfg.Node.DataDir)) // Create context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) diff --git a/pkg/config/config.go b/pkg/config/config.go index 94da2c7..85e595d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -109,10 +109,11 @@ func DefaultConfig() *Config { }, Discovery: DiscoveryConfig{ BootstrapPeers: []string{ - "/ip4/217.76.54.178/tcp/4001/p2p/12D3KooWKZnirPwNT4URtNSWK45f6vLkEs4xyUZ792F8Uj1oYnm1", - "/ip4/51.83.128.181/tcp/4001/p2p/12D3KooWBn2Zf1R8v9pEfmz7hDZ5b3oADxfejA3zJBYzKRCzgvhR", - "/ip4/155.133.27.199/tcp/4001/p2p/12D3KooWC69SBzM5QUgrLrfLWUykE8au32X5LwT7zwv9bixrQPm1", - "/ip4/217.76.56.2/tcp/4001/p2p/12D3KooWEiqJHvznxqJ5p2y8mUs6Ky6dfU1xTYFQbyKRCABfcZz4", + "/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj", + // "/ip4/217.76.54.178/tcp/4001/p2p/12D3KooWKZnirPwNT4URtNSWK45f6vLkEs4xyUZ792F8Uj1oYnm1", + // "/ip4/51.83.128.181/tcp/4001/p2p/12D3KooWBn2Zf1R8v9pEfmz7hDZ5b3oADxfejA3zJBYzKRCzgvhR", + // "/ip4/155.133.27.199/tcp/4001/p2p/12D3KooWC69SBzM5QUgrLrfLWUykE8au32X5LwT7zwv9bixrQPm1", + // "/ip4/217.76.56.2/tcp/4001/p2p/12D3KooWEiqJHvznxqJ5p2y8mUs6Ky6dfU1xTYFQbyKRCABfcZz4", }, BootstrapPort: 4001, // Default LibP2P port DiscoveryInterval: time.Second * 15, // Back to 15 seconds for testing diff --git a/pkg/config/validate.go b/pkg/config/validate.go new file mode 100644 index 0000000..6921ab4 --- /dev/null +++ b/pkg/config/validate.go @@ -0,0 +1,561 @@ +package config + +import ( + "fmt" + "net" + "os" + "path/filepath" + "strconv" + "strings" + + "github.com/multiformats/go-multiaddr" + manet "github.com/multiformats/go-multiaddr/net" +) + +// ValidationError represents a single validation error with context. +type ValidationError struct { + Path string // e.g., "discovery.bootstrap_peers[0]" + Message string // e.g., "invalid multiaddr" + Hint string // e.g., "expected /ip{4,6}/.../tcp//p2p/" +} + +func (e ValidationError) Error() string { + if e.Hint != "" { + return fmt.Sprintf("%s: %s; %s", e.Path, e.Message, e.Hint) + } + return fmt.Sprintf("%s: %s", e.Path, e.Message) +} + +// Validate performs comprehensive validation of the entire config. +// It aggregates all errors and returns them, allowing the caller to print all issues at once. +func (c *Config) Validate() []error { + var errs []error + + // Validate node config + errs = append(errs, c.validateNode()...) + // Validate database config + errs = append(errs, c.validateDatabase()...) + // Validate discovery config + errs = append(errs, c.validateDiscovery()...) + // Validate security config + errs = append(errs, c.validateSecurity()...) + // Validate logging config + errs = append(errs, c.validateLogging()...) + // Cross-field validations + errs = append(errs, c.validateCrossFields()...) + + return errs +} + +func (c *Config) validateNode() []error { + var errs []error + nc := c.Node + + // Validate type + if nc.Type != "bootstrap" && nc.Type != "node" { + errs = append(errs, ValidationError{ + Path: "node.type", + Message: fmt.Sprintf("must be one of [bootstrap node]; got %q", nc.Type), + }) + } + + // Validate listen_addresses + if len(nc.ListenAddresses) == 0 { + errs = append(errs, ValidationError{ + Path: "node.listen_addresses", + Message: "must not be empty", + }) + } + + seen := make(map[string]bool) + for i, addr := range nc.ListenAddresses { + path := fmt.Sprintf("node.listen_addresses[%d]", i) + + // Parse as multiaddr + ma, err := multiaddr.NewMultiaddr(addr) + if err != nil { + errs = append(errs, ValidationError{ + Path: path, + Message: fmt.Sprintf("invalid multiaddr: %v", err), + Hint: "expected /ip{4,6}/.../ tcp/", + }) + continue + } + + // Check for TCP and valid port + tcpAddr, err := manet.ToNetAddr(ma) + if err != nil { + errs = append(errs, ValidationError{ + Path: path, + Message: fmt.Sprintf("cannot convert multiaddr to network address: %v", err), + Hint: "ensure multiaddr contains /tcp/", + }) + continue + } + + tcpPort := tcpAddr.(*net.TCPAddr).Port + if tcpPort < 1 || tcpPort > 65535 { + errs = append(errs, ValidationError{ + Path: path, + Message: fmt.Sprintf("invalid TCP port %d", tcpPort), + Hint: "port must be between 1 and 65535", + }) + } + + if seen[addr] { + errs = append(errs, ValidationError{ + Path: path, + Message: "duplicate listen address", + }) + } + seen[addr] = true + } + + // Validate data_dir + if nc.DataDir == "" { + errs = append(errs, ValidationError{ + Path: "node.data_dir", + Message: "must not be empty", + }) + } else { + if err := validateDataDir(nc.DataDir); err != nil { + errs = append(errs, ValidationError{ + Path: "node.data_dir", + Message: err.Error(), + }) + } + } + + // Validate max_connections + if nc.MaxConnections <= 0 { + errs = append(errs, ValidationError{ + Path: "node.max_connections", + Message: fmt.Sprintf("must be > 0; got %d", nc.MaxConnections), + }) + } + + return errs +} + +func (c *Config) validateDatabase() []error { + var errs []error + dc := c.Database + + // Validate data_dir + if dc.DataDir == "" { + errs = append(errs, ValidationError{ + Path: "database.data_dir", + Message: "must not be empty", + }) + } else { + if err := validateDataDir(dc.DataDir); err != nil { + errs = append(errs, ValidationError{ + Path: "database.data_dir", + Message: err.Error(), + }) + } + } + + // Validate replication_factor + if dc.ReplicationFactor < 1 { + errs = append(errs, ValidationError{ + Path: "database.replication_factor", + Message: fmt.Sprintf("must be >= 1; got %d", dc.ReplicationFactor), + }) + } else if dc.ReplicationFactor%2 == 0 { + // Warn about even replication factor (Raft best practice: odd) + // For now we log a note but don't error + _ = fmt.Sprintf("note: database.replication_factor %d is even; Raft recommends odd numbers for quorum", dc.ReplicationFactor) + } + + // Validate shard_count + if dc.ShardCount < 1 { + errs = append(errs, ValidationError{ + Path: "database.shard_count", + Message: fmt.Sprintf("must be >= 1; got %d", dc.ShardCount), + }) + } + + // Validate max_database_size + if dc.MaxDatabaseSize < 0 { + errs = append(errs, ValidationError{ + Path: "database.max_database_size", + Message: fmt.Sprintf("must be >= 0; got %d", dc.MaxDatabaseSize), + }) + } + + // Validate rqlite_port + if dc.RQLitePort < 1 || dc.RQLitePort > 65535 { + errs = append(errs, ValidationError{ + Path: "database.rqlite_port", + Message: fmt.Sprintf("must be between 1 and 65535; got %d", dc.RQLitePort), + }) + } + + // Validate rqlite_raft_port + if dc.RQLiteRaftPort < 1 || dc.RQLiteRaftPort > 65535 { + errs = append(errs, ValidationError{ + Path: "database.rqlite_raft_port", + Message: fmt.Sprintf("must be between 1 and 65535; got %d", dc.RQLiteRaftPort), + }) + } + + // Ports must differ + if dc.RQLitePort == dc.RQLiteRaftPort { + errs = append(errs, ValidationError{ + Path: "database.rqlite_raft_port", + Message: fmt.Sprintf("must differ from database.rqlite_port (%d)", dc.RQLitePort), + }) + } + + // Validate rqlite_join_address context-dependently + if c.Node.Type == "node" { + if dc.RQLiteJoinAddress == "" { + errs = append(errs, ValidationError{ + Path: "database.rqlite_join_address", + Message: "required for node type (non-bootstrap)", + }) + } else { + if err := validateHostPort(dc.RQLiteJoinAddress); err != nil { + errs = append(errs, ValidationError{ + Path: "database.rqlite_join_address", + Message: err.Error(), + Hint: "expected format: host:port", + }) + } + } + } else if c.Node.Type == "bootstrap" { + if dc.RQLiteJoinAddress != "" { + errs = append(errs, ValidationError{ + Path: "database.rqlite_join_address", + Message: "must be empty for bootstrap type", + }) + } + } + + return errs +} + +func (c *Config) validateDiscovery() []error { + var errs []error + disc := c.Discovery + + // Validate discovery_interval + if disc.DiscoveryInterval <= 0 { + errs = append(errs, ValidationError{ + Path: "discovery.discovery_interval", + Message: fmt.Sprintf("must be > 0; got %v", disc.DiscoveryInterval), + }) + } + + // Validate bootstrap_port + if disc.BootstrapPort < 1 || disc.BootstrapPort > 65535 { + errs = append(errs, ValidationError{ + Path: "discovery.bootstrap_port", + Message: fmt.Sprintf("must be between 1 and 65535; got %d", disc.BootstrapPort), + }) + } + + // Validate bootstrap_peers context-dependently + if c.Node.Type == "node" { + if len(disc.BootstrapPeers) == 0 { + errs = append(errs, ValidationError{ + Path: "discovery.bootstrap_peers", + Message: "required for node type (must not be empty)", + }) + } + } + + // Validate each bootstrap peer multiaddr + seenPeers := make(map[string]bool) + for i, peer := range disc.BootstrapPeers { + path := fmt.Sprintf("discovery.bootstrap_peers[%d]", i) + + _, err := multiaddr.NewMultiaddr(peer) + if err != nil { + errs = append(errs, ValidationError{ + Path: path, + Message: fmt.Sprintf("invalid multiaddr: %v", err), + Hint: "expected /ip{4,6}/.../tcp//p2p/", + }) + continue + } + + // Check for /p2p/ component + if !strings.Contains(peer, "/p2p/") { + errs = append(errs, ValidationError{ + Path: path, + Message: "missing /p2p/ component", + Hint: "expected /ip{4,6}/.../tcp//p2p/", + }) + } + + // Extract TCP port by parsing the multiaddr string directly + // Look for /tcp/ in the peer string + tcpPortStr := extractTCPPort(peer) + if tcpPortStr == "" { + errs = append(errs, ValidationError{ + Path: path, + Message: "missing /tcp/ component", + Hint: "expected /ip{4,6}/.../tcp//p2p/", + }) + continue + } + + tcpPort, err := strconv.Atoi(tcpPortStr) + if err != nil || tcpPort < 1 || tcpPort > 65535 { + errs = append(errs, ValidationError{ + Path: path, + Message: fmt.Sprintf("invalid TCP port %s", tcpPortStr), + Hint: "port must be between 1 and 65535", + }) + } + + if seenPeers[peer] { + errs = append(errs, ValidationError{ + Path: path, + Message: "duplicate bootstrap peer", + }) + } + seenPeers[peer] = true + } + + // Validate http_adv_address + if disc.HttpAdvAddress != "" { + if err := validateHostOrHostPort(disc.HttpAdvAddress); err != nil { + errs = append(errs, ValidationError{ + Path: "discovery.http_adv_address", + Message: err.Error(), + Hint: "expected format: host or host:port", + }) + } + } + + // Validate raft_adv_address + if disc.RaftAdvAddress != "" { + if err := validateHostOrHostPort(disc.RaftAdvAddress); err != nil { + errs = append(errs, ValidationError{ + Path: "discovery.raft_adv_address", + Message: err.Error(), + Hint: "expected format: host or host:port", + }) + } + } + + return errs +} + +func (c *Config) validateSecurity() []error { + var errs []error + sec := c.Security + + // Validate logging level + if sec.EnableTLS { + if sec.PrivateKeyFile == "" { + errs = append(errs, ValidationError{ + Path: "security.private_key_file", + Message: "required when enable_tls is true", + }) + } else { + if err := validateFileReadable(sec.PrivateKeyFile); err != nil { + errs = append(errs, ValidationError{ + Path: "security.private_key_file", + Message: err.Error(), + }) + } + } + + if sec.CertificateFile == "" { + errs = append(errs, ValidationError{ + Path: "security.certificate_file", + Message: "required when enable_tls is true", + }) + } else { + if err := validateFileReadable(sec.CertificateFile); err != nil { + errs = append(errs, ValidationError{ + Path: "security.certificate_file", + Message: err.Error(), + }) + } + } + } + + return errs +} + +func (c *Config) validateLogging() []error { + var errs []error + log := c.Logging + + // Validate level + validLevels := map[string]bool{"debug": true, "info": true, "warn": true, "error": true} + if !validLevels[log.Level] { + errs = append(errs, ValidationError{ + Path: "logging.level", + Message: fmt.Sprintf("invalid value %q", log.Level), + Hint: "allowed values: debug, info, warn, error", + }) + } + + // Validate format + validFormats := map[string]bool{"json": true, "console": true} + if !validFormats[log.Format] { + errs = append(errs, ValidationError{ + Path: "logging.format", + Message: fmt.Sprintf("invalid value %q", log.Format), + Hint: "allowed values: json, console", + }) + } + + // Validate output_file + if log.OutputFile != "" { + dir := filepath.Dir(log.OutputFile) + if dir != "" && dir != "." { + if err := validateDirWritable(dir); err != nil { + errs = append(errs, ValidationError{ + Path: "logging.output_file", + Message: fmt.Sprintf("parent directory not writable: %v", err), + }) + } + } + } + + return errs +} + +func (c *Config) validateCrossFields() []error { + var errs []error + + // If node.type is invalid, don't run cross-checks + if c.Node.Type != "bootstrap" && c.Node.Type != "node" { + return errs + } + + // Cross-check rqlite_join_address vs node type + if c.Node.Type == "bootstrap" && c.Database.RQLiteJoinAddress != "" { + errs = append(errs, ValidationError{ + Path: "database.rqlite_join_address", + Message: "must be empty for bootstrap node type", + }) + } + + if c.Node.Type == "node" && c.Database.RQLiteJoinAddress == "" { + errs = append(errs, ValidationError{ + Path: "database.rqlite_join_address", + Message: "required for non-bootstrap node type", + }) + } + + return errs +} + +// Helper validation functions + +func validateDataDir(path string) error { + if path == "" { + return fmt.Errorf("must not be empty") + } + + if info, err := os.Stat(path); err == nil { + // Directory exists; check if it's a directory and writable + if !info.IsDir() { + return fmt.Errorf("path exists but is not a directory") + } + // Try to write a test file to check permissions + testFile := filepath.Join(path, ".write_test") + if err := os.WriteFile(testFile, []byte(""), 0644); err != nil { + return fmt.Errorf("directory not writable: %v", err) + } + os.Remove(testFile) + } else if os.IsNotExist(err) { + // Directory doesn't exist; check if parent is writable + parent := filepath.Dir(path) + if parent == "" || parent == "." { + parent = "." + } + if err := validateDirWritable(parent); err != nil { + return fmt.Errorf("parent directory not writable: %v", err) + } + } else { + return fmt.Errorf("cannot access path: %v", err) + } + + return nil +} + +func validateDirWritable(path string) error { + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("cannot access directory: %v", err) + } + if !info.IsDir() { + return fmt.Errorf("path is not a directory") + } + + // Try to write a test file + testFile := filepath.Join(path, ".write_test") + if err := os.WriteFile(testFile, []byte(""), 0644); err != nil { + return fmt.Errorf("directory not writable: %v", err) + } + os.Remove(testFile) + + return nil +} + +func validateFileReadable(path string) error { + _, err := os.Stat(path) + if err != nil { + return fmt.Errorf("cannot read file: %v", err) + } + return nil +} + +func validateHostPort(hostPort string) error { + parts := strings.Split(hostPort, ":") + if len(parts) != 2 { + return fmt.Errorf("expected format host:port") + } + + host := parts[0] + port := parts[1] + + if host == "" { + return fmt.Errorf("host must not be empty") + } + + portNum, err := strconv.Atoi(port) + if err != nil || portNum < 1 || portNum > 65535 { + return fmt.Errorf("port must be a number between 1 and 65535; got %q", port) + } + + return nil +} + +func validateHostOrHostPort(addr string) error { + // Try to parse as host:port first + if strings.Contains(addr, ":") { + return validateHostPort(addr) + } + + // Otherwise just check if it's a valid hostname/IP + if addr == "" { + return fmt.Errorf("address must not be empty") + } + + return nil +} + +func extractTCPPort(multiaddrStr string) string { + // Look for the /tcp/ protocol code + parts := strings.Split(multiaddrStr, "/") + for i := 0; i < len(parts); i++ { + if parts[i] == "tcp" { + // The port is the next part + if i+1 < len(parts) { + return parts[i+1] + } + break + } + } + return "" +} diff --git a/pkg/config/validate_test.go b/pkg/config/validate_test.go new file mode 100644 index 0000000..33de810 --- /dev/null +++ b/pkg/config/validate_test.go @@ -0,0 +1,409 @@ +package config + +import ( + "testing" + "time" +) + +func TestValidateNodeType(t *testing.T) { + tests := []struct { + name string + nodeType string + shouldError bool + }{ + {"bootstrap", "bootstrap", false}, + {"node", "node", false}, + {"invalid", "invalid-type", true}, + {"empty", "", true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: tt.nodeType, ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateListenAddresses(t *testing.T) { + tests := []struct { + name string + addresses []string + shouldError bool + }{ + {"valid single", []string{"/ip4/0.0.0.0/tcp/4001"}, false}, + {"valid ipv6", []string{"/ip6/::/tcp/4001"}, false}, + {"invalid port", []string{"/ip4/0.0.0.0/tcp/99999"}, true}, + {"invalid port zero", []string{"/ip4/0.0.0.0/tcp/0"}, true}, + {"invalid multiaddr", []string{"invalid"}, true}, + {"empty", []string{}, true}, + {"duplicate", []string{"/ip4/0.0.0.0/tcp/4001", "/ip4/0.0.0.0/tcp/4001"}, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: tt.addresses, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateReplicationFactor(t *testing.T) { + tests := []struct { + name string + replication int + shouldError bool + }{ + {"valid 1", 1, false}, + {"valid 3", 3, false}, + {"valid even", 2, false}, // warn but not error + {"invalid zero", 0, true}, + {"invalid negative", -1, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: tt.replication, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateRQLitePorts(t *testing.T) { + tests := []struct { + name string + httpPort int + raftPort int + shouldError bool + }{ + {"valid different", 5001, 7001, false}, + {"invalid same", 5001, 5001, true}, + {"invalid http port zero", 0, 7001, true}, + {"invalid raft port zero", 5001, 0, true}, + {"invalid http port too high", 99999, 7001, true}, + {"invalid raft port too high", 5001, 99999, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: tt.httpPort, RQLiteRaftPort: tt.raftPort, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateRQLiteJoinAddress(t *testing.T) { + tests := []struct { + name string + nodeType string + joinAddr string + shouldError bool + }{ + {"node with join", "node", "localhost:7001", false}, + {"node without join", "node", "", true}, + {"bootstrap with join", "bootstrap", "localhost:7001", true}, + {"bootstrap without join", "bootstrap", "", false}, + {"invalid join format", "node", "localhost", true}, + {"invalid join port", "node", "localhost:99999", true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: tt.nodeType, ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: tt.joinAddr}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateBootstrapPeers(t *testing.T) { + validPeer := "/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj" + tests := []struct { + name string + nodeType string + peers []string + shouldError bool + }{ + {"node with peer", "node", []string{validPeer}, false}, + {"node without peer", "node", []string{}, true}, + {"bootstrap with peer", "bootstrap", []string{validPeer}, false}, + {"bootstrap without peer", "bootstrap", []string{}, false}, + {"invalid multiaddr", "node", []string{"invalid"}, true}, + {"missing p2p", "node", []string{"/ip4/127.0.0.1/tcp/4001"}, true}, + {"duplicate peer", "node", []string{validPeer, validPeer}, true}, + {"invalid port", "node", []string{"/ip4/127.0.0.1/tcp/99999/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: tt.nodeType, ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: ""}, + Discovery: DiscoveryConfig{BootstrapPeers: tt.peers, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateLoggingLevel(t *testing.T) { + tests := []struct { + name string + level string + shouldError bool + }{ + {"debug", "debug", false}, + {"info", "info", false}, + {"warn", "warn", false}, + {"error", "error", false}, + {"invalid", "verbose", true}, + {"empty", "", true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: tt.level, Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateLoggingFormat(t *testing.T) { + tests := []struct { + name string + format string + shouldError bool + }{ + {"json", "json", false}, + {"console", "console", false}, + {"invalid", "text", true}, + {"empty", "", true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: tt.format}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateMaxConnections(t *testing.T) { + tests := []struct { + name string + maxConn int + shouldError bool + }{ + {"valid 50", 50, false}, + {"valid 1", 1, false}, + {"invalid zero", 0, true}, + {"invalid negative", -1, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: tt.maxConn}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateDiscoveryInterval(t *testing.T) { + tests := []struct { + name string + interval time.Duration + shouldError bool + }{ + {"valid 15s", 15 * time.Second, false}, + {"valid 1s", 1 * time.Second, false}, + {"invalid zero", 0, true}, + {"invalid negative", -5 * time.Second, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: tt.interval, BootstrapPort: 4001, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateBootstrapPort(t *testing.T) { + tests := []struct { + name string + port int + shouldError bool + }{ + {"valid 4001", 4001, false}, + {"valid 4002", 4002, false}, + {"invalid zero", 0, true}, + {"invalid too high", 99999, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Node: NodeConfig{Type: "node", ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4001"}, DataDir: ".", MaxConnections: 50}, + Database: DatabaseConfig{DataDir: ".", ReplicationFactor: 3, ShardCount: 16, MaxDatabaseSize: 1024, BackupInterval: 1 * time.Hour, RQLitePort: 5001, RQLiteRaftPort: 7001, RQLiteJoinAddress: "localhost:7001"}, + Discovery: DiscoveryConfig{BootstrapPeers: []string{"/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj"}, DiscoveryInterval: 15 * time.Second, BootstrapPort: tt.port, NodeNamespace: "default"}, + Logging: LoggingConfig{Level: "info", Format: "console"}, + } + errs := cfg.Validate() + if tt.shouldError && len(errs) == 0 { + t.Errorf("expected error, got none") + } + if !tt.shouldError && len(errs) > 0 { + t.Errorf("unexpected errors: %v", errs) + } + }) + } +} + +func TestValidateCompleteConfig(t *testing.T) { + // Test a complete valid config + validCfg := &Config{ + Node: NodeConfig{ + Type: "node", + ID: "node1", + ListenAddresses: []string{"/ip4/0.0.0.0/tcp/4002"}, + DataDir: ".", + MaxConnections: 50, + }, + Database: DatabaseConfig{ + DataDir: ".", + ReplicationFactor: 3, + ShardCount: 16, + MaxDatabaseSize: 1073741824, + BackupInterval: 24 * time.Hour, + RQLitePort: 5002, + RQLiteRaftPort: 7002, + RQLiteJoinAddress: "127.0.0.1:7001", + }, + Discovery: DiscoveryConfig{ + BootstrapPeers: []string{ + "/ip4/127.0.0.1/tcp/4001/p2p/12D3KooWHbcFcrGPXKUrHcxvd8MXEeUzRYyvY8fQcpEBxncSUwhj", + }, + DiscoveryInterval: 15 * time.Second, + BootstrapPort: 4001, + HttpAdvAddress: "127.0.0.1", + NodeNamespace: "default", + }, + Security: SecurityConfig{ + EnableTLS: false, + }, + Logging: LoggingConfig{ + Level: "info", + Format: "console", + }, + } + + errs := validCfg.Validate() + if len(errs) > 0 { + t.Errorf("valid config should not have errors: %v", errs) + } +} diff --git a/pkg/config/yaml.go b/pkg/config/yaml.go new file mode 100644 index 0000000..de40a9f --- /dev/null +++ b/pkg/config/yaml.go @@ -0,0 +1,19 @@ +package config + +import ( + "fmt" + "io" + + "gopkg.in/yaml.v3" +) + +// DecodeStrict decodes YAML from a reader and rejects any unknown fields. +// This ensures the YAML only contains recognized configuration keys. +func DecodeStrict(r io.Reader, out interface{}) error { + decoder := yaml.NewDecoder(r) + decoder.KnownFields(true) + if err := decoder.Decode(out); err != nil { + return fmt.Errorf("invalid config: %w", err) + } + return nil +} diff --git a/pkg/gateway/config_validate.go b/pkg/gateway/config_validate.go new file mode 100644 index 0000000..ebb4686 --- /dev/null +++ b/pkg/gateway/config_validate.go @@ -0,0 +1,117 @@ +package gateway + +import ( + "fmt" + "net" + "net/url" + "strconv" + "strings" + + "github.com/multiformats/go-multiaddr" + manet "github.com/multiformats/go-multiaddr/net" +) + +// ValidateConfig performs comprehensive validation of gateway configuration. +// It returns aggregated errors, allowing the caller to print all issues at once. +func (c *Config) ValidateConfig() []error { + var errs []error + + // Validate listen_addr + if c.ListenAddr == "" { + errs = append(errs, fmt.Errorf("gateway.listen_addr: must not be empty")) + } else { + if err := validateListenAddr(c.ListenAddr); err != nil { + errs = append(errs, fmt.Errorf("gateway.listen_addr: %v", err)) + } + } + + // Validate client_namespace + if c.ClientNamespace == "" { + errs = append(errs, fmt.Errorf("gateway.client_namespace: must not be empty")) + } + + // Validate bootstrap_peers if provided + seenPeers := make(map[string]bool) + for i, peer := range c.BootstrapPeers { + path := fmt.Sprintf("gateway.bootstrap_peers[%d]", i) + + ma, err := multiaddr.NewMultiaddr(peer) + if err != nil { + errs = append(errs, fmt.Errorf("%s: invalid multiaddr: %v; expected /ip{4,6}/.../tcp//p2p/", path, err)) + continue + } + + // Check for /p2p/ component + if !strings.Contains(peer, "/p2p/") { + errs = append(errs, fmt.Errorf("%s: missing /p2p/ component; expected /ip{4,6}/.../tcp//p2p/", path)) + } + + // Try to extract TCP addr to validate port + tcpAddr, err := manet.ToNetAddr(ma) + if err != nil { + errs = append(errs, fmt.Errorf("%s: cannot convert to network address: %v", path, err)) + continue + } + + tcpPort := tcpAddr.(*net.TCPAddr).Port + if tcpPort < 1 || tcpPort > 65535 { + errs = append(errs, fmt.Errorf("%s: invalid TCP port %d; port must be between 1 and 65535", path, tcpPort)) + } + + if seenPeers[peer] { + errs = append(errs, fmt.Errorf("%s: duplicate bootstrap peer", path)) + } + seenPeers[peer] = true + } + + // Validate rqlite_dsn if provided + if c.RQLiteDSN != "" { + if err := validateRQLiteDSN(c.RQLiteDSN); err != nil { + errs = append(errs, fmt.Errorf("gateway.rqlite_dsn: %v", err)) + } + } + + return errs +} + +// validateListenAddr checks if a listen address is valid (host:port format) +func validateListenAddr(addr string) error { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return fmt.Errorf("invalid format; expected host:port") + } + + portNum, err := strconv.Atoi(port) + if err != nil || portNum < 1 || portNum > 65535 { + return fmt.Errorf("port must be a number between 1 and 65535; got %q", port) + } + + // Allow empty host (for wildcard binds like :6001) + if host != "" && net.ParseIP(host) == nil { + // Try as hostname (may fail later during bind, but basic validation) + _, err := net.LookupHost(host) + if err != nil { + // Not an IP; assume it's a valid hostname for now + } + } + + return nil +} + +// validateRQLiteDSN checks if an RQLite DSN is a valid URL +func validateRQLiteDSN(dsn string) error { + u, err := url.Parse(dsn) + if err != nil { + return fmt.Errorf("invalid URL: %v", err) + } + + if u.Scheme != "http" && u.Scheme != "https" { + return fmt.Errorf("scheme must be http or https; got %q", u.Scheme) + } + + if u.Host == "" { + return fmt.Errorf("host must not be empty") + } + + return nil +}