Skip to content

Commit f49f55d

Browse files
snormoresnormore
committed
telemetry: state-ingest wip
1 parent fd6cb26 commit f49f55d

2 files changed

Lines changed: 340 additions & 0 deletions

File tree

‎telemetry/state-ingest/cmd/state-ingest/main.go‎

Lines changed: 227 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,227 @@
1+
package main
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"crypto/ed25519"
7+
"crypto/sha256"
8+
"encoding/hex"
9+
"encoding/json"
10+
"fmt"
11+
"io"
12+
"log"
13+
"net/http"
14+
"os"
15+
"strings"
16+
"time"
17+
18+
"github.com/aws/aws-sdk-go-v2/aws"
19+
"github.com/aws/aws-sdk-go-v2/config"
20+
"github.com/aws/aws-sdk-go-v2/service/s3"
21+
"github.com/mr-tron/base58/base58"
22+
)
23+
24+
type Metadata struct {
25+
SnapshotTimestamp string `json:"snapshot_timestamp"`
26+
Command string `json:"command"`
27+
DevicePubkey string `json:"device_pubkey"`
28+
Kind string `json:"kind"`
29+
}
30+
31+
type PushRequest struct {
32+
Metadata Metadata `json:"metadata"`
33+
Data json.RawMessage `json:"data"`
34+
}
35+
36+
type Server struct {
37+
s3 *s3.Client
38+
bucket string
39+
prefix string
40+
}
41+
42+
func sanitizePathComponent(s string) string {
43+
s = strings.TrimSpace(s)
44+
if s == "" {
45+
return "default"
46+
}
47+
s = strings.ReplaceAll(s, "/", "_")
48+
s = strings.ReplaceAll(s, " ", "_")
49+
return s
50+
}
51+
52+
func (s *Server) handlePush(w http.ResponseWriter, r *http.Request, kindOverride string) {
53+
if r.Method != http.MethodPost {
54+
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
55+
return
56+
}
57+
58+
pubkeyB58 := r.Header.Get("X-DoubleZero-Pubkey")
59+
sigB58 := r.Header.Get("X-DoubleZero-Signature")
60+
tsHeader := r.Header.Get("X-DoubleZero-Timestamp")
61+
if pubkeyB58 == "" || sigB58 == "" || tsHeader == "" {
62+
http.Error(w, "missing auth headers", http.StatusUnauthorized)
63+
return
64+
}
65+
66+
clientTS, err := time.Parse(time.RFC3339, tsHeader)
67+
if err != nil {
68+
http.Error(w, "invalid X-DoubleZero-Timestamp", http.StatusUnauthorized)
69+
return
70+
}
71+
if d := time.Since(clientTS); d < -5*time.Minute || d > 5*time.Minute {
72+
http.Error(w, "timestamp out of acceptable window", http.StatusUnauthorized)
73+
return
74+
}
75+
76+
pubkeyBytes, err := base58.Decode(pubkeyB58)
77+
if err != nil || len(pubkeyBytes) != ed25519.PublicKeySize {
78+
http.Error(w, "invalid X-DoubleZero-Pubkey", http.StatusUnauthorized)
79+
return
80+
}
81+
sigBytes, err := base58.Decode(sigB58)
82+
if err != nil || len(sigBytes) != ed25519.SignatureSize {
83+
http.Error(w, "invalid X-DoubleZero-Signature", http.StatusUnauthorized)
84+
return
85+
}
86+
87+
rawBody, err := io.ReadAll(r.Body)
88+
if err != nil {
89+
http.Error(w, "failed to read body", http.StatusBadRequest)
90+
return
91+
}
92+
r.Body.Close()
93+
r.Body = io.NopCloser(bytes.NewReader(rawBody))
94+
95+
hash := sha256.Sum256(rawBody)
96+
bodyHashHex := hex.EncodeToString(hash[:])
97+
98+
canonical := fmt.Sprintf(
99+
"DOUBLEZERO_STATE_PUSH_V1\nmethod:%s\npath:%s\ntimestamp:%s\nbody-sha256:%s\n",
100+
r.Method,
101+
r.URL.Path,
102+
tsHeader,
103+
bodyHashHex,
104+
)
105+
106+
if !ed25519.Verify(ed25519.PublicKey(pubkeyBytes), []byte(canonical), sigBytes) {
107+
http.Error(w, "invalid signature", http.StatusUnauthorized)
108+
return
109+
}
110+
111+
var req PushRequest
112+
if err := json.NewDecoder(bytes.NewReader(rawBody)).Decode(&req); err != nil {
113+
http.Error(w, "invalid json", http.StatusBadRequest)
114+
return
115+
}
116+
117+
if req.Metadata.DevicePubkey == "" || req.Metadata.SnapshotTimestamp == "" {
118+
http.Error(w, "missing metadata.device_pubkey or metadata.snapshot_timestamp", http.StatusBadRequest)
119+
return
120+
}
121+
if req.Metadata.DevicePubkey != pubkeyB58 {
122+
http.Error(w, "device_pubkey mismatch", http.StatusUnauthorized)
123+
return
124+
}
125+
126+
ts, err := time.Parse(time.RFC3339, req.Metadata.SnapshotTimestamp)
127+
if err != nil {
128+
http.Error(w, "invalid metadata.snapshot_timestamp (expect RFC3339)", http.StatusBadRequest)
129+
return
130+
}
131+
ts = ts.UTC()
132+
date := ts.Format("2006-01-02")
133+
hour := ts.Format("15")
134+
filename := ts.Format("20060102T150405Z") + ".json"
135+
136+
kind := kindOverride
137+
if kind == "" {
138+
kind = r.URL.Query().Get("kind")
139+
}
140+
if kind == "" {
141+
kind = req.Metadata.Kind
142+
}
143+
if kind == "" {
144+
kind = req.Metadata.Command
145+
}
146+
kind = sanitizePathComponent(kind)
147+
148+
device := sanitizePathComponent(req.Metadata.DevicePubkey)
149+
150+
keyPrefix := fmt.Sprintf("state/%s/device=%s/date=%s/hour=%s/", kind, device, date, hour)
151+
if s.prefix != "" {
152+
keyPrefix = s.prefix + "/" + keyPrefix
153+
}
154+
key := keyPrefix + filename
155+
156+
_, err = s.s3.PutObject(r.Context(), &s3.PutObjectInput{
157+
Bucket: aws.String(s.bucket),
158+
Key: aws.String(key),
159+
Body: bytes.NewReader(rawBody),
160+
ContentType: aws.String("application/json"),
161+
})
162+
if err != nil {
163+
log.Printf("s3 PutObject error: %v", err)
164+
http.Error(w, "failed to store payload", http.StatusInternalServerError)
165+
return
166+
}
167+
168+
w.Header().Set("Content-Type", "application/json")
169+
_ = json.NewEncoder(w).Encode(map[string]string{
170+
"status": "ok",
171+
"key": key,
172+
})
173+
}
174+
175+
func (s *Server) pushHandler(w http.ResponseWriter, r *http.Request) {
176+
s.handlePush(w, r, "")
177+
}
178+
179+
func (s *Server) pushWithKindHandler(w http.ResponseWriter, r *http.Request) {
180+
const base = "/v1/push/"
181+
if !strings.HasPrefix(r.URL.Path, base) {
182+
http.NotFound(w, r)
183+
return
184+
}
185+
rest := strings.TrimPrefix(r.URL.Path, base)
186+
parts := strings.SplitN(rest, "/", 2)
187+
kind := parts[0]
188+
if kind == "" {
189+
http.Error(w, "missing kind in path", http.StatusBadRequest)
190+
return
191+
}
192+
s.handlePush(w, r, kind)
193+
}
194+
195+
func main() {
196+
bucket := os.Getenv("S3_BUCKET")
197+
if bucket == "" {
198+
log.Fatal("S3_BUCKET is required")
199+
}
200+
prefix := os.Getenv("S3_PREFIX")
201+
202+
ctx := context.Background()
203+
cfg, err := config.LoadDefaultConfig(ctx)
204+
if err != nil {
205+
log.Fatalf("failed to load AWS config: %v", err)
206+
}
207+
s3Client := s3.NewFromConfig(cfg)
208+
209+
srv := &Server{
210+
s3: s3Client,
211+
bucket: bucket,
212+
prefix: prefix,
213+
}
214+
215+
mux := http.NewServeMux()
216+
mux.HandleFunc("/v1/push", srv.pushHandler)
217+
mux.HandleFunc("/v1/push/", srv.pushWithKindHandler)
218+
219+
port := os.Getenv("PORT")
220+
if port == "" {
221+
port = "8080"
222+
}
223+
log.Printf("listening on %s", port)
224+
if err := http.ListenAndServe(":"+port, mux); err != nil && err != http.ErrServerClosed {
225+
log.Fatal(err)
226+
}
227+
}

