Remove internal notification caller tokens

This commit is contained in:
Stephan D committed 2026-09-12 23:59:30 +02:00
1 parent 1fe173397f
commit caad7a3d5b
27 files changed
+112 -165

No files matched your search

+1 -2
View File
@@ -22,8 +22,7 @@ subject, text and HTML, bounded timeouts and safe error classification.
Requires Go 1.27, protoc, an SXCHNGE MongoDB database and a NATS server with
JetStream enabled. Copy `config.example.yml` to an untracked `config.yml` here.
Set `SXCHNGE_MONGO_URI`, `SXCHNGE_NATS_URL`, a unique token of at least 32 bytes for
each configured caller, and the selected email provider's API key environment
Set `SXCHNGE_MONGO_URI`, `SXCHNGE_NATS_URL` and the selected email provider's API key environment
variable. Configure a verified sender address and template
catalog. Paths are relative to the process working directory.
+9 -19
View File
@@ -4,7 +4,6 @@ import (
"bytes"
"io"
"os"
"strings"
"git.sendico.io/tech/sxchnge/notifications/bus"
"git.sendico.io/tech/sxchnge/notifications/email"
@@ -15,15 +14,14 @@ import (
)
type Config struct {
Queue bus.Config `yaml:"queue"`
Messaging messaging.Config `yaml:"messaging"`
Listen string `yaml:"listen"`
TLSCert string `yaml:"tls_cert"`
TLSKey string `yaml:"tls_key"`
Database db.Config `yaml:"database"`
Templates string `yaml:"templates"`
Callers map[string]string `yaml:"callers"` // caller -> token environment variable
Email email.Config `yaml:"email"`
Queue bus.Config `yaml:"queue"`
Messaging messaging.Config `yaml:"messaging"`
Listen string `yaml:"listen"`
TLSCert string `yaml:"tls_cert"`
TLSKey string `yaml:"tls_key"`
Database db.Config `yaml:"database"`
Templates string `yaml:"templates"`
Email email.Config `yaml:"email"`
}
func Load(path string) (Config, error) {
@@ -40,7 +38,7 @@ func Load(path string) (Config, error) {
if err = decoder.Decode(new(any)); err != io.EOF {
return c, merrors.InvalidArgument("unexpected configuration content")
}
if c.Listen == "" || c.Database.Driver != db.Mongo || c.Messaging.Driver != messaging.BBNats || c.Templates == "" || len(c.Callers) == 0 {
if c.Listen == "" || c.Database.Driver != db.Mongo || c.Messaging.Driver != messaging.BBNats || c.Templates == "" {
return c, merrors.InvalidArgument("missing notification configuration")
}
if (c.TLSCert == "") != (c.TLSKey == "") {
@@ -49,13 +47,5 @@ func Load(path string) (Config, error) {
if err = c.Queue.Validate(); err != nil {
return c, err
}
seen := map[string]bool{}
for caller, env := range c.Callers {
token := os.Getenv(env)
if strings.TrimSpace(caller) == "" || len(strings.TrimSpace(token)) < 32 || seen[token] {
return c, merrors.InvalidArgument("caller credentials must be unique and at least 32 bytes")
}
seen[token] = true
}
return c, nil
}
-8
View File
@@ -3,12 +3,10 @@ package app
import (
"os"
"path/filepath"
"strings"
"testing"
)
func TestExampleConfigAndSecrets(t *testing.T) {
t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", strings.Repeat("a", 32))
c, err := Load("../config.example.yml")
if err != nil {
t.Fatal(err)
@@ -16,11 +14,6 @@ func TestExampleConfigAndSecrets(t *testing.T) {
if c.Email.Driver != "resend" || c.Messaging.Driver != "NATS" || c.Database.Driver != "mongodb" {
t.Fatal("wrong driver configuration")
}
t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", "")
if _, err = Load("../config.example.yml"); err == nil {
t.Fatal("missing credentials accepted")
}
t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", strings.Repeat("a", 32))
data, err := os.ReadFile("../config.example.yml")
if err != nil {
t.Fatal(err)
@@ -35,7 +28,6 @@ func TestExampleConfigAndSecrets(t *testing.T) {
}
func TestPlaintextConfigAndIncompleteTLS(t *testing.T) {
t.Setenv("SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN", strings.Repeat("a", 32))
c, err := Load("../config.example.yml")
if err != nil {
t.Fatal(err)
+1 -6
View File
@@ -4,7 +4,6 @@ import (
"context"
"crypto/tls"
"net"
"os"
"sync"
"time"
@@ -63,11 +62,7 @@ func Run(ctx context.Context, c Config, logger mlogger.Logger) error {
return err
}
defer broker.Close()
tokens := map[string]string{}
for caller, env := range c.Callers {
tokens[caller] = os.Getenv(env)
}
server := grpc.NewServer(append(options, grpc.UnaryInterceptor(transport.Authenticate(tokens)), grpc.MaxRecvMsgSize(128<<10), grpc.MaxConcurrentStreams(64))...)
server := grpc.NewServer(append(options, grpc.MaxRecvMsgSize(128<<10), grpc.MaxConcurrentStreams(64))...)
pb.RegisterNotificationsServiceServer(server, transport.New(svc))
listener, err := net.Listen("tcp", c.Listen)
if err != nil {
-2
View File
@@ -12,8 +12,6 @@ messaging:
max_reconnects: -1
reconnect_wait: 2
templates: "templates.example.json"
callers:
product: "SXCHNGE_NOTIFICATIONS_PRODUCT_TOKEN"
email:
driver: resend # sendgrid or resend
settings:
+3 -3
View File
@@ -20,14 +20,14 @@ Internal Go consumers use the module
`git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1`. Within this
checkout, use a local replace to this directory and generate before building.
Calls require TLS and `authorization: Bearer <service-token>` metadata. The service
maps tokens to configured caller identities. Callers authorize user/destination
Calls use the trusted private network without caller authentication. The service
uses the existing product storage namespace. Callers authorize user/destination
ownership and preferences before sending; this contract is not a public user API.
Parameters are string values keyed by template parameter name. The entire request
is validated before persistence. An identical idempotency key/request returns the
original resource, even after completion or expiry. A changed request conflicts.
Keys are scoped to the authenticated caller and retained at least seven days;
Keys are scoped to the product namespace and retained at least seven days;
Mongo TTL deletion is asynchronous. Expiry defaults to 24 hours and cannot exceed
24 hours. One email target is accepted; other declared channels return UNIMPLEMENTED.
@@ -3,13 +3,13 @@ package sxchnge.notifications.v1;
option go_package = "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1;notificationsv1";
import "google/protobuf/timestamp.proto";
// Internal service: authenticate callers over TLS. Send durably queues work.
// Internal service on the trusted private network; no caller authentication. Send durably queues work.
service NotificationsService {
rpc Send(SendRequest) returns (SendResponse);
rpc Get(GetRequest) returns (GetResponse);
}
message SendRequest {
string idempotency_key = 1; // Required; scoped to authenticated caller, retained seven days.
string idempotency_key = 1; // Required; scoped to the product namespace, retained seven days.
string user_id = 2; // Caller resolves and authorizes recipient ownership.
string template_id = 3;
string template_version = 4;
+2 -7
View File
@@ -1,7 +1,6 @@
package service_test
import (
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
@@ -17,10 +16,9 @@ import (
"git.sendico.io/tech/sxchnge/notifications/transport"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/metadata"
)
// Exercise authenticated TLS transport against the real service and storage.
// Exercise optional TLS transport against the real service and storage.
func grpcClient(t *testing.T, svc *service.Service) pb.NotificationsServiceClient {
t.Helper()
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
@@ -42,7 +40,7 @@ func grpcClient(t *testing.T, svc *service.Service) pb.NotificationsServiceClien
if err != nil {
t.Fatal(err)
}
server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}})), grpc.UnaryInterceptor(transport.Authenticate(map[string]string{"product": "product-test-token", "other": "other-test-token"})))
server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}})))
pb.RegisterNotificationsServiceServer(server, transport.New(svc))
go func() { _ = server.Serve(listener) }()
t.Cleanup(server.Stop)
@@ -53,6 +51,3 @@ func grpcClient(t *testing.T, svc *service.Service) pb.NotificationsServiceClien
t.Cleanup(func() { _ = conn.Close() })
return pb.NewNotificationsServiceClient(conn)
}
func callerContext(ctx context.Context, caller string) context.Context {
return metadata.NewOutgoingContext(ctx, metadata.Pairs("authorization", "Bearer "+caller+"-test-token"))
}
@@ -26,8 +26,6 @@ import (
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.uber.org/zap"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
type sender struct {
@@ -233,11 +231,8 @@ func TestDurableDelivery(t *testing.T) {
}
client := grpcClient(t, svc)
rpcCtx := callerContext(ctx, "product")
rpcCtx := ctx
input := &pb.SendRequest{IdempotencyKey: "jetstream", UserId: "u", TemplateId: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "End to end & escaped"}, Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: "user@example.com"}}}}}
if _, e := client.Send(ctx, input); status.Code(e) != codes.Unauthenticated {
t.Fatal("unauthenticated request accepted", e)
}
sent, err := client.Send(rpcCtx, input)
if err != nil {
t.Fatal(err)
@@ -246,8 +241,8 @@ func TestDurableDelivery(t *testing.T) {
if err != nil || replay.Notification.Id != sent.Notification.Id {
t.Fatal("RPC idempotency failed", err)
}
if _, e := client.Get(callerContext(ctx, "other"), &pb.GetRequest{Id: sent.Notification.Id}); status.Code(e) != codes.NotFound {
t.Fatal("RPC caller isolation failed", e)
if _, e := client.Get(ctx, &pb.GetRequest{Id: sent.Notification.Id}); e != nil {
t.Fatal("RPC retrieval without credentials failed", e)
}
if err = broker.PublishDue(ctx); err != nil {
t.Fatal(err)
+3 -35
View File
@@ -2,21 +2,15 @@ package transport
import (
"context"
"crypto/sha256"
"crypto/subtle"
"errors"
"strings"
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
"git.sendico.io/tech/sxchnge/notifications/model"
merrors "git.sendico.io/tech/sxchnge/shared/errors"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
type callerKey struct{}
type API interface {
Send(context.Context, string, model.Request) (*model.Notification, error)
Get(context.Context, string, string) (*model.Notification, error)
@@ -26,14 +20,14 @@ type Server struct {
service API
}
// The private endpoint uses the existing product storage namespace without caller authentication.
func New(service API) *Server { return &Server{service: service} }
func (s *Server) Send(ctx context.Context, r *pb.SendRequest) (*pb.SendResponse, error) {
input, err := request(r)
if err != nil {
return nil, rpcError(err)
}
caller, _ := ctx.Value(callerKey{}).(string)
n, err := s.service.Send(ctx, caller, input)
n, err := s.service.Send(ctx, "product", input)
if err != nil {
return nil, rpcError(err)
}
@@ -43,39 +37,13 @@ func (s *Server) Get(ctx context.Context, r *pb.GetRequest) (*pb.GetResponse, er
if r == nil {
return nil, status.Error(codes.InvalidArgument, "missing request")
}
caller, _ := ctx.Value(callerKey{}).(string)
n, err := s.service.Get(ctx, caller, r.Id)
n, err := s.service.Get(ctx, "product", r.Id)
if err != nil {
return nil, rpcError(err)
}
return &pb.GetResponse{Notification: response(n)}, nil
}
// Tokens are mapped to stable server-configured caller identities, never supplied IDs.
func Authenticate(tokens map[string]string) grpc.UnaryServerInterceptor {
hashes := map[string][32]byte{}
for caller, token := range tokens {
hashes[caller] = sha256.Sum256([]byte(token))
}
return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
md, _ := metadata.FromIncomingContext(ctx)
values := md.Get("authorization")
if len(values) != 1 || !strings.HasPrefix(values[0], "Bearer ") {
return nil, status.Error(codes.Unauthenticated, "invalid service credentials")
}
hash := sha256.Sum256([]byte(strings.TrimPrefix(values[0], "Bearer ")))
identity := ""
for caller, expected := range hashes {
if subtle.ConstantTimeCompare(hash[:], expected[:]) == 1 {
identity = caller
}
}
if identity == "" {
return nil, status.Error(codes.Unauthenticated, "invalid service credentials")
}
return handler(context.WithValue(ctx, callerKey{}, identity), req)
}
}
func rpcError(err error) error {
if errors.Is(err, context.Canceled) {
return status.Error(codes.Canceled, "notification operation canceled")
@@ -15,7 +15,6 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
"google.golang.org/grpc/test/bufconn"
)
@@ -31,10 +30,9 @@ func (fakeAPI) Send(_ context.Context, caller string, r model.Request) (*model.N
func (fakeAPI) Get(context.Context, string, string) (*model.Notification, error) {
return nil, merrors.NoData("private details")
}
func TestGRPCAuthenticationAndMapping(t *testing.T) {
func TestGRPCWithoutAuthenticationAndMapping(t *testing.T) {
listener := bufconn.Listen(1 << 20)
token := strings.Repeat("a", 32)
server := grpc.NewServer(grpc.UnaryInterceptor(Authenticate(map[string]string{"product": token})))
server := grpc.NewServer()
pb.RegisterNotificationsServiceServer(server, New(fakeAPI{}))
go func() { _ = server.Serve(listener) }()
defer server.Stop()
@@ -45,10 +43,7 @@ func TestGRPCAuthenticationAndMapping(t *testing.T) {
defer conn.Close()
client := pb.NewNotificationsServiceClient(conn)
req := &pb.SendRequest{UserId: "u", Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: "user@example.com"}}}}}
if _, err = client.Send(context.Background(), req); status.Code(err) != codes.Unauthenticated {
t.Fatal(err)
}
ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("authorization", "Bearer "+token))
ctx := context.Background()
result, err := client.Send(ctx, req)
if err != nil {
t.Fatal(err)