sultan/pkg/server/payee.go

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 ""
}