Add txId in payload for RBAC check for asset indexing

aquarius/events/processors.py

@@ -59,8 +59,8 @@ def __init__(
         self._chain_id = chain_id
         self.metadata_proofs = None
 
-    def check_permission(self, publisher_address):
-        if not os.getenv("RBAC_SERVER_URL") or not publisher_address:
+    def check_permission(self, publisher_address, tx_id, asset):
+        if not os.getenv("RBAC_SERVER_URL") or not publisher_address or not tx_id:
             return True
 
         event_type = (
@@ -69,7 +69,7 @@ def check_permission(self, publisher_address):
             else "update"
         )
 
-        return RBAC.check_permission_rbac(event_type, publisher_address)
+        return RBAC.check_permission_rbac(event_type, publisher_address, tx_id, asset)
 
     def add_aqua_data(self, record):
         """Adds keys that are specific to Aquarius, on top of the DDO structure:
@@ -300,7 +300,7 @@ def process(self):
         except Exception:
             pass
 
-        permission = self.check_permission(sender_address)
+        permission = self.check_permission(sender_address, txid, asset)
         if not permission:
             error = "RBAC permission denied."
             logger.info(error)
@@ -446,7 +446,7 @@ def process(self):
         self.did = asset["id"]
         did, sender_address = self.did, self.sender_address
 
-        permission = self.check_permission(sender_address)
+        permission = self.check_permission(sender_address, txid, asset)
         if not permission:
             error = "RBAC permission denied."
             logger.info(error)

aquarius/rbac.py

@@ -71,11 +71,19 @@ def validate_ddo_rbac(data):
         ).json()
 
     @staticmethod
-    def check_permission_rbac(event_type, address):
+    def check_permission_rbac(event_type, address, tx_id, asset):
+        try:
+            chain_id = asset["chainId"]
+            nft_address = asset["nftAddress"]
+        except KeyError:
+            return False
+
         payload = {
             "eventType": event_type,
             "component": "metadatacache",
             "credentials": {"type": "address", "value": address},
+            "txid": tx_id,
+            "asset": {"chainId": chain_id, "nftAddress": nft_address},
         }
 
         try:

tests/helpers.py

@@ -115,7 +115,7 @@ def send_create_update_tx(name, ddo, flags, account):
     if flags[0] & 2:
         headers = {"Content-type": "application/octet-stream"}
         response = requests.post(
-            provider_url + "/api/services/encrypt?chainId={web3.chain_id}",
+            provider_url + f"/api/services/encrypt?chainId={web3.eth.chain_id}",
             data=compressed_document,
             headers=headers,
             timeout=5,

tests/test_events.py

@@ -62,7 +62,9 @@ def run_test(client, base_ddo_url, events_instance, flags):
     for service in published_ddo["services"]:
         assert service["datatokenAddress"] == erc20_address
         assert service["name"] in ["dataAssetAccess", "dataAssetComputingService"]
-    ddo_state = get_did_state(events_instance._es_instance, None, None, None, did)
+    ddo_state = get_did_state(
+        events_instance._es_instance, web3.eth.chain_id, None, None, did
+    )
     assert len(ddo_state["hits"]["hits"]) == 1
     assert ddo_state["hits"]["hits"][0]["_id"] == did
     assert ddo_state["hits"]["hits"][0]["_source"]["valid"] is True
@@ -122,7 +124,7 @@ def test_publish_unallowed_address(client, base_ddo_url, events_object):
 
 
 def test_publish_and_update_ddo_rbac(client, base_ddo_url, events_object, monkeypatch):
-    monkeypatch.setenv("RBAC_SERVER_URL", "http://localhost:3000")
+    monkeypatch.setenv("RBAC_SERVER_URL", "http://172.15.0.8:3000")
     run_test(client, base_ddo_url, events_object, 2)
 
 

tests/test_processors.py

@@ -81,7 +81,14 @@ def test_check_permission(monkeypatch):
     )
     with patch("requests.post") as mock:
         mock.side_effect = Exception("Boom!")
-        assert processor.check_permission("some_address") is False
+        asset = {"chainId": 8996, "nftAddress": "some_nft_address"}
+        assert processor.check_permission("some_address", "some tx id", asset) is False
+
+    # Test permission with empty asset
+    with patch("requests.post") as mock:
+        mock.side_effect = Exception("Boom!")
+        asset = {}
+        assert processor.check_permission("some_address", "some tx id", asset) is False
 
     # will affect the process() function too
     with pytest.raises(Exception):