kallithea Changeset - 7557da2252a3

Changeset - 7557da2252a3

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Søren Løvborg - 10 years ago 2015-07-26 14:10:16
kwi@kwi.dk

auth: construct AuthUser from either user_id or db.User object

If the caller already has the database User object, there's no reason
for AuthUser to look it up again.

The `api_key` lookup functionality is dropped, because 1) it's only
used in one place, and 2) it's simple enough for the caller to do the
lookup itself.

The `user_id` lookup functionality is kept, because 1) it's frequently
used, and 2) far from a simple `User.get(id)` lookup, it has a complex
interaction with UserModel. (That cleanup will have to wait for another
day.)

All calls of the form `AuthUser(user_id=x.user_id)` can be replaced with
`AuthUser(dbuser=x)`, assuming `x` is a db.User. However, verifying that
assumption requires a manual audit of every call site, since `x` might
also be another `AuthUser` object, for instance. Therefore, only the
most obvious call sites have been fixed here.

5 files changed with 17 insertions and 20 deletions:

kallithea/controllers/api/__init__.py

kallithea/lib/auth.py

kallithea/lib/base.py

kallithea/model/db.py

kallithea/tests/api/api_base.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/__init__.py

➞

Show inline comments

 # -*- 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.controllers.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~
 JSON RPC controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import inspect
 import logging
 import types
 import traceback
 import time
 from paste.response import replace_header
 from pylons.controllers import WSGIController
 from webob.exc import HTTPError
 from kallithea.model.db import User
 from kallithea.model import meta
 from kallithea.lib.compat import izip_longest, json
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.base import _get_ip_addr as _get_ip, _get_access_path
 from kallithea.lib.utils2 import safe_unicode, safe_str
 log = logging.getLogger('JSONRPC')
 class JSONRPCError(BaseException):
     def __init__(self, message):
         self.message = message
         super(JSONRPCError, self).__init__()
     def __str__(self):
         return safe_str(self.message)
 def jsonrpc_error(message, retid=None, code=None):
     """
     Generate a Response object with a JSON-RPC error body
     :param code:
     :param retid:
     :param message:
     """
     from pylons.controllers.util import Response
     return Response(
         body=json.dumps(dict(id=retid, result=None, error=message)),
         status=code,
         content_type='application/json'
+    )
 class JSONRPCController(WSGIController):
     """
      A WSGI-speaking JSON-RPC controller class
      See the specification:
      <http://json-rpc.org/wiki/specification>`.
      Valid controller return values should be json-serializable objects.
      Sub-classes should catch their exceptions and raise JSONRPCError
      if they want to pass meaningful errors to the client.
      """
     def _get_ip_addr(self, environ):
         return _get_ip(environ)
     def _get_method_args(self):
         """
         Return `self._rpc_args` to dispatched controller method
         chosen by __call__
         """
         return self._rpc_args
     def __call__(self, environ, start_response):
         """
         Parse the request body as JSON, look up the method on the
         controller and if it exists, dispatch to it.
         """
         try:
             return self._handle_request(environ, start_response)
         finally:
             meta.Session.remove()
     def _handle_request(self, environ, start_response):
         start = time.time()
         ip_addr = self.ip_addr = self._get_ip_addr(environ)
         self._req_id = None
         if 'CONTENT_LENGTH' not in environ:
             log.debug("No Content-Length")
             return jsonrpc_error(retid=self._req_id,
                                  message="No Content-Length in request")
         else:
             length = environ['CONTENT_LENGTH'] or 0
             length = int(environ['CONTENT_LENGTH'])
             log.debug('Content-Length: %s' % length)
         if length == 0:
             log.debug("Content-Length is 0")
             return jsonrpc_error(retid=self._req_id,
                                  message="Content-Length is 0")
         raw_body = environ['wsgi.input'].read(length)
         try:
             json_body = json.loads(raw_body)
         except ValueError, e:
             # catch JSON errors Here
             return jsonrpc_error(retid=self._req_id,
                                  message="JSON parse error ERR:%s RAW:%r"
                                  % (e, raw_body))
         # check AUTH based on API key
         try:
             self._req_api_key = json_body['api_key']
             self._req_id = json_body['id']
             self._req_method = json_body['method']
             self._request_params = json_body['args']
             if not isinstance(self._request_params, dict):
                 self._request_params = {}
             log.debug(
                 'method: %s, params: %s' % (self._req_method,
                                             self._request_params)
+            )
         except KeyError, e:
             return jsonrpc_error(retid=self._req_id,
                                  message='Incorrect JSON query missing %s' % e)
         # check if we can find this session using api_key
         try:
             u = User.get_by_api_key(self._req_api_key)
             if u is None:
                 return jsonrpc_error(retid=self._req_id,
                                      message='Invalid API key')
-            auth_u = AuthUser(u.user_id)
+            auth_u = AuthUser(dbuser=u)
             if not AuthUser.check_ip_allowed(auth_u, ip_addr):
                 return jsonrpc_error(retid=self._req_id,
                         message='request from IP:%s not allowed' % (ip_addr,))
             else:
                 log.info('Access for IP:%s allowed' % (ip_addr,))
         except Exception, e:
             return jsonrpc_error(retid=self._req_id,
                                  message='Invalid API key')
         self._error = None
         try:
             self._func = self._find_method()
         except AttributeError, e:
             return jsonrpc_error(retid=self._req_id,
                                  message=str(e))
         # now that we have a method, add self._req_params to
         # self.kargs and dispatch control to WGIController
         argspec = inspect.getargspec(self._func)
         arglist = argspec[0][1:]
         defaults = map(type, argspec[3] or [])
         default_empty = types.NotImplementedType
         # kw arguments required by this method
         func_kwargs = dict(izip_longest(reversed(arglist), reversed(defaults),
                                         fillvalue=default_empty))
         # this is little trick to inject logged in user for
         # perms decorators to work they expect the controller class to have
         # authuser attribute set
         self.authuser = auth_u
         # This attribute will need to be first param of a method that uses
         # api_key, which is translated to instance of user at that name
         USER_SESSION_ATTR = 'apiuser'
         if USER_SESSION_ATTR not in arglist:
             return jsonrpc_error(
                 retid=self._req_id,
                 message='This method [%s] does not support '
                          'authentication (missing %s param)' % (
                                     self._func.__name__, USER_SESSION_ATTR)
+            )
         # get our arglist and check if we provided them as args
         for arg, default in func_kwargs.iteritems():
             if arg == USER_SESSION_ATTR:
                 # USER_SESSION_ATTR is something translated from API key and
                 # this is checked before so we don't need validate it
                 continue
             # skip the required param check if it's default value is
             # NotImplementedType (default_empty)
             if default == default_empty and arg not in self._request_params:
                 return jsonrpc_error(
                     retid=self._req_id,
                     message=(
                         'Missing non optional `%s` arg in JSON DATA' % arg
+                    )
+                )
         self._rpc_args = {USER_SESSION_ATTR: u}
         self._rpc_args.update(self._request_params)
         self._rpc_args['action'] = self._req_method
         self._rpc_args['environ'] = environ
         self._rpc_args['start_response'] = start_response
         status = []
         headers = []
         exc_info = []
         def change_content(new_status, new_headers, new_exc_info=None):
             status.append(new_status)
             headers.extend(new_headers)
             exc_info.append(new_exc_info)
         output = WSGIController.__call__(self, environ, change_content)
         output = list(output)
         headers.append(('Content-Length', str(len(output[0]))))
         replace_header(headers, 'Content-Type', 'application/json')
         start_response(status[0], headers, exc_info[0])
         log.info('IP: %s Request to %s time: %.3fs' % (
             self._get_ip_addr(environ),
             safe_unicode(_get_access_path(environ)), time.time() - start)
+        )
         return output
     def _dispatch_call(self):
         """
         Implement dispatch interface specified by WSGIController
         """
         raw_response = ''
         try:
             raw_response = self._inspect_call(self._func)
             if isinstance(raw_response, HTTPError):
                 self._error = str(raw_response)
         except JSONRPCError, e:
             self._error = safe_str(e)
         except Exception, e:
             log.error('Encountered unhandled exception: %s'
                       % (traceback.format_exc(),))
             json_exc = JSONRPCError('Internal server error')
             self._error = safe_str(json_exc)
         if self._error is not None:
             raw_response = None
         response = dict(id=self._req_id, result=raw_response, error=self._error)
         try:
             return json.dumps(response)
         except TypeError, e:
             log.error('API FAILED. Error encoding response: %s' % e)
             return json.dumps(
                 dict(
                     id=self._req_id,
                     result=None,
                     error="Error encoding response"
+                )
+            )
     def _find_method(self):
         """
         Return method named by `self._req_method` in controller if able
         """
         log.debug('Trying to find JSON-RPC method: %s' % (self._req_method,))
         if self._req_method.startswith('_'):
             raise AttributeError("Method not allowed")
         try:
             func = getattr(self, self._req_method, None)
         except UnicodeEncodeError:
             raise AttributeError("Problem decoding unicode in requested "
                                  "method name.")
         if isinstance(func, types.MethodType):
             return func
         else:
             raise AttributeError("No such method: %s" % (self._req_method,))

