From 1bdd877051ab2ff229ed19cdcd71e05134abf740 Mon Sep 17 00:00:00 2001 From: "beug.k@beugsiud.de" Date: Tue, 13 Oct 2026 19:02:49 +0200 Subject: [PATCH] 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 --- server/internal/agenthub/hub.go | 155 +++++++++ server/internal/agenthub/hub_test.go | 63 ++++ server/internal/service/agentplane.go | 453 ++++++++++++++++++++++++++ server/internal/service/agents.go | 452 +++++++++++++++++++++++++ server/internal/service/revisions.go | 348 ++++++++++++++++++++ server/internal/service/service.go | 16 + 6 files changed, 1487 insertions(+) create mode 100644 server/internal/agenthub/hub.go create mode 100644 server/internal/agenthub/hub_test.go create mode 100644 server/internal/service/agentplane.go create mode 100644 server/internal/service/agents.go create mode 100644 server/internal/service/revisions.go diff --git a/server/internal/agenthub/hub.go b/server/internal/agenthub/hub.go new file mode 100644 index 0000000..5eec1f1 --- /dev/null +++ b/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]) +} diff --git a/server/internal/agenthub/hub_test.go b/server/internal/agenthub/hub_test.go new file mode 100644 index 0000000..fc399b4 --- /dev/null +++ b/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 +} diff --git a/server/internal/service/agentplane.go b/server/internal/service/agentplane.go new file mode 100644 index 0000000..85d5d07 --- /dev/null +++ b/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 +} diff --git a/server/internal/service/agents.go b/server/internal/service/agents.go new file mode 100644 index 0000000..8b76c4b --- /dev/null +++ b/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 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 +} diff --git a/server/internal/service/revisions.go b/server/internal/service/revisions.go new file mode 100644 index 0000000..c918115 --- /dev/null +++ b/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 +} diff --git a/server/internal/service/service.go b/server/internal/service/service.go index 40b550e..73804f3 100644 --- a/server/internal/service/service.go +++ b/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), } }