diff --git a/.gitignore b/.gitignore index 9fae5b7..f28516f 100644 --- a/.gitignore +++ b/.gitignore @@ -134,4 +134,7 @@ dmypy.json .idea # vscode specific -.vscode \ No newline at end of file +.vscode + +# temp files +.tmp \ No newline at end of file diff --git a/src/signify/core/authing.py b/src/signify/core/authing.py index 5aa459a..9989e88 100644 --- a/src/signify/core/authing.py +++ b/src/signify/core/authing.py @@ -122,6 +122,27 @@ def approveDelegation(self, agent): self.serder = eventing.interact(pre=self.serder.pre, dig=self.serder.said, sn=self.serder.sn + 1, data=[anchor]) return self.serder, [self.signer.sign(self.serder.raw, index=0).qb64] + def _decryptSaltQb64(self, decrypter, cipher): + """ + Support decrypting both Salter and Streamer codes + Salter are fixed size, 1AAH class salts and variable size are Streamer 4C class codes. + + Returns: + salt as qualified base 64 whether a Salter or Streamer primitive + """ + plain = decrypter.decrypt(cipher=cipher, bare=True) + qb64 = plain.decode("utf-8") if hasattr(plain, "decode") else plain + + try: + salter = signing.Salter(qb64=qb64) + except Exception as ex: + raise kering.ValidationError("decrypted sxlt is not a Salt_128") from ex + + if salter.qb64 != qb64: + raise kering.ValidationError("decrypted sxlt must contain exactly one Salt_128") + + return qb64 + def rotate(self, nbran, aids): """ Rotate passcode involves re-encrypting all saved AID salts for salty keyed AIDs and @@ -186,7 +207,8 @@ def rotate(self, nbran, aids): decrypter = signing.Decrypter(seed=nsigner.qb64) # decrypter with old salt # First encrypt and save old Salt in case we need a recovery - sxlt = encrypter.encrypt(prim=coring.Matter(qb64b=self.bran)).qb64 + sxlt = encrypter.encrypt(prim=coring.Matter(qb64b=self.bran), + code=coring.MtrDex.X25519_Cipher_Salt).qb64 data = dict( rot=rot.ked, @@ -201,7 +223,7 @@ def rotate(self, nbran, aids): if "salty" in aid: salty = aid["salty"] cipher = signing.Cipher(qb64=salty["sxlt"]) - dnxt = decrypter.decrypt(cipher=cipher).qb64 + dnxt = self._decryptSaltQb64(decrypter, cipher) # Now we have the AID salt, use it to verify against the current public keys acreator = keeping.SaltyCreator(dnxt, stem=salty["stem"], tier=salty["tier"]) @@ -211,7 +233,8 @@ def rotate(self, nbran, aids): if pubs != [signer.verfer.qb64 for signer in signers]: raise kering.ValidationError(f"unable to rotate, validation of salt to public keys {pubs} failed") - asxlt = encrypter.encrypt(prim=coring.Matter(qb64=dnxt)).qb64 + asxlt = encrypter.encrypt(prim=coring.Matter(qb64=dnxt), + code=coring.MtrDex.X25519_Cipher_Salt).qb64 keys[pre] = dict( sxlt=asxlt ) diff --git a/src/signify/core/keeping.py b/src/signify/core/keeping.py index 92ec73b..6f853ae 100644 --- a/src/signify/core/keeping.py +++ b/src/signify/core/keeping.py @@ -192,10 +192,12 @@ def __init__(self, salter, pidx, kidx=0, tier=Tiers.low, transferable=False, ste if bran is not None: bran = coring.MtrDex.Salt_128 + 'A' + bran[:21] self.creator = keeping.SaltyCreator(salt=bran, stem=stem, tier=tier) - self.sxlt = self.encrypter.encrypt(ser=self.creator.salt).qb64 + self.sxlt = self.encrypter.encrypt(ser=self.creator.salt, + code=coring.MtrDex.X25519_Cipher_Salt).qb64 elif sxlt is None: self.creator = keeping.SaltyCreator(stem=stem, tier=tier) - self.sxlt = self.encrypter.encrypt(ser=self.creator.salt).qb64 + self.sxlt = self.encrypter.encrypt(ser=self.creator.salt, + code=coring.MtrDex.X25519_Cipher_Salt).qb64 else: self.sxlt = sxlt ciph = signing.Cipher(qb64=self.sxlt) diff --git a/tests/core/test_authing.py b/tests/core/test_authing.py index 8197a3e..cf04eea 100644 --- a/tests/core/test_authing.py +++ b/tests/core/test_authing.py @@ -8,7 +8,7 @@ import pytest from keri import kering -from keri.core import serdering +from keri.core import coring, serdering, signing from keri.core.coring import Tiers from keri.kering import Kinds, versify from mockito import mock, unstub, expect, verifyNoUnwantedInteractions @@ -248,6 +248,49 @@ def test_controller_rotate_salty(): assert 'ELUvZ8aJEHAQE-0nsevyYTP98rBbGJUrTj5an-pCmwrK' in out['keys'] assert 'sxlt' in out['keys']['ELUvZ8aJEHAQE-0nsevyYTP98rBbGJUrTj5an-pCmwrK'] # type: ignore assert out['keys']['ELUvZ8aJEHAQE-0nsevyYTP98rBbGJUrTj5an-pCmwrK']['sxlt'] != "1AAH2R_SPhr_5vIBGGtyVamaGVDQAcYlgmwDOkJwM-q6Qw8K5NT7jLzJ0k6_7sa3oyKK33ym8JX1Il4MoUiy8ixYwsVWYhaU3sMT" # type: ignore + assert signing.Cipher(qb64=out["sxlt"]).code == coring.MtrDex.X25519_Cipher_Salt + assert signing.Cipher(qb64=out["keys"]["ELUvZ8aJEHAQE-0nsevyYTP98rBbGJUrTj5an-pCmwrK"]["sxlt"]).code == coring.MtrDex.X25519_Cipher_Salt + + +def test_ctlr_rotate_migrates_on_write_4C_to_1AAH_salty_sxlt_cipher(): + from keri.app import keeping + from signify.core.authing import Controller + + ctrl = Controller(bran="abcdefghijklmnop01234", tier=Tiers.low) + signer = ctrl.salter.signer(transferable=False) + encrypter = signing.Encrypter(verkey=signer.verfer.qb64) + aid_salter = signing.Salter(raw=b'fedcba9876543210') + creator = keeping.SaltyCreator(salt=aid_salter.qb64, stem="signify:aid", tier=Tiers.low) + signers = creator.create(codes=["A"], pidx=0, kidx=0, transferable=False) + + # Cipher starts as 4C due to KERIpy Encrypter.encrypt default. + sxlt = encrypter.encrypt(ser=aid_salter.qb64).qb64 + prefix = "legacy-pre" + + assert signing.Cipher(qb64=sxlt).code == coring.MtrDex.X25519_Cipher_L0 + + aid = { + "name": "aid1", + "prefix": prefix, + "salty": { + "pidx": 0, + "stem": "signify:aid", + "sxlt": sxlt, + "tier": Tiers.low, + "icodes": ["A"], + "kidx": 0, + "transferable": False, + }, + "state": { + "k": [signer.verfer.qb64 for signer in signers], + }, + } + # Rotate will read the 4C encoded CESR primitive and migrate it on-write to the 1AAH code + # that is compatible with SignifyTS. + out = ctrl.rotate(nbran="0123456789abcdefghijk", aids=[aid]) + + assert signing.Cipher(qb64=out["sxlt"]).code == coring.MtrDex.X25519_Cipher_Salt + assert signing.Cipher(qb64=out["keys"][prefix]["sxlt"]).code == coring.MtrDex.X25519_Cipher_Salt def test_controller_rotate_randy(): from signify.core.authing import Controller diff --git a/tests/core/test_keeping.py b/tests/core/test_keeping.py index 5332971..e05b54e 100644 --- a/tests/core/test_keeping.py +++ b/tests/core/test_keeping.py @@ -7,6 +7,7 @@ """ import pytest +from keri.core import coring as core_coring from mockito import mock, verifyNoUnwantedInteractions, unstub, expect, when @@ -234,7 +235,9 @@ def test_salty_keeper(): from keri.core.signing import Cipher mock_cipher = mock({'qb64': 'cipher qb64'}, spec=Cipher, strict=True) - expect(mock_encrypter, times=1).encrypt(ser='creator salt').thenReturn(mock_cipher) + expect(mock_encrypter, times=1).encrypt( + ser='creator salt', + code=core_coring.MtrDex.X25519_Cipher_Salt).thenReturn(mock_cipher) # test from signify.core.keeping import SaltyKeeper @@ -277,7 +280,9 @@ def test_salty_keeper_bran(): from keri.core.signing import Cipher mock_cipher = mock({'qb64': 'cipher qb64'}, spec=Cipher, strict=True) - expect(mock_encrypter, times=1).encrypt(ser='creator salt').thenReturn(mock_cipher) + expect(mock_encrypter, times=1).encrypt( + ser='creator salt', + code=core_coring.MtrDex.X25519_Cipher_Salt).thenReturn(mock_cipher) # test from signify.core.keeping import SaltyKeeper @@ -288,6 +293,36 @@ def test_salty_keeper_bran(): verifyNoUnwantedInteractions() unstub() + +def test_salty_keeper_bran_sxlt_uses_x25519_cipher_salt(): + from keri.core import signing + from signify.core.keeping import SaltyKeeper + + salter = signing.Salter(raw=b'0123456789abcdef') + sk = SaltyKeeper(salter, pidx=0, bran='0123456789abcdefghijk') + + cipher = signing.Cipher(qb64=sk.params()["sxlt"]) + assert cipher.code == core_coring.MtrDex.X25519_Cipher_Salt + + +def test_salty_keeper_accepts_stream_cipher_salt(): + from keri.core import signing + from signify.core.keeping import SaltyKeeper + + salter = signing.Salter(raw=b'0123456789abcdef') + signer = salter.signer(transferable=False) + encrypter = signing.Encrypter(verkey=signer.verfer.qb64) + aid_salter = signing.Salter(raw=b'fedcba9876543210') + sxlt = encrypter.encrypt(ser=aid_salter.qb64).qb64 # an unspecified code defaults to 4C variable size cipher + + assert signing.Cipher(qb64=sxlt).code == core_coring.MtrDex.X25519_Cipher_L0 + + sk = SaltyKeeper(salter, pidx=0, sxlt=sxlt) + + assert sk.params()["sxlt"] == sxlt + assert sk.creator.salt == aid_salter.qb64 + + def test_salty_keeper_sxlt(): # salty keep init mocks from keri.core.signing import Salter, Signer @@ -361,7 +396,9 @@ def test_salty_keeper_incept(): from keri.core.signing import Cipher mock_cipher = mock({'qb64': 'cipher qb64'}, spec=Cipher, strict=True) - expect(mock_encrypter, times=1).encrypt(ser='creator salt').thenReturn(mock_cipher) + expect(mock_encrypter, times=1).encrypt( + ser='creator salt', + code=core_coring.MtrDex.X25519_Cipher_Salt).thenReturn(mock_cipher) # incept mocks mock_incept_verfer = mock({'qb64': 'incept verfer qb64'}, spec=Verfer, strict=True) @@ -417,7 +454,9 @@ def test_salty_keeper_rotate(): from keri.core.signing import Cipher mock_cipher = mock({'qb64': 'cipher qb64'}, spec=Cipher, strict=True) - expect(mock_encrypter, times=1).encrypt(ser='creator salt').thenReturn(mock_cipher) + expect(mock_encrypter, times=1).encrypt( + ser='creator salt', + code=core_coring.MtrDex.X25519_Cipher_Salt).thenReturn(mock_cipher) # rotate mocks mock_rotate_verfer = mock({'qb64': 'rotate verfer qb64'}, spec=Verfer, strict=True) @@ -475,7 +514,9 @@ def test_salty_keeper_sign(): from keri.core.signing import Cipher mock_cipher = mock({'qb64': 'cipher qb64'}, spec=Cipher, strict=True) - expect(mock_encrypter, times=1).encrypt(ser='creator salt').thenReturn(mock_cipher) + expect(mock_encrypter, times=1).encrypt( + ser='creator salt', + code=core_coring.MtrDex.X25519_Cipher_Salt).thenReturn(mock_cipher) # sign mock expect(mock_creator, times=1).create(codes=['A'], pidx=0, kidx=0, transferable=False).thenReturn([mock_signer])