rattail/auth

   rattail/db/model.users

``rattail.auth``

================

.. automodule:: rattail.auth

   :members:

``rattail.db.model.users``

==========================

.. automodule:: rattail.db.model.users

  :members:

.. autoclass:: RoleImporter

   :members:

.. autoclass:: GlobalRoleImporter

   :members:

    def grant_permission(self, role, permission):

        """

        Grant a permission to the role.

        """

        if permission not in role.permissions:

            role.permissions.append(permission)

    def revoke_permission(self, role, permission):

        """

        Revoke a permission from the role.

        """

        if permission in role.permissions:

            role.permissions.remove(permission)

    def cache_permissions(self, session, principal,

                          include_guest=True,

                          include_authenticated=True):

        """

        Return a set of permission names, which represents all

        permissions effectively granted to the given principal.

        :param session: A SQLAlchemy session instance.

        :param principal: May be either a

           :class:`~rattail.db.model.users.User` or

           :class:`~rattail.db.model.users.Role` instance.  It is also

           expected that this may sometimes be ``None``, in which case

           the "Guest" role will typically be assumed.

        :param include_guest: Whether or not the "Guest" role should

           be included when checking permissions.  If ``False``, then

           Guest's permissions will *not* be consulted.

        :param include_authenticated: Whether or not the

           "Authenticated" role should be included when checking

           permissions.

        Note that if no ``principal`` is provided, and

        ``include_guest`` is set to ``False``, then no checks will

        actually be done, and the return value will be ``False``.

        """

        from rattail.db.auth import guest_role, authenticated_role

        # we will use any `roles` attribute which may be present.  in practice we

        # would be assuming a User in this case

        if hasattr(principal, 'roles'):

            roles = []

            for role in principal.roles:

                include = False

                if role.node_type:

                    if role.node_type == self.config.node_type():

                        include = True

                else:

                    include = True

                if include:

                    roles.append(role)

            # here our User assumption gets a little more explicit

            if include_authenticated:

                roles.append(authenticated_role(session))

        # otherwise a non-null principal is assumed to be a Role

        elif principal is not None:

            roles = [principal]

        # fallback assumption is "no roles"

        else:

            roles = []

        # maybe include guest roles

        if include_guest:

            roles.append(guest_role(session))

        # build the permissions cache

        cache = set()

        for role in roles:

            cache.update(role.permissions)

        return cache

    def has_permission(self, session, principal, permission,

                       include_guest=True,

                       include_authenticated=True):

        """

        Determine if a principal has been granted a permission.

        Under the hood this really just invokes

        :meth:`cache_permissions()` for the principal, and then checks

        to see if the given permission is included in that set.

        """

        perms = self.cache_permissions(session, principal,

                                       include_guest=include_guest,

                                       include_authenticated=include_authenticated)

        return permission in perms

from rattail.db import api, model

        auth = self.app.get_auth_handler()

        # add the GlobalRole importer, in place of Role, unless latter

        # is already present.

        if 'Role' not in importers:

            importers['Role'] = self.handler.get_importer('GlobalRole')

# -*- coding: utf-8 -*-

"""add role.sync_me

Revision ID: 8b78ef45a36c

Revises: 675a0034becc

Create Date: 2021-10-13 18:00:46.665524

"""

from __future__ import unicode_literals, absolute_import

# revision identifiers, used by Alembic.

revision = '8b78ef45a36c'

down_revision = '675a0034becc'

branch_labels = None

depends_on = None

from alembic import op

import sqlalchemy as sa

import rattail.db.types

def upgrade():

    # role

    op.add_column('role', sa.Column('node_type', sa.String(length=100), nullable=True))

    op.add_column('role', sa.Column('sync_me', sa.Boolean(), nullable=True))

    op.add_column('role_version', sa.Column('node_type', sa.String(length=100), autoincrement=False, nullable=True))

    op.add_column('role_version', sa.Column('sync_me', sa.Boolean(), autoincrement=False, nullable=True))

def downgrade():

    # role

    op.drop_column('role_version', 'sync_me')

    op.drop_column('role_version', 'node_type')

    op.drop_column('role', 'sync_me')

    op.drop_column('role', 'node_type')

