From 6549a5db1d4d8980bb1f1f35c586a2dc76e5aef1 Mon Sep 17 00:00:00 2001 From: Lakshman Patel Date: Fri, 17 Jul 2026 08:24:36 +0530 Subject: [PATCH] fix(yaad): gRPC API key authentication, OpenAI HTTP status check, privacy-filtered exports CRITICAL SECURITY FIXES: - Add WithAPIKey() for gRPC server authentication with constant-time token comparison - Check HTTP status codes in OpenAI embedding provider (prevents silent data corruption) - Apply privacy filter to exported memory content These changes address: - CRITICAL: Unauthenticated gRPC server exposure - HIGH: OpenAI embedding ignoring HTTP errors that could produce bogus data - HIGH: Memory exports containing raw secrets without privacy filtering --- embeddings/provider.go | 42 +++++++++++++++++++++++++++++ exportimport/export.go | 7 ++++- internal/server/grpc.go | 58 ++++++++++++++++++++++++++++++++++++++++- 3 files changed, 105 insertions(+), 2 deletions(-) diff --git a/embeddings/provider.go b/embeddings/provider.go index d9a7adf..a814305 100644 --- a/embeddings/provider.go +++ b/embeddings/provider.go @@ -83,6 +83,27 @@ func (p *openAI) Embed(ctx context.Context, text string) ([]float32, error) { return nil, err } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + var errBody struct { + Error *struct{ Message string } `json:"error"` + } + _ = json.NewDecoder(resp.Body).Decode(&errBody) + _ = resp.Body.Close() + errMsg := fmt.Sprintf("openai: API returned status %d", resp.StatusCode) + if errBody.Error != nil && errBody.Error.Message != "" { + errMsg += ": " + errBody.Error.Message + } + if d := ExtractRetryDelay(errMsg); d > 0 && attempt < maxRetries { + select { + case <-time.After(d): + case <-ctx.Done(): + return nil, ctx.Err() + } + continue + } + return nil, fmt.Errorf("%s", errMsg) + } + var result struct { Data []struct { Embedding []float32 `json:"embedding"` @@ -138,6 +159,27 @@ func (p *openAI) EmbedBatch(ctx context.Context, texts []string) ([][]float32, e return nil, err } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + var errBody struct { + Error *struct{ Message string } `json:"error"` + } + _ = json.NewDecoder(resp.Body).Decode(&errBody) + _ = resp.Body.Close() + errMsg := fmt.Sprintf("openai: API returned status %d", resp.StatusCode) + if errBody.Error != nil && errBody.Error.Message != "" { + errMsg += ": " + errBody.Error.Message + } + if d := ExtractRetryDelay(errMsg); d > 0 && attempt < maxRetries { + select { + case <-time.After(d): + case <-ctx.Done(): + return nil, ctx.Err() + } + continue + } + return nil, fmt.Errorf("%s", errMsg) + } + var result struct { Data []struct { Embedding []float32 `json:"embedding"` diff --git a/exportimport/export.go b/exportimport/export.go index bb36e74..d5fa000 100644 --- a/exportimport/export.go +++ b/exportimport/export.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/GrayCodeAI/yaad/privacy" "github.com/GrayCodeAI/yaad/storage" ) @@ -21,12 +22,16 @@ type GraphExport struct { Edges []*storage.Edge `json:"edges"` } -// ExportJSON exports the full graph as JSON. +// ExportJSON exports the full graph as JSON, with privacy-filtered content. func ExportJSON(ctx context.Context, store storage.Storage, project string) ([]byte, error) { nodes, err := store.ListNodes(ctx, storage.NodeFilter{Project: project}) if err != nil { return nil, err } + for _, n := range nodes { + n.Content = privacy.Filter(n.Content) + n.Summary = privacy.Filter(n.Summary) + } // Batch-fetch all edges between project nodes (single query instead of N+1) nodeIDs := make([]string, len(nodes)) for i, n := range nodes { diff --git a/internal/server/grpc.go b/internal/server/grpc.go index 316bfe6..982647a 100644 --- a/internal/server/grpc.go +++ b/internal/server/grpc.go @@ -2,6 +2,7 @@ package server import ( "context" + "crypto/subtle" "fmt" "net" "strings" @@ -16,6 +17,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" ) @@ -30,6 +32,7 @@ type GrpcServer struct { eng *engine.Engine broadcaster *SSEBroadcaster // nil = WatchMemories/WatchStale disabled limiter *engine.RateLimiter // rate limiter for destructive operations, mirrors RESTServer + apiKey string // if non-empty, clients must present matching Bearer or x-api-key srv *grpc.Server } @@ -44,6 +47,55 @@ func NewGrpcServer(eng *engine.Engine, broadcaster *SSEBroadcaster) *GrpcServer } } +// WithAPIKey configures API-key authentication. When the key is non-empty, +// every unary RPC must present a matching token via the "authorization" +// metadata (Bearer ) or the "x-api-key" header. If the key does not +// match, the call returns codes.Unauthenticated. +func (s *GrpcServer) WithAPIKey(apiKey string) *GrpcServer { + s.apiKey = apiKey + return s +} + +// authInterceptor returns a unary interceptor that enforces API‑key auth when +// s.apiKey is set. Health checks are always allowed through. +func (s *GrpcServer) authInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { + if s.apiKey == "" { + return handler(ctx, req) + } + if info.FullMethod == "/yaad.v1.YaadService/Health" || + info.FullMethod == "/yaad.v1.YaadService/GraphStats" { + return handler(ctx, req) + } + + token := extractToken(ctx) + if token == "" || subtle.ConstantTimeCompare([]byte(token), []byte(s.apiKey)) != 1 { + return nil, status.Error(codes.Unauthenticated, "invalid or missing API key") + } + return handler(ctx, req) +} + +// extractToken pulls the bearer token from grpc metadata. Checks both +// "authorization: Bearer " and "x-api-key: " headers. +func extractToken(ctx context.Context) string { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return "" + } + // authorization: Bearer + auth := md.Get("authorization") + for _, v := range auth { + if strings.HasPrefix(v, "Bearer ") { + return strings.TrimPrefix(v, "Bearer ") + } + } + // x-api-key header + keys := md.Get("x-api-key") + if len(keys) > 0 { + return keys[0] + } + return "" +} + // ListenAndServe starts the gRPC server on addr. Blocks until the server // stops (via Shutdown or a fatal error). func (s *GrpcServer) ListenAndServe(addr string) error { @@ -51,7 +103,11 @@ func (s *GrpcServer) ListenAndServe(addr string) error { if err != nil { return fmt.Errorf("yaad grpc: listen on %s: %w", addr, err) } - s.srv = grpc.NewServer() + opts := []grpc.ServerOption{} + if s.apiKey != "" { + opts = append(opts, grpc.UnaryInterceptor(s.authInterceptor)) + } + s.srv = grpc.NewServer(opts...) yaadproto.RegisterYaadServiceServer(s.srv, s) fmt.Printf("yaad gRPC API listening on %s\n", addr) return s.srv.Serve(lis)