Changeset - 87c2cd07166a
[Not reviewed]
default
0 5 0
Mads Kiilerich (mads) - 5 years ago 2020-11-01 23:04:25
mads@kiilerich.com
Grafted from: 90dd5831e497
lib: move get_all_user_repos to AuthUser
5 files changed with 13 insertions and 17 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/api/api.py
Show inline comments
 
@@ -1088,25 +1088,25 @@ class ApiController(JSONRPCController):
 
                        "description" :      "<description>",
 
                        "landing_rev":       "<landing_rev>",
 
                        "owner":             "<repo_owner>",
 
                        "fork_of":           "<name_of_fork_parent>",
 
                        "enable_downloads":  "<bool>",
 
                        "enable_statistics": "<bool>",
 
                      },
 
 
                    ]
 
            error:  null
 
        """
 
        if not HasPermissionAny('hg.admin')():
 
            repos = RepoModel().get_all_user_repos(user=request.authuser.user_id)
 
            repos = request.authuser.get_all_user_repos()
 
        else:
 
            repos = db.Repository.query()
 

	
 
        return [
 
            repo.get_api_data()
 
            for repo in repos
 
        ]
 

	
 
    # permission check inside
 
    def get_repo_nodes(self, repoid, revision, root_path,
 
                       ret_type='all'):
 
        """
kallithea/lib/auth.py
Show inline comments
 
@@ -458,24 +458,34 @@ class AuthUser(object):
 
                pass
 

	
 
        user_ips = db.UserIpMap.query().filter(db.UserIpMap.user_id == 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 get_all_user_repos(self):
 
        """
 
        Gets all repositories that user have at least read access
 
        """
 
        repos = [repo_name
 
            for repo_name, perm in self.repository_permissions.items()
 
            if perm in ['repository.read', 'repository.write', 'repository.admin']
 
            ]
 
        return db.Repository.query().filter(db.Repository.repo_name.in_(repos))
 

	
 

	
 
#==============================================================================
 
# CHECK DECORATORS
 
#==============================================================================
 

	
 
def _redirect_to_login(message=None):
 
    """Return an exception that must be raised. It will redirect to the login
 
    page which will redirect back to the current URL after authentication.
 
    The optional message will be shown in a flash message."""
 
    if message:
 
        webutils.flash(message, category='warning')
 
    p = request.path_qs
kallithea/lib/auth_modules/__init__.py
Show inline comments
 
@@ -13,25 +13,25 @@
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
Authentication modules
 
"""
 

	
 
import importlib
 
import logging
 
import traceback
 
from inspect import isfunction
 

	
 
from kallithea.lib.auth import AuthUser
 
from kallithea.lib.compat import hybrid_property
 
from kallithea.lib.utils2 import asbool, PasswordGenerator
 
from kallithea.lib.utils2 import PasswordGenerator, asbool
 
from kallithea.model import db, meta, validators
 
from kallithea.model.user import UserModel
 
from kallithea.model.user_group import UserGroupModel
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class LazyFormencode(object):
 
    def __init__(self, formencode_obj, *args, **kwargs):
 
        self.formencode_obj = formencode_obj
 
        self.args = args
kallithea/model/repo.py
Show inline comments
 
@@ -80,38 +80,24 @@ class RepoModel(object):
 
        repo = db.Repository.query() \
 
            .filter(db.Repository.repo_id == repo_id)
 
        return repo.scalar()
 

	
 
    def get_repo(self, repository):
 
        return db.Repository.guess_instance(repository)
 

	
 
    def get_by_repo_name(self, repo_name):
 
        repo = db.Repository.query() \
 
            .filter(db.Repository.repo_name == repo_name)
 
        return repo.scalar()
 

	
 
    def get_all_user_repos(self, user):
 
        """
 
        Gets all repositories that user have at least read access
 

	
 
        :param user:
 
        """
 
        from kallithea.lib.auth import AuthUser
 
        auth_user = AuthUser(dbuser=db.User.guess_instance(user))
 
        repos = [repo_name
 
            for repo_name, perm in auth_user.repository_permissions.items()
 
            if perm in ['repository.read', 'repository.write', 'repository.admin']
 
            ]
 
        return db.Repository.query().filter(db.Repository.repo_name.in_(repos))
 

	
 
    @classmethod
 
    def _render_datatable(cls, tmpl, *args, **kwargs):
 
        from tg import app_globals, request
 
        from tg import tmpl_context as c
 
        from tg.i18n import ugettext as _
 

	
 
        import kallithea.lib.helpers as h
 

	
 
        _tmpl_lookup = app_globals.mako_lookup
 
        template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
 

	
 
        tmpl = template.get_def(tmpl)
kallithea/tests/api/api_base.py
Show inline comments
 
@@ -679,25 +679,25 @@ class _BaseTestApi(object):
 
            repo.get_api_data()
 
            for repo in db.Repository.query()
 
        ])
 

	
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_get_repos_non_admin(self):
 
        id_, params = _build_data(self.apikey_regular, 'get_repos')
 
        response = api_call(self, params)
 

	
 
        expected = jsonify([
 
            repo.get_api_data()
 
            for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN)
 
            for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()
 
        ])
 

	
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,ret_type', [
 
        ('all', 'all'),
 
        ('dirs', 'dirs'),
 
        ('files', 'files'),
 
    ])
 
    def test_api_get_repo_nodes(self, name, ret_type):
 
        rev = 'tip'
 
        path = '/'
0 comments (0 inline, 0 general)