From 0ae5bed592bf4f1fb645be6391a4600f4e3d1c15 Mon Sep 17 00:00:00 2001
From: Daniel Hornung <d.hornung@indiscale.com>
Date: Tue, 11 May 2021 18:13:41 +0200
Subject: [PATCH] WIP ENH: Bloxberg utility
---
src/caosadvancedtools/bloxberg/__init__.py | 0
src/caosadvancedtools/bloxberg/bloxberg.py | 157 +++++++++++++++++++++
2 files changed, 157 insertions(+)
create mode 100644 src/caosadvancedtools/bloxberg/__init__.py
create mode 100644 src/caosadvancedtools/bloxberg/bloxberg.py
diff --git a/src/caosadvancedtools/bloxberg/__init__.py b/src/caosadvancedtools/bloxberg/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/src/caosadvancedtools/bloxberg/bloxberg.py b/src/caosadvancedtools/bloxberg/bloxberg.py
new file mode 100644
index 00000000..492b0fe8
--- /dev/null
+++ b/src/caosadvancedtools/bloxberg/bloxberg.py
@@ -0,0 +1,157 @@
+# This file is a part of the CaosDB Project.
+#
+# Copyright (C) 2021 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2021 Daniel Hornung <d.hornung@indiscale.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+"""Interaction with the Bloxberg blockchain.
+"""
+
+
+import hashlib
+import json
+import secrets
+
+import caosdb as db
+
+from ..models.parser import parse_model_from_string
+from . import swagger_client
+
+
+class Bloxberg:
+ """A Bloxberg instance can be used to obtain or verify certificates."""
+
+ def __init__(self, connection=None):
+ """A Bloxberg instance can be used to obtain or verify certificates.
+
+Parameters
+----------
+connection : dict
+A dict with the following keys:
+ - url : The bloxberg URL. Default is "https://qa.certify.bloxberg.org"
+ """
+ self._create_conf(connection)
+ self._api_client = swagger_client.ApiClient(configuration=self._conf)
+ self._api = swagger_client.CertificateApi(self._api_client)
+
+ def _create_conf(self, connection=None):
+ """Generate a Swagger configuration object."""
+ self._conf = swagger_client.Configuration()
+ if connection:
+ if "URL" in connection:
+ self._conf.host = connection["URL"]
+
+ def certify(self, entity):
+ """Attempt to certify the given `entity` and return a certificate Record.
+
+Parameters
+----------
+entity : caosdb.Entity
+The entity to be certified
+
+Returns
+-------
+out : caosdb.Record
+A BloxbergCertificate Record with all the necessary Properties.
+"""
+ # Calculate hash
+ pepper = str(secrets.randbits(1024))
+ entity.retrieve()
+ hasher = hashlib.sha256()
+ hasher.update(pepper)
+ hasher.update(str(entity).encode(encoding="utf8"))
+ entity_hash = "0x" + hasher.hexdigest()
+ print(entity_hash)
+ pubkey = "0x9858eC18a269EE69ebfD7C38eb297996827DDa98" # TODO The key of the API server?
+ # Create body
+ body = swagger_client.Batch(public_key=pubkey, crid=[entity_hash], crid_type="sha2-256",
+ enable_ipfs=False)
+ # Submit hash & obtain response
+ result = self._api.create_bloxberg_certificate_create_bloxberg_certificate_post(body=body)
+ cert = result[0]
+ json_s = json.dumps(cert.to_dict)
+
+ from IPython import embed; embed()
+ # Generate result Record
+ data_model =
+ cert_rec = db.Record().add_parent
+ # Extract information and put into result
+ # Return result
+ pass
+
+ def verify(self, certificate):
+ """Attempt to verify the certificate.
+
+A certificate passes verification if the Bloxberg instance says it is good. Typical use cases may
+also include the `validate` step to make sure that the certificate's original data exists and
+contains what it claimed to contain when the certificate was created.
+
+This method does nothing if the verification passes, else it raises an exception.
+
+Parameters
+----------
+certificate : caosdb.Record
+The BloxbergCertificate Record which shall be verified.
+
+ """
+ raise NotImplementedError("Bloxberg first needs to implement a verification API method.")
+
+ @staticmethod
+ def json_from_certificate(certificate, filename=None):
+ """Generate a qa.certify.bloxberg.org JSON string, optionally writing it to a file.
+
+Parameters
+----------
+certificate : caosdb.Record
+The BloxbergCertificate Record for which the JSON is generated.
+
+filename : str
+Write the JSON to this file.
+"""
+ content = {}
+
+ return content
+
+
+def ensure_data_model():
+ """Make sure that the data model fits our needs.
+
+ Most importantly, this means that a suitable RecordType "BoxbergCertificate" must exist.
+ """
+ model_yaml = """
+ BloxbergCertificate:
+ obligatory_properties:
+ pepper:
+ datatype: TEXT
+ hash:
+ datatype: TEXT
+ proofValue:
+ datatype: TEXT
+ certificateJSON:
+ datatype: TEXT
+ """
+ model = parse_model_from_string(model_yaml)
+ model.sync_data_model()
+
+
+def certify_entity(entity):
+ """Certify the given entity."""
+ if isinstance(entity, int):
+ entity = db.Entity(id=entity)
+
+ blx = Bloxberg()
+ print("Obtaining certificate...")
+ certificate = blx.certify(entity)
+ from IPython import embed; embed()
+
--
GitLab