Add CID to sha256 digest conversion support

Author: MythicManiac

README.md

@@ -40,6 +40,9 @@ assert result == "bafkreide5semuafsnds3ugrvm6fbwuyw2ijpj43gwjdxemstjkfozi37hq"
 
 ### Wrap an existing SHA 256 checksum as a CID
 
+**WARNING:** This will lead to an invalid CID if an invalid digest is provided.
+This is not possible to validate against without the original data.
+
 ```python
 from hashlib import sha256
 from ipfs_cid import cid_sha256_wrap_digest
@@ -50,6 +53,23 @@ result = cid_sha256_wrap_digest(digest)
 assert result == "bafkreide5semuafsnds3ugrvm6fbwuyw2ijpj43gwjdxemstjkfozi37hq"
 ```
 
+### Unwrap a compatible CID to a sha256 digest
+
+**NOTE:** The `cid_sha256_unwrap_digest` function will throw an `AttributeError`
+if the input CID is not using the same encoding parameters.
+
+```python
+from hashlib import sha256
+from ipfs_cid import cid_sha256_unwrap_digest
+
+data = b"Hello world"
+digest = sha256(data).digest()
+
+cid = "bafkreide5semuafsnds3ugrvm6fbwuyw2ijpj43gwjdxemstjkfozi37hq"
+result = cid_sha256_unwrap_digest(cid)
+assert result == digest
+```
+
 ## Encoding Format
 
 [The CID spec](https://github.com/multiformats/cid) supports multiple different

ipfs_cid/__init__.py

@@ -1 +1,2 @@
-from .encode import cid_sha256_hash, cid_sha256_hash_chunked, cid_sha256_wrap_digest
+from .cid import cid_sha256_unwrap_digest, cid_sha256_wrap_digest
+from .sha256 import cid_sha256_hash, cid_sha256_hash_chunked

ipfs_cid/cid.py

@@ -0,0 +1,32 @@
+from ipfs_cid.base32 import Base32Multibase
+
+MULTICODEC_CIDV1 = b"\x01"
+MULTICODEC_RAW_BINARY = b"\x55"
+MULTICODEC_SHA_2_256 = b"\x12"
+MULTICODEC_SHA_2_256_LENGTH = b"\x20"
+
+CID_PREFIX = b"".join(
+    [
+        MULTICODEC_CIDV1,
+        MULTICODEC_RAW_BINARY,
+        MULTICODEC_SHA_2_256,
+        MULTICODEC_SHA_2_256_LENGTH,
+    ],
+)
+
+
+def cid_sha256_wrap_digest(digest: bytes) -> str:
+    digestlen = len(digest).to_bytes(1, "big")
+    if digestlen != MULTICODEC_SHA_2_256_LENGTH:
+        raise AttributeError("Invalid digest length")
+    return Base32Multibase.encode(CID_PREFIX + digest)
+
+
+def cid_sha256_unwrap_digest(cid: str) -> bytes:
+    decoded = Base32Multibase.decode(cid)
+    if decoded[: len(CID_PREFIX)] != CID_PREFIX:
+        raise AttributeError("Unsupported CID format")
+    digest = decoded[len(CID_PREFIX) :]
+    if len(digest).to_bytes(1, "big") != MULTICODEC_SHA_2_256_LENGTH:
+        raise AttributeError("Unsupported CID format")
+    return digest

ipfs_cid/encode.py

@@ -0,0 +1,19 @@
+import array
+from hashlib import sha256
+from typing import Iterable, Union
+
+from ipfs_cid.cid import cid_sha256_wrap_digest
+
+ENCODEABLE_TYPES = Union[bytes, bytearray, memoryview, array.array]
+
+
+def cid_sha256_hash(data: ENCODEABLE_TYPES) -> str:
+    sha = sha256(data)
+    return cid_sha256_wrap_digest(sha.digest())
+
+
+def cid_sha256_hash_chunked(data: Iterable[ENCODEABLE_TYPES]) -> str:
+    sha = sha256()
+    for chunk in data:
+        sha.update(chunk)
+    return cid_sha256_wrap_digest(sha.digest())

ipfs_cid/sha256.py

@@ -0,0 +1,23 @@
+# Comparison targets were generated with Kubo CLI
+TEST_SETS = [
+    (
+        lambda: b"Hello world",
+        "bafkreide5semuafsnds3ugrvm6fbwuyw2ijpj43gwjdxemstjkfozi37hq",
+    ),
+    (
+        lambda: b"\x00" * 200000,
+        "bafkreicmxpm34df2nbmdk5k7qj3vq4c5wwsbhrkjjq2cmlgslfdkoptvqi",
+    ),
+    (
+        lambda: b"\x00" * 500000,
+        "bafkreidlw2xpv2ve4girfzlgwrt4imaumorqwcqvxhecjcqa5wonrzmunm",
+    ),
+    (
+        lambda: b"".join((x % 256).to_bytes(1, "big") for x in range(200000)),
+        "bafkreighu7ltw2gscebl67lntprhwqigjf7pzaizejf6x66snm3vkqn544",
+    ),
+    (
+        lambda: b"".join((x % 256).to_bytes(1, "big") for x in range(500000)),
+        "bafkreia7ldhwiil4nra2ymrkz66tyq2zzdwlypgeqvunj6d7sqd6gpmko4",
+    ),
+]

tests/data.py

@@ -0,0 +1,38 @@
+from hashlib import sha256
+from typing import Callable
+
+import pytest
+
+from ipfs_cid import cid_sha256_unwrap_digest, cid_sha256_wrap_digest
+from tests.data import TEST_SETS
+
+
+def test_cid_sha256_wrap_digest_invalid_length():
+    with pytest.raises(AttributeError, match="Invalid digest length"):
+        cid_sha256_wrap_digest(b"\x00" * 40)
+
+
+@pytest.mark.parametrize(
+    "cid",
+    [
+        "bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi",
+        "bafybeibnqmhw7bhvyalczhxzcgdhwk2lm3pgi4n2ctmpqgbnzyz6hpkkji",
+        "bafkreif2why4bwbiadz3gfjoik2ywzmumcnmebmec4evy4spctrxxd3m5ysa",
+    ],
+)
+def test_cid_sha256_unwrap_digest_unsupported_format(cid: str):
+    with pytest.raises(AttributeError, match="Unsupported CID format"):
+        cid_sha256_unwrap_digest(cid)
+
+
+# Comparison targets were generated with Kubo CLI
+@pytest.mark.parametrize(("data_generator", "expected"), TEST_SETS)
+def test_cid_sha256_wrap_unwrap_digest(
+    data_generator: Callable[[], bytes],
+    expected: str,
+):
+    digest = sha256(data_generator()).digest()
+    wrapped = cid_sha256_wrap_digest(digest)
+    assert wrapped == expected
+    unwrapped = cid_sha256_unwrap_digest(wrapped)
+    assert unwrapped == digest

tests/test_cid.py

@@ -0,0 +1,34 @@
+from hashlib import sha256
+from io import BytesIO
+from typing import Callable, Iterable
+
+import pytest
+
+from ipfs_cid import cid_sha256_hash, cid_sha256_hash_chunked, cid_sha256_unwrap_digest
+from tests.data import TEST_SETS
+
+
+@pytest.mark.parametrize(("data_generator", "expected"), TEST_SETS)
+def test_cid_sha256_hash(data_generator: Callable[[], bytes], expected: str) -> None:
+    data = data_generator()
+    expected_digest = sha256(data).digest()
+    cid = cid_sha256_hash(data)
+    assert cid == expected
+    assert cid_sha256_unwrap_digest(cid) == expected_digest
+
+
+@pytest.mark.parametrize(("data_generator", "expected"), TEST_SETS)
+def test_cid_sha256_hash_chunked(
+    data_generator: Callable[[], bytes],
+    expected: str,
+) -> None:
+    def as_chunks(stream: BytesIO, chunk_size: int) -> Iterable[bytes]:
+        while len((chunk := stream.read(chunk_size))) > 0:
+            yield chunk
+
+    data = data_generator()
+    expected_digest = sha256(data).digest()
+    buffer = BytesIO(data)
+    cid = cid_sha256_hash_chunked(as_chunks(buffer, 100))
+    assert cid == expected
+    assert cid_sha256_unwrap_digest(cid) == expected_digest