Accountant: Inventar, where, Checkmk, Abrechnung

This commit is contained in:
br committed 2026-10-18 17:08:58 +02:00
1 parent 9f16148cde
commit dd93277750
59 files changed
+7522 -113

No files matched your search

+1 -1
View File
@@ -73,7 +73,7 @@ func (s *Server) providerRoutes() http.Handler {
func (s *Server) handleEnroll(w http.ResponseWriter, r *http.Request) {
// Vor der Signaturprüfung: Fluten kosten so kaum etwas.
if ok, retry := s.enrollRL.allow(remoteIP(r)); !ok {
if ok, retry := s.enrollRL.Allow(remoteIP(r)); !ok {
w.Header().Set("Retry-After", strconv.Itoa(int(math.Ceil(retry.Seconds()))))
s.log.Warn("Enrollment gebremst (zu viele Versuche)", "ip", remoteIP(r))
http.Error(w, "zu viele Versuche", http.StatusTooManyRequests)
+401
View File
@@ -0,0 +1,401 @@
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 ""
}
+79
View File
@@ -0,0 +1,79 @@
package server
import (
"reflect"
"testing"
"time"
"sultan/pkg/accountant"
"sultan/pkg/registry"
)
// Aus dem gemeldeten Zustand eines Knotens entstehen die Berichtszeilen:
// Hosts, Ziel (IP und FQDN), Compose-Angaben, Produkt, down/aktiv.
func TestReportRoutes(t *testing.T) {
started := time.Date(2026, 10, 17, 8, 0, 0, 0, time.UTC)
st := registry.NodeState{
Host: &registry.HostInfo{FQDN: "webdocker11.lan.r25dc.de"},
Workloads: []registry.Workload{
{
Name: "kunde-a-wordpress-1", Host: "10.0.0.11",
Labels: map[string]string{
"traefik.enable": "true",
"traefik.http.routers.wp.rule": "Host(`www.kunde.de`) || Host(`kunde.de`)",
"traefik.http.services.wp.loadbalancer.server.port": "8081",
"sultan.accountant.billing": "web-hosting",
},
Meta: &registry.WorkloadMeta{ComposeProject: "kunde-a", ComposeService: "wordpress",
ComposeDir: "/srv/kunde-a", State: "running", StartedAt: started, Health: "healthy"},
},
{
Name: "olat", Host: "10.0.0.11",
Labels: map[string]string{
"traefik.enable": "true",
"traefik.http.routers.olat.rule": "Host(`olat.kunde.de`)",
"traefik.http.services.olat.loadbalancer.server.port": "8090",
},
Meta: &registry.WorkloadMeta{State: "exited"},
},
{
Name: "ca", Host: "10.0.0.11",
Labels: map[string]string{
"traefik.enable": "true",
"traefik.tcp.routers.ca.rule": "HostSNI(`ca.kunde.de`)",
"traefik.tcp.routers.ca.entrypoints": "websecure",
"traefik.tcp.routers.ca.tls.passthrough": "true",
"traefik.tcp.services.ca.loadbalancer.server.port": "9000",
},
},
{Name: "kaputt", Host: "10.0.0.11", Labels: map[string]string{"traefik.enable": "vielleicht"}},
},
DownServices: []string{"olat"},
}
active := map[string]bool{"http:kunde__webdocker11__wp": true, "tcp:kunde__webdocker11__ca": true}
got := reportRoutes("kunde", "webdocker11", st, func(proto, name string) bool { return active[proto+":"+name] })
want := []accountant.ReportRoute{
{Tenant: "kunde", Node: "webdocker11", Workload: "ca", Router: "ca", Protocol: "tcp",
Hosts: []string{"ca.kunde.de"}, Backend: "10.0.0.11:9000", BackendFQDN: "webdocker11.lan.r25dc.de:9000", Active: true},
{Tenant: "kunde", Node: "webdocker11", Workload: "kunde-a-wordpress-1", Router: "wp", Protocol: "http",
Hosts: []string{"kunde.de", "www.kunde.de"}, Backend: "http://10.0.0.11:8081",
BackendFQDN: "webdocker11.lan.r25dc.de:8081", Active: true, Product: "web-hosting",
ComposeProject: "kunde-a", ComposeService: "wordpress", ComposeDir: "/srv/kunde-a",
State: "running", StartedAt: started, Health: "healthy"},
{Tenant: "kunde", Node: "webdocker11", Workload: "olat", Router: "olat", Protocol: "http",
Hosts: []string{"olat.kunde.de"}, Backend: "http://10.0.0.11:8090",
BackendFQDN: "webdocker11.lan.r25dc.de:8090", Active: false, Down: true, State: "exited"},
}
if !reflect.DeepEqual(got, want) {
t.Errorf("reportRoutes =\n%+v\nerwartet\n%+v", got, want)
}
// Ohne Host-Infos (älterer Agent): kein FQDN-Ziel, Rest bleibt.
st.Host = nil
for _, r := range reportRoutes("kunde", "webdocker11", st, func(string, string) bool { return true }) {
if r.BackendFQDN != "" || r.Backend == "" {
t.Errorf("ohne Host-Infos: %+v", r)
}
}
}
+284
View File
@@ -0,0 +1,284 @@
package server
import (
"encoding/json"
"log/slog"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"sultan/pkg/accountant"
"sultan/pkg/crypto"
"sultan/pkg/ledger"
"sultan/pkg/registry"
)
// startAccountant startet einen echten Accountant; addr "" = freier Port.
func startAccountant(t *testing.T, dataDir, addr string) *accountant.Accountant {
t.Helper()
if dataDir == "" {
dataDir = filepath.Join(t.TempDir(), "accdata")
}
if addr == "" {
addr = "127.0.0.1:0"
}
a, err := accountant.New(ctx, accountant.Config{DataDir: dataDir, Addr: addr, MetricsAddr: "127.0.0.1:0", Logger: slog.New(slog.DiscardHandler)})
if err != nil {
t.Fatal(err)
}
if err := a.Start(); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { a.Shutdown(ctx) })
return a
}
// accInvite erzeugt eine Einladung wie `sultan accountant invite`.
func accInvite(t *testing.T, a *accountant.Accountant, payeeID string) string {
t.Helper()
l, err := ledger.Open(ctx, filepath.Join(a.DataDir(), "accountant.db"))
if err != nil {
t.Fatal(err)
}
defer l.Close()
tok, err := crypto.NewEnrollmentToken(a.Fingerprint(), "https://"+a.Addr())
if err != nil {
t.Fatal(err)
}
if err := l.AddInvite(ctx, tok.Secret, payeeID, time.Now().Add(time.Hour)); err != nil {
t.Fatal(err)
}
return tok.String()
}
func latestReport(t *testing.T, a *accountant.Accountant, payeeID string) (accountant.Report, time.Time, bool) {
t.Helper()
l, err := ledger.Open(ctx, filepath.Join(a.DataDir(), "accountant.db"))
if err != nil {
t.Fatal(err)
}
defer l.Close()
body, at, err := l.LatestReport(ctx, payeeID)
if err != nil {
return accountant.Report{}, time.Time{}, false
}
var r accountant.Report
if err := json.Unmarshal(body, &r); err != nil {
t.Fatal(err)
}
return r, at, true
}
// payeeConfig: Server meldet an den Accountant mit kurzen Intervallen.
func payeeConfig(token string, heartbeat time.Duration) func(*Config) {
return func(c *Config) {
c.AccountantToken = token
c.AccountantInterval = 20 * time.Millisecond
c.AccountantHeartbeat = heartbeat
c.Version = "v-test"
}
}
func metaState(domain string) registry.NodeState {
st := wpState(domain)
st.Workloads[0].Meta = &registry.WorkloadMeta{ComposeProject: "kunde-a", ComposeDir: "/srv/kunde-a", State: "running"}
st.Host = &registry.HostInfo{FQDN: "webdocker11.lan.r25dc.de", OS: "Debian GNU/Linux 13 (trixie)", DockerVersion: "29.8.2"}
return st
}
// Der ganze Weg: Einladung → Enrollment beim Start → Bericht mit Mandanten,
// Knoten und Routen samt FQDN-Ziel und Compose-Verzeichnis.
func TestPayee_EnrollsAndReports(t *testing.T) {
acc := startAccountant(t, "", "")
s, _ := startServer(t, payeeConfig(accInvite(t, acc, "webproxy11"), time.Hour))
agent, _ := connectedAgent(t, s, "node-1")
if r := agent.do(t, http.MethodPost, "/api/v1/state", metaState("kunde.de")); r.StatusCode != http.StatusNoContent {
t.Fatalf("State: %d", r.StatusCode)
}
var rep accountant.Report
eventually(t, "Bericht mit Zusatzdaten beim Accountant", func() bool {
r, _, ok := latestReport(t, acc, "webproxy11")
rep = r
return ok && len(r.Routes) == 1 && r.Routes[0].ComposeDir != ""
})
if rep.Version != "v-test" || rep.GeneratedAt.IsZero() {
t.Errorf("Kopf: %+v", rep)
}
if len(rep.Tenants) != 1 || rep.Tenants[0].ID != "kunde" || rep.Tenants[0].Tier != "pro" || rep.Tenants[0].Status != "active" {
t.Errorf("Mandanten: %+v", rep.Tenants)
}
if len(rep.Nodes) != 1 {
t.Fatalf("Knoten: %+v", rep.Nodes)
}
if n := rep.Nodes[0]; n.ID != "node-1" || n.Tenant != "kunde" || !n.Connected || n.FQDN != "webdocker11.lan.r25dc.de" ||
n.OS != "Debian GNU/Linux 13 (trixie)" || n.DockerVersion != "29.8.2" || n.Hostname != "host-node-1" {
t.Errorf("Knoten: %+v", n)
}
if r := rep.Routes[0]; r.Router != "wp" || strings.Join(r.Hosts, ",") != "kunde.de" || r.Backend != "http://100.64.0.42:80" ||
r.BackendFQDN != "webdocker11.lan.r25dc.de:80" || !r.Active || r.ComposeProject != "kunde-a" {
t.Errorf("Route: %+v", r)
}
// Identität des Payees liegt im Datenverzeichnis des Servers.
info, err := os.Stat(filepath.Join(s.cfg.DataDir, "payee.key"))
if err != nil || info.Mode().Perm() != 0o600 {
t.Errorf("payee.key: %v, %v", info, err)
}
raw, _ := os.ReadFile(filepath.Join(s.cfg.DataDir, "accountant.json"))
var id payeeIdentity
if err := json.Unmarshal(raw, &id); err != nil || id.PayeeID != "webproxy11" || id.Fingerprint != acc.Fingerprint() ||
id.URL != "https://"+acc.Addr() {
t.Errorf("accountant.json = %s (%v)", raw, err)
}
}
// Gesendet wird bei Änderungen und als Lebenszeichen – nicht bei jedem Prüfen.
func TestPayee_ReportsChangesNotEveryTick(t *testing.T) {
acc := startAccountant(t, "", "")
s, _ := startServer(t, payeeConfig(accInvite(t, acc, "webproxy11"), time.Hour))
agent, _ := connectedAgent(t, s, "node-1")
var first time.Time
eventually(t, "erster Bericht", func() bool {
r, at, ok := latestReport(t, acc, "webproxy11")
first = at
return ok && len(r.Routes) == 1
})
time.Sleep(200 * time.Millisecond) // zehn Prüfintervalle ohne Änderung
if _, at, _ := latestReport(t, acc, "webproxy11"); !at.Equal(first) {
t.Errorf("ohne Änderung erneut gesendet (%v → %v)", first, at)
}
st := wpState("kunde.de")
st.Workloads[0].Labels["traefik.http.routers.wp.rule"] = "Host(`kunde.de`) || Host(`www.kunde.de`)"
agent.do(t, http.MethodPost, "/api/v1/state", st)
eventually(t, "Änderung nachgemeldet", func() bool {
r, _, _ := latestReport(t, acc, "webproxy11")
return len(r.Routes) == 1 && len(r.Routes[0].Hosts) == 2
})
}
func TestPayee_Heartbeat(t *testing.T) {
acc := startAccountant(t, "", "")
s, _ := startServer(t, payeeConfig(accInvite(t, acc, "webproxy11"), 100*time.Millisecond))
connectedAgent(t, s, "node-1")
var first time.Time
eventually(t, "erster Bericht", func() bool {
_, at, ok := latestReport(t, acc, "webproxy11")
first = at
return ok
})
eventually(t, "Lebenszeichen ohne Änderung", func() bool {
_, at, _ := latestReport(t, acc, "webproxy11")
return at.After(first)
})
}
// Nach dem Neustart genügt accountant.json; die Einladung ist verbraucht.
func TestPayee_RestartWithoutToken(t *testing.T) {
acc := startAccountant(t, "", "")
dir := filepath.Join(t.TempDir(), "suldata")
s, _ := startServer(t, payeeConfig(accInvite(t, acc, "webproxy11"), time.Hour), func(c *Config) { c.DataDir = dir })
var first time.Time
eventually(t, "erster Bericht", func() bool {
_, at, ok := latestReport(t, acc, "webproxy11")
first = at
return ok
})
s.Shutdown(ctx)
startServer(t, payeeConfig("", 50*time.Millisecond), func(c *Config) { c.DataDir = dir })
eventually(t, "Bericht nach Neustart ohne Einladung", func() bool {
_, at, _ := latestReport(t, acc, "webproxy11")
return at.After(first)
})
}
// Ein nicht erreichbarer Accountant hält den Ingress nicht auf; der Server
// holt das Enrollment nach.
func TestPayee_AccountantDownAtStart(t *testing.T) {
accDir := filepath.Join(t.TempDir(), "accdata")
acc := startAccountant(t, accDir, "")
addr := acc.Addr()
token := accInvite(t, acc, "webproxy11")
acc.Shutdown(ctx)
s, logs := startServer(t, payeeConfig(token, time.Hour))
if r, err := http.Get("http://" + s.ProviderAddr() + "/healthz"); err != nil || r.StatusCode != http.StatusOK {
t.Fatalf("Server ohne Accountant nicht bereit: %v", err)
}
eventually(t, "Warnung im Log", func() bool { return strings.Contains(logs.String(), "Enrollment fehlgeschlagen") })
acc = startAccountant(t, accDir, addr)
eventually(t, "Enrollment nachgeholt", func() bool {
_, _, ok := latestReport(t, acc, "webproxy11")
return ok
})
}
// Schon bei Accountant A registriert: eine Einladung von B wird nicht
// eingelöst (kein stiller Wechsel), der Server läuft trotzdem.
func TestPayee_OtherAccountantRejected(t *testing.T) {
a := startAccountant(t, "", "")
dir := filepath.Join(t.TempDir(), "suldata")
s, _ := startServer(t, payeeConfig(accInvite(t, a, "webproxy11"), time.Hour), func(c *Config) { c.DataDir = dir })
eventually(t, "bei A registriert", func() bool { _, _, ok := latestReport(t, a, "webproxy11"); return ok })
s.Shutdown(ctx)
b := startAccountant(t, "", "")
_, logs := startServer(t, payeeConfig(accInvite(t, b, "webproxy11"), time.Hour), func(c *Config) { c.DataDir = dir })
eventually(t, "Fehler im Log", func() bool { return strings.Contains(logs.String(), "anderen Accountant") })
time.Sleep(100 * time.Millisecond)
if _, _, ok := latestReport(t, b, "webproxy11"); ok {
t.Error("Bericht an B trotz Registrierung bei A")
}
}
// --accountant überschreibt die Adresse aus der Einladung.
func TestPayee_URLOverride(t *testing.T) {
acc := startAccountant(t, "", "")
l, _ := ledger.Open(ctx, filepath.Join(acc.DataDir(), "accountant.db"))
tok, _ := crypto.NewEnrollmentToken(acc.Fingerprint(), "https://127.0.0.1:1") // falsche Adresse
l.AddInvite(ctx, tok.Secret, "webproxy11", time.Now().Add(time.Hour))
l.Close()
startServer(t, payeeConfig(tok.String(), time.Hour), func(c *Config) { c.AccountantURL = "https://" + acc.Addr() })
eventually(t, "Bericht über die überschriebene Adresse", func() bool {
_, _, ok := latestReport(t, acc, "webproxy11")
return ok
})
}
func TestPayee_ConfigErrors(t *testing.T) {
noURL, err := crypto.NewEnrollmentToken("PQeFR_XBBVMz70LWWvGsFsVrFT9SZVCvQk-HQvrBaTk", "")
if err != nil {
t.Fatal(err)
}
for name, mutate := range map[string]func(*Config){
"Einladung unlesbar": func(c *Config) { c.AccountantToken = "kaputt" },
"Adresse ohne TLS": func(c *Config) { c.AccountantURL = "http://acc:8444" },
"Einladung ohne jede Adresse": func(c *Config) { c.AccountantToken = noURL.String() },
"Adresse ohne Einladung": func(c *Config) { c.AccountantURL = "https://acc:8444" },
} {
t.Run(name, func(t *testing.T) {
cfg := Config{DataDir: filepath.Join(t.TempDir(), "suldata"), Logger: slog.New(slog.DiscardHandler)}
mutate(&cfg)
if s, err := New(ctx, cfg); err == nil {
s.Shutdown(ctx)
t.Error("erwartet Fehler")
}
})
}
}
// Ohne Einladung und ohne accountant.json meldet der Server nichts.
func TestPayee_Disabled(t *testing.T) {
s, _ := startServer(t)
time.Sleep(50 * time.Millisecond)
if _, err := os.Stat(filepath.Join(s.cfg.DataDir, "payee.key")); !os.IsNotExist(err) {
t.Errorf("payee.key ohne Accountant angelegt: %v", err)
}
}
+48 -2
View File
@@ -32,6 +32,7 @@ import (
"sultan/pkg/lander"
"sultan/pkg/notify"
"sultan/pkg/policy"
"sultan/pkg/ratelimit"
"sultan/pkg/registry"
"sultan/pkg/store"
"sultan/pkg/stream"
@@ -67,6 +68,16 @@ type Config struct {
// -1 = unbegrenzt) – gegen das Durchprobieren von Einladungen.
EnrollLimit int
// Accountant (Spec 5.3): Einladung (nur beim ersten Start) bzw.
// Adress-Override; ohne beides und ohne accountant.json meldet der
// Server nichts. Interval: Prüfen auf Änderungen (Default 10s),
// Heartbeat: Lebenszeichen ohne Änderung (Default 1h).
AccountantToken string
AccountantURL string
AccountantInterval time.Duration
AccountantHeartbeat time.Duration
Version string // für den Bericht (sultan version)
interceptorDerived bool // InterceptorURL folgt der gebundenen Lander-Adresse
}
@@ -99,6 +110,12 @@ func (c Config) withDefaults() Config {
if c.EnrollLimit == 0 {
c.EnrollLimit = 10
}
if c.AccountantInterval <= 0 {
c.AccountantInterval = 10 * time.Second
}
if c.AccountantHeartbeat <= 0 {
c.AccountantHeartbeat = time.Hour
}
if c.Logger == nil {
c.Logger = slog.New(slog.NewTextHandler(os.Stderr, nil))
}
@@ -141,7 +158,8 @@ type Server struct {
registry *registry.Registry
lander *lander.Lander
notifier notify.Target // nil = keine Benachrichtigungen
enrollRL *rateLimiter
enrollRL *ratelimit.Limiter
payee *payee // nil = meldet nicht an einen Accountant
vmu sync.Mutex
violations []time.Time // Zeitpunkte der Policy-Verstöße (für Checkmk)
@@ -151,6 +169,7 @@ type Server struct {
providerAddr string
landerAddr string
agentSrv *http.Server
payeeRunning bool
providerSrv *http.Server
landerSrv *http.Server
serveErr chan error
@@ -199,6 +218,11 @@ func New(ctx context.Context, cfg Config) (*Server, error) {
st.Close()
return nil, err
}
pay, err := newPayee(cfg, cfg.Logger)
if err != nil {
st.Close()
return nil, err
}
s := &Server{
cfg: cfg,
@@ -209,9 +233,13 @@ func New(ctx context.Context, cfg Config) (*Server, error) {
verifier: &auth.Verifier{Directory: st},
serveErr: make(chan error, 3),
stopRefresh: make(chan struct{}),
payee: pay,
}
if pay != nil {
pay.s = s
}
s.hub = &stream.Hub{OnConnect: s.nodeOnline, OnDisconnect: s.nodeOffline, SigningKey: master}
s.enrollRL = &rateLimiter{limit: cfg.EnrollLimit, window: time.Minute}
s.enrollRL = &ratelimit.Limiter{Limit: cfg.EnrollLimit, Window: time.Minute}
s.notifier = cfg.Notify
if s.notifier == nil {
var targets notify.Multi
@@ -260,6 +288,12 @@ func loadOrCreateMasterKey(path string) (ed25519.PrivateKey, error) {
// Fingerprint ist der Fingerprint des TLS-Zertifikats (für Enrollment-Tokens).
func (s *Server) Fingerprint() crypto.Fingerprint { return s.cert.Fingerprint }
// NodeState liefert den zuletzt gemeldeten Zustand eines verbundenen
// Knotens (Kopie, inkl. Zusatzdaten für das Inventar).
func (s *Server) NodeState(nodeID string) (registry.NodeState, bool) {
return s.registry.State(nodeID)
}
// MasterPublicKey ist der Schlüssel, mit dem Lizenzen geprüft werden.
func (s *Server) MasterPublicKey() ed25519.PublicKey {
return s.master.Public().(ed25519.PublicKey)
@@ -338,6 +372,12 @@ func (s *Server) Start() error {
go s.serve(func() error { return landerSrv.Serve(landerLn) })
}
go s.refreshLoop()
if s.payee != nil {
s.mu.Lock()
s.payeeRunning = true
s.mu.Unlock()
go s.payee.run()
}
s.log.Info("Sultan-Server gestartet",
"agent", s.AgentAddr(), "provider", s.ProviderAddr(), "lander", s.LanderAddr(),
@@ -405,6 +445,12 @@ func (s *Server) Shutdown(ctx context.Context) error {
errs = append(errs, srv.Shutdown(ctx))
}
}
s.mu.Lock()
payeeRunning := s.payeeRunning
s.mu.Unlock()
if payeeRunning {
s.payee.shutdown() // vor dem Schließen der Datenbank
}
errs = append(errs, s.store.Close())
s.shutdownErr = errors.Join(errs...)
s.log.Info("Sultan-Server beendet")