feat(store): agents, enrollment tokens, revisions, deployments and leases
Migration 00003 adds the PKI key table, agents, enrollment tokens, immutable configuration revisions with their files, deployment reports and DHCP leases, each with database constraints. Repositories and DB tests for all of them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
beug.k@beugsiud.de
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
e843502cbf
commit
8e76c0cf32
@@ -0,0 +1,293 @@
|
||||
// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/netip"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// PKI key purposes (table pki_keys).
|
||||
const (
|
||||
PKIAgentCA = "agent_ca"
|
||||
PKIBundleSigning = "bundle_signing"
|
||||
)
|
||||
|
||||
// PKIKey is a row of pki_keys. PrivateKey is sealed with the master key.
|
||||
type PKIKey struct {
|
||||
Purpose string
|
||||
Certificate []byte
|
||||
PublicKey []byte
|
||||
PrivateKey []byte
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// GetPKIKey returns the key material for purpose.
|
||||
func (t *Tx) GetPKIKey(ctx context.Context, purpose string) (*PKIKey, error) {
|
||||
var k PKIKey
|
||||
err := t.q.QueryRow(ctx, `SELECT purpose, certificate, public_key, private_key, created_at FROM pki_keys WHERE purpose = $1`,
|
||||
purpose).Scan(&k.Purpose, &k.Certificate, &k.PublicKey, &k.PrivateKey, &k.CreatedAt)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
return &k, nil
|
||||
}
|
||||
|
||||
// InsertPKIKeyIfAbsent stores k unless a key for its purpose exists already
|
||||
// (concurrent first starts: the first insert wins, everyone reads it back).
|
||||
func (t *Tx) InsertPKIKeyIfAbsent(ctx context.Context, k *PKIKey) (inserted bool, err error) {
|
||||
tag, err := t.q.Exec(ctx, `INSERT INTO pki_keys (purpose, certificate, public_key, private_key) VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (purpose) DO NOTHING`, k.Purpose, k.Certificate, k.PublicKey, k.PrivateKey)
|
||||
if err != nil {
|
||||
return false, mapErr(err)
|
||||
}
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
|
||||
// Agent is a row of the agents table.
|
||||
type Agent struct {
|
||||
ID string
|
||||
Role string
|
||||
Hostname string
|
||||
AgentVersion string
|
||||
ServiceVersion string
|
||||
Mode string
|
||||
EnrolledAt time.Time
|
||||
LastSeenAt *time.Time
|
||||
RemoteAddress *netip.Addr
|
||||
CertSerial string
|
||||
CertNotAfter time.Time
|
||||
AppliedRevision int64
|
||||
LastGoodRevision int64
|
||||
Health json.RawMessage
|
||||
RevokedAt *time.Time
|
||||
LeaseSequence int64
|
||||
}
|
||||
|
||||
const agentColumns = `id, role, hostname, agent_version, service_version, mode, enrolled_at, last_seen_at,
|
||||
remote_address, cert_serial, cert_not_after, applied_revision, last_good_revision, health, revoked_at, lease_sequence`
|
||||
|
||||
func scanAgent(row pgx.Row) (*Agent, error) {
|
||||
var a Agent
|
||||
var health []byte
|
||||
if err := row.Scan(&a.ID, &a.Role, &a.Hostname, &a.AgentVersion, &a.ServiceVersion, &a.Mode, &a.EnrolledAt,
|
||||
&a.LastSeenAt, &a.RemoteAddress, &a.CertSerial, &a.CertNotAfter, &a.AppliedRevision, &a.LastGoodRevision,
|
||||
&health, &a.RevokedAt, &a.LeaseSequence); err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
a.Health = health
|
||||
if a.RemoteAddress != nil {
|
||||
ip := a.RemoteAddress.Unmap()
|
||||
a.RemoteAddress = &ip
|
||||
}
|
||||
return &a, nil
|
||||
}
|
||||
|
||||
// InsertAgent creates an agent; the enrollment time is set.
|
||||
func (t *Tx) InsertAgent(ctx context.Context, a *Agent) error {
|
||||
row := t.q.QueryRow(ctx, `
|
||||
INSERT INTO agents (id, role, hostname, agent_version, remote_address, cert_serial, cert_not_after, last_seen_at)
|
||||
VALUES ($1, $2::agent_role, $3, $4, $5, $6, $7, now())
|
||||
RETURNING `+agentColumns,
|
||||
a.ID, a.Role, a.Hostname, a.AgentVersion, a.RemoteAddress, a.CertSerial, a.CertNotAfter)
|
||||
got, err := scanAgent(row)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
*a = *got
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAgent returns one agent.
|
||||
func (t *Tx) GetAgent(ctx context.Context, id string, forUpdate bool) (*Agent, error) {
|
||||
sql := `SELECT ` + agentColumns + ` FROM agents WHERE id = $1`
|
||||
if forUpdate {
|
||||
sql += ` FOR UPDATE`
|
||||
}
|
||||
return scanAgent(t.q.QueryRow(ctx, sql, id))
|
||||
}
|
||||
|
||||
// ListAgents returns all agents ordered by role and hostname.
|
||||
func (t *Tx) ListAgents(ctx context.Context) ([]*Agent, error) {
|
||||
rows, err := t.q.Query(ctx, `SELECT `+agentColumns+` FROM agents ORDER BY role, lower(hostname), id`)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*Agent
|
||||
for rows.Next() {
|
||||
a, err := scanAgent(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// RevokeAgent marks the agent revoked (keeps an earlier revocation time).
|
||||
func (t *Tx) RevokeAgent(ctx context.Context, id string, at time.Time) (*Agent, error) {
|
||||
return scanAgent(t.q.QueryRow(ctx, `UPDATE agents SET revoked_at = coalesce(revoked_at, $2) WHERE id = $1 RETURNING `+agentColumns, id, at))
|
||||
}
|
||||
|
||||
// DeleteAgent removes an agent with its deployments and leases.
|
||||
func (t *Tx) DeleteAgent(ctx context.Context, id string) error {
|
||||
tag, err := t.q.Exec(ctx, `DELETE FROM agents WHERE id = $1`, id)
|
||||
if err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateAgentCertificate records the serial of a renewed certificate. Only
|
||||
// non-revoked agents are updated.
|
||||
func (t *Tx) UpdateAgentCertificate(ctx context.Context, id, serial string, notAfter time.Time) error {
|
||||
tag, err := t.q.Exec(ctx, `UPDATE agents SET cert_serial = $2, cert_not_after = $3 WHERE id = $1 AND revoked_at IS NULL`,
|
||||
id, serial, notAfter)
|
||||
if err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// AgentStatus is reported by an agent. Nil fields keep the stored value.
|
||||
type AgentStatus struct {
|
||||
Hostname *string
|
||||
AgentVersion *string
|
||||
ServiceVersion *string
|
||||
AppliedRevision *int64
|
||||
LastGoodRevision *int64
|
||||
Health json.RawMessage
|
||||
RemoteAddress *netip.Addr
|
||||
SeenAt time.Time
|
||||
}
|
||||
|
||||
// UpdateAgentStatus stores reported status and the last contact. It returns
|
||||
// ErrNotFound for unknown and revoked agents.
|
||||
func (t *Tx) UpdateAgentStatus(ctx context.Context, id string, s AgentStatus) (*Agent, error) {
|
||||
return scanAgent(t.q.QueryRow(ctx, `
|
||||
UPDATE agents SET
|
||||
hostname = coalesce($2, hostname),
|
||||
agent_version = coalesce($3, agent_version),
|
||||
service_version = coalesce($4, service_version),
|
||||
applied_revision = coalesce($5, applied_revision),
|
||||
last_good_revision = coalesce($6, last_good_revision),
|
||||
health = coalesce($7::jsonb, health),
|
||||
remote_address = coalesce($8, remote_address),
|
||||
last_seen_at = $9
|
||||
WHERE id = $1 AND revoked_at IS NULL
|
||||
RETURNING `+agentColumns,
|
||||
id, s.Hostname, s.AgentVersion, s.ServiceVersion, s.AppliedRevision, s.LastGoodRevision,
|
||||
nullJSON(s.Health), s.RemoteAddress, s.SeenAt))
|
||||
}
|
||||
|
||||
// EnrollmentToken is a row of enrollment_tokens.
|
||||
type EnrollmentToken struct {
|
||||
ID string
|
||||
SecretHash []byte
|
||||
Role string
|
||||
Description string
|
||||
CreatedBy *string
|
||||
CreatedByName string
|
||||
CreatedAt time.Time
|
||||
ExpiresAt time.Time
|
||||
UsedAt *time.Time
|
||||
AgentID *string
|
||||
}
|
||||
|
||||
const tokenColumns = //nolint:gosec // a column list, not a credential
|
||||
`id, secret_hash, role, description, created_by, created_by_name, created_at, expires_at, used_at, agent_id`
|
||||
|
||||
func scanToken(row pgx.Row) (*EnrollmentToken, error) {
|
||||
var tk EnrollmentToken
|
||||
if err := row.Scan(&tk.ID, &tk.SecretHash, &tk.Role, &tk.Description, &tk.CreatedBy, &tk.CreatedByName,
|
||||
&tk.CreatedAt, &tk.ExpiresAt, &tk.UsedAt, &tk.AgentID); err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
return &tk, nil
|
||||
}
|
||||
|
||||
// InsertEnrollmentToken stores a new token (CreatedAt is taken from tk).
|
||||
func (t *Tx) InsertEnrollmentToken(ctx context.Context, tk *EnrollmentToken) error {
|
||||
_, err := t.q.Exec(ctx, `
|
||||
INSERT INTO enrollment_tokens (id, secret_hash, role, description, created_by, created_by_name, created_at, expires_at)
|
||||
VALUES ($1, $2, $3::agent_role, $4, $5, $6, $7, $8)`,
|
||||
tk.ID, tk.SecretHash, tk.Role, tk.Description, tk.CreatedBy, tk.CreatedByName, tk.CreatedAt, tk.ExpiresAt)
|
||||
return mapErr(err)
|
||||
}
|
||||
|
||||
// GetEnrollmentToken returns one token.
|
||||
func (t *Tx) GetEnrollmentToken(ctx context.Context, id string, forUpdate bool) (*EnrollmentToken, error) {
|
||||
sql := `SELECT ` + tokenColumns + ` FROM enrollment_tokens WHERE id = $1`
|
||||
if forUpdate {
|
||||
sql += ` FOR UPDATE`
|
||||
}
|
||||
return scanToken(t.q.QueryRow(ctx, sql, id))
|
||||
}
|
||||
|
||||
// ListEnrollmentTokens returns tokens that were used or expired after since
|
||||
// or are still unused and valid, newest first.
|
||||
func (t *Tx) ListEnrollmentTokens(ctx context.Context, since time.Time) ([]*EnrollmentToken, error) {
|
||||
rows, err := t.q.Query(ctx, `SELECT `+tokenColumns+` FROM enrollment_tokens
|
||||
WHERE (used_at IS NULL AND expires_at > $1) OR used_at > $1
|
||||
ORDER BY created_at DESC, id`, since)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*EnrollmentToken
|
||||
for rows.Next() {
|
||||
tk, err := scanToken(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, tk)
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// DeleteEnrollmentToken removes a token.
|
||||
func (t *Tx) DeleteEnrollmentToken(ctx context.Context, id string) error {
|
||||
tag, err := t.q.Exec(ctx, `DELETE FROM enrollment_tokens WHERE id = $1`, id)
|
||||
if err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkEnrollmentTokenUsed records the enrollment. It fails with ErrNotFound
|
||||
// if the token was used in the meantime.
|
||||
func (t *Tx) MarkEnrollmentTokenUsed(ctx context.Context, id, agentID string, at time.Time) error {
|
||||
tag, err := t.q.Exec(ctx, `UPDATE enrollment_tokens SET used_at = $3, agent_id = $2 WHERE id = $1 AND used_at IS NULL`,
|
||||
id, agentID, at)
|
||||
if err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteEnrollmentTokensBefore removes tokens that were used or expired
|
||||
// before t0 (housekeeping).
|
||||
func (t *Tx) DeleteEnrollmentTokensBefore(ctx context.Context, t0 time.Time) (int64, error) {
|
||||
tag, err := t.q.Exec(ctx, `DELETE FROM enrollment_tokens WHERE coalesce(used_at, expires_at) < $1`, t0)
|
||||
if err != nil {
|
||||
return 0, mapErr(err)
|
||||
}
|
||||
return tag.RowsAffected(), nil
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
||||
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.beugsiud.de/beug.k/DDI/server/internal/store"
|
||||
"git.beugsiud.de/beug.k/DDI/server/internal/store/storetest"
|
||||
)
|
||||
|
||||
const (
|
||||
agentA = "00000000-0000-4000-8000-00000000000a"
|
||||
agentB = "00000000-0000-4000-8000-00000000000b"
|
||||
)
|
||||
|
||||
func newAgent(t *testing.T, st *store.Store, id, role, host, serial string) *store.Agent {
|
||||
t.Helper()
|
||||
ip := netip.MustParseAddr("192.0.2.7")
|
||||
a := &store.Agent{ID: id, Role: role, Hostname: host, AgentVersion: "1.0", RemoteAddress: &ip, CertSerial: serial,
|
||||
CertNotAfter: time.Now().Add(14 * 24 * time.Hour)}
|
||||
if err := st.Q().InsertAgent(context.Background(), a); err != nil {
|
||||
t.Fatalf("insert agent: %v", err)
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
func TestPKIKeysInsertOnce(t *testing.T) {
|
||||
st := storetest.New(t)
|
||||
ctx := context.Background()
|
||||
k := &store.PKIKey{Purpose: store.PKIBundleSigning, PublicKey: []byte("pub1"), PrivateKey: []byte("sealed1")}
|
||||
if ok, err := st.Q().InsertPKIKeyIfAbsent(ctx, k); err != nil || !ok {
|
||||
t.Fatalf("first insert: %v %v", ok, err)
|
||||
}
|
||||
k2 := &store.PKIKey{Purpose: store.PKIBundleSigning, PublicKey: []byte("pub2"), PrivateKey: []byte("sealed2")}
|
||||
if ok, err := st.Q().InsertPKIKeyIfAbsent(ctx, k2); err != nil || ok {
|
||||
t.Fatalf("second insert: %v %v", ok, err)
|
||||
}
|
||||
got, err := st.Q().GetPKIKey(ctx, store.PKIBundleSigning)
|
||||
if err != nil || string(got.PublicKey) != "pub1" {
|
||||
t.Fatalf("get: %+v %v", got, err)
|
||||
}
|
||||
if _, err := st.Q().InsertPKIKeyIfAbsent(ctx, &store.PKIKey{Purpose: store.PKIAgentCA, PublicKey: []byte("x"), PrivateKey: []byte("y")}); err == nil {
|
||||
t.Error("CA without certificate accepted")
|
||||
}
|
||||
if _, err := st.Q().GetPKIKey(ctx, store.PKIAgentCA); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Errorf("missing key: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentsAndTokens(t *testing.T) {
|
||||
st := storetest.New(t)
|
||||
ctx := context.Background()
|
||||
q := st.Q()
|
||||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||||
|
||||
tk := &store.EnrollmentToken{ID: "0123456789abcdef", SecretHash: make([]byte, 32), Role: "dns", CreatedByName: "admin",
|
||||
CreatedAt: now, ExpiresAt: now.Add(time.Hour)}
|
||||
if err := q.InsertEnrollmentToken(ctx, tk); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bad := *tk
|
||||
bad.ID = "XYZ"
|
||||
if err := q.InsertEnrollmentToken(ctx, &bad); !store.IsConstraint(err, "enrollment_tokens_id_format") {
|
||||
t.Errorf("bad token id: %v", err)
|
||||
}
|
||||
|
||||
a := newAgent(t, st, agentA, "dns", "ns1", "0a")
|
||||
if a.Mode != "auto_pull" || a.EnrolledAt.IsZero() || a.RemoteAddress == nil || a.RemoteAddress.String() != "192.0.2.7" {
|
||||
t.Errorf("inserted agent %+v", a)
|
||||
}
|
||||
var h map[string]any
|
||||
if err := json.Unmarshal(a.Health, &h); err != nil || h["state"] != "unknown" {
|
||||
t.Errorf("default health %s", a.Health)
|
||||
}
|
||||
if err := st.Q().InsertAgent(ctx, &store.Agent{ID: agentB, Role: "dhcp", CertSerial: "0a", CertNotAfter: now}); !store.IsConstraint(err, "agents_cert_serial_unique") {
|
||||
t.Errorf("duplicate serial: %v", err)
|
||||
}
|
||||
newAgent(t, st, agentB, "dhcp", "dhcp1", "0b")
|
||||
|
||||
if err := q.MarkEnrollmentTokenUsed(ctx, tk.ID, agentA, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := q.MarkEnrollmentTokenUsed(ctx, tk.ID, agentB, now); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Errorf("second use: %v", err)
|
||||
}
|
||||
list, err := q.ListEnrollmentTokens(ctx, now.Add(-time.Minute))
|
||||
if err != nil || len(list) != 1 || list[0].AgentID == nil || *list[0].AgentID != agentA {
|
||||
t.Fatalf("tokens %+v %v", list, err)
|
||||
}
|
||||
if list, _ := q.ListEnrollmentTokens(ctx, now.Add(time.Minute)); len(list) != 0 {
|
||||
t.Errorf("old used token listed: %+v", list)
|
||||
}
|
||||
|
||||
host := "ns1.example.test"
|
||||
applied := int64(3)
|
||||
got, err := q.UpdateAgentStatus(ctx, agentA, store.AgentStatus{Hostname: &host, AppliedRevision: &applied,
|
||||
Health: json.RawMessage(`{"state":"ok","message":"","checks":[]}`), SeenAt: now})
|
||||
if err != nil || got.Hostname != host || got.AppliedRevision != 3 || got.AgentVersion != "1.0" || got.LastSeenAt == nil {
|
||||
t.Fatalf("status update %+v %v", got, err)
|
||||
}
|
||||
if err := q.UpdateAgentCertificate(ctx, agentA, "ff", now.Add(time.Hour)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := q.RevokeAgent(ctx, agentA, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := q.UpdateAgentStatus(ctx, agentA, store.AgentStatus{SeenAt: now}); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Errorf("status of revoked agent: %v", err)
|
||||
}
|
||||
if err := q.UpdateAgentCertificate(ctx, agentA, "fe", now); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Errorf("renewal of revoked agent: %v", err)
|
||||
}
|
||||
agents, err := q.ListAgents(ctx)
|
||||
if err != nil || len(agents) != 2 || agents[0].ID != agentA || agents[0].RevokedAt == nil || agents[0].CertSerial != "ff" {
|
||||
t.Fatalf("agents %+v %v", agents, err)
|
||||
}
|
||||
|
||||
if err := q.DeleteAgent(ctx, agentA); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, _ := q.GetEnrollmentToken(ctx, tk.ID, false); got == nil || got.AgentID != nil || got.UsedAt == nil {
|
||||
t.Errorf("token after agent deletion %+v", got)
|
||||
}
|
||||
if err := q.DeleteEnrollmentToken(ctx, tk.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := q.DeleteEnrollmentToken(ctx, tk.ID); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Errorf("second delete: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRevisionsAndDeployments(t *testing.T) {
|
||||
st := storetest.New(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||||
|
||||
if _, err := st.Q().LatestRevision(ctx); !errors.Is(err, store.ErrNotFound) {
|
||||
t.Fatalf("latest of none: %v", err)
|
||||
}
|
||||
files := []store.RevisionFile{
|
||||
{Role: "dns", Path: "ddi-suite.conf", Mode: 0o644, Content: []byte("conf")},
|
||||
{Role: "dns", Path: "zones/db.a.test", Mode: 0o644, CreateOnly: true, Content: []byte("zone")},
|
||||
{Role: "dhcp", Path: "kea-dhcp4.conf", Mode: 0o640, Content: []byte("{}")},
|
||||
}
|
||||
err := st.InTx(ctx, func(tx *store.Tx) error {
|
||||
if err := tx.LockRevisions(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.InsertRevision(ctx, &store.Revision{Number: 1, CreatedAt: now, CreatedByName: "admin",
|
||||
Warnings: json.RawMessage(`[{"code":"w"}]`)}, files)
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = st.InTx(ctx, func(tx *store.Tx) error {
|
||||
return tx.InsertRevision(ctx, &store.Revision{Number: 2, CreatedAt: now, CreatedByName: "admin"},
|
||||
[]store.RevisionFile{{Role: "dns", Path: "../x", Content: []byte("x")}})
|
||||
})
|
||||
if !store.IsConstraint(err, "config_revision_files_path") {
|
||||
t.Errorf("bad path: %v", err)
|
||||
}
|
||||
if err := st.Q().InsertRevision(ctx, &store.Revision{Number: 1 << 32, CreatedAt: now, CreatedByName: "admin"}, nil); !store.IsConstraint(err, "config_revisions_number") {
|
||||
t.Errorf("number out of range: %v", err)
|
||||
}
|
||||
if _, err := st.Pool().Exec(ctx, `UPDATE config_revisions SET comment = 'x'`); err == nil {
|
||||
t.Error("revision is mutable")
|
||||
}
|
||||
if _, err := st.Pool().Exec(ctx, `UPDATE config_revision_files SET content = 'x'`); err == nil {
|
||||
t.Error("revision file is mutable")
|
||||
}
|
||||
got, err := st.Q().RevisionFiles(ctx, 1, "dns")
|
||||
if err != nil || len(got) != 2 || got[1].Path != "zones/db.a.test" || !got[1].CreateOnly || got[0].Mode != 0o644 {
|
||||
t.Fatalf("dns files %+v %v", got, err)
|
||||
}
|
||||
if all, _ := st.Q().RevisionFiles(ctx, 1, ""); len(all) != 3 || all[2].Role != "dhcp" {
|
||||
t.Errorf("all files %+v", all)
|
||||
}
|
||||
revs, err := st.Q().ListRevisions(ctx, 10, 0)
|
||||
if err != nil || len(revs) != 1 || revs[0].DNSFiles != 2 || revs[0].DHCPFiles != 1 || revs[0].WarningCount != 1 {
|
||||
t.Fatalf("list %+v %v", revs, err)
|
||||
}
|
||||
if revs, _ := st.Q().ListRevisions(ctx, 10, 1); len(revs) != 0 {
|
||||
t.Errorf("before=1: %+v", revs)
|
||||
}
|
||||
|
||||
newAgent(t, st, agentA, "dns", "ns1", "01")
|
||||
for i, outcome := range []string{"rejected", "succeeded"} {
|
||||
d := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: outcome, ActiveRevision: int64(i),
|
||||
StartedAt: now, FinishedAt: now.Add(time.Second)}
|
||||
if err := st.Q().InsertDeployment(ctx, d); err != nil || d.ID == 0 {
|
||||
t.Fatalf("insert deployment: %v", err)
|
||||
}
|
||||
}
|
||||
bad := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "succeeded", StartedAt: now, FinishedAt: now.Add(-time.Second)}
|
||||
if err := st.Q().InsertDeployment(ctx, bad); !store.IsConstraint(err, "deployments_times") {
|
||||
t.Errorf("finished before started: %v", err)
|
||||
}
|
||||
ds, err := st.Q().ListDeployments(ctx, store.DeploymentFilter{AgentID: agentA, Limit: 1})
|
||||
if err != nil || len(ds) != 1 || ds[0].Outcome != "succeeded" || ds[0].AgentHostname != "ns1" {
|
||||
t.Fatalf("deployments %+v %v", ds, err)
|
||||
}
|
||||
if older, _ := st.Q().ListDeployments(ctx, store.DeploymentFilter{BeforeID: ds[0].ID}); len(older) != 1 || older[0].Outcome != "rejected" {
|
||||
t.Errorf("older %+v", older)
|
||||
}
|
||||
latest, err := st.Q().LatestDeployments(ctx)
|
||||
if err != nil || latest[agentA] == nil || latest[agentA].ID != ds[0].ID {
|
||||
t.Errorf("latest %+v %v", latest, err)
|
||||
}
|
||||
if err := st.Q().DeleteAgent(ctx, agentA); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ds, _ := st.Q().ListDeployments(ctx, store.DeploymentFilter{}); len(ds) != 0 {
|
||||
t.Errorf("deployments survived agent deletion: %+v", ds)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLeases(t *testing.T) {
|
||||
st := storetest.New(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC().Truncate(time.Microsecond)
|
||||
newAgent(t, st, agentA, "dhcp", "dhcp1", "01")
|
||||
newAgent(t, st, agentB, "dhcp", "dhcp2", "02")
|
||||
var netID string
|
||||
var subnet int32
|
||||
if err := st.Pool().QueryRow(ctx, `INSERT INTO networks (cidr, name) VALUES ('192.0.2.0/24', 'Office') RETURNING id, kea_subnet_id`).Scan(&netID, &subnet); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
lease := func(ip, mac, host, state string) store.Lease {
|
||||
return store.Lease{IP: netip.MustParseAddr(ip), MAC: mac, Hostname: host, State: state, ExpiresAt: &now, SubnetID: uint32(subnet)}
|
||||
}
|
||||
store1 := func(agent string, seq int64, replace bool, ls ...store.Lease) {
|
||||
t.Helper()
|
||||
err := st.InTx(ctx, func(tx *store.Tx) error {
|
||||
got, role, err := tx.LockLeaseSequence(ctx, agent)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if role != "dhcp" || got >= seq {
|
||||
t.Fatalf("sequence %d role %s", got, role)
|
||||
}
|
||||
return tx.StoreLeases(ctx, agent, seq, replace, ls, now)
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
store1(agentA, 1, true, lease("192.0.2.10", "00:11:22:33:44:55", "pc1", "active"), lease("192.0.2.11", "00:11:22:33:44:56", "pc2", "active"))
|
||||
store1(agentA, 2, false, lease("192.0.2.11", "00:11:22:33:44:56", "pc2", "released"), lease("192.0.2.12", "", "", "declined"))
|
||||
store1(agentB, 1, true, lease("192.0.2.10", "00:11:22:33:44:55", "pc1", "active"))
|
||||
|
||||
all, more, err := st.Q().ListLeases(ctx, store.LeaseFilter{})
|
||||
if err != nil || more || len(all) != 4 || all[0].NetworkID == nil || *all[0].NetworkID != netID {
|
||||
t.Fatalf("all leases %+v %v %v", all, more, err)
|
||||
}
|
||||
// The page holds two IPs: both agents' leases of .10 and the one of .11.
|
||||
page, more, err := st.Q().ListLeases(ctx, store.LeaseFilter{Limit: 2})
|
||||
if err != nil || !more || len(page) != 3 || page[2].IP.String() != "192.0.2.11" || page[2].State != "released" {
|
||||
t.Fatalf("page %+v %v %v", page, more, err)
|
||||
}
|
||||
after := page[2].IP
|
||||
rest, more, _ := st.Q().ListLeases(ctx, store.LeaseFilter{Limit: 2, After: &after})
|
||||
if more || len(rest) != 1 || rest[0].IP.String() != "192.0.2.12" {
|
||||
t.Errorf("rest %+v %v", rest, more)
|
||||
}
|
||||
if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{State: "active", Query: "pc1"}); len(ls) != 2 {
|
||||
t.Errorf("filtered %+v", ls)
|
||||
}
|
||||
if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{NetworkID: netID, Query: "44:56"}); len(ls) != 1 {
|
||||
t.Errorf("network+mac %+v", ls)
|
||||
}
|
||||
if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{NetworkID: "00000000-0000-4000-8000-000000000000"}); len(ls) != 0 {
|
||||
t.Errorf("other network %+v", ls)
|
||||
}
|
||||
|
||||
// A full sync replaces the agent's set.
|
||||
store1(agentA, 3, true, lease("192.0.2.20", "00:11:22:33:44:57", "pc3", "active"))
|
||||
if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{}); len(ls) != 2 {
|
||||
t.Errorf("after full sync %+v", ls)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/netip"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// MaxLeasePage bounds a page of ListLeases.
|
||||
const MaxLeasePage = 1000
|
||||
|
||||
// Lease is a DHCP lease reported by an agent. NetworkID is resolved through
|
||||
// the Kea subnet ID by ListLeases.
|
||||
type Lease struct {
|
||||
AgentID string
|
||||
IP netip.Addr
|
||||
MAC string
|
||||
Hostname string
|
||||
State string
|
||||
ExpiresAt *time.Time
|
||||
UpdatedAt *time.Time
|
||||
SubnetID uint32
|
||||
ClientID string
|
||||
ReportedAt time.Time
|
||||
NetworkID *string
|
||||
}
|
||||
|
||||
// LockLeaseSequence locks the agent row and returns its lease sequence and
|
||||
// role. Use it inside a transaction before StoreLeases.
|
||||
func (t *Tx) LockLeaseSequence(ctx context.Context, agentID string) (seq int64, role string, err error) {
|
||||
err = t.q.QueryRow(ctx, `SELECT lease_sequence, role FROM agents WHERE id = $1 AND revoked_at IS NULL FOR UPDATE`,
|
||||
agentID).Scan(&seq, &role)
|
||||
return seq, role, mapErr(err)
|
||||
}
|
||||
|
||||
// StoreLeases writes a lease batch of an agent and records its sequence.
|
||||
// With replace, the batch is the complete lease set of the agent.
|
||||
func (t *Tx) StoreLeases(ctx context.Context, agentID string, seq int64, replace bool, leases []Lease, at time.Time) error {
|
||||
if replace {
|
||||
if _, err := t.q.Exec(ctx, `DELETE FROM dhcp_leases WHERE agent_id = $1`, agentID); err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
}
|
||||
b := &pgx.Batch{}
|
||||
for _, l := range leases {
|
||||
b.Queue(`
|
||||
INSERT INTO dhcp_leases (agent_id, ip, mac, hostname, state, expires_at, updated_at, subnet_id, client_id, reported_at)
|
||||
VALUES ($1, $2, $3, $4, $5::lease_state, $6, $7, $8, $9, $10)
|
||||
ON CONFLICT (agent_id, ip) DO UPDATE SET mac = EXCLUDED.mac, hostname = EXCLUDED.hostname,
|
||||
state = EXCLUDED.state, expires_at = EXCLUDED.expires_at, updated_at = EXCLUDED.updated_at,
|
||||
subnet_id = EXCLUDED.subnet_id, client_id = EXCLUDED.client_id, reported_at = EXCLUDED.reported_at`,
|
||||
agentID, l.IP, l.MAC, l.Hostname, l.State, l.ExpiresAt, l.UpdatedAt, int64(l.SubnetID), l.ClientID, at)
|
||||
}
|
||||
if err := t.execBatch(ctx, b); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := t.q.Exec(ctx, `UPDATE agents SET lease_sequence = $2 WHERE id = $1`, agentID, seq)
|
||||
return mapErr(err)
|
||||
}
|
||||
|
||||
// LeaseFilter selects leases. Zero values mean "no restriction".
|
||||
type LeaseFilter struct {
|
||||
NetworkID string
|
||||
State string
|
||||
Query string // substring of IP, MAC or hostname
|
||||
After *netip.Addr
|
||||
Limit int
|
||||
}
|
||||
|
||||
// ListLeases returns leases ordered by IP and agent. Limit counts distinct
|
||||
// IPs, so leases of the same IP from several agents are never split across
|
||||
// pages; more reports whether further IPs follow.
|
||||
func (t *Tx) ListLeases(ctx context.Context, f LeaseFilter) (out []*Lease, more bool, err error) {
|
||||
var where []string
|
||||
var args []any
|
||||
add := func(cond string, v any) {
|
||||
args = append(args, v)
|
||||
where = append(where, fmt.Sprintf(cond, len(args)))
|
||||
}
|
||||
if f.NetworkID != "" {
|
||||
add("n.id = $%d::uuid", f.NetworkID)
|
||||
}
|
||||
if f.State != "" {
|
||||
add("l.state = $%d::lease_state", f.State)
|
||||
}
|
||||
if f.Query != "" {
|
||||
args = append(args, "%"+escapeLike(f.Query)+"%")
|
||||
n := len(args)
|
||||
where = append(where, fmt.Sprintf("(host(l.ip) ILIKE $%[1]d OR l.mac ILIKE $%[1]d OR l.hostname ILIKE $%[1]d)", n))
|
||||
}
|
||||
if f.After != nil {
|
||||
add("l.ip > $%d", *f.After)
|
||||
}
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > MaxLeasePage {
|
||||
limit = 100
|
||||
}
|
||||
from := ` FROM dhcp_leases l LEFT JOIN networks n ON n.kea_subnet_id = l.subnet_id AND l.subnet_id > 0`
|
||||
cond := ""
|
||||
if len(where) > 0 {
|
||||
cond = " WHERE " + strings.Join(where, " AND ")
|
||||
}
|
||||
// The page holds the first limit+1 distinct IPs; the extra one only
|
||||
// tells whether there is a next page.
|
||||
sql := `WITH page AS (SELECT DISTINCT l.ip` + from + cond + fmt.Sprintf(` ORDER BY l.ip LIMIT %d)`, limit+1) + `
|
||||
SELECT l.agent_id, l.ip, l.mac, l.hostname, l.state, l.expires_at, l.updated_at, l.subnet_id, l.client_id,
|
||||
l.reported_at, n.id` + from + ` JOIN page p ON p.ip = l.ip` + cond + ` ORDER BY l.ip, l.agent_id`
|
||||
rows, err := t.q.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, false, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
ips := 0
|
||||
var last netip.Addr
|
||||
for rows.Next() {
|
||||
var l Lease
|
||||
var subnet int64
|
||||
if err := rows.Scan(&l.AgentID, &l.IP, &l.MAC, &l.Hostname, &l.State, &l.ExpiresAt, &l.UpdatedAt, &subnet,
|
||||
&l.ClientID, &l.ReportedAt, &l.NetworkID); err != nil {
|
||||
return nil, false, mapErr(err)
|
||||
}
|
||||
l.IP = l.IP.Unmap()
|
||||
l.SubnetID = uint32(subnet) //nolint:gosec // 0..2^32-1 by constraint
|
||||
if ips == 0 || l.IP != last {
|
||||
ips++
|
||||
last = l.IP
|
||||
}
|
||||
if ips > limit {
|
||||
more = true
|
||||
continue
|
||||
}
|
||||
out = append(out, &l)
|
||||
}
|
||||
return out, more, mapErr(rows.Err())
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
-- SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
||||
-- Agents (M3, ADR 0021): PKI keys, enrollment tokens, agents, immutable
|
||||
-- configuration revisions, deployment reports and DHCP leases.
|
||||
|
||||
-- +goose Up
|
||||
|
||||
-- Key material of the agent PKI: the agent CA (ADR 0008) and the bundle
|
||||
-- signing key (ADR 0009). Private keys are sealed with the master key; the
|
||||
-- associated data binds them to their row (ADR 0015). Rows are created once
|
||||
-- (INSERT ... ON CONFLICT DO NOTHING, so concurrent first starts agree).
|
||||
CREATE TABLE pki_keys (
|
||||
purpose text PRIMARY KEY,
|
||||
certificate bytea,
|
||||
public_key bytea NOT NULL,
|
||||
private_key bytea NOT NULL,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
CONSTRAINT pki_keys_purpose CHECK (purpose IN ('agent_ca', 'bundle_signing')),
|
||||
CONSTRAINT pki_keys_ca_certificate CHECK (purpose <> 'agent_ca' OR certificate IS NOT NULL)
|
||||
);
|
||||
|
||||
CREATE TYPE agent_role AS ENUM ('dns', 'dhcp');
|
||||
CREATE TYPE agent_mode AS ENUM ('auto_pull', 'manual');
|
||||
|
||||
CREATE TABLE agents (
|
||||
id uuid PRIMARY KEY,
|
||||
role agent_role NOT NULL,
|
||||
hostname text NOT NULL DEFAULT '',
|
||||
agent_version text NOT NULL DEFAULT '',
|
||||
service_version text NOT NULL DEFAULT '',
|
||||
mode agent_mode NOT NULL DEFAULT 'auto_pull',
|
||||
enrolled_at timestamptz NOT NULL DEFAULT now(),
|
||||
last_seen_at timestamptz,
|
||||
remote_address inet,
|
||||
-- Serial number (lowercase hex) of the only certificate the agent may use.
|
||||
cert_serial text NOT NULL,
|
||||
cert_not_after timestamptz NOT NULL,
|
||||
applied_revision bigint NOT NULL DEFAULT 0,
|
||||
last_good_revision bigint NOT NULL DEFAULT 0,
|
||||
health jsonb NOT NULL DEFAULT '{"state": "unknown", "message": "", "checks": []}',
|
||||
revoked_at timestamptz,
|
||||
-- Highest lease batch sequence processed (ReportLeases).
|
||||
lease_sequence bigint NOT NULL DEFAULT 0,
|
||||
CONSTRAINT agents_cert_serial CHECK (cert_serial ~ '^[0-9a-f]{1,40}$'),
|
||||
CONSTRAINT agents_revisions CHECK (applied_revision BETWEEN 0 AND 4294967295
|
||||
AND last_good_revision BETWEEN 0 AND 4294967295),
|
||||
CONSTRAINT agents_lease_sequence CHECK (lease_sequence >= 0),
|
||||
CONSTRAINT agents_health_object CHECK (jsonb_typeof(health) = 'object')
|
||||
);
|
||||
CREATE UNIQUE INDEX agents_cert_serial_unique ON agents (cert_serial);
|
||||
|
||||
CREATE TABLE enrollment_tokens (
|
||||
id text PRIMARY KEY,
|
||||
-- SHA-256 of the secret part; the secret has 256 bits of entropy.
|
||||
secret_hash bytea NOT NULL,
|
||||
role agent_role NOT NULL,
|
||||
description text NOT NULL DEFAULT '',
|
||||
created_by uuid REFERENCES users (id) ON DELETE SET NULL,
|
||||
created_by_name text NOT NULL,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
expires_at timestamptz NOT NULL,
|
||||
used_at timestamptz,
|
||||
agent_id uuid REFERENCES agents (id) ON DELETE SET NULL,
|
||||
CONSTRAINT enrollment_tokens_id_format CHECK (id ~ '^[0-9a-f]{16}$'),
|
||||
CONSTRAINT enrollment_tokens_secret_hash CHECK (length(secret_hash) = 32),
|
||||
CONSTRAINT enrollment_tokens_expiry CHECK (expires_at > created_at),
|
||||
CONSTRAINT enrollment_tokens_agent_used CHECK (agent_id IS NULL OR used_at IS NOT NULL)
|
||||
);
|
||||
CREATE INDEX enrollment_tokens_created ON enrollment_tokens (created_at DESC);
|
||||
|
||||
-- A revision is immutable; its number is also the SOA serial (uint32).
|
||||
CREATE TABLE config_revisions (
|
||||
number bigint PRIMARY KEY,
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
created_by uuid REFERENCES users (id) ON DELETE SET NULL,
|
||||
created_by_name text NOT NULL,
|
||||
comment text NOT NULL DEFAULT '',
|
||||
warnings jsonb NOT NULL DEFAULT '[]',
|
||||
CONSTRAINT config_revisions_number CHECK (number BETWEEN 1 AND 4294967295),
|
||||
CONSTRAINT config_revisions_warnings_array CHECK (jsonb_typeof(warnings) = 'array')
|
||||
);
|
||||
|
||||
CREATE TABLE config_revision_files (
|
||||
revision bigint NOT NULL REFERENCES config_revisions (number) ON DELETE CASCADE,
|
||||
role agent_role NOT NULL,
|
||||
-- Position in the generator output (bundles list files in this order).
|
||||
ordinal integer NOT NULL,
|
||||
path text NOT NULL,
|
||||
mode integer NOT NULL,
|
||||
create_only boolean NOT NULL DEFAULT false,
|
||||
content bytea NOT NULL,
|
||||
PRIMARY KEY (revision, role, ordinal),
|
||||
CONSTRAINT config_revision_files_path_unique UNIQUE (revision, role, path),
|
||||
CONSTRAINT config_revision_files_ordinal CHECK (ordinal >= 0),
|
||||
CONSTRAINT config_revision_files_mode CHECK (mode BETWEEN 0 AND 511),
|
||||
CONSTRAINT config_revision_files_path CHECK (path <> '' AND length(path) <= 255
|
||||
AND left(path, 1) <> '/' AND path !~ '(^|/)\.')
|
||||
);
|
||||
|
||||
-- Revisions and their files never change once written.
|
||||
-- +goose StatementBegin
|
||||
CREATE FUNCTION config_revisions_immutable() RETURNS trigger
|
||||
LANGUAGE plpgsql AS $$
|
||||
BEGIN
|
||||
RAISE EXCEPTION 'configuration revisions are immutable'
|
||||
USING ERRCODE = 'check_violation', CONSTRAINT = 'config_revisions_immutable';
|
||||
END;
|
||||
$$;
|
||||
-- +goose StatementEnd
|
||||
CREATE TRIGGER config_revisions_immutable
|
||||
BEFORE UPDATE ON config_revisions
|
||||
FOR EACH ROW EXECUTE FUNCTION config_revisions_immutable();
|
||||
CREATE TRIGGER config_revision_files_immutable
|
||||
BEFORE UPDATE ON config_revision_files
|
||||
FOR EACH ROW EXECUTE FUNCTION config_revisions_immutable();
|
||||
|
||||
CREATE TYPE deployment_trigger AS ENUM ('pull', 'push');
|
||||
CREATE TYPE deployment_outcome AS ENUM ('succeeded', 'rejected', 'rolled_back', 'failed');
|
||||
|
||||
CREATE TABLE deployments (
|
||||
id bigserial PRIMARY KEY,
|
||||
agent_id uuid NOT NULL REFERENCES agents (id) ON DELETE CASCADE,
|
||||
-- deployment_id of the ApplyRevisionCommand, empty for pulls.
|
||||
correlation_id text NOT NULL DEFAULT '',
|
||||
revision bigint NOT NULL,
|
||||
trigger_kind deployment_trigger NOT NULL,
|
||||
outcome deployment_outcome NOT NULL,
|
||||
message text NOT NULL DEFAULT '',
|
||||
active_revision bigint NOT NULL DEFAULT 0,
|
||||
started_at timestamptz NOT NULL,
|
||||
finished_at timestamptz NOT NULL,
|
||||
reported_at timestamptz NOT NULL DEFAULT now(),
|
||||
CONSTRAINT deployments_revision CHECK (revision BETWEEN 1 AND 4294967295),
|
||||
CONSTRAINT deployments_active_revision CHECK (active_revision BETWEEN 0 AND 4294967295),
|
||||
CONSTRAINT deployments_times CHECK (finished_at >= started_at)
|
||||
);
|
||||
CREATE INDEX deployments_agent ON deployments (agent_id, id DESC);
|
||||
CREATE INDEX deployments_revision ON deployments (revision, id DESC);
|
||||
|
||||
-- Leases as last reported by each DHCP agent.
|
||||
CREATE TABLE dhcp_leases (
|
||||
agent_id uuid NOT NULL REFERENCES agents (id) ON DELETE CASCADE,
|
||||
ip inet NOT NULL,
|
||||
mac text NOT NULL DEFAULT '',
|
||||
hostname text NOT NULL DEFAULT '',
|
||||
state lease_state NOT NULL,
|
||||
expires_at timestamptz,
|
||||
updated_at timestamptz,
|
||||
subnet_id bigint NOT NULL DEFAULT 0,
|
||||
client_id text NOT NULL DEFAULT '',
|
||||
reported_at timestamptz NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (agent_id, ip),
|
||||
CONSTRAINT dhcp_leases_host_ip CHECK (masklen(ip) = CASE family(ip) WHEN 4 THEN 32 ELSE 128 END),
|
||||
CONSTRAINT dhcp_leases_subnet_id CHECK (subnet_id BETWEEN 0 AND 4294967295)
|
||||
);
|
||||
CREATE INDEX dhcp_leases_ip ON dhcp_leases (ip);
|
||||
CREATE INDEX dhcp_leases_subnet ON dhcp_leases (subnet_id);
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE dhcp_leases;
|
||||
DROP TABLE deployments;
|
||||
DROP TYPE deployment_outcome;
|
||||
DROP TYPE deployment_trigger;
|
||||
DROP TRIGGER config_revision_files_immutable ON config_revision_files;
|
||||
DROP TRIGGER config_revisions_immutable ON config_revisions;
|
||||
DROP FUNCTION config_revisions_immutable();
|
||||
DROP TABLE config_revision_files;
|
||||
DROP TABLE config_revisions;
|
||||
DROP TABLE enrollment_tokens;
|
||||
DROP TABLE agents;
|
||||
DROP TYPE agent_mode;
|
||||
DROP TYPE agent_role;
|
||||
DROP TABLE pki_keys;
|
||||
@@ -0,0 +1,266 @@
|
||||
// SPDX-License-Identifier: LicenseRef-DDI-Proprietary
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// MaxRevision is the highest revision number (it doubles as SOA serial).
|
||||
const MaxRevision = 1<<32 - 1
|
||||
|
||||
// revisionLockKey serializes revision creation (pg_advisory_xact_lock).
|
||||
const revisionLockKey = 0x64646901
|
||||
|
||||
// Revision is a row of config_revisions. DNSFiles, DHCPFiles and
|
||||
// WarningCount are filled by ListRevisions.
|
||||
type Revision struct {
|
||||
Number int64
|
||||
CreatedAt time.Time
|
||||
CreatedBy *string
|
||||
CreatedByName string
|
||||
Comment string
|
||||
Warnings json.RawMessage
|
||||
DNSFiles int
|
||||
DHCPFiles int
|
||||
WarningCount int
|
||||
}
|
||||
|
||||
// RevisionFile is one artifact of a revision.
|
||||
type RevisionFile struct {
|
||||
Role string
|
||||
Path string
|
||||
Mode uint32
|
||||
CreateOnly bool
|
||||
Content []byte
|
||||
}
|
||||
|
||||
// LockRevisions takes a transaction-scoped lock that serializes revision
|
||||
// creation. It must run inside a transaction.
|
||||
func (t *Tx) LockRevisions(ctx context.Context) error {
|
||||
_, err := t.q.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, int64(revisionLockKey))
|
||||
return mapErr(err)
|
||||
}
|
||||
|
||||
// LatestRevision returns the newest revision or ErrNotFound.
|
||||
func (t *Tx) LatestRevision(ctx context.Context) (*Revision, error) {
|
||||
return t.getRevision(ctx, `ORDER BY number DESC LIMIT 1`)
|
||||
}
|
||||
|
||||
// GetRevision returns one revision without files.
|
||||
func (t *Tx) GetRevision(ctx context.Context, number int64) (*Revision, error) {
|
||||
return t.getRevision(ctx, `WHERE number = $1`, number)
|
||||
}
|
||||
|
||||
func (t *Tx) getRevision(ctx context.Context, cond string, args ...any) (*Revision, error) {
|
||||
var r Revision
|
||||
var warnings []byte
|
||||
err := t.q.QueryRow(ctx, `SELECT number, created_at, created_by, created_by_name, comment, warnings
|
||||
FROM config_revisions `+cond, args...).Scan(&r.Number, &r.CreatedAt, &r.CreatedBy, &r.CreatedByName, &r.Comment, &warnings)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
r.Warnings = warnings
|
||||
return &r, nil
|
||||
}
|
||||
|
||||
// InsertRevision stores a revision with its files (in the given order per
|
||||
// role). CreatedAt is taken from r.
|
||||
func (t *Tx) InsertRevision(ctx context.Context, r *Revision, files []RevisionFile) error {
|
||||
warnings := r.Warnings
|
||||
if len(warnings) == 0 {
|
||||
warnings = json.RawMessage("[]")
|
||||
}
|
||||
if _, err := t.q.Exec(ctx, `INSERT INTO config_revisions (number, created_at, created_by, created_by_name, comment, warnings)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)`, r.Number, r.CreatedAt, r.CreatedBy, r.CreatedByName, r.Comment, []byte(warnings)); err != nil {
|
||||
return mapErr(err)
|
||||
}
|
||||
b := &pgx.Batch{}
|
||||
ordinal := map[string]int32{}
|
||||
for _, f := range files {
|
||||
n := ordinal[f.Role]
|
||||
ordinal[f.Role] = n + 1
|
||||
b.Queue(`INSERT INTO config_revision_files (revision, role, ordinal, path, mode, create_only, content)
|
||||
VALUES ($1, $2::agent_role, $3, $4, $5, $6, $7)`,
|
||||
r.Number, f.Role, n, f.Path, int32(f.Mode), f.CreateOnly, f.Content) //nolint:gosec // modes are at most 0o777 (checked by the database)
|
||||
}
|
||||
return t.execBatch(ctx, b)
|
||||
}
|
||||
|
||||
// RevisionFiles returns the files of a revision ordered by role and
|
||||
// generator order. role "" returns both roles.
|
||||
func (t *Tx) RevisionFiles(ctx context.Context, number int64, role string) ([]RevisionFile, error) {
|
||||
sql := `SELECT role, path, mode, create_only, content FROM config_revision_files WHERE revision = $1`
|
||||
args := []any{number}
|
||||
if role != "" {
|
||||
sql += ` AND role = $2::agent_role`
|
||||
args = append(args, role)
|
||||
}
|
||||
rows, err := t.q.Query(ctx, sql+` ORDER BY role, ordinal`, args...)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []RevisionFile{}
|
||||
for rows.Next() {
|
||||
var f RevisionFile
|
||||
var mode int32
|
||||
if err := rows.Scan(&f.Role, &f.Path, &mode, &f.CreateOnly, &f.Content); err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
f.Mode = uint32(mode) //nolint:gosec // 0..511 by constraint
|
||||
out = append(out, f)
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// ListRevisions returns up to limit revisions with a number below before
|
||||
// (0 = no bound), newest first.
|
||||
func (t *Tx) ListRevisions(ctx context.Context, limit int, before int64) ([]*Revision, error) {
|
||||
sql := `SELECT r.number, r.created_at, r.created_by, r.created_by_name, r.comment, jsonb_array_length(r.warnings),
|
||||
(SELECT count(*) FROM config_revision_files f WHERE f.revision = r.number AND f.role = 'dns'),
|
||||
(SELECT count(*) FROM config_revision_files f WHERE f.revision = r.number AND f.role = 'dhcp')
|
||||
FROM config_revisions r`
|
||||
args := []any{}
|
||||
if before > 0 {
|
||||
sql += ` WHERE r.number < $1`
|
||||
args = append(args, before)
|
||||
}
|
||||
sql += fmt.Sprintf(` ORDER BY r.number DESC LIMIT %d`, limit)
|
||||
rows, err := t.q.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*Revision
|
||||
for rows.Next() {
|
||||
var r Revision
|
||||
if err := rows.Scan(&r.Number, &r.CreatedAt, &r.CreatedBy, &r.CreatedByName, &r.Comment, &r.WarningCount,
|
||||
&r.DNSFiles, &r.DHCPFiles); err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
out = append(out, &r)
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// Deployment is a row of deployments; AgentHostname comes from the agent.
|
||||
type Deployment struct {
|
||||
ID int64
|
||||
AgentID string
|
||||
AgentHostname string
|
||||
CorrelationID string
|
||||
Revision int64
|
||||
Trigger string
|
||||
Outcome string
|
||||
Message string
|
||||
ActiveRevision int64
|
||||
StartedAt time.Time
|
||||
FinishedAt time.Time
|
||||
ReportedAt time.Time
|
||||
}
|
||||
|
||||
const deploymentColumns = `d.id, d.agent_id, a.hostname, d.correlation_id, d.revision, d.trigger_kind, d.outcome, d.message,
|
||||
d.active_revision, d.started_at, d.finished_at, d.reported_at`
|
||||
|
||||
func scanDeployment(row pgx.Row) (*Deployment, error) {
|
||||
var d Deployment
|
||||
if err := row.Scan(&d.ID, &d.AgentID, &d.AgentHostname, &d.CorrelationID, &d.Revision, &d.Trigger, &d.Outcome,
|
||||
&d.Message, &d.ActiveRevision, &d.StartedAt, &d.FinishedAt, &d.ReportedAt); err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
return &d, nil
|
||||
}
|
||||
|
||||
// InsertDeployment stores a deployment report; ID and ReportedAt are set.
|
||||
func (t *Tx) InsertDeployment(ctx context.Context, d *Deployment) error {
|
||||
return mapErr(t.q.QueryRow(ctx, `
|
||||
INSERT INTO deployments (agent_id, correlation_id, revision, trigger_kind, outcome, message, active_revision,
|
||||
started_at, finished_at)
|
||||
VALUES ($1, $2, $3, $4::deployment_trigger, $5::deployment_outcome, $6, $7, $8, $9)
|
||||
RETURNING id, reported_at`,
|
||||
d.AgentID, d.CorrelationID, d.Revision, d.Trigger, d.Outcome, d.Message, d.ActiveRevision,
|
||||
d.StartedAt, d.FinishedAt).Scan(&d.ID, &d.ReportedAt))
|
||||
}
|
||||
|
||||
// DeploymentFilter selects deployments. Zero values mean "no restriction".
|
||||
type DeploymentFilter struct {
|
||||
AgentID string
|
||||
Revision int64
|
||||
BeforeID int64
|
||||
Limit int
|
||||
}
|
||||
|
||||
// ListDeployments returns deployments newest first.
|
||||
func (t *Tx) ListDeployments(ctx context.Context, f DeploymentFilter) ([]*Deployment, error) {
|
||||
var where []string
|
||||
var args []any
|
||||
add := func(cond string, v any) {
|
||||
args = append(args, v)
|
||||
where = append(where, fmt.Sprintf(cond, len(args)))
|
||||
}
|
||||
if f.AgentID != "" {
|
||||
add("d.agent_id = $%d", f.AgentID)
|
||||
}
|
||||
if f.Revision > 0 {
|
||||
add("d.revision = $%d", f.Revision)
|
||||
}
|
||||
if f.BeforeID > 0 {
|
||||
add("d.id < $%d", f.BeforeID)
|
||||
}
|
||||
limit := f.Limit
|
||||
if limit <= 0 {
|
||||
limit = 100
|
||||
}
|
||||
sql := `SELECT ` + deploymentColumns + ` FROM deployments d JOIN agents a ON a.id = d.agent_id`
|
||||
if len(where) > 0 {
|
||||
sql += " WHERE " + strings.Join(where, " AND ")
|
||||
}
|
||||
sql += fmt.Sprintf(" ORDER BY d.id DESC LIMIT %d", limit)
|
||||
rows, err := t.q.Query(ctx, sql, args...)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*Deployment
|
||||
for rows.Next() {
|
||||
d, err := scanDeployment(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, d)
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// LatestDeployments returns the newest deployment of every agent that has
|
||||
// one, keyed by agent ID.
|
||||
func (t *Tx) LatestDeployments(ctx context.Context) (map[string]*Deployment, error) {
|
||||
rows, err := t.q.Query(ctx, `SELECT DISTINCT ON (d.agent_id) `+deploymentColumns+`
|
||||
FROM deployments d JOIN agents a ON a.id = d.agent_id ORDER BY d.agent_id, d.id DESC`)
|
||||
if err != nil {
|
||||
return nil, mapErr(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string]*Deployment{}
|
||||
for rows.Next() {
|
||||
d, err := scanDeployment(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[d.AgentID] = d
|
||||
}
|
||||
return out, mapErr(rows.Err())
|
||||
}
|
||||
|
||||
// LatestDeployment returns the newest deployment of an agent or ErrNotFound.
|
||||
func (t *Tx) LatestDeployment(ctx context.Context, agentID string) (*Deployment, error) {
|
||||
return scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+`
|
||||
FROM deployments d JOIN agents a ON a.id = d.agent_id WHERE d.agent_id = $1 ORDER BY d.id DESC LIMIT 1`, agentID))
|
||||
}
|
||||
@@ -134,6 +134,15 @@ type querier interface {
|
||||
Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
|
||||
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
|
||||
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
|
||||
SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
|
||||
}
|
||||
|
||||
// execBatch sends b and returns the first error.
|
||||
func (t *Tx) execBatch(ctx context.Context, b *pgx.Batch) error {
|
||||
if b.Len() == 0 {
|
||||
return nil
|
||||
}
|
||||
return mapErr(t.q.SendBatch(ctx, b).Close())
|
||||
}
|
||||
|
||||
// Tx is a repository bound to a transaction.
|
||||
|
||||
Reference in New Issue
Block a user