kallithea/lib/auth.py

➞

Show inline comments

@@ @@ -79,816 +79,812 @@ class PasswordGenerator(object): @@
     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 gen_password(self, length, alphabet=ALPHABETS_FULL):
         assert len(alphabet) <= 256, alphabet
         l = []
         while len(l) < length:
             i = ord(os.urandom(1))
             if i < len(alphabet):
                 l.append(alphabet[i])
         return ''.join(l)
 class KallitheaCrypto(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:
             return hashlib.sha256(str_).hexdigest()
         elif is_unix:
             import bcrypt
             return bcrypt.hashpw(str_, bcrypt.gensalt(10))
         else:
             raise Exception('Unknown or unsupported platform %s' \
                             % __platform__)
     @classmethod
     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
         """
         if is_windows:
             return hashlib.sha256(password).hexdigest() == hashed
         elif is_unix:
             import bcrypt
             return bcrypt.hashpw(password, hashed) == hashed
         else:
             raise Exception('Unknown or unsupported platform %s' \
                             % __platform__)
 def get_crypt_password(password):
     return KallitheaCrypto.hash_string(password)
 def check_password(password, hashed):
     return KallitheaCrypto.hash_check(password, hashed)
 def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,
                        explicit, algo):
     RK = 'repositories'
     GK = 'repositories_groups'
     UK = 'user_groups'
     GLOBAL = 'global'
     PERM_WEIGHTS = Permission.PERM_WEIGHTS
     permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
     def _choose_perm(new_perm, cur_perm):
         new_perm_val = PERM_WEIGHTS[new_perm]
         cur_perm_val = PERM_WEIGHTS[cur_perm]
         if algo == 'higherwin':
             if new_perm_val > cur_perm_val:
                 return new_perm
             return cur_perm
         elif algo == 'lowerwin':
             if new_perm_val < cur_perm_val:
                 return new_perm
             return cur_perm
     #======================================================================
     # fetch default permissions
     #======================================================================
     default_user = User.get_by_username('default', cache=True)
     default_user_id = default_user.user_id
     default_repo_perms = Permission.get_default_perms(default_user_id)
     default_repo_groups_perms = Permission.get_default_group_perms(default_user_id)
     default_user_group_perms = Permission.get_default_user_group_perms(default_user_id)
     if user_is_admin:
         #==================================================================
         # admin users have all rights;
         # based on default permissions, just set everything to admin
         #==================================================================
         permissions[GLOBAL].add('hg.admin')
         permissions[GLOBAL].add('hg.create.write_on_repogroup.true')
         # repositories
         for perm in default_repo_perms:
             r_k = perm.UserRepoToPerm.repository.repo_name
             p = 'repository.admin'
             permissions[RK][r_k] = p
         # repository groups
         for perm in default_repo_groups_perms:
             rg_k = perm.UserRepoGroupToPerm.group.group_name
             p = 'group.admin'
             permissions[GK][rg_k] = p
         # user groups
         for perm in default_user_group_perms:
             u_k = perm.UserUserGroupToPerm.user_group.users_group_name
             p = 'usergroup.admin'
             permissions[UK][u_k] = p
         return permissions
     #==================================================================
     # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
     #==================================================================
     # default global permissions taken from the default user
     default_global_perms = UserToPerm.query()\
         .filter(UserToPerm.user_id == default_user_id)\
         .options(joinedload(UserToPerm.permission))
     for perm in default_global_perms:
         permissions[GLOBAL].add(perm.permission.permission_name)
     # defaults for repositories, taken from default user
     for perm in default_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         if perm.Repository.private and not (perm.Repository.user_id == user_id):
             # disable defaults for private repos,
             p = 'repository.none'
         elif perm.Repository.user_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
         permissions[RK][r_k] = p
     # defaults for repository groups taken from default user permission
     # on given group
     for perm in default_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         permissions[GK][rg_k] = p
     # defaults for user groups taken from default user permission
     # on given user group
     for perm in default_user_group_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         permissions[UK][u_k] = p
     #======================================================================
     # !! OVERRIDE GLOBALS !! with user permissions if any found
     #======================================================================
     # those can be configured from groups or users explicitly
     _configurable = set([
         'hg.fork.none', 'hg.fork.repository',
         'hg.create.none', 'hg.create.repository',
         'hg.usergroup.create.false', 'hg.usergroup.create.true'
     ])
     # USER GROUPS comes first
     # user group global permissions
     user_perms_from_users_groups = Session().query(UserGroupToPerm)\
         .options(joinedload(UserGroupToPerm.permission))\
         .join((UserGroupMember, UserGroupToPerm.users_group_id ==
                UserGroupMember.users_group_id))\
         .filter(UserGroupMember.user_id == user_id)\
         .join((UserGroup, UserGroupMember.users_group_id ==
                UserGroup.users_group_id))\
         .filter(UserGroup.users_group_active == True)\
         .order_by(UserGroupToPerm.users_group_id)\
         .all()
     # need to group here by groups since user can be in more than
     # one group
     _grouped = [[x, list(y)] for x, y in
                 itertools.groupby(user_perms_from_users_groups,
                                   lambda x:x.users_group)]
     for gr, perms in _grouped:
         # since user can be in multiple groups iterate over them and
         # select the lowest permissions first (more explicit)
         ##TODO: do this^^
         if not gr.inherit_default_permissions:
             # NEED TO IGNORE all configurable permissions and
             # replace them with explicitly set
             permissions[GLOBAL] = permissions[GLOBAL]\
                                             .difference(_configurable)
         for perm in perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     # user specific global permissions
     user_perms = Session().query(UserToPerm)\
             .options(joinedload(UserToPerm.permission))\
             .filter(UserToPerm.user_id == user_id).all()
     if not user_inherit_default_permissions:
         # NEED TO IGNORE all configurable permissions and
         # replace them with explicitly set
         permissions[GLOBAL] = permissions[GLOBAL]\
                                         .difference(_configurable)
         for perm in user_perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     ## END GLOBAL PERMISSIONS
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORIES !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repositories permissions
     user_repo_perms_from_users_groups = \
      Session().query(UserGroupRepoToPerm, Permission, Repository,)\
         .join((Repository, UserGroupRepoToPerm.repository_id ==
                Repository.repo_id))\
         .join((Permission, UserGroupRepoToPerm.permission_id ==
                Permission.permission_id))\
         .join((UserGroup, UserGroupRepoToPerm.users_group_id ==
                UserGroup.users_group_id))\
         .filter(UserGroup.users_group_active == True)\
         .join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
                UserGroupMember.users_group_id))\
         .filter(UserGroupMember.user_id == user_id)\
         .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_perms_from_users_groups:
         r_k = perm.UserGroupRepoToPerm.repository.repo_name
         multiple_counter[r_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[RK][r_k]
         if perm.Repository.user_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             if multiple_counter[r_k] > 1:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     # user explicit permissions for repositories, overrides any specified
     # by the group permission
     user_repo_perms = Permission.get_default_perms(user_id)
     for perm in user_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         cur_perm = permissions[RK][r_k]
         # set admin if owner
         if perm.Repository.user_id == user_id:
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
             if not explicit:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORY GROUPS !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository groups and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repo groups permissions
     user_repo_group_perms_from_users_groups = \
      Session().query(UserGroupRepoGroupToPerm, Permission, RepoGroup)\
      .join((RepoGroup, UserGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
      .join((Permission, UserGroupRepoGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroup, UserGroupRepoGroupToPerm.users_group_id ==
             UserGroup.users_group_id))\
      .filter(UserGroup.users_group_active == True)\
      .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == user_id)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_group_perms_from_users_groups:
         g_k = perm.UserGroupRepoGroupToPerm.group.group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[GK][g_k] = p
     # user explicit permissions for repository groups
     user_repo_groups_perms = Permission.get_default_group_perms(user_id)
     for perm in user_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][rg_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[GK][rg_k] = p
     #======================================================================
     # !! PERMISSIONS FOR USER GROUPS !!
     #======================================================================
     # user group for user group permissions
     user_group_user_groups_perms = \
      Session().query(UserGroupUserGroupToPerm, Permission, UserGroup)\
      .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
             == UserGroup.users_group_id))\
      .join((Permission, UserGroupUserGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == user_id)\
      .join((UserGroup, UserGroupMember.users_group_id ==
             UserGroup.users_group_id), aliased=True, from_joinpoint=True)\
      .filter(UserGroup.users_group_active == True)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_group_user_groups_perms:
         g_k = perm.UserGroupUserGroupToPerm.target_user_group.users_group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[UK][g_k] = p
     #user explicit permission for user groups
     user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
     for perm in user_user_groups_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[UK][u_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[UK][u_k] = p
     return permissions
 def allowed_api_access(controller_name, whitelist=None, api_key=None):
     """
     Check if given controller_name is in whitelist API access
     """
     if not whitelist:
         from kallithea import CONFIG
         whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
                            sep=',')
         log.debug('whitelist of API access is: %s' % (whitelist))
     api_access_valid = controller_name in whitelist
     if api_access_valid:
         log.debug('controller:%s is in API whitelist' % (controller_name))
     else:
         msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
         if api_key:
             #if we use API key and don't have access it's a warning
             log.warning(msg)
         else:
             log.debug(msg)
     return api_access_valid
 class AuthUser(object):
     """
     Represents a Kallithea user, including various authentication and
     authorization information. Typically used to store the current user,
     but is also used as a generic user information data structure in
     parts of the code, e.g. user management.
     Constructed from user ID, API key or cookie dict, it looks
     up the matching database `User` and copies all attributes to itself,
     Constructed from a database `User` object, a user ID or cookie dict,
     it looks up the user (if needed) and copies all attributes to itself,
     adding various non-persistent data. If lookup fails but anonymous
     access to Kallithea is enabled, the default user is loaded instead.
     `AuthUser` does not by itself authenticate users and the constructor
     sets the `is_authenticated` field to False, except when falling back
     to the default anonymous user (if enabled). It's up to other parts
     of the code to check e.g. if a supplied password is correct, and if
     so, set `is_authenticated` to True.
     However, `AuthUser` does refuse to load a user that is not `active`.
     """
-    def __init__(self, user_id=None, api_key=None,
+    def __init__(self, user_id=None, dbuser=None,
             is_external_auth=False):
         self.is_authenticated = False
         self.is_external_auth = is_external_auth
         user_model = UserModel()
         self.anonymous_user = User.get_default_user(cache=True)
         is_user_loaded = False
         # These attributes will be overriden by fill_data, below, unless the
         # requested user cannot be found and the default anonymous user is
         # not enabled.
         self.user_id = None
         self.username = None
         self.api_key = None
         self.name = ''
         self.lastname = ''
         self.email = ''
         self.admin = False
         self.inherit_default_permissions = False
-        # lookup by userid
+        # Look up database user, if necessary.
         if user_id is not None:
             log.debug('Auth User lookup by USER ID %s' % user_id)
             is_user_loaded = user_model.fill_data(self, user_model.get(user_id))
             dbuser = user_model.get(user_id)
         else:
             # Note: dbuser is allowed to be None.
             log.debug('Auth User lookup by database user %s', dbuser)
         # try go get user by API key
         elif api_key:
             log.debug('Auth User lookup by API key %s' % api_key)
             is_user_loaded = user_model.fill_data(self, User.get_by_api_key(api_key))
         else:
             log.debug('No data in %s that could been used to log in' % self)
         is_user_loaded = user_model.fill_data(self, dbuser)
         # If user cannot be found, try falling back to anonymous.
         if not is_user_loaded:
             is_user_loaded =  user_model.fill_data(self, self.anonymous_user)
         # The anonymous user is always "logged in".
         if self.user_id == self.anonymous_user.user_id:
             self.is_authenticated = True
         if not self.username:
             self.username = 'None'
         log.debug('Auth User is now %s' % self)
     @LazyProperty
     def permissions(self):
         return self.__get_perms(user=self, cache=False)
     @property
     def api_keys(self):
         return self._get_api_keys()
     def __get_perms(self, user, explicit=True, algo='higherwin', cache=False):
         """
         Fills user permission attribute with permissions taken from database
         works for permissions given for repositories, and for permissions that
         are granted to groups
         :param user: `AuthUser` instance
         :param explicit: In case there are permissions both for user and a group
             that user is part of, explicit flag will define if user will
             explicitly override permissions from group, if it's False it will
             make decision based on the algo
         :param algo: algorithm to decide what permission should be choose if
             it's multiple defined, eg user in two different groups. It also
             decides if explicit flag is turned off how to specify the permission
             for case when user is in a group + have defined separate permission
         """
         user_id = user.user_id
         user_is_admin = user.is_admin
         user_inherit_default_permissions = user.inherit_default_permissions
         log.debug('Getting PERMISSION tree')
         compute = conditional_cache('short_term', 'cache_desc',
                                     condition=cache, func=_cached_perms_data)
         return compute(user_id, user_is_admin,
                        user_inherit_default_permissions, explicit, algo)
     def _get_api_keys(self):
         api_keys = [self.api_key]
         for api_key in UserApiKeys.query()\
                 .filter(UserApiKeys.user_id == self.user_id)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())).all():
             api_keys.append(api_key.api_key)
         return api_keys
     @property
     def is_admin(self):
         return self.admin
     @property
     def repositories_admin(self):
         """
         Returns list of repositories you're an admin of
         """
         return [x[0] for x in self.permissions['repositories'].iteritems()
                 if x[1] == 'repository.admin']
     @property
     def repository_groups_admin(self):
         """
         Returns list of repository groups you're an admin of
         """
         return [x[0] for x in self.permissions['repositories_groups'].iteritems()
                 if x[1] == 'group.admin']
     @property
     def user_groups_admin(self):
         """
         Returns list of user groups you're an admin of
         """
         return [x[0] for x in self.permissions['user_groups'].iteritems()
                 if x[1] == 'usergroup.admin']
     @staticmethod
     def check_ip_allowed(user, ip_addr):
         """
         Check if the given IP address (a `str`) is allowed for the given
         user (an `AuthUser` or `db.User`).
         """
         allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
             inherit_from_default=user.inherit_default_permissions)
         if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
             log.debug('IP:%s is in range of %s' % (ip_addr, allowed_ips))
             return True
         else:
             log.info('Access for IP:%s forbidden, '
                      'not in %s' % (ip_addr, allowed_ips))
             return False
     def __repr__(self):
         return "<AuthUser('id:%s[%s] auth:%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 to_cookie(self):
         """ Serializes this login session to a cookie `dict`. """
         return {
             'user_id': self.user_id,
             'is_authenticated': self.is_authenticated,
             'is_external_auth': self.is_external_auth,
+        }
     @staticmethod
     def from_cookie(cookie):
         """
         Deserializes an `AuthUser` from a cookie `dict`.
         """
         au = AuthUser(
             user_id=cookie.get('user_id'),
             is_external_auth=cookie.get('is_external_auth', False),
+        )
         if not au.is_authenticated and au.user_id is not None:
             # user is not authenticated and not empty
             au.set_authenticated(cookie.get('is_authenticated'))
         return au
     @classmethod
     def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):
         _set = set()
         if inherit_from_default:
             default_ips = UserIpMap.query().filter(UserIpMap.user ==
                                             User.get_default_user(cache=True))
             if cache:
                 default_ips = default_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_default"))
             # populate from default user
             for ip in default_ips:
                 try:
                     _set.add(ip.ip_addr)
                 except ObjectDeletedError:
                     # since we use heavy caching sometimes it happens that we get
                     # deleted objects here, we just skip them
                     pass
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         if cache:
             user_ips = user_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_%s" % user_id))
         for ip in user_ips:
             try:
                 _set.add(ip.ip_addr)
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 def set_available_permissions(config):
     """
     This function will propagate pylons globals with all available defined
     permission given in db. We don't want to check each time from db for new
     permissions since adding a new permission also requires application restart
     ie. to decorate new views with the newly created permission
     :param config: current pylons config instance
     """
     log.info('getting information about all available permissions')
     try:
         sa = meta.Session
         all_perms = sa.query(Permission).all()
         config['available_permissions'] = [x.permission_name for x in all_perms]
     finally:
         meta.Session.remove()
 #==============================================================================
 # CHECK DECORATORS
 #==============================================================================
 def redirect_to_login(message=None):
     from kallithea.lib import helpers as h
     p = url.current()
     if message:
         h.flash(h.literal(message), category='warning')
     log.debug('Redirecting to login page, origin: %s' % p)
     return redirect(url('login_home', came_from=p, **request.GET))
 class LoginRequired(object):
     """
     Must be logged in to execute this function else
     redirect to login page
     :param api_access: if enabled this checks only for valid auth token
         and grants access based on valid token
     """
     def __init__(self, api_access=False):
         self.api_access = api_access
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         controller = fargs[0]
         user = controller.authuser
         loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
         log.debug('Checking access for user %s @ %s' % (user, loc))
         if not AuthUser.check_ip_allowed(user, controller.ip_addr):
             return redirect_to_login(_('IP %s not allowed') % controller.ip_addr)
         # check if we used an API key and it's a valid one
         api_key = request.GET.get('api_key')
         if api_key is not None:
             # explicit controller is enabled or API is in our whitelist
             if self.api_access or allowed_api_access(loc, api_key=api_key):
                 if api_key in user.api_keys:
                     log.info('user %s authenticated with API key ****%s @ %s'
                              % (user, api_key[-4:], loc))
                     return func(*fargs, **fkwargs)
                 else:
                     log.warning('API key ****%s is NOT valid' % api_key[-4:])
                     return redirect_to_login(_('Invalid API key'))
             else:
                 # controller does not allow API access
                 log.warning('API access to %s is not allowed' % loc)
                 return abort(403)
         # CSRF protection - POSTs with session auth must contain correct token
         if request.POST and user.is_authenticated:
             token = request.POST.get(secure_form.token_key)
             if not token or token != secure_form.authentication_token():
                 log.error('CSRF check failed')
                 return abort(403)
         # regular user authentication
         if user.is_authenticated:
             log.info('user %s authenticated with regular auth @ %s' % (user, loc))
             return func(*fargs, **fkwargs)
         else:
             log.warning('user %s NOT authenticated with regular auth @ %s' % (user, loc))
             return redirect_to_login()
 class NotAnonymous(object):
     """
     Must be logged in to execute this function else
     redirect to login page"""
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         log.debug('Checking if user is not anonymous @%s' % cls)
         anonymous = self.user.username == User.DEFAULT_USER
         if anonymous:
             return redirect_to_login(_('You need to be a registered user to '
                     'perform this action'))
         else:
             return func(*fargs, **fkwargs)
 class PermsDecorator(object):
     """Base class for controller decorators"""
     def __init__(self, *required_perms):
         self.required_perms = set(required_perms)
         self.user_perms = None
     def __call__(self, func):
         return decorator(self.__wrapper, func)
     def __wrapper(self, func, *fargs, **fkwargs):
         cls = fargs[0]
         self.user = cls.authuser
         self.user_perms = self.user.permissions
         log.debug('checking %s permissions %s for %s %s',
            self.__class__.__name__, self.required_perms, cls, self.user)
         if self.check_permissions():
             log.debug('Permission granted for %s %s' % (cls, self.user))
             return func(*fargs, **fkwargs)
         else:
             log.debug('Permission denied for %s %s' % (cls, self.user))
             anonymous = self.user.username == User.DEFAULT_USER
             if anonymous:
                 return redirect_to_login(_('You need to be signed in to view this page'))
             else:
                 # redirect with forbidden ret code
                 return abort(403)
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
 class HasPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates. All of them
     have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         if self.required_perms.issubset(self.user_perms.get('global')):
             return True
         return False
 class HasPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates. In order to
     fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         if self.required_perms.intersection(self.user_perms.get('global')):
             return True
         return False
 class HasRepoPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates for specific
     repository. All of them have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         repo_name = get_repo_slug(request)
         try:
             user_perms = set([self.user_perms['repositories'][repo_name]])
         except KeyError:
             return False
         if self.required_perms.issubset(user_perms):
             return True
         return False
 class HasRepoPermissionAnyDecorator(PermsDecorator):
     """
     Checks for access permission for any of given predicates for specific
     repository. In order to fulfill the request any of predicates must be meet
     """
     def check_permissions(self):
         repo_name = get_repo_slug(request)
         try:
             user_perms = set([self.user_perms['repositories'][repo_name]])
         except KeyError:
             return False
         if self.required_perms.intersection(user_perms):
             return True
         return False
 class HasRepoGroupPermissionAllDecorator(PermsDecorator):
     """
     Checks for access permission for all given predicates for specific
     repository group. All of them have to be meet in order to fulfill the request
     """
     def check_permissions(self):
         group_name = get_repo_group_slug(request)
         try:
             user_perms = set([self.user_perms['repositories_groups'][group_name]])
         except KeyError:

kallithea/lib/base.py

➞

Show inline comments

 # -*- 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.lib.base
 ~~~~~~~~~~~~~~~~~~
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import time
 import traceback
 import webob.exc
 import paste.httpexceptions
 import paste.auth.basic
 import paste.httpheaders
 from pylons import config, tmpl_context as c, request, session, url
 from pylons.controllers import WSGIController
 from pylons.controllers.util import redirect
 from pylons.templating import render_mako as render  # don't remove this import
 from pylons.i18n.translation import _
 from kallithea import __version__, BACKENDS
 from kallithea.lib.utils2 import str2bool, safe_unicode, AttributeDict,\
     safe_str, safe_int
 from kallithea.lib import auth_modules
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.utils import get_repo_slug
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.vcs.exceptions import RepositoryError, EmptyRepositoryError, ChangesetDoesNotExistError
 from kallithea.model import meta
 from kallithea.model.db import Repository, Ui, User, Setting
 from kallithea.model.notification import NotificationModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.pull_request import PullRequestModel
 log = logging.getLogger(__name__)
 def _filter_proxy(ip):
     """
     HEADERS can have multiple ips inside the left-most being the original
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s' % (','.join(_ips), _first_ip))
         return _first_ip
     return ip
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     ip = environ.get(proxy_key)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(proxy_key2)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(def_key, '0.0.0.0')
     return _filter_proxy(ip)
 def _get_access_path(environ):
     path = environ.get('PATH_INFO')
     org_req = environ.get('pylons.original_request')
     if org_req:
         path = org_req.environ.get('PATH_INFO')
     return path
 def log_in_user(user, remember, is_external_auth):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     user.update_lastlogin()
     meta.Session().commit()
-    auth_user = AuthUser(user_id=user.user_id,
+    auth_user = AuthUser(dbuser=user,
                          is_external_auth=is_external_auth)
     auth_user.set_authenticated()
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
     session.save()
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
     # dumps session attrs back to cookie
     session._update_cookie_out()
     return auth_user
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
     def build_authentication(self):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication()
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication()
         auth = auth.strip().decode('base64')
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ):
                 return username
         return self.build_authentication()
     __call__ = authenticate
 class BaseVCSController(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         self.basepath = self.config['base_path']
         #authenticate this VCS request using authfunc
         self.authenticate = BasicAuth('', auth_modules.authenticate,
                                       config.get('auth_ret_code'))
         self.ip_addr = '0.0.0.0'
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     def _get_by_id(self, repo_name):
         """
         Gets a special pattern _<ID> from clone url and tries to replace it
         with a repository_name for support of _<ID> permanent URLs
         :param repo_name:
         """
         data = repo_name.split('/')
         if len(data) >= 2:
             from kallithea.lib.utils import get_repo_by_id
             by_id_match = get_repo_by_id(repo_name)
             if by_id_match:
                 data[1] = by_id_match
         return '/'.join(data)
     def _invalidate_cache(self, repo_name):
         """
         Sets cache for this repository for invalidation on next access
         :param repo_name: full repo name, also a cache key
         """
         ScmModel().mark_for_invalidation(repo_name)
     def _check_permission(self, action, user, repo_name, ip_addr=None):
         """
         Checks permissions using action (push/pull) user and repository
         name
         :param action: push or pull action
         :param user: `User` instance
         :param repo_name: repository name
         """
         # check IP
         ip_allowed = AuthUser.check_ip_allowed(user, ip_addr)
         if ip_allowed:
             log.info('Access for IP:%s allowed' % (ip_addr,))
         else:
             return False
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         else:
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(user,
                                                                   repo_name):
                 return False
         return True
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def _check_ssl(self, environ):
         """
         Checks the SSL check flag and returns False if SSL is not present
         and required True otherwise
         """
         #check if we have SSL required  ! if not it's a bad request !
         if str2bool(Ui.get_by_key('push_ssl').ui_value):
             org_proto = environ.get('wsgi._org_proto', environ['wsgi.url_scheme'])
             if org_proto != 'https':
                 log.debug('proto is %s and SSL is required BAD REQUEST !'
                           % org_proto)
                 return False
         return True
     def _check_locking_state(self, environ, action, repo, user_id):
         """
         Checks locking on this repository, if locking is enabled and lock is
         present returns a tuple of make_lock, locked, locked_by.
         make_lock can have 3 states None (do nothing) True, make lock
         False release lock, This value is later propagated to hooks, which
         do the locking. Think about this as signals passed to hooks what to do.
         """
         locked = False  # defines that locked error should be thrown to user
         make_lock = None
         repo = Repository.get_by_repo_name(repo)
         user = User.get(user_id)
         # this is kind of hacky, but due to how mercurial handles client-server
         # server see all operation on changeset; bookmarks, phases and
         # obsolescence marker in different transaction, we don't want to check
         # locking on those
         obsolete_call = environ['QUERY_STRING'] in ['cmd=listkeys',]
         locked_by = repo.locked
         if repo and repo.enable_locking and not obsolete_call:
             if action == 'push':
                 #check if it's already locked !, if it is compare users
                 user_id, _date = repo.locked
                 if user.user_id == user_id:
                     log.debug('Got push from user %s, now unlocking' % (user))
                     # unlock if we have push from user who locked
                     make_lock = False
                 else:
                     # we're not the same user who locked, ban with 423 !
                     locked = True
             if action == 'pull':
                 if repo.locked[0] and repo.locked[1]:
                     locked = True
                 else:
                     log.debug('Setting lock on repo %s by %s' % (repo, user))
                     make_lock = True
         else:
             log.debug('Repository %s do not have locking enabled' % (repo))
         log.debug('FINAL locking values make_lock:%s,locked:%s,locked_by:%s'
                   % (make_lock, locked, locked_by))
         return make_lock, locked, locked_by
     def __call__(self, environ, start_response):
         start = time.time()
         try:
             return self._handle_request(environ, start_response)
         finally:
             log = logging.getLogger('kallithea.' + self.__class__.__name__)
             log.debug('Request time: %.3fs' % (time.time() - start))
             meta.Session.remove()
 class BaseController(WSGIController):
     def __before__(self):
         """
         __before__ is called before controller methods and after __call__
         """
         c.kallithea_version = __version__
         rc_config = Setting.get_app_settings()
         # Visual options
         c.visual = AttributeDict({})
         ## DB stored
         c.visual.show_public_icon = str2bool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = str2bool(rc_config.get('show_private_icon'))
         c.visual.stylify_metatags = str2bool(rc_config.get('stylify_metatags'))
         c.visual.dashboard_items = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = str2bool(rc_config.get('repository_fields'))
         c.visual.show_version = str2bool(rc_config.get('show_version'))
         c.visual.use_gravatar = str2bool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl')
         ## INI stored
         c.visual.allow_repo_location_change = str2bool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = str2bool(config.get('allow_custom_hooks_settings', True))
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = BACKENDS.keys()
         c.unread_notifications = NotificationModel()\
                         .get_unread_cnt_for_user(c.authuser.user_id)
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequestModel().get_pullrequest_cnt_for_user(c.authuser.user_id)
         self.sa = meta.Session
         self.scm_model = ScmModel(self.sa)
     @staticmethod
     def _determine_auth_user(api_key, session_authuser):
         """
         Create an `AuthUser` object given the API key (if any) and the
         value of the authuser session cookie.
         """
         # Authenticate by API key
         if api_key:
             # when using API_KEY we are sure user exists.
             return AuthUser(api_key=api_key, is_external_auth=True)
             return AuthUser(dbuser=User.get_by_api_key(api_key),
                             is_external_auth=True)
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         if isinstance(session_authuser, dict):
             try:
                 return AuthUser.from_cookie(session_authuser)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw UserCreationError to signal issues with
                 # user creation. Explanation should be provided in the
                 # exception object.
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
         # Authenticate by auth_container plugin (if enabled)
         if any(
             auth_modules.importplugin(name).is_container_auth
             for name in Setting.get_auth_plugins()
         ):
             try:
                 auth_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if auth_info:
                     username = auth_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False,
                                        is_external_auth=True)
         # User is anonymous
         return AuthUser()
     def __call__(self, environ, start_response):
         """Invoke the Controller"""
         # WSGIController.__call__ dispatches to the Controller method
         # the request is routed to. This routing information is
         # available in environ['pylons.routes_dict']
         try:
             self.ip_addr = _get_ip_addr(environ)
             # make sure that we update permissions each time we call controller
             #set globals for auth user
             self.authuser = c.authuser = request.user = self._determine_auth_user(
                 request.GET.get('api_key'),
                 session.get('authuser'),
+            )
             log.info('IP: %s User: %s accessed %s',
                 self.ip_addr, self.authuser,
                 safe_unicode(_get_access_path(environ)),
+            )
             return WSGIController.__call__(self, environ, start_response)
         finally:
             meta.Session.remove()
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
     c.db_repo_scm_instance: instance of scm repository
     c.db_repo: instance of db
     c.repository_followers: number of followers
     c.repository_forks: number of forks
     c.repository_following: weather the current user is following the current repo
     """
     def __before__(self):
         super(BaseRepoController, self).__before__()
         if c.repo_name:  # extracted from routes
             _dbr = Repository.get_by_repo_name(c.repo_name)
             if not _dbr:
                 return
             log.debug('Found repository in database %s with state `%s`'
                       % (safe_unicode(_dbr), safe_unicode(_dbr.repo_state)))
             route = getattr(request.environ.get('routes.route'), 'name', '')
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
             if _dbr.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 return redirect(check_url)
             dbr = c.db_repo = _dbr
             c.db_repo_scm_instance = c.db_repo.scm_instance
             if c.db_repo_scm_instance is None:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
                 from kallithea.lib import helpers as h
                 h.flash(h.literal(_('Repository not found in the filesystem')),
                         category='error')
                 raise paste.httpexceptions.HTTPNotFound()
             # some globals counter for menu
             c.repository_followers = self.scm_model.get_followers(dbr)
             c.repository_forks = self.scm_model.get_forks(dbr)
             c.repository_pull_requests = self.scm_model.get_pull_requests(dbr)
             c.repository_following = self.scm_model.is_following_repo(
                                     c.repo_name, self.authuser.user_id)
     @staticmethod
     def _get_ref_rev(repo, ref_type, ref_name, returnempty=False):
         """
         Safe way to get changeset. If error occurs show error.
         """
         from kallithea.lib import helpers as h
         try:
             return repo.scm_instance.get_ref_revision(ref_type, ref_name)
         except EmptyRepositoryError as e:
             if returnempty:
                 return repo.scm_instance.EMPTY_CHANGESET
             h.flash(h.literal(_('There are no changesets yet')),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except ChangesetDoesNotExistError as e:
             h.flash(h.literal(_('Changeset not found')),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except RepositoryError as e:
             log.error(traceback.format_exc())
             h.flash(safe_str(e), category='error')
             raise webob.exc.HTTPBadRequest()

kallithea/model/db.py

➞

Show inline comments

@@ @@ -130,769 +130,769 @@ class BaseModel(object): @@
             raise HTTPNotFound
         res = cls.query().get(id_)
         if res is None:
             raise HTTPNotFound
         return res
     @classmethod
     def getAll(cls):
         # deprecated and left for backward compatibility
         return cls.get_all()
     @classmethod
     def get_all(cls):
         return cls.query().all()
     @classmethod
     def delete(cls, id_):
         obj = cls.query().get(id_)
         Session().delete(obj)
     def __repr__(self):
         if hasattr(self, '__unicode__'):
             # python repr needs to return str
             try:
                 return safe_str(self.__unicode__())
             except UnicodeDecodeError:
                 pass
         return '<DB:%s>' % (self.__class__.__name__)
 class Setting(Base, BaseModel):
     __tablename__ = DB_PREFIX + 'settings'
     __table_args__ = (
         UniqueConstraint('app_settings_name'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     SETTINGS_TYPES = {
         'str': safe_str,
         'int': safe_int,
         'unicode': safe_unicode,
         'bool': str2bool,
         'list': functools.partial(aslist, sep=',')
+    }
     DEFAULT_UPDATE_URL = ''
     app_settings_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     app_settings_name = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     _app_settings_value = Column("app_settings_value", String(4096, convert_unicode=False), nullable=True, unique=None, default=None)
     _app_settings_type = Column("app_settings_type", String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     def __init__(self, key='', val='', type='unicode'):
         self.app_settings_name = key
         self.app_settings_value = val
         self.app_settings_type = type
     @validates('_app_settings_value')
     def validate_settings_value(self, key, val):
         assert type(val) == unicode
         return val
     @hybrid_property
     def app_settings_value(self):
         v = self._app_settings_value
         _type = self.app_settings_type
         converter = self.SETTINGS_TYPES.get(_type) or self.SETTINGS_TYPES['unicode']
         return converter(v)
     @app_settings_value.setter
     def app_settings_value(self, val):
         """
         Setter that will always make sure we use unicode in app_settings_value
         :param val:
         """
         self._app_settings_value = safe_unicode(val)
     @hybrid_property
     def app_settings_type(self):
         return self._app_settings_type
     @app_settings_type.setter
     def app_settings_type(self, val):
         if val not in self.SETTINGS_TYPES:
             raise Exception('type must be one of %s got %s'
                             % (self.SETTINGS_TYPES.keys(), val))
         self._app_settings_type = val
     def __unicode__(self):
         return u"<%s('%s:%s[%s]')>" % (
             self.__class__.__name__,
             self.app_settings_name, self.app_settings_value, self.app_settings_type
+        )
     @classmethod
     def get_by_name(cls, key):
         return cls.query()\
             .filter(cls.app_settings_name == key).scalar()
     @classmethod
     def get_by_name_or_create(cls, key, val='', type='unicode'):
         res = cls.get_by_name(key)
         if res is None:
             res = cls(key, val, type)
         return res
     @classmethod
     def create_or_update(cls, key, val=Optional(''), type=Optional('unicode')):
         """
         Creates or updates Kallithea setting. If updates are triggered, it will only
         update parameters that are explicitly set. Optional instance will be skipped.
         :param key:
         :param val:
         :param type:
         :return:
         """
         res = cls.get_by_name(key)
         if res is None:
             val = Optional.extract(val)
             type = Optional.extract(type)
             res = cls(key, val, type)
         else:
             res.app_settings_name = key
             if not isinstance(val, Optional):
                 # update if set
                 res.app_settings_value = val
             if not isinstance(type, Optional):
                 # update if set
                 res.app_settings_type = type
         return res
     @classmethod
     def get_app_settings(cls, cache=False):
         ret = cls.query()
         if cache:
             ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
         if ret is None:
             raise Exception('Could not get application settings !')
         settings = {}
         for each in ret:
             settings[each.app_settings_name] = \
                 each.app_settings_value
         return settings
     @classmethod
     def get_auth_plugins(cls, cache=False):
         auth_plugins = cls.get_by_name("auth_plugins").app_settings_value
         return auth_plugins
     @classmethod
     def get_auth_settings(cls, cache=False):
         ret = cls.query()\
                 .filter(cls.app_settings_name.startswith('auth_')).all()
         fd = {}
         for row in ret:
             fd[row.app_settings_name] = row.app_settings_value
         return fd
     @classmethod
     def get_default_repo_settings(cls, cache=False, strip_prefix=False):
         ret = cls.query()\
                 .filter(cls.app_settings_name.startswith('default_')).all()
         fd = {}
         for row in ret:
             key = row.app_settings_name
             if strip_prefix:
                 key = remove_prefix(key, prefix='default_')
             fd.update({key: row.app_settings_value})
         return fd
     @classmethod
     def get_server_info(cls):
         import pkg_resources
         import platform
         import kallithea
         from kallithea.lib.utils import check_git_version
         mods = [(p.project_name, p.version) for p in pkg_resources.working_set]
         info = {
             'modules': sorted(mods, key=lambda k: k[0].lower()),
             'py_version': platform.python_version(),
             'platform': safe_unicode(platform.platform()),
             'kallithea_version': kallithea.__version__,
             'git_version': safe_unicode(check_git_version()),
             'git_path': kallithea.CONFIG.get('git_path')
+        }
         return info
 class Ui(Base, BaseModel):
     __tablename__ = DB_PREFIX + 'ui'
     __table_args__ = (
         UniqueConstraint('ui_key'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     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(Integer(), nullable=False, unique=True, primary_key=True)
     ui_section = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     ui_key = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     ui_value = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     ui_active = Column(Boolean(), nullable=True, unique=None, default=True)
     # def __init__(self, section='', key='', value=''):
     #     self.ui_section = section
     #     self.ui_key = key
     #     self.ui_value = value
     @classmethod
     def get_by_key(cls, key):
         return cls.query().filter(cls.ui_key == key).scalar()
     @classmethod
     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()
     @classmethod
     def get_custom_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]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_repos_location(cls):
         return cls.get_by_key('/').ui_value
     @classmethod
     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)
     def __repr__(self):
         return '<%s[%s]%s=>%s]>' % (self.__class__.__name__, self.ui_section,
                                     self.ui_key, self.ui_value)
 class User(Base, BaseModel):
     __tablename__ = 'users'
     __table_args__ = (
         UniqueConstraint('username'), UniqueConstraint('email'),
         Index('u_username_idx', 'username'),
         Index('u_email_idx', 'email'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     DEFAULT_USER = 'default'
     DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'
     user_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     username = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     password = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     active = Column(Boolean(), nullable=True, unique=None, default=True)
     admin = Column(Boolean(), nullable=True, unique=None, default=False)
     name = Column("firstname", String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     lastname = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     _email = Column("email", String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     last_login = Column(DateTime(timezone=False), nullable=True, unique=None, default=None)
     extern_type = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     extern_name = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     api_key = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     inherit_default_permissions = Column(Boolean(), nullable=False, unique=None, default=True)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data
     user_log = relationship('UserLog')
     user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
     repositories = relationship('Repository')
     repo_groups = relationship('RepoGroup')
     user_groups = relationship('UserGroup')
     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('UserGroupMember', 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')
     #extra API keys
     user_api_keys = relationship('UserApiKeys', cascade='all')
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
     @property
     def firstname(self):
         # alias for future
         return self.name
     @property
     def emails(self):
         other = UserEmailMap.query().filter(UserEmailMap.user==self).all()
         return [self.email] + [x.email for x in other]
     @property
     def api_keys(self):
         other = UserApiKeys.query().filter(UserApiKeys.user==self).all()
         return [self.api_key] + [x.api_key for x in other]
     @property
     def ip_addresses(self):
         ret = UserIpMap.query().filter(UserIpMap.user == self).all()
         return [x.ip_addr for x in ret]
     @property
     def full_name(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def full_name_or_username(self):
         """
         Show full name.
         If full name is not set, fall back to username.
         """
         return ('%s %s' % (self.firstname, self.lastname)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_name_and_username(self):
         """
         Show full name and username as 'Firstname Lastname (username)'.
         If full name is not set, fall back to username.
         """
         return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_contact(self):
         return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
     @property
     def short_contact(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
-        return AuthUser(user_id=self.user_id)
+        return AuthUser(dbuser=self)
     @hybrid_property
     def user_data(self):
         if not self._user_data:
             return {}
         try:
             return json.loads(self._user_data)
         except TypeError:
             return {}
     @user_data.setter
     def user_data(self, val):
         try:
             self._user_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.user_id, self.username)
     @classmethod
     def get_or_404(cls, id_, allow_default=True):
         '''
         Overridden version of BaseModel.get_or_404, with an extra check on
         the default user.
         '''
         user = super(User, cls).get_or_404(id_)
         if allow_default == False:
             if user.username == User.DEFAULT_USER:
                 raise DefaultUserException
         return user
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.username.ilike(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_user_%s" % _hash_key(username)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False, fallback=True):
         if len(api_key) != 40 or not api_key.isalnum():
             return None
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         res = q.scalar()
         if fallback and not res:
             #fallback to additional keys
             _res = UserApiKeys.query()\
                 .filter(UserApiKeys.api_key == api_key)\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time()))\
                 .first()
             if _res:
                 res = _res.user
         return res
     @classmethod
     def get_by_email(cls, email, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(cls.email.ilike(email))
         else:
             q = cls.query().filter(cls.email == email)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_email_key_%s" % email))
         ret = q.scalar()
         if ret is None:
             q = UserEmailMap.query()
             # try fetching in alternate email map
             if case_insensitive:
                 q = q.filter(UserEmailMap.email.ilike(email))
             else:
                 q = q.filter(UserEmailMap.email == email)
             q = q.options(joinedload(UserEmailMap.user))
             if cache:
                 q = q.options(FromCache("sql_cache_short",
                                         "get_email_map_key_%s" % email))
             ret = getattr(q.scalar(), 'user', None)
         return ret
     @classmethod
     def get_from_cs_author(cls, author):
         """
         Tries to get User objects out of commit author string
         :param author:
         """
         from kallithea.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 is not None:
                 return user
         # Maybe we can match by username?
         _author = author_name(author)
         user = cls.get_by_username(_author, case_insensitive=True)
         if user is not None:
             return user
     def update_lastlogin(self):
         """Update user lastlogin"""
         self.last_login = datetime.datetime.now()
         Session().add(self)
         log.debug('updated user %s lastlogin' % self.username)
     @classmethod
     def get_first_admin(cls):
         user = User.query().filter(User.admin == True).first()
         if user is None:
             raise Exception('Missing administrative account!')
         return user
     @classmethod
     def get_default_user(cls, cache=False):
         user = User.get_by_username(User.DEFAULT_USER, cache=cache)
         if user is None:
             raise Exception('Missing default account!')
         return user
     def get_api_data(self, details=False):
         """
         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,
             active=user.active,
             admin=user.admin,
+        )
         if details:
             data.update(dict(
                 extern_type=user.extern_type,
                 extern_name=user.extern_name,
                 api_key=user.api_key,
                 api_keys=user.api_keys,
                 last_login=user.last_login,
                 ip_addresses=user.ip_addresses
                 ))
         return data
     def __json__(self):
         data = dict(
             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())
         return data
 class UserApiKeys(Base, BaseModel):
     __tablename__ = 'user_api_keys'
     __table_args__ = (
         Index('uak_api_key_idx', 'api_key'),
         Index('uak_api_key_expires_idx', 'api_key', 'expires'),
         UniqueConstraint('api_key'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     __mapper_args__ = {}
     user_api_key_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
     api_key = Column(String(255, convert_unicode=False), nullable=False, unique=True)
     description = Column(UnicodeText(1024))
     expires = Column(Float(53), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     user = relationship('User')
     @property
     def expired(self):
         if self.expires == -1:
             return False
         return time.time() > self.expires
 class UserEmailMap(Base, BaseModel):
     __tablename__ = 'user_email_map'
     __table_args__ = (
         Index('uem_email_idx', 'email'),
         UniqueConstraint('email'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     __mapper_args__ = {}
     email_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
     _email = Column("email", String(255, convert_unicode=False), nullable=True, unique=False, default=None)
     user = relationship('User')
     @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
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
 class UserIpMap(Base, BaseModel):
     __tablename__ = 'user_ip_map'
     __table_args__ = (
         UniqueConstraint('user_id', 'ip_addr'),
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True}
+    )
     __mapper_args__ = {}
     ip_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
     ip_addr = Column(String(255, convert_unicode=False), nullable=True, unique=False, default=None)
     active = Column(Boolean(), nullable=True, unique=None, default=True)
     user = relationship('User')
     @classmethod
     def _get_ip_range(cls, ip_addr):
         from kallithea.lib import ipaddr
         net = ipaddr.IPNetwork(address=ip_addr)
         return [str(net.network), str(net.broadcast)]
     def __json__(self):
         return dict(
           ip_addr=self.ip_addr,
           ip_range=self._get_ip_range(self.ip_addr)
+        )
     def __unicode__(self):
         return u"<%s('user_id:%s=>%s')>" % (self.__class__.__name__,
                                             self.user_id, self.ip_addr)
 class UserLog(Base, BaseModel):
     __tablename__ = 'user_logs'
     __table_args__ = (
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True},
+    )
     user_log_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
     username = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     repository_name = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     user_ip = Column(String(255, convert_unicode=False), nullable=True, unique=None, default=None)
     action = Column(UnicodeText(1200000, convert_unicode=False), nullable=True, unique=None, default=None)
     action_date = Column(DateTime(timezone=False), nullable=True, unique=None, default=None)
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.repository_name,
                                       self.action)
     @property
     def action_as_day(self):
         return datetime.date(*self.action_date.timetuple()[:3])
     user = relationship('User')
     repository = relationship('Repository', cascade='')
 class UserGroup(Base, BaseModel):
     __tablename__ = 'users_groups'
     __table_args__ = (
         {'extend_existing': True, 'mysql_engine': 'InnoDB',
          'mysql_charset': 'utf8', 'sqlite_autoincrement': True},
+    )
     users_group_id = Column(Integer(), nullable=False, unique=True, primary_key=True)
     users_group_name = Column(String(255, convert_unicode=False), nullable=False, unique=True, default=None)
     user_group_description = Column(String(10000, convert_unicode=False), nullable=True, unique=None, default=None)
     users_group_active = Column(Boolean(), nullable=True, unique=None, default=None)
     inherit_default_permissions = Column("users_group_inherit_default_permissions", Boolean(), nullable=False, unique=None, default=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False, unique=False, default=None)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _group_data = Column("group_data", LargeBinary(), nullable=True)  # JSON data
     members = relationship('UserGroupMember', cascade="all, delete-orphan")
     users_group_to_perm = relationship('UserGroupToPerm', cascade='all')
     users_group_repo_to_perm = relationship('UserGroupRepoToPerm', cascade='all')
     users_group_repo_group_to_perm = relationship('UserGroupRepoGroupToPerm', cascade='all')
     user_user_group_to_perm = relationship('UserUserGroupToPerm ', cascade='all')
     user_group_user_group_to_perm = relationship('UserGroupUserGroupToPerm ', primaryjoin="UserGroupUserGroupToPerm.target_user_group_id==UserGroup.users_group_id", cascade='all')
     user = relationship('User')
     @hybrid_property
     def group_data(self):
         if not self._group_data:
             return {}
         try:
             return json.loads(self._group_data)
         except TypeError:
             return {}
     @group_data.setter
     def group_data(self, val):
         try:
             self._group_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.users_group_id,
                                       self.users_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, cache=False,
                           case_insensitive=False):
         if case_insensitive:
             q = cls.query().filter(cls.users_group_name.ilike(group_name))
         else:
             q = cls.query().filter(cls.users_group_name == group_name)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_group_%s" % _hash_key(group_name)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get(cls, user_group_id, cache=False):
         user_group = cls.query()
         if cache:
             user_group = user_group.options(FromCache("sql_cache_short",
                                     "get_users_group_%s" % user_group_id))
         return user_group.get(user_group_id)
     def get_api_data(self, with_members=True):
         user_group = self
         data = dict(
             users_group_id=user_group.users_group_id,
             group_name=user_group.users_group_name,
             group_description=user_group.user_group_description,
             active=user_group.users_group_active,
             owner=user_group.user.username,
+        )
         if with_members:
             members = []
             for user in user_group.members:
                 user = user.user
                 members.append(user.get_api_data())
             data['members'] = members
         return data

kallithea/tests/api/api_base.py

➞

Show inline comments

 # -*- 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/>.
 """
 tests for api. run with::
     KALLITHEA_WHOOSH_TEST_DISABLE=1 nosetests --with-coverage --cover-package=kallithea.controllers.api.api -x kallithea/tests/api
 """
 from __future__ import with_statement
 import os
 import random
 import mock
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.compat import json
 from kallithea.lib.auth import AuthUser
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.meta import Session
 from kallithea.model.scm import ScmModel
 from kallithea.model.gist import GistModel
 from kallithea.model.db import Repository, User, Setting
 from kallithea.lib.utils2 import time_to_datetime
 API_URL = '/_admin/api'
 TEST_USER_GROUP = 'test_user_group'
 TEST_REPO_GROUP = 'test_repo_group'
 fixture = Fixture()
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     :param random_id:
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 jsonify = lambda obj: json.loads(json.dumps(obj))
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 def api_call(test_obj, params):
     response = test_obj.app.post(API_URL, content_type='application/json',
                                  params=params)
     return response
 ## helpers
 def make_user_group(name=TEST_USER_GROUP):
     gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     UserGroupModel().add_user_to_group(user_group=gr,
                                        user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 def make_repo_group(name=TEST_REPO_GROUP):
     gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 class _BaseTestApi(object):
     REPO = None
     REPO_TYPE = None
     @classmethod
     def setup_class(cls):
         cls.usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@example.com',
             firstname='first',
             lastname='last'
+        )
         Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
     @classmethod
     def teardown_class(cls):
         pass
     def setUp(self):
         self.maxDiff = None
         make_user_group()
         make_repo_group()
     def tearDown(self):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
     def test_Optional_object(self):
         from kallithea.controllers.api.api import Optional
         option1 = Optional(None)
         self.assertEqual('<Optional:%s>' % None, repr(option1))
         self.assertEqual(option1(), None)
         self.assertEqual(1, Optional.extract(Optional(1)))
         self.assertEqual('trololo', Optional.extract('trololo'))
     def test_Optional_OAttr(self):
         from kallithea.controllers.api.api import Optional, OAttr
         option1 = Optional(OAttr('apiuser'))
         self.assertEqual('apiuser', Optional.extract(option1))
     def test_OAttr_object(self):
         from kallithea.controllers.api.api import OAttr
         oattr1 = OAttr('apiuser')
         self.assertEqual('<OptionalAttr:apiuser>', repr(oattr1))
         self.assertEqual(oattr1(), oattr1)
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
         expected = 'Invalid API key'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_args_is_null(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
     def test_api_args_is_bad(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
     def test_api_args_different_args(self):
         import string
         expected = {
             'ascii_letters': string.ascii_letters,
             'ws': string.whitespace,
             'printables': string.printable
+        }
         id_, params = _build_data(self.apikey, 'test', args=expected)
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
         self._compare_ok(id_, expected, response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         response = api_call(self, params)
         ret_all = []
         _users = User.query().filter(User.username != User.DEFAULT_USER) \
             .order_by(User.username).all()
         for usr in _users:
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
-        ret['permissions'] = AuthUser(usr.user_id).permissions
+        ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_pull(self):
         repo_name = 'test_pull'
         r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         r.clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)
         Session.add(r)
         Session.commit()
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_pull_error(self):
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO, )
         response = api_call(self, params)
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'repo_scan', crash)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos', )
         response = api_call(self, params)
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached()  # seed cache
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = {
             'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),
             'repository': self.REPO
+        }
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)
     def test_api_invalidate_cache_error(self):
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = 'Error occurred during cache invalidation action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache_regular_user_no_permission(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached() # seed cache
         id_, params = _build_data(self.apikey_regular, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = "repository `%s` does not exist" % (self.REPO,)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO, 'locked': True,
             'locked_since': response.json['result']['locked_since'],
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_by_non_admin(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': True,
                 'locked_since': response.json['result']['locked_since'],
                 'locked_by': self.TEST_USER_LOGIN,
                 'lock_state_changed': True,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (self.TEST_USER_LOGIN, repo_name, True))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       userid=TEST_USER_ADMIN_LOGIN,
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = 'userid is not the same as your user'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self):
         id_, params = _build_data(self.apikey_regular, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_release(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=False)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO,
             'locked': False,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, False))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_optional_userid(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         time_ = response.json['result']['locked_since']
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': time_,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_optional_locked(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO)
         response = api_call(self, params)
         time_ = response.json['result']['locked_since']
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': time_,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': False,
             'msg': ('Repo `%s` locked by `%s` on `%s`.'
                     % (self.REPO, TEST_USER_ADMIN_LOGIN,
                        json.dumps(time_to_datetime(time_))))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_optional_not_locked(self):
         repo_name = 'api_not_locked'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         self.assertEqual(repo.locked, [None, None])
         try:
             id_, params = _build_data(self.apikey, 'lock',
                                       repoid=repo.repo_id)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': False,
                 'locked_since': None,
                 'locked_by': None,
                 'lock_state_changed': False,
                 'msg': ('Repo `%s` not locked.' % (repo_name,))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'Error occurred locking repository `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks(self):
         id_, params = _build_data(self.apikey, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_one_locked_repo(self):
         repo_name = 'api_delete_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                                    cur_user=self.TEST_USER_LOGIN)
         Repository.lock(repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
         try:
             id_, params = _build_data(self.apikey, 'get_locks')
             response = api_call(self, params)
             expected = [repo.get_api_data()]
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_get_locks_with_one_locked_repo_for_specific_user(self):
         repo_name = 'api_delete_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                                    cur_user=self.TEST_USER_LOGIN)
         Repository.lock(repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
         try:
             id_, params = _build_data(self.apikey, 'get_locks',
                                       userid=self.TEST_USER_LOGIN)
             response = api_call(self, params)
             expected = [repo.get_api_data()]
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_get_locks_with_userid(self):
         id_, params = _build_data(self.apikey, 'get_locks',
                                   userid=TEST_USER_REGULAR_LOGIN)
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN,
                                   email='test@foo.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN + 'new',
                                   email=TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email, extern_name='internal')
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@example.com',

0 comments (0 inline, 0 general)