feat(service): agents, enrollment tokens, revisions, deployments and leases

Use cases for the REST API (agent list/revoke/delete, one-time tokens
with the ready enrollment command, immutable revisions with unchanged
detection, deployment and lease lists) and for the agent data plane
(enrollment, renewal, status, bundles, deployment and lease reports).
A hub of open agent streams pushes new revisions to auto_pull agents and
closes the stream of revoked agents.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/internal/agenthub/hub.go

@@ -0,0 +1,155 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+// Package agenthub tracks the open Connect streams of the agents, so that
+// the service layer can push commands to them and close the stream of a
+// revoked agent (ADR 0021).
+package agenthub
+
+import (
+	"crypto/rand"
+	"fmt"
+	"sync"
+
+	agentv1 "git.beugsiud.de/beug.k/DDI/proto/gen/ddi/agent/v1"
+)
+
+// queueSize bounds the commands waiting for one stream. A full queue drops
+// commands; the agent catches up through GetDesiredState.
+const queueSize = 16
+
+// Mode values (agents.mode).
+const (
+	ModeAutoPull = "auto_pull"
+	ModeManual   = "manual"
+)
+
+// Hub is the registry of open streams. It is safe for concurrent use.
+type Hub struct {
+	mu    sync.Mutex
+	conns map[string]*Conn
+}
+
+// New creates an empty hub.
+func New() *Hub { return &Hub{conns: map[string]*Conn{}} }
+
+// Conn is one open stream.
+type Conn struct {
+	AgentID string
+	Role    string
+	Mode    string
+
+	cmds   chan *agentv1.Command
+	done   chan struct{}
+	once   sync.Once
+	reason string
+}
+
+// Commands delivers the commands to send on the stream.
+func (c *Conn) Commands() <-chan *agentv1.Command { return c.cmds }
+
+// Done is closed when the server ends the stream (revocation, deletion or a
+// newer stream of the same agent).
+func (c *Conn) Done() <-chan struct{} { return c.done }
+
+// Reason says why Done was closed.
+func (c *Conn) Reason() string {
+	<-c.done
+	return c.reason
+}
+
+func (c *Conn) close(reason string) {
+	c.once.Do(func() {
+		c.reason = reason
+		close(c.done)
+	})
+}
+
+// Attach registers a stream. An older stream of the same agent is closed.
+func (h *Hub) Attach(agentID, role, mode string) *Conn {
+	c := &Conn{AgentID: agentID, Role: role, Mode: mode, cmds: make(chan *agentv1.Command, queueSize), done: make(chan struct{})}
+	h.mu.Lock()
+	old := h.conns[agentID]
+	h.conns[agentID] = c
+	h.mu.Unlock()
+	if old != nil {
+		old.close("replaced by a newer connection")
+	}
+	return c
+}
+
+// Detach removes c if it is still the registered stream of its agent.
+func (h *Hub) Detach(c *Conn) {
+	h.mu.Lock()
+	if h.conns[c.AgentID] == c {
+		delete(h.conns, c.AgentID)
+	}
+	h.mu.Unlock()
+	c.close("stream ended")
+}
+
+// Connected reports whether the agent has an open stream.
+func (h *Hub) Connected(agentID string) bool {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	_, ok := h.conns[agentID]
+	return ok
+}
+
+// Disconnect closes the agent's stream, if any.
+func (h *Hub) Disconnect(agentID, reason string) {
+	h.mu.Lock()
+	c := h.conns[agentID]
+	delete(h.conns, agentID)
+	h.mu.Unlock()
+	if c != nil {
+		c.close(reason)
+	}
+}
+
+// Count returns the number of open streams per role.
+func (h *Hub) Count() map[string]int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	out := map[string]int{}
+	for _, c := range h.conns {
+		out[c.Role]++
+	}
+	return out
+}
+
+// NotifyRevision sends an ApplyRevisionCommand to every connected agent in
+// auto_pull mode and returns how many were notified.
+func (h *Hub) NotifyRevision(revision uint64) int {
+	h.mu.Lock()
+	targets := make([]*Conn, 0, len(h.conns))
+	for _, c := range h.conns {
+		if c.Mode == ModeAutoPull {
+			targets = append(targets, c)
+		}
+	}
+	h.mu.Unlock()
+	n := 0
+	for _, c := range targets {
+		cmd := &agentv1.Command{
+			CommandId: NewID(),
+			Action: &agentv1.Command_ApplyRevision{ApplyRevision: &agentv1.ApplyRevisionCommand{
+				Revision: revision, DeploymentId: NewID(),
+			}},
+		}
+		select {
+		case c.cmds <- cmd:
+			n++
+		default: // queue full: the agent will pull
+		}
+	}
+	return n
+}
+
+// NewID returns a random UUID (version 4) in canonical lowercase form.
+func NewID() string {
+	var b [16]byte
+	_, _ = rand.Read(b[:]) // never fails (crypto/rand)
+	b[6] = b[6]&0x0f | 0x40
+	b[8] = b[8]&0x3f | 0x80
+	return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16])
+}

server/internal/agenthub/hub_test.go

@@ -0,0 +1,63 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package agenthub
+
+import (
+	"testing"
+
+	"git.beugsiud.de/beug.k/DDI/internal/agentpki"
+)
+
+func TestHub(t *testing.T) {
+	h := New()
+	a := h.Attach("a", "dns", ModeAutoPull)
+	b := h.Attach("b", "dhcp", ModeManual)
+	if !h.Connected("a") || !h.Connected("b") || h.Connected("c") {
+		t.Fatal("connected state")
+	}
+	if c := h.Count(); c["dns"] != 1 || c["dhcp"] != 1 {
+		t.Errorf("count %v", c)
+	}
+
+	if n := h.NotifyRevision(7); n != 1 {
+		t.Errorf("notified %d", n)
+	}
+	cmd := <-a.Commands()
+	if cmd.GetApplyRevision().GetRevision() != 7 || !agentpki.ValidAgentID(cmd.GetCommandId()) ||
+		!agentpki.ValidAgentID(cmd.GetApplyRevision().GetDeploymentId()) {
+		t.Errorf("command %v", cmd)
+	}
+	select {
+	case <-b.Commands():
+		t.Error("manual agent notified")
+	default:
+	}
+
+	// A full queue drops commands instead of blocking.
+	for range queueSize + 3 {
+		h.NotifyRevision(8)
+	}
+	if len(a.cmds) != queueSize {
+		t.Errorf("queue length %d", len(a.cmds))
+	}
+
+	// A newer stream replaces the old one.
+	a2 := h.Attach("a", "dns", ModeAutoPull)
+	if a.Reason() == "" {
+		t.Error("old stream not closed")
+	}
+	h.Detach(a) // stale detach must not remove the new stream
+	if !h.Connected("a") {
+		t.Error("stale detach removed the new stream")
+	}
+
+	h.Disconnect("a", "revoked")
+	if a2.Reason() != "revoked" || h.Connected("a") {
+		t.Errorf("disconnect: %q", a2.Reason())
+	}
+	h.Detach(b)
+	if h.Connected("b") || len(h.Count()) != 0 {
+		t.Error("detach")
+	}
+	h.Disconnect("unknown", "x") // no-op
+}

server/internal/service/agentplane.go

