Locazia: content encryption, indexation, deploy platform

app/__main__.py

@@ -85,8 +85,8 @@ if __name__ == '__main__':
         time.sleep(3)
 
         startup_fn = None
-        if startup_target == 'indexator':
+        if startup_target == 'indexer':
-            from app.core.background.indexator_service import main_fn as target_fn
+            from app.core.background.indexer_service import main_fn as target_fn
         elif startup_target == 'uploader':
             from app.core.background.uploader_service import main_fn as target_fn
         elif startup_target == 'ton_daemon':

app/api/routes/_blockchain.py

@@ -37,6 +37,8 @@ async def s_api_v1_blockchain_send_new_content_message(request):
     await ton_connect.restore_connection()
     assert ton_connect.connected, "No connected wallet"
 
+
+
     # await ton_connect._sdk_client.send_transaction({
     #     'valid_until': int(datetime.now().timestamp()),
     #     'messages': [

app/api/routes/_system.py

@@ -23,7 +23,7 @@ async def s_api_v1_node(request):  # /api/v1/node
         'indexer_height': 0,
         'services': {
             service_key: {
-                'status': service['status'],
+                'status': (service['status'] if (datetime.now() - service['timestamp']).total_seconds() < 30 else 'not working: timeout'),
                 'delay': round((datetime.now() - service['timestamp']).total_seconds(), 3) if service['timestamp'] else -1,
             }
             for service_key, service in request.app.ctx.memory.known_states.items()

app/api/routes/node_storage.py

@@ -1,7 +1,5 @@
 from sanic import response
 from app.core._config import UPLOADS_DIR
-from app.core.storage import db_session
-from app.core.content.content_id import ContentId
 from app.core._utils.resolve_content import resolve_content
 from app.core.models.node_storage import StoredContent
 from app.core.logger import make_log
@@ -10,6 +8,7 @@ from base58 import b58encode, b58decode
 from mimetypes import guess_type
 import os
 import hashlib
+import aiofiles
 
 
 async def s_api_v1_storage_post(request):
@@ -58,8 +57,8 @@ async def s_api_v1_storage_post(request):
         request.ctx.db_session.commit()
 
         file_path = os.path.join(UPLOADS_DIR, file_hash)
-        with open(file_path, "wb") as file:
+        async with aiofiles.open(file_path, "wb") as file:
-            file.write(file_content)
+            await file.write(file_content)
 
         new_content_id = new_content.cid
         new_cid = new_content_id.serialize_v1()

app/core/_blockchain/ton/connect.py

@@ -5,6 +5,7 @@ from hashlib import sha256
 
 from pytonconnect import TonConnect as ExternalLib_TonConnect
 from pytonconnect.storage import DefaultStorage
+from tonsdk.utils import Address
 
 from app.core._config import PROJECT_HOST
 from app.core.logger import make_log
@@ -114,7 +115,7 @@ class TonConnect:
                         network='ton',
                         wallet_key=f"{status['device'].get('app_name', 'UNKNOWN_NAME')}=={status['device'].get('app_version', '1.0')}",
                         connection_id=sha256(self.connection_key.encode()).hexdigest(),
-                        wallet_address=status['account']['address'],
+                        wallet_address=Address(status['account']['address']),
                         keys={
                             'connection_key': self.connection_key,
                         },

app/core/_blockchain/ton/platform.py

@@ -1,4 +1,4 @@
-from app.core._config import TESTNET, MY_PLATFORM_CONTRACT
+from app.core._config import TESTNET, MY_PLATFORM_CONTRACT, PROJECT_HOST
 from app.core._secrets import service_wallet
 from app.core._blockchain.ton.contracts.platform import Platform
 from app.core._blockchain.ton.contracts.cop_nft import COP_NFT
@@ -15,6 +15,6 @@ platform = Platform(
     blank_code=Cell.one_from_boc(Blank.code),
     cop_code=Cell.one_from_boc(COP_NFT.code),
 
-    collection_content_uri='https://music-gateway.letsw.app/api/platform-metadata.json',
+    collection_content_uri=f'{PROJECT_HOST}/api/platform-metadata.json',
     **kwargs
 )

app/core/_crypto/__init__.py

@@ -0,0 +1,2 @@
+from app.core._crypto.signer import Signer
+from app.core._crypto.cipher import Cipher

app/core/_crypto/cipher.py

@@ -0,0 +1,22 @@
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad, unpad
+import hashlib
+
+
+class AESCipher:
+    def __init__(self, seed):
+        assert len(seed) == 32, "Seed must be 32 bytes long"
+        self.private_key = hashlib.sha256(seed).digest()
+
+    def encrypt(self, data: bytes) -> bytes:
+        cipher = AES.new(self.private_key, AES.MODE_CBC)
+        ct_bytes = cipher.encrypt(pad(data, AES.block_size))
+        return cipher.iv + ct_bytes
+
+    def decrypt(self, encrypted_data: bytes) -> bytes:
+        iv = encrypted_data[:AES.block_size]
+        ct = encrypted_data[AES.block_size:]
+        cipher = AES.new(self.private_key, AES.MODE_CBC, iv)
+        pt = unpad(cipher.decrypt(ct), AES.block_size)
+        return pt
+

app/core/_crypto/content.py

@@ -0,0 +1,111 @@
+from app.core.models.node_storage import StoredContent
+from app.core.models.keys import KnownKey
+from app.core._config import PROJECT_HOST, UPLOADS_DIR
+from app.core._crypto.cipher import AESCipher
+from Crypto.Random import get_random_bytes
+from datetime import datetime
+from hashlib import sha256
+from base58 import b58encode, b58decode
+import aiofiles
+import os
+
+
+async def create_new_encryption_key(db_session, user_id: int = None) -> KnownKey:
+    randpart = get_random_bytes(32)
+    new_seed = randpart
+    new_seed_str = b58encode(new_seed).decode()
+    new_seed_hash_bin = sha256(new_seed).digest()
+    new_seed_hash = b58encode(new_seed_hash_bin).decode()
+    public_key = get_random_bytes(32)  # not used yet, algo is symmetric
+    public_key_str = b58encode(public_key).decode()
+    public_key_hash_bin = sha256(public_key).digest()
+    public_key_hash = b58encode(public_key_hash_bin).decode()
+
+    new_key = KnownKey(
+        type="CONTENT_ENCRYPTION_KEY",
+        seed=new_seed_str,
+        seed_hash=new_seed_hash,
+        public_key=public_key_str,
+        public_key_hash=public_key_hash,
+
+        algo="AES256",
+        meta={"I_user_id": user_id} if user_id else None,
+        created=datetime.now()
+    )
+    db_session.add(new_key)
+    db_session.commit()
+    new_key = db_session.query(KnownKey).filter(KnownKey.seed_hash == new_seed_hash).first()
+    assert new_key, "Key not created"
+    return new_key
+
+
+async def create_encrypted_content(
+    db_session, decrypted_content: StoredContent,
+) -> StoredContent:
+    encrypted_content = db_session.query(StoredContent).filter(
+        StoredContent.id == decrypted_content.id
+    ).first()
+    if encrypted_content:
+        return encrypted_content
+
+    encrypted_content = None
+    if decrypted_content.key is None:
+        key = await create_new_encryption_key(db_session, user_id=decrypted_content.user_id)
+        decrypted_content.key_id = key.id
+        db_session.commit()
+        decrypted_content = db_session.query(StoredContent).filter(
+            StoredContent.id == decrypted_content.id
+        ).first()
+        assert decrypted_content.key_id, "Key not assigned"
+
+    decrypted_path = os.path.join(UPLOADS_DIR, decrypted_content.hash)
+    async with aiofiles.open(decrypted_path, mode='rb') as file:
+        decrypted_bin = await file.read()
+
+    key = decrypted_content.key
+    cipher = AESCipher(key.seed_bin)
+
+    encrypted_bin = cipher.encrypt(decrypted_bin)
+    encrypted_hash_bin = sha256(encrypted_bin).digest()
+    encrypted_hash = b58encode(encrypted_hash_bin).decode()
+    encrypted_content = db_session.query(StoredContent).filter(
+        StoredContent.hash == encrypted_hash
+    ).first()
+    if encrypted_content:
+        return encrypted_content
+
+    encrypted_content = None
+
+    encrypted_meta = decrypted_content.meta
+    encrypted_meta["encrypt_algo"] = "AES256"
+
+    encrypted_content = StoredContent(
+        type="local/content_bin",
+        hash=encrypted_hash,
+        onchain_index=None,
+        filename=decrypted_content.filename,
+        meta=encrypted_meta,
+        user_id=decrypted_content.user_id,
+
+        encrypted=True,
+        decrypted_content_id=decrypted_content.id,
+        key_id=decrypted_content.key_id,
+
+        created=datetime.now(),
+    )
+    db_session.add(encrypted_content)
+    db_session.commit()
+
+    encrypted_path = os.path.join(UPLOADS_DIR, encrypted_hash)
+    async with aiofiles.open(encrypted_path, mode='wb') as file:
+        await file.write(encrypted_bin)
+
+    encrypted_content = db_session.query(StoredContent).filter(
+        StoredContent.hash == encrypted_hash
+    ).first()
+    assert encrypted_content, "Content not created"
+    return encrypted_content
+
+
+
+

app/core/background/indexator_service.py

@@ -1,50 +0,0 @@
-from app.core._utils.send_status import send_status
-from app.core._config import MY_PLATFORM_CONTRACT
-from app.core._blockchain.ton.toncenter import toncenter
-from app.core.models.node_storage import StoredContent
-from app.core.storage import db_session
-from app.core.logger import make_log
-import asyncio
-
-
-async def indexator_loop(platform_found: bool, seqno: int) -> [bool, int]:
-    if not platform_found:
-        platform_state = await toncenter.get_account(MY_PLATFORM_CONTRACT)
-        if not platform_state.get('code'):
-            make_log("TON", "Platform contract is not deployed, skipping loop", level="info")
-            await send_status("indexator", "not working: platform is not deployed")
-            return False, seqno
-        else:
-            platform_found = True
-
-    make_log("Indexator", "Service running", level="debug")
-    with db_session() as session:
-        last_known_index = session.query(StoredContent).order_by(StoredContent.onchain_index.desc()).first()
-        last_known_index = last_known_index.onchain_index if last_known_index >= 0 else 0
-        make_log("Indexator", f"Last known index: {last_known_index}", level="debug")
-
-    await send_status("indexator", f"working (seqno={seqno})")
-    return platform_found, seqno
-
-
-async def main_fn():
-    make_log("Indexator", "Service started", level="info")
-    platform_found = False
-    seqno = 0
-    while True:
-        try:
-            platform_found, seqno = await indexator_loop(platform_found, seqno)
-        except BaseException as e:
-            make_log("Indexator", f"Error: {e}", level="error")
-
-        await asyncio.sleep(5)
-        seqno += 1
-
-# if __name__ == '__main__':
-#     loop = asyncio.get_event_loop()
-#     loop.run_until_complete(main())
-#     loop.close()
-
-
-
-

app/core/background/indexer_service.py

@@ -0,0 +1,162 @@
+from app.core._utils.send_status import send_status
+from app.core._config import MY_PLATFORM_CONTRACT
+from app.core._blockchain.ton.toncenter import toncenter
+from app.core._blockchain.ton.platform import platform
+from app.core.models.node_storage import StoredContent
+from app.core.models.wallet_connection import WalletConnection
+from app.core.storage import db_session
+from app.core.logger import make_log
+from tonsdk.boc import begin_cell, Cell
+from tonsdk.utils import Address
+from base64 import b64encode, b64decode
+from base58 import b58encode, b58decode
+from datetime import datetime
+import asyncio
+
+
+async def indexer_loop(platform_found: bool, seqno: int) -> [bool, int]:
+    if not platform_found:
+        platform_state = await toncenter.get_account(platform.address.to_string(1, 1, 1))
+        if not platform_state.get('code'):
+            make_log("TON", "Platform contract is not deployed, skipping loop", level="info")
+            await send_status("Indexer", "not working: platform is not deployed")
+            return False, seqno
+        else:
+            platform_found = True
+
+    make_log("Indexer", "Service running", level="debug")
+    with db_session() as session:
+        last_known_index = session.query(StoredContent).filter(
+            StoredContent.type == "onchain/content"
+        ).order_by(StoredContent.onchain_index.desc()).first()
+        last_known_index = last_known_index.onchain_index if last_known_index >= 0 else 0
+        make_log("Indexer", f"Last known index: {last_known_index}", level="debug")
+        next_item_index = last_known_index + 1
+
+        resolve_item_result = await toncenter.run_get_method(platform.address.to_string(1, 1, 1), 'get_nft_address_by_index', [['num', hex(next_item_index)[2:]]])
+        if resolve_item_result.get('exit_code', -1) != 0:
+            make_log("Indexer", f"Resolve item error: {resolve_item_result}", level="error")
+            return platform_found, seqno
+
+        item_address_cell_b64 = resolve_item_result['stack'][0][1]
+        item_address_slice = Cell.one_from_boc(b64decode(item_address_cell_b64)).begin_parse()
+        item_address = item_address_slice.read_msg_addr()
+
+        item_get_data_result = await toncenter.run_get_method(item_address, 'indexator_data')
+        if item_get_data_result.get('exit_code', -1) != 0:
+            make_log("Indexer", f"Get item data error (maybe not deployed): {item_get_data_result}", level="debug")
+            return platform_found, seqno
+
+        assert item_get_data_result['stack'][0][0] == 'num', "Item type is not a number"
+        assert int(item_get_data_result['stack'][0][1], 16) == 1, "Item is not COP NFT"
+        item_returned_address = Cell.one_from_boc(b64decode(item_get_data_result['stack'][1][1])).begin_parse().read_msg_addr()
+        assert (
+            item_returned_address.to_string(1, 1, 1) == item_address.to_string(1, 1, 1)
+        ), "Item address mismatch"
+
+        assert item_get_data_result['stack'][2][0] == 'num', "Item index is not a number"
+        item_index = int(item_get_data_result['stack'][2][1], 16)
+        assert item_index == next_item_index, "Item index mismatch"
+
+        item_platform_address = Cell.one_from_boc(b64decode(item_get_data_result['stack'][3][1])).begin_parse().read_msg_addr()
+        assert item_platform_address.to_string(1, 1, 1) == Address(platform.address.to_string(1, 1, 1)).to_string(1, 1, 1), "Item platform address mismatch"
+
+        assert item_get_data_result['stack'][4][0] == 'num', "Item license type is not a number"
+        item_license_type = int(item_get_data_result['stack'][4][1], 16)
+        assert item_license_type == 0, "Item license type is not 0"
+
+        item_owner_address = Cell.one_from_boc(b64decode(item_get_data_result['stack'][5][1])).begin_parse().read_msg_addr()
+        item_values = Cell.one_from_boc(b64decode(item_get_data_result['stack'][6][1]))
+        item_derivates = Cell.one_from_boc(b64decode(item_get_data_result['stack'][7][1]))
+        item_platform_variables = Cell.one_from_boc(b64decode(item_get_data_result['stack'][8][1]))
+
+        item_values_slice = item_values.begin_parse()
+        item_content_hash_int = item_values_slice.read_uint(256)
+        item_content_hash = item_content_hash_int.to_bytes(32, 'big')
+        item_content_hash_str = b58encode(item_content_hash).decode()
+        item_metadata = item_values_slice.refs[0]
+        item_content = item_values_slice.refs[1]
+        item_metadata_str = item_metadata.bits.array.decode()
+        item_content_cid_str = item_content.refs[0].bits.array.decode()
+        item_content_cover_cid_str = item_content.refs[1].bits.array.decode()
+        item_content_metadata_cid_str = item_content.refs[2].bits.array.decode()
+
+        item_metadata_packed = {
+            'license_type': item_license_type,
+            'item_address': item_address.to_string(1, 1, 1),
+            'content_cid': item_content_cid_str,
+            'cover_cid': item_content_cover_cid_str,
+            'metadata_cid': item_content_metadata_cid_str,
+            'derivates': b58encode(item_derivates.to_boc(False)).decode(),
+            'platform_variables': b58encode(item_platform_variables.to_boc(False)).decode()
+        }
+
+        user_wallet_connection = None
+        if item_owner_address:
+            user_wallet_connection = session.query(WalletConnection).filter(
+                WalletConnection.wallet_address == item_owner_address.to_string(1, 1, 1)
+            ).first()
+
+        encrypted_stored_content = session.query(StoredContent).filter(
+            StoredContent.hash == item_content_hash_str,
+            StoredContent.onchain_index == None
+        ).first()
+        if encrypted_stored_content:
+            encrypted_stored_content_meta = encrypted_stored_content.meta
+            make_log("Indexer", f"Item already indexed: {item_content_hash_str}", level="debug")
+            encrypted_stored_content.type = "onchain/content"
+            encrypted_stored_content.onchain_index = item_index
+            encrypted_stored_content.owner_address = item_owner_address.to_string(1, 1, 1)
+            if user_wallet_connection:
+                encrypted_stored_content.user_id = user_wallet_connection.user_id
+
+            if encrypted_stored_content_meta != item_metadata_packed:
+                encrypted_stored_content.meta = item_metadata_packed
+
+            encrypted_stored_content.updated = datetime.now()
+            session.commit()
+            return platform_found, seqno
+
+        onchain_stored_content = StoredContent(
+            type="onchain/content_unknown",
+            hash=item_content_hash_str,
+            onchain_index=item_index,
+            owner_address=item_owner_address.to_string(1, 1, 1) if item_owner_address else None,
+            meta=item_metadata_packed,
+            user_id=user_wallet_connection.user_id if user_wallet_connection else None,
+            created=datetime.now(),
+            encrypted=True,
+            decrypted_content_id=None,
+            key_id=None,
+            updated=datetime.now()
+        )
+        session.add(onchain_stored_content)
+        session.commit()
+        make_log("Indexer", f"Item indexed: {item_content_hash_str}", level="info")
+        last_known_index += 1
+
+    await send_status("indexer", f"working (seqno={seqno}, height={last_known_index})")
+    return platform_found, seqno
+
+
+async def main_fn():
+    make_log("Indexer", "Service started", level="info")
+    platform_found = False
+    seqno = 0
+    while True:
+        try:
+            platform_found, seqno = await indexer_loop(platform_found, seqno)
+        except BaseException as e:
+            make_log("Indexer", f"Error: {e}", level="error")
+
+        await asyncio.sleep(5)
+        seqno += 1
+
+# if __name__ == '__main__':
+#     loop = asyncio.get_event_loop()
+#     loop.run_until_complete(main())
+#     loop.close()
+
+
+
+

app/core/background/ton_service.py

@@ -1,6 +1,7 @@
 from tonsdk.boc import begin_cell
 from app.core.logger import make_log
 from app.core._config import MY_FUND_ADDRESS, MY_PLATFORM_CONTRACT
+from app.core._blockchain.ton.platform import platform
 from app.core.storage import db_session
 from app.core._secrets import service_wallet
 from app.core._blockchain.ton.toncenter import toncenter
@@ -53,6 +54,26 @@ async def main_fn():
         )
         make_log("TON", "Withdraw command sent", level="info")
         await asyncio.sleep(10)
