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.
This commit is contained in:
parent
d379931bd1
commit
e04fe6e2c4
8 changed files with 743 additions and 22 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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{
|
||||
|
|
|
|||
19
go.mod
19
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
|
||||
)
|
||||
|
|
|
|||
35
go.sum
35
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=
|
||||
|
|
|
|||
458
internal/embednats/handlers.go
Normal file
458
internal/embednats/handlers.go
Normal file
|
|
@ -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
|
||||
}
|
||||
123
internal/embednats/server.go
Normal file
123
internal/embednats/server.go
Normal file
|
|
@ -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
|
||||
}
|
||||
100
internal/embednats/server_test.go
Normal file
100
internal/embednats/server_test.go
Normal file
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
|
|
@ -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 == "" {
|
||||
|
|
|
|||
Loading…
Reference in a new issue