@@ -0,0 +1,453 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package service
+
+import (
+	"context"
+	"crypto/subtle"
+	"crypto/x509"
+	"encoding/json"
+	"errors"
+	"math"
+	"net/netip"
+	"strings"
+	"time"
+
+	"git.beugsiud.de/beug.k/DDI/internal/bundle"
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
+	"git.beugsiud.de/beug.k/DDI/server/internal/pki"
+	"git.beugsiud.de/beug.k/DDI/server/internal/store"
+)
+
+// The methods in this file serve the agent API (data plane). They never
+// depend on the license: agents keep working in read-only mode.
+
+// Errors of the agent API.
+var (
+	ErrInvalidToken      = newError(KindUnauthorized, "invalid_token", "invalid, used or expired enrollment token")
+	ErrTokenRoleMismatch = newError(KindForbidden, "token_role_mismatch", "the enrollment token is for another role")
+	ErrInvalidCSR        = newError(KindInvalid, "invalid_csr", "invalid certificate signing request")
+	ErrAgentRevoked      = newError(KindForbidden, "agent_revoked", "the agent is revoked or unknown")
+	ErrNotDHCPAgent      = newError(KindForbidden, "not_dhcp_agent", "only DHCP agents report leases")
+)
+
+// Limits for values reported by agents.
+const (
+	maxHostnameLen = 253
+	maxVersionLen  = 100
+	maxMessageLen  = 64 << 10
+)
+
+// Agent intervals sent in Welcome/GetDesiredState (ADR 0021).
+const (
+	PullInterval      = 60 * time.Second
+	HeartbeatInterval = 30 * time.Second
+)
+
+// EnrollInput is an enrollment request of an agent.
+type EnrollInput struct {
+	Token        string
+	CSR          []byte
+	Role         string
+	Hostname     string
+	AgentVersion string
+	RemoteIP     *netip.Addr
+	RequestID    string
+}
+
+// EnrollResult is the outcome of a successful enrollment.
+type EnrollResult struct {
+	AgentID     string
+	Certificate *x509.Certificate
+}
+
+// EnrollAgent checks the one-time token, signs the CSR and registers the
+// agent. Attempts are rate limited per source IP.
+func (s *Service) EnrollAgent(ctx context.Context, in EnrollInput) (*EnrollResult, error) {
+	if s.opts.PKI == nil {
+		return nil, ErrAgentAPIUnavailable
+	}
+	ipKey := "unknown"
+	if in.RemoteIP != nil {
+		ipKey = in.RemoteIP.String()
+	}
+	if ok, wait := s.enrollLimiter.Allow(ipKey); !ok {
+		e := newError(KindRateLimited, "rate_limited", "too many enrollment attempts")
+		e.RetryAfter = int(math.Ceil(wait.Seconds()))
+		return nil, e
+	}
+	if !validRole(in.Role) {
+		return nil, invalidField("invalid_role", "role")
+	}
+	id, secret, ok := strings.Cut(in.Token, ".")
+	if !ok || !tokenIDPattern.MatchString(id) || secret == "" || len(secret) > 128 {
+		return nil, ErrInvalidToken
+	}
+	hash := hashTokenSecret(secret)
+	now := s.now()
+	var res EnrollResult
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		tk, err := tx.GetEnrollmentToken(ctx, id, true)
+		if errors.Is(err, store.ErrNotFound) {
+			return ErrInvalidToken
+		}
+		if err != nil {
+			return err
+		}
+		if subtle.ConstantTimeCompare(tk.SecretHash, hash) != 1 || tk.UsedAt != nil || !now.Before(tk.ExpiresAt) {
+			return ErrInvalidToken
+		}
+		if tk.Role != in.Role {
+			return ErrTokenRoleMismatch
+		}
+		agentID := agenthub.NewID()
+		cert, err := s.opts.PKI.SignAgentCSR(in.CSR, agentID)
+		if errors.Is(err, pki.ErrInvalidCSR) {
+			e := *ErrInvalidCSR
+			e.err = err
+			return &e
+		}
+		if err != nil {
+			return err
+		}
+		a := &store.Agent{ID: agentID, Role: in.Role, Hostname: cleanText(in.Hostname, maxHostnameLen),
+			AgentVersion: cleanText(in.AgentVersion, maxVersionLen), RemoteAddress: in.RemoteIP,
+			CertSerial: pki.SerialHex(cert.SerialNumber), CertNotAfter: cert.NotAfter}
+		if err := tx.InsertAgent(ctx, a); err != nil {
+			return err
+		}
+		if err := tx.MarkEnrollmentTokenUsed(ctx, id, agentID, now); err != nil {
+			if errors.Is(err, store.ErrNotFound) {
+				return ErrInvalidToken
+			}
+			return err
+		}
+		actor := &Actor{Username: "enrollment-token:" + id, IP: in.RemoteIP, RequestID: in.RequestID}
+		res = EnrollResult{AgentID: agentID, Certificate: cert}
+		return s.audit(ctx, tx, actor, "agent.enroll", "agent", agentID, a.Hostname, nil, s.agentDTO(a, 0, nil))
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &res, nil
+}
+
+// AgentIdentity is what the agent API checks on every call.
+type AgentIdentity struct {
+	ID         string
+	Role       string
+	Mode       string
+	CertSerial string
+	Revoked    bool
+}
+
+// AgentIdentity loads the identity of an agent.
+func (s *Service) AgentIdentity(ctx context.Context, id string) (*AgentIdentity, error) {
+	if !ValidUUID(id) {
+		return nil, ErrNotFound
+	}
+	a, err := s.st.Q().GetAgent(ctx, id, false)
+	if err != nil {
+		return nil, translate(err)
+	}
+	return &AgentIdentity{ID: a.ID, Role: a.Role, Mode: a.Mode, CertSerial: a.CertSerial, Revoked: a.RevokedAt != nil}, nil
+}
+
+// RenewAgentCertificate signs a new CSR for an agent and makes the new
+// certificate the only valid one.
+func (s *Service) RenewAgentCertificate(ctx context.Context, agentID string, csr []byte) (*x509.Certificate, error) {
+	if s.opts.PKI == nil {
+		return nil, ErrAgentAPIUnavailable
+	}
+	var cert *x509.Certificate
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		a, err := tx.GetAgent(ctx, agentID, true)
+		if errors.Is(err, store.ErrNotFound) {
+			return ErrAgentRevoked
+		}
+		if err != nil {
+			return err
+		}
+		if a.RevokedAt != nil {
+			return ErrAgentRevoked
+		}
+		cert, err = s.opts.PKI.SignAgentCSR(csr, agentID)
+		if errors.Is(err, pki.ErrInvalidCSR) {
+			e := *ErrInvalidCSR
+			e.err = err
+			return &e
+		}
+		if err != nil {
+			return err
+		}
+		return tx.UpdateAgentCertificate(ctx, agentID, pki.SerialHex(cert.SerialNumber), cert.NotAfter)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return cert, nil
+}
+
+// AgentSettings tell an agent what to do.
+type AgentSettings struct {
+	Mode            string
+	DesiredRevision uint64
+}
+
+// HelloInput is the status an agent reports when it opens its stream.
+type HelloInput struct {
+	Hostname         string
+	AgentVersion     string
+	ServiceVersion   string
+	AppliedRevision  uint64
+	LastGoodRevision uint64
+	Health           *AgentHealth
+	RemoteIP         *netip.Addr
+}
+
+func revisionPtr(n uint64) *int64 {
+	v := int64(min(n, store.MaxRevision)) //nolint:gosec // capped
+	return &v
+}
+
+func healthJSON(h *AgentHealth) json.RawMessage {
+	if h == nil {
+		return nil
+	}
+	c := *h
+	c.Message = cleanText(c.Message, 1000)
+	if c.Checks == nil {
+		c.Checks = []HealthCheck{}
+	}
+	for i := range c.Checks {
+		c.Checks[i].Name = cleanText(c.Checks[i].Name, 100)
+		c.Checks[i].Message = cleanText(c.Checks[i].Message, 1000)
+	}
+	raw, err := json.Marshal(c)
+	if err != nil {
+		return nil
+	}
+	return raw
+}
+
+// settings returns the agent's mode and desired revision.
+func (s *Service) settings(ctx context.Context, tx *store.Tx, a *store.Agent) (*AgentSettings, error) {
+	desired, err := latestRevisionNumber(ctx, tx)
+	if err != nil {
+		return nil, translate(err)
+	}
+	return &AgentSettings{Mode: a.Mode, DesiredRevision: uint64(desired)}, nil //nolint:gosec // non-negative
+}
+
+func agentStatusErr(err error) error {
+	if errors.Is(err, store.ErrNotFound) {
+		return ErrAgentRevoked
+	}
+	return translate(err)
+}
+
+// AgentHello records the status sent with Hello and returns the settings
+// for Welcome.
+func (s *Service) AgentHello(ctx context.Context, agentID string, in HelloInput) (*AgentSettings, error) {
+	host := cleanText(in.Hostname, maxHostnameLen)
+	av := cleanText(in.AgentVersion, maxVersionLen)
+	sv := cleanText(in.ServiceVersion, maxVersionLen)
+	st := store.AgentStatus{AgentVersion: &av, ServiceVersion: &sv, AppliedRevision: revisionPtr(in.AppliedRevision),
+		LastGoodRevision: revisionPtr(in.LastGoodRevision), Health: healthJSON(in.Health), RemoteAddress: in.RemoteIP,
+		SeenAt: s.now()}
+	if host != "" {
+		st.Hostname = &host
+	}
+	q := s.st.Q()
+	a, err := q.UpdateAgentStatus(ctx, agentID, st)
+	if err != nil {
+		return nil, agentStatusErr(err)
+	}
+	return s.settings(ctx, q, a)
+}
+
+// HeartbeatInput is a periodic status report.
+type HeartbeatInput struct {
+	AppliedRevision  uint64
+	LastGoodRevision uint64
+	Health           *AgentHealth
+}
+
+// AgentHeartbeat records a heartbeat.
+func (s *Service) AgentHeartbeat(ctx context.Context, agentID string, in HeartbeatInput) error {
+	_, err := s.st.Q().UpdateAgentStatus(ctx, agentID, store.AgentStatus{AppliedRevision: revisionPtr(in.AppliedRevision),
+		LastGoodRevision: revisionPtr(in.LastGoodRevision), Health: healthJSON(in.Health), SeenAt: s.now()})
+	if err != nil {
+		return agentStatusErr(err)
+	}
+	return nil
+}
+
+// AgentDesiredState records the contact and returns the agent's settings
+// (polling fallback).
+func (s *Service) AgentDesiredState(ctx context.Context, agentID string, remote *netip.Addr) (*AgentSettings, error) {
+	q := s.st.Q()
+	a, err := q.UpdateAgentStatus(ctx, agentID, store.AgentStatus{RemoteAddress: remote, SeenAt: s.now()})
+	if err != nil {
+		return nil, agentStatusErr(err)
+	}
+	return s.settings(ctx, q, a)
+}
+
+// RevisionArtifacts returns the files of a revision for one role and the
+// revision time (the bundle's created_at).
+func (s *Service) RevisionArtifacts(ctx context.Context, number uint64, role string) (time.Time, []bundle.Artifact, error) {
+	if number < 1 || number > store.MaxRevision {
+		return time.Time{}, nil, ErrNotFound
+	}
+	q := s.st.Q()
+	r, err := q.GetRevision(ctx, int64(number))
+	if err != nil {
+		return time.Time{}, nil, translate(err)
+	}
+	files, err := q.RevisionFiles(ctx, int64(number), role)
+	if err != nil {
+		return time.Time{}, nil, translate(err)
+	}
+	arts := make([]bundle.Artifact, len(files))
+	for i, f := range files {
+		arts[i] = bundle.Artifact{Path: f.Path, Mode: f.Mode, Content: f.Content, CreateOnly: f.CreateOnly}
+	}
+	return r.CreatedAt, arts, nil
+}
+
+// DeploymentReport is the outcome of applying a revision on an agent.
+type DeploymentReport struct {
+	CorrelationID  string
+	Revision       uint64
+	Trigger        string // pull, push
+	Outcome        string // succeeded, rejected, rolled_back, failed
+	Message        string
+	ActiveRevision uint64
+	StartedAt      time.Time
+	FinishedAt     time.Time
+}
+
+var (
+	validTriggers = map[string]bool{"pull": true, "push": true}
+	validOutcomes = map[string]bool{"succeeded": true, "rejected": true, "rolled_back": true, "failed": true}
+)
+
+// RecordDeployment stores a deployment report and updates the agent's
+// applied and last good revision.
+func (s *Service) RecordDeployment(ctx context.Context, agentID string, in DeploymentReport) error {
+	if in.Revision < 1 || in.Revision > store.MaxRevision || in.ActiveRevision > store.MaxRevision {
+		return invalidField("invalid_revision", "revision")
+	}
+	if !validTriggers[in.Trigger] || !validOutcomes[in.Outcome] {
+		return invalidField("invalid_value", "outcome")
+	}
+	now := s.now()
+	if in.FinishedAt.IsZero() {
+		in.FinishedAt = now
+	}
+	if in.StartedAt.IsZero() || in.StartedAt.After(in.FinishedAt) {
+		in.StartedAt = in.FinishedAt
+	}
+	st := store.AgentStatus{AppliedRevision: revisionPtr(in.ActiveRevision), SeenAt: now}
+	if in.Outcome == "succeeded" {
+		st.LastGoodRevision = revisionPtr(in.Revision)
+	}
+	return s.inTx(ctx, func(tx *store.Tx) error {
+		if _, err := tx.UpdateAgentStatus(ctx, agentID, st); err != nil {
+			return agentStatusErr(err)
+		}
+		d := &store.Deployment{
+			AgentID: agentID, CorrelationID: cleanText(in.CorrelationID, 100), Trigger: in.Trigger, Outcome: in.Outcome,
+			Message: cleanMessage(in.Message), StartedAt: in.StartedAt, FinishedAt: in.FinishedAt,
+			Revision:       int64(in.Revision),       //nolint:gosec // at most MaxRevision (checked above)
+			ActiveRevision: int64(in.ActiveRevision), //nolint:gosec // at most MaxRevision (checked above)
+		}
+		return tx.InsertDeployment(ctx, d)
+	})
+}
+
+// cleanMessage keeps line breaks (checker output) but drops other control
+// characters and bounds the size.
+func cleanMessage(s string) string {
+	s = strings.Map(func(r rune) rune {
+		if r == '\n' || r == '\t' {
+			return r
+		}
+		if r < 0x20 || r == 0x7f {
+			return -1
+		}
+		return r
+	}, strings.ToValidUTF8(s, "?"))
+	if len(s) > maxMessageLen {
+		s = strings.ToValidUTF8(s[:maxMessageLen], "")
+	}
+	return s
+}
+
+// LeaseInput is one lease of a batch.
+type LeaseInput struct {
+	IP        string
+	MAC       string
+	Hostname  string
+	State     string // active, expired, released, declined
+	ExpiresAt *time.Time
+	UpdatedAt *time.Time
+	SubnetID  uint32
+	ClientID  string
+}
+
+var validLeaseStates = map[string]bool{"active": true, "expired": true, "released": true, "declined": true}
+
+// LeaseBatchResult says what happened to a batch.
+type LeaseBatchResult struct {
+	Acknowledged uint64
+	// Duplicate is true if the batch was dropped as already processed.
+	Duplicate bool
+	// Skipped counts invalid leases that were ignored.
+	Skipped int
+}
+
+// StoreLeaseBatch stores a lease batch of a DHCP agent. Batches with a
+// sequence not above the last processed one are dropped; a full sync
+// replaces the agent's lease set.
+func (s *Service) StoreLeaseBatch(ctx context.Context, agentID string, seq uint64, fullSync bool, in []LeaseInput) (*LeaseBatchResult, error) {
+	if seq == 0 || seq > math.MaxInt64 {
+		return nil, invalidField("invalid_sequence", "sequence")
+	}
+	res := &LeaseBatchResult{}
+	leases := make([]store.Lease, 0, len(in))
+	seen := map[netip.Addr]bool{}
+	for _, l := range in {
+		ip, err := netip.ParseAddr(strings.TrimSpace(l.IP))
+		if err != nil || ip.Zone() != "" || !validLeaseStates[l.State] || seen[ip.Unmap()] {
+			res.Skipped++
+			continue
+		}
+		ip = ip.Unmap()
+		seen[ip] = true
+		leases = append(leases, store.Lease{IP: ip, MAC: strings.ToLower(cleanText(l.MAC, 128)),
+			Hostname: cleanText(l.Hostname, maxHostnameLen), State: l.State, ExpiresAt: l.ExpiresAt, UpdatedAt: l.UpdatedAt,
+			SubnetID: l.SubnetID, ClientID: cleanText(l.ClientID, 255)})
+	}
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		stored, role, err := tx.LockLeaseSequence(ctx, agentID)
+		if err != nil {
+			return agentStatusErr(err)
+		}
+		if role != RoleDHCP {
+			return ErrNotDHCPAgent
+		}
+		if int64(seq) <= stored { //nolint:gosec // checked above
+			res.Duplicate = true
+			res.Acknowledged = uint64(stored) //nolint:gosec // non-negative by constraint
+			return nil
+		}
+		res.Acknowledged = seq
+		return tx.StoreLeases(ctx, agentID, int64(seq), fullSync, leases, s.now()) //nolint:gosec // checked above
+	})
+	if err != nil {
+		return nil, err
+	}
+	return res, nil
+}

