feat: Notify the user of a modified device

lib/omemo_dart.dart

@@ -2,6 +2,7 @@ library omemo_dart;
 
 export 'src/double_ratchet/double_ratchet.dart';
 export 'src/errors.dart';
+export 'src/events.dart';
 export 'src/helpers.dart';
 export 'src/keys.dart';
 export 'src/omemo/bundle.dart';

lib/src/events.dart

@@ -0,0 +1,11 @@
+import 'package:omemo_dart/src/omemo/device.dart';
+
+abstract class OmemoEvent {}
+
+/// Triggered by the OmemoSessionManager when our own device bundle was modified
+/// and thus should be republished.
+class DeviceBundleModifiedEvent extends OmemoEvent {
+
+  DeviceBundleModifiedEvent(this.device);
+  final Device device;
+}

lib/src/helpers.dart

@@ -12,7 +12,7 @@ List<int> concat(List<List<int>> inputs) {
 
 /// Compares the two lists [a] and [b] and return true if [a] and [b] are index-by-index
 /// equal. Returns false, if they are not "equal";
-bool listsEqual(List<int> a, List<int> b) {
+bool listsEqual<T>(List<T> a, List<T> b) {
   // TODO(Unknown): Do we need to use a constant time comparison?
   if (a.length != b.length) return false;
 

lib/src/omemo/device.dart

@@ -47,7 +47,7 @@ class Device {
   /// This replaces the Onetime-Prekey with id [id] with a completely new one. Returns
   /// a new Device object that copies over everything but replaces said key.
   Future<Device> replaceOnetimePrekey(int id) async {
-    final newOpk = await OmemoKeyPair.generateNewPair(KeyPairType.x25519);
+    opks[id] = await OmemoKeyPair.generateNewPair(KeyPairType.x25519);
     
     return Device(
       id,
@@ -55,13 +55,7 @@ class Device {
       spk,
       spkId,
       spkSignature,
-      opks.map((keyId, opk) {
+      opks,
-        if (keyId == id) {
-          return MapEntry(id, newOpk);
-        }
-
-        return MapEntry(id, opk);
-      }),
     );
   }
 

lib/src/omemo/sessionmanager.dart

@@ -1,9 +1,11 @@
+import 'dart:async';
 import 'dart:convert';
 import 'package:collection/collection.dart';
 import 'package:cryptography/cryptography.dart';
 import 'package:omemo_dart/src/crypto.dart';
 import 'package:omemo_dart/src/double_ratchet/double_ratchet.dart';
 import 'package:omemo_dart/src/errors.dart';
+import 'package:omemo_dart/src/events.dart';
 import 'package:omemo_dart/src/helpers.dart';
 import 'package:omemo_dart/src/keys.dart';
 import 'package:omemo_dart/src/omemo/bundle.dart';
@@ -39,7 +41,12 @@ class EncryptedKey {
 
 class OmemoSessionManager {
 
-  OmemoSessionManager(this.device) : _ratchetMap = {}, _deviceMap = {}, _lock = Lock();
+  OmemoSessionManager(this._device)
+    : _ratchetMap = {},
+      _deviceMap = {},
+      _lock = Lock(),
+      _deviceLock = Lock(),
+      _eventStreamController = StreamController<OmemoEvent>.broadcast();
 
   /// Generate a new cryptographic identity.
   static Future<OmemoSessionManager> generateNewIdentity({ int opkAmount = 100 }) async {
@@ -58,8 +65,26 @@ class OmemoSessionManager {
   /// Mapping of a bare Jid to its Device Ids
   final Map<String, List<int>> _deviceMap;
 
-  /// Our own keys
+  /// The event bus of the session manager
-  Device device;
+  final StreamController<OmemoEvent> _eventStreamController;
+  
+  /// Our own keys...
+  // ignore: prefer_final_fields
+  Device _device;
+  /// and its lock
+  final Lock _deviceLock;
+
+  /// A stream that receives events regarding the session
+  Stream<OmemoEvent> get eventStream => _eventStreamController.stream;
+
+  Future<Device> getDevice() async {
+    Device? dev;
+    await _deviceLock.synchronized(() async {
+      dev = _device;
+    });
+
+    return dev!;
+  }
   
   /// Add a session [ratchet] with the [deviceId] to the internal tracking state.
   Future<void> addSession(String jid, int deviceId, OmemoDoubleRatchet ratchet) async {
@@ -84,6 +109,7 @@ class OmemoSessionManager {
   /// Create a ratchet session initiated by Alice to the user with Jid [jid] and the device
   /// [deviceId] from the bundle [bundle].
   Future<OmemoKeyExchange> addSessionFromBundle(String jid, int deviceId, OmemoBundle bundle) async {
+    final device = await getDevice();
     final kexResult = await x3dhFromBundle(
       bundle,
       device.ik,
@@ -105,8 +131,8 @@ class OmemoSessionManager {
 
   /// Build a new session with the user at [jid] with the device [deviceId] using data
   /// from the key exchange [kex].
-  // TODO(PapaTutuWawa): Replace the OPK
   Future<void> addSessionFromKeyExchange(String jid, int deviceId, OmemoKeyExchange kex) async {
+    final device = await getDevice();
     final kexResult = await x3dhFromInitialMessage(
       X3DHMessage(
         OmemoPublicKey.fromBytes(kex.ik!, KeyPairType.ed25519),
@@ -189,6 +215,7 @@ class OmemoSessionManager {
   /// <encrypted /> element.
   Future<String> decryptMessage(List<int> ciphertext, String senderJid, int senderDeviceId, List<EncryptedKey> keys) async {
     // Try to find a session we can decrypt with.
+    var device = await getDevice();
     final rawKey = keys.firstWhereOrNull((key) => key.rid == device.id);
     if (rawKey == null) {
       throw NotEncryptedForDeviceException();
@@ -206,6 +233,14 @@ class OmemoSessionManager {
       );
 
       authMessage = kex.message!;
+
+      // Replace the OPK
+      await _deviceLock.synchronized(() async {
+        device = await device.replaceOnetimePrekey(kex.pkId!);
+        _eventStreamController.add(
+          DeviceBundleModifiedEvent(device),
+        );
+      });
     } else {
       authMessage = OmemoAuthenticatedMessage.fromBuffer(decodedRawKey);
     }

test/omemo_test.dart

@@ -7,8 +7,15 @@ void main() {
     const bobJid = 'bob@other.server.example';
       
     // Alice and Bob generate their sessions
+    var deviceModified = false;
     final aliceSession = await OmemoSessionManager.generateNewIdentity(opkAmount: 1);
     final bobSession = await OmemoSessionManager.generateNewIdentity(opkAmount: 1);
+    final bobOpks = (await bobSession.getDevice()).opks.values.toList();
+    bobSession.eventStream.listen((event) {
+      if (event is DeviceBundleModifiedEvent) {
+        deviceModified = true;
+      }
+    });
 
     // Alice encrypts a message for Bob
     const messagePlaintext = 'Hello Bob!';
@@ -16,7 +23,7 @@ void main() {
       bobJid,
       messagePlaintext,
       newSessions: [
-        await bobSession.device.toBundle(),
+        await (await bobSession.getDevice()).toBundle(),
       ],
     );
     expect(aliceMessage.encryptedKeys.length, 1);
@@ -28,10 +35,17 @@ void main() {
     final bobMessage = await bobSession.decryptMessage(
       aliceMessage.ciphertext,
       aliceJid,
-      aliceSession.device.id,
+      (await aliceSession.getDevice()).id,
       aliceMessage.encryptedKeys,
     );
     expect(messagePlaintext, bobMessage);
+    // The event should be triggered
+    expect(deviceModified, true);
+    // Bob should have replaced his OPK
+    expect(
+      listsEqual(bobOpks, (await bobSession.getDevice()).opks.values.toList()),
+      false,
+    );
 
     // Bob responds to Alice
     const bobResponseText = 'Oh, hello Alice!';
@@ -47,7 +61,7 @@ void main() {
     final aliceReceivedMessage = await aliceSession.decryptMessage(
       bobResponseMessage.ciphertext,
       bobJid,
-      bobSession.device.id,
+      (await bobSession.getDevice()).id,
       bobResponseMessage.encryptedKeys,
     );
     expect(bobResponseText, aliceReceivedMessage);
@@ -69,8 +83,8 @@ void main() {
       bobJid,
       messagePlaintext,
       newSessions: [
-        await bobSession.device.toBundle(),
+        await (await bobSession.getDevice()).toBundle(),
-        await bobSession2.device.toBundle(),
+        await (await bobSession2.getDevice()).toBundle(),
       ],
     );
     expect(aliceMessage.encryptedKeys.length, 2);
@@ -84,7 +98,7 @@ void main() {
     final bobMessage = await bobSession.decryptMessage(
       aliceMessage.ciphertext,
       aliceJid,
-      aliceSession.device.id,
+      (await aliceSession.getDevice()).id,
       aliceMessage.encryptedKeys,
     );
     expect(messagePlaintext, bobMessage);
@@ -103,7 +117,7 @@ void main() {
     final aliceReceivedMessage = await aliceSession.decryptMessage(
       bobResponseMessage.ciphertext,
       bobJid,
-      bobSession.device.id,
+      (await bobSession.getDevice()).id,
       bobResponseMessage.encryptedKeys,
     );
     expect(bobResponseText, aliceReceivedMessage);