Changeset - e53f1db2b839
[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:
0 comments (0 inline, 0 general)
kallithea/model/ssh_key.py
Show inline comments
 
@@ -24,51 +24,53 @@ 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)
 

	
kallithea/tests/functional/test_admin_users.py
Show inline comments
 
@@ -532,49 +532,49 @@ class TestAdminUsersController(TestContr
 
        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):
kallithea/tests/functional/test_my_account.py
Show inline comments
 
@@ -264,32 +264,32 @@ class TestMyAccountController(TestContro
 

	
 
        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)