dev@locazia: add blockchain routes, edit resolve content system to cid class, add web2 docs

app/core/_config.py

@@ -28,3 +28,9 @@ LOG_FILEPATH = f"{LOG_DIR}/{_now_str}.log"
 WEB_APP_URLS = {
     'uploadContent': f"https://web2-client.vercel.app/uploadContent"
 }
+
+ALLOWED_CONTENT_TYPES = [
+    'image/jpeg', 'image/png', 'image/gif', 'image/webp',
+    'video/mp4', 'video/webm', 'video/ogg',
+    'audio/mpeg', 'audio/ogg', 'audio/wav',
+]

app/core/_utils/resolve_content.py

@@ -0,0 +1,9 @@
+from app.core.content.content_id import ContentId
+
+
+def resolve_content(content_id):  # -> [content, error]
+    try:
+        return ContentId.deserialize(content_id), None
+    except BaseException as e:
+        return None, f"{e}"
+

app/core/_utils/string_binary.py

@@ -0,0 +1,11 @@
+
+def string_to_bytes_fixed_size(src: str, size: int, encoding='utf-8') -> bytes:
+    assert type(src) is str, "src must be a string"
+    src_bin = src.encode(encoding)
+    assert len(src_bin) <= size, "src is too long"
+    return src_bin + b'\x00' * (size - len(src_bin))
+
+
+def bytes_to_string(src: bytes, encoding='utf-8') -> str:
+    assert type(src) is bytes, "src must be bytes"
+    return src.decode(encoding).rstrip('\x00')

app/core/content/__init__.py

@@ -0,0 +1 @@
+from app.core.content.content_id import ContentId

app/core/content/content_id.py

@@ -0,0 +1,68 @@
+from app.core._utils.string_binary import string_to_bytes_fixed_size, bytes_to_string
+from base58 import b58encode, b58decode
+from app.core._config import ALLOWED_CONTENT_TYPES
+
+# cid_v1#_ cid_version:int8 accept_type:uint120 content_sha256:uint256 onchain_index:uint128 = CIDv1;
+
+
+class ContentId:
+    def __init__(
+        self,
+        content_hash: bytes = None,  # only SHA256
+        onchain_index: int = None,
+        accept_type: str = 'image/jpeg'
+    ):
+        self.content_hash = content_hash
+        self.onchain_index = onchain_index or -1
+
+        self.accept_type = accept_type
+
+    def serialize_v1(self) -> str:
+        at_bin = string_to_bytes_fixed_size(self.accept_type, 15)
+        assert len(self.content_hash) == 32, "Invalid hash length"
+        if self.onchain_index < 0:
+            oi_bin = b''
+        else:
+            oi_bin = self.onchain_index.to_bytes(16, 'big', signed=False)
+            assert len(oi_bin) == 16, "Invalid onchain_index"
+
+        return b58encode(
+            (1).to_bytes(1, 'big')  # cid version
+            + at_bin
+            + self.content_hash
+            + oi_bin
+        ).decode()
+
+    @classmethod
+    def from_v1(cls, cid: str):
+        cid_bin = b58decode(cid)
+        (
+            cid_version,
+            accept_type,
+            content_sha256,
+            onchain_index
+        ) = (
+            int.from_bytes(cid_bin[0:1], 'big'),
+            bytes_to_string(cid_bin[1:16]),
+            cid_bin[16:48],
+            int.from_bytes(cid_bin[48:], 'big') if len(cid_bin) > 48 else -1
+        )
+        assert cid_version == 1, "Invalid version"
+        content_type = accept_type.split('/')
+        assert '/'.join(content_type[0:2]) in ALLOWED_CONTENT_TYPES, "Invalid accept type"
+        assert len(content_sha256) == 32, "Invalid hash length"
+        return cls(
+            content_hash=content_sha256,
+            onchain_index=onchain_index,
+            accept_type=accept_type
+        )
+
+    @classmethod
+    def deserialize(cls, cid: str):
+        cid_version = int.from_bytes(b58decode(cid)[0:1], 'big')
+        if cid_version == 1:
+            return cls.from_v1(cid)
+        else:
+            raise ValueError("Invalid cid version")
+
+

app/core/models/node_storage.py

@@ -4,12 +4,7 @@ from sqlalchemy.orm import relationship
 from .base import AlchemyBase
 from hashlib import sha256
 from base58 import b58encode, b58decode
-
-
-# DMY CID v1 specs
-# 1. int8 cid version
-# 2. int256 sha256 of content
-# 3. int128 onchain content index
+from app.core.content.content_id import ContentId
 
 
 class StoredContent(AlchemyBase):
@@ -37,13 +32,10 @@ class StoredContent(AlchemyBase):
     user = relationship('User', uselist=False, foreign_keys=[user_id])
     key = relationship('KnownKey', uselist=False, foreign_keys=[key_id])
 
-    def make_dmy_cid_v1(self) -> str:
-        vhash = sha256(
-            (1).to_bytes(1, 'big')  # cid version
-            + b58decode(hash)
-            + (self.onchain_index or 0).to_bytes(16, 'big')
-        ).digest()
-        return b58encode(vhash).decode()
-
-    def make_deeplink(self):
-        return f"dmy://storage?cid={self.make_dmy_cid_v1()}"
+    @property
+    def cid(self) -> ContentId:
+        return ContentId(
+            content_hash=b58decode(self.hash),
+            onchain_index=self.onchain_index,
+            accept_type=self.meta.get('content_type', 'image/jpeg')
+        )