server/internal/service/agents.go

@@ -0,0 +1,452 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package service
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"net"
+	"regexp"
+	"slices"
+	"strings"
+	"time"
+	"unicode"
+	"unicode/utf8"
+
+	"git.beugsiud.de/beug.k/DDI/internal/brand"
+	"git.beugsiud.de/beug.k/DDI/server/internal/store"
+)
+
+// Agent roles.
+const (
+	RoleDNS  = "dns"
+	RoleDHCP = "dhcp"
+)
+
+func validRole(r string) bool { return r == RoleDNS || r == RoleDHCP }
+
+// uuidPattern matches UUIDs as PostgreSQL prints them; IDs that do not
+// match cannot exist and are reported as not found (instead of a database
+// cast error).
+var uuidPattern = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
+
+// ValidUUID reports whether s is a canonical lowercase UUID.
+func ValidUUID(s string) bool { return uuidPattern.MatchString(s) }
+
+// ErrAgentAPIUnavailable is returned when the agent PKI is not set up.
+var ErrAgentAPIUnavailable = newError(KindUnavailable, "agent_api_unavailable", "the agent API is not configured")
+
+// HealthCheck is one named check reported by an agent.
+type HealthCheck struct {
+	Name    string `json:"name"`
+	State   string `json:"state"`
+	Message string `json:"message"`
+}
+
+// AgentHealth is the health of the managed service as reported by an agent.
+type AgentHealth struct {
+	State     string        `json:"state"`
+	Message   string        `json:"message"`
+	CheckedAt *time.Time    `json:"checked_at,omitempty"`
+	Checks    []HealthCheck `json:"checks"`
+}
+
+// Health states.
+const (
+	HealthUnknown  = "unknown"
+	HealthOK       = "ok"
+	HealthDegraded = "degraded"
+	HealthFailed   = "failed"
+)
+
+func decodeHealth(raw []byte) AgentHealth {
+	var h AgentHealth
+	if len(raw) == 0 || json.Unmarshal(raw, &h) != nil || h.State == "" {
+		h = AgentHealth{State: HealthUnknown}
+	}
+	if h.Checks == nil {
+		h.Checks = []HealthCheck{}
+	}
+	return h
+}
+
+// DeploymentDTO is a deployment result reported by an agent.
+type DeploymentDTO struct {
+	ID             int64     `json:"id"`
+	AgentID        string    `json:"agent_id"`
+	AgentHostname  string    `json:"agent_hostname"`
+	Revision       int64     `json:"revision"`
+	Trigger        string    `json:"trigger"`
+	Outcome        string    `json:"outcome"`
+	Message        string    `json:"message"`
+	ActiveRevision int64     `json:"active_revision"`
+	StartedAt      time.Time `json:"started_at"`
+	FinishedAt     time.Time `json:"finished_at"`
+}
+
+func deploymentDTO(d *store.Deployment) DeploymentDTO {
+	return DeploymentDTO{ID: d.ID, AgentID: d.AgentID, AgentHostname: d.AgentHostname, Revision: d.Revision,
+		Trigger: d.Trigger, Outcome: d.Outcome, Message: d.Message, ActiveRevision: d.ActiveRevision,
+		StartedAt: d.StartedAt, FinishedAt: d.FinishedAt}
+}
+
+// AgentDTO is an enrolled agent.
+type AgentDTO struct {
+	ID               string         `json:"id"`
+	Role             string         `json:"role"`
+	Hostname         string         `json:"hostname"`
+	AgentVersion     string         `json:"agent_version"`
+	ServiceVersion   string         `json:"service_version"`
+	Connected        bool           `json:"connected"`
+	RemoteAddress    string         `json:"remote_address,omitempty"`
+	Mode             string         `json:"mode"`
+	EnrolledAt       time.Time      `json:"enrolled_at"`
+	LastSeenAt       *time.Time     `json:"last_seen_at,omitempty"`
+	CertExpiresAt    time.Time      `json:"cert_expires_at"`
+	AppliedRevision  int64          `json:"applied_revision"`
+	LastGoodRevision int64          `json:"last_good_revision"`
+	DesiredRevision  int64          `json:"desired_revision"`
+	Health           AgentHealth    `json:"health"`
+	RevokedAt        *time.Time     `json:"revoked_at,omitempty"`
+	LastDeployment   *DeploymentDTO `json:"last_deployment,omitempty"`
+}
+
+func (s *Service) agentDTO(a *store.Agent, desired int64, last *store.Deployment) AgentDTO {
+	d := AgentDTO{
+		ID: a.ID, Role: a.Role, Hostname: a.Hostname, AgentVersion: a.AgentVersion, ServiceVersion: a.ServiceVersion,
+		Mode: a.Mode, EnrolledAt: a.EnrolledAt, LastSeenAt: a.LastSeenAt, CertExpiresAt: a.CertNotAfter,
+		AppliedRevision: a.AppliedRevision, LastGoodRevision: a.LastGoodRevision, DesiredRevision: desired,
+		Health: decodeHealth(a.Health), RevokedAt: a.RevokedAt,
+	}
+	if a.RemoteAddress != nil {
+		d.RemoteAddress = a.RemoteAddress.String()
+	}
+	if s.opts.Hub != nil {
+		d.Connected = s.opts.Hub.Connected(a.ID)
+	}
+	if last != nil {
+		dd := deploymentDTO(last)
+		d.LastDeployment = &dd
+	}
+	if a.RevokedAt != nil {
+		d.DesiredRevision = 0
+	}
+	return d
+}
+
+// latestRevisionNumber returns the newest revision number, 0 if none. In
+// M3 it is the desired revision of every agent (ADR 0021).
+func latestRevisionNumber(ctx context.Context, tx *store.Tx) (int64, error) {
+	r, err := tx.LatestRevision(ctx)
+	if errors.Is(err, store.ErrNotFound) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	return r.Number, nil
+}
+
+// ListAgents returns all agents, including revoked ones.
+func (s *Service) ListAgents(ctx context.Context) ([]AgentDTO, error) {
+	q := s.st.Q()
+	agents, err := q.ListAgents(ctx)
+	if err != nil {
+		return nil, translate(err)
+	}
+	desired, err := latestRevisionNumber(ctx, q)
+	if err != nil {
+		return nil, translate(err)
+	}
+	last, err := q.LatestDeployments(ctx)
+	if err != nil {
+		return nil, translate(err)
+	}
+	out := make([]AgentDTO, 0, len(agents))
+	for _, a := range agents {
+		out = append(out, s.agentDTO(a, desired, last[a.ID]))
+	}
+	return out, nil
+}
+
+// GetAgent returns one agent.
+func (s *Service) GetAgent(ctx context.Context, id string) (AgentDTO, error) {
+	if !ValidUUID(id) {
+		return AgentDTO{}, ErrNotFound
+	}
+	return s.loadAgentDTO(ctx, s.st.Q(), id)
+}
+
+func (s *Service) loadAgentDTO(ctx context.Context, tx *store.Tx, id string) (AgentDTO, error) {
+	a, err := tx.GetAgent(ctx, id, false)
+	if err != nil {
+		return AgentDTO{}, translate(err)
+	}
+	desired, err := latestRevisionNumber(ctx, tx)
+	if err != nil {
+		return AgentDTO{}, translate(err)
+	}
+	last, err := tx.LatestDeployment(ctx, id)
+	if err != nil && !errors.Is(err, store.ErrNotFound) {
+		return AgentDTO{}, translate(err)
+	}
+	return s.agentDTO(a, desired, last), nil
+}
+
+// RevokeAgent revokes the agent's certificate and closes its stream.
+func (s *Service) RevokeAgent(ctx context.Context, a *Actor, id string) (AgentDTO, error) {
+	if err := s.requireWritable(); err != nil {
+		return AgentDTO{}, err
+	}
+	if !ValidUUID(id) {
+		return AgentDTO{}, ErrNotFound
+	}
+	var out AgentDTO
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		before, err := tx.GetAgent(ctx, id, true)
+		if err != nil {
+			return err
+		}
+		if _, err := tx.RevokeAgent(ctx, id, s.now()); err != nil {
+			return err
+		}
+		if out, err = s.loadAgentDTO(ctx, tx, id); err != nil {
+			return err
+		}
+		if before.RevokedAt != nil {
+			return nil // already revoked: nothing changed
+		}
+		return s.audit(ctx, tx, a, "agent.revoke", "agent", id, before.Hostname, s.agentDTO(before, 0, nil), out)
+	})
+	if err != nil {
+		return AgentDTO{}, err
+	}
+	if s.opts.Hub != nil {
+		s.opts.Hub.Disconnect(id, "agent revoked")
+		out.Connected = false
+	}
+	return out, nil
+}
+
+// DeleteAgent removes an agent with its deployments and leases. Its
+// certificate becomes worthless with the row.
+func (s *Service) DeleteAgent(ctx context.Context, a *Actor, id string) error {
+	if err := s.requireWritable(); err != nil {
+		return err
+	}
+	if !ValidUUID(id) {
+		return ErrNotFound
+	}
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		before, err := tx.GetAgent(ctx, id, true)
+		if err != nil {
+			return err
+		}
+		if err := tx.DeleteAgent(ctx, id); err != nil {
+			return err
+		}
+		return s.audit(ctx, tx, a, "agent.delete", "agent", id, before.Hostname, s.agentDTO(before, 0, nil), nil)
+	})
+	if err != nil {
+		return err
+	}
+	if s.opts.Hub != nil {
+		s.opts.Hub.Disconnect(id, "agent deleted")
+	}
+	return nil
+}
+
+// Enrollment tokens (ADR 0021): "<id>.<secret>", id 16 hex digits, secret
+// 32 random bytes in base64url. Only SHA-256 of the secret is stored.
+const (
+	defaultTokenHours = 24
+	maxTokenHours     = 720
+	// tokenListWindow keeps used and expired tokens visible for a while.
+	tokenListWindow = 7 * 24 * time.Hour
+)
+
+var tokenIDPattern = regexp.MustCompile(`^[0-9a-f]{16}$`)
+
+// EnrollmentTokenDTO is a token without its secret.
+type EnrollmentTokenDTO struct {
+	ID            string     `json:"id"`
+	Role          string     `json:"role"`
+	Description   string     `json:"description"`
+	CreatedAt     time.Time  `json:"created_at"`
+	CreatedByName string     `json:"created_by_name"`
+	ExpiresAt     time.Time  `json:"expires_at"`
+	UsedAt        *time.Time `json:"used_at,omitempty"`
+	AgentID       string     `json:"agent_id,omitempty"`
+}
+
+func tokenDTO(t *store.EnrollmentToken) EnrollmentTokenDTO {
+	d := EnrollmentTokenDTO{ID: t.ID, Role: t.Role, Description: t.Description, CreatedAt: t.CreatedAt,
+		CreatedByName: t.CreatedByName, ExpiresAt: t.ExpiresAt, UsedAt: t.UsedAt}
+	if t.AgentID != nil {
+		d.AgentID = *t.AgentID
+	}
+	return d
+}
+
+// EnrollmentTokenCreated is returned once when a token is created.
+type EnrollmentTokenCreated struct {
+	Token         EnrollmentTokenDTO `json:"token"`
+	Secret        string             `json:"secret"`
+	CAFingerprint string             `json:"ca_fingerprint"`
+	ServerAddress string             `json:"server_address"`
+	Command       string             `json:"command"`
+}
+
+// TokenInput describes a new enrollment token.
+type TokenInput struct {
+	Role        string
+	Description string
+	// ValidHours defaults to 24.
+	ValidHours *int
+}
+
+func hashTokenSecret(secret string) []byte {
+	sum := sha256.Sum256([]byte(secret))
+	return sum[:]
+}
+
+// AgentBinary returns the agent binary name of a role.
+func AgentBinary(role string) string {
+	if role == RoleDHCP {
+		return brand.AgentDHCPBinary
+	}
+	return brand.AgentDNSBinary
+}
+
+// enrollCommand builds the command shown to the administrator.
+func (s *Service) enrollCommand(role, token string) string {
+	addr := s.opts.AgentAPIAddress
+	cmd := AgentBinary(role) + " enroll --server " + addr + " --token " + token + " --ca-hash " + s.opts.PKI.Fingerprint()
+	host, _, err := net.SplitHostPort(addr)
+	if err == nil && len(s.opts.AgentServerNames) > 0 && !slices.Contains(s.opts.AgentServerNames, host) {
+		cmd += " --server-name " + s.opts.AgentServerNames[0]
+	}
+	return cmd
+}
+
+// CreateEnrollmentToken creates a one-time token for a role.
+func (s *Service) CreateEnrollmentToken(ctx context.Context, a *Actor, in TokenInput) (*EnrollmentTokenCreated, error) {
+	if err := s.requireWritable(); err != nil {
+		return nil, err
+	}
+	if s.opts.PKI == nil || s.opts.AgentAPIAddress == "" {
+		return nil, ErrAgentAPIUnavailable
+	}
+	var problems []Problem
+	if !validRole(in.Role) {
+		problems = append(problems, Problem{Code: "invalid_role", Field: "role"})
+	}
+	desc := strings.TrimSpace(in.Description)
+	if utf8.RuneCountInString(desc) > 200 {
+		problems = append(problems, Problem{Code: "too_long", Field: "description"})
+	}
+	hours := defaultTokenHours
+	if in.ValidHours != nil {
+		hours = *in.ValidHours
+	}
+	if hours < 1 || hours > maxTokenHours {
+		problems = append(problems, Problem{Code: "out_of_range", Field: "valid_hours"})
+	}
+	if len(problems) > 0 {
+		return nil, invalid(problems...)
+	}
+
+	var idBytes [8]byte
+	var secretBytes [32]byte
+	if _, err := rand.Read(idBytes[:]); err != nil {
+		return nil, err
+	}
+	if _, err := rand.Read(secretBytes[:]); err != nil {
+		return nil, err
+	}
+	secret := base64.RawURLEncoding.EncodeToString(secretBytes[:])
+	now := s.now().UTC().Truncate(time.Microsecond)
+	tk := &store.EnrollmentToken{
+		ID: hex.EncodeToString(idBytes[:]), SecretHash: hashTokenSecret(secret), Role: in.Role, Description: desc,
+		CreatedByName: a.Username, CreatedAt: now, ExpiresAt: now.Add(time.Duration(hours) * time.Hour),
+	}
+	if a.UserID != "" {
+		uid := a.UserID
+		tk.CreatedBy = &uid
+	}
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		if err := tx.InsertEnrollmentToken(ctx, tk); err != nil {
+			return err
+		}
+		return s.audit(ctx, tx, a, "enrollment_token.create", "enrollment_token", tk.ID, tk.Description, nil, tokenDTO(tk))
+	})
+	if err != nil {
+		return nil, err
+	}
+	full := tk.ID + "." + secret
+	return &EnrollmentTokenCreated{
+		Token: tokenDTO(tk), Secret: full, CAFingerprint: s.opts.PKI.Fingerprint(),
+		ServerAddress: s.opts.AgentAPIAddress, Command: s.enrollCommand(tk.Role, full),
+	}, nil
+}
+
+// ListEnrollmentTokens returns open tokens and those used or expired within
+// the last 7 days, newest first.
+func (s *Service) ListEnrollmentTokens(ctx context.Context) ([]EnrollmentTokenDTO, error) {
+	list, err := s.st.Q().ListEnrollmentTokens(ctx, s.now().Add(-tokenListWindow))
+	if err != nil {
+		return nil, translate(err)
+	}
+	out := make([]EnrollmentTokenDTO, 0, len(list))
+	for _, t := range list {
+		out = append(out, tokenDTO(t))
+	}
+	return out, nil
+}
+
+// DeleteEnrollmentToken removes a token; an unused token becomes invalid.
+func (s *Service) DeleteEnrollmentToken(ctx context.Context, a *Actor, id string) error {
+	if err := s.requireWritable(); err != nil {
+		return err
+	}
+	if !tokenIDPattern.MatchString(id) {
+		return ErrNotFound
+	}
+	return s.inTx(ctx, func(tx *store.Tx) error {
+		tk, err := tx.GetEnrollmentToken(ctx, id, true)
+		if err != nil {
+			return err
+		}
+		if err := tx.DeleteEnrollmentToken(ctx, id); err != nil {
+			return err
+		}
+		return s.audit(ctx, tx, a, "enrollment_token.delete", "enrollment_token", id, tk.Description, tokenDTO(tk), nil)
+	})
+}
+
+// CleanupEnrollmentTokens removes tokens used or expired more than 30 days
+// ago.
+func (s *Service) CleanupEnrollmentTokens(ctx context.Context) (int64, error) {
+	n, err := s.st.Q().DeleteEnrollmentTokensBefore(ctx, s.now().Add(-30*24*time.Hour))
+	return n, translate(err)
+}
+
+// cleanText trims s, removes control characters and cuts it to max runes
+// (values reported by agents).
+func cleanText(s string, maxRunes int) string {
+	s = strings.Map(func(r rune) rune {
+		if unicode.IsControl(r) {
+			return -1
+		}
+		return r
+	}, strings.TrimSpace(s))
+	if utf8.RuneCountInString(s) > maxRunes {
+		s = string([]rune(s)[:maxRunes])
+	}
+	return s
+}

