diff --git a/cid/cid.py b/cid/cid.py index 7a604b6..a923e97 100644 --- a/cid/cid.py +++ b/cid/cid.py @@ -195,6 +195,19 @@ def __init__(self, multihash: str | bytes) -> None: """ :param bytes multihash: multihash for the CID """ + multihash_bytes = ensure_bytes(multihash) + # Validate multihash is sha2-256 with 32-byte digest + try: + mh_info = mh.decode(multihash_bytes) + except Exception as e: + raise ValueError(f"invalid multihash for CIDv0: {e}") from e + + if mh_info.code != 0x12 or mh_info.length != 32: + raise ValueError( + f"invalid hash for CIDv0: must be sha2-256 with 32-byte digest, " + f"got {mh_info.name} with {mh_info.length}-byte digest" + ) + super().__init__(0, self.CODEC, multihash) @property diff --git a/newsfragments/63.bugfix.rst b/newsfragments/63.bugfix.rst new file mode 100644 index 0000000..863f9a6 --- /dev/null +++ b/newsfragments/63.bugfix.rst @@ -0,0 +1 @@ +Validate multihash correctness in CIDv0 constructor. diff --git a/tests/test_cid.py b/tests/test_cid.py index d79fc6b..9554eb0 100644 --- a/tests/test_cid.py +++ b/tests/test_cid.py @@ -48,6 +48,18 @@ def test_encode(self, cid): def test_str(self, cid): assert str(cid) == ensure_unicode(cid.encode()) + def test_cidv0_rejects_non_sha256(self): + digest = hashlib.sha3_512(b"hello").digest() + mh_bytes = multihash.encode(digest, "sha3-512") + with pytest.raises(ValueError, match="must be sha2-256"): + CIDv0(mh_bytes) + + def test_cidv0_rejects_truncated_digest(self): + digest = hashlib.sha256(b"hello").digest()[:16] + mh_bytes = multihash.encode(digest, "sha2-256", 16) + with pytest.raises(ValueError, match="32-byte digest"): + CIDv0(mh_bytes) + class TestCIDv1: TEST_CODEC = "dag-pb" @@ -88,8 +100,8 @@ def test_cidv0_eq_cidv0(self, test_hash): def test_cidv0_neq(self): """Check for inequality for CIDv0 for different hashes""" - assert CIDv0(b"QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n") != CIDv0( - b"QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1o", + assert CIDv0(base58.b58decode(b"QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1n")) != CIDv0( + base58.b58decode(b"QmdfTbBqBPQ7VNxZEYEj14VmRuZBkqFbiwReogJgS1zR1o"), ) def test_cidv0_eq_cidv1(self, test_hash): diff --git a/tests/test_new_features.py b/tests/test_new_features.py index de98550..04c766e 100644 --- a/tests/test_new_features.py +++ b/tests/test_new_features.py @@ -202,7 +202,9 @@ def test_cid_set_has(self, cidv0): cid_set = CIDSet() cid_set.add(cidv0) assert cid_set.has(cidv0) - assert not cid_set.has(CIDv0(b"different")) + different_digest = hashlib.sha256(b"different").digest() + different_mh = multihash.encode(different_digest, "sha2-256") + assert not cid_set.has(CIDv0(different_mh)) def test_cid_set_remove(self, cidv0): """CIDSet.remove: removes CID from set""" @@ -265,7 +267,9 @@ def test_cid_set_contains(self, cidv0): cid_set = CIDSet() cid_set.add(cidv0) assert cidv0 in cid_set - assert CIDv0(b"different") not in cid_set + different_digest = hashlib.sha256(b"different").digest() + different_mh = multihash.encode(different_digest, "sha2-256") + assert CIDv0(different_mh) not in cid_set def test_cid_set_iter(self, cidv0, cidv1): """CIDSet.__iter__: makes set iterable"""