Session 4 complete - RedFlag update management platform

🚩 Private development - version retention only ✅ Complete web dashboard (React + TypeScript + TailwindCSS) ✅ Production-ready server backend (Go + Gin + PostgreSQL) ✅ Linux agent with APT + Docker scanning + local CLI tools ✅ JWT authentication and REST API ✅ Update discovery and approval workflow 🚧 Status: Alpha software - active development 📦 Purpose: Version retention during development ⚠️ Not for public use or deployment
2025-10-13 16:46:31 -04:00
commit 55b7d03010
57 changed files with 7326 additions and 0 deletions
--- a/aggregator-server/internal/database/db.go
+++ b/aggregator-server/internal/database/db.go
@@ -0,0 +1,76 @@
+package database
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+	_ "github.com/lib/pq"
+)
+
+// DB wraps the database connection
+type DB struct {
+	*sqlx.DB
+}
+
+// Connect establishes a connection to the PostgreSQL database
+func Connect(databaseURL string) (*DB, error) {
+	db, err := sqlx.Connect("postgres", databaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to database: %w", err)
+	}
+
+	// Configure connection pool
+	db.SetMaxOpenConns(25)
+	db.SetMaxIdleConns(5)
+
+	// Test the connection
+	if err := db.Ping(); err != nil {
+		return nil, fmt.Errorf("failed to ping database: %w", err)
+	}
+
+	return &DB{db}, nil
+}
+
+// Migrate runs database migrations
+func (db *DB) Migrate(migrationsPath string) error {
+	// Read migration files
+	files, err := os.ReadDir(migrationsPath)
+	if err != nil {
+		return fmt.Errorf("failed to read migrations directory: %w", err)
+	}
+
+	// Filter and sort .up.sql files
+	var migrationFiles []string
+	for _, file := range files {
+		if strings.HasSuffix(file.Name(), ".up.sql") {
+			migrationFiles = append(migrationFiles, file.Name())
+		}
+	}
+	sort.Strings(migrationFiles)
+
+	// Execute migrations
+	for _, filename := range migrationFiles {
+		path := filepath.Join(migrationsPath, filename)
+		content, err := os.ReadFile(path)
+		if err != nil {
+			return fmt.Errorf("failed to read migration %s: %w", filename, err)
+		}
+
+		if _, err := db.Exec(string(content)); err != nil {
+			return fmt.Errorf("failed to execute migration %s: %w", filename, err)
+		}
+
+		fmt.Printf("✓ Executed migration: %s\n", filename)
+	}
+
+	return nil
+}
+
+// Close closes the database connection
+func (db *DB) Close() error {
+	return db.DB.Close()
+}
--- a/aggregator-server/internal/database/migrations/001_initial_schema.down.sql
+++ b/aggregator-server/internal/database/migrations/001_initial_schema.down.sql
@@ -0,0 +1,11 @@
+-- Drop tables in reverse order (respecting foreign key constraints)
+DROP TABLE IF EXISTS agent_commands;
+DROP TABLE IF EXISTS users;
+DROP TABLE IF EXISTS agent_tags;
+DROP TABLE IF EXISTS update_logs;
+DROP TABLE IF EXISTS update_packages;
+DROP TABLE IF EXISTS agent_specs;
+DROP TABLE IF EXISTS agents;
+
+-- Drop extension
+DROP EXTENSION IF EXISTS "uuid-ossp";
--- a/aggregator-server/internal/database/migrations/001_initial_schema.up.sql
+++ b/aggregator-server/internal/database/migrations/001_initial_schema.up.sql
@@ -0,0 +1,127 @@
+-- Enable UUID extension
+CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
+
+-- Agents table
+CREATE TABLE agents (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    hostname VARCHAR(255) NOT NULL,
+    os_type VARCHAR(50) NOT NULL CHECK (os_type IN ('windows', 'linux', 'macos')),
+    os_version VARCHAR(100),
+    os_architecture VARCHAR(20),
+    agent_version VARCHAR(20) NOT NULL,
+    last_seen TIMESTAMP NOT NULL DEFAULT NOW(),
+    status VARCHAR(20) DEFAULT 'online' CHECK (status IN ('online', 'offline', 'error')),
+    metadata JSONB,
+    created_at TIMESTAMP DEFAULT NOW(),
+    updated_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX idx_agents_status ON agents(status);
+CREATE INDEX idx_agents_os_type ON agents(os_type);
+CREATE INDEX idx_agents_last_seen ON agents(last_seen);
+
+-- Agent specs
+CREATE TABLE agent_specs (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    agent_id UUID REFERENCES agents(id) ON DELETE CASCADE,
+    cpu_model VARCHAR(255),
+    cpu_cores INTEGER,
+    memory_total_mb INTEGER,
+    disk_total_gb INTEGER,
+    disk_free_gb INTEGER,
+    network_interfaces JSONB,
+    docker_installed BOOLEAN DEFAULT false,
+    docker_version VARCHAR(50),
+    package_managers TEXT[],
+    collected_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX idx_agent_specs_agent_id ON agent_specs(agent_id);
+
+-- Update packages
+CREATE TABLE update_packages (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    agent_id UUID REFERENCES agents(id) ON DELETE CASCADE,
+    package_type VARCHAR(50) NOT NULL,
+    package_name VARCHAR(500) NOT NULL,
+    package_description TEXT,
+    current_version VARCHAR(100),
+    available_version VARCHAR(100) NOT NULL,
+    severity VARCHAR(20) CHECK (severity IN ('critical', 'important', 'moderate', 'low', 'none')),
+    cve_list TEXT[],
+    kb_id VARCHAR(50),
+    repository_source VARCHAR(255),
+    size_bytes BIGINT,
+    status VARCHAR(30) DEFAULT 'pending' CHECK (status IN ('pending', 'approved', 'scheduled', 'installing', 'installed', 'failed', 'ignored')),
+    discovered_at TIMESTAMP DEFAULT NOW(),
+    approved_by VARCHAR(255),
+    approved_at TIMESTAMP,
+    scheduled_for TIMESTAMP,
+    installed_at TIMESTAMP,
+    error_message TEXT,
+    metadata JSONB,
+    UNIQUE(agent_id, package_type, package_name, available_version)
+);
+
+CREATE INDEX idx_updates_status ON update_packages(status);
+CREATE INDEX idx_updates_agent ON update_packages(agent_id);
+CREATE INDEX idx_updates_severity ON update_packages(severity);
+CREATE INDEX idx_updates_package_type ON update_packages(package_type);
+CREATE INDEX idx_updates_composite ON update_packages(status, severity, agent_id);
+
+-- Update logs
+CREATE TABLE update_logs (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    agent_id UUID REFERENCES agents(id) ON DELETE CASCADE,
+    update_package_id UUID REFERENCES update_packages(id) ON DELETE SET NULL,
+    action VARCHAR(50) NOT NULL,
+    result VARCHAR(20) NOT NULL CHECK (result IN ('success', 'failed', 'partial')),
+    stdout TEXT,
+    stderr TEXT,
+    exit_code INTEGER,
+    duration_seconds INTEGER,
+    executed_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX idx_logs_agent ON update_logs(agent_id);
+CREATE INDEX idx_logs_result ON update_logs(result);
+CREATE INDEX idx_logs_executed_at ON update_logs(executed_at DESC);
+
+-- Agent tags
+CREATE TABLE agent_tags (
+    agent_id UUID REFERENCES agents(id) ON DELETE CASCADE,
+    tag VARCHAR(100) NOT NULL,
+    PRIMARY KEY (agent_id, tag)
+);
+
+CREATE INDEX idx_agent_tags_tag ON agent_tags(tag);
+
+-- Users (for authentication)
+CREATE TABLE users (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    username VARCHAR(255) UNIQUE NOT NULL,
+    email VARCHAR(255) UNIQUE NOT NULL,
+    password_hash VARCHAR(255) NOT NULL,
+    role VARCHAR(50) DEFAULT 'user' CHECK (role IN ('admin', 'user', 'readonly')),
+    created_at TIMESTAMP DEFAULT NOW(),
+    last_login TIMESTAMP
+);
+
+CREATE INDEX idx_users_username ON users(username);
+CREATE INDEX idx_users_email ON users(email);
+
+-- Commands queue (for agent orchestration)
+CREATE TABLE agent_commands (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    agent_id UUID REFERENCES agents(id) ON DELETE CASCADE,
+    command_type VARCHAR(50) NOT NULL,
+    params JSONB,
+    status VARCHAR(20) DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'completed', 'failed')),
+    created_at TIMESTAMP DEFAULT NOW(),
+    sent_at TIMESTAMP,
+    completed_at TIMESTAMP,
+    result JSONB
+);
+
+CREATE INDEX idx_commands_agent_status ON agent_commands(agent_id, status);
+CREATE INDEX idx_commands_created_at ON agent_commands(created_at DESC);
--- a/aggregator-server/internal/database/queries/agents.go
+++ b/aggregator-server/internal/database/queries/agents.go
@@ -0,0 +1,83 @@
+package queries
+
+import (
+	"time"
+
+	"github.com/aggregator-project/aggregator-server/internal/models"
+	"github.com/google/uuid"
+	"github.com/jmoiron/sqlx"
+)
+
+type AgentQueries struct {
+	db *sqlx.DB
+}
+
+func NewAgentQueries(db *sqlx.DB) *AgentQueries {
+	return &AgentQueries{db: db}
+}
+
+// CreateAgent inserts a new agent into the database
+func (q *AgentQueries) CreateAgent(agent *models.Agent) error {
+	query := `
+		INSERT INTO agents (
+			id, hostname, os_type, os_version, os_architecture,
+			agent_version, last_seen, status, metadata
+		) VALUES (
+			:id, :hostname, :os_type, :os_version, :os_architecture,
+			:agent_version, :last_seen, :status, :metadata
+		)
+	`
+	_, err := q.db.NamedExec(query, agent)
+	return err
+}
+
+// GetAgentByID retrieves an agent by ID
+func (q *AgentQueries) GetAgentByID(id uuid.UUID) (*models.Agent, error) {
+	var agent models.Agent
+	query := `SELECT * FROM agents WHERE id = $1`
+	err := q.db.Get(&agent, query, id)
+	if err != nil {
+		return nil, err
+	}
+	return &agent, nil
+}
+
+// UpdateAgentLastSeen updates the agent's last_seen timestamp
+func (q *AgentQueries) UpdateAgentLastSeen(id uuid.UUID) error {
+	query := `UPDATE agents SET last_seen = $1, status = 'online' WHERE id = $2`
+	_, err := q.db.Exec(query, time.Now(), id)
+	return err
+}
+
+// ListAgents returns all agents with optional filtering
+func (q *AgentQueries) ListAgents(status, osType string) ([]models.Agent, error) {
+	var agents []models.Agent
+	query := `SELECT * FROM agents WHERE 1=1`
+	args := []interface{}{}
+	argIdx := 1
+
+	if status != "" {
+		query += ` AND status = $` + string(rune(argIdx+'0'))
+		args = append(args, status)
+		argIdx++
+	}
+	if osType != "" {
+		query += ` AND os_type = $` + string(rune(argIdx+'0'))
+		args = append(args, osType)
+	}
+
+	query += ` ORDER BY last_seen DESC`
+	err := q.db.Select(&agents, query, args...)
+	return agents, err
+}
+
+// MarkOfflineAgents marks agents as offline if they haven't checked in recently
+func (q *AgentQueries) MarkOfflineAgents(threshold time.Duration) error {
+	query := `
+		UPDATE agents
+		SET status = 'offline'
+		WHERE last_seen < $1 AND status = 'online'
+	`
+	_, err := q.db.Exec(query, time.Now().Add(-threshold))
+	return err
+}
--- a/aggregator-server/internal/database/queries/commands.go
+++ b/aggregator-server/internal/database/queries/commands.go
@@ -0,0 +1,79 @@
+package queries
+
+import (
+	"time"
+
+	"github.com/aggregator-project/aggregator-server/internal/models"
+	"github.com/google/uuid"
+	"github.com/jmoiron/sqlx"
+)
+
+type CommandQueries struct {
+	db *sqlx.DB
+}
+
+func NewCommandQueries(db *sqlx.DB) *CommandQueries {
+	return &CommandQueries{db: db}
+}
+
+// CreateCommand inserts a new command for an agent
+func (q *CommandQueries) CreateCommand(cmd *models.AgentCommand) error {
+	query := `
+		INSERT INTO agent_commands (
+			id, agent_id, command_type, params, status
+		) VALUES (
+			:id, :agent_id, :command_type, :params, :status
+		)
+	`
+	_, err := q.db.NamedExec(query, cmd)
+	return err
+}
+
+// GetPendingCommands retrieves pending commands for an agent
+func (q *CommandQueries) GetPendingCommands(agentID uuid.UUID) ([]models.AgentCommand, error) {
+	var commands []models.AgentCommand
+	query := `
+		SELECT * FROM agent_commands
+		WHERE agent_id = $1 AND status = 'pending'
+		ORDER BY created_at ASC
+		LIMIT 10
+	`
+	err := q.db.Select(&commands, query, agentID)
+	return commands, err
+}
+
+// MarkCommandSent updates a command's status to sent
+func (q *CommandQueries) MarkCommandSent(id uuid.UUID) error {
+	now := time.Now()
+	query := `
+		UPDATE agent_commands
+		SET status = 'sent', sent_at = $1
+		WHERE id = $2
+	`
+	_, err := q.db.Exec(query, now, id)
+	return err
+}
+
+// MarkCommandCompleted updates a command's status to completed
+func (q *CommandQueries) MarkCommandCompleted(id uuid.UUID, result models.JSONB) error {
+	now := time.Now()
+	query := `
+		UPDATE agent_commands
+		SET status = 'completed', completed_at = $1, result = $2
+		WHERE id = $3
+	`
+	_, err := q.db.Exec(query, now, result, id)
+	return err
+}
+
+// MarkCommandFailed updates a command's status to failed
+func (q *CommandQueries) MarkCommandFailed(id uuid.UUID, result models.JSONB) error {
+	now := time.Now()
+	query := `
+		UPDATE agent_commands
+		SET status = 'failed', completed_at = $1, result = $2
+		WHERE id = $3
+	`
+	_, err := q.db.Exec(query, now, result, id)
+	return err
+}
--- a/aggregator-server/internal/database/queries/updates.go
+++ b/aggregator-server/internal/database/queries/updates.go
@@ -0,0 +1,141 @@
+package queries
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/aggregator-project/aggregator-server/internal/models"
+	"github.com/google/uuid"
+	"github.com/jmoiron/sqlx"
+)
+
+type UpdateQueries struct {
+	db *sqlx.DB
+}
+
+func NewUpdateQueries(db *sqlx.DB) *UpdateQueries {
+	return &UpdateQueries{db: db}
+}
+
+// UpsertUpdate inserts or updates an update package
+func (q *UpdateQueries) UpsertUpdate(update *models.UpdatePackage) error {
+	query := `
+		INSERT INTO update_packages (
+			id, agent_id, package_type, package_name, package_description,
+			current_version, available_version, severity, cve_list, kb_id,
+			repository_source, size_bytes, status, metadata
+		) VALUES (
+			:id, :agent_id, :package_type, :package_name, :package_description,
+			:current_version, :available_version, :severity, :cve_list, :kb_id,
+			:repository_source, :size_bytes, :status, :metadata
+		)
+		ON CONFLICT (agent_id, package_type, package_name, available_version)
+		DO UPDATE SET
+			package_description = EXCLUDED.package_description,
+			current_version = EXCLUDED.current_version,
+			severity = EXCLUDED.severity,
+			cve_list = EXCLUDED.cve_list,
+			kb_id = EXCLUDED.kb_id,
+			repository_source = EXCLUDED.repository_source,
+			size_bytes = EXCLUDED.size_bytes,
+			metadata = EXCLUDED.metadata,
+			discovered_at = NOW()
+	`
+	_, err := q.db.NamedExec(query, update)
+	return err
+}
+
+// ListUpdates retrieves updates with filtering
+func (q *UpdateQueries) ListUpdates(filters *models.UpdateFilters) ([]models.UpdatePackage, int, error) {
+	var updates []models.UpdatePackage
+	whereClause := []string{"1=1"}
+	args := []interface{}{}
+	argIdx := 1
+
+	if filters.AgentID != nil {
+		whereClause = append(whereClause, fmt.Sprintf("agent_id = $%d", argIdx))
+		args = append(args, *filters.AgentID)
+		argIdx++
+	}
+	if filters.Status != "" {
+		whereClause = append(whereClause, fmt.Sprintf("status = $%d", argIdx))
+		args = append(args, filters.Status)
+		argIdx++
+	}
+	if filters.Severity != "" {
+		whereClause = append(whereClause, fmt.Sprintf("severity = $%d", argIdx))
+		args = append(args, filters.Severity)
+		argIdx++
+	}
+	if filters.PackageType != "" {
+		whereClause = append(whereClause, fmt.Sprintf("package_type = $%d", argIdx))
+		args = append(args, filters.PackageType)
+		argIdx++
+	}
+
+	// Get total count
+	countQuery := "SELECT COUNT(*) FROM update_packages WHERE " + strings.Join(whereClause, " AND ")
+	var total int
+	err := q.db.Get(&total, countQuery, args...)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	// Get paginated results
+	query := fmt.Sprintf(`
+		SELECT * FROM update_packages
+		WHERE %s
+		ORDER BY discovered_at DESC
+		LIMIT $%d OFFSET $%d
+	`, strings.Join(whereClause, " AND "), argIdx, argIdx+1)
+
+	limit := filters.PageSize
+	if limit == 0 {
+		limit = 50
+	}
+	offset := (filters.Page - 1) * limit
+	if offset < 0 {
+		offset = 0
+	}
+
+	args = append(args, limit, offset)
+	err = q.db.Select(&updates, query, args...)
+	return updates, total, err
+}
+
+// GetUpdateByID retrieves a single update by ID
+func (q *UpdateQueries) GetUpdateByID(id uuid.UUID) (*models.UpdatePackage, error) {
+	var update models.UpdatePackage
+	query := `SELECT * FROM update_packages WHERE id = $1`
+	err := q.db.Get(&update, query, id)
+	if err != nil {
+		return nil, err
+	}
+	return &update, nil
+}
+
+// ApproveUpdate marks an update as approved
+func (q *UpdateQueries) ApproveUpdate(id uuid.UUID, approvedBy string) error {
+	query := `
+		UPDATE update_packages
+		SET status = 'approved', approved_by = $1, approved_at = NOW()
+		WHERE id = $2 AND status = 'pending'
+	`
+	_, err := q.db.Exec(query, approvedBy, id)
+	return err
+}
+
+// CreateUpdateLog inserts an update log entry
+func (q *UpdateQueries) CreateUpdateLog(log *models.UpdateLog) error {
+	query := `
+		INSERT INTO update_logs (
+			id, agent_id, update_package_id, action, result,
+			stdout, stderr, exit_code, duration_seconds
+		) VALUES (
+			:id, :agent_id, :update_package_id, :action, :result,
+			:stdout, :stderr, :exit_code, :duration_seconds
+		)
+	`
+	_, err := q.db.NamedExec(query, log)
+	return err
+}