log.info('Populated Ui and Settings defaults')

    def create_user(self, username, password, email='', admin=False):

        log.info('creating user %s', username)

        UserModel().create_or_update(username, password, email,

                                     firstname='Kallithea', lastname='Admin',

                                     active=True, admin=admin,

                                     extern_type=User.DEFAULT_AUTH_TYPE)

    def create_default_user(self):

        log.info('creating default user')

        # create default user for handling default permissions.

                                            password=str(uuid.uuid1())[:20],

                                            email='anonymous@kallithea-scm.org',

                                            firstname='Anonymous',

                                            lastname='User')

        # based on configuration options activate/deactivate this user which

        # controls anonymous access

        if self.cli_args.get('public_access') is False:

            log.info('Public access disabled')

            user.active = False

            Session().commit()

    def create_permissions(self):

        """

        # module.(access|create|change|delete)_[name]

        # module.(none|read|write|admin)

        log.info('creating permissions')

        PermissionModel().create_permissions()

    def populate_default_permissions(self):

        """

        Populate default permissions. It will create only the default

        permissions that are missing, and not alter already defined ones

        """

        log.info('creating default user permissions')

            self.__class__.__name__,

            self.ui_section, self.ui_key, self.ui_value)

class User(Base, BaseDbModel):

    __tablename__ = 'users'

    __table_args__ = (

        Index('u_username_idx', 'username'),

        Index('u_email_idx', 'email'),

        _table_args_default_dict,

    )

    DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'

    # The name of the default auth type in extern_type, 'internal' lives in auth_internal.py

    DEFAULT_AUTH_TYPE = 'internal'

    user_id = Column(Integer(), primary_key=True)

    username = Column(String(255), nullable=False, unique=True)

    password = Column(String(255), nullable=False)

    active = Column(Boolean(), nullable=False, default=True)

    admin = Column(Boolean(), nullable=False, default=False)

    name = Column("firstname", Unicode(255), nullable=False)

    lastname = Column(Unicode(255), nullable=False)

    _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?

        return '%s %s <%s>' % (self.firstname, self.lastname, self.email)

    @property

    def short_contact(self):

        return '%s %s' % (self.firstname, self.lastname)

    @property

    def is_admin(self):

        return self.admin

    @hybrid_property

    def is_default_user(self):

    @hybrid_property

    def user_data(self):

        if not self._user_data:

            return {}

        try:

            return ext_json.loads(self._user_data)

        except TypeError:

            return {}

    @user_data.setter

        self.last_login = datetime.datetime.now()

        log.debug('updated user %s lastlogin', self.username)

    @classmethod

    def get_first_admin(cls):

        user = User.query().filter(User.admin == True).first()

        if user is None:

            raise Exception('Missing administrative account!')

        return user

    @classmethod

    def get_default_user(cls):

        if user is None:

            raise Exception('Missing default account!')

        return user

    def get_api_data(self, details=False):

        """

        Common function for generating user related data for API

        """

        user = self

        data = dict(

            user_id=user.user_id,

            username=user.username,

        def _make_perm(perm):

            new_perm = UserToPerm()

            new_perm.user = user

            new_perm.permission = Permission.get_by_key(perm)

            return new_perm

        def _get_group(perm_name):

            return '.'.join(perm_name.split('.')[:1])

        perms = UserToPerm.query().filter(UserToPerm.user == user).all()

        defined_perms_groups = set(_get_group(x.permission.permission_name) for x in perms)

        log.debug('GOT ALREADY DEFINED:%s', perms)

        if force:

            for perm in perms:

                Session().delete(perm)

            Session().commit()

            defined_perms_groups = []

        # For every default permission that needs to be created, we check if

        # its group is already defined. If it's not, we create default permission.

            gr = _get_group(perm_name)

            if gr not in defined_perms_groups:

                log.debug('GR:%s not found, creating permission %s',

                          gr, perm_name)

                new_perm = _make_perm(perm_name)

                Session().add(new_perm)

    def update(self, form_result):

        perm_user = User.get_by_username(username=form_result['perm_user_name'])

        try:

            # stage 1 set anonymous access

            # fill new permissions in order of how they were added

            for k in sorted(new_perms_group, key=lambda k: int(k)):

                perm_dict = new_perms_group[k]

                new_member = perm_dict.get('name')

                new_perm = perm_dict.get('perm')

                new_type = perm_dict.get('type')

                if new_member and new_perm and new_type:

                    perms_new.add((new_member, new_perm, new_type))

            for k, v in value.items():

                if k.startswith('u_perm_') or k.startswith('g_perm_'):

                    t = {'u': 'user',

                         'g': 'users_group'

                    }[k[0]]

                        if str2bool(value.get('repo_private')):

                            # set none for default when updating to

                            # private repo protects against form manipulation

                            v = EMPTY_PERM

            value['perms_updates'] = list(perms_update)

            value['perms_new'] = list(perms_new)

            # update permissions

            for k, v, t in perms_new:

                try:

                    if t == 'user':

                        _user_db = User.query() \

                            .filter(User.active == True) \

                            .filter(User.username == k).one()

                    if t == 'users_group':

        ret['members'] = members

        ret['followers'] = followers

        expected = ret

        try:

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

        finally:

            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)

    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          perm='repository.none')

        try:

            RepoModel().grant_user_permission(repo=self.REPO,

                                              user=self.TEST_USER_LOGIN,

                                              perm='repository.none')

            id_, params = _build_data(self.apikey_regular, 'get_repo',

                                      repoid=self.REPO)

            response = api_call(self, params)

            expected = 'repository `%s` does not exist' % (self.REPO)

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

        finally:

            RepoModel().grant_user_permission(repo=self.REPO,

                                              perm='repository.read')

    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_regular, 'fork_repo',

                                  repoid=self.REPO,

                                  fork_name=fork_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

        )

        response = api_call(self, params)

        expected = 'Only Kallithea admin can specify `owner` param'

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

        fixture.destroy_repo(fork_name)

    def test_api_fork_repo_non_admin_no_permission_to_fork(self):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          perm='repository.none')

        try:

            fork_name = 'api-repo-fork'

            id_, params = _build_data(self.apikey_regular, 'fork_repo',

                                      repoid=self.REPO,

                                      fork_name=fork_name,

            )

            response = api_call(self, params)

            expected = 'repository `%s` does not exist' % (self.REPO)

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

        finally:

            RepoModel().grant_user_permission(repo=self.REPO,

                                              perm='repository.read')

            fixture.destroy_repo(fork_name)

    @base.parametrize('name,perm', [

        ('read', 'repository.read'),

        ('write', 'repository.write'),

        ('admin', 'repository.admin'),

    ])

    def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):

        fork_name = 'api-repo-fork'

        # regardless of base repository permission, forking is disallowed

        # when repository creation is disabled

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_in_group_without_needed_permissions(self):

        usr = self.log_user(base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR_PASS)

        # avoid spurious RepoGroup DetachedInstanceError ...

        session_csrf_secret_token = self.session_csrf_secret_token()

        # revoke

        user_model = UserModel()

        # disable fork and create on default user

        # disable on regular user

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.none')

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.none')

        Session().commit()

        ## create GROUP

        group_name = 'reg_sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description='test',

        self.log_user()

        repo = Repository.get_by_repo_name(self.REPO)

        response = self.app.post(base.url('edit_repo_advanced_fork', repo_name=self.REPO),

                                params=dict(id_fork_of=repo.repo_id, _session_csrf_secret_token=self.session_csrf_secret_token()))

        self.checkSessionFlash(response,

                               'An error occurred during this operation')

    def test_create_on_top_level_without_permissions(self):

        usr = self.log_user(base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR_PASS)

        # revoke

        user_model = UserModel()

        # disable fork and create on default user

        # disable on regular user

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.none')

        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')

        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.none')

        Session().commit()

        user = User.get(usr['user_id'])

        repo_name = self.NEW_REPO + 'no_perms'

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestPermissions(base.TestController):

    @classmethod

    def setup_class(cls):

        # recreate default user to get a clean start

                                                     force=True)

        Session().commit()

    def setup_method(self, method):

        self.u1 = UserModel().create_or_update(

            username='u1', password='qweqwe',

            email='u1@example.com', firstname='u1', lastname='u1'

        )

        self.u2 = UserModel().create_or_update(

            username='u2', password='qweqwe',

            email='u2@example.com', firstname='u2', lastname='u2'

        )