+        return await main_fn()
+
+    platform_state = await toncenter.get_account(platform.address.to_string(1, 1, 1))
+    if not platform_state.get('code'):
+        make_log("TON", "Platform contract is not deployed, send deploy transaction..", level="info")
+        await toncenter.send_boc(
+            service_wallet.create_transfer_message(
+                [{
+                    'address': platform.address.to_string(1, 1, 1),
+                    'amount': int(0.08 * 10 ** 9),
+                    'send_mode': 1,
+                    'payload': begin_cell().store_uint(0, 32).store_uint(0, 64).end_cell(),
+                    'state_init': platform.create_state_init()['state_init']
+                }], sw_seqno_value
+            )['message'].to_boc(False)
+        )
+
+        await send_status("ton_daemon", "working: deploying platform")
+        await asyncio.sleep(15)
+        return await main_fn()
 
     while True:
         try:

app/core/content/utils.py

@@ -0,0 +1,6 @@
+from app.core.models.node_storage import StoredContent
+
+
+async def create_metadata_for_item(**kwargs):
+
+

app/core/models/keys.py

@@ -1,5 +1,6 @@
 from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, JSON, Boolean
 from sqlalchemy.orm import relationship
+from base58 import b58decode
 
 from .base import AlchemyBase
 
@@ -9,14 +10,25 @@ class KnownKey(AlchemyBase):
 
     id = Column(Integer, autoincrement=True, primary_key=True)
     type = Column(String(32), nullable=False, default="NOT_SPECIFIED")
-    seed = Column(String(6144), nullable=True, default=None)
+    seed = Column(String(6144), nullable=True, default=None, unique=True)
-    seed_hash = Column(String(64), nullable=True, default=None)  # base58
+    seed_hash = Column(String(64), nullable=True, default=None, unique=True)  # base58
     public_key = Column(String(6144), nullable=False, unique=True)
     public_key_hash = Column(String(64), nullable=False, unique=True)  # base58
 
     algo = Column(String(32), nullable=True, default=None)
     meta = Column(JSON, nullable=False, default={})
+    # {
+    #   "I_user_id": TRUSTED_USER_ID,
+    # }
 
     created = Column(DateTime, nullable=False, default=0)
 
     # stored_content = relationship('StoredContent', back_populates='key')
+
+    @property
+    def seed_bin(self) -> bytes:
+        return b58decode(self.seed)
+
+    @property
+    def public_key_bin(self) -> bytes:
+        return b58decode(self.public_key)

app/core/models/memory.py

@@ -24,7 +24,7 @@ class Memory:
             #     "status": "no status",
             #     "timestamp": None
             # },
-            "indexator": {
+            "indexer": {
                 "status": "no status",
                 "timestamp": None
             },

app/core/models/node_storage.py

@@ -20,19 +20,24 @@ class StoredContent(AlchemyBase):
     meta = Column(JSON, nullable=False, default={})
 
     user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
+    owner_address = Column(String(1024), nullable=True)
 
     btfs_cid = Column(String(1024), nullable=True)
     ipfs_cid = Column(String(1024), nullable=True)
     telegram_cid = Column(String(1024), nullable=True)
 
     created = Column(DateTime, nullable=False, default=0)
+    updated = Column(DateTime, nullable=False, default=0)
     disabled = Column(DateTime, nullable=False, default=0)
     disabled_by = Column(Integer, ForeignKey('users.id'), nullable=True, default=None)
 
+    encrypted = Column(Boolean, nullable=False, default=False)
+    decrypted_content_id = Column(Integer, ForeignKey('node_storage.id'), nullable=True, default=None)
     key_id = Column(Integer, ForeignKey('known_keys.id'), nullable=True, default=None)
 
     user = relationship('User', uselist=False, foreign_keys=[user_id])
     key = relationship('KnownKey', uselist=False, foreign_keys=[key_id])
+    decrypted_content = relationship('StoredContent', uselist=False, foreign_keys=[decrypted_content])
 
     @property
     def cid(self) -> ContentId:
@@ -56,5 +61,5 @@ class StoredContent(AlchemyBase):
             "hash": self.hash,
             "cid": self.cid.serialize_v1(),
             "status": self.status,
-            "meta": self.meta,
+            "meta": self.meta
         }

docker-compose.yml

@@ -33,11 +33,11 @@ services:
       maria_db:
         condition: service_healthy
 
-  indexator:
+  indexer:
     build:
       context: .
       dockerfile: Dockerfile
-    command: python -m app indexator
+    command: python -m app indexer
     env_file:
       - .env
     links:

requirements.txt

@@ -11,3 +11,4 @@ httpx==0.25.0
 docker==7.0.0
 pycryptodome==3.20.0
 pynacl==1.5.0
+aiofiles==23.2.1