From e04fe6e2c4d70d01315f0e281fa72711f9144105 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Tue, 29 Sep 2026 12:11:47 -0400 Subject: [PATCH] feat: embedded NATS with VeraeTime handlers and stdio MCP Start in-process nats-server on VERAE_NATS_LISTEN (default 127.0.0.1:4223; empty disables). Queue-subscribe verae.ts.request|verify|hash.report with ledger/veraetime routing and authctx.Resolve from authorization/access_token. Publish verae.ts.completed on core NATS only (no JetStream). Wire embed into main alongside Streamable HTTP; stdio MCP via go-sdk StdioTransport. --- README.md | 5 +- cmd/verae-mcp-timestamping-server/main.go | 16 + go.mod | 19 +- go.sum | 35 +- internal/embednats/handlers.go | 458 ++++++++++++++++++++++ internal/embednats/server.go | 123 ++++++ internal/embednats/server_test.go | 100 +++++ internal/server/server.go | 9 +- 8 files changed, 743 insertions(+), 22 deletions(-) create mode 100644 internal/embednats/handlers.go create mode 100644 internal/embednats/server.go create mode 100644 internal/embednats/server_test.go diff --git a/README.md b/README.md index ea71242..ea9f1ff 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ cd verae-mcp-timestamping-server-go go build -o bin/verae-mcp-timestamping-server ./cmd/verae-mcp-timestamping-server ``` -Requires Go 1.25+. Design: `docs/superpowers/specs/2026-09-29-go-mcp-timestamping-server-design.md`. +Requires Go 1.26+. Design: `docs/superpowers/specs/2026-09-29-go-mcp-timestamping-server-design.md`. ## Usage @@ -30,10 +30,11 @@ Requires Go 1.25+. Design: `docs/superpowers/specs/2026-09-29-go-mcp-timestampin VERAE_MCP_TRANSPORT=streamable-http \ VERAE_MCP_HOST=127.0.0.1 \ VERAE_MCP_PORT=8001 \ +VERAE_NATS_LISTEN=127.0.0.1:4223 \ ./bin/verae-mcp-timestamping-server ``` -Point clients at `http://127.0.0.1:8001/mcp`. Global stamps use `https://api.veraetime.net` with auth from `~/.verae/auth.properties` (or `VERAE_AUTH_PROPERTIES`). +Point clients at `http://127.0.0.1:8001/mcp`. Global stamps use `https://api.veraetime.net` with auth from `~/.verae/auth.properties` (or `VERAE_AUTH_PROPERTIES`). Embedded NATS (default `127.0.0.1:4223`; set `VERAE_NATS_LISTEN=` empty to disable) serves `verae.ts.*` with the same ledger/VeraeTime routing; job-complete is core `verae.ts.completed` only. For stdio MCP: `-transport stdio` or `VERAE_MCP_TRANSPORT=stdio`. ## Contributing diff --git a/cmd/verae-mcp-timestamping-server/main.go b/cmd/verae-mcp-timestamping-server/main.go index 21f8ff9..83d6dbe 100644 --- a/cmd/verae-mcp-timestamping-server/main.go +++ b/cmd/verae-mcp-timestamping-server/main.go @@ -10,6 +10,7 @@ import ( "os/signal" "syscall" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/embednats" "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/ledger" "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/server" "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/tools" @@ -32,6 +33,21 @@ func main() { global := veraetime.NewClient(cfg.GlobalURL, veraetime.WithAuthPath(cfg.AuthPath)) keeper := veraetime.NewKeeper(global, cfg.AuthPath) + emb, err := embednats.Start(embednats.Options{ + Listen: cfg.NATSListen, + Ledger: led, + Global: global, + Keeper: keeper, + RequireBearer: cfg.RequireBearer, + }) + if err != nil { + log.Fatalf("embednats: %v", err) + } + if emb != nil { + defer emb.Shutdown() + log.Printf("embedded NATS listening on %s", emb.ClientURL()) + } + switch cfg.Transport { case "streamable-http", "http": srv, err := server.New(server.Options{ diff --git a/go.mod b/go.mod index ec208a0..84810c7 100644 --- a/go.mod +++ b/go.mod @@ -1,23 +1,28 @@ module git.georgelambert.org/marchon/verae-mcp-timestamping-server-go -go 1.25.0 +go 1.26.0 require ( github.com/modelcontextprotocol/go-sdk v1.8.0 - github.com/nats-io/nats.go v1.39.1 + github.com/nats-io/nats-server/v2 v2.15.0 + github.com/nats-io/nats.go v1.51.0 ) require ( + github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/google/jsonschema-go v0.4.3 // indirect - github.com/klauspost/compress v1.17.11 // indirect - github.com/nats-io/nkeys v0.4.9 // indirect + github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.2 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/segmentio/asm v1.1.3 // indirect github.com/segmentio/encoding v0.5.4 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect - golang.org/x/crypto v0.31.0 // indirect + golang.org/x/crypto v0.57.0 // indirect golang.org/x/oauth2 v0.35.0 // indirect golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.41.0 // indirect - golang.org/x/time v0.15.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.16.0 // indirect ) diff --git a/go.sum b/go.sum index c78d84b..dd43c97 100644 --- a/go.sum +++ b/go.sum @@ -1,17 +1,27 @@ +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0= github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= -github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc= -github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/modelcontextprotocol/go-sdk v1.8.0 h1:KIvahhYqwtbeniWVPs3TcXEA7b8jEtwfBpOTAI+Urx4= github.com/modelcontextprotocol/go-sdk v1.8.0/go.mod h1:dL7u98E/zjJTGzEq+j30jQ8K2k1mb6LeAH4inEcSGts= -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/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= +github.com/nats-io/nats-server/v2 v2.15.0 h1:M99yf0y05rTr46/qc/Is6ZAowI58Ryp2SjufLCUeVJc= +github.com/nats-io/nats-server/v2 v2.15.0/go.mod h1:5qLF4CDGzZVFt//3fUrY1ePpwbi05r7QHPNroSUtolk= +github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI= +github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= @@ -20,15 +30,16 @@ github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfv github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= -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/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= -golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= diff --git a/internal/embednats/handlers.go b/internal/embednats/handlers.go new file mode 100644 index 0000000..4340847 --- /dev/null +++ b/internal/embednats/handlers.go @@ -0,0 +1,458 @@ +package embednats + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/authctx" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/ledger" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/veraetime" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/pkg/digest" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/pkg/tsnats" + "github.com/nats-io/nats.go" +) + +type handlers struct { + nc *nats.Conn + ledger *ledger.FileLedger + global *veraetime.Client + keeper *veraetime.Keeper + requireBearer bool +} + +func (h *handlers) subscribe() error { + if _, err := h.nc.QueueSubscribe(tsnats.SubjectRequest, queueGroup, h.onRequest); err != nil { + return fmt.Errorf("embednats: subscribe %s: %w", tsnats.SubjectRequest, err) + } + if _, err := h.nc.QueueSubscribe(tsnats.SubjectVerify, queueGroup, h.onVerify); err != nil { + return fmt.Errorf("embednats: subscribe %s: %w", tsnats.SubjectVerify, err) + } + if _, err := h.nc.QueueSubscribe(tsnats.SubjectReport, queueGroup, h.onReport); err != nil { + return fmt.Errorf("embednats: subscribe %s: %w", tsnats.SubjectReport, err) + } + return h.nc.Flush() +} + +func (h *handlers) onRequest(msg *nats.Msg) { + var req map[string]any + if err := json.Unmarshal(msg.Data, &req); err != nil { + replyErr(msg, "invalid json") + return + } + if err := digest.AssertDigestOnly(req); err != nil { + replyErr(msg, err.Error()) + return + } + hash, _ := req["hash"].(string) + if hash == "" { + if s, ok := req["sha256"].(string); ok { + hash = s + } + } + normalized, err := digest.NormalizeSHA256(hash) + if err != nil { + replyErr(msg, err.Error()) + return + } + scope, _ := req["scope"].(string) + scope = strings.TrimSpace(scope) + if scope == "" { + scope = "local" + } + cubeID, _ := req["cube_id"].(string) + orgID, _ := req["org_id"].(string) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + switch scope { + case "global": + tok, err := h.resolveBearer(ctx, req) + if err != nil { + replyErr(msg, err.Error()) + return + } + if h.global == nil { + replyErr(msg, "global VeraeTime client not configured") + return + } + rec, err := h.global.Stamp(ctx, normalized, tok) + if err != nil { + replyErr(msg, err.Error()) + return + } + if cubeID != "" { + rec.CubeID = cubeID + } + if orgID != "" { + rec.OrgID = orgID + } + out := []tsnats.Receipt{toTSNATSFromGlobal(rec)} + replyJSON(msg, out) + h.publishCompleted(out[0]) + case "local", "shared": + if h.ledger == nil { + replyErr(msg, "ledger not configured") + return + } + recs, err := h.ledger.Stamp(normalized, scope, cubeID, orgID) + if err != nil { + replyErr(msg, err.Error()) + return + } + out := toTSNATSFromLocalList(recs) + replyJSON(msg, out) + if len(out) > 0 { + h.publishCompleted(out[0]) + } + default: + replyErr(msg, "scope must be local, shared, or global") + } +} + +func (h *handlers) onVerify(msg *nats.Msg) { + var req map[string]any + if err := json.Unmarshal(msg.Data, &req); err != nil { + replyErr(msg, "invalid json") + return + } + receiptID, _ := req["receipt_id"].(string) + want, _ := req["want_scope"].(string) + want = strings.TrimSpace(want) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + useGlobal := want == "global" || looksLikeCID(receiptID) + if useGlobal { + h.replyVerifyGlobal(ctx, msg, req, receiptID) + return + } + if h.ledger != nil { + vr := h.ledger.Verify(receiptID, want) + if vr.Valid { + replyJSON(msg, verifyPayload(vr.Valid, vr.Error, vr.Receipt)) + return + } + if want == "" && h.global != nil { + if h.tryVerifyGlobal(ctx, msg, req, receiptID) { + return + } + } + replyJSON(msg, verifyPayload(vr.Valid, vr.Error, vr.Receipt)) + return + } + h.replyVerifyGlobal(ctx, msg, req, receiptID) +} + +func (h *handlers) tryVerifyGlobal(ctx context.Context, msg *nats.Msg, req map[string]any, receiptID string) bool { + tok, err := h.resolveBearer(ctx, req) + if err != nil || h.global == nil { + return false + } + vr, err := h.global.Verify(ctx, receiptID, tok) + if err != nil || !vr.Valid { + return false + } + replyJSON(msg, verifyPayloadGlobal(vr)) + return true +} + +func (h *handlers) replyVerifyGlobal(ctx context.Context, msg *nats.Msg, req map[string]any, receiptID string) { + if h.global == nil { + replyJSON(msg, map[string]any{"valid": false, "error": "global VeraeTime client not configured"}) + return + } + tok, err := h.resolveBearer(ctx, req) + if err != nil { + replyErr(msg, err.Error()) + return + } + vr, err := h.global.Verify(ctx, receiptID, tok) + if err != nil { + replyErr(msg, err.Error()) + return + } + replyJSON(msg, verifyPayloadGlobal(vr)) +} + +func (h *handlers) onReport(msg *nats.Msg) { + var req map[string]any + if err := json.Unmarshal(msg.Data, &req); err != nil { + replyErr(msg, "invalid json") + return + } + if err := digest.AssertDigestOnly(req); err != nil { + replyErr(msg, err.Error()) + return + } + hash, _ := req["hash"].(string) + if hash == "" { + if s, ok := req["sha256"].(string); ok { + hash = s + } + } + normalized, err := digest.NormalizeSHA256(hash) + if err != nil { + replyErr(msg, err.Error()) + return + } + orgID, _ := req["org_id"].(string) + want, _ := req["want_scope"].(string) + want = strings.TrimSpace(want) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if want == "global" { + recs, err := h.hashReportGlobal(ctx, req, normalized) + if err != nil { + replyErr(msg, err.Error()) + return + } + replyJSON(msg, recs) + return + } + + var local []tsnats.Receipt + if h.ledger != nil { + recs, err := h.ledger.HashReport(normalized, orgID, want) + if err != nil { + replyErr(msg, err.Error()) + return + } + local = toTSNATSFromLocalList(recs) + } + if want != "" { + replyJSON(msg, local) + return + } + central, err := h.hashReportGlobal(ctx, req, normalized) + if err != nil { + replyJSON(msg, local) + return + } + seen := make(map[string]struct{}, len(local)) + for _, r := range local { + seen[r.ReceiptID] = struct{}{} + } + merged := append([]tsnats.Receipt{}, local...) + for _, r := range central { + if _, ok := seen[r.ReceiptID]; ok { + continue + } + merged = append(merged, r) + } + replyJSON(msg, merged) +} + +func (h *handlers) hashReportGlobal(ctx context.Context, req map[string]any, dig string) ([]tsnats.Receipt, error) { + if h.global == nil { + return nil, fmt.Errorf("global VeraeTime client not configured") + } + tok, err := h.resolveBearer(ctx, req) + if err != nil { + return nil, err + } + recs, err := h.global.HashReport(ctx, dig, tok) + if err != nil { + return nil, err + } + out := make([]tsnats.Receipt, 0, len(recs)) + for i := range recs { + out = append(out, toTSNATSFromGlobal(&recs[i])) + } + return out, nil +} + +func (h *handlers) resolveBearer(ctx context.Context, req map[string]any) (string, error) { + tok := messageBearer(req) + return authctx.Resolve(ctx, tok, h.requireBearer, h.keeper) +} + +func (h *handlers) publishCompleted(rec tsnats.Receipt) { + if h.nc == nil || rec.ReceiptID == "" { + return + } + ev := tsnats.CompletedEvent{ + JobID: rec.ReceiptID, + ReceiptID: rec.ReceiptID, + Hash: rec.Hash, + } + raw, err := json.Marshal(ev) + if err != nil { + return + } + _ = h.nc.Publish(tsnats.SubjectCompleted, raw) +} + +func messageBearer(req map[string]any) string { + if req == nil { + return "" + } + if v, ok := req["authorization"].(string); ok { + if t := stripBearer(v); t != "" { + return t + } + } + if v, ok := req["access_token"].(string); ok { + if t := stripBearer(v); t != "" { + return t + } + } + return "" +} + +func stripBearer(s string) string { + s = strings.TrimSpace(s) + const prefix = "bearer " + if len(s) > len(prefix) && strings.EqualFold(s[:len(prefix)], prefix) { + return strings.TrimSpace(s[len(prefix):]) + } + return s +} + +func replyJSON(msg *nats.Msg, v any) { + if msg == nil || msg.Reply == "" { + return + } + raw, err := json.Marshal(v) + if err != nil { + replyErr(msg, "marshal: "+err.Error()) + return + } + _ = msg.Respond(raw) +} + +func replyErr(msg *nats.Msg, errMsg string) { + raw, _ := json.Marshal(map[string]string{"error": errMsg}) + _ = msg.Respond(raw) +} + +func verifyPayload(valid bool, errMsg string, rec *ledger.Receipt) map[string]any { + if valid && rec != nil { + out := map[string]any{ + "valid": true, + "receipt_id": rec.ReceiptID, + "hash": rec.Hash, + "scope": rec.Scope, + "created_at": rec.CreatedAt, + "merkle": rec.Merkle, + } + if rec.CubeID != "" { + out["cube_id"] = rec.CubeID + } + if rec.OrgID != "" { + out["org_id"] = rec.OrgID + } + if rec.RestampOf != "" { + out["restamp_of"] = rec.RestampOf + } + return out + } + out := map[string]any{"valid": false} + if errMsg != "" { + out["error"] = errMsg + } + if rec != nil { + out["receipt_id"] = rec.ReceiptID + out["hash"] = rec.Hash + out["scope"] = rec.Scope + out["created_at"] = rec.CreatedAt + out["merkle"] = rec.Merkle + } + return out +} + +func verifyPayloadGlobal(vr veraetime.VerifyResult) map[string]any { + if vr.Valid && vr.Receipt != nil { + r := vr.Receipt + out := map[string]any{ + "valid": true, + "receipt_id": r.ReceiptID, + "hash": r.Hash, + "scope": r.Scope, + "created_at": r.CreatedAt, + "merkle": r.Merkle, + } + if r.CubeID != "" { + out["cube_id"] = r.CubeID + } + if r.OrgID != "" { + out["org_id"] = r.OrgID + } + if r.RestampOf != "" { + out["restamp_of"] = r.RestampOf + } + return out + } + out := map[string]any{"valid": false} + if vr.Error != "" { + out["error"] = vr.Error + } + return out +} + +func toTSNATSFromLocal(r ledger.Receipt) tsnats.Receipt { + return tsnats.Receipt{ + ReceiptID: r.ReceiptID, + Hash: r.Hash, + Scope: r.Scope, + CubeID: r.CubeID, + OrgID: r.OrgID, + CreatedAt: r.CreatedAt, + Merkle: r.Merkle, + RestampOf: r.RestampOf, + } +} + +func toTSNATSFromLocalList(recs []ledger.Receipt) []tsnats.Receipt { + out := make([]tsnats.Receipt, len(recs)) + for i := range recs { + out[i] = toTSNATSFromLocal(recs[i]) + } + return out +} + +func toTSNATSFromGlobal(r *veraetime.Receipt) tsnats.Receipt { + if r == nil { + return tsnats.Receipt{} + } + return tsnats.Receipt{ + ReceiptID: r.ReceiptID, + Hash: r.Hash, + Scope: r.Scope, + CubeID: r.CubeID, + OrgID: r.OrgID, + CreatedAt: r.CreatedAt, + Merkle: r.Merkle, + RestampOf: r.RestampOf, + } +} + +// looksLikeCID mirrors tools.looksLikeCID (CID-like → VeraeTime). +func looksLikeCID(receiptID string) bool { + id := strings.TrimSpace(receiptID) + if id == "" { + return false + } + lower := strings.ToLower(id) + if strings.HasPrefix(lower, "qm") || + strings.HasPrefix(lower, "bafy") || + strings.HasPrefix(lower, "bafk") || + strings.HasPrefix(lower, "bagu") { + return true + } + if len(id) == 64 { + for _, c := range lower { + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return true + } + } + return false + } + return true +} diff --git a/internal/embednats/server.go b/internal/embednats/server.go new file mode 100644 index 0000000..87b80d2 --- /dev/null +++ b/internal/embednats/server.go @@ -0,0 +1,123 @@ +package embednats + +import ( + "fmt" + "net" + "strconv" + "strings" + "time" + + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/ledger" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/veraetime" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" +) + +const ( + // DefaultListen is used when VERAE_NATS_LISTEN is unset. + DefaultListen = "127.0.0.1:4223" + queueGroup = "verae-ts" +) + +// Options configures the in-process NATS server and subject handlers. +type Options struct { + // Listen is host:port for client connections. Empty disables embed. + // Port 0 picks an ephemeral port (tests). + Listen string + Ledger *ledger.FileLedger + Global *veraetime.Client + Keeper *veraetime.Keeper + RequireBearer bool +} + +// Server is an embedded nats-server with timestamp subject handlers. +type Server struct { + ns *server.Server + nc *nats.Conn +} + +// Start runs an in-process nats-server on opts.Listen and registers handlers. +// When Listen is empty, returns (nil, nil). +func Start(opts Options) (*Server, error) { + listen := strings.TrimSpace(opts.Listen) + if listen == "" { + return nil, nil + } + host, port, err := splitHostPort(listen) + if err != nil { + return nil, err + } + + ns, err := server.NewServer(&server.Options{ + Host: host, + Port: port, + NoLog: true, + NoSigs: true, + }) + if err != nil { + return nil, fmt.Errorf("embednats: NewServer: %w", err) + } + go ns.Start() + if !ns.ReadyForConnections(5 * time.Second) { + ns.Shutdown() + return nil, fmt.Errorf("embednats: not ready for connections on %s", listen) + } + + nc, err := nats.Connect(ns.ClientURL()) + if err != nil { + ns.Shutdown() + return nil, fmt.Errorf("embednats: connect: %w", err) + } + + h := &handlers{ + nc: nc, + ledger: opts.Ledger, + global: opts.Global, + keeper: opts.Keeper, + requireBearer: opts.RequireBearer, + } + if err := h.subscribe(); err != nil { + nc.Close() + ns.Shutdown() + return nil, err + } + + return &Server{ns: ns, nc: nc}, nil +} + +// ClientURL returns the nats:// URL clients should dial. +func (s *Server) ClientURL() string { + if s == nil || s.ns == nil { + return "" + } + return s.ns.ClientURL() +} + +// Shutdown stops handlers and the embedded server. +func (s *Server) Shutdown() { + if s == nil { + return + } + if s.nc != nil { + s.nc.Close() + } + if s.ns != nil { + s.ns.Shutdown() + s.ns.WaitForShutdown() + } +} + +func splitHostPort(listen string) (string, int, error) { + host, portStr, err := net.SplitHostPort(listen) + if err != nil { + return "", 0, fmt.Errorf("embednats: listen %q: %w", listen, err) + } + port, err := strconv.Atoi(portStr) + if err != nil { + return "", 0, fmt.Errorf("embednats: listen port %q: %w", portStr, err) + } + if host == "" { + host = "127.0.0.1" + } + return host, port, nil +} diff --git a/internal/embednats/server_test.go b/internal/embednats/server_test.go new file mode 100644 index 0000000..b11cd50 --- /dev/null +++ b/internal/embednats/server_test.go @@ -0,0 +1,100 @@ +package embednats_test + +import ( + "context" + "path/filepath" + "sync" + "testing" + "time" + + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/embednats" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/internal/ledger" + "git.georgelambert.org/marchon/verae-mcp-timestamping-server-go/pkg/tsnats" +) + +const testDigest = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + +func TestEmbed_StampLocalViaTSNATS(t *testing.T) { + led, err := ledger.Open(filepath.Join(t.TempDir(), "ledger.json")) + if err != nil { + t.Fatal(err) + } + + emb, err := embednats.Start(embednats.Options{ + Listen: "127.0.0.1:0", + Ledger: led, + }) + if err != nil { + t.Fatalf("Start: %v", err) + } + defer emb.Shutdown() + + url := emb.ClientURL() + if url == "" { + t.Fatal("empty ClientURL") + } + + client, err := tsnats.Connect(url, tsnats.WithTimeout(3*time.Second)) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer client.Close() + + var ( + mu sync.Mutex + got tsnats.CompletedEvent + done = make(chan struct{}) + ) + sub, err := client.SubscribeCompleted(context.Background(), func(ev tsnats.CompletedEvent) { + mu.Lock() + got = ev + mu.Unlock() + select { + case <-done: + default: + close(done) + } + }) + if err != nil { + t.Fatalf("SubscribeCompleted: %v", err) + } + defer sub.Unsubscribe() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + receipts, err := client.Stamp(ctx, testDigest, tsnats.StampOptions{Scope: "local"}) + if err != nil { + t.Fatalf("Stamp: %v", err) + } + if len(receipts) != 1 { + t.Fatalf("receipts=%+v", receipts) + } + if receipts[0].Hash != testDigest || receipts[0].Scope != "local" || receipts[0].ReceiptID == "" { + t.Fatalf("receipt=%+v", receipts[0]) + } + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("expected verae.ts.completed publish") + } + mu.Lock() + defer mu.Unlock() + if got.ReceiptID != receipts[0].ReceiptID { + t.Fatalf("completed receipt_id=%q want %q", got.ReceiptID, receipts[0].ReceiptID) + } + if got.Hash != "" && got.Hash != testDigest { + t.Fatalf("completed hash=%q", got.Hash) + } +} + +func TestEmbed_DisabledWhenListenEmpty(t *testing.T) { + emb, err := embednats.Start(embednats.Options{Listen: ""}) + if err != nil { + t.Fatalf("Start empty: %v", err) + } + if emb != nil { + t.Fatal("expected nil embed when Listen empty") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 7d1b9de..c9e24a7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -106,12 +106,15 @@ type Config struct { GlobalURL string AuthPath string LedgerPath string + NATSListen string CORS bool } // LoadConfig merges flag defaults with environment overrides. // Env wins when set: VERAE_MCP_TRANSPORT, VERAE_MCP_HOST, VERAE_MCP_PORT, -// VERAE_MCP_REQUIRE_BEARER, VERAE_GLOBAL_URL, VERAE_AUTH_PROPERTIES, VERAE_LEDGER_PATH, VERAE_MCP_CORS. +// VERAE_MCP_REQUIRE_BEARER, VERAE_GLOBAL_URL, VERAE_AUTH_PROPERTIES, VERAE_LEDGER_PATH, +// VERAE_NATS_LISTEN, VERAE_MCP_CORS. +// VERAE_NATS_LISTEN defaults to 127.0.0.1:4223; set empty to disable embedded NATS. func LoadConfig(transport, host string, port int) Config { cfg := Config{ Transport: transport, @@ -120,6 +123,7 @@ func LoadConfig(transport, host string, port int) Config { GlobalURL: veraetime.DefaultBaseURL, AuthPath: veraetime.DefaultAuthPath(), LedgerPath: ledger.DefaultPath(), + NATSListen: "127.0.0.1:4223", } if v := strings.TrimSpace(os.Getenv("VERAE_MCP_TRANSPORT")); v != "" { cfg.Transport = v @@ -141,6 +145,9 @@ func LoadConfig(transport, host string, port int) Config { if v := strings.TrimSpace(os.Getenv("VERAE_LEDGER_PATH")); v != "" { cfg.LedgerPath = v } + if _, set := os.LookupEnv("VERAE_NATS_LISTEN"); set { + cfg.NATSListen = strings.TrimSpace(os.Getenv("VERAE_NATS_LISTEN")) + } cfg.RequireBearer = envTruthy("VERAE_MCP_REQUIRE_BEARER") cfg.CORS = envTruthy("VERAE_MCP_CORS") if cfg.Transport == "" {