George Lambert
e04fe6e2c4
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.
458 lines
11 KiB
Go
458 lines
11 KiB
Go
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
|
|
}
|