Adds proofs, pending downloaded bytes (#424)

README.md

@@ -99,6 +99,7 @@ Refer to the [API.md](API.md) file for endpoints and payloads.
 * `LOG_CFG` and `LOG_LEVEL` define the location of the log file and logging leve, respectively
 * `IPFS_GATEWAY` defines ipfs gateway for resolving urls
 * `AUTHORIZED_DECRYPTERS` list of authorized addresses that are allowed to decrypt chain data. Use it to restrict access only to certain callers (e.g. custom Aquarius instance). Empty by default, meaning all decrypters are authorized.
+* `USE_CHAIN_PROOF` or `USE_HTTP_PROOF` set a mechanism for saving proof-of-download information. For any present true-ish value of `USE_CHAIN_PROOF`, the proof is sent on-chain. When defining `USE_HTTP_PROOF` the env var must configure a HTTP endpoint that accepts a POST request.
 
 
 #### Before you commit

ocean_provider/routes/consume.py

@@ -2,14 +2,18 @@
 # Copyright 2021 Ocean Protocol Foundation
 # SPDX-License-Identifier: Apache-2.0
 #
+import json
 import logging
 
 from flask import jsonify, request
 from flask_sieve import validate
 from ocean_provider.myapp import app
 from ocean_provider.requests_session import get_requests_session
 from ocean_provider.user_nonce import get_nonce, update_nonce
-from ocean_provider.utils.asset import get_asset_from_metadatastore
+from ocean_provider.utils.asset import (
+    get_asset_from_metadatastore,
+    check_asset_consumable,
+)
 from ocean_provider.utils.basics import (
     LocalFileAdapter,
     get_provider_wallet,
@@ -20,12 +24,12 @@
 from ocean_provider.utils.compute_environments import check_environment_exists
 from ocean_provider.utils.datatoken import validate_order
 from ocean_provider.utils.error_responses import error_response
+from ocean_provider.utils.proof import send_proof
 from ocean_provider.utils.provider_fees import get_provider_fees, get_c2d_environments
 from ocean_provider.utils.services import ServiceType
 from ocean_provider.utils.url import append_userdata, check_url_details
 from ocean_provider.utils.util import (
     build_download_response,
-    check_asset_consumable,
     get_download_url,
     get_request_data,
     get_service_files_list,
@@ -332,4 +336,26 @@ def download():
     )
     logger.info(f"download response = {response}")
 
+    provider_proof_data = json.dumps(
+        {
+            "documentId": did,
+            "serviceId": service_id,
+            "fileIndex": file_index,
+            "downloadedBytes": 0,  # TODO
+        },
+        separators=(",", ":"),
+    )
+
+    consumer_data = f'{did}{data.get("nonce")}'
+
+    send_proof(
+        web3=get_web3(),
+        order_tx_id=_tx.hash,
+        provider_data=provider_proof_data,
+        consumer_data=consumer_data,
+        consumer_signature=data.get("signature"),
+        consumer_address=consumer_address,
+        datatoken_address=service.datatoken_address,
+    )
+
     return response

ocean_provider/utils/accounts.py

@@ -77,9 +77,11 @@ def sign_message(message, wallet):
     :return: signature
     """
     keys_pk = keys.PrivateKey(wallet.key)
+    hexable = Web3.toBytes(text=message) if isinstance(message, str) else message
+
     message_hash = Web3.solidityKeccak(
         ["bytes"],
-        [Web3.toBytes(text=message)],
+        [Web3.toHex(hexable)],
     )
     prefix = "\x19Ethereum Signed Message:\n32"
     signable_hash = Web3.solidityKeccak(

ocean_provider/utils/asset.py

@@ -6,6 +6,9 @@
 import requests
 from typing import Optional
 
+from jsonsempai import magic  # noqa: F401
+from artifacts import ERC721Template
+from ocean_provider.utils.basics import get_web3
 from ocean_provider.utils.consumable import ConsumableCodes
 from ocean_provider.utils.credentials import AddressCredential
 from ocean_provider.utils.services import Service
@@ -89,3 +92,25 @@ def get_asset_from_metadatastore(metadata_url, document_id):
     response = requests.get(url)
 
     return Asset(response.json()) if response.status_code == 200 else None
+
+
+def check_asset_consumable(asset, consumer_address, logger, custom_url=None):
+    if not asset.nft or "address" not in asset.nft:
+        return False, "Asset malformed"
+
+    dt_contract = get_web3().eth.contract(
+        abi=ERC721Template.abi, address=asset.nft["address"]
+    )
+
+    if dt_contract.caller.getMetaData()[2] != 0:
+        return False, "Asset is not consumable."
+
+    code = asset.is_consumable({"type": "address", "value": consumer_address})
+
+    if code == ConsumableCodes.OK:  # is consumable
+        return True, ""
+
+    message = f"Error: Access to asset {asset.did} was denied with code: {code}."
+    logger.error(message, exc_info=1)
+
+    return False, message

ocean_provider/utils/proof.py

@@ -0,0 +1,63 @@
+import os
+import requests
+from ocean_provider.utils.basics import get_provider_wallet
+from ocean_provider.utils.accounts import sign_message
+from ocean_provider.utils.datatoken import get_datatoken_contract
+from ocean_provider.utils.util import sign_and_send
+from web3.main import Web3
+
+
+def send_proof(
+    web3,
+    order_tx_id,
+    provider_data,
+    consumer_data,
+    consumer_signature,
+    consumer_address,
+    datatoken_address,
+):
+    if not os.getenv("USE_CHAIN_PROOF") and not os.getenv("USE_HTTP_PROOF"):
+        return
+
+    provider_wallet = get_provider_wallet()
+    provider_signature = sign_message(provider_data, provider_wallet)
+
+    if os.getenv("USE_HTTP_PROOF"):
+        payload = {
+            "orderTxId": order_tx_id.hex(),
+            "providerData": provider_data,
+            "providerSignature": provider_signature,
+            "consumerData": consumer_data,
+            "consumerSignature": consumer_signature,
+            "consumerAddress": consumer_address,
+        }
+
+        try:
+            requests.post(os.getenv("USE_HTTP_PROOF"), payload)
+
+            return True
+        except Exception:
+            pass
+
+        return
+
+    datatoken_contract = get_datatoken_contract(web3, datatoken_address)
+    provider_message = order_tx_id + Web3.toBytes(text=provider_data)
+    provider_signature = sign_message(provider_message, provider_wallet)
+
+    consumer_message = Web3.toBytes(text=consumer_data)
+
+    tx = datatoken_contract.functions.orderExecuted(
+        order_tx_id,
+        Web3.toBytes(text=provider_data),
+        provider_signature,
+        consumer_message,
+        consumer_signature,
+        consumer_address,
+    ).buildTransaction(
+        {"from": provider_wallet.address, "gasPrice": int(web3.eth.gas_price * 1.1)}
+    )
+
+    _, transaction_id = sign_and_send(web3, tx, provider_wallet)
+
+    return transaction_id

ocean_provider/utils/util.py

@@ -9,19 +9,20 @@
 import os
 from cgi import parse_header
 from urllib.parse import urljoin
-
+from typing import Tuple
 import werkzeug
-from jsonsempai import magic  # noqa: F401
-from artifacts import ERC721Template
+
 from eth_account.signers.local import LocalAccount
 from eth_keys import KeyAPI
 from eth_keys.backends import NativeECCBackend
+from eth_typing.encoding import HexStr
 from flask import Response
-from ocean_provider.utils.basics import get_web3
-from ocean_provider.utils.consumable import ConsumableCodes
 from ocean_provider.utils.encryption import do_decrypt
 from ocean_provider.utils.services import Service
 from ocean_provider.utils.url import is_safe_url
+from web3 import Web3
+from web3.types import TxParams, TxReceipt
+
 
 logger = logging.getLogger(__name__)
 keys = KeyAPI(NativeECCBackend)
@@ -151,23 +152,36 @@ def get_download_url(url_object):
     return urljoin(os.getenv("IPFS_GATEWAY"), urljoin("ipfs/", url_object["hash"]))
 
 
-def check_asset_consumable(asset, consumer_address, logger, custom_url=None):
-    if not asset.nft or "address" not in asset.nft:
-        return False, "Asset malformed"
+def sign_tx(web3, tx, private_key):
+    """
+    :param web3: Web3 object instance
+    :param tx: transaction
+    :param private_key: Private key of the account
+    :return: rawTransaction (str)
+    """
+    account = web3.eth.account.from_key(private_key)
+    nonce = web3.eth.get_transaction_count(account.address)
+    tx["nonce"] = nonce
+    signed_tx = web3.eth.account.sign_transaction(tx, private_key)
+
+    return signed_tx.rawTransaction
 
-    dt_contract = get_web3().eth.contract(
-        abi=ERC721Template.abi, address=asset.nft["address"]
-    )
 
-    if dt_contract.caller.getMetaData()[2] != 0:
-        return False, "Asset is not consumable."
+def sign_and_send(
+    web3: Web3, transaction: TxParams, from_account: LocalAccount
+) -> Tuple[HexStr, TxReceipt]:
+    """Returns the transaction id and transaction receipt."""
+    transaction_signed = sign_tx(web3, transaction, from_account.key)
+    transaction_hash = web3.eth.send_raw_transaction(transaction_signed)
+    transaction_id = Web3.toHex(transaction_hash)
 
-    code = asset.is_consumable({"type": "address", "value": consumer_address})
+    return transaction_hash, transaction_id
 
-    if code == ConsumableCodes.OK:  # is consumable
-        return True, ""
 
-    message = f"Error: Access to asset {asset.did} was denied with code: {code}."
-    logger.error(message, exc_info=1)
+def sign_send_and_wait_for_receipt(
+    web3: Web3, transaction: TxParams, from_account: LocalAccount
+) -> Tuple[HexStr, TxReceipt]:
+    """Returns the transaction id and transaction receipt."""
+    transaction_hash, transaction_id = sign_and_send(web3, transaction, from_account)
 
-    return False, message
+    return (transaction_id, web3.eth.wait_for_transaction_receipt(transaction_hash))

ocean_provider/validation/algo.py

@@ -7,7 +7,10 @@
 
 from ocean_provider.constants import BaseURLs
 from ocean_provider.serializers import StageAlgoSerializer
-from ocean_provider.utils.asset import get_asset_from_metadatastore
+from ocean_provider.utils.asset import (
+    get_asset_from_metadatastore,
+    check_asset_consumable,
+)
 from ocean_provider.utils.basics import get_config, get_metadata_url
 from ocean_provider.utils.datatoken import (
     record_consume_request,
@@ -16,7 +19,6 @@
 )
 from ocean_provider.utils.url import append_userdata
 from ocean_provider.utils.util import (
-    check_asset_consumable,
     get_service_files_list,
     msg_hash,
 )

setup.py

@@ -68,6 +68,7 @@
     "flake8",
     "isort",
     "black==22.1.0",
+    "click==8.0.4",
     "pre-commit",
     "licenseheaders",
     "pytest-env",

tests/test_helpers.py

@@ -30,6 +30,7 @@
 from ocean_provider.utils.did import compute_did_from_data_nft_address_and_chain_id
 from ocean_provider.utils.encryption import do_encrypt
 from ocean_provider.utils.services import Service, ServiceType
+from ocean_provider.utils.util import sign_send_and_wait_for_receipt, sign_tx
 from tests.helpers.ddo_dict_builders import (
     build_credentials_dict,
     build_ddo_dict,
@@ -40,7 +41,7 @@
 )
 from web3.logs import DISCARD
 from web3.main import Web3
-from web3.types import TxParams, TxReceipt
+from web3.types import TxReceipt
 
 logger = logging.getLogger(__name__)
 BLACK_HOLE_ADDRESS = "0x0000000000000000000000000000000000000000"
@@ -50,20 +51,6 @@ def get_gas_price(web3) -> int:
     return int(web3.eth.gas_price * 1.1)
 
 
-def sign_tx(web3, tx, private_key):
-    """
-    :param web3: Web3 object instance
-    :param tx: transaction
-    :param private_key: Private key of the account
-    :return: rawTransaction (str)
-    """
-    account = web3.eth.account.from_key(private_key)
-    nonce = web3.eth.get_transaction_count(account.address)
-    tx["nonce"] = nonce
-    signed_tx = web3.eth.account.sign_transaction(tx, private_key)
-    return signed_tx.rawTransaction
-
-
 def deploy_contract(w3, _json, private_key, *args):
     """
     :param w3: Web3 object instance
@@ -94,16 +81,6 @@ def get_ocean_token_address(web3: Web3) -> HexAddress:
     return get_contract_address(get_config().address_file, "Ocean", 8996)
 
 
-def sign_send_and_wait_for_receipt(
-    web3: Web3, transaction: TxParams, from_account: LocalAccount
-) -> Tuple[HexStr, TxReceipt]:
-    """Returns the transaction id and transaction receipt."""
-    transaction_signed = sign_tx(web3, transaction, from_account.key)
-    transaction_hash = web3.eth.send_raw_transaction(transaction_signed)
-    transaction_id = Web3.toHex(transaction_hash)
-    return (transaction_id, web3.eth.wait_for_transaction_receipt(transaction_hash))
-
-
 def deploy_data_nft(
     web3: Web3,
     name: str,