server/internal/service/revisions.go

@@ -0,0 +1,348 @@
+// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
+
+package service
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"net/netip"
+	"strconv"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"git.beugsiud.de/beug.k/DDI/server/internal/configgen"
+	"git.beugsiud.de/beug.k/DDI/server/internal/store"
+)
+
+// Page sizes.
+const (
+	defaultRevisionPage   = 50
+	MaxRevisionPage       = 500
+	defaultDeploymentPage = 50
+	MaxDeploymentPage     = 500
+	defaultLeasePage      = 100
+)
+
+// ErrRevisionLimit is returned when the revision numbers are exhausted.
+var ErrRevisionLimit = newError(KindConflict, "revision_limit", "the highest revision number has been reached")
+
+// RevisionSummaryDTO is a revision without files.
+type RevisionSummaryDTO struct {
+	Number        int64     `json:"number"`
+	CreatedAt     time.Time `json:"created_at"`
+	CreatedByName string    `json:"created_by_name"`
+	Comment       string    `json:"comment"`
+	DNSFiles      int       `json:"dns_files"`
+	DHCPFiles     int       `json:"dhcp_files"`
+	Warnings      int       `json:"warnings"`
+}
+
+// RevisionDTO is a revision with its files.
+type RevisionDTO struct {
+	Number        int64               `json:"number"`
+	CreatedAt     time.Time           `json:"created_at"`
+	CreatedByName string              `json:"created_by_name"`
+	Comment       string              `json:"comment"`
+	DNS           []GeneratedFile     `json:"dns"`
+	DHCP          []GeneratedFile     `json:"dhcp"`
+	Warnings      []configgen.Warning `json:"warnings"`
+}
+
+// RevisionPage is a page of revisions, newest first.
+type RevisionPage struct {
+	Items      []RevisionSummaryDTO `json:"items"`
+	NextBefore *int64               `json:"next_before"`
+}
+
+func revisionDTO(r *store.Revision, files []store.RevisionFile) RevisionDTO {
+	d := RevisionDTO{Number: r.Number, CreatedAt: r.CreatedAt, CreatedByName: r.CreatedByName, Comment: r.Comment,
+		DNS: []GeneratedFile{}, DHCP: []GeneratedFile{}, Warnings: []configgen.Warning{}}
+	if len(r.Warnings) > 0 {
+		_ = json.Unmarshal(r.Warnings, &d.Warnings) // written by us; a broken value shows no warnings
+		if d.Warnings == nil {
+			d.Warnings = []configgen.Warning{}
+		}
+	}
+	for _, f := range files {
+		g := GeneratedFile{Path: f.Path, Content: string(f.Content), CreateOnly: f.CreateOnly}
+		if f.Role == RoleDHCP {
+			d.DHCP = append(d.DHCP, g)
+		} else {
+			d.DNS = append(d.DNS, g)
+		}
+	}
+	return d
+}
+
+// revisionFiles flattens a generator run into stored files (DNS first, as
+// the database returns them).
+func revisionFiles(g *Generated) []store.RevisionFile {
+	out := make([]store.RevisionFile, 0, len(g.DNS)+len(g.DHCP))
+	for _, a := range g.DNS {
+		out = append(out, store.RevisionFile{Role: RoleDNS, Path: a.Path, Mode: a.Mode, CreateOnly: a.CreateOnly, Content: a.Content})
+	}
+	for _, a := range g.DHCP {
+		out = append(out, store.RevisionFile{Role: RoleDHCP, Path: a.Path, Mode: a.Mode, CreateOnly: a.CreateOnly, Content: a.Content})
+	}
+	return out
+}
+
+func sameFiles(a, b []store.RevisionFile) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		x, y := a[i], b[i]
+		if x.Role != y.Role || x.Path != y.Path || x.Mode != y.Mode || x.CreateOnly != y.CreateOnly || !bytes.Equal(x.Content, y.Content) {
+			return false
+		}
+	}
+	return true
+}
+
+// generationProblem reports the first part that could not be generated.
+func generationProblem(g *Generated) error {
+	for _, p := range []struct {
+		part string
+		err  *GeneratorProblem
+	}{{RoleDNS, g.DNSError}, {RoleDHCP, g.DHCPError}} {
+		if p.err == nil {
+			continue
+		}
+		params := map[string]string{"part": p.part}
+		for k, v := range p.err.Params {
+			params[k] = v
+		}
+		return &Error{Kind: KindInvalid, Code: p.err.Code, Message: "the " + p.part + " configuration cannot be generated",
+			Params: params}
+	}
+	return nil
+}
+
+// CreateRevision generates the configuration and stores it as a new
+// revision (ADR 0021). If the output equals the latest revision, that one
+// is returned with created false. Agents in auto_pull mode are notified.
+func (s *Service) CreateRevision(ctx context.Context, a *Actor, comment string) (*RevisionDTO, bool, error) {
+	if err := s.requireWritable(); err != nil {
+		return nil, false, err
+	}
+	comment = strings.TrimSpace(comment)
+	if utf8.RuneCountInString(comment) > 500 {
+		return nil, false, invalidField("too_long", "comment")
+	}
+	var out RevisionDTO
+	created := false
+	err := s.inTx(ctx, func(tx *store.Tx) error {
+		if err := tx.LockRevisions(ctx); err != nil {
+			return err
+		}
+		latest, err := tx.LatestRevision(ctx)
+		if err != nil && !errors.Is(err, store.ErrNotFound) {
+			return err
+		}
+		next := int64(1)
+		if latest != nil {
+			// Generate with the latest number first: identical output
+			// means nothing changed since then.
+			g, err := s.generate(ctx, uint32(latest.Number)) //nolint:gosec // 1..2^32-1 by constraint
+			if err != nil {
+				return err
+			}
+			if err := generationProblem(g); err != nil {
+				return err
+			}
+			stored, err := tx.RevisionFiles(ctx, latest.Number, "")
+			if err != nil {
+				return err
+			}
+			if sameFiles(revisionFiles(g), stored) {
+				out = revisionDTO(latest, stored)
+				return nil
+			}
+			if latest.Number >= store.MaxRevision {
+				return ErrRevisionLimit
+			}
+			next = latest.Number + 1
+		}
+		g, err := s.generate(ctx, uint32(next)) //nolint:gosec // 1..2^32-1, checked above
+		if err != nil {
+			return err
+		}
+		if err := generationProblem(g); err != nil {
+			return err
+		}
+		warnings, err := json.Marshal(g.Warnings)
+		if err != nil {
+			return err
+		}
+		r := &store.Revision{Number: next, CreatedAt: s.now().UTC().Truncate(time.Microsecond), CreatedByName: a.Username,
+			Comment: comment, Warnings: warnings}
+		if a.UserID != "" {
+			uid := a.UserID
+			r.CreatedBy = &uid
+		}
+		files := revisionFiles(g)
+		if err := tx.InsertRevision(ctx, r, files); err != nil {
+			return err
+		}
+		out = revisionDTO(r, files)
+		created = true
+		summary := RevisionSummaryDTO{Number: r.Number, CreatedAt: r.CreatedAt, CreatedByName: r.CreatedByName,
+			Comment: r.Comment, DNSFiles: len(g.DNS), DHCPFiles: len(g.DHCP), Warnings: len(g.Warnings)}
+		return s.audit(ctx, tx, a, "revision.create", "revision", strconv.FormatInt(r.Number, 10),
+			"#"+strconv.FormatInt(r.Number, 10), nil, summary)
+	})
+	if err != nil {
+		return nil, false, err
+	}
+	if created && s.opts.Hub != nil {
+		n := s.opts.Hub.NotifyRevision(uint64(out.Number)) //nolint:gosec // positive
+		s.log.Info("revision created", "revision", out.Number, "notified_agents", n, "request_id", a.RequestID)
+	}
+	return &out, created, nil
+}
+
+// ListRevisions returns revisions newest first. before (0 = none) pages.
+func (s *Service) ListRevisions(ctx context.Context, limit int, before int64) (*RevisionPage, error) {
+	if limit <= 0 || limit > MaxRevisionPage {
+		limit = defaultRevisionPage
+	}
+	list, err := s.st.Q().ListRevisions(ctx, limit+1, before)
+	if err != nil {
+		return nil, translate(err)
+	}
+	page := &RevisionPage{Items: []RevisionSummaryDTO{}}
+	for i, r := range list {
+		if i == limit {
+			n := list[limit-1].Number
+			page.NextBefore = &n
+			break
+		}
+		page.Items = append(page.Items, RevisionSummaryDTO{Number: r.Number, CreatedAt: r.CreatedAt,
+			CreatedByName: r.CreatedByName, Comment: r.Comment, DNSFiles: r.DNSFiles, DHCPFiles: r.DHCPFiles,
+			Warnings: r.WarningCount})
+	}
+	return page, nil
+}
+
+// GetRevision returns a revision with its files.
+func (s *Service) GetRevision(ctx context.Context, number int64) (*RevisionDTO, error) {
+	if number < 1 || number > store.MaxRevision {
+		return nil, ErrNotFound
+	}
+	q := s.st.Q()
+	r, err := q.GetRevision(ctx, number)
+	if err != nil {
+		return nil, translate(err)
+	}
+	files, err := q.RevisionFiles(ctx, number, "")
+	if err != nil {
+		return nil, translate(err)
+	}
+	d := revisionDTO(r, files)
+	return &d, nil
+}
+
+// DeploymentPage is a page of deployments, newest first.
+type DeploymentPage struct {
+	Items        []DeploymentDTO `json:"items"`
+	NextBeforeID *int64          `json:"next_before_id"`
+}
+
+// DeploymentFilter selects deployments.
+type DeploymentFilter struct {
+	AgentID  string
+	Revision int64
+	BeforeID int64
+	Limit    int
+}
+
+// ListDeployments returns deployment reports newest first.
+func (s *Service) ListDeployments(ctx context.Context, f DeploymentFilter) (*DeploymentPage, error) {
+	page := &DeploymentPage{Items: []DeploymentDTO{}}
+	if f.AgentID != "" && !ValidUUID(f.AgentID) {
+		return page, nil // no such agent
+	}
+	limit := f.Limit
+	if limit <= 0 || limit > MaxDeploymentPage {
+		limit = defaultDeploymentPage
+	}
+	list, err := s.st.Q().ListDeployments(ctx, store.DeploymentFilter{AgentID: f.AgentID, Revision: f.Revision,
+		BeforeID: f.BeforeID, Limit: limit + 1})
+	if err != nil {
+		return nil, translate(err)
+	}
+	for i, d := range list {
+		if i == limit {
+			id := list[limit-1].ID
+			page.NextBeforeID = &id
+			break
+		}
+		page.Items = append(page.Items, deploymentDTO(d))
+	}
+	return page, nil
+}
+
+// LeaseDTO is a lease reported by a DHCP agent.
+type LeaseDTO struct {
+	IP         netip.Addr `json:"ip"`
+	MAC        string     `json:"mac"`
+	Hostname   string     `json:"hostname"`
+	State      string     `json:"state"`
+	ExpiresAt  *time.Time `json:"expires_at,omitempty"`
+	UpdatedAt  *time.Time `json:"updated_at,omitempty"`
+	SubnetID   uint32     `json:"subnet_id"`
+	NetworkID  string     `json:"network_id,omitempty"`
+	ClientID   string     `json:"client_id,omitempty"`
+	AgentID    string     `json:"agent_id"`
+	ReportedAt time.Time  `json:"reported_at"`
+}
+
+// LeasePage is a page of leases sorted by IP.
+type LeasePage struct {
+	Items     []LeaseDTO `json:"items"`
+	NextAfter *string    `json:"next_after"`
+}
+
+// LeaseFilter selects leases.
+type LeaseFilter struct {
+	NetworkID string
+	State     string
+	Query     string
+	After     *netip.Addr
+	Limit     int
+}
+
+// ListLeases returns leases sorted by IP. Leases of one IP reported by
+// several agents always share a page.
+func (s *Service) ListLeases(ctx context.Context, f LeaseFilter) (*LeasePage, error) {
+	page := &LeasePage{Items: []LeaseDTO{}}
+	if f.NetworkID != "" && !ValidUUID(f.NetworkID) {
+		return page, nil // no such network
+	}
+	limit := f.Limit
+	if limit <= 0 || limit > store.MaxLeasePage {
+		limit = defaultLeasePage
+	}
+	list, more, err := s.st.Q().ListLeases(ctx, store.LeaseFilter{NetworkID: f.NetworkID, State: f.State, Query: f.Query,
+		After: f.After, Limit: limit})
+	if err != nil {
+		return nil, translate(err)
+	}
+	for _, l := range list {
+		d := LeaseDTO{IP: l.IP, MAC: l.MAC, Hostname: l.Hostname, State: l.State, ExpiresAt: l.ExpiresAt,
+			UpdatedAt: l.UpdatedAt, SubnetID: l.SubnetID, ClientID: l.ClientID, AgentID: l.AgentID, ReportedAt: l.ReportedAt}
+		if l.NetworkID != nil {
+			d.NetworkID = *l.NetworkID
+		}
+		page.Items = append(page.Items, d)
+	}
+	if more && len(list) > 0 {
+		next := list[len(list)-1].IP.String()
+		page.NextAfter = &next
+	}
+	return page, nil
+}

