kallithea Changeset - 3cab6bc45cc3

Changeset - 3cab6bc45cc3

Parent rev.

Child rev.

[Not reviewed]

stable

0 7 0

Mads Kiilerich - 6 years ago 2019-12-29 15:11:13
mads@kiilerich.com

Grafted from: 7e353be4c536

ssh: use fingerprint when deleting public keys

Avoid relying on a database index of the full public key string.

7 files changed with 13 insertions and 13 deletions:

kallithea/controllers/admin/my_account.py

kallithea/controllers/admin/users.py

kallithea/model/ssh_key.py

kallithea/templates/admin/my_account/my_account_ssh_keys.html

kallithea/templates/admin/users/user_edit_ssh_keys.html

kallithea/tests/functional/test_admin_users.py

kallithea/tests/functional/test_my_account.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/my_account.py

➞

Show inline comments

@@ @@ -264,33 +264,33 @@ class MyAccountController(BaseController @@
     @IfSshEnabled
     def my_account_ssh_keys(self):
         c.active = 'ssh_keys'
         self.__load_data()
         c.user_ssh_keys = SshKeyModel().get_ssh_keys(request.authuser.user_id)
         return render('admin/my_account/my_account.html')
     @IfSshEnabled
     def my_account_ssh_keys_add(self):
         description = request.POST.get('description')
         public_key = request.POST.get('public_key')
         try:
             new_ssh_key = SshKeyModel().create(request.authuser.user_id,
                                                description, public_key)
             Session().commit()
             SshKeyModel().write_authorized_keys()
             h.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
         except SshKeyModelException as errors:
             h.flash(errors.message, category='error')
         raise HTTPFound(location=url('my_account_ssh_keys'))
     @IfSshEnabled
     def my_account_ssh_keys_delete(self):
-        public_key = request.POST.get('del_public_key')
+        fingerprint = request.POST.get('del_public_key_fingerprint')
         try:
-            SshKeyModel().delete(public_key, request.authuser.user_id)
+            SshKeyModel().delete(fingerprint, request.authuser.user_id)
             Session().commit()
             SshKeyModel().write_authorized_keys()
             h.flash(_("SSH key successfully deleted"), category='success')
         except SshKeyModelException as errors:
             h.flash(errors.message, category='error')
         raise HTTPFound(location=url('my_account_ssh_keys'))

kallithea/controllers/admin/users.py

➞

Show inline comments

@@ @@ -441,33 +441,33 @@ class UsersController(BaseController): @@
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @IfSshEnabled
     def ssh_keys_add(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         description = request.POST.get('description')
         public_key = request.POST.get('public_key')
         try:
             new_ssh_key = SshKeyModel().create(c.user.user_id,
                                                description, public_key)
             Session().commit()
             SshKeyModel().write_authorized_keys()
             h.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
         except SshKeyModelException as errors:
             h.flash(errors.message, category='error')
         raise HTTPFound(location=url('edit_user_ssh_keys', id=c.user.user_id))
     @IfSshEnabled
     def ssh_keys_delete(self, id):
         c.user = self._get_user_or_raise_if_default(id)
-        public_key = request.POST.get('del_public_key')
+        fingerprint = request.POST.get('del_public_key_fingerprint')
         try:
-            SshKeyModel().delete(public_key, c.user.user_id)
+            SshKeyModel().delete(fingerprint, c.user.user_id)
             Session().commit()
             SshKeyModel().write_authorized_keys()
             h.flash(_("SSH key successfully deleted"), category='success')
         except SshKeyModelException as errors:
             h.flash(errors.message, category='error')
         raise HTTPFound(location=url('edit_user_ssh_keys', id=c.user.user_id))

kallithea/model/ssh_key.py

➞

Show inline comments

@@ @@ -51,63 +51,63 @@ class SshKeyModel(object): @@
         Will raise SshKeyModelException on errors
         """
         try:
             keytype, pub, comment = ssh.parse_pub_key(public_key)
         except ssh.SshKeyParseError as e:
             raise SshKeyModelException(_('SSH key %r is invalid: %s') % (safe_str(public_key), e.message))
         if not description.strip():
             description = comment.strip()
         user = User.guess_instance(user)
         new_ssh_key = UserSshKeys()
         new_ssh_key.user_id = user.user_id
         new_ssh_key.description = description
         new_ssh_key.public_key = public_key
         for ssh_key in UserSshKeys.query().filter(UserSshKeys.fingerprint == new_ssh_key.fingerprint).all():
             raise SshKeyModelException(_('SSH key %s is already used by %s') %
                                        (new_ssh_key.fingerprint, ssh_key.user.username))
         Session().add(new_ssh_key)
         return new_ssh_key
-    def delete(self, public_key, user=None):
+    def delete(self, fingerprint, user=None):
         """
         Deletes given public_key, if user is set it also filters the object for
         deletion by given user.
         Deletes ssh key with given fingerprint. If user is set, it also filters
         the object for deletion by given user.
         Will raise SshKeyModelException on errors
         """
-        ssh_key = UserSshKeys.query().filter(UserSshKeys._public_key == public_key)
+        ssh_key = UserSshKeys.query().filter(UserSshKeys.fingerprint == fingerprint)
         if user:
             user = User.guess_instance(user)
             ssh_key = ssh_key.filter(UserSshKeys.user_id == user.user_id)
         ssh_key = ssh_key.scalar()
         if ssh_key is None:
-            raise SshKeyModelException(_('SSH key %r not found') % safe_str(public_key))
+            raise SshKeyModelException(_('SSH key with fingerprint %r found') % safe_str(fingerprint))
         Session().delete(ssh_key)
     def get_ssh_keys(self, user):
         user = User.guess_instance(user)
         user_ssh_keys = UserSshKeys.query() \
             .filter(UserSshKeys.user_id == user.user_id).all()
         return user_ssh_keys
     def write_authorized_keys(self):
         if not str2bool(config.get('ssh_enabled', False)):
             log.error("Will not write SSH authorized_keys file - ssh_enabled is not configured")
             return
         authorized_keys = config.get('ssh_authorized_keys')
         kallithea_cli_path = config.get('kallithea_cli_path', 'kallithea-cli')
         if not authorized_keys:
             log.error('Cannot write SSH authorized_keys file - ssh_authorized_keys is not configured')
             return
         log.info('Writing %s', authorized_keys)
         authorized_keys_dir = os.path.dirname(authorized_keys)
         try:
             os.makedirs(authorized_keys_dir)
             os.chmod(authorized_keys_dir, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) # ~/.ssh/ must be 0700
         except OSError as exception:

kallithea/templates/admin/my_account/my_account_ssh_keys.html

➞

Show inline comments

@@ @@ -2,49 +2,49 @@ @@
     %if c.user_ssh_keys:
         <tr>
             <th>${_('Fingerprint')}</th>
             <th>${_('Description')}</th>
             <th>${_('Last Used')}</th>
             <th>${_('Action')}</th>
         </tr>
         %for ssh_key in c.user_ssh_keys:
           <tr>
             <td>
                 ${ssh_key.fingerprint}
             </td>
             <td>
                 ${ssh_key.description}
             </td>
             <td>
               %if ssh_key.last_seen:
                 ${h.fmt_date(ssh_key.last_seen)}
               %else:
                 ${_('Never')}
               %endif
             </td>
             <td>
                 ${h.form(url('my_account_ssh_keys_delete'))}
-                    ${h.hidden('del_public_key', ssh_key.public_key)}
+                    ${h.hidden('del_public_key_fingerprint', ssh_key.fingerprint)}
                     <button class="btn btn-danger btn-xs" type="submit"
                             onclick="return confirm('${_('Confirm to remove this SSH key: %s') % ssh_key.fingerprint}');">
                         <i class="icon-trashcan"></i>
                         ${_('Remove')}
                     </button>
                 ${h.end_form()}
             </td>
           </tr>
         %endfor
     %else:
         <tr>
             <td>
                 <div class="ip">${_('No SSH keys have been added')}</div>
             </td>
         </tr>
     %endif
 </table>
 <div>
     ${h.form(url('my_account_ssh_keys'))}
     <div class="form">
             <div class="form-group">
                 <label class="control-label">${_('New SSH key')}</label>
             </div>

kallithea/templates/admin/users/user_edit_ssh_keys.html

➞

Show inline comments

@@ @@ -2,49 +2,49 @@ @@
     %if c.user_ssh_keys:
         <tr>
             <th>${_('Fingerprint')}</th>
             <th>${_('Description')}</th>
             <th>${_('Last Used')}</th>
             <th>${_('Action')}</th>
         </tr>
         %for ssh_key in c.user_ssh_keys:
           <tr>
             <td>
                 ${ssh_key.fingerprint}
             </td>
             <td>
                 ${ssh_key.description}
             </td>
             <td>
               %if ssh_key.last_seen:
                 ${h.fmt_date(ssh_key.last_seen)}
               %else:
                 ${_('Never')}
               %endif
             </td>
             <td>
                 ${h.form(url('edit_user_ssh_keys_delete', id=c.user.user_id))}
-                    ${h.hidden('del_public_key', ssh_key.public_key)}
+                    ${h.hidden('del_public_key_fingerprint', ssh_key.fingerprint)}
                     <button class="btn btn-danger btn-xs" type="submit"
                             onclick="return confirm('${_('Confirm to remove this SSH key: %s') % ssh_key.fingerprint}');">
                         <i class="icon-trashcan"></i>
                         ${_('Remove')}
                     </button>
                 ${h.end_form()}
             </td>
           </tr>
         %endfor
     %else:
         <tr>
             <td>
                 <div class="ip">${_('No SSH keys have been added')}</div>
             </td>
         </tr>
     %endif
 </table>
 <div>
     ${h.form(url('edit_user_ssh_keys', id=c.user.user_id))}
     <div class="form">
             <div class="form-group">
                 <label class="control-label">${_('New SSH key')}</label>
             </div>

kallithea/tests/functional/test_admin_users.py

➞

Show inline comments

@@ @@ -535,49 +535,49 @@ class TestAdminUsersController(TestContr @@
         assert ssh_key.fingerprint == fingerprint
         assert ssh_key.description == description
         Session().delete(ssh_key)
         Session().commit()
     def test_remove_ssh_key(self):
         description = u''
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user()
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         user_id = user.user_id
         response = self.app.post(url('edit_user_ssh_keys', id=user_id),
                                  {'description': description,
                                   'public_key': public_key,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response.follow()
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.description == u'me@localhost'
         response = self.app.post(url('edit_user_ssh_keys_delete', id=user_id),
-                                 {'del_public_key': ssh_key.public_key,
+                                 {'del_public_key_fingerprint': ssh_key.fingerprint,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key successfully deleted')
         keys = UserSshKeys.query().all()
         assert 0 == len(keys)
 class TestAdminUsersController_unittest(TestController):
     """ Unit tests for the users controller """
     def test_get_user_or_raise_if_default(self, monkeypatch, test_context_fixture):
         # flash complains about an non-existing session
         def flash_mock(*args, **kwargs):
             pass
         monkeypatch.setattr(h, 'flash', flash_mock)
         u = UsersController()
         # a regular user should work correctly
         user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         assert u._get_user_or_raise_if_default(user.user_id) == user
         # the default user should raise
         with pytest.raises(HTTPNotFound):
             u._get_user_or_raise_if_default(User.get_default_user().user_id)

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

@@ @@ -268,29 +268,29 @@ class TestMyAccountController(TestContro @@
         user_id = response.session['authuser']['user_id']
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.fingerprint == fingerprint
         assert ssh_key.description == description
         Session().delete(ssh_key)
         Session().commit()
     def test_my_account_remove_ssh_key(self):
         description = u''
         public_key = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== me@localhost'
         fingerprint = u'Ke3oUCNJM87P0jJTb3D+e3shjceP2CqMpQKVd75E9I8'
         self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         response = self.app.post(url('my_account_ssh_keys'),
                                  {'description': description,
                                   'public_key': public_key,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response.follow()
         user_id = response.session['authuser']['user_id']
         ssh_key = UserSshKeys.query().filter(UserSshKeys.user_id == user_id).one()
         assert ssh_key.description == u'me@localhost'
         response = self.app.post(url('my_account_ssh_keys_delete'),
-                                 {'del_public_key': ssh_key.public_key,
+                                 {'del_public_key_fingerprint': ssh_key.fingerprint,
                                   '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'SSH key successfully deleted')
         keys = UserSshKeys.query().all()
         assert 0 == len(keys)

0 comments (0 inline, 0 general)