Changeset - 66c208bf56fe
[Not reviewed]
default
0 4 1
Tim Freund - 11 years ago 2014-11-17 20:40:35
tim@freunds.net
ssh: user management of ssh keys

Add user interface for managing SSH based access.

The work in this commit is based heavily off of the existing API key
code for the sake of consistency.

Updates to use Bootstrap, request.authuser, POST methods and pytest by Anton
Schur <tonich.sh@gmail.com>.

Additional Bootstrap fixes by Dominik Ruf.

The original code has been heavily modified by Mads Kiilerich.
5 files changed with 145 insertions and 2 deletions:
0 comments (0 inline, 0 general)
kallithea/config/routing.py
Show inline comments
 
@@ -337,48 +337,55 @@ def make_map(config):
 

	
 
        m.connect("my_account_repos", "/my_account/repos",
 
                  action="my_account_repos", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_watched", "/my_account/watched",
 
                  action="my_account_watched", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_perms", "/my_account/perms",
 
                  action="my_account_perms", conditions=dict(method=["GET"]))
 

	
 
        m.connect("my_account_emails", "/my_account/emails",
 
                  action="my_account_emails", conditions=dict(method=["GET"]))
 
        m.connect("my_account_emails", "/my_account/emails",
 
                  action="my_account_emails_add", conditions=dict(method=["POST"]))
 
        m.connect("my_account_emails_delete", "/my_account/emails/delete",
 
                  action="my_account_emails_delete", conditions=dict(method=["POST"]))
 

	
 
        m.connect("my_account_api_keys", "/my_account/api_keys",
 
                  action="my_account_api_keys", conditions=dict(method=["GET"]))
 
        m.connect("my_account_api_keys", "/my_account/api_keys",
 
                  action="my_account_api_keys_add", conditions=dict(method=["POST"]))
 
        m.connect("my_account_api_keys_delete", "/my_account/api_keys/delete",
 
                  action="my_account_api_keys_delete", conditions=dict(method=["POST"]))
 

	
 
        m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
 
                  action="my_account_ssh_keys", conditions=dict(method=["GET"]))
 
        m.connect("my_account_ssh_keys", "/my_account/ssh_keys",
 
                  action="my_account_ssh_keys_add", conditions=dict(method=["POST"]))
 
        m.connect("my_account_ssh_keys_delete", "/my_account/ssh_keys/delete",
 
                  action="my_account_ssh_keys_delete", conditions=dict(method=["POST"]))
 

	
 
    # ADMIN GIST
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/gists') as m:
 
        m.connect("gists", "/gists",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("gists", "/gists",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_gist", "/gists/new",
 
                  action="new", conditions=dict(method=["GET"]))
 

	
 
        m.connect("gist_delete", "/gists/{gist_id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("edit_gist", "/gists/{gist_id}/edit",
 
                  action="edit", conditions=dict(method=["GET", "POST"]))
 
        m.connect("edit_gist_check_revision", "/gists/{gist_id}/edit/check_revision",
 
                  action="check_revision", conditions=dict(method=["POST"]))
 

	
 
        m.connect("gist", "/gists/{gist_id}",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("gist_rev", "/gists/{gist_id}/{revision}",
 
                  revision="tip",
 
                  action="show", conditions=dict(method=["GET"]))
 
        m.connect("formatted_gist", "/gists/{gist_id}/{revision}/{format}",
 
                  revision="tip",
kallithea/controllers/admin/my_account.py
Show inline comments
 
@@ -18,55 +18,56 @@ kallithea.controllers.admin.my_account
 
my account controller for Kallithea admin
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: August 20, 2013
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import logging
 
import traceback
 
import formencode
 

	
 
from sqlalchemy import func
 
from formencode import htmlfill
 
from tg import request, tmpl_context as c
 
from tg.i18n import ugettext as _
 
from webob.exc import HTTPFound
 

	
 
from kallithea.config.routing import url
 
from kallithea.lib import helpers as h
 
from kallithea.lib import auth_modules
 
from kallithea.lib.auth import LoginRequired, AuthUser
 
from kallithea.lib.base import BaseController, render
 
from kallithea.lib.base import BaseController, render, IfSshEnabled
 
from kallithea.lib.utils2 import generate_api_key, safe_int
 
from kallithea.model.db import Repository, UserEmailMap, User, UserFollowing
 
from kallithea.model.forms import UserForm, PasswordChangeForm
 
from kallithea.model.user import UserModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.api_key import ApiKeyModel
 
from kallithea.model.ssh_key import SshKeyModel
 
from kallithea.model.meta import Session
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class MyAccountController(BaseController):
 
    """REST Controller styled on the Atom Publishing Protocol"""
 
    # To properly map this controller, ensure your config/routing.py
 
    # file has a resource setup:
 
    #     map.resource('setting', 'settings', controller='admin/settings',
 
    #         path_prefix='/admin', name_prefix='admin_')
 

	
 
    @LoginRequired()
 
    def _before(self, *args, **kwargs):
 
        super(MyAccountController, self)._before(*args, **kwargs)
 

	
 
    def __load_data(self):
 
        c.user = User.get(request.authuser.user_id)
 
        if c.user.is_default_user:
 
            h.flash(_("You can't edit this user since it's"
 
                      " crucial for entire application"), category='warning')
 
            raise HTTPFound(location=url('users'))
 

	
 
    def _load_my_repos_data(self, watched=False):
 
@@ -238,24 +239,49 @@ class MyAccountController(BaseController
 
                                                     show_expired=show_expired)
 
        return render('admin/my_account/my_account.html')
 

	
 
    def my_account_api_keys_add(self):
 
        lifetime = safe_int(request.POST.get('lifetime'), -1)
 
        description = request.POST.get('description')
 
        ApiKeyModel().create(request.authuser.user_id, description, lifetime)
 
        Session().commit()
 
        h.flash(_("API key successfully created"), category='success')
 
        raise HTTPFound(location=url('my_account_api_keys'))
 

	
 
    def my_account_api_keys_delete(self):
 
        api_key = request.POST.get('del_api_key')
 
        if request.POST.get('del_api_key_builtin'):
 
            user = User.get(request.authuser.user_id)
 
            user.api_key = generate_api_key()
 
            Session().commit()
 
            h.flash(_("API key successfully reset"), category='success')
 
        elif api_key:
 
            ApiKeyModel().delete(api_key, request.authuser.user_id)
 
            Session().commit()
 
            h.flash(_("API key successfully deleted"), category='success')
 

	
 
        raise HTTPFound(location=url('my_account_api_keys'))
 

	
 
    @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')
 
        new_ssh_key = SshKeyModel().create(request.authuser.user_id,
 
                                           description, public_key)
 
        Session().commit()
 
        h.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
 
        raise HTTPFound(location=url('my_account_ssh_keys'))
 

	
 
    @IfSshEnabled
 
    def my_account_ssh_keys_delete(self):
 
        public_key = request.POST.get('del_public_key')
 
        SshKeyModel().delete(public_key, request.authuser.user_id)
 
        Session().commit()
 
        h.flash(_("SSH key successfully deleted"), category='success')
 
        raise HTTPFound(location=url('my_account_ssh_keys'))
kallithea/templates/admin/my_account/my_account.html
Show inline comments
 
@@ -4,37 +4,40 @@
 
<%block name="title">
 
    ${_('My Account')} ${request.authuser.username}
 
</%block>
 

	
 
<%def name="breadcrumbs_links()">
 
    ${_('My Account')}
 
</%def>
 

	
 
<%block name="header_menu">
 
    ${self.menu('admin')}
 
</%block>
 

	
 
<%def name="main()">
 
<div class="panel panel-primary">
 
    <div class="panel-heading">
 
        ${self.breadcrumbs()}
 
    </div>
 

	
 
    ##main
 
    <div class="panel-body settings">
 
        <ul class="nav nav-pills nav-stacked">
 
            <li class="${'active' if c.active=='profile' else ''}"><a href="${h.url('my_account')}">${_('Profile')}</a></li>
 
            <li class="${'active' if c.active=='emails' else ''}"><a href="${h.url('my_account_emails')}">${_('Email Addresses')}</a></li>
 
            <li class="${'active' if c.active=='password' else ''}"><a href="${h.url('my_account_password')}">${_('Password')}</a></li>
 
            %if c.ssh_enabled:
 
              <li class="${'active' if c.active=='ssh_keys' else ''}"><a href="${h.url('my_account_ssh_keys')}">${_('SSH Keys')}</a></li>
 
            %endif
 
            <li class="${'active' if c.active=='api_keys' else ''}"><a href="${h.url('my_account_api_keys')}">${_('API Keys')}</a></li>
 
            <li class="${'active' if c.active=='repos' else ''}"><a href="${h.url('my_account_repos')}">${_('Owned Repositories')}</a></li>
 
            <li class="${'active' if c.active=='watched' else ''}"><a href="${h.url('my_account_watched')}">${_('Watched Repositories')}</a></li>
 
            <li class="${'active' if c.active=='perms' else ''}"><a href="${h.url('my_account_perms')}">${_('Show Permissions')}</a></li>
 
        </ul>
 

	
 
        <div>
 
            <%include file="/admin/my_account/my_account_${c.active}.html"/>
 
        </div>
 
    </div>
 
</div>
 

	
 
</%def>
kallithea/templates/admin/my_account/my_account_ssh_keys.html
Show inline comments
 
new file 100644
 
<table class="table">
 
    %if c.user_ssh_keys:
 
        <tr>
 
            <th>${_('Fingerprint')}</th>
 
            <th>${_('Description')}</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>
 
                ${h.form(url('my_account_ssh_keys_delete'))}
 
                    ${h.hidden('del_public_key', ssh_key.public_key)}
 
                    <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>
 
            <div class="form-group">
 
                <label class="control-label" for="public_key">${_('Public key')}:</label>
 
                <div>
 
                    ${h.textarea('public_key', '', class_='form-control', placeholder=_('Public key (contents of e.g. ~/.ssh/id_rsa.pub)'), cols=80, rows=5)}
 
                </div>
 
            </div>
 
            <div class="form-group">
 
                <label class="control-label" for="description">${_('Description')}:</label>
 
                <div>
 
                    ${h.text('description', class_='form-control', placeholder=_('Description'))}
 
                </div>
 
            </div>
 
            <div class="form-group">
 
                <div class="buttons">
 
                    ${h.submit('save', _('Add'), class_="btn btn-default")}
 
                    ${h.reset('reset', _('Reset'), class_="btn btn-default")}
 
                </div>
 
            </div>
 
    </div>
 
    ${h.end_form()}
 
</div>
kallithea/tests/functional/test_my_account.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys
 
from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys, UserSshKeys
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib import helpers as h
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 

	
 
from tg.util.webtest import test_context
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestMyAccountController(TestController):
 
    test_user_1 = 'testme'
 

	
 
    @classmethod
 
    def teardown_class(cls):
 
        if User.get_by_username(cls.test_user_1):
 
            UserModel().delete(cls.test_user_1)
 
            Session().commit()
 

	
 
    def test_my_account(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account'))
 

	
 
@@ -228,24 +228,68 @@ class TestMyAccountController(TestContro
 

	
 
        # now delete our key
 
        keys = UserApiKeys.query().all()
 
        assert 1 == len(keys)
 

	
 
        response = self.app.post(url('my_account_api_keys_delete'),
 
                 {'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'API key successfully deleted')
 
        keys = UserApiKeys.query().all()
 
        assert 0 == len(keys)
 

	
 
    def test_my_account_reset_main_api_key(self):
 
        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
 

	
 
        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)