kallithea Changeset - 3563c47e52fd

Changeset - 3563c47e52fd

Parent rev.

Child rev.

[Not reviewed]

beta

0 4 0

Marcin Kuzminski - 13 years ago 2013-01-13 22:55:56
marcin@python-works.com

Implemented API calls for non-admin users for locking/unlocking repositories

4 files changed with 163 insertions and 17 deletions:

docs/api/api.rst

rhodecode/controllers/api/api.py

rhodecode/lib/auth.py

103

rhodecode/tests/api/api_base.py

0 comments (0 inline, 0 general)

docs/api/api.rst

➞

Show inline comments

 .. _api:
 ===
 API
 ===
 Starting from RhodeCode version 1.2 a simple API was implemented.
 There's a single schema for calling all api methods. API is implemented
 with JSON protocol both ways. An url to send API request to RhodeCode is
 <your_server>/_admin/api
 API ACCESS FOR WEB VIEWS
 ++++++++++++++++++++++++
 API access can also be turned on for each web view in RhodeCode that is
 decorated with `@LoginRequired` decorator. To enable API access simple change
 the standard login decorator to `@LoginRequired(api_access=True)`.
 After this change, a rhodecode view can be accessed without login by adding a
 GET parameter `?api_key=<api_key>` to url. By default this is only
 enabled on RSS/ATOM feed views.
 API ACCESS
 ++++++++++
 All clients are required to send JSON-RPC spec JSON data::
+    {
         "id:"<id>",
         "api_key":"<api_key>",
         "method":"<method_name>",
         "args":{"<arg_key>":"<arg_val>"}
+    }
 Example call for autopulling remotes repos using curl::
     curl https://server.com/_admin/api -X POST -H 'content-type:text/plain' --data-binary '{"id":1,"api_key":"xe7cdb2v278e4evbdf5vs04v832v0efvcbcve4a3","method":"pull","args":{"repo":"CPython"}}'
 Simply provide
  - *id* A value of any type, which is used to match the response with the request that it is replying to.
  - *api_key* for access and permission validation.
  - *method* is name of method to call
  - *args* is an key:value list of arguments to pass to method
 .. note::
     api_key can be found in your user account page
 RhodeCode API will return always a JSON-RPC response::
