From 0a2d3177c68e2dbfdf978a2e1b5665e678ff89c3 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Tue, 29 Sep 2026 11:52:35 -0400 Subject: [PATCH] feat: pkg/tsnats digests-only NATS timestamp client --- go.mod | 12 +- go.sum | 12 ++ pkg/tsnats/client.go | 398 ++++++++++++++++++++++++++++++++++++++ pkg/tsnats/client_test.go | 267 +++++++++++++++++++++++++ pkg/tsnats/subjects.go | 9 + 5 files changed, 697 insertions(+), 1 deletion(-) create mode 100644 go.sum create mode 100644 pkg/tsnats/client.go create mode 100644 pkg/tsnats/client_test.go create mode 100644 pkg/tsnats/subjects.go diff --git a/go.mod b/go.mod index 274cb05..3b2855d 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,13 @@ module git.georgelambert.org/marchon/verae-mcp-timestamping-server-go -go 1.22 +go 1.22.0 + +require github.com/nats-io/nats.go v1.39.1 + +require ( + github.com/klauspost/compress v1.17.11 // indirect + github.com/nats-io/nkeys v0.4.9 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.31.0 // indirect + golang.org/x/sys v0.28.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..7ba462d --- /dev/null +++ b/go.sum @@ -0,0 +1,12 @@ +github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc= +github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0= +github.com/nats-io/nats.go v1.39.1 h1:oTkfKBmz7W047vRxV762M67ZdXeOtUgvbBaNoQ+3PPk= +github.com/nats-io/nats.go v1.39.1/go.mod h1:MgRb8oOdigA6cYpEPhXJuRVH6UE/V4jblJ2jQ27IXYM= +github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0= +github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U= +golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= +golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/pkg/tsnats/client.go b/pkg/tsnats/client.go new file mode 100644 index 0000000..622a476 --- /dev/null +++ b/pkg/tsnats/client.go @@ -0,0 +1,398 @@ +package tsnats + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/nats-io/nats.go" + + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/pkg/digest" +) + +const defaultTimeout = 2 * time.Second + +// Requester is the request/reply surface used by Stamp/Verify/HashReport. +// Inject a mock in unit tests; Connect wraps *nats.Conn. +type Requester interface { + Request(ctx context.Context, subj string, data []byte) ([]byte, error) +} + +// MessageHandler receives raw message payloads from a core NATS subscription. +type MessageHandler func(data []byte) + +// SubConn is the subscribe surface for SubscribeCompleted. +type SubConn interface { + Subscribe(subj string, cb MessageHandler) (Subscription, error) +} + +// Subscription is returned by SubscribeCompleted. +type Subscription interface { + Unsubscribe() error +} + +// Receipt is a timestamp receipt from request/reply or hash.report. +type Receipt struct { + ReceiptID string `json:"receipt_id"` + Hash string `json:"hash"` + Scope string `json:"scope"` + CubeID string `json:"cube_id,omitempty"` + OrgID string `json:"org_id,omitempty"` + CreatedAt string `json:"created_at"` + Merkle string `json:"merkle"` + RestampOf string `json:"restamp_of,omitempty"` +} + +// VerifyResult is the outcome of Verify. +type VerifyResult struct { + Valid bool + Error string + Receipt *Receipt +} + +// CompletedEvent is a fire-and-forget job-complete notification on SubjectCompleted. +type CompletedEvent struct { + JobID string `json:"jobId"` + ReceiptID string `json:"receipt_id"` + Hash string `json:"hash,omitempty"` +} + +// StampOptions configures Stamp. Authorization/AccessToken are optional +// passthrough fields for embedded handlers (JSON keys authorization / access_token). +type StampOptions struct { + Scope string + CubeID string + OrgID string + Authorization string + AccessToken string +} + +// VerifyOptions configures Verify. +type VerifyOptions struct { + WantScope string + Authorization string + AccessToken string +} + +// ReportOptions configures HashReport. +type ReportOptions struct { + OrgID string + WantScope string + Authorization string + AccessToken string +} + +// Option configures Client construction. +type Option func(*Client) + +// WithTimeout sets the request timeout (default 2s). +func WithTimeout(d time.Duration) Option { + return func(c *Client) { + if d > 0 { + c.timeout = d + } + } +} + +// WithRequester injects a Requester (tests or custom transports). +func WithRequester(r Requester) Option { + return func(c *Client) { c.req = r } +} + +// WithSubscriber injects a SubConn for SubscribeCompleted. +func WithSubscriber(s SubConn) Option { + return func(c *Client) { c.sub = s } +} + +// Client is a digests-only NATS timestamping client (core NATS only; no JetStream). +type Client struct { + req Requester + sub SubConn + timeout time.Duration + nc *nats.Conn +} + +// NewClient builds a Client with an injected Requester (unit tests). +func NewClient(r Requester, opts ...Option) *Client { + c := &Client{req: r, timeout: defaultTimeout} + for _, opt := range opts { + opt(c) + } + return c +} + +// Connect dials a NATS URL and returns a Client using core request/reply and subscribe. +func Connect(url string, opts ...Option) (*Client, error) { + nc, err := nats.Connect(url) + if err != nil { + return nil, err + } + wrap := &natsAdapter{nc: nc} + c := &Client{req: wrap, sub: wrap, timeout: defaultTimeout, nc: nc} + for _, opt := range opts { + opt(c) + } + if c.req == nil { + c.req = wrap + } + if c.sub == nil { + c.sub = wrap + } + return c, nil +} + +// Close closes the underlying NATS connection when opened via Connect. +func (c *Client) Close() { + if c.nc != nil { + c.nc.Close() + } +} + +// Timeout returns the request timeout. +func (c *Client) Timeout() time.Duration { + return c.timeout +} + +// Stamp publishes a digests-only stamp request on SubjectRequest. +func (c *Client) Stamp(ctx context.Context, sha256 string, opts StampOptions) ([]Receipt, error) { + h, err := digest.NormalizeSHA256(sha256) + if err != nil { + return nil, err + } + scope := opts.Scope + if scope == "" { + scope = "local" + } + payload := map[string]any{ + "hash": h, + "scope": scope, + "cube_id": opts.CubeID, + "org_id": opts.OrgID, + } + putAuth(payload, opts.Authorization, opts.AccessToken) + raw, err := c.request(ctx, SubjectRequest, payload) + if err != nil { + return nil, err + } + return parseStampReply(raw) +} + +// Verify asks SubjectVerify for a receipt by id. +func (c *Client) Verify(ctx context.Context, receiptID string, opts VerifyOptions) (VerifyResult, error) { + if receiptID == "" { + return VerifyResult{Valid: false, Error: "receipt_id required"}, nil + } + payload := map[string]any{ + "receipt_id": receiptID, + "want_scope": opts.WantScope, + } + putAuth(payload, opts.Authorization, opts.AccessToken) + raw, err := c.request(ctx, SubjectVerify, payload) + if err != nil { + return VerifyResult{}, err + } + return parseVerifyReply(raw) +} + +// HashReport asks SubjectReport for receipts for a digest. +func (c *Client) HashReport(ctx context.Context, sha256 string, opts ReportOptions) ([]Receipt, error) { + h, err := digest.NormalizeSHA256(sha256) + if err != nil { + return nil, err + } + payload := map[string]any{ + "hash": h, + "org_id": opts.OrgID, + "want_scope": opts.WantScope, + } + putAuth(payload, opts.Authorization, opts.AccessToken) + raw, err := c.request(ctx, SubjectReport, payload) + if err != nil { + return nil, err + } + return parseReportReply(raw) +} + +// SubscribeCompleted core-subscribes to SubjectCompleted (not JetStream). +func (c *Client) SubscribeCompleted(ctx context.Context, handler func(CompletedEvent)) (Subscription, error) { + if c.sub == nil { + return nil, fmt.Errorf("tsnats: no subscriber configured") + } + if handler == nil { + return nil, fmt.Errorf("tsnats: handler required") + } + _ = ctx + return c.sub.Subscribe(SubjectCompleted, func(data []byte) { + var ev CompletedEvent + if err := json.Unmarshal(data, &ev); err != nil { + return + } + handler(ev) + }) +} + +func (c *Client) request(ctx context.Context, subj string, payload map[string]any) ([]byte, error) { + if c.req == nil { + return nil, fmt.Errorf("tsnats: no requester configured") + } + data, err := json.Marshal(payload) + if err != nil { + return nil, err + } + ctx, cancel := withTimeout(ctx, c.timeout) + defer cancel() + return c.req.Request(ctx, subj, data) +} + +func putAuth(payload map[string]any, authorization, accessToken string) { + if authorization != "" { + payload["authorization"] = authorization + } + if accessToken != "" { + payload["access_token"] = accessToken + } +} + +func withTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + if _, ok := ctx.Deadline(); ok { + return context.WithCancel(ctx) + } + return context.WithTimeout(ctx, d) +} + +func parseStampReply(raw []byte) ([]Receipt, error) { + var asList []Receipt + if err := json.Unmarshal(raw, &asList); err == nil && (len(asList) > 0 || isJSONArray(raw)) { + return asList, nil + } + var obj map[string]json.RawMessage + if err := json.Unmarshal(raw, &obj); err != nil { + return nil, fmt.Errorf("tsnats: stamp reply: %w", err) + } + if errRaw, ok := obj["error"]; ok { + var msg string + _ = json.Unmarshal(errRaw, &msg) + if msg == "" { + msg = string(errRaw) + } + return nil, fmt.Errorf("%s", msg) + } + if _, hasReceipt := obj["receipt_id"]; hasReceipt { + var rec Receipt + if err := json.Unmarshal(raw, &rec); err != nil { + return nil, err + } + return []Receipt{rec}, nil + } + if jobRaw, ok := obj["jobId"]; ok { + var jobID string + _ = json.Unmarshal(jobRaw, &jobID) + return nil, fmt.Errorf("stamp pending jobId=%s; use SubscribeCompleted", jobID) + } + return nil, fmt.Errorf("tsnats: unrecognized stamp reply") +} + +func parseVerifyReply(raw []byte) (VerifyResult, error) { + var obj map[string]any + if err := json.Unmarshal(raw, &obj); err != nil { + return VerifyResult{Valid: false, Error: "invalid verify payload"}, nil + } + if v, ok := obj["valid"]; ok { + if b, _ := v.(bool); !b { + errMsg, _ := obj["error"].(string) + return VerifyResult{Valid: false, Error: errMsg}, nil + } + } + if errMsg, ok := obj["error"].(string); ok && errMsg != "" { + if valid, _ := obj["valid"].(bool); !valid { + return VerifyResult{Valid: false, Error: errMsg}, nil + } + } + if _, ok := obj["receipt_id"]; ok { + var rec Receipt + if err := json.Unmarshal(raw, &rec); err != nil { + return VerifyResult{Valid: false, Error: err.Error()}, nil + } + return VerifyResult{Valid: true, Receipt: &rec}, nil + } + return VerifyResult{Valid: false, Error: "unknown receipt"}, nil +} + +func parseReportReply(raw []byte) ([]Receipt, error) { + var asList []Receipt + if err := json.Unmarshal(raw, &asList); err == nil && (len(asList) > 0 || isJSONArray(raw)) { + return asList, nil + } + var obj map[string]json.RawMessage + if err := json.Unmarshal(raw, &obj); err != nil { + return nil, nil + } + if errRaw, ok := obj["error"]; ok { + var msg string + _ = json.Unmarshal(errRaw, &msg) + if msg == "" { + msg = string(errRaw) + } + return nil, fmt.Errorf("%s", msg) + } + if receiptsRaw, ok := obj["receipts"]; ok { + var receipts []Receipt + if err := json.Unmarshal(receiptsRaw, &receipts); err != nil { + return nil, err + } + return receipts, nil + } + if timelineRaw, ok := obj["timeline"]; ok { + var events []map[string]json.RawMessage + if err := json.Unmarshal(timelineRaw, &events); err != nil { + return nil, nil + } + var out []Receipt + for _, ev := range events { + recRaw, ok := ev["receipt"] + if !ok { + continue + } + var rec Receipt + if err := json.Unmarshal(recRaw, &rec); err != nil { + continue + } + out = append(out, rec) + } + return out, nil + } + return nil, nil +} + +func isJSONArray(raw []byte) bool { + for _, b := range raw { + if b == ' ' || b == '\t' || b == '\n' || b == '\r' { + continue + } + return b == '[' + } + return false +} + +type natsAdapter struct { + nc *nats.Conn +} + +func (a *natsAdapter) Request(ctx context.Context, subj string, data []byte) ([]byte, error) { + msg, err := a.nc.RequestWithContext(ctx, subj, data) + if err != nil { + return nil, err + } + return msg.Data, nil +} + +func (a *natsAdapter) Subscribe(subj string, cb MessageHandler) (Subscription, error) { + sub, err := a.nc.Subscribe(subj, func(msg *nats.Msg) { + cb(msg.Data) + }) + if err != nil { + return nil, err + } + return sub, nil +} diff --git a/pkg/tsnats/client_test.go b/pkg/tsnats/client_test.go new file mode 100644 index 0000000..596e2bd --- /dev/null +++ b/pkg/tsnats/client_test.go @@ -0,0 +1,267 @@ +package tsnats + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" +) + +const testDigest = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + +type mockRequester struct { + mu sync.Mutex + lastSubj string + lastData []byte + reply []byte + err error +} + +func (m *mockRequester) Request(ctx context.Context, subj string, data []byte) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.lastSubj = subj + m.lastData = append([]byte(nil), data...) + if m.err != nil { + return nil, m.err + } + return m.reply, nil +} + +type mockSubConn struct { + mu sync.Mutex + subj string + handler MessageHandler + unsub bool +} + +func (m *mockSubConn) Subscribe(subj string, cb MessageHandler) (Subscription, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.subj = subj + m.handler = cb + return &mockSubscription{parent: m}, nil +} + +type mockSubscription struct { + parent *mockSubConn +} + +func (s *mockSubscription) Unsubscribe() error { + s.parent.mu.Lock() + defer s.parent.mu.Unlock() + s.parent.unsub = true + return nil +} + +func TestSubjects(t *testing.T) { + if SubjectRequest != "verae.ts.request" { + t.Fatalf("SubjectRequest=%q", SubjectRequest) + } + if SubjectVerify != "verae.ts.verify" { + t.Fatalf("SubjectVerify=%q", SubjectVerify) + } + if SubjectReport != "verae.hash.report" { + t.Fatalf("SubjectReport=%q", SubjectReport) + } + if SubjectCompleted != "verae.ts.completed" { + t.Fatalf("SubjectCompleted=%q", SubjectCompleted) + } +} + +func TestStamp_SendsJSONAndParsesReceiptList(t *testing.T) { + reply, _ := json.Marshal([]map[string]any{{ + "receipt_id": "rid-1", + "hash": testDigest, + "scope": "global", + "created_at": "2026-01-01T00:00:00Z", + "merkle": "m1", + }}) + mock := &mockRequester{reply: reply} + c := NewClient(mock) + + receipts, err := c.Stamp(context.Background(), testDigest, StampOptions{Scope: "global"}) + if err != nil { + t.Fatalf("Stamp: %v", err) + } + if mock.lastSubj != SubjectRequest { + t.Fatalf("subject=%q want %q", mock.lastSubj, SubjectRequest) + } + var body map[string]any + if err := json.Unmarshal(mock.lastData, &body); err != nil { + t.Fatalf("request json: %v", err) + } + if body["hash"] != testDigest || body["scope"] != "global" { + t.Fatalf("request body=%v", body) + } + if len(receipts) != 1 || receipts[0].ReceiptID != "rid-1" || receipts[0].Hash != testDigest { + t.Fatalf("receipts=%+v", receipts) + } +} + +func TestStamp_NormalizesDigestAndRejectsInvalid(t *testing.T) { + reply, _ := json.Marshal(map[string]any{ + "receipt_id": "rid-1", + "hash": testDigest, + "scope": "local", + "created_at": "t", + "merkle": "m", + }) + mock := &mockRequester{reply: reply} + c := NewClient(mock) + + _, err := c.Stamp(context.Background(), "sha256:"+strings.ToUpper(testDigest), StampOptions{Scope: "local"}) + if err != nil { + t.Fatalf("Stamp: %v", err) + } + var body map[string]any + _ = json.Unmarshal(mock.lastData, &body) + if body["hash"] != testDigest { + t.Fatalf("normalized hash=%v", body["hash"]) + } + + beforeSubj, beforeData := mock.lastSubj, string(mock.lastData) + _, err = c.Stamp(context.Background(), "notadigest", StampOptions{}) + if err == nil { + t.Fatal("expected digest error") + } + if mock.lastSubj != beforeSubj || string(mock.lastData) != beforeData { + t.Fatal("invalid digest must not publish") + } +} + +func TestStamp_IncludesBearerFields(t *testing.T) { + reply, _ := json.Marshal(map[string]any{ + "receipt_id": "rid-1", + "hash": testDigest, + "scope": "global", + "created_at": "t", + "merkle": "m", + }) + mock := &mockRequester{reply: reply} + c := NewClient(mock) + _, err := c.Stamp(context.Background(), testDigest, StampOptions{ + Scope: "global", + Authorization: "tok-a", + AccessToken: "tok-b", + }) + if err != nil { + t.Fatalf("Stamp: %v", err) + } + var body map[string]any + _ = json.Unmarshal(mock.lastData, &body) + if body["authorization"] != "tok-a" || body["access_token"] != "tok-b" { + t.Fatalf("auth fields missing: %v", body) + } +} + +func TestStamp_ErrorAndJobID(t *testing.T) { + mock := &mockRequester{reply: []byte(`{"error":"boom"}`)} + c := NewClient(mock) + _, err := c.Stamp(context.Background(), testDigest, StampOptions{Scope: "local"}) + if err == nil || !strings.Contains(err.Error(), "boom") { + t.Fatalf("want boom error, got %v", err) + } + + mock.reply = []byte(`{"jobId":"job-9"}`) + _, err = c.Stamp(context.Background(), testDigest, StampOptions{Scope: "local"}) + if err == nil || !strings.Contains(err.Error(), "job-9") { + t.Fatalf("want pending jobId error, got %v", err) + } +} + +func TestVerify_ParsesReceiptAndInvalid(t *testing.T) { + mock := &mockRequester{reply: []byte(`{"receipt_id":"rid-1","hash":"` + testDigest + `","scope":"local","created_at":"t","merkle":"m"}`)} + c := NewClient(mock) + res, err := c.Verify(context.Background(), "rid-1", VerifyOptions{WantScope: "local"}) + if err != nil { + t.Fatalf("Verify: %v", err) + } + if mock.lastSubj != SubjectVerify { + t.Fatalf("subject=%q", mock.lastSubj) + } + var body map[string]any + _ = json.Unmarshal(mock.lastData, &body) + if body["receipt_id"] != "rid-1" || body["want_scope"] != "local" { + t.Fatalf("body=%v", body) + } + if !res.Valid || res.Receipt == nil || res.Receipt.ReceiptID != "rid-1" { + t.Fatalf("result=%+v", res) + } + + mock.reply = []byte(`{"valid":false,"error":"unknown"}`) + res, err = c.Verify(context.Background(), "missing", VerifyOptions{}) + if err != nil { + t.Fatalf("Verify invalid: %v", err) + } + if res.Valid || res.Error != "unknown" { + t.Fatalf("result=%+v", res) + } +} + +func TestHashReport_ParsesReceiptsWrapper(t *testing.T) { + mock := &mockRequester{reply: []byte(`{"receipts":[{"receipt_id":"rid-1","hash":"` + testDigest + `","scope":"local","created_at":"t","merkle":"m"}]}`)} + c := NewClient(mock) + receipts, err := c.HashReport(context.Background(), testDigest, ReportOptions{OrgID: "org-1", WantScope: "local"}) + if err != nil { + t.Fatalf("HashReport: %v", err) + } + if mock.lastSubj != SubjectReport { + t.Fatalf("subject=%q", mock.lastSubj) + } + var body map[string]any + _ = json.Unmarshal(mock.lastData, &body) + if body["hash"] != testDigest || body["org_id"] != "org-1" || body["want_scope"] != "local" { + t.Fatalf("body=%v", body) + } + if len(receipts) != 1 || receipts[0].ReceiptID != "rid-1" { + t.Fatalf("receipts=%+v", receipts) + } +} + +func TestSubscribeCompleted(t *testing.T) { + sub := &mockSubConn{} + c := NewClient(&mockRequester{}, WithSubscriber(sub)) + var got CompletedEvent + var wg sync.WaitGroup + wg.Add(1) + s, err := c.SubscribeCompleted(context.Background(), func(ev CompletedEvent) { + got = ev + wg.Done() + }) + if err != nil { + t.Fatalf("SubscribeCompleted: %v", err) + } + if sub.subj != SubjectCompleted { + t.Fatalf("subj=%q", sub.subj) + } + sub.handler([]byte(`{"jobId":"j1","receipt_id":"r1","hash":"` + testDigest + `"}`)) + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("handler not called") + } + if got.JobID != "j1" || got.ReceiptID != "r1" || got.Hash != testDigest { + t.Fatalf("event=%+v", got) + } + if err := s.Unsubscribe(); err != nil { + t.Fatalf("Unsubscribe: %v", err) + } + if !sub.unsub { + t.Fatal("expected unsubscribe") + } +} + +func TestDefaultTimeout(t *testing.T) { + c := NewClient(&mockRequester{}) + if c.Timeout() != 2*time.Second { + t.Fatalf("timeout=%v", c.Timeout()) + } +} diff --git a/pkg/tsnats/subjects.go b/pkg/tsnats/subjects.go new file mode 100644 index 0000000..90a9885 --- /dev/null +++ b/pkg/tsnats/subjects.go @@ -0,0 +1,9 @@ +package tsnats + +// Core NATS subjects for timestamping. Job-complete is ephemeral core NATS only. +const ( + SubjectRequest = "verae.ts.request" + SubjectVerify = "verae.ts.verify" + SubjectReport = "verae.hash.report" + SubjectCompleted = "verae.ts.completed" +)