use hybrid ed25519 + pq crypto signatures for session authentication

internal/api/syncapi/cmdstreamutil.go

@@ -1,7 +1,6 @@
 package syncapi
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -111,14 +110,16 @@ func (s *bidiSyncCommandStream) ReceiveWithinDuration(ctx context.Context, d tim
 }
 
 // ConnectStream bridges the channel-based bidiSyncCommandStream to a real transport.
-// It first performs an ECDH key exchange on the raw transport to establish an encrypted
-// session, then starts the send/recv pump loop over the encrypted channel.
-func (s *bidiSyncCommandStream) ConnectStream(ctx context.Context, stream syncCommandStreamTrait) error {
+// It first performs a post-quantum KEM handshake on the raw transport to
+// establish an encrypted session, then starts the send/recv pump loop over the
+// encrypted channel. isInitiator must be true on the side that opens the
+// connection (the client) and false on the side that accepts it (the server).
+func (s *bidiSyncCommandStream) ConnectStream(ctx context.Context, stream syncCommandStreamTrait, isInitiator bool) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	// Perform ECDH key exchange on the raw transport before starting the pump.
-	transport, err := establishEncryption(stream)
+	// Perform the PQ KEM handshake on the raw transport before starting the pump.
+	transport, err := establishEncryption(stream, isInitiator)
 	if err != nil {
 		// Signal termination so any goroutine parked in ReceiveWithinDuration
 		// (e.g. runSync waiting for the handshake reply) wakes up immediately
@@ -164,52 +165,92 @@ func (s *bidiSyncCommandStream) ConnectStream(ctx context.Context, stream syncCo
 	}
 }
 
-// establishEncryption performs an ECDH key exchange on the raw transport and
-// returns an encrypted stream wrapper. Each side generates an ephemeral ECDH P-256
-// key pair, exchanges public keys, and derives a shared AES-256-GCM session key.
-// The handshake (identity authentication) runs over the encrypted channel afterward.
-func establishEncryption(stream syncCommandStreamTrait) (syncCommandStreamTrait, error) {
-	keyPair, err := cryptoutil.GenerateECDHKeyPair()
-	if err != nil {
-		return nil, NewSyncErrorInternal(fmt.Errorf("generating ephemeral ECDH key: %w", err))
-	}
+// establishEncryption performs a post-quantum KEM handshake on the raw
+// transport and returns an encrypted stream wrapper. The KEM ciphersuite is
+// hard-pinned to TransportProtocolVersion and tied to the wire format.
+//
+// Flow: the initiator generates an ephemeral hybrid (ML-KEM-1024 + ECDH-P384)
+// HPKE keypair and sends its public key. The responder encapsulates against
+// it and replies with the encapsulation. Both sides derive an AES-256-GCM
+// session key via the HPKE Export interface. The handshake (identity
+// authentication) runs over the encrypted channel afterward.
+//
+// Initiators use nonce prefix 0x00; responders use 0x01 — fixed by role to
+// avoid nonce reuse. isInitiator must be true on the connecting side (client)
+// and false on the accepting side (server).
+func establishEncryption(stream syncCommandStreamTrait, isInitiator bool) (syncCommandStreamTrait, error) {
+	if isInitiator {
+		recipient, pubBytes, err := cryptoutil.NewTransportRecipient()
+		if err != nil {
+			return nil, NewSyncErrorInternal(fmt.Errorf("generating ephemeral KEM key: %w", err))
+		}
 
-	// Send our ephemeral ECDH public key
-	if err := stream.Send(&v1sync.SyncStreamItem{
-		Action: &v1sync.SyncStreamItem_EstablishSharedSecret{
-			EstablishSharedSecret: &v1sync.SyncStreamItem_SyncEstablishSharedSecret{
-				EcdhPublicKey: keyPair.Public.Bytes(),
+		if err := stream.Send(&v1sync.SyncStreamItem{
+			Action: &v1sync.SyncStreamItem_EstablishSharedSecret{
+				EstablishSharedSecret: &v1sync.SyncStreamItem_SyncEstablishSharedSecret{
+					ProtocolVersion: cryptoutil.TransportProtocolVersion,
+					KemPublicKey:    pubBytes,
+				},
 			},
-		},
-	}); err != nil {
-		return nil, NewSyncErrorDisconnected(fmt.Errorf("sending ECDH public key: %w", err))
+		}); err != nil {
+			return nil, NewSyncErrorDisconnected(fmt.Errorf("sending KEM public key: %w", err))
+		}
+
+		peerMsg, err := stream.Receive()
+		if err != nil {
+			return nil, NewSyncErrorDisconnected(fmt.Errorf("receiving KEM encapsulation: %w", err))
+		}
+		peerSecret := peerMsg.GetEstablishSharedSecret()
+		if peerSecret == nil {
+			return nil, NewSyncErrorProtocol(fmt.Errorf("expected KEM key exchange, got %T", peerMsg.GetAction()))
+		}
+		if peerSecret.GetProtocolVersion() != cryptoutil.TransportProtocolVersion {
+			return nil, NewSyncErrorProtocol(fmt.Errorf("unsupported transport protocol version %d (this build requires v%d, post-quantum)", peerSecret.GetProtocolVersion(), cryptoutil.TransportProtocolVersion))
+		}
+		if len(peerSecret.GetKemEncapsulation()) == 0 {
+			return nil, NewSyncErrorProtocol(errors.New("responder did not send KEM encapsulation"))
+		}
+
+		sess, err := recipient.Decapsulate(peerSecret.GetKemEncapsulation())
+		if err != nil {
+			return nil, NewSyncErrorProtocol(fmt.Errorf("decapsulating KEM: %w", err))
+		}
+
+		zap.L().Info("encrypted sync session established (initiator)")
+		return newEncryptedStream(stream, sess), nil
 	}
 
-	// Receive the peer's ephemeral ECDH public key
 	peerMsg, err := stream.Receive()
 	if err != nil {
-		return nil, NewSyncErrorDisconnected(fmt.Errorf("receiving ECDH public key: %w", err))
+		return nil, NewSyncErrorDisconnected(fmt.Errorf("receiving KEM public key: %w", err))
 	}
 	peerSecret := peerMsg.GetEstablishSharedSecret()
 	if peerSecret == nil {
-		return nil, NewSyncErrorProtocol(fmt.Errorf("expected ECDH key exchange, got %T", peerMsg.GetAction()))
+		return nil, NewSyncErrorProtocol(fmt.Errorf("expected KEM key exchange, got %T", peerMsg.GetAction()))
+	}
+	if peerSecret.GetProtocolVersion() != cryptoutil.TransportProtocolVersion {
+		return nil, NewSyncErrorProtocol(fmt.Errorf("unsupported transport protocol version %d (this build requires v%d, post-quantum)", peerSecret.GetProtocolVersion(), cryptoutil.TransportProtocolVersion))
+	}
+	if len(peerSecret.GetKemPublicKey()) == 0 {
+		return nil, NewSyncErrorProtocol(errors.New("initiator did not send KEM public key"))
 	}
 
-	peerECDHPub, err := cryptoutil.ParseECDHPublicKey(peerSecret.GetEcdhPublicKey())
+	enc, sess, err := cryptoutil.EncapsulateToTransport(peerSecret.GetKemPublicKey())
 	if err != nil {
-		return nil, NewSyncErrorProtocol(fmt.Errorf("parsing peer ECDH public key: %w", err))
+		return nil, NewSyncErrorProtocol(fmt.Errorf("encapsulating to KEM public key: %w", err))
 	}
 
-	// Derive AES-256-GCM session key
-	gcm, err := cryptoutil.DeriveSessionKey(keyPair.Private, peerECDHPub)
-	if err != nil {
-		return nil, NewSyncErrorProtocol(fmt.Errorf("deriving session key: %w", err))
+	if err := stream.Send(&v1sync.SyncStreamItem{
+		Action: &v1sync.SyncStreamItem_EstablishSharedSecret{
+			EstablishSharedSecret: &v1sync.SyncStreamItem_SyncEstablishSharedSecret{
+				ProtocolVersion:  cryptoutil.TransportProtocolVersion,
+				KemEncapsulation: enc,
+			},
+		},
+	}); err != nil {
+		return nil, NewSyncErrorDisconnected(fmt.Errorf("sending KEM encapsulation: %w", err))
 	}
 
-	// Determine nonce direction: side with smaller public key uses prefix 0x00
-	localIsSmaller := bytes.Compare(keyPair.Public.Bytes(), peerECDHPub.Bytes()) < 0
-
-	zap.L().Info("encrypted sync session established")
-
-	return newEncryptedStream(stream, gcm, localIsSmaller), nil
+	zap.L().Info("encrypted sync session established (responder)")
+	return newEncryptedStream(stream, sess), nil
 }

internal/api/syncapi/encryption.go

@@ -7,23 +7,20 @@ import (
 	"sync"
 
 	"github.com/garethgeorge/backrest/gen/go/v1sync"
+	"github.com/garethgeorge/backrest/internal/cryptoutil"
 	"google.golang.org/protobuf/proto"
 )
 
-// encryptedStream wraps a syncCommandStreamTrait with AES-256-GCM encryption.
-// Outgoing SyncStreamItems are serialized, encrypted, and sent as SyncActionEncrypted.
-// Incoming SyncActionEncrypted messages are decrypted and deserialized back to SyncStreamItems.
+// encryptedStream wraps a syncCommandStreamTrait with AES-256-GCM encryption
+// using the per-direction AEADs derived during the transport handshake.
 //
-// To avoid nonce reuse (since both sides share the same key), each direction
-// uses a different nonce prefix byte: the side with the lexicographically smaller
-// ECDH public key uses prefix 0x00 for sending and expects 0x01 for receiving,
-// and vice versa.
+// Each direction has an independent key (initiator-to-responder vs
+// responder-to-initiator), so a counter-based nonce starting at zero is
+// sufficient: there is no shared (key, nonce) space to collide in.
 type encryptedStream struct {
 	inner syncCommandStreamTrait
-	gcm   cipher.AEAD
-
-	sendPrefix byte
-	recvPrefix byte
+	send  cipher.AEAD
+	recv  cipher.AEAD
 
 	sendMu      sync.Mutex
 	sendCounter uint64
@@ -32,18 +29,11 @@ type encryptedStream struct {
 	recvCounter uint64
 }
 
-func newEncryptedStream(inner syncCommandStreamTrait, gcm cipher.AEAD, localIsSmaller bool) *encryptedStream {
-	var sendPrefix, recvPrefix byte
-	if localIsSmaller {
-		sendPrefix, recvPrefix = 0x00, 0x01
-	} else {
-		sendPrefix, recvPrefix = 0x01, 0x00
-	}
+func newEncryptedStream(inner syncCommandStreamTrait, sess *cryptoutil.TransportSession) *encryptedStream {
 	return &encryptedStream{
-		inner:      inner,
-		gcm:        gcm,
-		sendPrefix: sendPrefix,
-		recvPrefix: recvPrefix,
+		inner: inner,
+		send:  sess.Send,
+		recv:  sess.Recv,
 	}
 }
 
@@ -54,11 +44,11 @@ func (s *encryptedStream) Send(item *v1sync.SyncStreamItem) error {
 	}
 
 	s.sendMu.Lock()
-	nonce := s.makeNonce(s.sendPrefix, s.sendCounter)
+	nonce := makeNonce(s.send.NonceSize(), s.sendCounter)
 	s.sendCounter++
 	s.sendMu.Unlock()
 
-	ciphertext := s.gcm.Seal(nil, nonce, plaintext, nil)
+	ciphertext := s.send.Seal(nil, nonce, plaintext, nil)
 
 	return s.inner.Send(&v1sync.SyncStreamItem{
 		Action: &v1sync.SyncStreamItem_Encrypted{
@@ -82,22 +72,22 @@ func (s *encryptedStream) Receive() (*v1sync.SyncStreamItem, error) {
 	}
 
 	s.recvMu.Lock()
-	expectedNonce := s.makeNonce(s.recvPrefix, s.recvCounter)
+	expectedNonce := makeNonce(s.recv.NonceSize(), s.recvCounter)
 	s.recvCounter++
 	s.recvMu.Unlock()
 
-	if len(encrypted.Nonce) != s.gcm.NonceSize() {
-		return nil, fmt.Errorf("invalid nonce size: got %d, want %d", len(encrypted.Nonce), s.gcm.NonceSize())
+	if len(encrypted.Nonce) != s.recv.NonceSize() {
+		return nil, fmt.Errorf("invalid nonce size: got %d, want %d", len(encrypted.Nonce), s.recv.NonceSize())
 	}
 
-	// Verify nonce matches expected counter to prevent replay/reorder attacks
+	// Verify nonce matches expected counter to prevent replay/reorder attacks.
 	for i := range expectedNonce {
 		if expectedNonce[i] != encrypted.Nonce[i] {
 			return nil, fmt.Errorf("nonce mismatch: possible replay or reorder attack")
 		}
 	}
 
-	plaintext, err := s.gcm.Open(nil, encrypted.Nonce, encrypted.Ciphertext, nil)
+	plaintext, err := s.recv.Open(nil, encrypted.Nonce, encrypted.Ciphertext, nil)
 	if err != nil {
 		return nil, fmt.Errorf("decrypt message: %w", err)
 	}
@@ -110,11 +100,11 @@ func (s *encryptedStream) Receive() (*v1sync.SyncStreamItem, error) {
 	return &inner, nil
 }
 
-// makeNonce creates a 12-byte GCM nonce. The first byte is the direction prefix
-// (0x00 or 0x01), bytes 1-3 are zero, and bytes 4-11 are the counter in big-endian.
-func (s *encryptedStream) makeNonce(prefix byte, counter uint64) []byte {
-	nonce := make([]byte, s.gcm.NonceSize()) // 12 bytes for GCM
-	nonce[0] = prefix
-	binary.BigEndian.PutUint64(nonce[4:], counter)
+// makeNonce builds an N-byte AES-GCM nonce by big-endian-encoding the counter
+// in the trailing 8 bytes; the leading bytes are zero. The counter never
+// repeats within a single session direction, so the nonce never repeats.
+func makeNonce(size int, counter uint64) []byte {
+	nonce := make([]byte, size)
+	binary.BigEndian.PutUint64(nonce[size-8:], counter)
 	return nonce
 }

internal/api/syncapi/encryption_test.go

@@ -1,6 +1,7 @@
 package syncapi
 
 import (
+	"crypto/rand"
 	"sync"
 	"testing"
 
@@ -32,28 +33,33 @@ func newFakeStreamPair() (*fakeStream, *fakeStream) {
 	return &fakeStream{sendCh: ab, recvCh: ba}, &fakeStream{sendCh: ba, recvCh: ab}
 }
 
+// runHandshake performs the post-quantum KEM handshake between an initiator
+// and responder over a fakeStream pair, returning the resulting sessions in
+// (initiator, responder) order.
+func runHandshake(t *testing.T) (initiatorSess, responderSess *cryptoutil.TransportSession) {
+	t.Helper()
+	recipient, pubBytes, err := cryptoutil.NewTransportRecipient()
+	if err != nil {
+		t.Fatalf("NewTransportRecipient: %v", err)
+	}
+	enc, respSess, err := cryptoutil.EncapsulateToTransport(pubBytes)
+	if err != nil {
+		t.Fatalf("EncapsulateToTransport: %v", err)
+	}
+	initSess, err := recipient.Decapsulate(enc)
+	if err != nil {
+		t.Fatalf("Decapsulate: %v", err)
+	}
+	return initSess, respSess
+}
+
 func TestEncryptedStream_RoundTrip(t *testing.T) {
-	alice, err := cryptoutil.GenerateECDHKeyPair()
-	if err != nil {
-		t.Fatal(err)
-	}
-	bob, err := cryptoutil.GenerateECDHKeyPair()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	gcm, err := cryptoutil.DeriveSessionKey(alice.Private, bob.Public)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	aliceIsSmaller := string(alice.Public.Bytes()) < string(bob.Public.Bytes())
+	initSess, respSess := runHandshake(t)
 
 	transportA, transportB := newFakeStreamPair()
-	encA := newEncryptedStream(transportA, gcm, aliceIsSmaller)
-	encB := newEncryptedStream(transportB, gcm, !aliceIsSmaller)
+	encA := newEncryptedStream(transportA, initSess)
+	encB := newEncryptedStream(transportB, respSess)
 
-	// Send a heartbeat from A to B
 	sendItem := &v1sync.SyncStreamItem{
 		Action: &v1sync.SyncStreamItem_Heartbeat{
 			Heartbeat: &v1sync.SyncStreamItem_SyncActionHeartbeat{},
@@ -81,15 +87,11 @@ func TestEncryptedStream_RoundTrip(t *testing.T) {
 }
 
 func TestEncryptedStream_BidirectionalMultiMessage(t *testing.T) {
-	alice, _ := cryptoutil.GenerateECDHKeyPair()
-	bob, _ := cryptoutil.GenerateECDHKeyPair()
-	gcm, _ := cryptoutil.DeriveSessionKey(alice.Private, bob.Public)
-
-	aliceIsSmaller := string(alice.Public.Bytes()) < string(bob.Public.Bytes())
+	initSess, respSess := runHandshake(t)
 
 	transportA, transportB := newFakeStreamPair()
-	encA := newEncryptedStream(transportA, gcm, aliceIsSmaller)
-	encB := newEncryptedStream(transportB, gcm, !aliceIsSmaller)
+	encA := newEncryptedStream(transportA, initSess)
+	encB := newEncryptedStream(transportB, respSess)
 
 	heartbeat := &v1sync.SyncStreamItem{
 		Action: &v1sync.SyncStreamItem_Heartbeat{
@@ -97,7 +99,6 @@ func TestEncryptedStream_BidirectionalMultiMessage(t *testing.T) {
 		},
 	}
 
-	// Send 5 messages A→B sequentially, then 5 messages B→A sequentially
 	var wg sync.WaitGroup
 
 	// A→B direction
@@ -146,11 +147,13 @@ func TestEstablishEncryption_Integration(t *testing.T) {
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
-		encA, errA = establishEncryption(transportA)
+		// A is the initiator (client side).
+		encA, errA = establishEncryption(transportA, true)
 	}()
 	go func() {
 		defer wg.Done()
-		encB, errB = establishEncryption(transportB)
+		// B is the responder (server side).
+		encB, errB = establishEncryption(transportB, false)
 	}()
 	wg.Wait()
 
@@ -161,7 +164,6 @@ func TestEstablishEncryption_Integration(t *testing.T) {
 		t.Fatalf("establish B: %v", errB)
 	}
 
-	// Verify encrypted communication works
 	heartbeat := &v1sync.SyncStreamItem{
 		Action: &v1sync.SyncStreamItem_Heartbeat{
 			Heartbeat: &v1sync.SyncStreamItem_SyncActionHeartbeat{},
@@ -186,3 +188,29 @@ func TestEstablishEncryption_Integration(t *testing.T) {
 		t.Fatalf("expected heartbeat, got %T", recv.GetAction())
 	}
 }
+
+func TestEstablishEncryption_ProtocolVersionMismatch(t *testing.T) {
+	testutil.InstallZapLogger(t)
+	transportA, transportB := newFakeStreamPair()
+
+	// Responder receives a handshake with the wrong protocol version and
+	// must reject it with a protocol error.
+	junkPub := make([]byte, 32)
+	if _, err := rand.Read(junkPub); err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		_ = transportA.Send(&v1sync.SyncStreamItem{
+			Action: &v1sync.SyncStreamItem_EstablishSharedSecret{
+				EstablishSharedSecret: &v1sync.SyncStreamItem_SyncEstablishSharedSecret{
+					ProtocolVersion: cryptoutil.TransportProtocolVersion + 1,
+					KemPublicKey:    junkPub,
+				},
+			},
+		})
+	}()
+
+	if _, err := establishEncryption(transportB, false); err == nil {
+		t.Fatal("expected protocol version mismatch to fail handshake")
+	}
+}

internal/api/syncapi/syncclient.go

@@ -115,7 +115,7 @@ func (c *SyncClient) RunSync(ctx context.Context) {
 			cmdStream.SendErrorAndTerminate(err)
 		}()
 
-		connectErr := cmdStream.ConnectStream(ctx, c.client.Sync(ctx))
+		connectErr := cmdStream.ConnectStream(ctx, c.client.Sync(ctx), true /* isInitiator: client side */)
 		if connectErr != nil {
 			c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, connectErr)
 			var syncErr *SyncError

internal/api/syncapi/syncserver.go

@@ -64,7 +64,7 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre
 		cmdStream.SendErrorAndTerminate(err)
 	}()
 
-	if err := cmdStream.ConnectStream(ctx, stream); err != nil {
+	if err := cmdStream.ConnectStream(ctx, stream, false /* isInitiator: server side */); err != nil {
 		zap.S().Errorf("sync handler stream error: %v", err)
 		var syncErr *SyncError
 		if errors.As(err, &syncErr) {