Integrate email notifications into signup and login (#38)
This commit is contained in:
Stephan D committed
1 parent
326d58f705
commit
ad5a061164
55 files changed
+1859
-3464
No files matched your search
@@ -0,0 +1,88 @@
|
||||
// Package notifications adapts the internal notification wire contract to product models.
|
||||
package notifications
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Address string `yaml:"address"`
|
||||
TokenEnv string `yaml:"token_env"`
|
||||
CAFile string `yaml:"ca_file"`
|
||||
ServerName string `yaml:"server_name"`
|
||||
ActivationURL string `yaml:"activation_url"`
|
||||
}
|
||||
type Message struct {
|
||||
Key, UserID, Email, Locale, Template string
|
||||
Parameters map[string]string
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
type Sender interface {
|
||||
Send(context.Context, Message) error
|
||||
}
|
||||
type Client struct {
|
||||
conn *grpc.ClientConn
|
||||
rpc pb.NotificationsServiceClient
|
||||
token string
|
||||
}
|
||||
|
||||
func Open(c Config) (*Client, error) {
|
||||
token := os.Getenv(c.TokenEnv)
|
||||
if c.Address == "" || len(token) < 32 {
|
||||
return nil, merrors.InvalidArgument("notification address and caller token of at least 32 bytes are required")
|
||||
}
|
||||
tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12, ServerName: c.ServerName}
|
||||
if c.CAFile != "" {
|
||||
pem, err := os.ReadFile(c.CAFile)
|
||||
if err != nil {
|
||||
return nil, merrors.InternalWrap(err, "read notification CA")
|
||||
}
|
||||
pool, err := x509.SystemCertPool()
|
||||
if err != nil {
|
||||
pool = x509.NewCertPool()
|
||||
}
|
||||
if !pool.AppendCertsFromPEM(pem) {
|
||||
return nil, merrors.InvalidArgument("invalid notification CA")
|
||||
}
|
||||
tlsConfig.RootCAs = pool
|
||||
}
|
||||
conn, err := grpc.NewClient(c.Address, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
|
||||
if err != nil {
|
||||
return nil, merrors.InternalWrap(err, "create notification client")
|
||||
}
|
||||
return &Client{conn: conn, rpc: pb.NewNotificationsServiceClient(conn), token: token}, nil
|
||||
}
|
||||
func (c *Client) Close() error { return c.conn.Close() }
|
||||
func (c *Client) Send(ctx context.Context, m Message) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
ctx = metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+c.token)
|
||||
res, err := c.rpc.Send(ctx, request(m))
|
||||
if err != nil {
|
||||
return merrors.ServiceUnavailableWrap(err, "notification queue unavailable")
|
||||
}
|
||||
if res.GetNotification().GetId() == "" || len(res.GetNotification().GetDeliveries()) != 1 {
|
||||
return merrors.ServiceUnavailable("notification queue returned no receipt")
|
||||
}
|
||||
for _, d := range res.Notification.Deliveries {
|
||||
switch d.State {
|
||||
case pb.DeliveryState_DELIVERY_STATE_FAILED, pb.DeliveryState_DELIVERY_STATE_EXPIRED, pb.DeliveryState_DELIVERY_STATE_UNKNOWN:
|
||||
return merrors.ServiceUnavailable("notification could not be delivered")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func request(m Message) *pb.SendRequest {
|
||||
return &pb.SendRequest{IdempotencyKey: m.Key, UserId: m.UserID, TemplateId: m.Template, TemplateVersion: "1", Locale: m.Locale, Parameters: m.Parameters, ExpiresAt: timestamppb.New(m.ExpiresAt), Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: m.Email}}}}}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package notifications
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/pem"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
||||
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
type rpcStub struct {
|
||||
pb.NotificationsServiceClient
|
||||
request *pb.SendRequest
|
||||
auth string
|
||||
result *pb.SendResponse
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *rpcStub) Send(ctx context.Context, r *pb.SendRequest, _ ...grpc.CallOption) (*pb.SendResponse, error) {
|
||||
s.request = r
|
||||
md, _ := metadata.FromOutgoingContext(ctx)
|
||||
s.auth = md.Get("authorization")[0]
|
||||
return s.result, s.err
|
||||
}
|
||||
func TestSendMapsOwnedContractAndRejectsFailedReceipt(t *testing.T) {
|
||||
rpc := &rpcStub{result: &pb.SendResponse{Notification: &pb.Notification{Id: "job", Deliveries: []*pb.Delivery{{State: pb.DeliveryState_DELIVERY_STATE_QUEUED}}}}}
|
||||
client := &Client{rpc: rpc, token: "service-secret"}
|
||||
expiry := time.Now().Add(time.Minute)
|
||||
m := Message{Key: "key", UserID: "customer", Email: "user@example.com", Locale: "ru", Template: "auth.login", Parameters: map[string]string{"code": "123456"}, ExpiresAt: expiry}
|
||||
require.NoError(t, client.Send(context.Background(), m))
|
||||
require.Equal(t, "Bearer service-secret", rpc.auth)
|
||||
require.Equal(t, m.Email, rpc.request.Targets[0].GetEmail().Address)
|
||||
require.Equal(t, "1", rpc.request.TemplateVersion)
|
||||
require.Equal(t, m.Parameters, rpc.request.Parameters)
|
||||
require.WithinDuration(t, expiry, rpc.request.ExpiresAt.AsTime(), time.Nanosecond)
|
||||
rpc.result.Notification.Deliveries[0].State = pb.DeliveryState_DELIVERY_STATE_UNKNOWN
|
||||
require.ErrorIs(t, client.Send(context.Background(), m), merrors.ErrServiceUnavailable)
|
||||
rpc.result = nil
|
||||
require.ErrorIs(t, client.Send(context.Background(), m), merrors.ErrServiceUnavailable)
|
||||
}
|
||||
func TestOpenRequiresCredentialsAndValidCA(t *testing.T) {
|
||||
_, err := Open(Config{Address: "localhost:8443", TokenEnv: "ABSENT_NOTIFICATION_TEST_TOKEN"})
|
||||
require.ErrorIs(t, err, merrors.ErrInvalidArg)
|
||||
t.Setenv("TEST_NOTIFICATION_TOKEN", "test-token-that-is-at-least-32-bytes")
|
||||
_, err = Open(Config{Address: "localhost:8443", TokenEnv: "TEST_NOTIFICATION_TOKEN", CAFile: "missing.pem"})
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
type tlsRPC struct {
|
||||
pb.UnimplementedNotificationsServiceServer
|
||||
auth chan string
|
||||
}
|
||||
|
||||
func (s *tlsRPC) Send(ctx context.Context, _ *pb.SendRequest) (*pb.SendResponse, error) {
|
||||
md, _ := metadata.FromIncomingContext(ctx)
|
||||
s.auth <- md.Get("authorization")[0]
|
||||
return &pb.SendResponse{Notification: &pb.Notification{Id: "tls-job", Deliveries: []*pb.Delivery{{State: pb.DeliveryState_DELIVERY_STATE_QUEUED}}}}, nil
|
||||
}
|
||||
func TestOpenUsesVerifiedTLSAndCallerAuthentication(t *testing.T) {
|
||||
fixture := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
|
||||
cert := fixture.TLS.Certificates[0]
|
||||
fixture.Close()
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12})))
|
||||
rpc := &tlsRPC{auth: make(chan string, 1)}
|
||||
pb.RegisterNotificationsServiceServer(server, rpc)
|
||||
go func() { _ = server.Serve(listener) }()
|
||||
t.Cleanup(server.Stop)
|
||||
ca := filepath.Join(t.TempDir(), "ca.pem")
|
||||
require.NoError(t, os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Certificate[0]}), 0600))
|
||||
t.Setenv("TEST_TLS_CALLER_TOKEN", "test-service-token-with-at-least-32-bytes")
|
||||
client, err := Open(Config{Address: listener.Addr().String(), TokenEnv: "TEST_TLS_CALLER_TOKEN", CAFile: ca})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = client.Close() })
|
||||
require.NoError(t, client.Send(context.Background(), Message{ExpiresAt: time.Now().Add(time.Minute)}))
|
||||
require.Equal(t, "Bearer test-service-token-with-at-least-32-bytes", <-rpc.auth)
|
||||
}
|
||||
Reference in new issue
Block a user