Merge branch 'develop' into main

1 files changed, 108 insertions, 0 deletions

diff --git a/bsfs/utils/uuid.py b/bsfs/utils/uuid.py
new file mode 100644
index 0000000..6366b18
--- /dev/null
+++ b/bsfs/utils/uuid.py
@@ -0,0 +1,108 @@
+"""
+
+Part of the BlackStar filesystem (bsfs) module.
+A copy of the license is provided with the project.
+Author: Matthias Baumgartner, 2022
+"""
+# imports
+from collections import abc
+import hashlib
+import os
+import platform
+import random
+import threading
+import time
+import typing
+import uuid
+
+# constants
+HASH = hashlib.sha256
+
+# exports
+__all__: typing.Sequence[str] = [
+    'UCID',
+    'UUID',
+    ]
+
+
+## code ##
+
+class UUID(abc.Iterator, abc.Callable): # type: ignore [misc] # abc.Callable "is an invalid base class"
+    """Generate 256-bit universally unique IDs.
+
+    This is a 'best-effort' kind of implementation that tries to ensure global
+    uniqueness, even tough actual uniqueness cannot be guaranteed.
+    The approach is different from python's uuid module (which implements
+    RFC 4122) in that it generates longer UUIDs and in that it cannot be
+    reconstructed whether two UUIDs were generated on the same system.
+
+    The ID is a cryptographic hash over several components:
+    * host
+    * system
+    * process
+    * thread
+    * random
+    * time
+    * cpu cycles
+    * content (if available)
+
+    """
+
+    # host identifier
+    host: str
+
+    # system identifier
+    system: str
+
+    # process identifier
+    process: str
+
+    # thread identifier
+    thread: str
+
+    def __init__(self, seed: typing.Optional[int] = None):
+        # initialize static components
+        self.host = str(uuid.getnode())
+        self.system = '-'.join(platform.uname())
+        self.process = str(os.getpid())
+        self.thread = str(threading.get_ident())
+        # initialize random component
+        random.seed(seed)
+
+    def __call__(self, content: typing.Optional[str] = None) -> str: # pylint: disable=arguments-differ
+        """Return a globally unique ID."""
+        # content component
+        content = str(content) if content is not None else ''
+        # time component
+        now = str(time.time())
+        # clock component
+        clk = str(time.perf_counter())
+        # random component
+        rnd = str(random.random())
+        # build the token from all available components
+        token = self.host + self.system + self.process + self.thread + rnd + now + clk + content
+        # return the token's hash
+        return HASH(token.encode('ascii', 'ignore')).hexdigest()
+
+    def __iter__(self) -> typing.Iterator[str]:
+        """Iterate indefinitely over universally unique IDs."""
+        return self
+
+    def __next__(self) -> str:
+        """Generate universally unique IDs."""
+        return self()
+
+
+class UCID():
+    """Generate 256-bit content IDs.
+
+    Effectively computes a cryptographic hash over the content.
+
+    """
+    @staticmethod
+    def from_path(path: str) -> str:
+        """Read the content from a file."""
+        with open(path, 'rb') as ifile:
+            return HASH(ifile.read()).hexdigest()
+
+## EOF ##