# -*- coding: utf-8 -*-
"""
rhodecode.controllers.login
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Login controller for rhodeocode
:created_on: Apr 22, 2010
:author: marcink
:copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
:license: GPLv3, see COPYING for more details.
# 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/>.
import logging
import formencode
import datetime
import urlparse
from formencode import htmlfill
from webob.exc import HTTPFound
from pylons.i18n.translation import _
from pylons.controllers.util import abort, redirect
from pylons import request, response, session, tmpl_context as c, url
import rhodecode.lib.helpers as h
from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator
from rhodecode.lib.base import BaseController, render
from rhodecode.model.db import User
from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
from rhodecode.model.user import UserModel
from rhodecode.model.meta import Session
log = logging.getLogger(__name__)
class LoginController(BaseController):
def __before__(self):
super(LoginController, self).__before__()
def index(self):
# redirect if already logged in
c.came_from = request.GET.get('came_from')
not_default = self.rhodecode_user.username != 'default'
ip_allowed = self.rhodecode_user.ip_allowed
if self.rhodecode_user.is_authenticated and not_default and ip_allowed:
return redirect(url('home'))
if request.POST:
# import Login Form validator class
login_form = LoginForm()
try:
session.invalidate()
c.form_result = login_form.to_python(dict(request.POST))
# form checks for username/password, now we're authenticated
username = c.form_result['username']
user = User.get_by_username(username, case_insensitive=True)
auth_user = AuthUser(user.user_id)
auth_user.set_authenticated()
cs = auth_user.get_cookie_store()
session['rhodecode_user'] = cs
user.update_lastlogin()
Session().commit()
# If they want to be remembered, update the cookie
if c.form_result['remember'] is not False:
_year = (datetime.datetime.now() +
datetime.timedelta(seconds=60 * 60 * 24 * 365))
session._set_cookie_expires(_year)
session.save()
log.info('user %s is now authenticated and stored in '
'session, session attrs %s' % (username, cs))
# dumps session attrs back to cookie
session._update_cookie_out()
# we set new cookie
headers = None
if session.request['set_cookie']:
# send set-cookie headers back to response to update cookie
headers = [('Set-Cookie', session.request['cookie_out'])]
allowed_schemes = ['http', 'https']
if c.came_from:
parsed = urlparse.urlparse(c.came_from)
server_parsed = urlparse.urlparse(url.current())
if parsed.scheme and parsed.scheme not in allowed_schemes:
log.error(
'Suspicious URL scheme detected %s for url %s' %
(parsed.scheme, parsed))
c.came_from = url('home')
elif server_parsed.netloc != parsed.netloc:
log.error('Suspicious NETLOC detected %s for url %s'
'server url is: %s' %
(parsed.netloc, parsed, server_parsed))
raise HTTPFound(location=c.came_from, headers=headers)
else:
raise HTTPFound(location=url('home'), headers=headers)
except formencode.Invalid, errors:
return htmlfill.render(
render('/login.html'),
defaults=errors.value,
errors=errors.error_dict or {},
prefix_error=False,
encoding="UTF-8")
return render('/login.html')
@HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
'hg.register.manual_activate')
def register(self):
c.auto_active = False
for perm in User.get_by_username('default').user_perms:
if perm.permission.permission_name == 'hg.register.auto_activate':
c.auto_active = True
break
c.auto_active = 'hg.register.auto_activate' in User.get_by_username('default')\
.AuthUser.permissions['global']
register_form = RegisterForm()()
form_result = register_form.to_python(dict(request.POST))
form_result['active'] = c.auto_active
UserModel().create_registration(form_result)
h.flash(_('You have successfully registered into rhodecode'),
h.flash(_('You have successfully registered into RhodeCode'),
category='success')
return redirect(url('login_home'))
render('/register.html'),
return render('/register.html')
def password_reset(self):
password_reset_form = PasswordResetForm()()
form_result = password_reset_form.to_python(dict(request.POST))
UserModel().reset_password_link(form_result)
h.flash(_('Your password reset link was sent'),
render('/password_reset.html'),
return render('/password_reset.html')
def password_reset_confirmation(self):
if request.GET and request.GET.get('key'):
user = User.get_by_api_key(request.GET.get('key'))
data = dict(email=user.email)
UserModel().reset_password(data)
h.flash(_('Your password reset was successful, '
'new password has been sent to your email'),
except Exception, e:
log.error(e)
return redirect(url('reset_password'))
def logout(self):
session.delete()
log.info('Logging out and deleting session for user')
redirect(url('home'))
@@ -37,412 +37,416 @@ from pylons.i18n.translation import _
from sqlalchemy.orm.exc import ObjectDeletedError
from rhodecode import __platform__, is_windows, is_unix
from rhodecode.lib.utils2 import str2bool, safe_unicode
from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError
from rhodecode.lib.utils import get_repo_slug, get_repos_group_slug
from rhodecode.lib.auth_ldap import AuthLdap
from rhodecode.model import meta
from rhodecode.model.db import Permission, RhodeCodeSetting, User, UserIpMap
from rhodecode.lib.caching_query import FromCache
class PasswordGenerator(object):
This is a simple class for generating password from different sets of
characters
usage::
passwd_gen = PasswordGenerator()
#print 8-letter password containing only big and small letters
of alphabet
passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
ALPHABETS_NUM = r'''1234567890'''
ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
+ ALPHABETS_NUM + ALPHABETS_SPECIAL
ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
def __init__(self, passwd=''):
self.passwd = passwd
def gen_password(self, length, type_=None):
if type_ is None:
type_ = self.ALPHABETS_FULL
self.passwd = ''.join([random.choice(type_) for _ in xrange(length)])
return self.passwd
class RhodeCodeCrypto(object):
@classmethod
def hash_string(cls, str_):
Cryptographic function used for password hashing based on pybcrypt
or pycrypto in windows
:param password: password to hash
if is_windows:
from hashlib import sha256
return sha256(str_).hexdigest()
elif is_unix:
import bcrypt
return bcrypt.hashpw(str_, bcrypt.gensalt(10))
raise Exception('Unknown or unsupported platform %s' \
% __platform__)
def hash_check(cls, password, hashed):
Checks matching password with it's hashed value, runs different
implementation based on platform it runs on
:param password: password
:param hashed: password in hashed form
return sha256(password).hexdigest() == hashed
return bcrypt.hashpw(password, hashed) == hashed
def get_crypt_password(password):
return RhodeCodeCrypto.hash_string(password)
def check_password(password, hashed):
return RhodeCodeCrypto.hash_check(password, hashed)
def generate_api_key(str_, salt=None):
Generates API KEY from given string
:param str_:
:param salt:
if salt is None:
salt = _RandomNameSequence().next()
return hashlib.sha1(str_ + salt).hexdigest()
def authfunc(environ, username, password):
Dummy authentication wrapper function used in Mercurial and Git for
access control.
:param environ: needed only for using in Basic auth
return authenticate(username, password)
def authenticate(username, password):
Authentication function used for access control,
firstly checks for db authentication then if ldap is enabled for ldap
authentication, also creates ldap user if not in database
:param username: username
user_model = UserModel()
user = User.get_by_username(username)
log.debug('Authenticating user using RhodeCode account')
if user is not None and not user.ldap_dn:
if user.active:
if user.username == 'default' and user.active:
log.info('user %s authenticated correctly as anonymous user' %
username)
return True
elif user.username == username and check_password(password,
user.password):
log.info('user %s authenticated correctly' % username)
log.warning('user %s tried auth but is disabled' % username)
log.debug('Regular authentication failed')
user_obj = User.get_by_username(username, case_insensitive=True)
if user_obj is not None and not user_obj.ldap_dn:
log.debug('this user already exists as non ldap')
return False
ldap_settings = RhodeCodeSetting.get_ldap_settings()
#======================================================================
# FALLBACK TO LDAP AUTH IF ENABLE
if str2bool(ldap_settings.get('ldap_active')):
log.debug("Authenticating user using ldap")
kwargs = {
'server': ldap_settings.get('ldap_host', ''),
'base_dn': ldap_settings.get('ldap_base_dn', ''),
'port': ldap_settings.get('ldap_port'),
'bind_dn': ldap_settings.get('ldap_dn_user'),
'bind_pass': ldap_settings.get('ldap_dn_pass'),
'tls_kind': ldap_settings.get('ldap_tls_kind'),
'tls_reqcert': ldap_settings.get('ldap_tls_reqcert'),
'ldap_filter': ldap_settings.get('ldap_filter'),
'search_scope': ldap_settings.get('ldap_search_scope'),
'attr_login': ldap_settings.get('ldap_attr_login'),
'ldap_version': 3,
}
log.debug('Checking for ldap authentication')
aldap = AuthLdap(**kwargs)
(user_dn, ldap_attrs) = aldap.authenticate_ldap(username,
password)
log.debug('Got ldap DN response %s' % user_dn)
get_ldap_attr = lambda k: ldap_attrs.get(ldap_settings\
.get(k), [''])[0]
user_attrs = {
'name': safe_unicode(get_ldap_attr('ldap_attr_firstname')),
'lastname': safe_unicode(get_ldap_attr('ldap_attr_lastname')),
'email': get_ldap_attr('ldap_attr_email'),
'active': 'hg.register.auto_activate' in User\
.get_by_username('default').AuthUser.permissions['global']
# don't store LDAP password since we don't need it. Override
# with some random generated password
_password = PasswordGenerator().gen_password(length=8)
# create this user on the fly if it doesn't exist in rhodecode
# database
if user_model.create_ldap(username, _password, user_dn,
user_attrs):
log.info('created new ldap user %s' % username)
except (LdapUsernameError, LdapPasswordError,):
pass
except (Exception,):
log.error(traceback.format_exc())
def login_container_auth(username):
if user is None:
'name': username,
'lastname': None,
'email': None,
user = UserModel().create_for_container_auth(username, user_attrs)
if not user:
return None
log.info('User %s was created by container authentication' % username)
if not user.active:
log.debug('User %s is now logged in by container authentication',
user.username)
return user
def get_container_username(environ, config, clean_username=False):
Get's the container_auth username (or email). It tries to get username
from REMOTE_USER if container_auth_enabled is enabled, if that fails
it tries to get username from HTTP_X_FORWARDED_USER if proxypass_auth_enabled
is enabled. clean_username extracts the username from this data if it's
having @ in it.
:param environ:
:param config:
:param clean_username:
username = None
if str2bool(config.get('container_auth_enabled', False)):
from paste.httpheaders import REMOTE_USER
username = REMOTE_USER(environ)
log.debug('extracted REMOTE_USER:%s' % (username))
if not username and str2bool(config.get('proxypass_auth_enabled', False)):
username = environ.get('HTTP_X_FORWARDED_USER')
log.debug('extracted HTTP_X_FORWARDED_USER:%s' % (username))
if username and clean_username:
# Removing realm and domain from username
username = username.partition('@')[0]
username = username.rpartition('\\')[2]
log.debug('Received username %s from container' % username)
return username
class CookieStoreWrapper(object):
def __init__(self, cookie_store):
self.cookie_store = cookie_store
def __repr__(self):
return 'CookieStore<%s>' % (self.cookie_store)
def get(self, key, other=None):
if isinstance(self.cookie_store, dict):
return self.cookie_store.get(key, other)
elif isinstance(self.cookie_store, AuthUser):
return self.cookie_store.__dict__.get(key, other)
class AuthUser(object):
A simple object that handles all attributes of user in RhodeCode
It does lookup based on API key,given user, or user present in session
Then it fills all required information for such user. It also checks if
anonymous access is enabled and if so, it returns default user as logged
in
def __init__(self, user_id=None, api_key=None, username=None, ip_addr=None):
self.user_id = user_id
self.api_key = None
self.username = username
self.ip_addr = ip_addr
self.name = ''
self.lastname = ''
self.email = ''
self.is_authenticated = False
self.admin = False
self.inherit_default_permissions = False
self.permissions = {}
self._api_key = api_key
self.propagate_data()
self._instance = None
def propagate_data(self):
self.anonymous_user = User.get_by_username('default', cache=True)
is_user_loaded = False
# try go get user by api key
if self._api_key and self._api_key != self.anonymous_user.api_key:
log.debug('Auth User lookup by API KEY %s' % self._api_key)
is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
# lookup by userid
elif (self.user_id is not None and
self.user_id != self.anonymous_user.user_id):
log.debug('Auth User lookup by USER ID %s' % self.user_id)
is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
# lookup by username
elif self.username and \
str2bool(config.get('container_auth_enabled', False)):
log.debug('Auth User lookup by USER NAME %s' % self.username)
dbuser = login_container_auth(self.username)
if dbuser is not None:
log.debug('filling all attributes to object')
for k, v in dbuser.get_dict().items():
setattr(self, k, v)
self.set_authenticated()
is_user_loaded = True
log.debug('No data in %s that could been used to log in' % self)
if not is_user_loaded:
# if we cannot authenticate user try anonymous
if self.anonymous_user.active is True:
user_model.fill_data(self, user_id=self.anonymous_user.user_id)
# then we set this user is logged in
self.is_authenticated = True
self.user_id = None
self.username = None
if not self.username:
self.username = 'None'
log.debug('Auth User is now %s' % self)
user_model.fill_perms(self)
@property
def is_admin(self):
return self.admin
def ip_allowed(self):
Checks if ip_addr used in constructor is allowed from defined list of
allowed ip_addresses for user
:returns: boolean, True if ip is in allowed ip range
#check IP
allowed_ips = AuthUser.get_allowed_ips(self.user_id, cache=True)
if check_ip_access(source_ip=self.ip_addr, allowed_ips=allowed_ips):
log.debug('IP:%s is in range of %s' % (self.ip_addr, allowed_ips))
log.info('Access for IP:%s forbidden, '
'not in %s' % (self.ip_addr, allowed_ips))
return "<AuthUser('id:%s:%s|%s')>" % (self.user_id, self.username,
self.is_authenticated)
def set_authenticated(self, authenticated=True):
if self.user_id != self.anonymous_user.user_id:
self.is_authenticated = authenticated
def get_cookie_store(self):
return {'username': self.username,
'user_id': self.user_id,
'is_authenticated': self.is_authenticated}
def from_cookie_store(cls, cookie_store):
Creates AuthUser from a cookie store
:param cls:
:param cookie_store:
user_id = cookie_store.get('user_id')
username = cookie_store.get('username')
api_key = cookie_store.get('api_key')
return AuthUser(user_id, api_key, username)
def get_allowed_ips(cls, user_id, cache=False):
_set = set()
user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
if cache:
user_ips = user_ips.options(FromCache("sql_cache_short",
@@ -210,384 +210,393 @@ class RhodeCodeSetting(Base, BaseModel):
ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
if not ret:
raise Exception('Could not get application settings !')
settings = {}
for each in ret:
settings['rhodecode_' + each.app_settings_name] = \
each.app_settings_value
return settings
def get_ldap_settings(cls, cache=False):
ret = cls.query()\
.filter(cls.app_settings_name.startswith('ldap_')).all()
fd = {}
for row in ret:
fd.update({row.app_settings_name: row.app_settings_value})
return fd
def get_default_repo_settings(cls, cache=False, strip_prefix=False):
.filter(cls.app_settings_name.startswith('default_')).all()
key = row.app_settings_name
if strip_prefix:
key = remove_prefix(key, prefix='default_')
fd.update({key: row.app_settings_value})
class RhodeCodeUi(Base, BaseModel):
__tablename__ = 'rhodecode_ui'
__table_args__ = (
UniqueConstraint('ui_key'),
{'extend_existing': True, 'mysql_engine': 'InnoDB',
'mysql_charset': 'utf8'}
)
HOOK_UPDATE = 'changegroup.update'
HOOK_REPO_SIZE = 'changegroup.repo_size'
HOOK_PUSH = 'changegroup.push_logger'
HOOK_PRE_PUSH = 'prechangegroup.pre_push'
HOOK_PULL = 'outgoing.pull_logger'
HOOK_PRE_PULL = 'preoutgoing.pre_pull'
ui_id = Column("ui_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
ui_section = Column("ui_section", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_key = Column("ui_key", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_value = Column("ui_value", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_active = Column("ui_active", Boolean(), nullable=True, unique=None, default=True)
def get_by_key(cls, key):
return cls.query().filter(cls.ui_key == key).scalar()
def get_builtin_hooks(cls):
q = cls.query()
q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,
cls.HOOK_PULL, cls.HOOK_PRE_PULL]))
return q.all()
def get_custom_hooks(cls):
q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
q = q.filter(cls.ui_section == 'hooks')
def get_repos_location(cls):
return cls.get_by_key('/').ui_value
def create_or_update_hook(cls, key, val):
new_ui = cls.get_by_key(key) or cls()
new_ui.ui_section = 'hooks'
new_ui.ui_active = True
new_ui.ui_key = key
new_ui.ui_value = val
Session().add(new_ui)
return '<DB:%s[%s:%s]>' % (self.__class__.__name__, self.ui_key,
self.ui_value)
class User(Base, BaseModel):
__tablename__ = 'users'
UniqueConstraint('username'), UniqueConstraint('email'),
Index('u_username_idx', 'username'),
Index('u_email_idx', 'email'),
DEFAULT_USER = 'default'
DEFAULT_PERMISSIONS = [
'hg.register.manual_activate', 'hg.create.repository',
'hg.fork.repository', 'repository.read', 'group.read'
]
user_id = Column("user_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
username = Column("username", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
password = Column("password", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
active = Column("active", Boolean(), nullable=True, unique=None, default=True)
admin = Column("admin", Boolean(), nullable=True, unique=None, default=False)
name = Column("firstname", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
lastname = Column("lastname", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
_email = Column("email", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
last_login = Column("last_login", DateTime(timezone=False), nullable=True, unique=None, default=None)
ldap_dn = Column("ldap_dn", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
api_key = Column("api_key", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
inherit_default_permissions = Column("inherit_default_permissions", Boolean(), nullable=False, unique=None, default=True)
user_log = relationship('UserLog')
user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
repositories = relationship('Repository')
user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')
repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')
repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')
group_member = relationship('UsersGroupMember', cascade='all')
notifications = relationship('UserNotification', cascade='all')
# notifications assigned to this user
user_created_notifications = relationship('Notification', cascade='all')
# comments created by this user
user_comments = relationship('ChangesetComment', cascade='all')
#extra emails for this user
user_emails = relationship('UserEmailMap', cascade='all')
@hybrid_property
def email(self):
return self._email
@email.setter
def email(self, val):
self._email = val.lower() if val else None
def firstname(self):
# alias for future
return self.name
def emails(self):
other = UserEmailMap.query().filter(UserEmailMap.user==self).all()
return [self.email] + [x.email for x in other]
def ip_addresses(self):
ret = UserIpMap.query().filter(UserIpMap.user == self).all()
return [x.ip_addr for x in ret]
def username_and_name(self):
return '%s (%s %s)' % (self.username, self.firstname, self.lastname)
def full_name(self):
return '%s %s' % (self.firstname, self.lastname)
def full_name_or_username(self):
return ('%s %s' % (self.firstname, self.lastname)
if (self.firstname and self.lastname) else self.username)
def full_contact(self):
return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
def short_contact(self):
def AuthUser(self):
Returns instance of AuthUser for this user
from rhodecode.lib.auth import AuthUser
return AuthUser(user_id=self.user_id, api_key=self.api_key,
username=self.username)
def __unicode__(self):
return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
self.user_id, self.username)
def get_by_username(cls, username, case_insensitive=False, cache=False):
if case_insensitive:
q = cls.query().filter(cls.username.ilike(username))
q = cls.query().filter(cls.username == username)
q = q.options(FromCache(
"sql_cache_short",
"get_user_%s" % _hash_key(username)
return q.scalar()
def get_by_api_key(cls, api_key, cache=False):
q = cls.query().filter(cls.api_key == api_key)
q = q.options(FromCache("sql_cache_short",
"get_api_key_%s" % api_key))
def get_by_email(cls, email, case_insensitive=False, cache=False):
q = cls.query().filter(cls.email.ilike(email))
q = cls.query().filter(cls.email == email)
"get_email_key_%s" % email))
ret = q.scalar()
if ret is None:
q = UserEmailMap.query()
# try fetching in alternate email map
q = q.filter(UserEmailMap.email.ilike(email))
q = q.filter(UserEmailMap.email == email)
q = q.options(joinedload(UserEmailMap.user))
"get_email_map_key_%s" % email))
ret = getattr(q.scalar(), 'user', None)
return ret
def get_from_cs_author(cls, author):
Tries to get User objects out of commit author string
:param author:
from rhodecode.lib.helpers import email, author_name
# Valid email in the attribute passed, see if they're in the system
_email = email(author)
if _email:
user = cls.get_by_email(_email, case_insensitive=True)
if user:
# Maybe we can match by username?
_author = author_name(author)
user = cls.get_by_username(_author, case_insensitive=True)
def update_lastlogin(self):
"""Update user lastlogin"""
self.last_login = datetime.datetime.now()
Session().add(self)
log.debug('updated user %s lastlogin' % self.username)
def get_api_data(self):
Common function for generating user related data for API
user = self
data = dict(
user_id=user.user_id,
username=user.username,
firstname=user.name,
lastname=user.lastname,
email=user.email,
emails=user.emails,
api_key=user.api_key,
active=user.active,
admin=user.admin,
ldap_dn=user.ldap_dn,
last_login=user.last_login,
ip_addresses=user.ip_addresses
return data
def __json__(self):
full_name=self.full_name,
full_name_or_username=self.full_name_or_username,
short_contact=self.short_contact,
full_contact=self.full_contact
data.update(self.get_api_data())
class UserEmailMap(Base, BaseModel):
__tablename__ = 'user_email_map'
Index('uem_email_idx', 'email'),
UniqueConstraint('email'),
__mapper_args__ = {}
email_id = Column("email_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
_email = Column("email", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
user = relationship('User', lazy='joined')
@validates('_email')
def validate_email(self, key, email):
# check if this email is not main one
main_email = Session().query(User).filter(User.email == email).scalar()
if main_email is not None:
raise AttributeError('email %s is present is user table' % email)
return email
class UserIpMap(Base, BaseModel):
__tablename__ = 'user_ip_map'
UniqueConstraint('user_id', 'ip_addr'),
ip_id = Column("ip_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
ip_addr = Column("ip_addr", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
def _get_ip_range(cls, ip_addr):
from rhodecode.lib import ipaddr
net = ipaddr.IPNetwork(address=ip_addr)
return [str(net.network), str(net.broadcast)]
return dict(
ip_addr=self.ip_addr,
ip_range=self._get_ip_range(self.ip_addr)
class UserLog(Base, BaseModel):
__tablename__ = 'user_logs'
'mysql_charset': 'utf8'},
user_log_id = Column("user_log_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True)
repository_name = Column("repository_name", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
user_ip = Column("user_ip", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action = Column("action", UnicodeText(1200000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action_date = Column("action_date", DateTime(timezone=False), nullable=True, unique=None, default=None)
def action_as_day(self):
return datetime.date(*self.action_date.timetuple()[:3])
Status change: