fix: use random IVs for stored keys (put/get), matching the Dart SDK

.github/workflows/interop.yml

@@ -0,0 +1,85 @@
+# DRAFT — opt-in cross-SDK IV interop test (Python at_client <-> Dart at_client).
+# Manual only (workflow_dispatch). Review before enabling on PRs: it starts the
+# ephemeral environment container, onboards two test atSigns via CRAM, installs the
+# SDK, and runs test/interop_test.py with AT_INTEROP=1.
+name: interop (Python <-> Dart)
+
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  interop:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    services:
+      ee:
+        image: atsigncompany/ephemeral
+        env:
+          DNS_FQDN: vip.ve.atsign.zone
+          FIRST_PORT: 2500
+        ports:
+          - 64:64
+          - 2500-2540:2500-2540
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Resolve the EE FQDN to localhost (matches the cert CN)
+        run: echo "127.0.0.1 vip.ve.atsign.zone" | sudo tee -a /etc/hosts
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+      - uses: dart-lang/setup-dart@v1
+
+      - name: Install the SDK (repo generates README.PyPI.md at publish time)
+        run: |
+          cp README.md README.PyPI.md
+          pip install .
+
+      - name: Wait for the EE root, then onboard @alpha and @bravo (CRAM)
+        run: |
+          export HOME=/tmp/eehome
+          mkdir -p "$HOME/.atsign/keys"
+          # give the root time to register the atServers
+          sleep 30
+          cid=$(docker ps --filter ancestor=atsigncompany/ephemeral --format '{{.ID}}' | head -1)
+          python - "$cid" <<'PY'
+          import subprocess, sys
+          from at_client.connections import Address, AtRootConnection, AtSecondaryConnection
+          from at_client.common import AtSign
+          from at_client.util import AuthUtil, OnboardingUtil, KeysUtil
+          cid = sys.argv[1]
+          root = Address.from_string("vip.ve.atsign.zone:64")
+          for name in ("alpha", "bravo"):
+              atsign = AtSign("@" + name)
+              cram = subprocess.check_output(
+                  ["docker", "exec", cid, "cat", f"/atsign/atservers/{name}/CRAM"]).decode().strip()
+              sec = AtRootConnection.get_instance(root.host, root.port).find_secondary(atsign)
+              conn = AtSecondaryConnection(sec); conn.connect()
+              auth, ob = AuthUtil(), OnboardingUtil()
+              auth.authenticate_with_cram(conn, atsign, cram)
+              keys = {}
+              ob.generate_self_encryption_key(keys); ob.generate_pkam_keypair(keys)
+              ob.generate_encryption_keypair(keys); KeysUtil.save_keys(atsign, keys)
+              ob.store_pkam_public_key(conn, keys)
+              auth.authenticate_with_pkam(conn, atsign, KeysUtil.load_keys(atsign))
+              ob.store_public_encryption_key(conn, atsign.without_prefix, keys)
+              ob.delete_cram_key(conn)
+              print("onboarded @" + name)
+          PY
+
+      - name: dart pub get (interop helper)
+        run: dart pub get
+        working-directory: test/interop
+
+      - name: Run interop test
+        run: |
+          # Start from clean Dart local storage (keeps .atKeys). Harmless on a fresh
+          # runner; prevents stale-key errors when re-running against a recreated EE.
+          rm -rf /tmp/eehome/.atsign/storage
+          HOME=/tmp/eehome AT_INTEROP=1 \
+            AT_ROOT=vip.ve.atsign.zone:64 AT_ROOT_DOMAIN=vip.ve.atsign.zone \
+            python -m unittest discover -s test -p 'interop_test.py' -v

at_client/atclient.py

@@ -21,6 +21,11 @@
 from .common.keys import AtKey, Keys, SharedKey, PrivateHiddenKey, PublicKey, SelfKey
 from .util.authutil import AuthUtil
 
+# The 16-byte all-zero IV used before random IVs existed. Matches the Dart SDK's
+# AtChopsUtil.generateIVLegacy() / getIV(null); used to decrypt legacy data (no ivNonce).
+LEGACY_IV = b"\x00" * 16
+
+
 class AtClient(ABC):
     def __init__(self, atsign:AtSign, root_address:Address=Address("root.atsign.org", 64), secondary_address:Address=None, queue:Queue=None, verbose:bool = False):
         self.atsign = atsign
@@ -174,6 +179,13 @@ def get_encryption_key_shared_by_other(self, shared_key: SharedKey):
 
 
     def put(self, key, value):
+        # Generate the IV once here, before dispatching to the per-type encryptor —
+        # mirroring the Dart SDK's AtClientImpl._putInternal, which sets a random ivNonce
+        # for every put ahead of encryption. This guarantees every encrypted put gets a
+        # random IV regardless of key type; the per-type _put_* methods keep a `?=`
+        # backstop so direct calls are safe too. Public keys aren't encrypted, so no IV.
+        if isinstance(key, (SharedKey, SelfKey)) and key.metadata.iv_nonce is None:
+            key.metadata.iv_nonce = EncryptionUtil.generate_iv_nonce()
         if isinstance(key, SharedKey):
             return self._put_shared_key(key, value)
         elif isinstance(key, SelfKey):
@@ -186,8 +198,16 @@ def put(self, key, value):
     def _put_self_key(self, key: SelfKey, value: str):
         key.metadata.data_signature = EncryptionUtil.sign_sha256_rsa(value, self.keys[KeysUtil.encryption_private_key_name])
 
+        # Generate a random IV per self key too (stored as ivNonce in metadata). Dart's
+        # current SelfKeyEncryption does NOT do this (it uses the zero IV) — that's a
+        # security gap on the Dart side; doing it here is interop-safe because get falls
+        # back to the legacy zero IV when ivNonce is absent, and Dart's self decrypt
+        # already honors ivNonce when present.
+        if key.metadata.iv_nonce is None:
+            key.metadata.iv_nonce = EncryptionUtil.generate_iv_nonce()
+        self_key = self.keys[KeysUtil.self_encryption_key_name]
         try:
-            cipher_text = EncryptionUtil.aes_encrypt_from_base64(value, self.keys[KeysUtil.self_encryption_key_name])
+            cipher_text = EncryptionUtil.aes_encrypt_from_base64(value, self_key, key.metadata.iv_nonce)
         except Exception as e:
             raise AtEncryptionException(f"Failed to encrypt value with self encryption key - {e}")
 
@@ -218,7 +238,11 @@ def _put_shared_key(self, key: SharedKey, value: str):
             share_to_encryption_key = self.get_encryption_key_shared_by_me(key)
 
             what = "encrypt value with shared encryption key"
-            cipher_text = EncryptionUtil.aes_encrypt_from_base64(value, share_to_encryption_key)
+            # Match the Dart SDK: shared keys always get a random IV, persisted as
+            # ivNonce in the key metadata (which is serialized into the update command).
+            if key.metadata.iv_nonce is None:
+                key.metadata.iv_nonce = EncryptionUtil.generate_iv_nonce()
+            cipher_text = EncryptionUtil.aes_encrypt_from_base64(value, share_to_encryption_key, key.metadata.iv_nonce)
         except Exception as e:
             raise AtEncryptionException(f"Failed to {what} - {e}")
 
@@ -255,6 +279,17 @@ def get_lookup_response(self, command: str):
         return fetched
 
 
+    @staticmethod
+    def _iv_from_fetched(fetched) -> bytes:
+        """Read the IV from a fetched (all) lookup response's metadata.
+
+        Returns the ivNonce bytes if present, else the legacy all-zero IV — matching
+        the Dart SDK's `metadata.ivNonce != null ? fromBase64 : generateIVLegacy()`.
+        """
+        meta = fetched.get("metaData") or {}
+        iv_b64 = meta.get("ivNonce")
+        return base64.b64decode(iv_b64) if iv_b64 else LEGACY_IV
+
     def _get_self_key(self, key: SelfKey):
         command = LlookupVerbBuilder().with_at_key(key, LlookupVerbBuilder.Type.ALL).build()
 
@@ -263,8 +298,9 @@ def _get_self_key(self, key: SelfKey):
         decrypted_value = None
         encrypted_value = fetched["data"]
         self_encryption_key = self.keys[KeysUtil.self_encryption_key_name]
+        iv = self._iv_from_fetched(fetched)
         try:
-            decrypted_value = EncryptionUtil.aes_decrypt_from_base64(encrypted_value, self_encryption_key)
+            decrypted_value = EncryptionUtil.aes_decrypt_from_base64(encrypted_value, self_encryption_key, iv)
         except Exception as e:
             raise AtDecryptionException(f"Failed to {command} - {e}")
 
@@ -296,38 +332,31 @@ def _get_shared_key(self, key: SharedKey):
     def _get_shared_by_me_with_other(self, shared_key: SharedKey):
         share_encryption_key = self.get_encryption_key_shared_by_me(shared_key)
 
-        raw_response = None
-        command = "llookup:" + str(shared_key)
-        try:
-            raw_response = self.secondary_connection.execute_command(command, True)
-        except (AtKeyNotFoundException, AtInternalServerException) as e: raise e
-        except AtSecondaryConnectException as e:
-            raise AtSecondaryConnectException(f"Failed to execute {command} - {e}")
+        # Use llookup:all so we get the metadata (ivNonce) alongside the value.
+        command = "llookup:all:" + str(shared_key)
+        fetched = self.get_lookup_response(command)
+        iv = self._iv_from_fetched(fetched)
 
         try:
-            return EncryptionUtil.aes_decrypt_from_base64(raw_response.get_raw_data_response(), share_encryption_key)
+            return EncryptionUtil.aes_decrypt_from_base64(fetched["data"], share_encryption_key, iv)
         except Exception as e:
             raise AtDecryptionException(f"Failed to decrypt value with shared encryption key - {e}")
 
     def _get_shared_by_other_with_me(self, shared_key:SharedKey):
-        what = None
         share_encryption_key = self.get_encryption_key_shared_by_other(shared_key)
 
-        raw_response = None
-        command = "lookup:" + shared_key.name
+        # Use lookup:all so we get the metadata (ivNonce) alongside the value.
+        command = "lookup:all:" + shared_key.name
         if shared_key.get_namespace() is not None and shared_key.get_namespace():
             command += "." + shared_key.get_namespace()
         command += str(shared_key.shared_by)
-        try:
-            raw_response = self.secondary_connection.execute_command(command, True)
-        except Exception as e:
-            raise AtSecondaryConnectException(f"Failed to execute {command} - {e}")
+        fetched = self.get_lookup_response(command)
+        iv = self._iv_from_fetched(fetched)
 
-        what = "decrypt value with shared encryption key"
         try:
-            return EncryptionUtil.aes_decrypt_from_base64(raw_response.get_raw_data_response(), share_encryption_key)
+            return EncryptionUtil.aes_decrypt_from_base64(fetched["data"], share_encryption_key, iv)
         except Exception as e:
-            raise AtDecryptionException(f"Failed to {what} - {e}")
+            raise AtDecryptionException(f"Failed to decrypt value with shared encryption key - {e}")
 
     def delete(self, key):
         if isinstance(key, SharedKey) or isinstance(key, SelfKey) or isinstance(key, PublicKey):

at_client/common/metadata.py

@@ -68,8 +68,11 @@ def from_json(json_str):
         metadata.shared_key_enc = data.get('sharedKeyEnc')
         metadata.pub_key_cs = data.get('pubKeyCS')
         metadata.encoding = data.get('encoding')
-        metadata.iv_nonce = data.get('ivNonce')
-
+        # ivNonce travels as base64 on the wire; keep it as raw bytes internally so it
+        # round-trips with generate_iv_nonce()/__str__ and is usable directly as an IV.
+        _iv = data.get('ivNonce')
+        metadata.iv_nonce = binascii.a2b_base64(_iv) if _iv else None
+
         return metadata
 
     @staticmethod
@@ -100,7 +103,9 @@ def from_dict(data_dict):
         metadata.shared_key_enc = data_dict.get('sharedKeyEnc')
         metadata.pub_key_cs = data_dict.get('pubKeyCS')
         metadata.encoding = data_dict.get('encoding')
-        metadata.iv_nonce = data_dict.get('ivNonce')
+        # ivNonce travels as base64; keep raw bytes internally (see from_json).
+        _iv = data_dict.get('ivNonce')
+        metadata.iv_nonce = binascii.a2b_base64(_iv) if _iv else None
 
         return metadata
 

at_client/util/verbbuilder.py

@@ -181,6 +181,7 @@ def set_metadata(self, metadata):
         self.shared_key_enc = metadata.shared_key_enc
         self.pub_key_cs = metadata.pub_key_cs
         self.encoding = metadata.encoding
+        self.iv_nonce = metadata.iv_nonce
         return self
 
     def with_at_key(self, at_key, value):
@@ -228,6 +229,7 @@ def _build_metadata_str(self):
         metadata.shared_key_enc = self.shared_key_enc
         metadata.pub_key_cs = self.pub_key_cs
         metadata.encoding = self.encoding
+        metadata.iv_nonce = getattr(self, "iv_nonce", None)
         return str(metadata)
 
 class LlookupVerbBuilder: