sir_melonchenandClaude Opus 5.5
2b04661792
Implements the M5 endpoints of the contract (ADR 0023): clusters with member validation, transfer keys and HA credentials, TSIG keys with audited secret access and rotation, explicit zones and typed records, the DHCP cluster of networks, the cluster of agents and the default clusters and HA password file of the generator settings. Revisions are stored per target with sealed secret files and sealed source secrets (restored for rollbacks); bundles carry an agent's own target or the default group. Active leases are adopted as dynamic addresses. The derivation is called through a temporary adapter until derive implements the M5 contract. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
576 lines
18 KiB
Go
576 lines
18 KiB
Go
// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
|
|
|
package service
|
|
|
|
import (
|
|
"context"
|
|
"crypto/subtle"
|
|
"crypto/x509"
|
|
"encoding/json"
|
|
"errors"
|
|
"math"
|
|
"net/netip"
|
|
"reflect"
|
|
"strconv"
|
|
"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
|
|
}
|
|
s.publishAgent(res.AgentID)
|
|
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
|
|
// PushDeploymentID is set while a push of DesiredRevision is open
|
|
// (decimal deployment ID, ADR 0022).
|
|
PushDeploymentID string
|
|
}
|
|
|
|
// 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, desired revision and open push. A
|
|
// pending push of another revision is stale (the desired revision moved on
|
|
// without it) and is superseded.
|
|
func (s *Service) settings(ctx context.Context, tx *store.Tx, a *store.Agent) (*AgentSettings, error) {
|
|
latest, err := latestRevisionNumber(ctx, tx)
|
|
if err != nil {
|
|
return nil, translate(err)
|
|
}
|
|
desired := desiredRevision(a, latest)
|
|
out := &AgentSettings{Mode: a.Mode, DesiredRevision: uint64(desired)} //nolint:gosec // non-negative
|
|
p, err := tx.PendingDeployment(ctx, a.ID)
|
|
if errors.Is(err, store.ErrNotFound) {
|
|
return out, nil
|
|
}
|
|
if err != nil {
|
|
return nil, translate(err)
|
|
}
|
|
if p.Revision == desired {
|
|
out.PushDeploymentID = strconv.FormatInt(p.ID, 10)
|
|
return out, nil
|
|
}
|
|
stale, err := tx.SupersedeDeployments(ctx, store.SupersedeFilter{AgentID: a.ID, KeepRevision: desired},
|
|
s.now().UTC().Truncate(time.Microsecond))
|
|
if err != nil {
|
|
return nil, translate(err)
|
|
}
|
|
for _, d := range stale {
|
|
s.publishDeployment(d)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// pushDeploymentID parses the deployment ID of a server-created push
|
|
// (decimal); IDs of M3-style notifications (UUIDs) do not parse.
|
|
func pushDeploymentID(s string) (int64, bool) {
|
|
if s == "" || len(s) > 19 || strings.Trim(s, "0123456789") != "" {
|
|
return 0, false
|
|
}
|
|
id, err := strconv.ParseInt(s, 10, 64)
|
|
return id, err == nil && id > 0
|
|
}
|
|
|
|
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)
|
|
}
|
|
s.publishAgent(agentID)
|
|
return s.settings(ctx, q, a)
|
|
}
|
|
|
|
// AgentDisconnected tells the UI that an agent's stream has ended.
|
|
func (s *Service) AgentDisconnected(agentID string) { s.publishAgent(agentID) }
|
|
|
|
// statusChanged reports whether a heartbeat changed what the UI shows
|
|
// (revisions or health apart from the check time).
|
|
func statusChanged(before, after *store.Agent) bool {
|
|
if before.AppliedRevision != after.AppliedRevision || before.LastGoodRevision != after.LastGoodRevision {
|
|
return true
|
|
}
|
|
hb, ha := decodeHealth(before.Health), decodeHealth(after.Health)
|
|
hb.CheckedAt, ha.CheckedAt = nil, nil
|
|
return !reflect.DeepEqual(hb, ha)
|
|
}
|
|
|
|
// 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 {
|
|
q := s.st.Q()
|
|
before, err := q.GetAgent(ctx, agentID, false)
|
|
if err != nil {
|
|
return agentStatusErr(err)
|
|
}
|
|
after, err := 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)
|
|
}
|
|
if statusChanged(before, after) {
|
|
s.publishAgent(agentID)
|
|
}
|
|
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 an agent and the
|
|
// revision time (the bundle's created_at): the files of the target with the
|
|
// agent's ID if the revision has one, otherwise those of the role's
|
|
// default group (ADR 0023). Secret files are decrypted; only the signed
|
|
// bundle carries them.
|
|
func (s *Service) RevisionArtifacts(ctx context.Context, number uint64, role, agentID string) (time.Time, []bundle.Artifact, error) {
|
|
if number < 1 || number > store.MaxRevision {
|
|
return time.Time{}, nil, ErrNotFound
|
|
}
|
|
q := s.st.Q()
|
|
n := int64(number)
|
|
r, err := q.GetRevision(ctx, n)
|
|
if err != nil {
|
|
return time.Time{}, nil, translate(err)
|
|
}
|
|
target := ""
|
|
if ValidUUID(agentID) {
|
|
ok, err := q.RevisionHasTarget(ctx, n, role, agentID)
|
|
if err != nil {
|
|
return time.Time{}, nil, translate(err)
|
|
}
|
|
if ok {
|
|
target = agentID
|
|
}
|
|
}
|
|
files, err := q.RevisionTargetFiles(ctx, n, role, target)
|
|
if err != nil {
|
|
return time.Time{}, nil, translate(err)
|
|
}
|
|
if files, err = s.openFiles(n, files); err != nil {
|
|
return time.Time{}, nil, 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)
|
|
}
|
|
revision := int64(in.Revision) //nolint:gosec // at most MaxRevision (checked above)
|
|
message := cleanMessage(in.Message)
|
|
var recorded *store.Deployment
|
|
err := s.inTx(ctx, func(tx *store.Tx) error {
|
|
// A report with the ID of a server-created push completes that
|
|
// deployment (ADR 0022).
|
|
if id, ok := pushDeploymentID(in.CorrelationID); ok {
|
|
d, completed, err := tx.CompleteDeployment(ctx, id, agentID, revision, store.DeploymentResult{
|
|
Outcome: in.Outcome, Message: message, StartedAt: in.StartedAt, FinishedAt: in.FinishedAt,
|
|
ActiveRevision: int64(in.ActiveRevision), //nolint:gosec // at most MaxRevision (checked above)
|
|
})
|
|
switch {
|
|
case err == nil && !completed:
|
|
return nil // resent report: nothing changes
|
|
case err == nil:
|
|
recorded = d
|
|
if _, err := tx.UpdateAgentStatus(ctx, agentID, st); err != nil {
|
|
return agentStatusErr(err)
|
|
}
|
|
return nil
|
|
case !errors.Is(err, store.ErrNotFound):
|
|
return err
|
|
}
|
|
// Unknown ID: stored as a report like any other.
|
|
}
|
|
finished := in.FinishedAt
|
|
d := &store.Deployment{
|
|
AgentID: agentID, CorrelationID: cleanText(in.CorrelationID, 100), Trigger: in.Trigger, Outcome: in.Outcome,
|
|
Message: message, StartedAt: in.StartedAt, FinishedAt: &finished, Revision: revision,
|
|
ActiveRevision: int64(in.ActiveRevision), //nolint:gosec // at most MaxRevision (checked above)
|
|
}
|
|
inserted, err := tx.InsertDeployment(ctx, d)
|
|
if err != nil || !inserted {
|
|
// A resent report (same agent, revision and start) changes
|
|
// nothing; it must not roll back the agent's state either.
|
|
return err
|
|
}
|
|
recorded = d
|
|
if _, err := tx.UpdateAgentStatus(ctx, agentID, st); err != nil {
|
|
return agentStatusErr(err)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if recorded != nil {
|
|
s.publishDeployment(recorded)
|
|
s.publishAgent(agentID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 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
|
|
now := s.now().UTC().Truncate(time.Microsecond)
|
|
if err := tx.StoreLeases(ctx, agentID, int64(seq), fullSync, leases, now); err != nil { //nolint:gosec // checked above
|
|
return err
|
|
}
|
|
return s.adoptLeases(ctx, tx, leases, now)
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return res, nil
|
|
}
|