#  Copyright © 2010-2021 Lance Edgar

    sync_me = sa.Column(sa.Boolean(), nullable=True, doc="""

    Flag indicating that this Role (and its user-ship, and

    permissions) should be synced across all nodes.

    So if set, when the role changes at one node that change should

    propagate to all other nodes.  This includes "proper" changes e.g.

    to the role name, but also when any users are added to or removed

    from the role, that fact also should propagate.  Additionally,

    when permissions are granted to or revoked from the role, that

    should propagate.

    See also :attr:`node_type`.

    """)

    node_type = sa.Column(sa.String(length=100), nullable=True, doc="""

    Type of node for which this role is applicable.  This is probably

    only useful if the :attr:`sync_me` flag is set.

    If set, this value must match a node's configured type, or else it

    will be ignored by that node.  See also

    :meth:`~rattail.config.RattailConfig.node_type()`.  If there is no

    value set for this field then the role will be honored by all

    nodes in which it exists (which is just one unless ``sync_me`` is

    set, in which case all nodes would have it).

    It is useful in combination with ``sync_me`` in that it allows a

    certain role to be "global" (synced) and yet only be "effective"

    for certain nodes.  Probably the most common scenario is where you

    have a host node and several store nodes, and you want to manage

    the store roles "globally" but avoid granting unwanted access to

    the host node.  So you'd set the ``sync_me`` flag but also set

    ``node_type`` to e.g. ``'store'``.

    """)

        self.app = config.get_app() if config else None

class RoleImporter(ToRattail):

    """

    Role data importer.

    """

    model_class = model.Role

        importers['Role'] = RoleImporter

        importers['GlobalRole'] = GlobalRoleImporter

            'Role',

            'GlobalRole',

class RoleImporter(FromRattail, model.RoleImporter):

    pass

class GlobalRoleImporter(RoleImporter):

    """

    Role importer which only will handle roles which have the

    :attr:`~rattail.db.model.users.Role.sync_me` flag set.  (So it

    syncs those roles but avoids others.)

    """

    @property

    def supported_fields(self):

        fields = list(super(GlobalRoleImporter, self).supported_fields)

        fields.extend([

            'permissions',

            'users',

        ])

        return fields

    # nb. we must override both cache_query() and query() b/c they use

    # different sessions!

    def cache_query(self):

        """

        Return the query to be used when caching "local" data.

        """

        query = super(GlobalRoleImporter, self).cache_query()

        model = self.model

        # only want roles which are *meant* to be synced

        query = query.filter(model.Role.sync_me == True)

        return query

    def query(self):

        query = super(GlobalRoleImporter, self).query()

        model = self.model

        # only want roles which are *meant* to be synced

        query = query.filter(model.Role.sync_me == True)

        return query

    # nb. we do not need to override normalize_host_object() b/c it

    # just calls normalize_local_object() by default

    def normalize_local_object(self, role):

        # only want roles which are *meant* to be synced

        if not role.sync_me:

            return

        data = super(GlobalRoleImporter, self).normalize_local_object(role)

        if data:

            # users

            if 'users' in self.fields:

                data['users'] = sorted([user.uuid for user in role.users])

            # permissions

            if 'permissions' in self.fields:

                auth = self.app.get_auth_handler()

                perms = auth.cache_permissions(self.session, role,

                                               include_guest=False)

                data['permissions'] = sorted(perms)

            return data

    def update_object(self, role, host_data, local_data=None, **kwargs):

        role = super(GlobalRoleImporter, self).update_object(role, host_data,

                                                             local_data=local_data,

                                                             **kwargs)

        model = self.model

        # users

        if 'users' in self.fields:

            new_users = host_data['users']

            old_users = local_data['users'] if local_data else []

            changed = False

            # add some users

            for new_user in new_users:

                if new_user not in old_users:

                    user = self.session.query(model.User).get(new_user)

                    user.roles.append(role)

                    changed = True

            # remove some users

            for old_user in old_users:

                if old_user not in new_users:

                    user = self.session.query(model.User).get(old_user)

                    user.roles.remove(role)

                    changed = True

            if changed:

                # also record a change to the role, for datasync.

                # this is done "just in case" the role is to be

                # synced to all nodes

                if self.session.rattail_record_changes:

                    self.session.add(model.Change(class_name='Role',

                                                  instance_uuid=role.uuid,

                                                  deleted=False))

        # permissions

        if 'permissions' in self.fields:

            auth = self.app.get_auth_handler()

            new_perms = host_data['permissions']

            old_perms = local_data['permissions'] if local_data else []

            # grant permissions

            for new_perm in new_perms:

                if new_perm not in old_perms:

                    auth.grant_permission(role, new_perm)

            # revoke permissions

            for old_perm in old_perms:

                if old_perm not in new_perms:

                    auth.revoke_permission(role, old_perm)

        return role