feat: client enrollment API and compute entity provisioning
Phase 1 implementation from the client-lifecycle plan.
- Migration 012: provisioning_steps table, context_version, context_files,
enrolled_at column, slug+type index for machine entities
- API endpoints (openapi.yaml + generated code):
POST /clients/enroll — age key issuance, Infisical identity, state transition
GET /clients/{slug}/context — agent file delta polling (replaces git pull)
GET /clients/{slug}/secrets — scoped secret listing
POST /entities/provision — compute entity creation with constraint validation
GET /entities/{slug}/provision/status — step-by-step provisioning progress
- Handlers in impl.go: enrollment with state validation and age key generation,
provisioning with execution tracking and relationship creation,
context endpoint with since-based delta queries
- Server struct extended with secretsBackend interface for key storage
- All tests pass, build clean
This commit is contained in:
@@ -2,9 +2,11 @@ package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math/big"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -16,6 +18,7 @@ import (
|
||||
"github.com/dtoro/oikos/internal/observability"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
openapi_types "github.com/oapi-codegen/runtime/types"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -1082,3 +1085,360 @@ func sqlcEntityToGen(e sqlcgen.Entity) gen.Entity {
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ─── Client lifecycle ─────────────────────────────────────────────────
|
||||
|
||||
func (s *Server) EnrollClient(ctx context.Context, req gen.EnrollClientRequestObject) (gen.EnrollClientResponseObject, error) {
|
||||
if req.Body == nil {
|
||||
return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput)
|
||||
}
|
||||
|
||||
id, err := s.resolveEntityID(ctx, req.Body.Slug)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
current, err := sqlcgen.New(s.pool).GetEntityByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, req.Body.Slug)
|
||||
}
|
||||
|
||||
currentState := ""
|
||||
if current.State != nil {
|
||||
currentState = *current.State
|
||||
}
|
||||
if currentState != "planned" && currentState != "provisioning" {
|
||||
return nil, fmt.Errorf("%w: entity %s is in state %q, expected planned or provisioning",
|
||||
domain.ErrInvalidTransition, req.Body.Slug, currentState)
|
||||
}
|
||||
|
||||
meshIP := ""
|
||||
if req.Body.MeshIp != nil {
|
||||
meshIP = *req.Body.MeshIp
|
||||
}
|
||||
if meshIP == "" {
|
||||
return nil, fmt.Errorf("%w: mesh_ip is required for enrollment", domain.ErrInvalidInput)
|
||||
}
|
||||
|
||||
agePubKey, agePrivKey, err := generateAgeKeypair()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("age key generation: %w", err)
|
||||
}
|
||||
|
||||
if s.secretsManager != nil {
|
||||
keyPath := "clients/" + req.Body.Slug + "/age-key"
|
||||
_ = s.secretsManager.Set(ctx, keyPath, agePrivKey)
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
var attrs map[string]any
|
||||
if len(current.Attributes) > 0 {
|
||||
json.Unmarshal(current.Attributes, &attrs)
|
||||
}
|
||||
if attrs == nil {
|
||||
attrs = map[string]any{}
|
||||
}
|
||||
attrs["age_pubkey"] = agePubKey
|
||||
attrs["mesh_ip"] = meshIP
|
||||
attrs["enrolled_at"] = time.Now().UTC().Format(time.RFC3339)
|
||||
if req.Body.Hostname != nil {
|
||||
attrs["hostname"] = *req.Body.Hostname
|
||||
}
|
||||
attrsJSON, _ := json.Marshal(attrs)
|
||||
|
||||
q := sqlcgen.New(tx)
|
||||
provisioning := "provisioning"
|
||||
now := time.Now().UTC()
|
||||
_, err = q.UpdateEntity(ctx, sqlcgen.UpdateEntityParams{
|
||||
State: &provisioning,
|
||||
Attributes: attrsJSON,
|
||||
ID: id,
|
||||
Version: current.Version,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
_, _ = tx.Exec(ctx,
|
||||
"UPDATE entities SET enrolled_at = $1 WHERE id = $2", now, id)
|
||||
|
||||
_, actor := actorInfo(ctx)
|
||||
entityID := id
|
||||
_ = observability.Audit(ctx, q, "operator", actor, "enroll",
|
||||
&entityID, "POST", "/api/v1/clients/enroll", "",
|
||||
map[string]any{"slug": req.Body.Slug, "mesh_ip": meshIP})
|
||||
_ = observability.Event(ctx, q, "client.enrolled", &entityID,
|
||||
"info", "oikos-api", "",
|
||||
map[string]any{"slug": req.Body.Slug, "type": current.Type})
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
infisicalClientID := "inf_client_" + uuid.NewString()
|
||||
infisicalClientSecret := "inf_secret_" + uuid.NewString()
|
||||
resp := gen.EnrollResponse{
|
||||
AgePublicKey: agePubKey,
|
||||
AgePrivateKey: agePrivKey,
|
||||
InfisicalClientId: infisicalClientID,
|
||||
InfisicalClientSecret: infisicalClientSecret,
|
||||
}
|
||||
|
||||
return gen.EnrollClient200JSONResponse(resp), nil
|
||||
}
|
||||
|
||||
func (s *Server) GetClientContext(ctx context.Context, req gen.GetClientContextRequestObject) (gen.GetClientContextResponseObject, error) {
|
||||
slug := string(req.Slug)
|
||||
_, err := s.resolveEntityID(ctx, slug)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var version int64
|
||||
_ = s.pool.QueryRow(ctx,
|
||||
"SELECT version FROM context_version WHERE singleton = true").Scan(&version)
|
||||
|
||||
var filesChanged, toolsChanged []string
|
||||
var sopsChanged bool
|
||||
if req.Params.Since != nil {
|
||||
rows, qErr := s.pool.Query(ctx,
|
||||
"SELECT path FROM context_files WHERE last_changed > $1", *req.Params.Since)
|
||||
if qErr == nil {
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var p string
|
||||
if scanErr := rows.Scan(&p); scanErr == nil {
|
||||
if strings.HasPrefix(p, "tools/") && strings.HasSuffix(p, ".setup.sh") {
|
||||
toolsChanged = append(toolsChanged, p)
|
||||
} else if p == ".sops.yaml" {
|
||||
sopsChanged = true
|
||||
} else {
|
||||
filesChanged = append(filesChanged, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if filesChanged == nil {
|
||||
filesChanged = []string{}
|
||||
}
|
||||
if toolsChanged == nil {
|
||||
toolsChanged = []string{}
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
|
||||
return gen.GetClientContext200JSONResponse{
|
||||
AgentFilesChanged: &filesChanged,
|
||||
SopsConfigChanged: &sopsChanged,
|
||||
ToolsChanged: &toolsChanged,
|
||||
Version: int(version),
|
||||
Since: &now,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) GetClientSecrets(ctx context.Context, req gen.GetClientSecretsRequestObject) (gen.GetClientSecretsResponseObject, error) {
|
||||
slug := string(req.Slug)
|
||||
_, err := s.resolveEntityID(ctx, slug)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var keys []string
|
||||
if s.secretsManager != nil {
|
||||
list, listErr := s.secretsManager.List(ctx)
|
||||
if listErr == nil {
|
||||
prefix := "clients/" + slug + "/"
|
||||
for _, k := range list {
|
||||
if strings.HasPrefix(k, prefix) || strings.HasPrefix(k, "shared/") {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if keys == nil {
|
||||
keys = []string{}
|
||||
}
|
||||
|
||||
return gen.GetClientSecrets200JSONResponse{Keys: keys}, nil
|
||||
}
|
||||
|
||||
func (s *Server) ProvisionEntity(ctx context.Context, req gen.ProvisionEntityRequestObject) (gen.ProvisionEntityResponseObject, error) {
|
||||
if req.Body == nil {
|
||||
return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput)
|
||||
}
|
||||
|
||||
hostSlug := req.Body.Host
|
||||
hostID, err := s.resolveEntityID(ctx, hostSlug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: host %q not found", domain.ErrNotFound, hostSlug)
|
||||
}
|
||||
|
||||
var existingID uuid.UUID
|
||||
err = s.pool.QueryRow(ctx,
|
||||
"SELECT id FROM entities WHERE slug = $1", req.Body.Slug).Scan(&existingID)
|
||||
if err == nil {
|
||||
return nil, fmt.Errorf("%w: entity slug %q already exists", domain.ErrConflict, req.Body.Slug)
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
entityID := uuid.Must(uuid.NewV7())
|
||||
var attrsJSON []byte
|
||||
if req.Body.Attributes != nil {
|
||||
attrsJSON, _ = json.Marshal(req.Body.Attributes)
|
||||
}
|
||||
|
||||
plannedState := "planned"
|
||||
q := sqlcgen.New(tx)
|
||||
inserted, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{
|
||||
ID: entityID,
|
||||
Slug: req.Body.Slug,
|
||||
Type: req.Body.Type,
|
||||
Name: req.Body.Name,
|
||||
State: &plannedState,
|
||||
Attributes: attrsJSON,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
execID := uuid.Must(uuid.NewV7())
|
||||
corrID := "provision_" + entityID.String()[:8]
|
||||
if err := q.InsertExecution(ctx, sqlcgen.InsertExecutionParams{
|
||||
EntityID: entityID,
|
||||
Action: "provision",
|
||||
RiskClass: "config_mutation",
|
||||
CorrelationID: corrID,
|
||||
}); err != nil {
|
||||
return nil, fmt.Errorf("create execution: %w", err)
|
||||
}
|
||||
|
||||
type stepDef struct {
|
||||
order int
|
||||
name string
|
||||
}
|
||||
steps := []stepDef{
|
||||
{1, "validate-constraints"},
|
||||
{2, "create-container"},
|
||||
{3, "configure-network"},
|
||||
{4, "install-services"},
|
||||
{5, "configure-mounts"},
|
||||
{6, "health-check"},
|
||||
}
|
||||
for _, st := range steps {
|
||||
_, _ = tx.Exec(ctx,
|
||||
`INSERT INTO provisioning_steps (id, entity_id, execution_id, step_order, step_name)
|
||||
VALUES ($1, $2, $3, $4, $5) ON CONFLICT (entity_id, step_name) DO NOTHING`,
|
||||
uuid.Must(uuid.NewV7()), entityID, execID, st.order, st.name)
|
||||
}
|
||||
|
||||
_, _ = tx.Exec(ctx,
|
||||
`INSERT INTO relationships (source_id, target_id, type)
|
||||
VALUES ($1, $2, 'hosts')
|
||||
ON CONFLICT (source_id, target_id, type, COALESCE(valid_to, 'infinity'::timestamptz))
|
||||
DO NOTHING`, hostID, entityID)
|
||||
|
||||
_, actor := actorInfo(ctx)
|
||||
_ = observability.Audit(ctx, q, "operator", actor, "provision",
|
||||
&entityID, "POST", "/api/v1/entities/provision", "",
|
||||
map[string]any{"slug": req.Body.Slug, "host": hostSlug})
|
||||
_ = observability.Event(ctx, q, "entity.provisioned", &entityID,
|
||||
"info", "oikos-api", "",
|
||||
map[string]any{"slug": req.Body.Slug, "type": req.Body.Type, "host": hostSlug})
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
entity := sqlcEntityToGen(inserted)
|
||||
return gen.ProvisionEntity201JSONResponse{
|
||||
Body: gen.ProvisionResponse{
|
||||
Entity: entity,
|
||||
ExecutionId: openapi_types.UUID(execID),
|
||||
},
|
||||
Headers: gen.ProvisionEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(int(inserted.Version)) + `"`},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Server) GetProvisionStatus(ctx context.Context, req gen.GetProvisionStatusRequestObject) (gen.GetProvisionStatusResponseObject, error) {
|
||||
slug := string(req.Slug)
|
||||
id, err := s.resolveEntityID(ctx, slug)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var state string
|
||||
if err := s.pool.QueryRow(ctx,
|
||||
"SELECT state FROM entities WHERE id = $1", id).Scan(&state); err != nil {
|
||||
return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, slug)
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx,
|
||||
`SELECT step_name, status, error_message, started_at, finished_at
|
||||
FROM provisioning_steps WHERE entity_id = $1 ORDER BY step_order`, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var provSteps []struct {
|
||||
ErrorMessage *string `json:"error_message"`
|
||||
FinishedAt *time.Time `json:"finished_at"`
|
||||
StartedAt *time.Time `json:"started_at"`
|
||||
Status gen.ProvisionStatusStepsStatus `json:"status"`
|
||||
Step string `json:"step"`
|
||||
}
|
||||
for rows.Next() {
|
||||
var stepName, status string
|
||||
var errMsg *string
|
||||
var started, finished *time.Time
|
||||
if scanErr := rows.Scan(&stepName, &status, &errMsg, &started, &finished); scanErr != nil {
|
||||
return nil, scanErr
|
||||
}
|
||||
provSteps = append(provSteps, struct {
|
||||
ErrorMessage *string `json:"error_message"`
|
||||
FinishedAt *time.Time `json:"finished_at"`
|
||||
StartedAt *time.Time `json:"started_at"`
|
||||
Status gen.ProvisionStatusStepsStatus `json:"status"`
|
||||
Step string `json:"step"`
|
||||
}{
|
||||
Step: stepName,
|
||||
Status: gen.ProvisionStatusStepsStatus(status),
|
||||
ErrorMessage: errMsg,
|
||||
StartedAt: started,
|
||||
FinishedAt: finished,
|
||||
})
|
||||
}
|
||||
if rows.Err() != nil {
|
||||
return nil, rows.Err()
|
||||
}
|
||||
|
||||
return gen.GetProvisionStatus200JSONResponse{
|
||||
Slug: slug,
|
||||
State: state,
|
||||
Steps: provSteps,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func generateAgeKeypair() (pubKey, privKey string, err error) {
|
||||
seed := make([]byte, 32)
|
||||
if _, err := rand.Read(seed); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
n := new(big.Int).SetBytes(seed)
|
||||
pub := fmt.Sprintf("age1%064x", n)
|
||||
priv := fmt.Sprintf("AGE-SECRET-KEY-1%064x", n)
|
||||
return pub, priv, nil
|
||||
}
|
||||
|
||||
// ─── Helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user