Files
TA-Deployment-Broker/internal/app/fleet.go
T
2026-07-28 19:57:39 -05:00

414 lines
13 KiB
Go

package app
import (
"context"
"crypto/hmac"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"io"
"net/http"
"regexp"
"strings"
"time"
)
const fleetCredentialPrefix = "Bearer "
var (
installationIDPattern = regexp.MustCompile(
`^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`,
)
hexCredentialPattern = regexp.MustCompile(`^[0-9a-f]{64}$`)
fleetValuePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._+~,:/() -]*$`)
fleetHostnamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,252}$`)
errorCodePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]*$`)
)
type fleetRequest struct {
SchemaVersion int `json:"schema_version"`
InstallationID string `json:"installation_id"`
Hostname string `json:"hostname,omitempty"`
Credential string `json:"credential,omitempty"`
Event string `json:"event,omitempty"`
Result string `json:"result,omitempty"`
ProxMenuVersion string `json:"proxmenu_version,omitempty"`
GitCommit string `json:"git_commit,omitempty"`
PVEVersion string `json:"pve_version,omitempty"`
OSVersion string `json:"os_version,omitempty"`
KernelVersion string `json:"kernel_version,omitempty"`
Architecture string `json:"architecture,omitempty"`
Clustered bool `json:"clustered"`
ErrorCode string `json:"error_code,omitempty"`
DurationSeconds int `json:"duration_seconds,omitempty"`
}
type fleetHostRecord struct {
InstallationID string
Hostname string
VerifiedAt sql.NullTime
FirstSeenAt time.Time
LastSeenAt time.Time
LastSuccessAt sql.NullTime
LastEvent string
LastResult string
LastErrorCode string
ProxMenuVersion string
GitCommit string
PVEVersion string
OSVersion string
KernelVersion string
Architecture string
Clustered bool
}
func (s *Server) handleFleetRegister(w http.ResponseWriter, r *http.Request) {
request, ok := decodeFleetRequest(w, r, true)
if !ok {
return
}
sourceHash := s.fleetSourceHash(s.clientIP(r))
var recent int
if err := s.db.QueryRowContext(
r.Context(),
`SELECT COUNT(*) FROM fleet_hosts
WHERE registration_source_hash = ?
AND first_seen_at > datetime('now', '-1 day')`,
sourceHash,
).Scan(&recent); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to register host"})
return
}
credentialHash := hashValue(request.Credential)
var existingHash []byte
err := s.db.QueryRowContext(
r.Context(),
`SELECT credential_hash FROM fleet_hosts WHERE installation_id = ?`,
request.InstallationID,
).Scan(&existingHash)
switch {
case errors.Is(err, sql.ErrNoRows):
if recent >= 25 {
w.Header().Set("Retry-After", "86400")
writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": "registration limit reached"})
return
}
if err := s.insertFleetHost(r.Context(), request, credentialHash[:], sourceHash); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to register host"})
return
}
writeJSON(w, http.StatusCreated, map[string]string{"status": "registered"})
case err != nil:
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to register host"})
case !hmac.Equal(existingHash, credentialHash[:]):
writeJSON(w, http.StatusForbidden, map[string]string{"error": "host credential is invalid"})
default:
if err := s.updateFleetHost(r.Context(), request, false); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to update host"})
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "registered"})
}
}
func (s *Server) handleFleetEvent(w http.ResponseWriter, r *http.Request) {
request, ok := decodeFleetRequest(w, r, false)
if !ok {
return
}
authorization := strings.TrimSpace(r.Header.Get("Authorization"))
if !strings.HasPrefix(authorization, fleetCredentialPrefix) {
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "host credential required"})
return
}
credential := strings.TrimSpace(strings.TrimPrefix(authorization, fleetCredentialPrefix))
if !hexCredentialPattern.MatchString(credential) {
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "host credential required"})
return
}
credentialHash := hashValue(credential)
var authorized int
if err := s.db.QueryRowContext(
r.Context(),
`SELECT COUNT(*) FROM fleet_hosts
WHERE installation_id = ? AND credential_hash = ?`,
request.InstallationID,
credentialHash[:],
).Scan(&authorized); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to authenticate host"})
return
}
if authorized != 1 {
writeJSON(w, http.StatusForbidden, map[string]string{"error": "host credential is invalid"})
return
}
if err := s.updateFleetHost(r.Context(), request, true); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "unable to record event"})
return
}
w.WriteHeader(http.StatusNoContent)
}
func decodeFleetRequest(w http.ResponseWriter, r *http.Request, registration bool) (fleetRequest, bool) {
var request fleetRequest
r.Body = http.MaxBytesReader(w, r.Body, 16<<10)
decoder := json.NewDecoder(r.Body)
decoder.DisallowUnknownFields()
if err := decoder.Decode(&request); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request"})
return request, false
}
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request"})
return request, false
}
request.InstallationID = strings.ToLower(strings.TrimSpace(request.InstallationID))
request.Hostname = strings.TrimSpace(request.Hostname)
request.Credential = strings.ToLower(strings.TrimSpace(request.Credential))
request.Event = strings.TrimSpace(request.Event)
request.Result = strings.TrimSpace(request.Result)
request.ErrorCode = strings.TrimSpace(request.ErrorCode)
if request.SchemaVersion != 1 ||
!installationIDPattern.MatchString(request.InstallationID) ||
(request.Hostname != "" && !fleetHostnamePattern.MatchString(request.Hostname)) ||
(registration && !hexCredentialPattern.MatchString(request.Credential)) ||
!validFleetMetadata(request) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid host data"})
return request, false
}
if registration {
request.Event = "installed"
request.Result = "success"
} else if !validFleetEvent(request.Event, request.Result, request.ErrorCode) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid event data"})
return request, false
}
return request, true
}
func validFleetMetadata(request fleetRequest) bool {
values := []struct {
value string
limit int
}{
{request.ProxMenuVersion, 64},
{request.GitCommit, 64},
{request.PVEVersion, 128},
{request.OSVersion, 128},
{request.KernelVersion, 128},
{request.Architecture, 32},
}
for _, candidate := range values {
if len(candidate.value) > candidate.limit ||
(candidate.value != "" && !fleetValuePattern.MatchString(candidate.value)) {
return false
}
}
return request.DurationSeconds >= 0 && request.DurationSeconds <= 31*24*60*60
}
func validFleetEvent(event, result, errorCode string) bool {
switch event {
case "run_started", "run_completed", "run_failed", "upgraded":
default:
return false
}
switch result {
case "started", "success", "warning", "failure":
default:
return false
}
return errorCode == "" || (len(errorCode) <= 64 && errorCodePattern.MatchString(errorCode))
}
func (s *Server) insertFleetHost(
ctx context.Context,
request fleetRequest,
credentialHash []byte,
sourceHash []byte,
) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if _, err := tx.ExecContext(
ctx,
`INSERT INTO fleet_hosts
(installation_id, hostname, credential_hash, registration_source_hash,
proxmenu_version, git_commit, pve_version, os_version,
kernel_version, architecture, clustered)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
request.InstallationID, request.Hostname, credentialHash, sourceHash,
request.ProxMenuVersion, request.GitCommit, request.PVEVersion,
request.OSVersion, request.KernelVersion, request.Architecture,
request.Clustered,
); err != nil {
return err
}
if _, err := tx.ExecContext(
ctx,
`INSERT INTO fleet_events
(installation_id, event_type, result, proxmenu_version)
VALUES (?, 'installed', 'success', ?)`,
request.InstallationID, request.ProxMenuVersion,
); err != nil {
return err
}
return tx.Commit()
}
func (s *Server) updateFleetHost(ctx context.Context, request fleetRequest, recordEvent bool) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
result, err := tx.ExecContext(
ctx,
`UPDATE fleet_hosts SET
last_seen_at = CURRENT_TIMESTAMP,
last_success_at = CASE
WHEN ? = 'run_completed' AND ? = 'success' THEN CURRENT_TIMESTAMP
ELSE last_success_at END,
last_event = CASE WHEN ? THEN ? ELSE last_event END,
last_result = CASE WHEN ? THEN ? ELSE last_result END,
last_error_code = CASE WHEN ? THEN ? ELSE last_error_code END,
hostname = CASE WHEN ? <> '' THEN ? ELSE hostname END,
proxmenu_version = ?,
git_commit = ?,
pve_version = ?,
os_version = ?,
kernel_version = ?,
architecture = ?,
clustered = ?,
updated_at = CURRENT_TIMESTAMP
WHERE installation_id = ?`,
request.Event, request.Result,
recordEvent, request.Event,
recordEvent, request.Result,
recordEvent, request.ErrorCode,
request.Hostname, request.Hostname,
request.ProxMenuVersion, request.GitCommit, request.PVEVersion,
request.OSVersion, request.KernelVersion, request.Architecture,
request.Clustered, request.InstallationID,
)
if err != nil {
return err
}
affected, _ := result.RowsAffected()
if affected != 1 {
return sql.ErrNoRows
}
if recordEvent {
if _, err := tx.ExecContext(
ctx,
`INSERT INTO fleet_events
(installation_id, event_type, result, proxmenu_version,
error_code, duration_seconds)
VALUES (?, ?, ?, ?, ?, ?)`,
request.InstallationID, request.Event, request.Result,
request.ProxMenuVersion, request.ErrorCode, request.DurationSeconds,
); err != nil {
return err
}
}
return tx.Commit()
}
func (s *Server) fleetSourceHash(sourceIP string) []byte {
mac := hmac.New(sha256.New, s.cfg.CookieSecret)
_, _ = mac.Write([]byte("fleet-registration:"))
_, _ = mac.Write([]byte(sourceIP))
return mac.Sum(nil)
}
func (s *Server) verifyFleetInstallation(ctx context.Context, installationID string) {
installationID = strings.ToLower(strings.TrimSpace(installationID))
if !installationIDPattern.MatchString(installationID) {
return
}
_, _ = s.db.ExecContext(
ctx,
`UPDATE fleet_hosts
SET verified_at = COALESCE(verified_at, CURRENT_TIMESTAMP),
last_seen_at = CURRENT_TIMESTAMP,
updated_at = CURRENT_TIMESTAMP
WHERE installation_id = ?`,
installationID,
)
}
func (s *Server) listFleetHosts(r *http.Request) ([]fleetHostRecord, error) {
rows, err := s.db.QueryContext(
r.Context(),
`SELECT installation_id, hostname, verified_at, first_seen_at, last_seen_at,
last_success_at, last_event, last_result, last_error_code,
proxmenu_version, git_commit, pve_version, os_version,
kernel_version, architecture, clustered
FROM fleet_hosts
ORDER BY last_seen_at DESC
LIMIT 500`,
)
if err != nil {
return nil, err
}
defer rows.Close()
var records []fleetHostRecord
for rows.Next() {
var record fleetHostRecord
if err := rows.Scan(
&record.InstallationID, &record.Hostname, &record.VerifiedAt, &record.FirstSeenAt,
&record.LastSeenAt, &record.LastSuccessAt, &record.LastEvent,
&record.LastResult, &record.LastErrorCode, &record.ProxMenuVersion,
&record.GitCommit, &record.PVEVersion, &record.OSVersion,
&record.KernelVersion, &record.Architecture, &record.Clustered,
); err != nil {
return nil, err
}
records = append(records, record)
}
return records, rows.Err()
}
func (s *Server) handleFleetPortal(w http.ResponseWriter, r *http.Request) {
tech, err := s.currentTechnician(r)
if err != nil {
http.Redirect(w, r, "/", http.StatusSeeOther)
return
}
hosts, err := s.listFleetHosts(r)
if err != nil {
http.Error(w, "unable to list hosts", http.StatusInternalServerError)
return
}
s.render(w, "hosts.html", pageData{
Title: "ProxMenu Hosts",
Technician: tech,
CSRFToken: tech.CSRFToken,
FleetHosts: hosts,
CurrentView: "hosts",
})
}
func shortInstallationID(value string) string {
if len(value) <= 8 {
return value
}
return value[:8]
}
func shortCommit(value string) string {
value = strings.TrimSpace(value)
if len(value) <= 12 {
return value
}
if _, err := hex.DecodeString(value[:12]); err != nil {
return value
}
return value[:12]
}