+    {
         "id":<id>, # matching id sent by request
         "result": "<result>"|null, # JSON formatted result, null if any errors
         "error": "null"|<error_message> # JSON formatted error (if any)
+    }
 All responses from API will be `HTTP/1.0 200 OK`, if there's an error while
 calling api *error* key from response will contain failure description
 and result will be null.
 API CLIENT
 ++++++++++
 From version 1.4 RhodeCode adds a script that allows to easily
 communicate with API. After installing RhodeCode a `rhodecode-api` script
 will be available.
 To get started quickly simply run::
   rhodecode-api _create_config --apikey=<youapikey> --apihost=<rhodecode host>
 This will create a file named .config in the directory you executed it storing
 json config file with credentials. You can skip this step and always provide
 both of the arguments to be able to communicate with server
 after that simply run any api command for example get_repo::
  rhodecode-api get_repo
  calling {"api_key": "<apikey>", "id": 75, "args": {}, "method": "get_repo"} to http://127.0.0.1:5000
  rhodecode said:
  {'error': 'Missing non optional `repoid` arg in JSON DATA',
   'id': 75,
   'result': None}
 Ups looks like we forgot to add an argument
 Let's try again now giving the repoid as parameters::
     rhodecode-api get_repo repoid:rhodecode
     calling {"api_key": "<apikey>", "id": 39, "args": {"repoid": "rhodecode"}, "method": "get_repo"} to http://127.0.0.1:5000
     rhodecode said:
     {'error': None,
      'id': 39,
      'result': <json data...>}
 API METHODS
 +++++++++++
 pull
 ----
 Pulls given repo from remote location. Can be used to automatically keep
 remote repos up to date. This command can be executed only using api_key
 belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "pull"
     args :    {
                 "repoid" : "<reponame or repo_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "Pulled from `<reponame>`"
     error :  null
 rescan_repos
 ------------
 Dispatch rescan repositories action. If remove_obsolete is set
 RhodeCode will delete repos that are in database but not in the filesystem.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "rescan_repos"
     args :    {
                 "remove_obsolete" : "<boolean = Optional(False)>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "{'added': [<list of names of added repos>],
                'removed': [<list of names of removed repos>]}"
     error :  null
 lock
 ----
 Set locking state on given repository by given user.
 Set locking state on given repository by given user. If userid param is skipped
 , then it is set to id of user whos calling this method.
 This command can be executed only using api_key belonging to user with admin
 rights.
+rights or regular user that have admin or write access to repository.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "lock"
     args :    {
                 "repoid" : "<reponame or repo_id>"
                 "userid" : "<user_id or username>",
+                "userid" : "<user_id or username = Optional(=apiuser)>",
                 "locked" : "<bool true|false>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : "User `<username>` set lock state for repo `<reponame>` to `true|false`"
     error :  null
 show_ip
 -------
 Shows IP address as seen from RhodeCode server, together with all
 defined IP addresses for given user.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "show_ip"
     args :    {
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : {
                  "ip_addr_server": <ip_from_clien>",
                  "user_ips": [
+                                {
                                    "ip_addr": "<ip_with_mask>",
                                    "ip_range": ["<start_ip>", "<end_ip>"],
                                 },
                                 ...
+                             ]
+             }
     error :  null
 get_user
 --------
 Get's an user by username or user_id, Returns empty result if user is not found.
 This command can be executed only using api_key belonging to user with admin
 rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_user"
     args :    {
                 "userid" : "<username or user_id>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: None if user does not exist or
+            {
                 "user_id" :     "<user_id>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
                 "permissions": {
                     "global": ["hg.create.repository",
                                "repository.read",
                                "hg.register.manual_activate"],
                     "repositories": {"repo1": "repository.none"},
                     "repositories_groups": {"Group1": "group.read"}
                  },
+            }
     error:  null
 get_users
 ---------
 Lists all existing users. This command can be executed only using api_key
 belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_users"
     args :    { }
 OUTPUT::
     id : <id_given_in_input>
     result: [
+              {
                 "user_id" :     "<user_id>",
                 "username" :    "<username>",
                 "firstname":    "<firstname>",
                 "lastname" :    "<lastname>",
                 "email" :       "<email>",
                 "emails":       "<list_of_all_additional_emails>",
                 "ip_addresses": "<list_of_ip_addresses_for_user>",
                 "active" :      "<bool>",
                 "admin" :       "<bool>",
                 "ldap_dn" :     "<ldap_dn>",
                 "last_login":   "<last_login>",
               },
     	      …
+            ]
     error:  null
 create_user
 -----------
 Creates new user. This command can
 be executed only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_user"
     args :    {
                 "username" :  "<username>",
                 "email" :     "<useremail>",
                 "password" :  "<password>",
                 "firstname" : "<firstname> = Optional(None)",
                 "lastname" :  "<lastname> = Optional(None)",
                 "active" :    "<bool> = Optional(True)",
                 "admin" :     "<bool> = Optional(False)",
                 "ldap_dn" :   "<ldap_dn> = Optional(None)"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg" : "created new user `<username>`",
               "user": {
                 "user_id" :  "<user_id>",
                 "username" : "<username>",
                 "firstname": "<firstname>",
                 "lastname" : "<lastname>",
                 "email" :    "<email>",
                 "emails":    "<list_of_all_additional_emails>",
                 "active" :   "<bool>",
                 "admin" :    "<bool>",
                 "ldap_dn" :  "<ldap_dn>",
                 "last_login": "<last_login>",
               },
+            }
     error:  null
 update_user
 -----------
 updates given user if such user exists. This command can
 be executed only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "update_user"
     args :    {
                 "userid" : "<user_id or username>",
                 "username" :  "<username> = Optional",
                 "email" :     "<useremail> = Optional",
                 "password" :  "<password> = Optional",
                 "firstname" : "<firstname> = Optional",
                 "lastname" :  "<lastname> = Optional",
                 "active" :    "<bool> = Optional",
                 "admin" :     "<bool> = Optional",
                 "ldap_dn" :   "<ldap_dn> = Optional"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg" : "updated user ID:<userid> <username>",
               "user": {
                 "user_id" :  "<user_id>",
                 "username" : "<username>",
                 "firstname": "<firstname>",
                 "lastname" : "<lastname>",
                 "email" :    "<email>",
                 "emails":    "<list_of_all_additional_emails>",
                 "active" :   "<bool>",
                 "admin" :    "<bool>",
                 "ldap_dn" :  "<ldap_dn>",
                 "last_login": "<last_login>",
               },
+            }
     error:  null
 delete_user
 -----------
 deletes givenuser if such user exists. This command can
 be executed only using api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "delete_user"
     args :    {
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg" : "deleted user ID:<userid> <username>",
               "user": null
+            }
     error:  null
 get_users_group
 ---------------
 Gets an existing users group. This command can be executed only using api_key
 belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_users_group"
     args :    {
                 "usersgroupid" : "<users group id or name>"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result : None if group not exist
+             {
                "users_group_id" : "<id>",
                "group_name" :     "<groupname>",
                "active":          "<bool>",
                "members" :  [
+                              {
                                 "user_id" :  "<user_id>",
                                 "username" : "<username>",
                                 "firstname": "<firstname>",
                                 "lastname" : "<lastname>",
                                 "email" :    "<email>",
                                 "emails":    "<list_of_all_additional_emails>",
                                 "active" :   "<bool>",
                                 "admin" :    "<bool>",
                                 "ldap_dn" :  "<ldap_dn>",
                                 "last_login": "<last_login>",
                               },
                               …
+                            ]
+             }
     error : null
 get_users_groups
 ----------------
 Lists all existing users groups. This command can be executed only using
 api_key belonging to user with admin rights.
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "get_users_groups"
     args :    { }
 OUTPUT::
     id : <id_given_in_input>
     result : [
+               {
                "users_group_id" : "<id>",
                "group_name" :     "<groupname>",
                "active":          "<bool>",
                },
                …
+              ]
     error : null
 create_users_group
 ------------------
 Creates new users group. This command can be executed only using api_key
 belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "create_users_group"
     args:     {
                 "group_name":  "<groupname>",
                 "active":"<bool> = Optional(True)"
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "msg": "created new users group `<groupname>`",
               "users_group": {
                      "users_group_id" : "<id>",
                      "group_name" :     "<groupname>",
                      "active":          "<bool>",
                },
+            }
     error:  null
 add_user_to_users_group
 -----------------------
 Adds a user to a users group. If user exists in that group success will be
 `false`. This command can be executed only using api_key
 belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"
     method :  "add_user_users_group"
     args:     {
                 "usersgroupid" : "<users group id or name>",
                 "userid" : "<user_id or username>",
+              }
 OUTPUT::
     id : <id_given_in_input>
     result: {
               "success": True|False # depends on if member is in group
               "msg": "added member `<username>` to users group `<groupname>` |
                       User is already in that group"
+            }
     error:  null
 remove_user_from_users_group
 ----------------------------
 Removes a user from a users group. If user is not in given group success will
 be `false`. This command can be executed only
 using api_key belonging to user with admin rights
 INPUT::
     id : <id_for_response>
     api_key : "<api_key>"

rhodecode/controllers/api/api.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.api
     ~~~~~~~~~~~~~~~~~~~~~~~~~
     API controller for RhodeCode
     :created_on: Aug 20, 2011
     :author: marcink
     :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
 # as published by the Free Software Foundation; version 2
 # of the License or (at your opinion) any later version of the license.
+#
 # 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, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 # MA  02110-1301, USA.
 import traceback
 import logging
 from pylons.controllers.util import abort
 from rhodecode.controllers.api import JSONRPCController, JSONRPCError
 from rhodecode.lib.auth import HasPermissionAllDecorator, \
     HasPermissionAnyDecorator, PasswordGenerator, AuthUser
 from rhodecode.lib.auth import PasswordGenerator, AuthUser, \
     HasPermissionAllDecorator, HasPermissionAnyDecorator, \
     HasPermissionAnyApi, HasRepoPermissionAnyApi
 from rhodecode.lib.utils import map_groups, repo2db_mapper
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.users_group import UsersGroupModel
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.db import Repository, RhodeCodeSetting, UserIpMap
 log = logging.getLogger(__name__)
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute
     """
     def __init__(self, attr_name):
         self.attr_name = attr_name
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
     def __call__(self):
         return self
 #alias
 OAttr = OptionalAttr
 class Optional(object):
     """
     Defines an optional parameter::
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
     is equivalent of::
         param = Optional.extract(param)
     """
     def __init__(self, type_):
         self.type_ = type_
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
     def __call__(self):
         return self.getval()
     def getval(self):
         """
         returns value from this Optional instance
         """
         return self.type_
     @classmethod
     def extract(cls, val):
         if isinstance(val, cls):
             return val.getval()
         return val
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
     :param userid:
     """
     user = UserModel().get_user(userid)
     if user is None:
         raise JSONRPCError("user `%s` does not exist" % userid)
     return user
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
     :param userid:
     """
     repo = RepoModel().get_repo(repoid)
     if repo is None:
         raise JSONRPCError('repository `%s` does not exist' % (repoid))
     return repo
 def get_users_group_or_error(usersgroupid):
     """
     Get users group by id or name or return JsonRPCError if not found
     :param userid:
     """
     users_group = UsersGroupModel().get_group(usersgroupid)
     if users_group is None:
         raise JSONRPCError('users group `%s` does not exist' % usersgroupid)
     return users_group
 def get_perm_or_error(permid):
     """
     Get permission by id or name or return JsonRPCError if not found
     :param userid:
     """
     perm = PermissionModel().get_permission_by_name(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid))
     return perm
 class ApiController(JSONRPCController):
     """
     API Controller
     Each method needs to have USER as argument this is then based on given
     API_KEY propagated as instance of user object
     Preferably this should be first argument also
     Each function should also **raise** JSONRPCError for any
     errors that happens
     """
     @HasPermissionAllDecorator('hg.admin')
     def pull(self, apiuser, repoid):
         """
         Dispatch pull action on given repo
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         try:
             ScmModel().pull_changes(repo.repo_name,
                                     self.rhodecode_user.username)
             return 'Pulled from `%s`' % repo.repo_name
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Unable to pull changes from `%s`' % repo.repo_name
+            )
     @HasPermissionAllDecorator('hg.admin')
     def rescan_repos(self, apiuser, remove_obsolete=Optional(False)):
         """
         Dispatch rescan repositories action. If remove_obsolete is set
         than also delete repos that are in database but not in the filesystem.
         aka "clean zombies"
         :param apiuser:
         :param remove_obsolete:
         """
         try:
             rm_obsolete = Optional.extract(remove_obsolete)
             added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                             remove_obsolete=rm_obsolete)
             return {'added': added, 'removed': removed}
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during rescan repositories action'
+            )
     @HasPermissionAllDecorator('hg.admin')
     def lock(self, apiuser, repoid, userid, locked):
     def lock(self, apiuser, repoid, locked, userid=Optional(OAttr('apiuser'))):
         """
         Set locking state on particular repository by given user
         Set locking state on particular repository by given user, if
         this command is runned by non-admin account userid is set to user
         who is calling this method
         :param apiuser:
         :param repoid:
         :param userid:
         :param locked:
         """
         repo = get_repo_or_error(repoid)
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         elif HasRepoPermissionAnyApi('repository.admin',
                                      'repository.write')(user=apiuser,
                                                          repo_name=repo.repo_name):
             #make sure normal user does not pass userid, he is not allowed to do that
             if not isinstance(userid, Optional):
                 raise JSONRPCError(
                     'Only RhodeCode admin can specify `userid` params'
+                )
         else:
             return abort(403)
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         locked = bool(locked)
         try:
             if locked:
                 Repository.lock(repo, user.user_id)
             else:
                 Repository.unlock(repo)
             return ('User `%s` set lock state for repo `%s` to `%s`'
                     % (user.username, repo.repo_name, locked))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred locking repository `%s`' % repo.repo_name
+            )
     @HasPermissionAllDecorator('hg.admin')
     def show_ip(self, apiuser, userid):
         """
         Shows IP address as seen from RhodeCode server, together with all
         defined IP addresses for given user
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             ip_addr_server=self.ip_addr,
             user_ips=ips
+        )
     @HasPermissionAllDecorator('hg.admin')
     def get_user(self, apiuser, userid):
         """"
         Get a user by username
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         data = user.get_api_data()
         data['permissions'] = AuthUser(user_id=user.user_id).permissions
         return data
     @HasPermissionAllDecorator('hg.admin')
     def get_users(self, apiuser):
         """"
         Get all users
         :param apiuser:
         """
         result = []
         for user in UserModel().get_all():
             result.append(user.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_user(self, apiuser, username, email, password,
                     firstname=Optional(None), lastname=Optional(None),
                     active=Optional(True), admin=Optional(False),
                     ldap_dn=Optional(None)):
         """
         Create new user
         :param apiuser:
         :param username:
         :param email:
         :param password:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         """
         if UserModel().get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % username)
         if UserModel().get_by_email(email, case_insensitive=True):
             raise JSONRPCError("email `%s` already exist" % email)
         if Optional.extract(ldap_dn):
             # generate temporary password if ldap_dn
             password = PasswordGenerator().gen_password(length=8)
         try:
             user = UserModel().create_or_update(
                 username=Optional.extract(username),
                 password=Optional.extract(password),
                 email=Optional.extract(email),
                 firstname=Optional.extract(firstname),
                 lastname=Optional.extract(lastname),
                 active=Optional.extract(active),
                 admin=Optional.extract(admin),
                 ldap_dn=Optional.extract(ldap_dn)
+            )
             Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % username)
     @HasPermissionAllDecorator('hg.admin')
     def update_user(self, apiuser, userid, username=Optional(None),
                     email=Optional(None), firstname=Optional(None),
                     lastname=Optional(None), active=Optional(None),
                     admin=Optional(None), ldap_dn=Optional(None),
                     password=Optional(None)):
         """
         Updates given user
         :param apiuser:
         :param userid:
         :param username:
         :param email:
         :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         :param password:
         """
         user = get_user_or_error(userid)
         # call function and store only updated arguments
         updates = {}
         def store_update(attr, name):
             if not isinstance(attr, Optional):
                 updates[name] = attr
         try:
             store_update(username, 'username')
             store_update(password, 'password')
             store_update(email, 'email')
             store_update(firstname, 'name')
             store_update(lastname, 'lastname')
             store_update(active, 'active')
             store_update(admin, 'admin')
             store_update(ldap_dn, 'ldap_dn')
             user = UserModel().update_user(user, **updates)
             Session().commit()
             return dict(
                 msg='updated user ID:%s %s' % (user.user_id, user.username),
                 user=user.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % userid)
     @HasPermissionAllDecorator('hg.admin')
     def delete_user(self, apiuser, userid):
         """"
         Deletes an user
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         try:
             UserModel().delete(userid)
             Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete ID:%s %s' % (user.user_id,
                                                               user.username))
     @HasPermissionAllDecorator('hg.admin')
     def get_users_group(self, apiuser, usersgroupid):
         """"
         Get users group by name or id
         :param apiuser:
         :param usersgroupid:
         """
         users_group = get_users_group_or_error(usersgroupid)
         data = users_group.get_api_data()
         members = []
         for user in users_group.members:
             user = user.user
             members.append(user.get_api_data())
         data['members'] = members
         return data
     @HasPermissionAllDecorator('hg.admin')
     def get_users_groups(self, apiuser):
         """"
         Get all users groups
         :param apiuser:
         """
         result = []
         for users_group in UsersGroupModel().get_all():
             result.append(users_group.get_api_data())
         return result
     @HasPermissionAllDecorator('hg.admin')
     def create_users_group(self, apiuser, group_name, active=Optional(True)):
         """
         Creates an new usergroup
         :param apiuser:
         :param group_name:
         :param active:
         """
         if UsersGroupModel().get_by_name(group_name):
             raise JSONRPCError("users group `%s` already exist" % group_name)
         try:
             active = Optional.extract(active)
             ug = UsersGroupModel().create(name=group_name, active=active)
             Session().commit()
             return dict(
                 msg='created new users group `%s`' % group_name,
                 users_group=ug.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create group `%s`' % group_name)
     @HasPermissionAllDecorator('hg.admin')
     def add_user_to_users_group(self, apiuser, usersgroupid, userid):
         """"
         Add a user to a users group
         :param apiuser:
         :param usersgroupid:
         :param userid:
         """
         user = get_user_or_error(userid)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             ugm = UsersGroupModel().add_user_to_group(users_group, user)
             success = True if ugm != True else False
             msg = 'added member `%s` to users group `%s`' % (
                         user.username, users_group.users_group_name
+                    )
             msg = msg if success else 'User is already in that group'
             Session().commit()
             return dict(
                 success=success,
                 msg=msg
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to add member to users group `%s`' % (
                     users_group.users_group_name
+                )
+            )
     @HasPermissionAllDecorator('hg.admin')
     def remove_user_from_users_group(self, apiuser, usersgroupid, userid):
         """
         Remove user from a group
         :param apiuser:
         :param usersgroupid:
         :param userid:
         """
         user = get_user_or_error(userid)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             success = UsersGroupModel().remove_user_from_group(users_group,
                                                                user)
             msg = 'removed member `%s` from users group `%s`' % (
                         user.username, users_group.users_group_name
+                    )
             msg = msg if success else "User wasn't in group"
             Session().commit()
             return dict(success=success, msg=msg)
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to remove member from users group `%s`' % (
                         users_group.users_group_name
+                    )
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def get_repo(self, apiuser, repoid):
         """"
         Get repository by name
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = user.get_api_data()
             user_data['type'] = "user"
             user_data['permission'] = perm
             members.append(user_data)
         for users_group in repo.users_group_to_perm:
             perm = users_group.permission.permission_name
             users_group = users_group.users_group
             users_group_data = users_group.get_api_data()
             users_group_data['type'] = "users_group"
             users_group_data['permission'] = perm
             members.append(users_group_data)
         data = repo.get_api_data()
         data['members'] = members
         return data
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def get_repos(self, apiuser):
         """"
         Get all repositories
         :param apiuser:
         """
         result = []
         for repo in RepoModel().get_all():
             result.append(repo.get_api_data())
         return result
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def get_repo_nodes(self, apiuser, repoid, revision, root_path,
                        ret_type='all'):
         """
         returns a list of nodes and it's children
         for a given path at given revision. It's possible to specify ret_type
         to show only files or dirs
         :param apiuser:
         :param repoid: name or id of repository
         :param revision: revision for which listing should be done
         :param root_path: path from which start displaying
         :param ret_type: return type 'all|files|dirs' nodes
         """
         repo = get_repo_or_error(repoid)
         try:
             _d, _f = ScmModel().get_nodes(repo, revision, root_path,
                                           flat=False)
             _map = {
                 'all': _d + _f,
                 'files': _f,
                 'dirs': _d,
+            }
             return _map[ret_type]
         except KeyError:
             raise JSONRPCError('ret_type must be one of %s' % _map.keys())
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
+            )
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def create_repo(self, apiuser, repo_name, owner, repo_type=Optional('hg'),
                     description=Optional(''), private=Optional(False),
                     clone_uri=Optional(None), landing_rev=Optional('tip'),
                     enable_statistics=Optional(False),
                     enable_locking=Optional(False),
                     enable_downloads=Optional(False)):
         """
         Create repository, if clone_url is given it makes a remote clone
         if repo_name is withina  group name the groups will be created
         automatically if they aren't present
         :param apiuser:
         :param repo_name:
         :param onwer:
         :param repo_type:
         :param description:
         :param private:
         :param clone_uri:
         :param landing_rev:
         """
         owner = get_user_or_error(owner)
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
         defs = RhodeCodeSetting.get_default_repo_settings(strip_prefix=True)
         if isinstance(private, Optional):
             private = defs.get('repo_private') or Optional.extract(private)
         if isinstance(repo_type, Optional):
             repo_type = defs.get('repo_type')
         if isinstance(enable_statistics, Optional):
             enable_statistics = defs.get('repo_enable_statistics')
         if isinstance(enable_locking, Optional):
             enable_locking = defs.get('repo_enable_locking')
         if isinstance(enable_downloads, Optional):
             enable_downloads = defs.get('repo_enable_downloads')
         clone_uri = Optional.extract(clone_uri)
         description = Optional.extract(description)
         landing_rev = Optional.extract(landing_rev)
         try:
             # create structure of groups and return the last group
             group = map_groups(repo_name)
             repo = RepoModel().create_repo(
                 repo_name=repo_name,
                 repo_type=repo_type,
                 description=description,
                 owner=owner,
                 private=private,
                 clone_uri=clone_uri,
                 repos_group=group,
                 landing_rev=landing_rev,
                 enable_statistics=enable_statistics,
                 enable_downloads=enable_downloads,
                 enable_locking=enable_locking
+            )
             Session().commit()
             return dict(
                 msg="Created new repository `%s`" % (repo.repo_name),
                 repo=repo.get_api_data()
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create repository `%s`' % repo_name)
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def fork_repo(self, apiuser, repoid, fork_name, owner,
                   description=Optional(''), copy_permissions=Optional(False),
                   private=Optional(False), landing_rev=Optional('tip')):
         repo = get_repo_or_error(repoid)
         repo_name = repo.repo_name
         owner = get_user_or_error(owner)
         _repo = RepoModel().get_by_repo_name(fork_name)
         if _repo:
             type_ = 'fork' if _repo.fork else 'repo'
             raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
         try:
             # create structure of groups and return the last group
             group = map_groups(fork_name)
             form_data = dict(
                 repo_name=fork_name,
                 repo_name_full=fork_name,
                 repo_group=group,
                 repo_type=repo.repo_type,
                 description=Optional.extract(description),
                 private=Optional.extract(private),
                 copy_permissions=Optional.extract(copy_permissions),
                 landing_rev=Optional.extract(landing_rev),
                 update_after_clone=False,
                 fork_parent_id=repo.repo_id,
+            )
             RepoModel().create_fork(form_data, cur_user=owner)
             return dict(
                 msg='Created fork of `%s` as `%s`' % (repo.repo_name,
                                                       fork_name),
                 success=True  # cannot return the repo data here since fork
                               # cann be done async
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to fork repository `%s` as `%s`' % (repo_name,
                                                             fork_name)
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def delete_repo(self, apiuser, repoid):
         """
         Deletes a given repository
         :param apiuser:
         :param repoid:
         """
         repo = get_repo_or_error(repoid)
         try:
             RepoModel().delete(repo)
             Session().commit()
             return dict(
                 msg='Deleted repository `%s`' % repo.repo_name,
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to delete repository `%s`' % repo.repo_name
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def grant_user_permission(self, apiuser, repoid, userid, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found
         :param repoid:
         :param userid:
         :param perm:
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         perm = get_perm_or_error(perm)
         try:
             RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
             Session().commit()
             return dict(
                 msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                     perm.permission_name, user.username, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
+                )
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def revoke_user_permission(self, apiuser, repoid, userid):
         """
         Revoke permission for user on given repository
         :param apiuser:
         :param repoid:
         :param userid:
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         try:
             RepoModel().revoke_user_permission(repo=repo, user=user)
             Session().commit()
             return dict(
                 msg='Revoked perm for user: `%s` in repo: `%s`' % (
                     user.username, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
+                )
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def grant_users_group_permission(self, apiuser, repoid, usersgroupid,
                                      perm):
         """
         Grant permission for users group on given repository, or update
         existing one if found
         :param apiuser:
         :param repoid:
         :param usersgroupid:
         :param perm:
         """
         repo = get_repo_or_error(repoid)
         perm = get_perm_or_error(perm)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             RepoModel().grant_users_group_permission(repo=repo,
                                                      group_name=users_group,
                                                      perm=perm)
             Session().commit()
             return dict(
                 msg='Granted perm: `%s` for users group: `%s` in '
                     'repo: `%s`' % (
                     perm.permission_name, users_group.users_group_name,
                     repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for users group: `%s` in '
                 'repo: `%s`' % (
                     usersgroupid, repo.repo_name
+                )
+            )
-    @HasPermissionAnyDecorator('hg.admin')
+    @HasPermissionAllDecorator('hg.admin')
     def revoke_users_group_permission(self, apiuser, repoid, usersgroupid):
         """
         Revoke permission for users group on given repository
         :param apiuser:
         :param repoid:
         :param usersgroupid:
         """
         repo = get_repo_or_error(repoid)
         users_group = get_users_group_or_error(usersgroupid)
         try:
             RepoModel().revoke_users_group_permission(repo=repo,
                                                       group_name=users_group)
             Session().commit()
             return dict(
                 msg='Revoked perm for users group: `%s` in repo: `%s`' % (
                     users_group.users_group_name, repo.repo_name
                 ),
                 success=True
+            )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for users group: `%s` in '
                 'repo: `%s`' % (
                     users_group.users_group_name, repo.repo_name
+                )
+            )

rhodecode/lib/auth.py

➞

Show inline comments

@@ @@ -482,398 +482,501 @@ class LoginRequired(object): @@
         cls = fargs[0]
         user = cls.rhodecode_user
         loc = "%s:%s" % (cls.__class__.__name__, func.__name__)
         #check IP
         ip_access_ok = True
         if not user.ip_allowed:
             from rhodecode.lib import helpers as h
             h.flash(h.literal(_('IP %s not allowed' % (user.ip_addr))),
                     category='warning')
             ip_access_ok = False
         api_access_ok = False
         if self.api_access:
             log.debug('Checking API KEY access for %s' % cls)
             if user.api_key == request.GET.get('api_key'):
                 api_access_ok = True
             else:
                 log.debug("API KEY token not valid")
         log.debug('Checking if %s is authenticated @ %s' % (user.username, loc))
         if (user.is_authenticated or api_access_ok) and ip_access_ok:
             reason = 'RegularAuth' if user.is_authenticated else 'APIAuth'
             log.info('user %s is authenticated and granted access to %s '
                      'using %s' % (user.username, loc, reason)
+            )
             return func(*fargs, **fkwargs)
         else:
             log.warn('user %s NOT authenticated on func: %s' % (
                 user, loc)
+            )
             p = url.current()
             log.debug('redirecting to login page with %s' % p)
             return redirect(url('login_home', came_from=p))
 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.rhodecode_user
         log.debug('Checking if user is not anonymous @%s' % cls)
         anonymous = self.user.username == 'default'
         if anonymous:
             p = url.current()
             import rhodecode.lib.helpers as h
             h.flash(_('You need to be a registered user to '
                       'perform this action'),
                     category='warning')
             return redirect(url('login_home', came_from=p))
         else:
             return func(*fargs, **fkwargs)
 class PermsDecorator(object):
     """Base class for controller decorators"""
     def __init__(self, *required_perms):
         available_perms = config['available_permissions']
         for perm in required_perms:
             if perm not in available_perms:
                 raise Exception("'%s' permission is not defined" % perm)
         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.rhodecode_user
         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 == 'default'
             if anonymous:
                 p = url.current()
                 import rhodecode.lib.helpers as h
                 h.flash(_('You need to be a signed in to '
                           'view this page'),
                         category='warning')
                 return redirect(url('login_home', came_from=p))
             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 HasReposGroupPermissionAllDecorator(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):
         group_name = get_repos_group_slug(request)
         try:
             user_perms = set([self.user_perms['repositories_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.issubset(user_perms):
             return True
         return False
 class HasReposGroupPermissionAnyDecorator(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):
         group_name = get_repos_group_slug(request)
         try:
             user_perms = set([self.user_perms['repositories_groups'][group_name]])
         except KeyError:
             return False
         if self.required_perms.intersection(user_perms):
             return True
         return False
 #==============================================================================
 # CHECK FUNCTIONS
 #==============================================================================
 class PermsFunction(object):
     """Base function for other check functions"""
     def __init__(self, *perms):
         available_perms = config['available_permissions']
         for perm in perms:
             if perm not in available_perms:
                 raise Exception("'%s' permission is not defined" % perm)
         self.required_perms = set(perms)
         self.user_perms = None
         self.repo_name = None
         self.group_name = None
     def __call__(self, check_Location=''):
         user = request.user
         cls_name = self.__class__.__name__
         check_scope = {
             'HasPermissionAll': '',
             'HasPermissionAny': '',
             'HasRepoPermissionAll': 'repo:%s' % self.repo_name,
             'HasRepoPermissionAny': 'repo:%s' % self.repo_name,
             'HasReposGroupPermissionAll': 'group:%s' % self.group_name,
             'HasReposGroupPermissionAny': 'group:%s' % self.group_name,
         }.get(cls_name, '?')
         log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
                   self.required_perms, user, check_scope,
                   check_Location or 'unspecified location')
         if not user:
             log.debug('Empty request user')
             return False
         self.user_perms = user.permissions
         if self.check_permissions():
             log.debug('Permission to %s granted for user: %s @ %s', self.repo_name, user,
                       check_Location or 'unspecified location')
             return True
         else:
             log.debug('Permission to %s denied for user: %s @ %s', self.repo_name, user,
                         check_Location or 'unspecified location')
             return False
     def check_permissions(self):
         """Dummy function for overriding"""
         raise Exception('You have to write this function in child class')
 class HasPermissionAll(PermsFunction):
     def check_permissions(self):
         if self.required_perms.issubset(self.user_perms.get('global')):
             return True
         return False
 class HasPermissionAny(PermsFunction):
     def check_permissions(self):
         if self.required_perms.intersection(self.user_perms.get('global')):
             return True
         return False
 class HasRepoPermissionAll(PermsFunction):
     def __call__(self, repo_name=None, check_Location=''):
         self.repo_name = repo_name
         return super(HasRepoPermissionAll, self).__call__(check_Location)
     def check_permissions(self):
         if not self.repo_name:
             self.repo_name = get_repo_slug(request)
         try:
             self._user_perms = set(
                 [self.user_perms['repositories'][self.repo_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 class HasRepoPermissionAny(PermsFunction):
     def __call__(self, repo_name=None, check_Location=''):
         self.repo_name = repo_name
         return super(HasRepoPermissionAny, self).__call__(check_Location)
     def check_permissions(self):
         if not self.repo_name:
             self.repo_name = get_repo_slug(request)
         try:
             self._user_perms = set(
                 [self.user_perms['repositories'][self.repo_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
 class HasReposGroupPermissionAny(PermsFunction):
     def __call__(self, group_name=None, check_Location=''):
         self.group_name = group_name
         return super(HasReposGroupPermissionAny, self).__call__(check_Location)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['repositories_groups'][self.group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.intersection(self._user_perms):
             return True
         return False
 class HasReposGroupPermissionAll(PermsFunction):
     def __call__(self, group_name=None, check_Location=''):
         self.group_name = group_name
         return super(HasReposGroupPermissionAll, self).__call__(check_Location)
     def check_permissions(self):
         try:
             self._user_perms = set(
                 [self.user_perms['repositories_groups'][self.group_name]]
+            )
         except KeyError:
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 #==============================================================================
 # SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
 #==============================================================================
 class HasPermissionAnyMiddleware(object):
     def __init__(self, *perms):
         self.required_perms = set(perms)
     def __call__(self, user, repo_name):
         # repo_name MUST be unicode, since we handle keys in permission
         # dict by unicode
         repo_name = safe_unicode(repo_name)
         usr = AuthUser(user.user_id)
         try:
             self.user_perms = set([usr.permissions['repositories'][repo_name]])
         except Exception:
             log.error('Exception while accessing permissions %s' %
                       traceback.format_exc())
             self.user_perms = set()
         self.username = user.username
         self.repo_name = repo_name
         return self.check_permissions()
     def check_permissions(self):
         log.debug('checking VCS protocol '
                   'permissions %s for user:%s repository:%s', self.user_perms,
                                                 self.username, self.repo_name)
         if self.required_perms.intersection(self.user_perms):
             log.debug('permission granted for user:%s on repo:%s' % (
                           self.username, self.repo_name
+                     )
+            )
             return True
         log.debug('permission denied for user:%s on repo:%s' % (
                       self.username, self.repo_name
+                 )
+        )
         return False
 #==============================================================================
 # SPECIAL VERSION TO HANDLE API AUTH
 #==============================================================================
 class _BaseApiPerm(object):
     def __init__(self, *perms):
         self.required_perms = set(perms)
     def __call__(self, check_location='unspecified', user=None, repo_name=None):
         cls_name = self.__class__.__name__
         check_scope = 'user:%s, repo:%s' % (user, repo_name)
         log.debug('checking cls:%s %s %s @ %s', cls_name,
                   self.required_perms, check_scope, check_location)
         if not user:
             log.debug('Empty User passed into arguments')
             return False
         ## process user
         if not isinstance(user, AuthUser):
             user = AuthUser(user.user_id)
         if self.check_permissions(user.permissions, repo_name):
             log.debug('Permission to %s granted for user: %s @ %s', repo_name,
                       user, check_location)
             return True
         else:
             log.debug('Permission to %s denied for user: %s @ %s', repo_name,
                       user, check_location)
             return False
     def check_permissions(self, perm_defs, repo_name):
         """
         implement in child class should return True if permissions are ok,
         False otherwise
         :param perm_defs: dict with permission definitions
         :param repo_name: repo name
         """
         raise NotImplementedError()
 class HasPermissionAllApi(_BaseApiPerm):
     def __call__(self, user, check_location=''):
         return super(HasPermissionAllApi, self)\
             .__call__(check_location=check_location, user=user)
     def check_permissions(self, perm_defs, repo):
         if self.required_perms.issubset(perm_defs.get('global')):
             return True
         return False
 class HasPermissionAnyApi(_BaseApiPerm):
     def __call__(self, user, check_location=''):
         return super(HasPermissionAnyApi, self)\
             .__call__(check_location=check_location, user=user)
     def check_permissions(self, perm_defs, repo):
         if self.required_perms.intersection(perm_defs.get('global')):
             return True
         return False
 class HasRepoPermissionAllApi(_BaseApiPerm):
     def __call__(self, user, repo_name, check_location=''):
         return super(HasRepoPermissionAllApi, self)\
             .__call__(check_location=check_location, user=user,
                       repo_name=repo_name)
     def check_permissions(self, perm_defs, repo_name):
         try:
             self._user_perms = set(
                 [perm_defs['repositories'][repo_name]]
+            )
         except KeyError:
             log.warning(traceback.format_exc())
             return False
         if self.required_perms.issubset(self._user_perms):
             return True
         return False
 class HasRepoPermissionAnyApi(_BaseApiPerm):
     def __call__(self, user, repo_name, check_location=''):
         return super(HasRepoPermissionAnyApi, self)\
             .__call__(check_location=check_location, user=user,
                       repo_name=repo_name)
     def check_permissions(self, perm_defs, repo_name):
         try:
             _user_perms = set(
                 [perm_defs['repositories'][repo_name]]
+            )
         except KeyError:
             log.warning(traceback.format_exc())
             return False
         if self.required_perms.intersection(_user_perms):
             return True
         return False
 def check_ip_access(source_ip, allowed_ips=None):
     """
     Checks if source_ip is a subnet of any of allowed_ips.
     :param source_ip:
     :param allowed_ips: list of allowed ips together with mask
     """
     from rhodecode.lib import ipaddr
     log.debug('checking if ip:%s is subnet of %s' % (source_ip, allowed_ips))
     if isinstance(allowed_ips, (tuple, list, set)):
         for ip in allowed_ips:
             if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
                 return True
     return False

rhodecode/tests/api/api_base.py

➞

Show inline comments

 from __future__ import with_statement
 import random
 import mock
 from rhodecode.tests import *
 from rhodecode.lib.compat import json
 from rhodecode.lib.auth import AuthUser
 from rhodecode.model.user import UserModel
 from rhodecode.model.users_group import UsersGroupModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.db import Repository
 API_URL = '/_admin/api'
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     :param random_id:
     :type 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
 TEST_USERS_GROUP = 'test_users_group'
 def make_users_group(name=TEST_USERS_GROUP):
     gr = UsersGroupModel().create(name=name)
     UsersGroupModel().add_user_to_group(users_group=gr,
                                         user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 def destroy_users_group(name=TEST_USERS_GROUP):
     UsersGroupModel().delete(users_group=name, force=True)
     Session().commit()
 def create_repo(repo_name, repo_type):
     # create new repo
     form_data = _get_repo_create_params(
                     repo_name_full=repo_name,
                     repo_description='description %s' % repo_name,
+                )
     cur_user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
     r = RepoModel().create(form_data, cur_user)
     Session().commit()
     return r
 def create_fork(fork_name, fork_type, fork_of):
     fork = RepoModel(Session())._get_repo(fork_of)
     r = create_repo(fork_name, fork_type)
     r.fork = fork
     Session().add(r)
     Session().commit()
     return r
 def destroy_repo(repo_name):
     RepoModel().delete(repo_name)
     Session().commit()
 class BaseTestApi(object):
     REPO = None
     REPO_TYPE = None
     @classmethod
     def setUpClass(self):
         self.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         self.apikey = self.usr.api_key
         self.TEST_USER = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@api.rhodecode.org',
             firstname='first',
             lastname='last'
+        )
         Session().commit()
         self.TEST_USER_LOGIN = self.TEST_USER.username
     @classmethod
     def teardownClass(self):
         pass
     def setUp(self):
         self.maxDiff = None
         make_users_group()
     def tearDown(self):
         destroy_users_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(self):
 #        from rhodecode.controllers.api.api import Optional
 #        option1 = Optional(None)
 #        self.assertEqual('<Optional:%s>' % None, repr(option1))
+#
 #        self.assertEqual(1, Optional.extract(Optional(1)))
 #        self.assertEqual('trololo', Optional.extract('trololo'))
     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_user')
         response = api_call(self, params)
         expected = 'Missing non optional `userid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users',)
         response = api_call(self, params)
         ret_all = []
         for usr in UserModel().get_all():
             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 = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(usr.user_id).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_pull(self):
         #TODO: issues with rhodecode_extras here.. not sure why !
         pass
 #        repo_name = 'test_pull'
 #        r = create_repo(repo_name, self.REPO_TYPE)
 #        r.clone_uri = TEST_self.REPO
 #        Session.add(r)
 #        Session.commit()
+#
 #        id_, params = _build_data(self.apikey, 'pull',
 #                                  repoid=repo_name,)
 #        response = self.app.post(API_URL, content_type='application/json',
 #                                 params=params)
+#
 #        expected = 'Pulled from `%s`' % repo_name
 #        self._compare_ok(id_, expected, given=response.body)
+#
 #        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_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 = ('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_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 = ('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)
         expected = ('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)
     @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_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 = UserModel().get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         UserModel().delete(usr.user_id)
         Session().commit()
     @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@rhodecode.org',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = api_call(self, params)
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'delete', crash)
     def test_api_delete_user_when_exception_happened(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@rhodecode.org',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username,)
         response = api_call(self, params)
         ret = 'failed to delete ID:%s %s' % (usr.user_id,
                                              usr.username)
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     @parameterized.expand([('firstname', 'new_username'),
                            ('lastname', 'new_username'),
                            ('email', 'new_username'),
                            ('admin', True),
                            ('admin', False),
                            ('ldap_dn', 'test'),
                            ('ldap_dn', None),
                            ('active', False),
                            ('active', True),
                            ('password', 'newpass')
                            ])
     def test_api_update_user(self, name, expected):
         usr = UserModel().get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, self.TEST_USER_LOGIN),
         'user': jsonify(UserModel()\
                             .get_by_username(self.TEST_USER_LOGIN)\
                             .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_by_user_id(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = {
         'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),
         'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'update_user', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo(self):
         new_group = 'some_new_group'
         make_users_group(new_group)
         RepoModel().grant_users_group_permission(repo=self.REPO,
                                                  group_name=new_group,
                                                  perm='repository.read')
         Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = user.get_api_data()
             user_data['type'] = "user"
             user_data['permission'] = perm
             members.append(user_data)
         for users_group in repo.users_group_to_perm:
             perm = users_group.permission.permission_name
             users_group = users_group.users_group
             users_group_data = users_group.get_api_data()
             users_group_data['type'] = "users_group"
             users_group_data['permission'] = perm
             members.append(users_group_data)
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_users_group(new_group)
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
         result = []
         for repo in RepoModel().get_all():
             result.append(repo.get_api_data())
         ret = jsonify(result)
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @parameterized.expand([('all', 'all'),
                            ('dirs', 'dirs'),
                            ('files', 'files'), ])
     def test_api_get_repo_nodes(self, name, ret_type):
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = json.loads(response.body)['result']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_revisions(self):
         rev = 'i-dont-exist'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,)
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_path(self):
         rev = 'tip'
         path = '/idontexits'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,)
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_ret_type(self):
         rev = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         expected = 'ret_type must be one of %s' % (['files', 'dirs', 'all'])
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo(self):
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'repo': jsonify(repo.get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(repo_name)
     def test_api_create_repo_unknown_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=owner,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo_exists(self):
         repo_name = self.REPO
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create_repo', crash)
     def test_api_create_repo_exception_occurred(self):
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                     repo_name=repo_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
                                     repo_type='hg',
+                                  )
         response = api_call(self, params)
         expected = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_repo(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'delete_repo',
                                   repoid=repo_name,)
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_delete_repo_exception_occurred(self):
         repo_name = 'api_delete_me'
         create_repo(repo_name, self.REPO_TYPE)
         try:
             with mock.patch.object(RepoModel, 'delete', crash):
                 id_, params = _build_data(self.apikey, 'delete_repo',
                                           repoid=repo_name,)
                 response = api_call(self, params)
                 expected = 'failed to delete repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
         finally:
             destroy_repo(repo_name)
     def test_api_fork_repo(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                     repoid=self.REPO,
                                     fork_name=fork_name,
                                     owner=TEST_USER_ADMIN_LOGIN,
+                                  )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):

0 comments (0 inline, 0 general)