package server // Payee: der Server meldet sein Inventar an den Accountant (Spec 5.3, // Roadmap 6c) – mit demselben Sicherheitsmodell wie Agent → Server: // eigener Schlüssel payee.key, Pinning des Accountant-Fingerprints aus der // Einladung, signierte Requests, nur TLS. Ein nicht erreichbarer Accountant // hält den Ingress nie auf: Enrollment und Berichte werden im Hintergrund // wiederholt. import ( "bytes" "cmp" "context" "crypto/ed25519" "crypto/sha256" "encoding/json" "errors" "fmt" "io" "log/slog" "net" "net/http" "net/url" "os" "path/filepath" "slices" "strings" "time" "sultan/pkg/accountant" "sultan/pkg/auth" "sultan/pkg/crypto" "sultan/pkg/policy" "sultan/pkg/registry" "sultan/pkg/traefik" ) // payeeIdentity ist accountant.json im Datenverzeichnis des Servers. type payeeIdentity struct { URL string `json:"url"` Fingerprint crypto.Fingerprint `json:"fingerprint"` PayeeID string `json:"payee_id"` } type payee struct { s *Server log *slog.Logger dir string key ed25519.PrivateKey token *crypto.EnrollmentToken // nil = schon registriert url string // tatsächlich genutzte Adresse fp crypto.Fingerprint // gepinnter Accountant id *payeeIdentity // nil bis zum Enrollment // conflict: Einladung eines anderen Accountants bei bestehender // Registrierung – es wird nichts gemeldet. conflict error interval, heartbeat time.Duration hc *http.Client lastHash [32]byte lastSent time.Time lastErr string // nur geänderte Fehler loggen stop chan struct{} done chan struct{} } // newPayee liest die Accountant-Konfiguration; nil = Server meldet nicht. // Fehler gibt es nur für unbrauchbare Angaben (Tippfehler), nie für einen // nicht erreichbaren Accountant. func newPayee(cfg Config, log *slog.Logger) (*payee, error) { p := &payee{ log: log, dir: cfg.DataDir, interval: cfg.AccountantInterval, heartbeat: cfg.AccountantHeartbeat, stop: make(chan struct{}), done: make(chan struct{}), } if raw := strings.TrimSpace(cfg.AccountantToken); raw != "" { tok, err := crypto.ParseEnrollmentToken(raw) if err != nil { return nil, fmt.Errorf("server: --accountant-token: %w", err) } p.token = &tok } var override string if cfg.AccountantURL != "" { u, err := crypto.ValidateServerURL(cfg.AccountantURL) if err != nil { return nil, fmt.Errorf("server: --accountant: %w", err) } override = u } if data, err := os.ReadFile(p.path("accountant.json")); err == nil { var id payeeIdentity if err := json.Unmarshal(data, &id); err != nil || id.Fingerprint == "" || id.PayeeID == "" { return nil, fmt.Errorf("server: accountant.json beschädigt – Datei löschen und neu einladen") } p.id = &id } else if !errors.Is(err, os.ErrNotExist) { return nil, fmt.Errorf("server: accountant.json: %w", err) } switch { case p.token == nil && p.id == nil: if override != "" { return nil, errors.New("server: --accountant ohne Einladung (--accountant-token) und ohne accountant.json") } return nil, nil case p.id != nil: p.fp, p.url = p.id.Fingerprint, cmp.Or(override, p.id.URL) if p.token != nil && p.token.Fingerprint != p.id.Fingerprint { p.conflict = fmt.Errorf("bereits bei einem anderen Accountant registriert (Fingerprint %s) – für den neuen accountant.json und payee.key löschen", p.id.Fingerprint) } p.token = nil // eine stehen gebliebene Einladung desselben Accountants ist harmlos default: p.fp, p.url = p.token.Fingerprint, cmp.Or(override, p.token.ServerURL) if p.url == "" { return nil, errors.New("server: Einladung ohne Accountant-Adresse – --accountant=https://…:8444 angeben") } } key, err := loadOrCreateKey(p.path("payee.key")) if err != nil { return nil, err } p.key = key p.hc = &http.Client{ Timeout: 30 * time.Second, Transport: &http.Transport{TLSClientConfig: crypto.PinnedTLSConfig(p.fp)}, } return p, nil } func (p *payee) path(name string) string { return filepath.Join(p.dir, name) } func loadOrCreateKey(path string) (ed25519.PrivateKey, error) { if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) { key, err := auth.GenerateKey() if err != nil { return nil, err } if err := auth.WritePrivateKey(path, key); err != nil { return nil, fmt.Errorf("server: %s anlegen: %w", filepath.Base(path), err) } } key, err := auth.ReadPrivateKey(path) if err != nil { return nil, fmt.Errorf("server: %s laden: %w", filepath.Base(path), err) } return key, nil } // run arbeitet, bis stop geschlossen wird: Enrollment (falls nötig), dann // Bericht bei Änderung bzw. als Lebenszeichen. func (p *payee) run() { defer close(p.done) if p.conflict != nil { p.log.Error("Accountant: Berichte abgeschaltet", "err", p.conflict) return } ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { <-p.stop; cancel() }() t := time.NewTicker(p.interval) defer t.Stop() for { if p.id == nil { p.report(p.enroll(ctx), "Accountant: Enrollment fehlgeschlagen – neuer Versuch folgt") } if p.id != nil { p.report(p.maybeSend(ctx), "Accountant: Bericht fehlgeschlagen – neuer Versuch folgt") } select { case <-p.stop: return case <-t.C: } } } // report loggt Fehler nur, wenn sie sich ändern (kein Log-Spam bei Ausfall). func (p *payee) report(err error, msg string) { if err == nil { if p.lastErr != "" { p.log.Info("Accountant wieder erreichbar", "url", p.url) } p.lastErr = "" return } if errors.Is(err, context.Canceled) || err.Error() == p.lastErr { return } p.lastErr = err.Error() p.log.Warn(msg, "url", p.url, "err", err) } func (p *payee) post(ctx context.Context, path string, body any) (*http.Response, error) { raw, err := json.Marshal(body) if err != nil { return nil, err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.url+path, bytes.NewReader(raw)) if err != nil { return nil, err } req.Header.Set("Content-Type", "application/json") if err := (&auth.Signer{Key: p.key}).Sign(req); err != nil { return nil, err } return p.hc.Do(req) } func responseError(resp *http.Response) error { msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) return fmt.Errorf("Accountant antwortet %d: %s", resp.StatusCode, strings.TrimSpace(string(msg))) } func (p *payee) enroll(ctx context.Context) error { hostname, _ := os.Hostname() resp, err := p.post(ctx, accountant.PathEnroll, accountant.EnrollRequest{Token: p.token.Secret, Hostname: cmp.Or(hostname, "sultan-server")}) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusCreated { return responseError(resp) } var er accountant.EnrollResponse if err := json.NewDecoder(resp.Body).Decode(&er); err != nil || !policy.ValidID(er.PayeeID) { return fmt.Errorf("Antwort des Accountants unlesbar") } id := payeeIdentity{URL: p.url, Fingerprint: p.fp, PayeeID: er.PayeeID} raw, _ := json.MarshalIndent(id, "", " ") if err := os.WriteFile(p.path("accountant.json"), append(raw, '\n'), 0o600); err != nil { return fmt.Errorf("accountant.json schreiben: %w", err) } p.id, p.token = &id, nil p.log.Info("Beim Accountant registriert", "payee", er.PayeeID, "url", p.url) return nil } // maybeSend sendet den Schnappschuss, wenn er sich geändert hat oder das // Lebenszeichen fällig ist. func (p *payee) maybeSend(ctx context.Context) error { rep, err := p.s.buildReport(ctx) if err != nil { return err } raw, _ := json.Marshal(rep) // ohne Zeitpunkt: nur der Inhalt zählt hash := sha256.Sum256(raw) if hash == p.lastHash && time.Since(p.lastSent) < p.heartbeat { return nil } rep.GeneratedAt = time.Now().UTC() resp, err := p.post(ctx, accountant.PathReport, rep) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusNoContent { return responseError(resp) } p.lastHash, p.lastSent = hash, time.Now() return nil } func (p *payee) shutdown() { close(p.stop) <-p.done } // buildReport erzeugt den Schnappschuss (ohne GeneratedAt). func (s *Server) buildReport(ctx context.Context) (accountant.Report, error) { rep := accountant.Report{Version: s.cfg.Version, Tenants: []accountant.ReportTenant{}, Nodes: []accountant.ReportNode{}, Routes: []accountant.ReportRoute{}} tenants, err := s.store.ListTenants(ctx) if err != nil { return rep, err } for _, t := range tenants { rep.Tenants = append(rep.Tenants, accountant.ReportTenant{ID: t.ID, Name: t.Name, Tier: string(t.Tier), Status: string(t.Status)}) } nodes, err := s.store.ListNodes(ctx, "") if err != nil { return rep, err } active := activeRouters(s.registry.Config()) now := time.Now() for _, n := range nodes { rn := accountant.ReportNode{ ID: n.ID, Tenant: n.TenantID, Hostname: n.Hostname, IP: n.IPAddress, Status: string(n.Status), Connected: s.hub.Connected(n.ID), Maintenance: n.Maintenance.Active(now), LastSeen: n.LastSeen.UTC(), } if st, ok := s.registry.State(n.ID); ok { if h := st.Host; h != nil { rn.FQDN, rn.OS, rn.DockerVersion, rn.NetBirdIP = h.FQDN, h.OS, h.DockerVersion, h.NetBirdIP } rep.Routes = append(rep.Routes, reportRoutes(n.TenantID, n.ID, st, active)...) } rep.Nodes = append(rep.Nodes, rn) } return rep, nil } // activeRouters: ob ein (qualifizierter) Router in der Traefik-Konfiguration steht. func activeRouters(c *traefik.Configuration) func(proto, name string) bool { return func(proto, name string) bool { switch { case c == nil: return false case proto == "tcp": return c.TCP != nil && c.TCP.Routers[name] != nil default: return c.HTTP != nil && c.HTTP.Routers[name] != nil } } } // reportRoutes übersetzt den Zustand eines Knotens in Berichtszeilen, sortiert // nach Workload, Protokoll, Router. Workloads mit unlesbaren Labels fehlen // (die Registry meldet sie als Verstoß). func reportRoutes(tenantID, nodeID string, st registry.NodeState, active func(proto, name string) bool) []accountant.ReportRoute { fqdn := "" if st.Host != nil { fqdn = st.Host.FQDN } down := map[string]bool{} for _, d := range st.DownServices { down[d] = true } prefix := tenantID + "__" + nodeID + "__" var out []accountant.ReportRoute for _, w := range st.Workloads { cfg, err := traefik.FromLabels(w.Host, w.Labels) if err != nil { continue } base := accountant.ReportRoute{Tenant: tenantID, Node: nodeID, Workload: w.Name, Product: labelValue(w.Labels, accountant.LabelBilling)} if m := w.Meta; m != nil { base.ComposeProject, base.ComposeService, base.ComposeDir = m.ComposeProject, m.ComposeService, m.ComposeDir base.State, base.StartedAt, base.Health = m.State, m.StartedAt, m.Health } if cfg.HTTP != nil { for name, r := range cfg.HTTP.Routers { route := base route.Router, route.Protocol = name, "http" route.Hosts = sortedHosts(policy.HostsFromRule(r.Rule)) if svc := cfg.HTTP.Services[r.Service]; svc != nil && svc.LoadBalancer != nil && len(svc.LoadBalancer.Servers) > 0 { route.Backend = svc.LoadBalancer.Servers[0].URL if u, err := url.Parse(route.Backend); err == nil && fqdn != "" && u.Port() != "" { route.BackendFQDN = net.JoinHostPort(fqdn, u.Port()) } } route.Active, route.Down = active("http", prefix+name), down[r.Service] out = append(out, route) } } if cfg.TCP != nil { for name, r := range cfg.TCP.Routers { route := base route.Router, route.Protocol = name, "tcp" route.Hosts = sortedHosts(policy.HostsFromSNIRule(r.Rule)) if svc := cfg.TCP.Services[r.Service]; svc != nil && svc.LoadBalancer != nil && len(svc.LoadBalancer.Servers) > 0 { route.Backend = svc.LoadBalancer.Servers[0].Address if _, port, err := net.SplitHostPort(route.Backend); err == nil && fqdn != "" { route.BackendFQDN = net.JoinHostPort(fqdn, port) } } route.Active, route.Down = active("tcp", prefix+name), down[r.Service] out = append(out, route) } } } slices.SortFunc(out, func(a, b accountant.ReportRoute) int { return cmp.Or(strings.Compare(a.Workload, b.Workload), strings.Compare(a.Protocol, b.Protocol), strings.Compare(a.Router, b.Router)) }) return out } func sortedHosts(hosts []string, err error) []string { if err != nil { return nil } out := make([]string, 0, len(hosts)) for _, h := range hosts { out = append(out, strings.ToLower(h)) } slices.Sort(out) return slices.Compact(out) } // labelValue sucht einen Label-Wert ohne Beachtung der Groß-/Kleinschreibung. func labelValue(labels map[string]string, key string) string { for k, v := range labels { if strings.EqualFold(k, key) { return strings.TrimSpace(v) } } return "" }