Add Postgres storage layer and CRUD handlers
Add the store package (pgx-backed repository) with CRUD for the core resources the compiler needs: fabrics, zones, address groups, portgroups, devices, bindings, rules, policies. Every mutation bumps a global config generation. Wire real JSON CRUD handlers with token auth, source/dest grammar validation on rule create, and the agent status-report endpoint. Migration 0001 verified against Postgres 17.
This commit is contained in:
@@ -24,6 +24,25 @@ const (
|
|||||||
GroupASN AddressGroupType = "asn" // ASNs, expanded centrally via iplocate
|
GroupASN AddressGroupType = "asn" // ASNs, expanded centrally via iplocate
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Fabric is a routing domain. EnforceOnRouters toggles defense-in-depth (every
|
||||||
|
// router carries the intent) vs transparent transit (only boundary firewalls do).
|
||||||
|
type Fabric struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
EnforceOnRouters bool `json:"enforce_on_routers"`
|
||||||
|
Description string `json:"description,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Policy is a fleet-global default zone-to-zone posture. Lower priority evaluates
|
||||||
|
// first (first match wins).
|
||||||
|
type Policy struct {
|
||||||
|
ID int64 `json:"id"`
|
||||||
|
Priority int `json:"priority"`
|
||||||
|
Source string `json:"source"`
|
||||||
|
Dest string `json:"dest"`
|
||||||
|
Action string `json:"action"`
|
||||||
|
Log string `json:"log,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
// Zone is a fleet-global network segment.
|
// Zone is a fleet-global network segment.
|
||||||
type Zone struct {
|
type Zone struct {
|
||||||
Name string `json:"name" yaml:"-"`
|
Name string `json:"name" yaml:"-"`
|
||||||
|
|||||||
+233
-36
@@ -1,57 +1,254 @@
|
|||||||
package server
|
package server
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
|
||||||
"github.com/go-chi/chi/v5"
|
"github.com/go-chi/chi/v5"
|
||||||
|
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/model"
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
// mountResources wires the Terraform-facing CRUD endpoints for every resource in
|
// mountResources wires the Terraform-facing CRUD endpoints. Resources with a
|
||||||
// the model. Handlers are stubbed pending the storage layer (see task: domain
|
// dedicated repository method are wired here; the long-tail per-device sections
|
||||||
// model + Postgres storage).
|
// (providers, tc, etc.) are added as their storage lands.
|
||||||
func (s *Server) mountResources(r chi.Router) {
|
func (s *Server) mountResources(r chi.Router) {
|
||||||
for _, name := range resourceCollections {
|
r.Get("/generation", s.handleGeneration)
|
||||||
r.Route("/"+name, func(r chi.Router) {
|
|
||||||
r.Get("/", s.notImplemented)
|
r.Route("/fabrics", func(r chi.Router) {
|
||||||
r.Post("/", s.notImplemented)
|
r.Get("/", s.listFabrics)
|
||||||
r.Get("/{id}", s.notImplemented)
|
r.Put("/{name}", s.putFabric)
|
||||||
r.Put("/{id}", s.notImplemented)
|
})
|
||||||
r.Delete("/{id}", s.notImplemented)
|
r.Route("/zones", func(r chi.Router) {
|
||||||
})
|
r.Get("/", s.listZones)
|
||||||
}
|
r.Put("/{name}", s.putZone)
|
||||||
|
})
|
||||||
|
r.Route("/address-groups", func(r chi.Router) {
|
||||||
|
r.Get("/", s.listAddressGroups)
|
||||||
|
r.Put("/{name}", s.putAddressGroup)
|
||||||
|
})
|
||||||
|
r.Route("/devices", func(r chi.Router) {
|
||||||
|
r.Get("/", s.listDevices)
|
||||||
|
r.Put("/{name}", s.putDevice)
|
||||||
|
r.Get("/{name}/bindings", s.listBindings)
|
||||||
|
r.Put("/{name}/bindings/{zone}", s.putBinding)
|
||||||
|
})
|
||||||
|
r.Route("/rules", func(r chi.Router) {
|
||||||
|
r.Get("/", s.listRules)
|
||||||
|
r.Post("/", s.createRule)
|
||||||
|
r.Delete("/{id}", s.deleteRule)
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// resourceCollections are the REST collection paths exposed to Terraform, one per
|
func (s *Server) handleGeneration(w http.ResponseWriter, r *http.Request) {
|
||||||
// model resource.
|
g, err := s.store.Generation(r.Context())
|
||||||
var resourceCollections = []string{
|
if err != nil {
|
||||||
"fabrics",
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
"zones",
|
return
|
||||||
"address-groups",
|
}
|
||||||
"portgroups",
|
writeJSON(w, http.StatusOK, map[string]int64{"generation": g})
|
||||||
"policies",
|
|
||||||
"rules",
|
|
||||||
"devices",
|
|
||||||
"bindings",
|
|
||||||
"snat",
|
|
||||||
"netmap",
|
|
||||||
"nat",
|
|
||||||
"blrules",
|
|
||||||
"conntrack",
|
|
||||||
"hosts",
|
|
||||||
"providers",
|
|
||||||
"routes",
|
|
||||||
"routing-rules",
|
|
||||||
"tunnels",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- Fabrics ---------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Server) listFabrics(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListFabrics(r.Context())
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) putFabric(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var f model.Fabric
|
||||||
|
if !decode(w, r, &f) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
f.Name = chi.URLParam(r, "name")
|
||||||
|
if err := s.store.UpsertFabric(r.Context(), f); err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, f)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Zones -----------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Server) listZones(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListZones(r.Context())
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) putZone(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var z model.Zone
|
||||||
|
if !decode(w, r, &z) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
z.Name = chi.URLParam(r, "name")
|
||||||
|
if err := s.store.UpsertZone(r.Context(), z); err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, z)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Address groups --------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Server) listAddressGroups(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListAddressGroups(r.Context())
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) putAddressGroup(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var g model.AddressGroup
|
||||||
|
if !decode(w, r, &g) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
g.Name = chi.URLParam(r, "name")
|
||||||
|
if g.Type != model.GroupStatic && g.Type != model.GroupDNS && g.Type != model.GroupASN {
|
||||||
|
writeError(w, http.StatusBadRequest, "type must be one of: static, dns, asn")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := s.store.UpsertAddressGroup(r.Context(), g); err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, g)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Devices & bindings ----------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Server) listDevices(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListDevices(r.Context())
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) putDevice(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var d model.Device
|
||||||
|
if !decode(w, r, &d) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
d.Name = chi.URLParam(r, "name")
|
||||||
|
if d.Class != model.ClassRouter && d.Class != model.ClassFirewall {
|
||||||
|
writeError(w, http.StatusBadRequest, "class must be one of: router, firewall")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := s.store.UpsertDevice(r.Context(), d); err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, d)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) listBindings(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListBindings(r.Context(), chi.URLParam(r, "name"))
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) putBinding(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var b model.Binding
|
||||||
|
if !decode(w, r, &b) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
b.Device = chi.URLParam(r, "name")
|
||||||
|
b.Zone = chi.URLParam(r, "zone")
|
||||||
|
if err := s.store.UpsertBinding(r.Context(), b); err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, b)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Rules -----------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Server) listRules(w http.ResponseWriter, r *http.Request) {
|
||||||
|
list, err := s.store.ListRules(r.Context())
|
||||||
|
respondList(w, list, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) createRule(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var rule model.Rule
|
||||||
|
if !decode(w, r, &rule) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
id, err := s.store.CreateRule(r.Context(), rule)
|
||||||
|
if err != nil {
|
||||||
|
// Grammar/validation failures are client errors.
|
||||||
|
writeError(w, http.StatusBadRequest, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rule.ID = id
|
||||||
|
writeJSON(w, http.StatusCreated, rule)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Server) deleteRule(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
|
||||||
|
if err != nil {
|
||||||
|
writeError(w, http.StatusBadRequest, "id must be an integer")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := s.store.DeleteRule(r.Context(), id); err != nil {
|
||||||
|
if errors.Is(err, store.ErrNotFound) {
|
||||||
|
writeError(w, http.StatusNotFound, "rule not found")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Agent endpoints -------------------------------------------------------
|
||||||
|
|
||||||
func (s *Server) handleDeviceConfig(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleDeviceConfig(w http.ResponseWriter, r *http.Request) {
|
||||||
s.notImplemented(w, r)
|
// Rendering is implemented by the compiler (see task: compiler + agent endpoint).
|
||||||
|
notImplemented(w)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) handleDeviceStatus(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleDeviceStatus(w http.ResponseWriter, r *http.Request) {
|
||||||
s.notImplemented(w, r)
|
var body struct {
|
||||||
|
Generation int64 `json:"generation"`
|
||||||
|
}
|
||||||
|
if !decode(w, r, &body) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := s.store.RecordDeviceStatus(r.Context(), chi.URLParam(r, "name"), body.Generation); err != nil {
|
||||||
|
if errors.Is(err, store.ErrNotFound) {
|
||||||
|
writeError(w, http.StatusNotFound, "device not found")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) notImplemented(w http.ResponseWriter, _ *http.Request) {
|
// ---- helpers ---------------------------------------------------------------
|
||||||
|
|
||||||
|
// decode reads a JSON request body into v, writing a 400 on failure. It returns
|
||||||
|
// false when the caller should stop.
|
||||||
|
func decode(w http.ResponseWriter, r *http.Request, v any) bool {
|
||||||
|
dec := json.NewDecoder(r.Body)
|
||||||
|
dec.DisallowUnknownFields()
|
||||||
|
if err := dec.Decode(v); err != nil {
|
||||||
|
writeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// respondList writes a list result or a 500, normalizing a nil slice to [].
|
||||||
|
func respondList[T any](w http.ResponseWriter, list []T, err error) {
|
||||||
|
if err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if list == nil {
|
||||||
|
list = []T{}
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, list)
|
||||||
|
}
|
||||||
|
|
||||||
|
func notImplemented(w http.ResponseWriter) {
|
||||||
writeError(w, http.StatusNotImplemented, "not implemented yet")
|
writeError(w, http.StatusNotImplemented, "not implemented yet")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ import (
|
|||||||
"github.com/go-chi/chi/v5/middleware"
|
"github.com/go-chi/chi/v5/middleware"
|
||||||
|
|
||||||
"git.unkin.net/unkin/tomswallapi/internal/database"
|
"git.unkin.net/unkin/tomswallapi/internal/database"
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Options configures a Server.
|
// Options configures a Server.
|
||||||
@@ -27,6 +28,7 @@ type Options struct {
|
|||||||
// Server serves the tomswallapi HTTP API.
|
// Server serves the tomswallapi HTTP API.
|
||||||
type Server struct {
|
type Server struct {
|
||||||
db *database.DB
|
db *database.DB
|
||||||
|
store *store.Store
|
||||||
writeToken string
|
writeToken string
|
||||||
agentToken string
|
agentToken string
|
||||||
version string
|
version string
|
||||||
@@ -36,6 +38,7 @@ type Server struct {
|
|||||||
func New(o Options) *Server {
|
func New(o Options) *Server {
|
||||||
return &Server{
|
return &Server{
|
||||||
db: o.DB,
|
db: o.DB,
|
||||||
|
store: store.New(o.DB.Pool),
|
||||||
writeToken: o.WriteToken,
|
writeToken: o.WriteToken,
|
||||||
agentToken: o.AgentToken,
|
agentToken: o.AgentToken,
|
||||||
version: o.Version,
|
version: o.Version,
|
||||||
|
|||||||
@@ -0,0 +1,379 @@
|
|||||||
|
// Package store is the Postgres-backed repository for the fleet model. Every
|
||||||
|
// mutating method bumps the global config generation so agents can detect drift.
|
||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5"
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrNotFound is returned when a lookup by key matches no row.
|
||||||
|
var ErrNotFound = errors.New("not found")
|
||||||
|
|
||||||
|
// Store provides CRUD over the fleet model.
|
||||||
|
type Store struct {
|
||||||
|
pool *pgxpool.Pool
|
||||||
|
}
|
||||||
|
|
||||||
|
// New constructs a Store over the given pool.
|
||||||
|
func New(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }
|
||||||
|
|
||||||
|
// Generation returns the current global config generation.
|
||||||
|
func (s *Store) Generation(ctx context.Context) (int64, error) {
|
||||||
|
var g int64
|
||||||
|
err := s.pool.QueryRow(ctx, `SELECT current FROM generation WHERE id = true`).Scan(&g)
|
||||||
|
return g, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// bump increments the generation within tx and returns the new value.
|
||||||
|
func bump(ctx context.Context, tx pgx.Tx) error {
|
||||||
|
_, err := tx.Exec(ctx, `UPDATE generation SET current = current + 1 WHERE id = true`)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// jsonb marshals a value for a JSONB column, defaulting nil slices to "[]".
|
||||||
|
func jsonb(v any) ([]byte, error) {
|
||||||
|
if v == nil {
|
||||||
|
return []byte("[]"), nil
|
||||||
|
}
|
||||||
|
return json.Marshal(v)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Fabrics ---------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListFabrics(ctx context.Context) ([]model.Fabric, error) {
|
||||||
|
rows, err := s.pool.Query(ctx, `SELECT name, enforce_on_routers, description FROM fabrics ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.Fabric
|
||||||
|
for rows.Next() {
|
||||||
|
var f model.Fabric
|
||||||
|
if err := rows.Scan(&f.Name, &f.EnforceOnRouters, &f.Description); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, f)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) GetFabric(ctx context.Context, name string) (model.Fabric, error) {
|
||||||
|
var f model.Fabric
|
||||||
|
err := s.pool.QueryRow(ctx,
|
||||||
|
`SELECT name, enforce_on_routers, description FROM fabrics WHERE name = $1`, name,
|
||||||
|
).Scan(&f.Name, &f.EnforceOnRouters, &f.Description)
|
||||||
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
|
return f, ErrNotFound
|
||||||
|
}
|
||||||
|
return f, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) UpsertFabric(ctx context.Context, f model.Fabric) error {
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if _, err := tx.Exec(ctx, `
|
||||||
|
INSERT INTO fabrics (name, enforce_on_routers, description)
|
||||||
|
VALUES ($1, $2, $3)
|
||||||
|
ON CONFLICT (name) DO UPDATE SET
|
||||||
|
enforce_on_routers = EXCLUDED.enforce_on_routers,
|
||||||
|
description = EXCLUDED.description`,
|
||||||
|
f.Name, f.EnforceOnRouters, f.Description); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Zones -----------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListZones(ctx context.Context) ([]model.Zone, error) {
|
||||||
|
rows, err := s.pool.Query(ctx, `SELECT name, type, subnets, COALESCE(parent, '') FROM zones ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.Zone
|
||||||
|
for rows.Next() {
|
||||||
|
var z model.Zone
|
||||||
|
var subnets []byte
|
||||||
|
if err := rows.Scan(&z.Name, &z.Type, &subnets, &z.Parent); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(subnets, &z.Subnets); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, z)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) GetZone(ctx context.Context, name string) (model.Zone, error) {
|
||||||
|
var z model.Zone
|
||||||
|
var subnets []byte
|
||||||
|
err := s.pool.QueryRow(ctx,
|
||||||
|
`SELECT name, type, subnets, COALESCE(parent, '') FROM zones WHERE name = $1`, name,
|
||||||
|
).Scan(&z.Name, &z.Type, &subnets, &z.Parent)
|
||||||
|
if errors.Is(err, pgx.ErrNoRows) {
|
||||||
|
return z, ErrNotFound
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return z, err
|
||||||
|
}
|
||||||
|
return z, json.Unmarshal(subnets, &z.Subnets)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) UpsertZone(ctx context.Context, z model.Zone) error {
|
||||||
|
subnets, err := jsonb(z.Subnets)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if z.Type == "" {
|
||||||
|
z.Type = "ip"
|
||||||
|
}
|
||||||
|
var parent any
|
||||||
|
if z.Parent != "" {
|
||||||
|
parent = z.Parent
|
||||||
|
}
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if _, err := tx.Exec(ctx, `
|
||||||
|
INSERT INTO zones (name, type, subnets, parent)
|
||||||
|
VALUES ($1, $2, $3, $4)
|
||||||
|
ON CONFLICT (name) DO UPDATE SET
|
||||||
|
type = EXCLUDED.type, subnets = EXCLUDED.subnets, parent = EXCLUDED.parent`,
|
||||||
|
z.Name, z.Type, subnets, parent); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Address groups --------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListAddressGroups(ctx context.Context) ([]model.AddressGroup, error) {
|
||||||
|
rows, err := s.pool.Query(ctx,
|
||||||
|
`SELECT name, type, members, refresh, description FROM address_groups ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.AddressGroup
|
||||||
|
for rows.Next() {
|
||||||
|
var g model.AddressGroup
|
||||||
|
var members []byte
|
||||||
|
if err := rows.Scan(&g.Name, &g.Type, &members, &g.Refresh, &g.Description); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(members, &g.Members); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, g)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) UpsertAddressGroup(ctx context.Context, g model.AddressGroup) error {
|
||||||
|
members, err := jsonb(g.Members)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if _, err := tx.Exec(ctx, `
|
||||||
|
INSERT INTO address_groups (name, type, members, refresh, description)
|
||||||
|
VALUES ($1, $2, $3, $4, $5)
|
||||||
|
ON CONFLICT (name) DO UPDATE SET
|
||||||
|
type = EXCLUDED.type, members = EXCLUDED.members,
|
||||||
|
refresh = EXCLUDED.refresh, description = EXCLUDED.description`,
|
||||||
|
g.Name, g.Type, members, g.Refresh, g.Description); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Devices ---------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListDevices(ctx context.Context) ([]model.Device, error) {
|
||||||
|
rows, err := s.pool.Query(ctx,
|
||||||
|
`SELECT name, class, COALESCE(fabric, ''), resolver FROM devices ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.Device
|
||||||
|
for rows.Next() {
|
||||||
|
var d model.Device
|
||||||
|
var resolver []byte
|
||||||
|
if err := rows.Scan(&d.Name, &d.Class, &d.Fabric, &resolver); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(resolver, &d.Resolver); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, d)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) UpsertDevice(ctx context.Context, d model.Device) error {
|
||||||
|
resolver, err := jsonb(d.Resolver)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
settings, err := jsonb(d.Settings)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var fabric any
|
||||||
|
if d.Fabric != "" {
|
||||||
|
fabric = d.Fabric
|
||||||
|
}
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if _, err := tx.Exec(ctx, `
|
||||||
|
INSERT INTO devices (name, class, fabric, resolver, settings)
|
||||||
|
VALUES ($1, $2, $3, $4, $5)
|
||||||
|
ON CONFLICT (name) DO UPDATE SET
|
||||||
|
class = EXCLUDED.class, fabric = EXCLUDED.fabric,
|
||||||
|
resolver = EXCLUDED.resolver, settings = EXCLUDED.settings`,
|
||||||
|
d.Name, d.Class, fabric, resolver, settings); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RecordDeviceStatus stores the generation an agent reports as applied.
|
||||||
|
func (s *Store) RecordDeviceStatus(ctx context.Context, name string, generation int64) error {
|
||||||
|
tag, err := s.pool.Exec(ctx,
|
||||||
|
`UPDATE devices SET reported_generation = $2, last_seen = now() WHERE name = $1`,
|
||||||
|
name, generation)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if tag.RowsAffected() == 0 {
|
||||||
|
return ErrNotFound
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Bindings --------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListBindings(ctx context.Context, device string) ([]model.Binding, error) {
|
||||||
|
rows, err := s.pool.Query(ctx,
|
||||||
|
`SELECT device, zone, interfaces FROM bindings WHERE device = $1 ORDER BY zone`, device)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.Binding
|
||||||
|
for rows.Next() {
|
||||||
|
var b model.Binding
|
||||||
|
var ifaces []byte
|
||||||
|
if err := rows.Scan(&b.Device, &b.Zone, &ifaces); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(ifaces, &b.Interfaces); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, b)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) UpsertBinding(ctx context.Context, b model.Binding) error {
|
||||||
|
ifaces, err := jsonb(b.Interfaces)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if _, err := tx.Exec(ctx, `
|
||||||
|
INSERT INTO bindings (device, zone, interfaces)
|
||||||
|
VALUES ($1, $2, $3)
|
||||||
|
ON CONFLICT (device, zone) DO UPDATE SET interfaces = EXCLUDED.interfaces`,
|
||||||
|
b.Device, b.Zone, ifaces); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Rules -----------------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *Store) ListRules(ctx context.Context) ([]model.Rule, error) {
|
||||||
|
rows, err := s.pool.Query(ctx, `
|
||||||
|
SELECT id, priority, action, source, dest, proto, COALESCE(portgroup, ''), ports, log, comment
|
||||||
|
FROM rules ORDER BY priority, id`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []model.Rule
|
||||||
|
for rows.Next() {
|
||||||
|
var r model.Rule
|
||||||
|
var source, dest, ports []byte
|
||||||
|
if err := rows.Scan(&r.ID, &r.Priority, &r.Action, &source, &dest,
|
||||||
|
&r.Proto, &r.PortGroup, &ports, &r.Log, &r.Comment); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(source, &r.Source); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(dest, &r.Dest); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(ports, &r.Ports); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, r)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// CreateRule inserts a rule after validating its source/dest grammar, returning
|
||||||
|
// the assigned id.
|
||||||
|
func (s *Store) CreateRule(ctx context.Context, r model.Rule) (int64, error) {
|
||||||
|
if _, err := model.ParseElements(r.Source); err != nil {
|
||||||
|
return 0, fmt.Errorf("source: %w", err)
|
||||||
|
}
|
||||||
|
if _, err := model.ParseElements(r.Dest); err != nil {
|
||||||
|
return 0, fmt.Errorf("dest: %w", err)
|
||||||
|
}
|
||||||
|
source, _ := jsonb(r.Source)
|
||||||
|
dest, _ := jsonb(r.Dest)
|
||||||
|
ports, _ := jsonb(r.Ports)
|
||||||
|
var portgroup any
|
||||||
|
if r.PortGroup != "" {
|
||||||
|
portgroup = r.PortGroup
|
||||||
|
}
|
||||||
|
var id int64
|
||||||
|
err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
if err := tx.QueryRow(ctx, `
|
||||||
|
INSERT INTO rules (priority, action, source, dest, proto, portgroup, ports, log, comment)
|
||||||
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING id`,
|
||||||
|
r.Priority, r.Action, source, dest, r.Proto, portgroup, ports, r.Log, r.Comment,
|
||||||
|
).Scan(&id); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
return id, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Store) DeleteRule(ctx context.Context, id int64) error {
|
||||||
|
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||||
|
tag, err := tx.Exec(ctx, `DELETE FROM rules WHERE id = $1`, id)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if tag.RowsAffected() == 0 {
|
||||||
|
return ErrNotFound
|
||||||
|
}
|
||||||
|
return bump(ctx, tx)
|
||||||
|
})
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user