ec30be7246
- DB Migration: api_keys bekommt 'permissions' + 'customer_id' Spalten - Permissions: agent | read | write | admin - agent: nur /api/v1/agent/ws + /api/v1/agent/heartbeat - read: GET-Endpoints (optional customer-scoped) - write: alles ausser Terminal - admin: voll (Terminal bleibt JWT-only) - Bestehende Keys werden auf 'admin' migriert (Abwaertskompatibilitaet) - Frontend: API-Key Verwaltung zeigt Permissions-Badge + Kunden-Scope - API-Client: createAPIKey nimmt jetzt permissions + customer_id
1384 lines
41 KiB
Go
1384 lines
41 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/cynfo/rmm-backend/models"
|
|
|
|
_ "github.com/jackc/pgx/v5/stdlib"
|
|
)
|
|
|
|
type ifaceSnapshot struct {
|
|
rxBytes float64
|
|
txBytes float64
|
|
ts time.Time
|
|
}
|
|
|
|
type Database struct {
|
|
db *sql.DB
|
|
lastIfaceSnap map[string]map[string]ifaceSnapshot // agentID -> ifaceName -> snapshot
|
|
}
|
|
|
|
type DBConfig struct {
|
|
Host string `yaml:"host"`
|
|
Port int `yaml:"port"`
|
|
User string `yaml:"user"`
|
|
Password string `yaml:"password"`
|
|
DBName string `yaml:"dbname"`
|
|
SSLMode string `yaml:"sslmode"`
|
|
}
|
|
|
|
func New(cfg DBConfig) (*Database, error) {
|
|
dsn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=%s",
|
|
cfg.Host, cfg.Port, cfg.User, cfg.Password, cfg.DBName, cfg.SSLMode)
|
|
|
|
db, err := sql.Open("pgx", dsn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Datenbank oeffnen: %w", err)
|
|
}
|
|
|
|
// Connection Pool konfigurieren
|
|
db.SetMaxOpenConns(25)
|
|
db.SetMaxIdleConns(5)
|
|
db.SetConnMaxLifetime(5 * time.Minute)
|
|
|
|
// Verbindung testen
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
if err := db.PingContext(ctx); err != nil {
|
|
return nil, fmt.Errorf("Datenbank-Verbindung fehlgeschlagen: %w", err)
|
|
}
|
|
|
|
d := &Database{db: db, lastIfaceSnap: make(map[string]map[string]ifaceSnapshot)}
|
|
if err := d.migrate(); err != nil {
|
|
return nil, fmt.Errorf("Migration fehlgeschlagen: %w", err)
|
|
}
|
|
|
|
log.Printf("PostgreSQL verbunden: %s@%s:%d/%s", cfg.User, cfg.Host, cfg.Port, cfg.DBName)
|
|
return d, nil
|
|
}
|
|
|
|
func (d *Database) migrate() error {
|
|
// Relationale Tabellen
|
|
schema := `
|
|
CREATE TABLE IF NOT EXISTS agents (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
hostname TEXT NOT NULL,
|
|
ip TEXT NOT NULL,
|
|
opnsense_version TEXT NOT NULL DEFAULT '',
|
|
agent_version TEXT NOT NULL DEFAULT '',
|
|
registered_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
last_heartbeat TIMESTAMPTZ
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS system_data (
|
|
agent_id TEXT PRIMARY KEY REFERENCES agents(id) ON DELETE CASCADE,
|
|
data_json JSONB NOT NULL,
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS config_backups (
|
|
id SERIAL PRIMARY KEY,
|
|
agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
|
|
hash TEXT NOT NULL,
|
|
size INTEGER NOT NULL,
|
|
config_xml TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_config_backups_agent ON config_backups(agent_id, created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_config_backups_hash ON config_backups(agent_id, hash);
|
|
|
|
CREATE TABLE IF NOT EXISTS agent_events (
|
|
id SERIAL PRIMARY KEY,
|
|
agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
|
|
event_type TEXT NOT NULL,
|
|
message TEXT,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_agent_events_agent ON agent_events(agent_id, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS customers (
|
|
id SERIAL PRIMARY KEY,
|
|
number TEXT NOT NULL UNIQUE,
|
|
name TEXT NOT NULL,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id SERIAL PRIMARY KEY,
|
|
username TEXT NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
display_name TEXT NOT NULL DEFAULT '',
|
|
role TEXT NOT NULL DEFAULT 'admin',
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
last_login TIMESTAMPTZ
|
|
);
|
|
`
|
|
|
|
if _, err := d.db.Exec(schema); err != nil {
|
|
return fmt.Errorf("Schema anlegen: %w", err)
|
|
}
|
|
|
|
// TOTP 2FA Spalten
|
|
d.db.Exec("ALTER TABLE users ADD COLUMN IF NOT EXISTS totp_secret TEXT")
|
|
d.db.Exec("ALTER TABLE users ADD COLUMN IF NOT EXISTS totp_enabled BOOLEAN NOT NULL DEFAULT false")
|
|
|
|
// Agents um customer_id erweitern
|
|
d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS customer_id INTEGER REFERENCES customers(id)")
|
|
d.db.Exec("CREATE INDEX IF NOT EXISTS idx_agents_customer ON agents(customer_id)")
|
|
|
|
// Agents um agent_version und update_requested erweitern
|
|
d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS agent_version TEXT NOT NULL DEFAULT ''")
|
|
d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS update_requested BOOLEAN NOT NULL DEFAULT FALSE")
|
|
d.db.Exec("ALTER TABLE agents ADD COLUMN IF NOT EXISTS platform TEXT NOT NULL DEFAULT 'freebsd'")
|
|
|
|
// Agent-Firmware Tabelle (Multi-Plattform)
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS agent_firmware (
|
|
version TEXT PRIMARY KEY,
|
|
hash TEXT NOT NULL,
|
|
binary_data BYTEA NOT NULL,
|
|
uploaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
)`)
|
|
// Platform-Spalte hinzufuegen (default freebsd fuer bestehende Eintraege)
|
|
d.db.Exec("ALTER TABLE agent_firmware DROP CONSTRAINT IF EXISTS agent_firmware_pkey")
|
|
d.db.Exec("ALTER TABLE agent_firmware ADD COLUMN IF NOT EXISTS platform TEXT NOT NULL DEFAULT 'freebsd'")
|
|
// Neuer Primary Key: version + platform
|
|
d.db.Exec(`DO $$ BEGIN
|
|
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'agent_firmware_version_platform_key') THEN
|
|
ALTER TABLE agent_firmware ADD CONSTRAINT agent_firmware_version_platform_key UNIQUE (version, platform);
|
|
END IF;
|
|
END $$`)
|
|
|
|
// Metrics Hypertable
|
|
metricsSchema := `
|
|
CREATE TABLE IF NOT EXISTS metrics (
|
|
time TIMESTAMPTZ NOT NULL,
|
|
agent_id TEXT NOT NULL,
|
|
metric TEXT NOT NULL,
|
|
value DOUBLE PRECISION NOT NULL,
|
|
tags JSONB
|
|
);
|
|
`
|
|
if _, err := d.db.Exec(metricsSchema); err != nil {
|
|
return fmt.Errorf("Metrics-Tabelle anlegen: %w", err)
|
|
}
|
|
|
|
// In Hypertable umwandeln (ignoriert wenn schon Hypertable)
|
|
var isHypertable bool
|
|
err := d.db.QueryRow(`
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM timescaledb_information.hypertables
|
|
WHERE hypertable_name = 'metrics'
|
|
)
|
|
`).Scan(&isHypertable)
|
|
|
|
if err != nil || !isHypertable {
|
|
_, err = d.db.Exec(`SELECT create_hypertable('metrics', 'time', if_not_exists => TRUE)`)
|
|
if err != nil {
|
|
log.Printf("Hypertable-Warnung (ggf. schon vorhanden): %v", err)
|
|
} else {
|
|
log.Println("Metrics Hypertable erstellt")
|
|
}
|
|
}
|
|
|
|
// Indices fuer Metrics
|
|
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_metrics_agent_metric ON metrics(agent_id, metric, time DESC)`)
|
|
|
|
// API-Keys Tabelle
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS api_keys (
|
|
id SERIAL PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
key TEXT NOT NULL UNIQUE,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
last_used TIMESTAMPTZ
|
|
)`)
|
|
// Migration: customer_id + permissions Spalten hinzufuegen falls nicht vorhanden
|
|
d.db.Exec(`ALTER TABLE api_keys ADD COLUMN IF NOT EXISTS customer_id INTEGER REFERENCES customers(id) ON DELETE SET NULL`)
|
|
d.db.Exec(`ALTER TABLE api_keys ADD COLUMN IF NOT EXISTS permissions TEXT NOT NULL DEFAULT 'admin'`)
|
|
// Bestehende Keys auf 'admin' setzen (Abwaertskompatibilitaet)
|
|
d.db.Exec(`UPDATE api_keys SET permissions = 'admin' WHERE permissions = ''`)
|
|
|
|
// Installer-Pakete (install.sh + Plugin als ZIP pro Plattform)
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS installers (
|
|
id SERIAL PRIMARY KEY,
|
|
platform TEXT NOT NULL UNIQUE,
|
|
filename TEXT NOT NULL,
|
|
data BYTEA NOT NULL,
|
|
sha256 TEXT NOT NULL,
|
|
size INTEGER NOT NULL,
|
|
uploaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
)`)
|
|
|
|
// Audit-Log Tabelle
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS audit_log (
|
|
id SERIAL PRIMARY KEY,
|
|
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
user_name TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
target_type TEXT,
|
|
target_id TEXT,
|
|
target_name TEXT,
|
|
details TEXT,
|
|
source_ip TEXT
|
|
)`)
|
|
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_audit_log_timestamp ON audit_log(timestamp DESC)`)
|
|
|
|
// System Settings Tabelle
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS system_settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL,
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
)`)
|
|
// Default-Keys anlegen (leere Werte)
|
|
for _, k := range []string{"backend_url_internal", "backend_url_external", "api_key"} {
|
|
d.db.Exec(`INSERT INTO system_settings (key, value) VALUES ($1, '') ON CONFLICT DO NOTHING`, k)
|
|
}
|
|
|
|
// Tasks Tabelle
|
|
d.db.Exec(`CREATE TABLE IF NOT EXISTS tasks (
|
|
id SERIAL PRIMARY KEY,
|
|
agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
|
|
action TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'scheduled',
|
|
scheduled_at TIMESTAMPTZ NOT NULL,
|
|
started_at TIMESTAMPTZ,
|
|
finished_at TIMESTAMPTZ,
|
|
result JSONB,
|
|
webhook_url TEXT,
|
|
webhook_sent BOOLEAN DEFAULT FALSE,
|
|
webhook_response TEXT,
|
|
created_by TEXT,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
|
)`)
|
|
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_tasks_status_scheduled ON tasks(status, scheduled_at)`)
|
|
|
|
// Tasks: recurring jobs columns
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS name TEXT DEFAULT ''`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_type TEXT DEFAULT 'once'`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_time TEXT DEFAULT '02:00'`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_weekday INT`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_monthday INT`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS reboot BOOLEAN DEFAULT FALSE`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS health_check BOOLEAN DEFAULT FALSE`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS health_check_timeout INT DEFAULT 600`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS security_only BOOLEAN DEFAULT FALSE`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active BOOLEAN DEFAULT TRUE`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS last_run TIMESTAMPTZ`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS next_run TIMESTAMPTZ`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_count INT DEFAULT 0`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS script TEXT`)
|
|
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS script_timeout INT DEFAULT 300`)
|
|
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_tasks_active_next_run ON tasks(active, next_run) WHERE active = TRUE`)
|
|
|
|
// Retention Policy (90 Tage)
|
|
d.setupRetention()
|
|
|
|
// Compression Policy
|
|
d.setupCompression()
|
|
|
|
return nil
|
|
}
|
|
|
|
func (d *Database) setupRetention() {
|
|
// Pruefen ob schon eine Policy existiert
|
|
var count int
|
|
d.db.QueryRow(`
|
|
SELECT COUNT(*) FROM timescaledb_information.jobs
|
|
WHERE hypertable_name = 'metrics' AND proc_name = 'policy_retention'
|
|
`).Scan(&count)
|
|
|
|
if count == 0 {
|
|
_, err := d.db.Exec(`SELECT add_retention_policy('metrics', INTERVAL '90 days')`)
|
|
if err != nil {
|
|
log.Printf("Retention Policy Warnung: %v", err)
|
|
} else {
|
|
log.Println("Retention Policy: 90 Tage fuer metrics")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (d *Database) setupCompression() {
|
|
// Compression aktivieren
|
|
d.db.Exec(`
|
|
ALTER TABLE metrics SET (
|
|
timescaledb.compress,
|
|
timescaledb.compress_segmentby = 'agent_id, metric',
|
|
timescaledb.compress_orderby = 'time DESC'
|
|
)
|
|
`)
|
|
|
|
var count int
|
|
d.db.QueryRow(`
|
|
SELECT COUNT(*) FROM timescaledb_information.jobs
|
|
WHERE hypertable_name = 'metrics' AND proc_name = 'policy_compression'
|
|
`).Scan(&count)
|
|
|
|
if count == 0 {
|
|
_, err := d.db.Exec(`SELECT add_compression_policy('metrics', INTERVAL '7 days')`)
|
|
if err != nil {
|
|
log.Printf("Compression Policy Warnung: %v", err)
|
|
} else {
|
|
log.Println("Compression Policy: nach 7 Tagen fuer metrics")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (d *Database) Close() error {
|
|
return d.db.Close()
|
|
}
|
|
|
|
// --- Agent Methoden ---
|
|
|
|
func (d *Database) RegisterAgent(agent *models.Agent) error {
|
|
_, err := d.db.Exec(`
|
|
INSERT INTO agents (id, name, hostname, ip, opnsense_version, agent_version, platform, registered_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name=EXCLUDED.name,
|
|
hostname=EXCLUDED.hostname,
|
|
ip=EXCLUDED.ip,
|
|
opnsense_version=EXCLUDED.opnsense_version,
|
|
agent_version=EXCLUDED.agent_version,
|
|
platform=EXCLUDED.platform
|
|
`, agent.ID, agent.Name, agent.Hostname, agent.IP, agent.OPNsenseVersion, agent.AgentVersion, agent.Platform, agent.RegisteredAt.UTC())
|
|
return err
|
|
}
|
|
|
|
func (d *Database) SaveSystemData(agentID string, data *models.SystemData) error {
|
|
now := time.Now().UTC()
|
|
|
|
jsonData, err := json.Marshal(data)
|
|
if err != nil {
|
|
return fmt.Errorf("JSON Marshal: %w", err)
|
|
}
|
|
|
|
tx, err := d.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
// Heartbeat aktualisieren
|
|
res, err := tx.Exec("UPDATE agents SET last_heartbeat = $1 WHERE id = $2", now, agentID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, _ := res.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("Agent %s nicht gefunden", agentID)
|
|
}
|
|
|
|
// Systemdaten upsert
|
|
_, err = tx.Exec(`
|
|
INSERT INTO system_data (agent_id, data_json, updated_at)
|
|
VALUES ($1, $2, $3)
|
|
ON CONFLICT(agent_id) DO UPDATE SET
|
|
data_json=EXCLUDED.data_json,
|
|
updated_at=EXCLUDED.updated_at
|
|
`, agentID, string(jsonData), now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Metriken extrahieren und schreiben
|
|
if err := d.writeMetrics(tx, agentID, now, data); err != nil {
|
|
log.Printf("Metriken schreiben Warnung: %v", err)
|
|
// Kein Abbruch — Systemdaten sind wichtiger
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// writeMetrics extrahiert Metriken aus den Systemdaten und schreibt sie in die Hypertable
|
|
func (d *Database) writeMetrics(tx *sql.Tx, agentID string, ts time.Time, data *models.SystemData) error {
|
|
stmt, err := tx.Prepare(`INSERT INTO metrics (time, agent_id, metric, value, tags) VALUES ($1, $2, $3, $4, $5)`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
// CPU
|
|
if data.CPU.UsagePercent > 0 {
|
|
stmt.Exec(ts, agentID, "cpu_usage", data.CPU.UsagePercent, nil)
|
|
}
|
|
|
|
// Memory
|
|
if data.Memory.TotalBytes > 0 {
|
|
stmt.Exec(ts, agentID, "memory_total_bytes", float64(data.Memory.TotalBytes), nil)
|
|
stmt.Exec(ts, agentID, "memory_used_bytes", float64(data.Memory.UsedBytes), nil)
|
|
usedPercent := float64(data.Memory.UsedBytes) / float64(data.Memory.TotalBytes) * 100
|
|
stmt.Exec(ts, agentID, "memory_used_percent", usedPercent, nil)
|
|
}
|
|
|
|
// Uptime
|
|
if data.UptimeSeconds > 0 {
|
|
stmt.Exec(ts, agentID, "uptime_seconds", float64(data.UptimeSeconds), nil)
|
|
}
|
|
|
|
// Disks
|
|
for _, disk := range data.Disks {
|
|
if disk.TotalBytes > 0 {
|
|
usedPercent := float64(disk.UsedBytes) / float64(disk.TotalBytes) * 100
|
|
tags, _ := json.Marshal(map[string]string{"mountpoint": disk.MountPoint, "dataset": disk.Filesystem})
|
|
stmt.Exec(ts, agentID, "disk_used_percent", usedPercent, string(tags))
|
|
stmt.Exec(ts, agentID, "disk_used_bytes", float64(disk.UsedBytes), string(tags))
|
|
}
|
|
}
|
|
|
|
// Network Interfaces — kumulierte Bytes + abgeleitete Rate (bytes/s)
|
|
if _, ok := d.lastIfaceSnap[agentID]; !ok {
|
|
d.lastIfaceSnap[agentID] = make(map[string]ifaceSnapshot)
|
|
}
|
|
for _, iface := range data.NetworkInterfaces {
|
|
rx := float64(iface.RxBytes)
|
|
tx := float64(iface.TxBytes)
|
|
tags, _ := json.Marshal(map[string]string{"interface": iface.Name})
|
|
tagsStr := string(tags)
|
|
if rx > 0 || tx > 0 {
|
|
stmt.Exec(ts, agentID, "interface_rx_bytes", rx, tagsStr)
|
|
stmt.Exec(ts, agentID, "interface_tx_bytes", tx, tagsStr)
|
|
}
|
|
// Rate berechnen wenn vorheriger Snapshot vorhanden
|
|
if prev, ok := d.lastIfaceSnap[agentID][iface.Name]; ok {
|
|
elapsed := ts.Sub(prev.ts).Seconds()
|
|
if elapsed > 0 && elapsed < 300 { // max 5 min Lücke ignorieren
|
|
rxRate := (rx - prev.rxBytes) / elapsed
|
|
txRate := (tx - prev.txBytes) / elapsed
|
|
if rxRate >= 0 && txRate >= 0 { // negative Werte ignorieren (Reboot)
|
|
stmt.Exec(ts, agentID, "interface_rx_bps", rxRate, tagsStr)
|
|
stmt.Exec(ts, agentID, "interface_tx_bps", txRate, tagsStr)
|
|
}
|
|
}
|
|
}
|
|
d.lastIfaceSnap[agentID][iface.Name] = ifaceSnapshot{rxBytes: rx, txBytes: tx, ts: ts}
|
|
}
|
|
|
|
// PF States
|
|
if data.PFStates != nil && data.PFStates.HardLimit > 0 {
|
|
stmt.Exec(ts, agentID, "pf_states_current", float64(data.PFStates.CurrentEntries), nil)
|
|
stmt.Exec(ts, agentID, "pf_states_limit", float64(data.PFStates.HardLimit), nil)
|
|
usedPct := float64(data.PFStates.CurrentEntries) / float64(data.PFStates.HardLimit) * 100
|
|
stmt.Exec(ts, agentID, "pf_states_used_percent", usedPct, nil)
|
|
}
|
|
|
|
// Gateways (Loss und Delay sind Strings wie "0.0%" und "1.2ms")
|
|
for _, gw := range data.Gateways {
|
|
if gw.Delay != "" {
|
|
tags, _ := json.Marshal(map[string]string{"gateway": gw.Name})
|
|
// Delay parsen ("1.2ms" -> 1.2)
|
|
delayStr := strings.TrimSuffix(strings.TrimSpace(gw.Delay), "ms")
|
|
if delay, err := strconv.ParseFloat(delayStr, 64); err == nil {
|
|
stmt.Exec(ts, agentID, "gateway_rtt_ms", delay, string(tags))
|
|
}
|
|
// Loss parsen ("0.0%" -> 0.0)
|
|
lossStr := strings.TrimSuffix(strings.TrimSpace(gw.Loss), "%")
|
|
if loss, err := strconv.ParseFloat(lossStr, 64); err == nil {
|
|
stmt.Exec(ts, agentID, "gateway_loss_percent", loss, string(tags))
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (d *Database) GetAgents() ([]models.Agent, error) {
|
|
rows, err := d.db.Query("SELECT id, name, hostname, ip, opnsense_version, agent_version, platform, registered_at, last_heartbeat, customer_id, update_requested FROM agents ORDER BY name")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var agents []models.Agent
|
|
for rows.Next() {
|
|
var a models.Agent
|
|
var lastHB sql.NullTime
|
|
var custID sql.NullInt64
|
|
|
|
if err := rows.Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.AgentVersion, &a.Platform, &a.RegisteredAt, &lastHB, &custID, &a.UpdateRequested); err != nil {
|
|
return nil, err
|
|
}
|
|
if lastHB.Valid {
|
|
a.LastHeartbeat = &lastHB.Time
|
|
}
|
|
if custID.Valid {
|
|
id := int(custID.Int64)
|
|
a.CustomerID = &id
|
|
}
|
|
agents = append(agents, a)
|
|
}
|
|
if agents == nil {
|
|
agents = []models.Agent{}
|
|
}
|
|
return agents, rows.Err()
|
|
}
|
|
|
|
func (d *Database) GetAgentByHostname(hostname string) (*models.Agent, error) {
|
|
var a models.Agent
|
|
var lastHB sql.NullTime
|
|
|
|
err := d.db.QueryRow(
|
|
"SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat FROM agents WHERE hostname = $1 LIMIT 1",
|
|
hostname,
|
|
).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB)
|
|
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if lastHB.Valid {
|
|
a.LastHeartbeat = &lastHB.Time
|
|
}
|
|
return &a, nil
|
|
}
|
|
|
|
func (d *Database) GetAgentByName(name string) (*models.Agent, error) {
|
|
var a models.Agent
|
|
var lastHB sql.NullTime
|
|
|
|
err := d.db.QueryRow(
|
|
"SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat FROM agents WHERE name = $1 LIMIT 1",
|
|
name,
|
|
).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB)
|
|
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if lastHB.Valid {
|
|
a.LastHeartbeat = &lastHB.Time
|
|
}
|
|
return &a, nil
|
|
}
|
|
|
|
func (d *Database) GetAgent(id string) (*models.Agent, *models.SystemData, error) {
|
|
var a models.Agent
|
|
var lastHB sql.NullTime
|
|
var custID sql.NullInt64
|
|
|
|
err := d.db.QueryRow(
|
|
"SELECT id, name, hostname, ip, opnsense_version, agent_version, platform, registered_at, last_heartbeat, customer_id, update_requested FROM agents WHERE id = $1",
|
|
id,
|
|
).Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.AgentVersion, &a.Platform, &a.RegisteredAt, &lastHB, &custID, &a.UpdateRequested)
|
|
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if lastHB.Valid {
|
|
a.LastHeartbeat = &lastHB.Time
|
|
}
|
|
if custID.Valid {
|
|
cid := int(custID.Int64)
|
|
a.CustomerID = &cid
|
|
}
|
|
|
|
// Systemdaten laden
|
|
var dataJSON sql.NullString
|
|
err = d.db.QueryRow("SELECT data_json FROM system_data WHERE agent_id = $1", id).Scan(&dataJSON)
|
|
if err != nil && err != sql.ErrNoRows {
|
|
return nil, nil, err
|
|
}
|
|
|
|
var sysData *models.SystemData
|
|
if dataJSON.Valid {
|
|
sysData = &models.SystemData{}
|
|
if err := json.Unmarshal([]byte(dataJSON.String), sysData); err != nil {
|
|
return &a, nil, nil
|
|
}
|
|
}
|
|
|
|
return &a, sysData, nil
|
|
}
|
|
|
|
func (d *Database) GetSystemData(agentID string) (*models.SystemData, error) {
|
|
var dataJSON string
|
|
err := d.db.QueryRow("SELECT data_json FROM system_data WHERE agent_id = $1", agentID).Scan(&dataJSON)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var data models.SystemData
|
|
if err := json.Unmarshal([]byte(dataJSON), &data); err != nil {
|
|
return nil, err
|
|
}
|
|
return &data, nil
|
|
}
|
|
|
|
func (d *Database) UpdateAgentVersion(id, version string) error {
|
|
_, err := d.db.Exec("UPDATE agents SET agent_version = $1 WHERE id = $2", version, id)
|
|
return err
|
|
}
|
|
|
|
// Update-Request Flag setzen/lesen/loeschen
|
|
func (d *Database) SetUpdateRequest(id string, requested bool) error {
|
|
_, err := d.db.Exec("UPDATE agents SET update_requested = $1 WHERE id = $2", requested, id)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) IsUpdateRequested(id string) bool {
|
|
var requested bool
|
|
err := d.db.QueryRow("SELECT update_requested FROM agents WHERE id = $1", id).Scan(&requested)
|
|
return err == nil && requested
|
|
}
|
|
|
|
func (d *Database) ClearUpdateRequest(id string) error {
|
|
return d.SetUpdateRequest(id, false)
|
|
}
|
|
|
|
// Alle Agents mit Update-Flag setzen
|
|
func (d *Database) SetUpdateRequestAll(requested bool) (int64, error) {
|
|
res, err := d.db.Exec("UPDATE agents SET update_requested = $1", requested)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return res.RowsAffected()
|
|
}
|
|
|
|
// Agent-Version direkt aus DB lesen (vor dem Update durch Heartbeat)
|
|
func (d *Database) GetAgentVersionFromDB(id string) string {
|
|
var version string
|
|
d.db.QueryRow("SELECT agent_version FROM agents WHERE id = $1", id).Scan(&version)
|
|
return version
|
|
}
|
|
|
|
// Agent-Firmware speichern (pro Plattform)
|
|
func (d *Database) SaveAgentFirmware(version, platform, hash string, binary []byte) error {
|
|
_, err := d.db.Exec(`
|
|
INSERT INTO agent_firmware (version, platform, hash, binary_data, uploaded_at)
|
|
VALUES ($1, $2, $3, $4, NOW())
|
|
ON CONFLICT(version, platform) DO UPDATE SET hash=EXCLUDED.hash, binary_data=EXCLUDED.binary_data, uploaded_at=NOW()
|
|
`, version, platform, hash, binary)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetLatestFirmware(platform string) (string, string, []byte, error) {
|
|
var version, hash string
|
|
var binary []byte
|
|
err := d.db.QueryRow("SELECT version, hash, binary_data FROM agent_firmware WHERE platform = $1 ORDER BY uploaded_at DESC LIMIT 1", platform).Scan(&version, &hash, &binary)
|
|
if err != nil {
|
|
return "", "", nil, err
|
|
}
|
|
return version, hash, binary, nil
|
|
}
|
|
|
|
func (d *Database) GetFirmwareInfo(platform string) (string, string, int, error) {
|
|
var version, hash string
|
|
var size int
|
|
err := d.db.QueryRow("SELECT version, hash, length(binary_data) FROM agent_firmware WHERE platform = $1 ORDER BY uploaded_at DESC LIMIT 1", platform).Scan(&version, &hash, &size)
|
|
if err != nil {
|
|
return "", "", 0, err
|
|
}
|
|
return version, hash, size, nil
|
|
}
|
|
|
|
// Alle Plattform-Firmwares auflisten
|
|
func (d *Database) GetAllFirmwareInfo() ([]map[string]interface{}, error) {
|
|
rows, err := d.db.Query(`
|
|
SELECT DISTINCT ON (platform) platform, version, hash, length(binary_data) as size, uploaded_at
|
|
FROM agent_firmware ORDER BY platform, uploaded_at DESC
|
|
`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var results []map[string]interface{}
|
|
for rows.Next() {
|
|
var platform, version, hash string
|
|
var size int
|
|
var uploadedAt time.Time
|
|
if err := rows.Scan(&platform, &version, &hash, &size, &uploadedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
results = append(results, map[string]interface{}{
|
|
"platform": platform,
|
|
"version": version,
|
|
"hash": hash,
|
|
"size": size,
|
|
"uploaded_at": uploadedAt,
|
|
})
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
// --- Installer ---
|
|
|
|
func (d *Database) SaveInstaller(platform, filename, sha256hash string, data []byte) error {
|
|
_, err := d.db.Exec(`
|
|
INSERT INTO installers (platform, filename, data, sha256, size, uploaded_at)
|
|
VALUES ($1, $2, $3, $4, $5, NOW())
|
|
ON CONFLICT (platform) DO UPDATE SET filename=$2, data=$3, sha256=$4, size=$5, uploaded_at=NOW()
|
|
`, platform, filename, data, sha256hash, len(data))
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetInstaller(platform string) (string, string, []byte, error) {
|
|
var filename, sha256hash string
|
|
var data []byte
|
|
err := d.db.QueryRow("SELECT filename, sha256, data FROM installers WHERE platform = $1", platform).Scan(&filename, &sha256hash, &data)
|
|
return filename, sha256hash, data, err
|
|
}
|
|
|
|
func (d *Database) GetInstallerInfo(platform string) (string, string, int, string, error) {
|
|
var filename, sha256hash string
|
|
var size int
|
|
var uploadedAt time.Time
|
|
err := d.db.QueryRow("SELECT filename, sha256, size, uploaded_at FROM installers WHERE platform = $1", platform).Scan(&filename, &sha256hash, &size, &uploadedAt)
|
|
return filename, sha256hash, size, uploadedAt.Format(time.RFC3339), err
|
|
}
|
|
|
|
func (d *Database) GetAllInstallerInfo() ([]map[string]interface{}, error) {
|
|
rows, err := d.db.Query("SELECT platform, filename, sha256, size, uploaded_at FROM installers ORDER BY platform")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var results []map[string]interface{}
|
|
for rows.Next() {
|
|
var platform, filename, sha256hash string
|
|
var size int
|
|
var uploadedAt time.Time
|
|
if err := rows.Scan(&platform, &filename, &sha256hash, &size, &uploadedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
results = append(results, map[string]interface{}{
|
|
"platform": platform,
|
|
"filename": filename,
|
|
"sha256": sha256hash,
|
|
"size": size,
|
|
"uploaded_at": uploadedAt,
|
|
})
|
|
}
|
|
return results, nil
|
|
}
|
|
|
|
func (d *Database) DeleteAgent(id string) (bool, error) {
|
|
// metrics hat keinen FK-Cascade → manuell löschen
|
|
d.db.Exec("DELETE FROM metrics WHERE agent_id = $1", id)
|
|
// agents löschen → CASCADE löscht system_data, tasks, config_backups, agent_events
|
|
res, err := d.db.Exec("DELETE FROM agents WHERE id = $1", id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := res.RowsAffected()
|
|
return rows > 0, nil
|
|
}
|
|
|
|
// --- Config Backup Methoden ---
|
|
|
|
func (d *Database) SaveConfigBackup(agentID, hash string, size int, configXML string) (int64, bool, error) {
|
|
var existingID int64
|
|
err := d.db.QueryRow("SELECT id FROM config_backups WHERE agent_id = $1 AND hash = $2 LIMIT 1", agentID, hash).Scan(&existingID)
|
|
if err == nil {
|
|
return existingID, false, nil
|
|
}
|
|
|
|
var id int64
|
|
err = d.db.QueryRow(
|
|
"INSERT INTO config_backups (agent_id, hash, size, config_xml) VALUES ($1, $2, $3, $4) RETURNING id",
|
|
agentID, hash, size, configXML,
|
|
).Scan(&id)
|
|
if err != nil {
|
|
return 0, false, fmt.Errorf("Backup speichern: %w", err)
|
|
}
|
|
return id, true, nil
|
|
}
|
|
|
|
func (d *Database) GetConfigBackups(agentID string, limit int) ([]map[string]interface{}, error) {
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
rows, err := d.db.Query(
|
|
"SELECT id, hash, size, created_at FROM config_backups WHERE agent_id = $1 ORDER BY created_at DESC LIMIT $2",
|
|
agentID, limit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var backups []map[string]interface{}
|
|
for rows.Next() {
|
|
var id int64
|
|
var hash string
|
|
var size int
|
|
var createdAt time.Time
|
|
if err := rows.Scan(&id, &hash, &size, &createdAt); err != nil {
|
|
return nil, err
|
|
}
|
|
backups = append(backups, map[string]interface{}{
|
|
"id": id,
|
|
"hash": hash,
|
|
"size": size,
|
|
"created_at": createdAt.Format(time.RFC3339),
|
|
})
|
|
}
|
|
if backups == nil {
|
|
backups = []map[string]interface{}{}
|
|
}
|
|
return backups, rows.Err()
|
|
}
|
|
|
|
func (d *Database) GetConfigBackup(agentID string, backupID int64) (map[string]interface{}, error) {
|
|
var hash, configXML string
|
|
var size int
|
|
var id int64
|
|
var createdAt time.Time
|
|
|
|
err := d.db.QueryRow(
|
|
"SELECT id, hash, size, config_xml, created_at FROM config_backups WHERE agent_id = $1 AND id = $2",
|
|
agentID, backupID,
|
|
).Scan(&id, &hash, &size, &configXML, &createdAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]interface{}{
|
|
"id": id,
|
|
"hash": hash,
|
|
"size": size,
|
|
"config_xml": configXML,
|
|
"created_at": createdAt.Format(time.RFC3339),
|
|
}, nil
|
|
}
|
|
|
|
func (d *Database) GetLatestConfigBackup(agentID string) (map[string]interface{}, error) {
|
|
var hash, configXML string
|
|
var size int
|
|
var id int64
|
|
var createdAt time.Time
|
|
|
|
err := d.db.QueryRow(
|
|
"SELECT id, hash, size, config_xml, created_at FROM config_backups WHERE agent_id = $1 ORDER BY created_at DESC LIMIT 1",
|
|
agentID,
|
|
).Scan(&id, &hash, &size, &configXML, &createdAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]interface{}{
|
|
"id": id,
|
|
"hash": hash,
|
|
"size": size,
|
|
"config_xml": configXML,
|
|
"created_at": createdAt.Format(time.RFC3339),
|
|
}, nil
|
|
}
|
|
|
|
func (d *Database) DeleteConfigBackup(agentID string, backupID int64) (bool, error) {
|
|
res, err := d.db.Exec("DELETE FROM config_backups WHERE agent_id = $1 AND id = $2", agentID, backupID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := res.RowsAffected()
|
|
return rows > 0, nil
|
|
}
|
|
|
|
// --- Metrics Abfragen ---
|
|
|
|
// GetMetrics holt Metriken fuer einen Agent in einem Zeitraum
|
|
func (d *Database) GetMetrics(agentID, metric, tagsFilter string, from, to time.Time, limit int) ([]map[string]interface{}, error) {
|
|
if limit <= 0 {
|
|
limit = 1000
|
|
}
|
|
|
|
query := `SELECT time, metric, value, tags FROM metrics
|
|
WHERE agent_id = $1 AND time >= $2 AND time <= $3`
|
|
args := []interface{}{agentID, from, to}
|
|
|
|
if metric != "" {
|
|
query += " AND metric = $4"
|
|
args = append(args, metric)
|
|
}
|
|
|
|
if tagsFilter != "" {
|
|
query += fmt.Sprintf(" AND tags @> $%d::jsonb", len(args)+1)
|
|
args = append(args, tagsFilter)
|
|
}
|
|
|
|
query += " ORDER BY time DESC LIMIT $" + fmt.Sprintf("%d", len(args)+1)
|
|
args = append(args, limit)
|
|
|
|
rows, err := d.db.Query(query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var results []map[string]interface{}
|
|
for rows.Next() {
|
|
var ts time.Time
|
|
var m string
|
|
var v float64
|
|
var tags sql.NullString
|
|
if err := rows.Scan(&ts, &m, &v, &tags); err != nil {
|
|
return nil, err
|
|
}
|
|
entry := map[string]interface{}{
|
|
"time": ts.Format(time.RFC3339),
|
|
"metric": m,
|
|
"value": v,
|
|
}
|
|
if tags.Valid {
|
|
var t map[string]interface{}
|
|
json.Unmarshal([]byte(tags.String), &t)
|
|
entry["tags"] = t
|
|
}
|
|
results = append(results, entry)
|
|
}
|
|
if results == nil {
|
|
results = []map[string]interface{}{}
|
|
}
|
|
return results, rows.Err()
|
|
}
|
|
|
|
// GetMetricsSummary holt aggregierte Metriken (avg/min/max) in Zeitbuckts
|
|
func (d *Database) GetMetricsSummary(agentID, metric, bucket string, from, to time.Time) ([]map[string]interface{}, error) {
|
|
if bucket == "" {
|
|
bucket = "5 minutes"
|
|
}
|
|
|
|
rows, err := d.db.Query(`
|
|
SELECT time_bucket($1::interval, time) AS t,
|
|
AVG(value) AS avg_value,
|
|
MIN(value) AS min_value,
|
|
MAX(value) AS max_value,
|
|
COUNT(*) AS samples
|
|
FROM metrics
|
|
WHERE agent_id = $2 AND metric = $3 AND time >= $4 AND time <= $5
|
|
GROUP BY t
|
|
ORDER BY t DESC
|
|
`, bucket, agentID, metric, from, to)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var results []map[string]interface{}
|
|
for rows.Next() {
|
|
var t time.Time
|
|
var avg, min, max float64
|
|
var samples int
|
|
if err := rows.Scan(&t, &avg, &min, &max, &samples); err != nil {
|
|
return nil, err
|
|
}
|
|
results = append(results, map[string]interface{}{
|
|
"time": t.Format(time.RFC3339),
|
|
"avg": avg,
|
|
"min": min,
|
|
"max": max,
|
|
"samples": samples,
|
|
})
|
|
}
|
|
if results == nil {
|
|
results = []map[string]interface{}{}
|
|
}
|
|
return results, rows.Err()
|
|
}
|
|
|
|
// --- Agent Events Methoden ---
|
|
|
|
// InsertAgentEvent schreibt ein Event in die agent_events Tabelle
|
|
func (d *Database) InsertAgentEvent(agentID, eventType, message string) error {
|
|
_, err := d.db.Exec(
|
|
"INSERT INTO agent_events (agent_id, event_type, message) VALUES ($1, $2, $3)",
|
|
agentID, eventType, message,
|
|
)
|
|
if err != nil {
|
|
log.Printf("Event schreiben fehlgeschlagen: %v", err)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// GetAgentEvents holt Events fuer einen bestimmten Agent
|
|
func (d *Database) GetAgentEvents(agentID string, eventType string, limit int) ([]models.AgentEvent, error) {
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
|
|
query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events WHERE agent_id = $1"
|
|
args := []interface{}{agentID}
|
|
|
|
if eventType != "" {
|
|
query += " AND event_type = $2"
|
|
args = append(args, eventType)
|
|
}
|
|
|
|
query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", len(args)+1)
|
|
args = append(args, limit)
|
|
|
|
rows, err := d.db.Query(query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var events []models.AgentEvent
|
|
for rows.Next() {
|
|
var e models.AgentEvent
|
|
if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
events = append(events, e)
|
|
}
|
|
if events == nil {
|
|
events = []models.AgentEvent{}
|
|
}
|
|
return events, rows.Err()
|
|
}
|
|
|
|
// GetAllAgentEvents holt Events fuer alle Agents
|
|
func (d *Database) GetAllAgentEvents(eventType string, limit int) ([]models.AgentEvent, error) {
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
|
|
query := "SELECT id, agent_id, event_type, COALESCE(message, ''), created_at FROM agent_events"
|
|
args := []interface{}{}
|
|
|
|
if eventType != "" {
|
|
query += " WHERE event_type = $1"
|
|
args = append(args, eventType)
|
|
}
|
|
|
|
query += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", len(args)+1)
|
|
args = append(args, limit)
|
|
|
|
rows, err := d.db.Query(query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var events []models.AgentEvent
|
|
for rows.Next() {
|
|
var e models.AgentEvent
|
|
if err := rows.Scan(&e.ID, &e.AgentID, &e.EventType, &e.Message, &e.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
events = append(events, e)
|
|
}
|
|
if events == nil {
|
|
events = []models.AgentEvent{}
|
|
}
|
|
return events, rows.Err()
|
|
}
|
|
|
|
// --- Agent-Customer Zuordnung ---
|
|
|
|
func (d *Database) AssignAgentCustomer(agentID string, customerID int) error {
|
|
_, err := d.db.Exec("UPDATE agents SET customer_id = $1 WHERE id = $2", customerID, agentID)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) UnassignAgentCustomer(agentID string) error {
|
|
_, err := d.db.Exec("UPDATE agents SET customer_id = NULL WHERE id = $1", agentID)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetAgentsByCustomer(customerID int) ([]models.Agent, error) {
|
|
rows, err := d.db.Query("SELECT id, name, hostname, ip, opnsense_version, registered_at, last_heartbeat, customer_id FROM agents WHERE customer_id = $1 ORDER BY name", customerID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var agents []models.Agent
|
|
for rows.Next() {
|
|
var a models.Agent
|
|
var lastHB sql.NullTime
|
|
var custID sql.NullInt64
|
|
if err := rows.Scan(&a.ID, &a.Name, &a.Hostname, &a.IP, &a.OPNsenseVersion, &a.RegisteredAt, &lastHB, &custID); err != nil {
|
|
return nil, err
|
|
}
|
|
if lastHB.Valid {
|
|
a.LastHeartbeat = &lastHB.Time
|
|
}
|
|
if custID.Valid {
|
|
cid := int(custID.Int64)
|
|
a.CustomerID = &cid
|
|
}
|
|
agents = append(agents, a)
|
|
}
|
|
if agents == nil {
|
|
agents = []models.Agent{}
|
|
}
|
|
return agents, rows.Err()
|
|
}
|
|
|
|
// GetAgentStatus berechnet den Status eines Agents aus last_heartbeat
|
|
func (d *Database) GetAgentStatus(agentID string) string {
|
|
var lastHB sql.NullTime
|
|
err := d.db.QueryRow("SELECT last_heartbeat FROM agents WHERE id = $1", agentID).Scan(&lastHB)
|
|
if err != nil || !lastHB.Valid {
|
|
return "unknown"
|
|
}
|
|
return models.CalculateAgentStatus(&lastHB.Time)
|
|
}
|
|
|
|
// ==================== API-Keys ====================
|
|
|
|
// Permissions: "agent" | "read" | "write" | "admin"
|
|
// - agent: nur /api/v1/agent/ws + /api/v1/agent/heartbeat (fuer Agents auf Kundenservern)
|
|
// - read: GET-Endpoints, optional auf customer_id beschraenkt
|
|
// - write: alles ausser Terminal (fuer externe Integrationen)
|
|
// - admin: voll (kein Terminal — das ist JWT-only)
|
|
type APIKey struct {
|
|
ID int `json:"id"`
|
|
Name string `json:"name"`
|
|
Key string `json:"key"`
|
|
Permissions string `json:"permissions"`
|
|
CustomerID *int `json:"customer_id,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
LastUsed *time.Time `json:"last_used,omitempty"`
|
|
}
|
|
|
|
func (d *Database) ListAPIKeys() ([]APIKey, error) {
|
|
rows, err := d.db.Query("SELECT id, name, key, permissions, customer_id, created_at, last_used FROM api_keys ORDER BY created_at")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var keys []APIKey
|
|
for rows.Next() {
|
|
var k APIKey
|
|
var lastUsed sql.NullTime
|
|
var custID sql.NullInt64
|
|
if err := rows.Scan(&k.ID, &k.Name, &k.Key, &k.Permissions, &custID, &k.CreatedAt, &lastUsed); err != nil {
|
|
return nil, err
|
|
}
|
|
if lastUsed.Valid {
|
|
k.LastUsed = &lastUsed.Time
|
|
}
|
|
if custID.Valid {
|
|
id := int(custID.Int64)
|
|
k.CustomerID = &id
|
|
}
|
|
keys = append(keys, k)
|
|
}
|
|
if keys == nil {
|
|
keys = []APIKey{}
|
|
}
|
|
return keys, rows.Err()
|
|
}
|
|
|
|
func (d *Database) CreateAPIKey(name, key, permissions string, customerID *int) (*APIKey, error) {
|
|
var k APIKey
|
|
var custID sql.NullInt64
|
|
if customerID != nil {
|
|
custID = sql.NullInt64{Int64: int64(*customerID), Valid: true}
|
|
}
|
|
var scannedCustID sql.NullInt64
|
|
err := d.db.QueryRow(
|
|
"INSERT INTO api_keys (name, key, permissions, customer_id) VALUES ($1, $2, $3, $4) RETURNING id, name, key, permissions, customer_id, created_at",
|
|
name, key, permissions, custID,
|
|
).Scan(&k.ID, &k.Name, &k.Key, &k.Permissions, &scannedCustID, &k.CreatedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if scannedCustID.Valid {
|
|
id := int(scannedCustID.Int64)
|
|
k.CustomerID = &id
|
|
}
|
|
return &k, nil
|
|
}
|
|
|
|
func (d *Database) DeleteAPIKey(id int) error {
|
|
result, err := d.db.Exec("DELETE FROM api_keys WHERE id = $1", id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("API-Key nicht gefunden")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// GetAPIKeyInfo gibt Key + Permissions + CustomerID zurueck (fuer Middleware)
|
|
func (d *Database) GetAPIKeyInfo(key string) (*APIKey, error) {
|
|
var k APIKey
|
|
var custID sql.NullInt64
|
|
var lastUsed sql.NullTime
|
|
err := d.db.QueryRow(
|
|
"SELECT id, name, key, permissions, customer_id, created_at, last_used FROM api_keys WHERE key = $1",
|
|
key,
|
|
).Scan(&k.ID, &k.Name, &k.Key, &k.Permissions, &custID, &k.CreatedAt, &lastUsed)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if custID.Valid {
|
|
id := int(custID.Int64)
|
|
k.CustomerID = &id
|
|
}
|
|
return &k, nil
|
|
}
|
|
|
|
func (d *Database) GetAllAPIKeys() ([]string, error) {
|
|
rows, err := d.db.Query("SELECT key FROM api_keys")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var keys []string
|
|
for rows.Next() {
|
|
var k string
|
|
if err := rows.Scan(&k); err != nil {
|
|
return nil, err
|
|
}
|
|
keys = append(keys, k)
|
|
}
|
|
return keys, rows.Err()
|
|
}
|
|
|
|
func (d *Database) UpdateAPIKeyLastUsed(key string) {
|
|
d.db.Exec("UPDATE api_keys SET last_used = NOW() WHERE key = $1", key)
|
|
}
|
|
|
|
// MigrateConfigAPIKeys migriert API-Keys aus der Config in die DB (einmalig)
|
|
// --- Audit Log ---
|
|
|
|
func (d *Database) AddAuditLog(userName, action, targetType, targetID, targetName, details, sourceIP string) error {
|
|
_, err := d.db.Exec(
|
|
`INSERT INTO audit_log (user_name, action, target_type, target_id, target_name, details, source_ip) VALUES ($1, $2, $3, $4, $5, $6, $7)`,
|
|
userName, action, targetType, targetID, targetName, details, sourceIP,
|
|
)
|
|
if err != nil {
|
|
log.Printf("Audit-Log schreiben fehlgeschlagen: %v", err)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetAuditLogs(limit int, offset int, action string, agentID string) ([]models.AuditLogEntry, int, error) {
|
|
if limit <= 0 {
|
|
limit = 100
|
|
}
|
|
if limit > 500 {
|
|
limit = 500
|
|
}
|
|
|
|
where := []string{}
|
|
args := []interface{}{}
|
|
argIdx := 1
|
|
|
|
if action != "" {
|
|
where = append(where, fmt.Sprintf("action = $%d", argIdx))
|
|
args = append(args, action)
|
|
argIdx++
|
|
}
|
|
if agentID != "" {
|
|
where = append(where, fmt.Sprintf("target_id = $%d", argIdx))
|
|
args = append(args, agentID)
|
|
argIdx++
|
|
}
|
|
|
|
whereClause := ""
|
|
if len(where) > 0 {
|
|
whereClause = "WHERE " + strings.Join(where, " AND ")
|
|
}
|
|
|
|
// Total count
|
|
var total int
|
|
countQuery := "SELECT COUNT(*) FROM audit_log " + whereClause
|
|
err := d.db.QueryRow(countQuery, args...).Scan(&total)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
// Data
|
|
query := fmt.Sprintf("SELECT id, timestamp, user_name, action, COALESCE(target_type,''), COALESCE(target_id,''), COALESCE(target_name,''), COALESCE(details,''), COALESCE(source_ip,'') FROM audit_log %s ORDER BY timestamp DESC LIMIT $%d OFFSET $%d", whereClause, argIdx, argIdx+1)
|
|
args = append(args, limit, offset)
|
|
|
|
rows, err := d.db.Query(query, args...)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var entries []models.AuditLogEntry
|
|
for rows.Next() {
|
|
var e models.AuditLogEntry
|
|
var ts time.Time
|
|
if err := rows.Scan(&e.ID, &ts, &e.UserName, &e.Action, &e.TargetType, &e.TargetID, &e.TargetName, &e.Details, &e.SourceIP); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
e.Timestamp = ts.Format(time.RFC3339)
|
|
entries = append(entries, e)
|
|
}
|
|
if entries == nil {
|
|
entries = []models.AuditLogEntry{}
|
|
}
|
|
return entries, total, rows.Err()
|
|
}
|
|
|
|
// --- System Settings ---
|
|
|
|
func (d *Database) GetSettings() (map[string]string, error) {
|
|
rows, err := d.db.Query("SELECT key, value FROM system_settings ORDER BY key")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
settings := make(map[string]string)
|
|
for rows.Next() {
|
|
var k, v string
|
|
if err := rows.Scan(&k, &v); err != nil {
|
|
return nil, err
|
|
}
|
|
settings[k] = v
|
|
}
|
|
return settings, rows.Err()
|
|
}
|
|
|
|
func (d *Database) GetSetting(key string) (string, error) {
|
|
var value string
|
|
err := d.db.QueryRow("SELECT value FROM system_settings WHERE key = $1", key).Scan(&value)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func (d *Database) SetSetting(key, value string) error {
|
|
_, err := d.db.Exec(`
|
|
INSERT INTO system_settings (key, value, updated_at) VALUES ($1, $2, NOW())
|
|
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()
|
|
`, key, value)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) MigrateConfigAPIKeys(configKeys []string) {
|
|
for _, key := range configKeys {
|
|
var exists bool
|
|
d.db.QueryRow("SELECT EXISTS(SELECT 1 FROM api_keys WHERE key = $1)", key).Scan(&exists)
|
|
if !exists {
|
|
d.db.Exec("INSERT INTO api_keys (name, key) VALUES ($1, $2)", "Default (migriert)", key)
|
|
log.Printf("API-Key aus Config migriert: %s...", key[:8])
|
|
}
|
|
}
|
|
}
|