managed_fields.extend(['username', 'firstname', 'lastname', 'email'])

        c.readonly = lambda n: 'readonly' if n in managed_fields else None

        defaults = c.user.get_dict()

        update = False

        if request.POST:

            _form = UserForm(edit=True,

                             old_data={'user_id': request.authuser.user_id,

                                       'email': request.authuser.email})()

            form_result = {}

            try:

                post_data = dict(request.POST)

                post_data['new_password'] = ''

                post_data['password_confirmation'] = ''

                form_result = _form.to_python(post_data)

                # skip updating those attrs for my account

                skip_attrs = ['admin', 'active', 'extern_type', 'extern_name',

                              'new_password', 'password_confirmation',

                             ] + managed_fields

                UserModel().update(request.authuser.user_id, form_result,

                                   skip_attrs=skip_attrs)

                h.flash(_('Your account was updated successfully'),

                        category='success')

                Session().commit()

                update = True

            except formencode.Invalid as errors:

                return htmlfill.render(

                    render('admin/my_account/my_account.html'),

                    defaults=errors.value,

                    errors=errors.error_dict or {},

                    prefix_error=False,

                    encoding="UTF-8",

                    force_defaults=False)

            except Exception:

                log.error(traceback.format_exc())

                h.flash(_('Error occurred during update of user %s')

                        % form_result.get('username'), category='error')

        if update:

            raise HTTPFound(location='my_account')

        return htmlfill.render(

            render('admin/my_account/my_account.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    def my_account_password(self):

        c.active = 'password'

        self.__load_data()

        managed_fields = auth_modules.get_managed_fields(c.user)

        c.can_change_password = 'password' not in managed_fields

        if request.POST and c.can_change_password:

            _form = PasswordChangeForm(request.authuser.username)()

            try:

                form_result = _form.to_python(request.POST)

                UserModel().update(request.authuser.user_id, form_result)

                Session().commit()

                h.flash(_("Successfully updated password"), category='success')

            except formencode.Invalid as errors:

                return htmlfill.render(

                    render('admin/my_account/my_account.html'),

                    defaults=errors.value,

                    errors=errors.error_dict or {},

                    prefix_error=False,

                    encoding="UTF-8",

                    force_defaults=False)

            except Exception:

                log.error(traceback.format_exc())

                h.flash(_('Error occurred during update of user password'),

                        category='error')

        return render('admin/my_account/my_account.html')

    def my_account_repos(self):

        c.active = 'repos'

        self.__load_data()

        # data used to render the grid

        c.data = self._load_my_repos_data()

        return render('admin/my_account/my_account.html')

    def my_account_watched(self):

        c.active = 'watched'

        self.__load_data()

        # data used to render the grid

        c.data = self._load_my_repos_data(watched=True)

        return render('admin/my_account/my_account.html')

    def my_account_perms(self):

        c.active = 'perms'

        self.__load_data()

        c.perm_user = AuthUser(user_id=request.authuser.user_id)

        return render('admin/my_account/my_account.html')

    def my_account_emails(self):

        c.active = 'emails'

        self.__load_data()

        c.user_email_map = UserEmailMap.query() \

            .filter(UserEmailMap.user == c.user).all()

        return render('admin/my_account/my_account.html')

    def my_account_emails_add(self):

        email = request.POST.get('new_email')

        try:

            UserModel().add_extra_email(request.authuser.user_id, email)

            Session().commit()

            h.flash(_("Added email %s to user") % email, category='success')

        except formencode.Invalid as error:

            msg = error.error_dict['email']

            h.flash(msg, category='error')

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred during email saving'),

                    category='error')

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

    def my_account_emails_delete(self):

        email_id = request.POST.get('del_email_id')

        user_model = UserModel()

        user_model.delete_extra_email(request.authuser.user_id, email_id)

        Session().commit()

        h.flash(_("Removed email from user"), category='success')

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

    def my_account_api_keys(self):

        c.active = 'api_keys'

        self.__load_data()

        show_expired = True

        c.lifetime_values = [

            (str(-1), _('Forever')),

            (str(5), _('5 minutes')),

            (str(60), _('1 hour')),

            (str(60 * 24), _('1 day')),

            (str(60 * 24 * 30), _('1 month')),

        ]

        c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]

        c.user_api_keys = ApiKeyModel().get_api_keys(request.authuser.user_id,

                                                     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')

        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):

        try:

            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'))

    def delete_api_key(self, id):

        c.user = self._get_user_or_raise_if_default(id)

        api_key = request.POST.get('del_api_key')

        if request.POST.get('del_api_key_builtin'):

            c.user.api_key = generate_api_key()

            Session().commit()

            h.flash(_("API key successfully reset"), category='success')

        elif api_key:

            ApiKeyModel().delete(api_key, c.user.user_id)

            Session().commit()

            h.flash(_("API key successfully deleted"), category='success')

        raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))

    def update_account(self, id):

        pass

    def edit_perms(self, id):

        c.user = self._get_user_or_raise_if_default(id)

        c.active = 'perms'

        c.perm_user = AuthUser(dbuser=c.user)

        umodel = UserModel()

        defaults = c.user.get_dict()

        defaults.update({

            'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),

            'create_user_group_perm': umodel.has_perm(c.user,

                                                      'hg.usergroup.create.true'),

            'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),

        })

        return htmlfill.render(

            render('admin/users/user_edit.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    def update_perms(self, id):

        user = self._get_user_or_raise_if_default(id)

        try:

            form = CustomDefaultPermissionsForm()()

            form_result = form.to_python(request.POST)

            user_model = UserModel()

            defs = UserToPerm.query() \

                .filter(UserToPerm.user == user) \

                .all()

            for ug in defs:

                Session().delete(ug)

            if form_result['create_repo_perm']:

                user_model.grant_perm(id, 'hg.create.repository')

            else:

                user_model.grant_perm(id, 'hg.create.none')

            if form_result['create_user_group_perm']:

                user_model.grant_perm(id, 'hg.usergroup.create.true')

            else:

                user_model.grant_perm(id, 'hg.usergroup.create.false')

            if form_result['fork_repo_perm']:

                user_model.grant_perm(id, 'hg.fork.repository')

            else:

                user_model.grant_perm(id, 'hg.fork.none')

            h.flash(_("Updated permissions"), category='success')

            Session().commit()

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred during permissions saving'),

                    category='error')

        raise HTTPFound(location=url('edit_user_perms', id=id))

    def edit_emails(self, id):

        c.user = self._get_user_or_raise_if_default(id)

        c.active = 'emails'

        c.user_email_map = UserEmailMap.query() \

            .filter(UserEmailMap.user == c.user).all()

        defaults = c.user.get_dict()

        return htmlfill.render(

            render('admin/users/user_edit.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    def add_email(self, id):

        user = self._get_user_or_raise_if_default(id)

        email = request.POST.get('new_email')

        user_model = UserModel()

        try:

            user_model.add_extra_email(id, email)

            Session().commit()

            h.flash(_("Added email %s to user") % email, category='success')

        except formencode.Invalid as error:

            msg = error.error_dict['email']

            h.flash(msg, category='error')

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred during email saving'),

                    category='error')

        raise HTTPFound(location=url('edit_user_emails', id=id))

    def delete_email(self, id):

        user = self._get_user_or_raise_if_default(id)

        email_id = request.POST.get('del_email_id')

        user_model = UserModel()

        user_model.delete_extra_email(id, email_id)

        Session().commit()

        h.flash(_("Removed email from user"), category='success')

        raise HTTPFound(location=url('edit_user_emails', id=id))

    def edit_ips(self, id):

        c.user = self._get_user_or_raise_if_default(id)

        c.active = 'ips'

        c.user_ip_map = UserIpMap.query() \

            .filter(UserIpMap.user == c.user).all()

        c.default_user_ip_map = UserIpMap.query() \

            .filter(UserIpMap.user == User.get_default_user()).all()

        defaults = c.user.get_dict()

        return htmlfill.render(

            render('admin/users/user_edit.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    def add_ip(self, id):

        ip = request.POST.get('new_ip')

        user_model = UserModel()

        try:

            user_model.add_extra_ip(id, ip)

            Session().commit()

            h.flash(_("Added IP address %s to user whitelist") % ip, category='success')

        except formencode.Invalid as error:

            msg = error.error_dict['ip']

            h.flash(msg, category='error')

        except Exception:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred while adding IP address'),

                    category='error')

        if 'default_user' in request.POST:

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

        raise HTTPFound(location=url('edit_user_ips', id=id))

    def delete_ip(self, id):

        ip_id = request.POST.get('del_ip_id')

        user_model = UserModel()

        user_model.delete_extra_ip(id, ip_id)

        Session().commit()

        h.flash(_("Removed IP address from user whitelist"), category='success')

        if 'default_user' in request.POST:

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

        raise HTTPFound(location=url('edit_user_ips', id=id))

    @IfSshEnabled

    def edit_ssh_keys(self, id):

        c.user = self._get_user_or_raise_if_default(id)

        c.active = 'ssh_keys'

        c.user_ssh_keys = SshKeyModel().get_ssh_keys(c.user.user_id)

        defaults = c.user.get_dict()

        return htmlfill.render(

            render('admin/users/user_edit.html'),

            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)

        try:

            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))

# -*- 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 errno

import logging

import os

import stat

import tempfile

from tg import config

from tg.i18n import ugettext as _

from kallithea.lib import ssh

from kallithea.lib.utils2 import safe_str, str2bool

from kallithea.model.db import User, UserSshKeys

from kallithea.model.meta import Session

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:

            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

        """

        Will raise SshKeyModelException on errors

        """

        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:

        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:

            if exception.errno != errno.EEXIST:

                raise

        # Now, test that the directory is or was created in a readable way by previous.

        if not (os.path.isdir(authorized_keys_dir) and

                os.access(authorized_keys_dir, os.W_OK)):

            raise Exception("Directory of authorized_keys cannot be written to so authorized_keys file %s cannot be written" % (authorized_keys))

        # Make sure we don't overwrite a key file with important content

        if os.path.exists(authorized_keys):

            with open(authorized_keys) as f:

                for l in f:

                    if not l.strip() or l.startswith('#'):

                        pass # accept empty lines and comments

                    elif ssh.SSH_OPTIONS in l and ' ssh-serve ' in l:

                        pass # Kallithea entries are ok to overwrite

                    else:

                        raise Exception("Safety check failed, found %r in %s - please review and remove it" % (l.strip(), authorized_keys))

        fh, tmp_authorized_keys = tempfile.mkstemp('.authorized_keys', dir=os.path.dirname(authorized_keys))

        with os.fdopen(fh, 'w') as f:

            for key in UserSshKeys.query().join(UserSshKeys.user).filter(User.active == True):

                f.write(ssh.authorized_keys_line(kallithea_cli_path, config['__file__'], key))

        os.chmod(tmp_authorized_keys, stat.S_IRUSR | stat.S_IWUSR)

        # This preliminary remove is needed for Windows, not for Unix.

        # TODO In Python 3, the remove+rename sequence below should become os.replace.

        if os.path.exists(authorized_keys):

            os.remove(authorized_keys)

        os.rename(tmp_authorized_keys, authorized_keys)

<table class="table">

    %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'))}

                    <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>

<table class="table">

    %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))}

                    <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>

            <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>

        self.log_user()

        perm_none = Permission.get_by_key('hg.fork.none')

        perm_fork = Permission.get_by_key('hg.fork.repository')

        user = UserModel().create_or_update(username='dummy', password='qwe',

                                            email='dummy', firstname=u'a',

                                            lastname=u'b')

        Session().commit()

        uid = user.user_id

        try:

            # User should have None permission on creation repository

            assert UserModel().has_perm(user, perm_none) == False

            assert UserModel().has_perm(user, perm_fork) == False

            response = self.app.post(url('edit_user_perms_update', id=uid),

                                     params=dict(_session_csrf_secret_token=self.session_csrf_secret_token()))

            perm_none = Permission.get_by_key('hg.create.none')

            perm_create = Permission.get_by_key('hg.create.repository')

            # User should have None permission on creation repository

            assert UserModel().has_perm(uid, perm_none) == True

            assert UserModel().has_perm(uid, perm_create) == False

        finally:

            UserModel().delete(uid)

            Session().commit()

    def test_ips(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        response = self.app.get(url('edit_user_ips', id=user.user_id))

        response.mustcontain('All IP addresses are allowed')

    @parametrize('test_name,ip,ip_range,failure', [

        ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),

        ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),

        ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),

        ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),

        ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),

        ('127_bad_ip', 'foobar', 'foobar', True),

    ])

    def test_add_ip(self, test_name, ip, ip_range, failure, auto_clear_ip_permissions):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_ips_update', id=user_id),

                                 params=dict(new_ip=ip, _session_csrf_secret_token=self.session_csrf_secret_token()))

        if failure:

            self.checkSessionFlash(response, 'Please enter a valid IPv4 or IPv6 address')

            response = self.app.get(url('edit_user_ips', id=user_id))

            response.mustcontain(no=[ip])

            response.mustcontain(no=[ip_range])

        else:

            response = self.app.get(url('edit_user_ips', id=user_id))

            response.mustcontain(ip)

            response.mustcontain(ip_range)

    def test_delete_ip(self, auto_clear_ip_permissions):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        ip = '127.0.0.1/32'

        ip_range = '127.0.0.1 - 127.0.0.1'

        with test_context(self.app):

            new_ip = UserModel().add_extra_ip(user_id, ip)

            Session().commit()

        new_ip_id = new_ip.ip_id

        response = self.app.get(url('edit_user_ips', id=user_id))

        response.mustcontain(ip)

        response.mustcontain(ip_range)

        self.app.post(url('edit_user_ips_delete', id=user_id),

                      params=dict(del_ip_id=new_ip_id, _session_csrf_secret_token=self.session_csrf_secret_token()))

        response = self.app.get(url('edit_user_ips', id=user_id))

        response.mustcontain('All IP addresses are allowed')

        response.mustcontain(no=[ip])

        response.mustcontain(no=[ip_range])

    def test_api_keys(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        response = self.app.get(url('edit_user_api_keys', id=user.user_id))

        response.mustcontain(user.api_key)

        response.mustcontain('Expires: Never')

    @parametrize('desc,lifetime', [

        ('forever', -1),

        ('5mins', 60*5),

        ('30days', 60*60*24*30),

    ])

    def test_add_api_keys(self, desc, lifetime):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_api_keys_update', id=user_id),

                 {'description': desc, 'lifetime': lifetime, '_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        try:

            response = response.follow()

            user = User.get(user_id)

            for api_key in user.api_keys:

                response.mustcontain(api_key)

        finally:

            for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():

                Session().delete(api_key)

                Session().commit()

    def test_remove_api_key(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        response = self.app.post(url('edit_user_api_keys_update', id=user_id),

                {'description': 'desc', 'lifetime': -1, '_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        response = response.follow()

        # now delete our key

        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

        assert 1 == len(keys)

        response = self.app.post(url('edit_user_api_keys_delete', id=user_id),

                 {'del_api_key': keys[0].api_key, '_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'API key successfully deleted')

        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()

        assert 0 == len(keys)

    def test_reset_main_api_key(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        user_id = user.user_id

        api_key = user.api_key

        response = self.app.get(url('edit_user_api_keys', id=user_id))

        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, '_session_csrf_secret_token': self.session_csrf_secret_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'