sultan/pkg/agent/events.go

package agent

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"

	"sultan/pkg/auth"
	"sultan/pkg/license"
	"sultan/pkg/server"
	"sultan/pkg/stream"
)

// currentSigner liefert den aktuellen Signierer (rotate_key tauscht ihn aus).
func (a *Agent) currentSigner() *auth.Signer {
	a.keyMu.RLock()
	defer a.keyMu.RUnlock()
	return a.signer
}

func (a *Agent) setSigner(s *auth.Signer) {
	a.keyMu.Lock()
	a.signer = s
	a.keyMu.Unlock()
}

// handleEvent verarbeitet Events vom Server. Nur Events mit gültiger
// Signatur des Master-Keys werden beachtet; Fehler beenden nie den Stream.
func (a *Agent) handleEvent(ev stream.Event) error {
	pub, err := auth.DecodePublicKey(a.id.MasterPublicKey)
	if err != nil {
		a.log.Error("Master-Key in node.json unlesbar – Event verworfen", "err", err)
		return nil
	}
	if err := stream.VerifyEvent(ev, pub); err != nil {
		a.log.Warn("Event ohne gültige Server-Signatur verworfen", "type", ev.Type, "err", err)
		return nil
	}
	switch ev.Type {
	case stream.EventLeaseRenew:
		a.recordLease(ev.Payload, pub)
	case stream.EventNodeCommand:
		cmd, err := stream.ParseNodeCommand(ev)
		if err != nil {
			a.log.Warn("Befehl unlesbar", "err", err)
			return nil
		}
		a.runCommand(cmd)
	default:
		a.log.Info("Ereignis vom Server", "type", ev.Type)
	}
	return nil
}

// recordLease prüft die Lease, legt sie als lease.json ab und protokolliert
// sie (vorerst ohne weitere Konsequenz, Spec 4a).
func (a *Agent) recordLease(doc []byte, pub ed25519.PublicKey) {
	l, err := license.ParseLease(doc, pub)
	if err != nil {
		a.log.Warn("Lease ungültig – verworfen", "err", err)
		return
	}
	if l.NodeID != a.id.NodeID {
		a.log.Warn("Lease für anderen Knoten – verworfen", "node", l.NodeID)
		return
	}
	tmp := a.path("lease.json.tmp")
	if err := os.WriteFile(tmp, doc, 0o600); err == nil {
		err = os.Rename(tmp, a.path("lease.json"))
	}
	if err != nil {
		a.log.Warn("lease.json nicht gespeichert", "err", err)
	}
	a.log.Info("Lease erhalten", "valid_until", l.ValidUntil, "renew_after", l.RenewAfter, "tier", l.Tier)
}

func (a *Agent) runCommand(cmd stream.NodeCommand) {
	switch cmd.Command {
	case stream.CmdResync:
		a.log.Info("Server fordert Zustand an (resync)")
		if a.kick != nil {
			a.kick()
		}
	case stream.CmdRotateKey:
		a.log.Info("Server fordert Schlüsselwechsel an", "reason", cmd.Reason)
		ctx := a.runCtx
		if ctx == nil {
			ctx = context.Background()
		}
		if err := a.rotateKey(ctx); err != nil {
			a.log.Error("Schlüsselwechsel fehlgeschlagen", "err", err)
		}
	case stream.CmdConfig:
		a.updateServerURLs(cmd.ServerURLs)
	case stream.CmdDrain, stream.CmdMaintenance, stream.CmdResume:
		a.log.Info("Hinweis vom Server", "command", cmd.Command, "reason", cmd.Reason)
	default:
		a.log.Warn("Unbekannter Befehl ignoriert", "command", cmd.Command)
	}
}

// rotateKey erzeugt einen neuen Schlüssel (node.key.new) und meldet ihn.
func (a *Agent) rotateKey(ctx context.Context) error {
	newPath := a.path("node.key.new")
	if _, err := os.Stat(newPath); errors.Is(err, os.ErrNotExist) {
		key, err := auth.GenerateKey()
		if err != nil {
			return err
		}
		if err := auth.WritePrivateKey(newPath, key); err != nil {
			return err
		}
	}
	return a.completeRotation(ctx)
}

// completeRotation meldet node.key.new beim Server und benennt die Datei erst
// danach um. Abbruch-sicher: Lehnt der Server den alten Schlüssel ab (er hat
// den neuen schon), wird mit dem neuen wiederholt.
func (a *Agent) completeRotation(ctx context.Context) error {
	newPath := a.path("node.key.new")
	newKey, err := auth.ReadPrivateKey(newPath)
	if err != nil {
		return err
	}
	body, _ := json.Marshal(server.RotateKeyRequest{
		PublicKey: auth.EncodePublicKey(newKey.Public().(ed25519.PublicKey)),
		Proof:     auth.RotationProof(a.id.NodeID, newKey),
	})
	newSigner := &auth.Signer{Key: newKey}
	status, err := a.postSigned(ctx, "/api/v1/rotate-key", body, a.currentSigner())
	if err == nil && status == http.StatusUnauthorized {
		status, err = a.postSigned(ctx, "/api/v1/rotate-key", body, newSigner)
	}
	if err != nil {
		return err
	}
	if status != http.StatusNoContent {
		return fmt.Errorf("Server lehnt Schlüsselwechsel ab (%d)", status)
	}
	if err := os.Rename(newPath, a.path("node.key")); err != nil {
		return err
	}
	a.setSigner(newSigner)
	a.log.Info("Knotenschlüssel gewechselt")
	return nil
}

func (a *Agent) postSigned(ctx context.Context, path string, body []byte, signer *auth.Signer) (int, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.currentServerURL()+path, bytes.NewReader(body))
	if err != nil {
		return 0, err
	}
	req.Header.Set("Content-Type", "application/json")
	if err := signer.Sign(req); err != nil {
		return 0, err
	}
	resp, err := a.httpClient().Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 400 && resp.StatusCode != http.StatusUnauthorized {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
		return resp.StatusCode, fmt.Errorf("Server antwortet %d: %s", resp.StatusCode, strings.TrimSpace(string(msg)))
	}
	return resp.StatusCode, nil
}