Merge pull request #6619 from TheThingsNetwork/feature/pba-tokens-minify

Optimize Packet Broker Agent uplink token serialization

CHANGELOG.md

@@ -21,6 +21,7 @@ For details about compatibility between different releases, see the **Commitment
 
 - Users can now request a new email for the account validation from time to time instead of once per validation, the interval between email requests is determined by `is.user-registration.contact-info-validation.retry-interval` and by default it is an hour.
 - Traffic related correlation IDs have been simplified. Previously one correlation ID per component was added as traffic passed through different stack components. Now a singular correlation ID relating to the entry point of the message will be added (such as `gs:uplink:xxx` for uplinks, or `as:downlink:xxx` for downlinks), and subsequent components will no longer add any extra correlation IDs (such as `ns:uplink:xxx` or `as:up:xxx`). The uplink entry points are `pba` and `gs`, while the downlink entry points are `pba`, `ns` and `as`.
+- Packet Broker Agent uplink tokens are now serialized in a more efficient manner.
 
 ### Deprecated
 

api/ttn/lorawan/v3/api.md

@@ -604,6 +604,10 @@
   - [Message `ListHomeNetworkRoutingPoliciesRequest`](#ttn.lorawan.v3.ListHomeNetworkRoutingPoliciesRequest)
   - [Message `ListPacketBrokerHomeNetworksRequest`](#ttn.lorawan.v3.ListPacketBrokerHomeNetworksRequest)
   - [Message `ListPacketBrokerNetworksRequest`](#ttn.lorawan.v3.ListPacketBrokerNetworksRequest)
+  - [Message `PacketBrokerAgentCompoundUplinkToken`](#ttn.lorawan.v3.PacketBrokerAgentCompoundUplinkToken)
+  - [Message `PacketBrokerAgentEncryptedPayload`](#ttn.lorawan.v3.PacketBrokerAgentEncryptedPayload)
+  - [Message `PacketBrokerAgentGatewayUplinkToken`](#ttn.lorawan.v3.PacketBrokerAgentGatewayUplinkToken)
+  - [Message `PacketBrokerAgentUplinkToken`](#ttn.lorawan.v3.PacketBrokerAgentUplinkToken)
   - [Message `PacketBrokerDefaultGatewayVisibility`](#ttn.lorawan.v3.PacketBrokerDefaultGatewayVisibility)
   - [Message `PacketBrokerDefaultRoutingPolicy`](#ttn.lorawan.v3.PacketBrokerDefaultRoutingPolicy)
   - [Message `PacketBrokerDevAddrBlock`](#ttn.lorawan.v3.PacketBrokerDevAddrBlock)
@@ -8693,6 +8697,36 @@ organization registrations.
 | `tenant_id_contains` | <p>`string.max_len`: `100`</p> |
 | `name_contains` | <p>`string.max_len`: `100`</p> |
 
+### <a name="ttn.lorawan.v3.PacketBrokerAgentCompoundUplinkToken">Message `PacketBrokerAgentCompoundUplinkToken`</a>
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| `gateway` | [`bytes`](#bytes) |  |  |
+| `forwarder` | [`bytes`](#bytes) |  |  |
+| `agent` | [`PacketBrokerAgentUplinkToken`](#ttn.lorawan.v3.PacketBrokerAgentUplinkToken) |  |  |
+
+### <a name="ttn.lorawan.v3.PacketBrokerAgentEncryptedPayload">Message `PacketBrokerAgentEncryptedPayload`</a>
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| `ciphertext` | [`bytes`](#bytes) |  |  |
+| `nonce` | [`bytes`](#bytes) |  |  |
+
+### <a name="ttn.lorawan.v3.PacketBrokerAgentGatewayUplinkToken">Message `PacketBrokerAgentGatewayUplinkToken`</a>
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| `gateway_uid` | [`string`](#string) |  |  |
+| `token` | [`bytes`](#bytes) |  |  |
+
+### <a name="ttn.lorawan.v3.PacketBrokerAgentUplinkToken">Message `PacketBrokerAgentUplinkToken`</a>
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| `forwarder_net_id` | [`bytes`](#bytes) |  |  |
+| `forwarder_tenant_id` | [`string`](#string) |  |  |
+| `forwarder_cluster_id` | [`string`](#string) |  |  |
+
 ### <a name="ttn.lorawan.v3.PacketBrokerDefaultGatewayVisibility">Message `PacketBrokerDefaultGatewayVisibility`</a>
 
 | Field | Type | Label | Description |

api/ttn/lorawan/v3/packetbrokeragent.proto

@@ -465,3 +465,25 @@ service Pba {
     option (google.api.http) = {get: "/pba/forwarders/policies"};
   }
 }
+
+message PacketBrokerAgentUplinkToken {
+  bytes forwarder_net_id = 1;
+  string forwarder_tenant_id = 2;
+  string forwarder_cluster_id = 3;
+}
+
+message PacketBrokerAgentGatewayUplinkToken {
+  string gateway_uid = 1;
+  bytes token = 2;
+}
+
+message PacketBrokerAgentCompoundUplinkToken {
+  bytes gateway = 1;
+  bytes forwarder = 2;
+  PacketBrokerAgentUplinkToken agent = 3;
+}
+
+message PacketBrokerAgentEncryptedPayload {
+  bytes ciphertext = 1;
+  bytes nonce = 2;
+}

pkg/packetbrokeragent/agent.go

@@ -17,6 +17,8 @@ package packetbrokeragent
 
 import (
 	"context"
+	"crypto/aes"
+	"crypto/cipher"
 	"encoding/hex"
 	"fmt"
 	"sync"
@@ -90,7 +92,7 @@ type uplinkMessage struct {
 
 type downlinkMessage struct {
 	context.Context
-	*agentUplinkToken
+	*ttnpb.PacketBrokerAgentUplinkToken
 	*packetbroker.DownlinkMessage
 }
 
@@ -270,25 +272,33 @@ func New(c *component.Component, conf *Config, opts ...Option) (*Agent, error) {
 			logger.WithField("token_key", hex.EncodeToString(a.forwarderConfig.TokenKey)).Warn("No token key configured, generated a random one")
 		}
 		var (
-			enc jose.ContentEncryption
-			alg jose.KeyAlgorithm
+			legacyEncryption   jose.ContentEncryption
+			legacyKeyAlgorithm jose.KeyAlgorithm
 		)
 		switch l := len(a.forwarderConfig.TokenKey); l {
 		case 16:
-			enc, alg = jose.A128GCM, jose.A128GCMKW
+			legacyEncryption, legacyKeyAlgorithm = jose.A128GCM, jose.A128GCMKW
 		case 32:
-			enc, alg = jose.A256GCM, jose.A256GCMKW
+			legacyEncryption, legacyKeyAlgorithm = jose.A256GCM, jose.A256GCMKW
 		default:
 			return nil, errTokenKey.WithAttributes("length", l).New()
 		}
 		var err error
-		a.forwarderConfig.TokenEncrypter, err = jose.NewEncrypter(enc, jose.Recipient{
-			Algorithm: alg,
+		a.forwarderConfig.LegacyTokenEncrypter, err = jose.NewEncrypter(legacyEncryption, jose.Recipient{
+			Algorithm: legacyKeyAlgorithm,
 			Key:       a.forwarderConfig.TokenKey,
 		}, nil)
 		if err != nil {
 			return nil, errTokenKey.WithCause(err)
 		}
+		blockCipher, err := aes.NewCipher(a.forwarderConfig.TokenKey)
+		if err != nil {
+			return nil, errTokenKey.WithCause(err)
+		}
+		a.forwarderConfig.TokenAEAD, err = cipher.NewGCM(blockCipher)
+		if err != nil {
+			return nil, errTokenKey.WithCause(err)
+		}
 	}
 
 	if a.homeNetworkConfig.Enable {
@@ -1109,21 +1119,22 @@ func (a *Agent) homeNetworkPublisher(conn *grpc.ClientConn) workerpool.Handler[*
 	client := routingpb.NewHomeNetworkDataClient(conn)
 	h := func(ctx context.Context, down *downlinkMessage) {
 		tenantID := a.tenantIDExtractor(down.Context)
-		msg, token := down.DownlinkMessage, down.agentUplinkToken
+		msg, token := down.DownlinkMessage, down.PacketBrokerAgentUplinkToken
+		forwarderNetID := types.MustNetID(token.ForwarderNetId)
 		ctx = log.NewContextWithFields(ctx, log.Fields(
-			"forwarder_net_id", token.ForwarderNetID,
-			"forwarder_tenant_id", token.ForwarderTenantID,
-			"forwarder_cluster_id", token.ForwarderClusterID,
+			"forwarder_net_id", forwarderNetID,
+			"forwarder_tenant_id", token.ForwarderTenantId,
+			"forwarder_cluster_id", token.ForwarderClusterId,
 			"home_network_tenant_id", tenantID,
 		))
 		logger := log.FromContext(ctx)
 		req := &routingpb.PublishDownlinkMessageRequest{
 			HomeNetworkNetId:     a.netID.MarshalNumber(),
 			HomeNetworkTenantId:  tenantID,
 			HomeNetworkClusterId: a.homeNetworkClusterID,
-			ForwarderNetId:       token.ForwarderNetID.MarshalNumber(),
-			ForwarderTenantId:    token.ForwarderTenantID,
-			ForwarderClusterId:   token.ForwarderClusterID,
+			ForwarderNetId:       forwarderNetID.MarshalNumber(),
+			ForwarderTenantId:    token.ForwarderTenantId,
+			ForwarderClusterId:   token.ForwarderClusterId,
 			Message:              msg,
 		}
 		ctx, cancel := context.WithTimeout(ctx, publishMessageTimeout)
@@ -1135,7 +1146,7 @@ func (a *Agent) homeNetworkPublisher(conn *grpc.ClientConn) workerpool.Handler[*
 			logger.WithField("message_id", res.Id).Debug("Published downlink message")
 			pbaMetrics.downlinkForwarded.WithLabelValues(ctx,
 				a.netID.String(), tenantID, a.homeNetworkClusterID,
-				token.ForwarderNetID.String(), token.ForwarderTenantID, token.ForwarderClusterID,
+				forwarderNetID.String(), token.ForwarderTenantId, token.ForwarderClusterId,
 			).Inc()
 		}
 	}

pkg/packetbrokeragent/agent_test.go

@@ -17,7 +17,10 @@ package packetbrokeragent_test
 import (
 	"bytes"
 	"context"
-	"encoding/json"
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/rand"
+	"io"
 	"strconv"
 	"testing"
 	"time"
@@ -41,12 +44,12 @@ import (
 	"go.thethings.network/lorawan-stack/v3/pkg/util/test"
 	"go.thethings.network/lorawan-stack/v3/pkg/util/test/assertions/should"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/fieldmaskpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
-	"gopkg.in/square/go-jose.v2"
 )
 
 var (
@@ -143,10 +146,8 @@ func TestForwarder(t *testing.T) {
 
 	gs := test.Must(mock.NewGatewayServer(c))
 	tokenKey := bytes.Repeat([]byte{0x42}, 16)
-	tokenEncrypter := test.Must(jose.NewEncrypter(jose.A128GCM, jose.Recipient{
-		Algorithm: jose.A128GCMKW,
-		Key:       tokenKey,
-	}, nil)).(jose.Encrypter)
+	blockCipher := test.Must(aes.NewCipher(tokenKey))
+	aead := test.Must(cipher.NewGCM(blockCipher))
 	test.Must(New(c, &Config{
 		IAMAddress:          iamAddr.String(),
 		ControlPlaneAddress: cpAddr.String(),
@@ -161,7 +162,7 @@ func TestForwarder(t *testing.T) {
 				Limit: 1,
 			},
 			TokenKey:          tokenKey,
-			TokenEncrypter:    tokenEncrypter,
+			TokenAEAD:         aead,
 			IncludeGatewayEUI: true,
 			IncludeGatewayID:  true,
 			HashGatewayID:     true,
@@ -421,12 +422,16 @@ func TestForwarder(t *testing.T) {
 	t.Run("Downlink", func(t *testing.T) {
 		a := assertions.New(t)
 
-		token := test.Must(json.Marshal(GatewayUplinkToken{
-			GatewayUID: unique.ID(ctx, &ttnpb.GatewayIdentifiers{GatewayId: "test-gateway"}),
+		plainToken := test.Must(proto.Marshal(&ttnpb.PacketBrokerAgentGatewayUplinkToken{
+			GatewayUid: unique.ID(ctx, &ttnpb.GatewayIdentifiers{GatewayId: "test-gateway"}),
 			Token:      []byte{0x1, 0x2, 0x3, 0x4},
 		}))
-		tokenObj := test.Must(tokenEncrypter.Encrypt(token))
-		tokenCompact := test.Must(tokenObj.CompactSerialize())
+		tokenNonce := make([]byte, aead.NonceSize())
+		test.Must(io.ReadFull(rand.Reader, tokenNonce))
+		token := test.Must(proto.Marshal(&ttnpb.PacketBrokerAgentEncryptedPayload{
+			Ciphertext: aead.Seal(nil, tokenNonce, plainToken, nil),
+			Nonce:      tokenNonce,
+		}))
 
 		dp.ForwarderDown <- &packetbroker.RoutedDownlinkMessage{
 			ForwarderNetId:      0x000013,
@@ -449,7 +454,7 @@ func TestForwarder(t *testing.T) {
 					DataRate:  packetbroker.NewLoRaDataRate(12, 125000, ""),
 				},
 				Rx1Delay:           durationpb.New(5 * time.Second),
-				GatewayUplinkToken: []byte(tokenCompact),
+				GatewayUplinkToken: token,
 			},
 		}
 
@@ -813,10 +818,10 @@ func TestHomeNetwork(t *testing.T) {
 								Longitude: 4.8,
 								Altitude:  2,
 							},
-							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &AgentUplinkToken{
-								ForwarderNetID:     [3]byte{0x0, 0x0, 0x42},
-								ForwarderTenantID:  "foo-tenant",
-								ForwarderClusterID: "test",
+							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &ttnpb.PacketBrokerAgentUplinkToken{
+								ForwarderNetId:     []byte{0x0, 0x0, 0x42},
+								ForwarderTenantId:  "foo-tenant",
+								ForwarderClusterId: "test",
 							})),
 						},
 						{
@@ -839,10 +844,10 @@ func TestHomeNetwork(t *testing.T) {
 							Rssi:            -43,
 							Snr:             10.6,
 							FrequencyOffset: 1,
-							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &AgentUplinkToken{
-								ForwarderNetID:     [3]byte{0x0, 0x0, 0x42},
-								ForwarderTenantID:  "foo-tenant",
-								ForwarderClusterID: "test",
+							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &ttnpb.PacketBrokerAgentUplinkToken{
+								ForwarderNetId:     []byte{0x0, 0x0, 0x42},
+								ForwarderTenantId:  "foo-tenant",
+								ForwarderClusterId: "test",
 							})),
 						},
 					},
@@ -950,10 +955,10 @@ func TestHomeNetwork(t *testing.T) {
 							ChannelRssi: 4.2,
 							Rssi:        4.2,
 							Snr:         -5.5,
-							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &AgentUplinkToken{
-								ForwarderNetID:     [3]byte{0x0, 0x0, 0x42},
-								ForwarderTenantID:  "foo-tenant",
-								ForwarderClusterID: "test",
+							UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &ttnpb.PacketBrokerAgentUplinkToken{
+								ForwarderNetId:     []byte{0x0, 0x0, 0x42},
+								ForwarderTenantId:  "foo-tenant",
+								ForwarderClusterId: "test",
 							})),
 						},
 					},
@@ -1019,10 +1024,10 @@ func TestHomeNetwork(t *testing.T) {
 					DownlinkPaths: []*ttnpb.DownlinkPath{
 						{
 							Path: &ttnpb.DownlinkPath_UplinkToken{
-								UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &AgentUplinkToken{
-									ForwarderNetID:     [3]byte{0x0, 0x0, 0x42},
-									ForwarderTenantID:  "foo-tenant",
-									ForwarderClusterID: "test",
+								UplinkToken: test.Must(WrapUplinkTokens([]byte("test-token"), nil, &ttnpb.PacketBrokerAgentUplinkToken{
+									ForwarderNetId:     []byte{0x0, 0x0, 0x42},
+									ForwarderTenantId:  "foo-tenant",
+									ForwarderClusterId: "test",
 								})),
 							},
 						},

pkg/packetbrokeragent/config.go

@@ -15,6 +15,7 @@
 package packetbrokeragent
 
 import (
+	"crypto/cipher"
 	"time"
 
 	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
@@ -74,14 +75,15 @@ type OAuth2Config struct {
 
 // ForwarderConfig defines configuration of the Forwarder role.
 type ForwarderConfig struct {
-	Enable            bool             `name:"enable" description:"Enable Forwarder role"`
-	WorkerPool        WorkerPoolConfig `name:"worker-pool" description:"Workers pool configuration"`
-	TokenKey          []byte           `name:"token-key" description:"AES 128 or 256-bit key for encrypting tokens"`
-	TokenEncrypter    jose.Encrypter   `name:"-"`
-	IncludeGatewayEUI bool             `name:"include-gateway-eui" description:"Include the gateway EUI in forwarded metadata"`
-	IncludeGatewayID  bool             `name:"include-gateway-id" description:"Include the gateway ID in forwarded metadata"`
-	HashGatewayID     bool             `name:"hash-gateway-id" description:"Hash the gateway ID (if forwarded in the metadata)"`
-	GatewayOnlineTTL  time.Duration    `name:"gateway-online-ttl" description:"Time-to-live of online status reported to Packet Broker"`
+	Enable               bool             `name:"enable" description:"Enable Forwarder role"`
+	WorkerPool           WorkerPoolConfig `name:"worker-pool" description:"Workers pool configuration"`
+	TokenKey             []byte           `name:"token-key" description:"AES 128 or 256-bit key for encrypting tokens"`
+	LegacyTokenEncrypter jose.Encrypter   `name:"-"`
+	TokenAEAD            cipher.AEAD      `name:"-"`
+	IncludeGatewayEUI    bool             `name:"include-gateway-eui" description:"Include the gateway EUI in forwarded metadata"`          // nolint:lll
+	IncludeGatewayID     bool             `name:"include-gateway-id" description:"Include the gateway ID in forwarded metadata"`            // nolint:lll
+	HashGatewayID        bool             `name:"hash-gateway-id" description:"Hash the gateway ID (if forwarded in the metadata)"`         // nolint:lll
+	GatewayOnlineTTL     time.Duration    `name:"gateway-online-ttl" description:"Time-to-live of online status reported to Packet Broker"` // nolint:lll
 }
 
 // HomeNetworkConfig defines the configuration of the Home Network role.

pkg/packetbrokeragent/grpc_nspba.go

@@ -54,9 +54,9 @@ func (s *nsPbaServer) PublishDownlink(ctx context.Context, down *ttnpb.DownlinkM
 	}
 
 	ctxMsg := &downlinkMessage{
-		Context:          s.contextDecoupler.FromRequestContext(ctx),
-		agentUplinkToken: token,
-		DownlinkMessage:  msg,
+		Context:                      s.contextDecoupler.FromRequestContext(ctx),
+		PacketBrokerAgentUplinkToken: token,
+		DownlinkMessage:              msg,
 	}
 	select {
 	case <-ctx.Done():