package main import ( "crypto/rand" "crypto/tls" "crypto/x509" "encoding/json" "flag" "fmt" "net/http" "os" "os/exec" "path/filepath" "strings" "time" "git.netcell-it.de/projekte/edgeguard-native/internal/cluster/clustertls" ) const ( pgVersion = "16" pgCluster = "main" pgDataDir = "/var/lib/postgresql/16/main" pgHBAPath = "/etc/postgresql/16/main/pg_hba.conf" pgConfD = "/etc/postgresql/16/main/conf.d" egReplSecret = "/var/lib/edgeguard/pg-replication-secret" egReplUser = "edgeguard_replicator" egPubName = "edgeguard_shared" egSubName = "edgeguard_sub" // Tabellen die NODE-SPEZIFISCH sind und NICHT repliziert werden. // Jede Node hat eigene Interfaces, IPs, Routen, VIP-Einstellungen, // Listener-Adressen, Token-Tracking und Audit-Log. // Analog zu OPNsense: Interface-IPs und Hostname bleiben immer lokal. ) // localOnlyTables listet alle Tabellen die nicht in die Replikations- // Publication aufgenommen werden. Alles andere wird automatisch repliziert. var localOnlyTables = []string{ "ha_nodes", // Node-Identität, Status "network_interfaces", // Eigene Interfaces (eth0, eth1 …) "ip_addresses", // Eigene IP-Adressen (unterschiedlich pro Node!) "static_routes", // Node-spezifisches Routing "cluster_settings", // VIP-Interface kann pro Node unterschiedlich sein "dns_settings", // listen_addresses ist node-spezifisch "ntp_settings", // listen_addresses ist node-spezifisch "system_settings", // Hostname, Maintenance-Mode etc. "join_tokens_used", // Token-Tracking nur auf Primary relevant "audit_log", // Lokales Audit-Protokoll "alert_events", // Lokale Laufzeit-Events "backups", // Backup-Historie ist per-Node "goose_db_version", // Migration-Tracking, internes Tool-State } // cmdClusterInitReplication richtet PG auf dieser Node als Logical-Replication- // Primary ein. Idempotent — kann gefahrlos mehrfach laufen. // // Ablauf: // 1. edgeguard_replicator-Rolle anlegen/aktualisieren // 2. Passwort → /var/lib/edgeguard/pg-replication-secret // 3. conf.d/edgeguard-replication.conf mit wal_level=logical schreiben // 4. pg_hba.conf für Replikations-Verbindungen aktualisieren // 5. SELECT-Grants auf alle geteilten Tabellen // 6. PUBLICATION erstellen (alle Tabellen außer localOnlyTables) // 7. PG reload func cmdClusterInitReplication(args []string) int { fs := flag.NewFlagSet("cluster-init-replication", flag.ContinueOnError) fs.SetOutput(os.Stderr) if err := fs.Parse(args); err != nil { return 2 } // 1. Passwort generieren pass, err := generatePassword(32) if err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: generate password:", err) return 1 } // 2. edgeguard_replicator-Rolle anlegen/updaten roleSQL := fmt.Sprintf(`DO $$ BEGIN IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = '%s') THEN CREATE ROLE %s REPLICATION LOGIN PASSWORD '%s'; ELSE ALTER ROLE %s PASSWORD '%s'; END IF; END $$`, egReplUser, egReplUser, pass, egReplUser, pass) if err := psqlExec(roleSQL); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: create replication role:", err) return 1 } fmt.Printf("✓ Replication-Rolle %q angelegt/aktualisiert\n", egReplUser) // 3. Passwort speichern if err := os.MkdirAll(filepath.Dir(egReplSecret), 0o750); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: mkdir:", err) return 1 } if err := os.WriteFile(egReplSecret, []byte(pass), 0o600); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: write secret:", err) return 1 } fmt.Printf("✓ Replication-Secret gespeichert: %s\n", egReplSecret) // 4. conf.d/edgeguard-replication.conf schreiben // wal_level=logical ist eine Obermenge von replica — unterstützt // sowohl Logical Replication als auch ggfs. physisches WAL-Archiving. if err := os.MkdirAll(pgConfD, 0o755); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: conf.d mkdir:", err) return 1 } replConf := `# EdgeGuard Logical Replication — automatisch generiert # Nicht manuell bearbeiten; wird von edgeguard-ctl cluster-init-replication verwaltet. wal_level = logical max_wal_senders = 5 max_replication_slots = 5 max_logical_replication_workers = 4 wal_keep_size = 512MB ` confPath := filepath.Join(pgConfD, "edgeguard-replication.conf") if err := os.WriteFile(confPath, []byte(replConf), 0o644); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: write postgresql conf:", err) return 1 } fmt.Printf("✓ %s geschrieben (wal_level=logical)\n", confPath) // 5. pg_hba.conf aktualisieren if err := ensureHBAReplication(); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: pg_hba.conf:", err) return 1 } fmt.Printf("✓ %s aktualisiert\n", pgHBAPath) // 6. PG reload (damit wal_level + pg_hba aktiv werden) if out, err := exec.Command("pg_ctlcluster", pgVersion, pgCluster, "reload").CombinedOutput(); err != nil { fmt.Fprintf(os.Stderr, "cluster-init-replication: pg reload failed: %v\n%s\n", err, out) return 1 } fmt.Printf("✓ PostgreSQL %s/%s neu geladen\n", pgVersion, pgCluster) // 7. SELECT-Grants: edgeguard_replicator muss alle zu replizierenden // Tabellen lesen können. DEFAULT PRIVILEGES sichert zukünftige Tabellen. grantSQL := fmt.Sprintf(` GRANT SELECT ON ALL TABLES IN SCHEMA public TO %s; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO %s; `, egReplUser, egReplUser) if err := psqlDBExec("edgeguard", grantSQL); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: grant SELECT:", err) return 1 } fmt.Printf("✓ SELECT auf alle Tabellen für %q gewährt\n", egReplUser) // 8. PUBLICATION erstellen — alle public-Tabellen außer localOnlyTables. // Idempotent: DROP IF EXISTS + CREATE. if err := createPublication(); err != nil { fmt.Fprintln(os.Stderr, "cluster-init-replication: create publication:", err) return 1 } fmt.Printf("✓ PUBLICATION %q erstellt\n", egPubName) fmt.Println() fmt.Println("Nächste Schritte:") fmt.Println(" 1) Auf dem Secondary: edgeguard-ctl cluster-setup-standby ") fmt.Println(" 2) Cluster-Settings (VIP) auf BEIDEN Nodes separat konfigurieren") fmt.Println(" → Settings → Cluster → VIP/Keepalived") return 0 } // createPublication baut die PUBLICATION dynamisch aus allen Tabellen // im public-Schema minus localOnlyTables. Idempotent: löscht eine // bestehende Publication gleichen Namens zuerst. func createPublication() error { // Alle Tabellen im public-Schema ermitteln listSQL := `SELECT tablename FROM pg_tables WHERE schemaname = 'public' ORDER BY tablename` out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", listSQL}) if err != nil { return fmt.Errorf("list tables: %w", err) } excluded := make(map[string]bool) for _, t := range localOnlyTables { excluded[t] = true } var tables []string for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { t := strings.TrimSpace(line) if t == "" || excluded[t] { continue } tables = append(tables, t) } if len(tables) == 0 { return fmt.Errorf("keine Tabellen für Publication gefunden") } dropSQL := fmt.Sprintf("DROP PUBLICATION IF EXISTS %s;", egPubName) if err := psqlDBExec("edgeguard", dropSQL); err != nil { return fmt.Errorf("drop old publication: %w", err) } createSQL := fmt.Sprintf("CREATE PUBLICATION %s FOR TABLE %s;", egPubName, strings.Join(tables, ", ")) if err := psqlDBExec("edgeguard", createSQL); err != nil { return fmt.Errorf("create publication: %w", err) } return nil } // ensureHBAReplication fügt Einträge für die Replikations-Verbindung // in pg_hba.conf ein. Für Logical Replication brauchen wir einen // normalen "host edgeguard"-Eintrag (nicht "host replication"). // Idempotent via Marker-Kommentar. func ensureHBAReplication() error { data, err := os.ReadFile(pgHBAPath) if err != nil { return fmt.Errorf("read: %w", err) } const marker = "# EdgeGuard replication" if strings.Contains(string(data), marker) { return nil } // Logical Replication: Subscriber verbindet sich auf die DB (nicht "replication"-Typ) // Physical/WAL-Archiving: "replication"-Typ bleibt für Kompatibilität entry := fmt.Sprintf(` %s host edgeguard %s 0.0.0.0/0 scram-sha-256 host edgeguard %s ::/0 scram-sha-256 host replication %s 0.0.0.0/0 scram-sha-256 host replication %s ::/0 scram-sha-256 `, marker, egReplUser, egReplUser, egReplUser, egReplUser) f, err := os.OpenFile(pgHBAPath, os.O_APPEND|os.O_WRONLY, 0o640) if err != nil { return fmt.Errorf("open: %w", err) } defer f.Close() _, err = f.WriteString(entry) return err } // cmdClusterSetupStandby richtet diesen Node als Logical-Replication- // Subscriber ein. Der Secondary behält seine eigene beschreibbare PG- // Instanz — nur die geteilten Tabellen werden vom Primary repliziert. // Node-spezifische Tabellen (Interfaces, IPs, Routen, VIP-Settings …) // bleiben lokal und werden NICHT überschrieben. Analog zu OPNsense's // HA-Sync: Interface-IPs und Hostname bleiben immer per-Node konfiguriert. // // Voraussetzungen: // - cluster-join erfolgreich (TLS-Certs in /var/lib/edgeguard/cluster-tls/) // - Primary hat cluster-init-replication ausgeführt // - Dieser Node hat edgeguard-api schon gelaufen (Migrations ausgeführt) // // Ablauf: // 1. Replication-Credentials via mTLS vom Primary holen // 2. Bestehende Subscription löschen (idempotent) // 3. SUBSCRIPTION auf Primary erstellen (copy_data=true → Initialkopiierung) // 4. Warten bis Initialkopiierung abgeschlossen // 5. render-config ausführen damit Service-Configs den neuen Stand reflektieren func cmdClusterSetupStandby(args []string) int { fs := flag.NewFlagSet("cluster-setup-standby", flag.ContinueOnError) agentPort := fs.Int("agent-port", 8443, "mTLS agent port on primary") tlsDir := fs.String("tls-dir", clustertls.DefaultDir, "Verzeichnis mit ca.crt + peer.{crt,key}") fs.SetOutput(os.Stderr) if err := fs.Parse(args); err != nil { return 2 } if fs.NArg() < 1 { fmt.Fprintln(os.Stderr, "usage: edgeguard-ctl cluster-setup-standby ") return 2 } primaryHost := fs.Arg(0) // 1. Replication-Credentials vom Primary holen creds, err := fetchReplicationCreds(primaryHost, *agentPort, *tlsDir) if err != nil { fmt.Fprintf(os.Stderr, "cluster-setup-standby: replication-creds: %v\n", err) return 1 } fmt.Printf("✓ Replication-Credentials von %s:%d erhalten\n", primaryHost, *agentPort) // 2. Bestehende Subscription löschen (idempotent) dropSQL := fmt.Sprintf(` DO $$ BEGIN IF EXISTS (SELECT FROM pg_subscription WHERE subname = '%s') THEN ALTER SUBSCRIPTION %s DISABLE; ALTER SUBSCRIPTION %s SET (slot_name = NONE); DROP SUBSCRIPTION %s; END IF; END $$;`, egSubName, egSubName, egSubName, egSubName) if err := psqlDBExec("edgeguard", dropSQL); err != nil { // Nicht fatal — wenn PG noch keine Subscription kennt ist das OK fmt.Printf(" → keine bestehende Subscription gefunden (ok)\n") } else { fmt.Println("✓ Bestehende Subscription entfernt") } // 3. SUBSCRIPTION erstellen // sslmode=require: Verbindung zwischen Cluster-Nodes soll immer verschlüsselt sein. // copy_data=true: Initialkopiierung aller geteilten Tabellen vom Primary. connStr := fmt.Sprintf( "host=%s port=%d user=%s password=%s dbname=edgeguard sslmode=require", creds.Host, creds.Port, creds.User, creds.Password, ) createSQL := fmt.Sprintf( "CREATE SUBSCRIPTION %s CONNECTION '%s' PUBLICATION %s WITH (copy_data = true, enabled = true);", egSubName, connStr, egPubName, ) if err := psqlDBExec("edgeguard", createSQL); err != nil { fmt.Fprintf(os.Stderr, "cluster-setup-standby: create subscription: %v\n", err) return 1 } fmt.Printf("✓ SUBSCRIPTION %q erstellt — Initialkopiierung läuft\n", egSubName) // 4. Warten bis Initialkopiierung abgeschlossen fmt.Print("→ Warte auf Initialkopiierung") deadline := time.Now().Add(5 * time.Minute) for time.Now().Before(deadline) { pendingSQL := fmt.Sprintf(` SELECT COUNT(*) FROM pg_subscription_rel WHERE srsubid = (SELECT oid FROM pg_subscription WHERE subname = '%s') AND srsubstate != 'r';`, egSubName) out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", pendingSQL}) if err == nil && strings.TrimSpace(string(out)) == "0" { break } fmt.Print(".") time.Sleep(3 * time.Second) } fmt.Println() // Finale Prüfung checkSQL := fmt.Sprintf(` SELECT COUNT(*) FROM pg_subscription_rel WHERE srsubid = (SELECT oid FROM pg_subscription WHERE subname = '%s') AND srsubstate != 'r';`, egSubName) if out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", checkSQL}); err == nil { if n := strings.TrimSpace(string(out)); n != "0" { fmt.Fprintf(os.Stderr, "cluster-setup-standby: %s Tabellen noch nicht synchronisiert — prüfe PG-Logs\n", n) fmt.Println(" → Subscription läuft trotzdem weiter im Hintergrund") } else { fmt.Println("✓ Alle geteilten Tabellen synchronisiert") } } // 5. render-config ausführen fmt.Println("→ Service-Configs neu rendern...") if out, err := exec.Command("edgeguard-ctl", "render-config", "--no-reload").CombinedOutput(); err != nil { fmt.Fprintf(os.Stderr, "cluster-setup-standby: render-config: %v\n%s\n", err, out) fmt.Println(" → Manuell nachholen: edgeguard-ctl render-config") } else { fmt.Print(string(out)) fmt.Println("✓ Service-Configs aktualisiert") } fmt.Println() fmt.Println("✓ Logical Replication eingerichtet.") fmt.Println() fmt.Println("Was repliziert wird (automatisch, in Echtzeit):") fmt.Println(" Domains, Backends, Firewall-Rules, WireGuard, DNS-Zones,") fmt.Println(" TLS-Certs, Users, Forward-Proxy, NTP-Pools, ...") fmt.Println() fmt.Println("Was NICHT repliziert wird (bleibt pro Node konfiguriert):") fmt.Println(" Netzwerk-Interfaces, IP-Adressen, Routen,") fmt.Println(" Cluster-Settings (VIP-Interface!), DNS/NTP-Listen-Adressen") fmt.Println() fmt.Println("Nächste Schritte:") fmt.Println(" 1) sudo systemctl restart edgeguard-api") fmt.Println(" 2) VIP/Keepalived auf BEIDEN Nodes separat konfigurieren:") fmt.Println(" Settings → Cluster → VIP/Keepalived") fmt.Println(" 3) Bei Failover: edgeguard-ctl promote (auf dem Secondary)") return 0 } // pgReplicationCreds sind die Credentials die der Primary via mTLS zurückgibt. type pgReplicationCreds struct { Host string `json:"host"` Port int `json:"port"` User string `json:"user"` Password string `json:"password"` } // fetchReplicationCreds ruft GET /agent/cluster/pg-replication-info via mTLS ab. func fetchReplicationCreds(host string, agentPort int, tlsDir string) (*pgReplicationCreds, error) { caPath := filepath.Join(tlsDir, "ca.crt") certPath := filepath.Join(tlsDir, "peer.crt") keyPath := filepath.Join(tlsDir, "peer.key") caCert, err := os.ReadFile(caPath) if err != nil { return nil, fmt.Errorf("read ca.crt: %w", err) } pool := x509.NewCertPool() pool.AppendCertsFromPEM(caCert) cert, err := tls.LoadX509KeyPair(certPath, keyPath) if err != nil { return nil, fmt.Errorf("load peer cert: %w", err) } client := &http.Client{ Timeout: 15 * time.Second, Transport: &http.Transport{ TLSClientConfig: &tls.Config{ RootCAs: pool, Certificates: []tls.Certificate{cert}, }, }, } url := fmt.Sprintf("https://%s:%d/agent/cluster/pg-replication-info", host, agentPort) resp, err := client.Get(url) if err != nil { return nil, fmt.Errorf("GET %s: %w", url, err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("GET %s: HTTP %d", url, resp.StatusCode) } var result struct { Data pgReplicationCreds `json:"data"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return nil, fmt.Errorf("decode response: %w", err) } return &result.Data, nil } // generatePassword erzeugt ein kryptographisch sicheres Passwort. func generatePassword(n int) (string, error) { const charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" buf := make([]byte, n) if _, err := rand.Read(buf); err != nil { return "", err } for i, b := range buf { buf[i] = charset[int(b)%len(charset)] } return string(buf), nil } // psqlDBExec führt SQL in der angegebenen Datenbank als postgres-Superuser aus. func psqlDBExec(db, sql string) error { _, err := psqlDBRun(db, []string{"-v", "ON_ERROR_STOP=1", "-c", sql}) return err } // psqlDBRun führt psql-Kommandos gegen eine bestimmte Datenbank aus. func psqlDBRun(db string, args []string) ([]byte, error) { baseArgs := []string{"-d", db} return psqlRun(append(baseArgs, args...)) }