kallithea Changeset - e53f1db2b839

Changeset - e53f1db2b839

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Christian Oyarzun - 11 years ago 2014-11-17 20:42:45
oyarzun@gmail.com

ssh: when adding a new public key, if no description is provided and the SSH key itself has a comment, use that as description

Based on work by Ilya Beda <ir4y.ix@gmail.com> on
https://bitbucket.org/ir4y/rhodecode/commits/branch/ssh_server_support , also
heavily modified by Mads Kiilerich.

3 files changed with 5 insertions and 3 deletions:

kallithea/model/ssh_key.py

kallithea/tests/functional/test_admin_users.py

kallithea/tests/functional/test_my_account.py

0 comments (0 inline, 0 general)

kallithea/model/ssh_key.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU 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 General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.ssh_key
 ~~~~~~~~~~~~~~~~~~~~~~~
 SSH key model for Kallithea
 """
 import logging
 from tg.i18n import ugettext as _
 from kallithea.lib.utils2 import safe_str
 from kallithea.model.db import UserSshKeys, User
 from kallithea.model.meta import Session
 from kallithea.lib import ssh
 log = logging.getLogger(__name__)
 class SshKeyModelException(Exception):
     """Exception raised by SshKeyModel methods to report errors"""
 class SshKeyModel(object):
     def create(self, user, description, public_key):
         """
         :param user: user or user_id
         :param description: description of SshKey
         :param publickey: public key text
         Will raise SshKeyModelException on errors
         """
         try:
             ssh.parse_pub_key(public_key)
+            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):
         """
         Deletes given public_key, 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)
         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))
         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

kallithea/tests/functional/test_admin_users.py

➞

Show inline comments

@@ @@ -508,97 +508,97 @@ class TestAdminUsersController(TestContr @@
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('edit_user_api_keys_delete', id=user_id),
                  {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])
     def test_add_ssh_key(self):
         description = u'something'
         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,
                                   '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response = response.follow()
         response.mustcontain(fingerprint)
         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_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,
                                   '_authentication_token': self.authentication_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 == description
+        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,
                                   '_authentication_token': self.authentication_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)
 class TestAdminUsersControllerForDefaultUser(TestController):
     """
     Edit actions on the default user are not allowed.
     Validate that they throw a 404 exception.
     """
     def test_edit_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user', id=user.user_id), status=404)
     def test_edit_advanced_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)
     # API keys
     def test_edit_api_keys_default_user(self):
         self.log_user()
         user = User.get_default_user()
         response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)

kallithea/tests/functional/test_my_account.py

➞

Show inline comments

@@ @@ -240,56 +240,56 @@ class TestMyAccountController(TestContro @@
         usr = self.log_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS)
         user = User.get(usr['user_id'])
         api_key = user.api_key
         response = self.app.get(url('my_account_api_keys'))
         response.mustcontain(api_key)
         response.mustcontain('Expires: Never')
         response = self.app.post(url('my_account_api_keys_delete'),
                  {'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'API key successfully reset')
         response = response.follow()
         response.mustcontain(no=[api_key])
     def test_my_account_add_ssh_key(self):
         description = u'something'
         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,
                                   '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'SSH key %s successfully added' % fingerprint)
         response = response.follow()
         response.mustcontain(fingerprint)
         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,
                                   '_authentication_token': self.authentication_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 == description
+        assert ssh_key.description == u'me@localhost'
         response = self.app.post(url('my_account_ssh_keys_delete'),
                                  {'del_public_key': ssh_key.public_key,
                                   '_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'SSH key successfully deleted')
         keys = UserSshKeys.query().all()
         assert 0 == len(keys)

0 comments (0 inline, 0 general)