server/internal/service/service.go

@@ -9,8 +9,10 @@ import (
 	"net/netip"
 	"time"
 
+	"git.beugsiud.de/beug.k/DDI/server/internal/agenthub"
 	"git.beugsiud.de/beug.k/DDI/server/internal/auth"
 	"git.beugsiud.de/beug.k/DDI/server/internal/license"
+	"git.beugsiud.de/beug.k/DDI/server/internal/pki"
 	"git.beugsiud.de/beug.k/DDI/server/internal/secrets"
 	"git.beugsiud.de/beug.k/DDI/server/internal/store"
 )
@@ -40,6 +42,16 @@ type Options struct {
 	SessionIdle     time.Duration
 	Now             func() time.Time
 	Logger          *slog.Logger
+
+	// Agent API (ADR 0021). Without PKI, enrollment tokens cannot be
+	// created and agents cannot enroll.
+	PKI *pki.Authority
+	// Hub tracks the agents' open streams (optional).
+	Hub *agenthub.Hub
+	// AgentAPIAddress is the host:port shown in the enrollment command.
+	AgentAPIAddress string
+	// AgentServerNames are the SANs of the agent-port certificate.
+	AgentServerNames []string
 }
 
 // Service implements the use cases.
@@ -51,6 +63,8 @@ type Service struct {
 
 	ipLimiter   *auth.Limiter
 	userLimiter *auth.Limiter
+	// enrollLimiter bounds enrollment attempts per source IP.
+	enrollLimiter *auth.Limiter
 }
 
 // New creates a Service.
@@ -73,6 +87,8 @@ func New(opts Options) *Service {
 		// generous because several users may share an address.
 		ipLimiter:   auth.NewLimiter(20, 15*time.Second),
 		userLimiter: auth.NewLimiter(5, time.Minute),
+		// Enrollment: 10 attempts, then one every 30 s per source IP.
+		enrollLimiter: auth.NewLimiter(10, 30*time.Second),
 	}
 }