Make sure that all ssh keys are unique, regardless of the user

README.md

@@ -115,6 +115,7 @@ Professional support for Rdiffweb is available by contacting [IKUS Soft](https:/
 * Sent email notification to user when a new SSH Key get added - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
 * Ratelimit "Resend code to my email" in Two-Factor Authentication view - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
 * Username are not case-insensitive - credits to [raiders0786](https://www.linkedin.com/in/chirag-agrawal-770488144/)
+* Make sure that all ssh keys are unique, regardless of the user - credit to [Nehal Pillai](https://www.linkedin.com/in/nehal-pillai-02a854172)
 
 Breaking changes:
 

rdiffweb/core/model/__init__.py

@@ -19,12 +19,12 @@
 import sys
 
 import cherrypy
-from sqlalchemy import event
+from sqlalchemy import event, func
 from sqlalchemy.exc import IntegrityError
 
 from ._repo import RepoObject  # noqa
 from ._session import DbSession, SessionObject  # noqa
-from ._sshkey import SshKey  # noqa
+from ._sshkey import SshKey, sshkey_fingerprint_index  # noqa
 from ._token import Token  # noqa
 from ._user import DuplicateSSHKeyError, UserObject, user_username_index  # noqa
 
@@ -137,3 +137,22 @@ def db_after_create(target, connection, **kw):
             logger.error(msg)
             print(msg, file=sys.stderr)
             raise SystemExit(12)
+
+    # Fix SSH Key uniqueness - since 2.5.4
+    if not _index_exists(connection, 'sshkey_fingerprint_index'):
+        duplicate_sshkeys = (
+            SshKey.query.with_entities(SshKey.fingerprint)
+            .group_by(SshKey.fingerprint)
+            .having(func.count(SshKey.fingerprint) > 1)
+        ).all()
+        try:
+            sshkey_fingerprint_index.create()
+        except IntegrityError:
+            msg = (
+                'Failure to upgrade your database to make SSH Keys unique. '
+                'You must downgrade and deleted duplicate SSH Keys. '
+                '%s' % '\n'.join([str(k) for k in duplicate_sshkeys]),
+            )
+            logger.error(msg)
+            print(msg, file=sys.stderr)
+            raise SystemExit(12)

rdiffweb/core/model/_sshkey.py

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import cherrypy
-from sqlalchemy import Column, Integer, Text
+from sqlalchemy import Column, Index, Integer, Text
 
 Base = cherrypy.tools.db.get_base()
 
@@ -27,3 +27,7 @@ class SshKey(Base):
     fingerprint = Column('Fingerprint', Text)
     key = Column('Key', Text, unique=True, primary_key=True)
     userid = Column('UserID', Integer, nullable=False)
+
+
+# Make finger print unique
+sshkey_fingerprint_index = Index('sshkey_fingerprint_index', SshKey.fingerprint, unique=True)

rdiffweb/core/model/_user.py

@@ -159,7 +159,7 @@ def add_authorizedkey(self, key, comment=None):
         assert key
         key = authorizedkeys.check_publickey(key)
 
-        # Remove option, replace comments.
+        # Remove option & Remove comment for SQL storage
         key = authorizedkeys.AuthorizedKey(
             options=None, keytype=key.keytype, key=key.key, comment=comment or key.comment
         )
@@ -176,7 +176,8 @@ def add_authorizedkey(self, key, comment=None):
             # Also look in database.
             logger.info("add key [%s] to [%s] database", key, self.username)
             try:
-                SshKey(userid=self.userid, fingerprint=key.fingerprint, key=key.getvalue()).add().flush()
+                sshkey = SshKey(userid=self.userid, fingerprint=key.fingerprint, key=key.getvalue())
+                sshkey.add().flush()
             except IntegrityError:
                 raise DuplicateSSHKeyError(
                     _("Duplicate key. This key already exists or is associated to another user.")

rdiffweb/core/model/tests/test_user.py

@@ -325,15 +325,47 @@ def test_add_authorizedkey_without_file(self):
     def test_add_authorizedkey_duplicate(self):
         # Read the pub key
         key = self._read_ssh_key()
-        # Add the key to the user
+        # Given a user with a SSH Key
         userobj = UserObject.get_user(self.USERNAME)
         userobj.add_authorizedkey(key)
         userobj.commit()
-        # Add the same key
+
+        # When adding the same identical key.
+        # Then an error is raised
         with self.assertRaises(DuplicateSSHKeyError):
             userobj.add_authorizedkey(key)
             userobj.commit()
 
+    def test_add_authorizedkey_duplicate_new_comment(self):
+        # Read the pub key
+        key = self._read_ssh_key()
+        # Given a user with a SSH Key
+        userobj = UserObject.get_user(self.USERNAME)
+        userobj.add_authorizedkey(key)
+        userobj.commit()
+
+        # When adding the same key with a different comment
+        # Then an error is raised
+        with self.assertRaises(DuplicateSSHKeyError):
+            userobj.add_authorizedkey(key, comment="new comment")
+            userobj.commit()
+
+    def test_add_authorizedkey_duplicate_new_user(self):
+        # Read the pub key
+        key = self._read_ssh_key()
+        # Given a user with a SSH Key
+        userobj = UserObject.get_user(self.USERNAME)
+        userobj.add_authorizedkey(key)
+        userobj.commit()
+
+        # When adding the same key to a different user
+        # Then an error is raised
+        newuser = UserObject.add_user("newuser")
+        newuser.commit()
+        with self.assertRaises(DuplicateSSHKeyError):
+            newuser.add_authorizedkey(key, comment="new comment")
+            newuser.commit()
+
     def test_add_authorizedkey_with_file(self):
         """
         Add an ssh key for a user with an authorizedkey file.