users-admin: moved views into pyramid for editing emails and ips.
- added audit-logging - added few additional tests
This commit is contained in:
parent
37c0465603
commit
effa0f53ce
13 changed files with 426 additions and 256 deletions
|
|
@ -64,6 +64,11 @@ def admin_routes(config):
|
|||
name='admin_settings_sessions_cleanup',
|
||||
pattern='/settings/sessions/cleanup')
|
||||
|
||||
# global permissions
|
||||
config.add_route(
|
||||
name='admin_permissions_ips',
|
||||
pattern='/permissions/ips')
|
||||
|
||||
# users admin
|
||||
config.add_route(
|
||||
name='users',
|
||||
|
|
@ -84,6 +89,28 @@ def admin_routes(config):
|
|||
name='edit_user_auth_tokens_delete',
|
||||
pattern='/users/{user_id:\d+}/edit/auth_tokens/delete')
|
||||
|
||||
# user emails
|
||||
config.add_route(
|
||||
name='edit_user_emails',
|
||||
pattern='/users/{user_id:\d+}/edit/emails')
|
||||
config.add_route(
|
||||
name='edit_user_emails_add',
|
||||
pattern='/users/{user_id:\d+}/edit/emails/new')
|
||||
config.add_route(
|
||||
name='edit_user_emails_delete',
|
||||
pattern='/users/{user_id:\d+}/edit/emails/delete')
|
||||
|
||||
# user IPs
|
||||
config.add_route(
|
||||
name='edit_user_ips',
|
||||
pattern='/users/{user_id:\d+}/edit/ips')
|
||||
config.add_route(
|
||||
name='edit_user_ips_add',
|
||||
pattern='/users/{user_id:\d+}/edit/ips/new')
|
||||
config.add_route(
|
||||
name='edit_user_ips_delete',
|
||||
pattern='/users/{user_id:\d+}/edit/ips/delete')
|
||||
|
||||
# user groups management
|
||||
config.add_route(
|
||||
name='edit_user_groups_management',
|
||||
|
|
|
|||
|
|
@ -20,7 +20,9 @@
|
|||
|
||||
import pytest
|
||||
|
||||
from rhodecode.model.db import User, UserApiKeys
|
||||
from rhodecode.model.db import User, UserApiKeys, UserEmailMap
|
||||
from rhodecode.model.meta import Session
|
||||
from rhodecode.model.user import UserModel
|
||||
|
||||
from rhodecode.tests import (
|
||||
TestController, TEST_USER_REGULAR_LOGIN, assert_session_flash)
|
||||
|
|
@ -44,6 +46,20 @@ def route_path(name, params=None, **kwargs):
|
|||
ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens/new',
|
||||
'edit_user_auth_tokens_delete':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/auth_tokens/delete',
|
||||
|
||||
'edit_user_emails':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/emails',
|
||||
'edit_user_emails_add':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/emails/new',
|
||||
'edit_user_emails_delete':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/emails/delete',
|
||||
|
||||
'edit_user_ips':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips',
|
||||
'edit_user_ips_add':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
|
||||
'edit_user_ips_delete':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
|
||||
}[name].format(**kwargs)
|
||||
|
||||
if params:
|
||||
|
|
@ -135,8 +151,131 @@ class TestAdminUsersView(TestController):
|
|||
|
||||
response = self.app.post(
|
||||
route_path('edit_user_auth_tokens_delete', user_id=user_id),
|
||||
{'del_auth_token': keys[0].api_key, 'csrf_token': self.csrf_token})
|
||||
{'del_auth_token': keys[0].user_api_key_id,
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
assert_session_flash(response, 'Auth token successfully deleted')
|
||||
keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
|
||||
assert 2 == len(keys)
|
||||
|
||||
def test_ips(self):
|
||||
self.log_user()
|
||||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
response = self.app.get(route_path('edit_user_ips', user_id=user.user_id))
|
||||
response.mustcontain('All IP addresses are allowed')
|
||||
|
||||
@pytest.mark.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_ips_add(self, user_util, test_name, ip, ip_range, failure):
|
||||
self.log_user()
|
||||
user = user_util.create_user(username=test_name)
|
||||
user_id = user.user_id
|
||||
|
||||
response = self.app.post(
|
||||
route_path('edit_user_ips_add', user_id=user_id),
|
||||
params={'new_ip': ip, 'csrf_token': self.csrf_token})
|
||||
|
||||
if failure:
|
||||
assert_session_flash(
|
||||
response, 'Please enter a valid IPv4 or IpV6 address')
|
||||
response = self.app.get(route_path('edit_user_ips', user_id=user_id))
|
||||
|
||||
response.mustcontain(no=[ip])
|
||||
response.mustcontain(no=[ip_range])
|
||||
|
||||
else:
|
||||
response = self.app.get(route_path('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain(ip)
|
||||
response.mustcontain(ip_range)
|
||||
|
||||
def test_ips_delete(self, user_util):
|
||||
self.log_user()
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
ip = '127.0.0.1/32'
|
||||
ip_range = '127.0.0.1 - 127.0.0.1'
|
||||
new_ip = UserModel().add_extra_ip(user_id, ip)
|
||||
Session().commit()
|
||||
new_ip_id = new_ip.ip_id
|
||||
|
||||
response = self.app.get(route_path('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain(ip)
|
||||
response.mustcontain(ip_range)
|
||||
|
||||
self.app.post(
|
||||
route_path('edit_user_ips_delete', user_id=user_id),
|
||||
params={'del_ip_id': new_ip_id, 'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(route_path('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain('All IP addresses are allowed')
|
||||
response.mustcontain(no=[ip])
|
||||
response.mustcontain(no=[ip_range])
|
||||
|
||||
def test_emails(self):
|
||||
self.log_user()
|
||||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
response = self.app.get(route_path('edit_user_emails', user_id=user.user_id))
|
||||
response.mustcontain('No additional emails specified')
|
||||
|
||||
def test_emails_add(self, user_util):
|
||||
self.log_user()
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
|
||||
self.app.post(
|
||||
route_path('edit_user_emails_add', user_id=user_id),
|
||||
params={'new_email': 'example@rhodecode.com',
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(route_path('edit_user_emails', user_id=user_id))
|
||||
response.mustcontain('example@rhodecode.com')
|
||||
|
||||
def test_emails_add_existing_email(self, user_util, user_regular):
|
||||
existing_email = user_regular.email
|
||||
|
||||
self.log_user()
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
|
||||
response = self.app.post(
|
||||
route_path('edit_user_emails_add', user_id=user_id),
|
||||
params={'new_email': existing_email,
|
||||
'csrf_token': self.csrf_token})
|
||||
assert_session_flash(
|
||||
response, 'This e-mail address is already taken')
|
||||
|
||||
response = self.app.get(route_path('edit_user_emails', user_id=user_id))
|
||||
response.mustcontain(no=[existing_email])
|
||||
|
||||
def test_emails_delete(self, user_util):
|
||||
self.log_user()
|
||||
user = user_util.create_user()
|
||||
user_id = user.user_id
|
||||
|
||||
self.app.post(
|
||||
route_path('edit_user_emails_add', user_id=user_id),
|
||||
params={'new_email': 'example@rhodecode.com',
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(route_path('edit_user_emails', user_id=user_id))
|
||||
response.mustcontain('example@rhodecode.com')
|
||||
|
||||
user_email = UserEmailMap.query()\
|
||||
.filter(UserEmailMap.email == 'example@rhodecode.com') \
|
||||
.filter(UserEmailMap.user_id == user_id)\
|
||||
.one()
|
||||
|
||||
del_email_id = user_email.email_id
|
||||
self.app.post(
|
||||
route_path('edit_user_emails_delete', user_id=user_id),
|
||||
params={'del_email_id': del_email_id,
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(route_path('edit_user_emails', user_id=user_id))
|
||||
response.mustcontain(no=['example@rhodecode.com'])
|
||||
|
|
@ -20,15 +20,16 @@
|
|||
|
||||
import logging
|
||||
import datetime
|
||||
import formencode
|
||||
|
||||
from pyramid.httpexceptions import HTTPFound
|
||||
from pyramid.view import view_config
|
||||
from sqlalchemy.sql.functions import coalesce
|
||||
|
||||
from rhodecode.lib.helpers import Page
|
||||
from rhodecode.lib.ext_json import json
|
||||
|
||||
from rhodecode.apps._base import BaseAppView, DataGridAppView
|
||||
|
||||
from rhodecode.lib import audit_logger
|
||||
from rhodecode.lib.ext_json import json
|
||||
from rhodecode.lib.auth import (
|
||||
LoginRequired, HasPermissionAllDecorator, CSRFRequired)
|
||||
from rhodecode.lib import helpers as h
|
||||
|
|
@ -37,7 +38,7 @@ from rhodecode.lib.utils2 import safe_int, safe_unicode
|
|||
from rhodecode.model.auth_token import AuthTokenModel
|
||||
from rhodecode.model.user import UserModel
|
||||
from rhodecode.model.user_group import UserGroupModel
|
||||
from rhodecode.model.db import User, or_
|
||||
from rhodecode.model.db import User, or_, UserIpMap, UserEmailMap, UserApiKeys
|
||||
from rhodecode.model.meta import Session
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
|
@ -194,15 +195,23 @@ class AdminUsersView(BaseAppView, DataGridAppView):
|
|||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
|
||||
user_data = c.user.get_api_data()
|
||||
lifetime = safe_int(self.request.POST.get('lifetime'), -1)
|
||||
description = self.request.POST.get('description')
|
||||
role = self.request.POST.get('role')
|
||||
|
||||
token = AuthTokenModel().create(
|
||||
c.user.user_id, description, lifetime, role)
|
||||
token_data = token.get_api_data()
|
||||
|
||||
self.maybe_attach_token_scope(token)
|
||||
audit_logger.store(
|
||||
action='user.edit.token.add',
|
||||
action_data={'data': {'token': token_data, 'user': user_data}},
|
||||
user=self._rhodecode_user, )
|
||||
Session().commit()
|
||||
|
||||
h.flash(_("Auth token successfully created"), category='success')
|
||||
|
|
@ -220,16 +229,214 @@ class AdminUsersView(BaseAppView, DataGridAppView):
|
|||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
user_data = c.user.get_api_data()
|
||||
|
||||
del_auth_token = self.request.POST.get('del_auth_token')
|
||||
|
||||
if del_auth_token:
|
||||
token = UserApiKeys.get_or_404(del_auth_token, pyramid_exc=True)
|
||||
token_data = token.get_api_data()
|
||||
|
||||
AuthTokenModel().delete(del_auth_token, c.user.user_id)
|
||||
audit_logger.store(
|
||||
action='user.edit.token.delete',
|
||||
action_data={'data': {'token': token_data, 'user': user_data}},
|
||||
user=self._rhodecode_user,)
|
||||
Session().commit()
|
||||
h.flash(_("Auth token successfully deleted"), category='success')
|
||||
|
||||
return HTTPFound(h.route_path('edit_user_auth_tokens', user_id=user_id))
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@view_config(
|
||||
route_name='edit_user_emails', request_method='GET',
|
||||
renderer='rhodecode:templates/admin/users/user_edit.mako')
|
||||
def emails(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
|
||||
c.active = 'emails'
|
||||
c.user_email_map = UserEmailMap.query() \
|
||||
.filter(UserEmailMap.user == c.user).all()
|
||||
|
||||
return self._get_template_context(c)
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@CSRFRequired()
|
||||
@view_config(
|
||||
route_name='edit_user_emails_add', request_method='POST')
|
||||
def emails_add(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
|
||||
email = self.request.POST.get('new_email')
|
||||
user_data = c.user.get_api_data()
|
||||
try:
|
||||
UserModel().add_extra_email(c.user.user_id, email)
|
||||
audit_logger.store_web(
|
||||
'user.edit.email.add',
|
||||
action_data={'email': email, 'user': user_data},
|
||||
user=self._rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Added new email address `%s` for user account") % email,
|
||||
category='success')
|
||||
except formencode.Invalid as error:
|
||||
msg = error.error_dict['email']
|
||||
h.flash(msg, category='error')
|
||||
except Exception:
|
||||
log.exception("Exception during email saving")
|
||||
h.flash(_('An error occurred during email saving'),
|
||||
category='error')
|
||||
raise HTTPFound(h.route_path('edit_user_emails', user_id=user_id))
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@CSRFRequired()
|
||||
@view_config(
|
||||
route_name='edit_user_emails_delete', request_method='POST')
|
||||
def emails_delete(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
|
||||
email_id = self.request.POST.get('del_email_id')
|
||||
user_model = UserModel()
|
||||
|
||||
email = UserEmailMap.query().get(email_id).email
|
||||
user_data = c.user.get_api_data()
|
||||
user_model.delete_extra_email(c.user.user_id, email_id)
|
||||
audit_logger.store_web(
|
||||
'user.edit.email.delete',
|
||||
action_data={'email': email, 'user': user_data},
|
||||
user=self._rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Removed email address from user account"),
|
||||
category='success')
|
||||
raise HTTPFound(h.route_path('edit_user_emails', user_id=user_id))
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@view_config(
|
||||
route_name='edit_user_ips', request_method='GET',
|
||||
renderer='rhodecode:templates/admin/users/user_edit.mako')
|
||||
def ips(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
self._redirect_for_default_user(c.user.username)
|
||||
|
||||
c.active = 'ips'
|
||||
c.user_ip_map = UserIpMap.query() \
|
||||
.filter(UserIpMap.user == c.user).all()
|
||||
|
||||
c.inherit_default_ips = c.user.inherit_default_permissions
|
||||
c.default_user_ip_map = UserIpMap.query() \
|
||||
.filter(UserIpMap.user == User.get_default_user()).all()
|
||||
|
||||
return self._get_template_context(c)
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@CSRFRequired()
|
||||
@view_config(
|
||||
route_name='edit_user_ips_add', request_method='POST')
|
||||
def ips_add(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
# NOTE(marcink): this view is allowed for default users, as we can
|
||||
# edit their IP white list
|
||||
|
||||
user_model = UserModel()
|
||||
desc = self.request.POST.get('description')
|
||||
try:
|
||||
ip_list = user_model.parse_ip_range(
|
||||
self.request.POST.get('new_ip'))
|
||||
except Exception as e:
|
||||
ip_list = []
|
||||
log.exception("Exception during ip saving")
|
||||
h.flash(_('An error occurred during ip saving:%s' % (e,)),
|
||||
category='error')
|
||||
added = []
|
||||
user_data = c.user.get_api_data()
|
||||
for ip in ip_list:
|
||||
try:
|
||||
user_model.add_extra_ip(c.user.user_id, ip, desc)
|
||||
audit_logger.store_web(
|
||||
'user.edit.ip.add',
|
||||
action_data={'ip': ip, 'user': user_data},
|
||||
user=self._rhodecode_user)
|
||||
Session().commit()
|
||||
added.append(ip)
|
||||
except formencode.Invalid as error:
|
||||
msg = error.error_dict['ip']
|
||||
h.flash(msg, category='error')
|
||||
except Exception:
|
||||
log.exception("Exception during ip saving")
|
||||
h.flash(_('An error occurred during ip saving'),
|
||||
category='error')
|
||||
if added:
|
||||
h.flash(
|
||||
_("Added ips %s to user whitelist") % (', '.join(ip_list), ),
|
||||
category='success')
|
||||
if 'default_user' in self.request.POST:
|
||||
# case for editing global IP list we do it for 'DEFAULT' user
|
||||
raise HTTPFound(h.route_path('admin_permissions_ips'))
|
||||
raise HTTPFound(h.route_path('edit_user_ips', user_id=user_id))
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@CSRFRequired()
|
||||
@view_config(
|
||||
route_name='edit_user_ips_delete', request_method='POST')
|
||||
def ips_delete(self):
|
||||
_ = self.request.translate
|
||||
c = self.load_default_context()
|
||||
|
||||
user_id = self.request.matchdict.get('user_id')
|
||||
c.user = User.get_or_404(user_id, pyramid_exc=True)
|
||||
# NOTE(marcink): this view is allowed for default users, as we can
|
||||
# edit their IP white list
|
||||
|
||||
ip_id = self.request.POST.get('del_ip_id')
|
||||
user_model = UserModel()
|
||||
user_data = c.user.get_api_data()
|
||||
ip = UserIpMap.query().get(ip_id).ip_addr
|
||||
user_model.delete_extra_ip(c.user.user_id, ip_id)
|
||||
audit_logger.store_web(
|
||||
'user.edit.ip.delete',
|
||||
action_data={'ip': ip, 'user': user_data},
|
||||
user=self._rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Removed ip address from user whitelist"), category='success')
|
||||
|
||||
if 'default_user' in self.request.POST:
|
||||
# case for editing global IP list we do it for 'DEFAULT' user
|
||||
raise HTTPFound(h.route_path('admin_permissions_ips'))
|
||||
raise HTTPFound(h.route_path('edit_user_ips', user_id=user_id))
|
||||
|
||||
@LoginRequired()
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@view_config(
|
||||
|
|
@ -301,8 +508,8 @@ class AdminUsersView(BaseAppView, DataGridAppView):
|
|||
kw['filter'] = filter_term
|
||||
return self.request.current_route_path(_query=kw)
|
||||
|
||||
c.audit_logs = Page(user_log, page=p, items_per_page=10,
|
||||
url=url_generator)
|
||||
c.audit_logs = h.Page(
|
||||
user_log, page=p, items_per_page=10, url=url_generator)
|
||||
c.filter_term = filter_term
|
||||
return self._get_template_context(c)
|
||||
|
||||
|
|
|
|||
|
|
@ -287,19 +287,6 @@ def make_map(config):
|
|||
m.connect('edit_user_perms_summary', '/users/{user_id}/edit/permissions_summary',
|
||||
action='edit_perms_summary', conditions={'method': ['GET']})
|
||||
|
||||
m.connect('edit_user_emails', '/users/{user_id}/edit/emails',
|
||||
action='edit_emails', conditions={'method': ['GET']})
|
||||
m.connect('edit_user_emails', '/users/{user_id}/edit/emails',
|
||||
action='add_email', conditions={'method': ['PUT']})
|
||||
m.connect('edit_user_emails', '/users/{user_id}/edit/emails',
|
||||
action='delete_email', conditions={'method': ['DELETE']})
|
||||
|
||||
m.connect('edit_user_ips', '/users/{user_id}/edit/ips',
|
||||
action='edit_ips', conditions={'method': ['GET']})
|
||||
m.connect('edit_user_ips', '/users/{user_id}/edit/ips',
|
||||
action='add_ip', conditions={'method': ['PUT']})
|
||||
m.connect('edit_user_ips', '/users/{user_id}/edit/ips',
|
||||
action='delete_ip', conditions={'method': ['DELETE']})
|
||||
|
||||
# ADMIN USER GROUPS REST ROUTES
|
||||
with rmap.submapper(path_prefix=ADMIN_PREFIX,
|
||||
|
|
|
|||
|
|
@ -491,153 +491,3 @@ class UsersController(BaseController):
|
|||
|
||||
return render('admin/users/user_edit.mako')
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
def edit_emails(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
if c.user.username == User.DEFAULT_USER:
|
||||
h.flash(_("You can't edit this user"), category='warning')
|
||||
return redirect(h.route_path('users'))
|
||||
|
||||
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.mako'),
|
||||
defaults=defaults,
|
||||
encoding="UTF-8",
|
||||
force_defaults=False)
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@auth.CSRFRequired()
|
||||
def add_email(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
|
||||
email = request.POST.get('new_email')
|
||||
user_model = UserModel()
|
||||
user_data = c.user.get_api_data()
|
||||
try:
|
||||
user_model.add_extra_email(user_id, email)
|
||||
audit_logger.store_web(
|
||||
'user.edit.email.add',
|
||||
action_data={'email': email, 'user': user_data},
|
||||
user=c.rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Added new email address `%s` for user account") % email,
|
||||
category='success')
|
||||
except formencode.Invalid as error:
|
||||
msg = error.error_dict['email']
|
||||
h.flash(msg, category='error')
|
||||
except Exception:
|
||||
log.exception("Exception during email saving")
|
||||
h.flash(_('An error occurred during email saving'),
|
||||
category='error')
|
||||
return redirect(url('edit_user_emails', user_id=user_id))
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@auth.CSRFRequired()
|
||||
def delete_email(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
email_id = request.POST.get('del_email_id')
|
||||
user_model = UserModel()
|
||||
|
||||
email = UserEmailMap.query().get(email_id).email
|
||||
user_data = c.user.get_api_data()
|
||||
user_model.delete_extra_email(user_id, email_id)
|
||||
audit_logger.store_web(
|
||||
'user.edit.email.delete',
|
||||
action_data={'email': email, 'user': user_data},
|
||||
user=c.rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Removed email address from user account"), category='success')
|
||||
return redirect(url('edit_user_emails', user_id=user_id))
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
def edit_ips(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
if c.user.username == User.DEFAULT_USER:
|
||||
h.flash(_("You can't edit this user"), category='warning')
|
||||
return redirect(h.route_path('users'))
|
||||
|
||||
c.active = 'ips'
|
||||
c.user_ip_map = UserIpMap.query() \
|
||||
.filter(UserIpMap.user == c.user).all()
|
||||
|
||||
c.inherit_default_ips = c.user.inherit_default_permissions
|
||||
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.mako'),
|
||||
defaults=defaults,
|
||||
encoding="UTF-8",
|
||||
force_defaults=False)
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@auth.CSRFRequired()
|
||||
def add_ip(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
user_model = UserModel()
|
||||
try:
|
||||
ip_list = user_model.parse_ip_range(request.POST.get('new_ip'))
|
||||
except Exception as e:
|
||||
ip_list = []
|
||||
log.exception("Exception during ip saving")
|
||||
h.flash(_('An error occurred during ip saving:%s' % (e,)),
|
||||
category='error')
|
||||
|
||||
desc = request.POST.get('description')
|
||||
added = []
|
||||
user_data = c.user.get_api_data()
|
||||
for ip in ip_list:
|
||||
try:
|
||||
user_model.add_extra_ip(user_id, ip, desc)
|
||||
audit_logger.store_web(
|
||||
'user.edit.ip.add',
|
||||
action_data={'ip': ip, 'user': user_data},
|
||||
user=c.rhodecode_user)
|
||||
Session().commit()
|
||||
added.append(ip)
|
||||
except formencode.Invalid as error:
|
||||
msg = error.error_dict['ip']
|
||||
h.flash(msg, category='error')
|
||||
except Exception:
|
||||
log.exception("Exception during ip saving")
|
||||
h.flash(_('An error occurred during ip saving'),
|
||||
category='error')
|
||||
if added:
|
||||
h.flash(
|
||||
_("Added ips %s to user whitelist") % (', '.join(ip_list), ),
|
||||
category='success')
|
||||
if 'default_user' in request.POST:
|
||||
return redirect(url('admin_permissions_ips'))
|
||||
return redirect(url('edit_user_ips', user_id=user_id))
|
||||
|
||||
@HasPermissionAllDecorator('hg.admin')
|
||||
@auth.CSRFRequired()
|
||||
def delete_ip(self, user_id):
|
||||
user_id = safe_int(user_id)
|
||||
c.user = User.get_or_404(user_id)
|
||||
|
||||
ip_id = request.POST.get('del_ip_id')
|
||||
user_model = UserModel()
|
||||
user_data = c.user.get_api_data()
|
||||
ip = UserIpMap.query().get(ip_id).ip_addr
|
||||
user_model.delete_extra_ip(user_id, ip_id)
|
||||
audit_logger.store_web(
|
||||
'user.edit.ip.delete',
|
||||
action_data={'ip': ip, 'user': user_data},
|
||||
user=c.rhodecode_user)
|
||||
Session().commit()
|
||||
h.flash(_("Removed ip address from user whitelist"), category='success')
|
||||
|
||||
if 'default_user' in request.POST:
|
||||
return redirect(url('admin_permissions_ips'))
|
||||
return redirect(url('edit_user_ips', user_id=user_id))
|
||||
|
|
|
|||
|
|
@ -73,11 +73,18 @@ function registerRCRoutes() {
|
|||
pyroutes.register('admin_settings_system_update', '_admin/settings/system/updates', []);
|
||||
pyroutes.register('admin_settings_sessions', '_admin/settings/sessions', []);
|
||||
pyroutes.register('admin_settings_sessions_cleanup', '_admin/settings/sessions/cleanup', []);
|
||||
pyroutes.register('admin_permissions_ips', '_admin/permissions/ips', []);
|
||||
pyroutes.register('users', '_admin/users', []);
|
||||
pyroutes.register('users_data', '_admin/users_data', []);
|
||||
pyroutes.register('edit_user_auth_tokens', '_admin/users/%(user_id)s/edit/auth_tokens', ['user_id']);
|
||||
pyroutes.register('edit_user_auth_tokens_add', '_admin/users/%(user_id)s/edit/auth_tokens/new', ['user_id']);
|
||||
pyroutes.register('edit_user_auth_tokens_delete', '_admin/users/%(user_id)s/edit/auth_tokens/delete', ['user_id']);
|
||||
pyroutes.register('edit_user_emails', '_admin/users/%(user_id)s/edit/emails', ['user_id']);
|
||||
pyroutes.register('edit_user_emails_add', '_admin/users/%(user_id)s/edit/emails/new', ['user_id']);
|
||||
pyroutes.register('edit_user_emails_delete', '_admin/users/%(user_id)s/edit/emails/delete', ['user_id']);
|
||||
pyroutes.register('edit_user_ips', '_admin/users/%(user_id)s/edit/ips', ['user_id']);
|
||||
pyroutes.register('edit_user_ips_add', '_admin/users/%(user_id)s/edit/ips/new', ['user_id']);
|
||||
pyroutes.register('edit_user_ips_delete', '_admin/users/%(user_id)s/edit/ips/delete', ['user_id']);
|
||||
pyroutes.register('edit_user_groups_management', '_admin/users/%(user_id)s/edit/groups_management', ['user_id']);
|
||||
pyroutes.register('edit_user_groups_management_updates', '_admin/users/%(user_id)s/edit/edit_user_groups_management/updates', ['user_id']);
|
||||
pyroutes.register('edit_user_audit_logs', '_admin/users/%(user_id)s/edit/audit', ['user_id']);
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@
|
|||
<td class="td-iprange"><div class="ip">${h.ip_range(ip.ip_addr)}</div></td>
|
||||
<td class="td-description"><div class="ip">${ip.description}</div></td>
|
||||
<td class="td-action">
|
||||
${h.secure_form(url('edit_user_ips', user_id=c.user.user_id),method='delete')}
|
||||
${h.secure_form(h.route_path('edit_user_ips_delete', user_id=c.user.user_id), method='POST')}
|
||||
${h.hidden('del_ip_id',ip.ip_id)}
|
||||
${h.hidden('default_user', 'True')}
|
||||
${h.submit('remove_',_('Delete'),id="remove_ip_%s" % ip.ip_id,
|
||||
|
|
@ -40,7 +40,7 @@
|
|||
</table>
|
||||
</div>
|
||||
|
||||
${h.secure_form(url('edit_user_ips', user_id=c.user.user_id),method='put')}
|
||||
${h.secure_form(h.route_path('edit_user_ips_add', user_id=c.user.user_id), method='POST')}
|
||||
<div class="form">
|
||||
<!-- fields -->
|
||||
<div class="fields">
|
||||
|
|
|
|||
|
|
@ -40,8 +40,8 @@
|
|||
<li class="${'active' if c.active=='advanced' else ''}"><a href="${h.url('edit_user_advanced', user_id=c.user.user_id)}">${_('Advanced')}</a></li>
|
||||
<li class="${'active' if c.active=='global_perms' else ''}"><a href="${h.url('edit_user_global_perms', user_id=c.user.user_id)}">${_('Global permissions')}</a></li>
|
||||
<li class="${'active' if c.active=='perms_summary' else ''}"><a href="${h.url('edit_user_perms_summary', user_id=c.user.user_id)}">${_('Permissions summary')}</a></li>
|
||||
<li class="${'active' if c.active=='emails' else ''}"><a href="${h.url('edit_user_emails', user_id=c.user.user_id)}">${_('Emails')}</a></li>
|
||||
<li class="${'active' if c.active=='ips' else ''}"><a href="${h.url('edit_user_ips', user_id=c.user.user_id)}">${_('Ip Whitelist')}</a></li>
|
||||
<li class="${'active' if c.active=='emails' else ''}"><a href="${h.route_path('edit_user_emails', user_id=c.user.user_id)}">${_('Emails')}</a></li>
|
||||
<li class="${'active' if c.active=='ips' else ''}"><a href="${h.route_path('edit_user_ips', user_id=c.user.user_id)}">${_('Ip Whitelist')}</a></li>
|
||||
<li class="${'active' if c.active=='groups' else ''}"><a href="${h.route_path('edit_user_groups_management', user_id=c.user.user_id)}">${_('User Groups Management')}</a></li>
|
||||
<li class="${'active' if c.active=='audit' else ''}"><a href="${h.route_path('edit_user_audit_logs', user_id=c.user.user_id)}">${_('User audit')}</a></li>
|
||||
</ul>
|
||||
|
|
|
|||
|
|
@ -38,10 +38,10 @@
|
|||
%endif
|
||||
</td>
|
||||
<td class="td-action">
|
||||
${h.secure_form(h.route_path('edit_user_auth_tokens_delete', user_id=c.user.user_id), method='post')}
|
||||
${h.hidden('del_auth_token',auth_token.api_key)}
|
||||
${h.secure_form(h.route_path('edit_user_auth_tokens_delete', user_id=c.user.user_id), method='POST')}
|
||||
${h.hidden('del_auth_token', auth_token.user_api_key_id)}
|
||||
<button class="btn btn-link btn-danger" type="submit"
|
||||
onclick="return confirm('${_('Confirm to remove this auth token: %s') % auth_token.api_key}');">
|
||||
onclick="return confirm('${_('Confirm to remove this auth token: %s') % auth_token.token_obfuscated}');">
|
||||
${_('Delete')}
|
||||
</button>
|
||||
${h.end_form()}
|
||||
|
|
@ -55,7 +55,7 @@
|
|||
</div>
|
||||
|
||||
<div class="user_auth_tokens">
|
||||
${h.secure_form(h.route_path('edit_user_auth_tokens_add', user_id=c.user.user_id), method='post')}
|
||||
${h.secure_form(h.route_path('edit_user_auth_tokens_add', user_id=c.user.user_id), method='POST')}
|
||||
<div class="form form-vertical">
|
||||
<!-- fields -->
|
||||
<div class="fields">
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@
|
|||
<span class="user email">${em.email}</span>
|
||||
</td>
|
||||
<td class="td-action">
|
||||
${h.secure_form(url('edit_user_emails', user_id=c.user.user_id),method='delete')}
|
||||
${h.hidden('del_email_id',em.email_id)}
|
||||
${h.secure_form(h.route_path('edit_user_emails_delete', user_id=c.user.user_id), method='POST')}
|
||||
${h.hidden('del_email_id', em.email_id)}
|
||||
<button class="btn btn-link btn-danger" type="submit"
|
||||
onclick="return confirm('${_('Confirm to delete this email: %s') % em.email}');">
|
||||
${_('Delete')}
|
||||
|
|
@ -46,7 +46,7 @@
|
|||
</table>
|
||||
</div>
|
||||
|
||||
${h.secure_form(url('edit_user_emails', user_id=c.user.user_id),method='put')}
|
||||
${h.secure_form(h.route_path('edit_user_emails_add', user_id=c.user.user_id), method='POST')}
|
||||
<div class="form">
|
||||
<!-- fields -->
|
||||
<div class="fields">
|
||||
|
|
|
|||
|
|
@ -30,9 +30,9 @@
|
|||
<td class="td-iprange"><div class="ip">${h.ip_range(ip.ip_addr)}</div></td>
|
||||
<td class="td-description"><div class="ip">${ip.description}</div></td>
|
||||
<td class="td-action">
|
||||
${h.secure_form(url('edit_user_ips', user_id=c.user.user_id),method='delete')}
|
||||
${h.hidden('del_ip_id',ip.ip_id)}
|
||||
${h.submit('remove_',_('Delete'),id="remove_ip_%s" % ip.ip_id,
|
||||
${h.secure_form(h.route_path('edit_user_ips_delete', user_id=c.user.user_id), method='POST')}
|
||||
${h.hidden('del_ip_id', ip.ip_id)}
|
||||
${h.submit('remove_', _('Delete'),id="remove_ip_%s" % ip.ip_id,
|
||||
class_="btn btn-link btn-danger", onclick="return confirm('"+_('Confirm to delete this ip: %s') % ip.ip_addr+"');")}
|
||||
${h.end_form()}
|
||||
</td>
|
||||
|
|
@ -51,7 +51,7 @@
|
|||
</div>
|
||||
|
||||
<div>
|
||||
${h.secure_form(url('edit_user_ips', user_id=c.user.user_id),method='put')}
|
||||
${h.secure_form(h.route_path('edit_user_ips_add', user_id=c.user.user_id), method='POST')}
|
||||
<div class="form">
|
||||
<!-- fields -->
|
||||
<div class="fields">
|
||||
|
|
|
|||
|
|
@ -25,6 +25,24 @@ from rhodecode.tests import (
|
|||
TestController, url, clear_all_caches, assert_session_flash)
|
||||
|
||||
|
||||
def route_path(name, params=None, **kwargs):
|
||||
import urllib
|
||||
from rhodecode.apps._base import ADMIN_PREFIX
|
||||
|
||||
base_url = {
|
||||
'edit_user_ips':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips',
|
||||
'edit_user_ips_add':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips/new',
|
||||
'edit_user_ips_delete':
|
||||
ADMIN_PREFIX + '/users/{user_id}/edit/ips/delete',
|
||||
}[name].format(**kwargs)
|
||||
|
||||
if params:
|
||||
base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
|
||||
return base_url
|
||||
|
||||
|
||||
class TestAdminPermissionsController(TestController):
|
||||
|
||||
@pytest.fixture(scope='class', autouse=True)
|
||||
|
|
@ -181,10 +199,9 @@ class TestAdminPermissionsController(TestController):
|
|||
|
||||
# ADD
|
||||
default_user_id = User.get_default_user().user_id
|
||||
response = self.app.post(
|
||||
url('edit_user_ips', user_id=default_user_id),
|
||||
params={'new_ip': '127.0.0.0/24', '_method': 'put',
|
||||
'csrf_token': self.csrf_token})
|
||||
self.app.post(
|
||||
route_path('edit_user_ips_add', user_id=default_user_id),
|
||||
params={'new_ip': '127.0.0.0/24', 'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(url('admin_permissions_ips'))
|
||||
response.mustcontain('127.0.0.0/24')
|
||||
|
|
@ -196,9 +213,8 @@ class TestAdminPermissionsController(TestController):
|
|||
default_user_id).first().ip_id
|
||||
|
||||
response = self.app.post(
|
||||
url('edit_user_ips', user_id=default_user_id),
|
||||
params={'_method': 'delete', 'del_ip_id': del_ip_id,
|
||||
'csrf_token': self.csrf_token})
|
||||
route_path('edit_user_ips_delete', user_id=default_user_id),
|
||||
params={'del_ip_id': del_ip_id, 'csrf_token': self.csrf_token})
|
||||
|
||||
assert_session_flash(response, 'Removed ip address from user whitelist')
|
||||
|
||||
|
|
|
|||
|
|
@ -510,66 +510,3 @@ class TestAdminUsersController(TestController):
|
|||
element = assert_response.get_element(css_selector)
|
||||
assert element.value == default_permissions[permission]
|
||||
|
||||
def test_ips(self):
|
||||
self.log_user()
|
||||
user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
|
||||
response = self.app.get(url('edit_user_ips', user_id=user.user_id))
|
||||
response.mustcontain('All IP addresses are allowed')
|
||||
|
||||
@pytest.mark.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):
|
||||
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', user_id=user_id),
|
||||
params={'new_ip': ip, '_method': 'put',
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
if failure:
|
||||
assert_session_flash(
|
||||
response, 'Please enter a valid IPv4 or IpV6 address')
|
||||
response = self.app.get(url('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain(no=[ip])
|
||||
response.mustcontain(no=[ip_range])
|
||||
|
||||
else:
|
||||
response = self.app.get(url('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain(ip)
|
||||
response.mustcontain(ip_range)
|
||||
|
||||
# cleanup
|
||||
for del_ip in UserIpMap.query().filter(
|
||||
UserIpMap.user_id == user_id).all():
|
||||
Session().delete(del_ip)
|
||||
Session().commit()
|
||||
|
||||
def test_delete_ip(self):
|
||||
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'
|
||||
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', user_id=user_id))
|
||||
response.mustcontain(ip)
|
||||
response.mustcontain(ip_range)
|
||||
|
||||
self.app.post(url('edit_user_ips', user_id=user_id),
|
||||
params={'_method': 'delete', 'del_ip_id': new_ip_id,
|
||||
'csrf_token': self.csrf_token})
|
||||
|
||||
response = self.app.get(url('edit_user_ips', user_id=user_id))
|
||||
response.mustcontain('All IP addresses are allowed')
|
||||
response.mustcontain(no=[ip])
|
||||
response.mustcontain(no=[ip_range])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue