add a check for hash uniqueness across all tables

Author: laowantong

sqlab/__init__.py

@@ -1 +1 @@
-__version__ = "0.5.3"
+__version__ = "0.5.4"

sqlab/cmd_create.py

@@ -79,6 +79,25 @@ def run(config: dict):
     sql_dump.write(data_inserts_queries)
     db.execute_non_select(data_inserts_queries)
 
+    # Check that the hash columns contain unique values across all tables. An alternative would be
+    # to declare each hash column as UNIQUE, but:
+    # 1. The uniqueness would not be cross-table.
+    # 2. In MySQL, the hash column is calculated from all other columns except the auto-incremented
+    #    primary key column. In a stateful game like SQL Island, the player is instructed to insert
+    #    a row in the table inhabitant. If they repeat the same insertion, the hash will be the
+    #    same (although the personid will be different), raising an IntegrityError.
+
+    seen_hashes = {}  # use a dictionary for better warning messages
+    for table_name in db.get_table_names():
+        query = f"SELECT * FROM {table_name};"
+        (_, _, rows) = db.execute_select(query)
+        for row in rows:
+            if (hash := row[-1]) in seen_hashes:
+                (t1, v1) = seen_hashes[hash]
+                (t2, v2) = (table_name, row[:-1])
+                print(f"{WARNING}Hash collision:\n    {t1}: {v1}\n    {t2}: {v2}\nhave same hash {hash}.{RESET}")
+            seen_hashes[hash] = (table_name, row[:-1])
+
     sql_dump.write(db.fk_constraints_queries)
 
     # If the source is a notebook, parse it and populate the `records` list.

sqlab/database.py

@@ -35,6 +35,12 @@ def get_headers(self, table: str, keep_auto_increment_columns=True) -> list[str]
         """
         raise NotImplementedError
 
+    def get_table_names(self) -> list[str]:
+        """Return the names of all the tables in the DB, except the utility tables.
+        These include those starting with "sqlab_", and, in SQLite, the virtual tables
+        decrypt and sqlean_define."""
+        raise NotImplementedError
+
     def encrypt(self, plain: str, token: int) -> str:
         """Return the encrypted version of the given plain text."""
         raise NotImplementedError

sqlab/dbms/mysql/database.py

@@ -37,6 +37,17 @@ def get_headers(self, table: str, keep_auto_increment_columns=False) -> list[str
             headers = [row[0] for row in cursor]
         return headers
 
+    def get_table_names(self) -> list[str]:
+        query = f"""
+            SELECT table_name
+            FROM information_schema.tables
+            WHERE table_schema = "{self.cnx.database}"
+                AND table_name NOT LIKE 'sqlab_%';
+        """
+        with self.cnx.cursor() as cursor:
+            cursor.execute(query)
+            return [row[0] for row in cursor]
+    
     def encrypt(self, clear_text, token):
         """In MySQL, the function aes_encrypt() takes a numeric key."""
         with self.cnx.cursor() as cursor:

sqlab/dbms/postgresql/database.py

@@ -39,6 +39,17 @@ def get_headers(self, table, keep_auto_increment_columns=True):
             headers = [row[0] for row in cursor.fetchall()]
         return headers
 
+    def get_table_names(self) -> list[str]:
+        query = """
+            SELECT table_name
+            FROM information_schema.tables
+            WHERE table_schema = 'public'
+                AND table_name NOT LIKE 'sqlab_%';
+        """
+        with self.cnx.cursor() as cursor:
+            cursor.execute(query)
+            return [row[0] for row in cursor]
+
     def encrypt(self, clear_text, token):
         """
         In PostgreSQL, the function pgp_sym_encrypt() takes a textual key, not a numeric one.

sqlab/dbms/sqlite/database.py

@@ -44,6 +44,18 @@ def get_headers(self, table: str, keep_auto_increment_columns=True) -> list[str]
         headers = [header for header in headers if header != "hash"]
         return headers
 
+    def get_table_names(self) -> list[str]:
+        query = """
+            SELECT name
+            FROM sqlite_master
+            WHERE type = 'table'
+                AND name NOT LIKE 'sqlab_%'
+                AND name NOT IN ('sqlean_define', 'decrypt');
+        """
+        cursor = self.cnx.cursor()
+        cursor.execute(query)
+        return [row[0] for row in cursor.fetchall()]
+
     def encrypt(self, clear_text, token):
         query = f"SELECT encode(sha256({token}), 'hex') || encode(brotli({repr(clear_text)}), 'hex');"
         return repr(self.execute_select(query)[2][0][0])