‎telemetry/state-ingest/pkg/ingest/client.go‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
package ingest
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"crypto/ed25519"
7+
"crypto/sha256"
8+
"encoding/hex"
9+
"encoding/json"
10+
"fmt"
11+
"io"
12+
"net/http"
13+
"net/url"
14+
"time"
15+
16+
"github.com/gagliardetto/solana-go"
17+
"github.com/mr-tron/base58/base58"
18+
)
19+
20+
type Metadata struct {
21+
SnapshotTimestamp string `json:"snapshot_timestamp"`
22+
Command string `json:"command"`
23+
DevicePubkey solana.PublicKey `json:"device_pubkey"`
24+
Kind string `json:"kind"`
25+
}
26+
27+
type PushRequest struct {
28+
Metadata Metadata `json:"metadata"`
29+
Data interface{} `json:"data"`
30+
}
31+
32+
type Client struct {
33+
baseURL *url.URL
34+
httpClient *http.Client
35+
pubkey solana.PublicKey
36+
privKey ed25519.PrivateKey
37+
}
38+
39+
func NewClient(rawBaseURL string, pubkey solana.PublicKey, privKey ed25519.PrivateKey) (*Client, error) {
40+
u, err := url.Parse(rawBaseURL)
41+
if err != nil {
42+
return nil, err
43+
}
44+
return &Client{
45+
baseURL: u,
46+
httpClient: &http.Client{Timeout: 10 * time.Second},
47+
pubkey: pubkey,
48+
privKey: privKey,
49+
}, nil
50+
}
51+
52+
func (c *Client) Push(ctx context.Context, kind string, req PushRequest) error {
53+
if req.Metadata.DevicePubkey.IsZero() {
54+
req.Metadata.DevicePubkey = c.pubkey
55+
}
56+
if req.Metadata.SnapshotTimestamp == "" {
57+
req.Metadata.SnapshotTimestamp = time.Now().UTC().Format(time.RFC3339)
58+
}
59+
if req.Metadata.Kind == "" {
60+
req.Metadata.Kind = kind
61+
}
62+
63+
path := "/v1/push"
64+
if kind != "" {
65+
path = "/v1/push/" + kind
66+
}
67+
68+
fullURL := *c.baseURL
69+
fullURL.Path = path
70+
71+
bodyBytes, err := json.Marshal(req)
72+
if err != nil {
73+
return fmt.Errorf("marshal body: %w", err)
74+
}
75+
76+
ts := time.Now().UTC().Format(time.RFC3339)
77+
78+
h := sha256.Sum256(bodyBytes)
79+
bodyHashHex := hex.EncodeToString(h[:])
80+
81+
canonical := fmt.Sprintf(
82+
"DOUBLEZERO_STATE_PUSH_V1\nmethod:%s\npath:%s\ntimestamp:%s\nbody-sha256:%s\n",
83+
"POST",
84+
path,
85+
ts,
86+
bodyHashHex,
87+
)
88+
89+
sig := ed25519.Sign(c.privKey, []byte(canonical))
90+
sigB58 := base58.Encode(sig)
91+
92+
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, fullURL.String(), bytes.NewReader(bodyBytes))
93+
if err != nil {
94+
return fmt.Errorf("build request: %w", err)
95+
}
96+
httpReq.Header.Set("Content-Type", "application/json")
97+
httpReq.Header.Set("X-DoubleZero-Pubkey", c.pubkey.String())
98+
httpReq.Header.Set("X-DoubleZero-Signature", sigB58)
99+
httpReq.Header.Set("X-DoubleZero-Timestamp", ts)
100+
101+
resp, err := c.httpClient.Do(httpReq)
102+
if err != nil {
103+
return fmt.Errorf("do request: %w", err)
104+
}
105+
defer resp.Body.Close()
106+
107+
if resp.StatusCode >= 300 {
108+
b, _ := io.ReadAll(resp.Body)
109+
return fmt.Errorf("server error: %s: %s", resp.Status, string(b))
110+
}
111+
112+
return nil
113+
}

0 commit comments

Comments
 (0)