codekingpro/portable-devtools

1# -*- coding: utf-8 -*-2"""3    flask_principal4    ~~~~~~~~~~~~~~~5 6    Identity management for Flask.7 8    :copyright: (c) 2012 by Ali Afshar.9    :license: MIT, see LICENSE for more details.10 11"""12 13from __future__ import with_statement14 15__version__ = '0.4.0'16 17import sys18 19from functools import partial, wraps20from collections import deque21 22from collections import namedtuple23 24from flask import g, session, current_app, abort, request25from flask.signals import Namespace26 27PY3 = sys.version_info[0] == 328 29signals = Namespace()30 31 32identity_changed = signals.signal('identity-changed', doc="""33Signal sent when the identity for a request has been changed.34 35Actual name: ``identity-changed``36 37Authentication providers should send this signal when authentication has been38successfully performed. Flask-Principal connects to this signal and39causes the identity to be saved in the session.40 41For example::42 43    from flaskext.principal import Identity, identity_changed44 45    def login_view(req):46        username = req.form.get('username')47        # check the credentials48        identity_changed.send(app, identity=Identity(username))49""")50 51 52identity_loaded = signals.signal('identity-loaded', doc="""53Signal sent when the identity has been initialised for a request.54 55Actual name: ``identity-loaded``56 57Identity information providers should connect to this signal to perform two58major activities:59 60    1. Populate the identity object with the necessary authorization provisions.61    2. Load any additional user information.62 63For example::64 65    from flaskext.principal import identity_loaded, RoleNeed, UserNeed66 67    @identity_loaded.connect68    def on_identity_loaded(sender, identity):69        # Get the user information from the db70        user = db.get(identity.name)71        # Update the roles that a user can provide72        for role in user.roles:73            identity.provides.add(RoleNeed(role.name))74        # Save the user somewhere so we only look it up once75        identity.user = user76""")77 78 79Need = namedtuple('Need', ['method', 'value'])80"""A required need81 82This is just a named tuple, and practically any tuple will do.83 84The ``method`` attribute can be used to look up element 0, and the ``value``85attribute can be used to look up element 1.86"""87 88 89UserNeed = partial(Need, 'id')90UserNeed.__doc__ = """A need with the method preset to `"id"`."""91 92 93RoleNeed = partial(Need, 'role')94RoleNeed.__doc__ = """A need with the method preset to `"role"`."""95 96 97TypeNeed = partial(Need, 'type')98TypeNeed.__doc__ = """A need with the method preset to `"type"`."""99 100 101ActionNeed = partial(Need, 'action')102TypeNeed.__doc__ = """A need with the method preset to `"action"`."""103 104 105ItemNeed = namedtuple('ItemNeed', ['method', 'value', 'type'])106"""A required item need107 108An item need is just a named tuple, and practically any tuple will do. In109addition to other Needs, there is a type, for example this could be specified110as::111 112    ItemNeed('update', 27, 'posts')113    ('update', 27, 'posts') # or like this114 115And that might describe the permission to update a particular blog post. In116reality, the developer is free to choose whatever convention the permissions117are.118"""119 120 121class PermissionDenied(RuntimeError):122    """Permission denied to the resource"""123 124 125class Identity(object):126    """Represent the user's identity.127 128    :param id: The user id129    :param auth_type: The authentication type used to confirm the user's130                      identity.131 132    The identity is used to represent the user's identity in the system. This133    object is created on login, or on the start of the request as loaded from134    the user's session.135 136    Once loaded it is sent using the `identity-loaded` signal, and should be137    populated with additional required information.138 139    Needs that are provided by this identity should be added to the `provides`140    set after loading.141    """142    def __init__(self, id, auth_type=None):143        self.id = id144        self.auth_type = auth_type145        self.provides = set()146 147    def can(self, permission):148        """Whether the identity has access to the permission.149 150        :param permission: The permission to test provision for.151        """152        return permission.allows(self)153 154    def __repr__(self):155        return '<{0} id="{1}" auth_type="{2}" provides={3}>'.format(156            self.__class__.__name__, self.id, self.auth_type, self.provides157        )158 159 160class AnonymousIdentity(Identity):161    """An anonymous identity"""162 163    def __init__(self):164        Identity.__init__(self, None)165 166 167class IdentityContext(object):168    """The context of an identity for a permission.169 170    .. note:: The principal is usually created by the flaskext.Permission.require method171              call for normal use-cases.172 173    The principal behaves as either a context manager or a decorator. The174    permission is checked for provision in the identity, and if available the175    flow is continued (context manager) or the function is executed (decorator).176    """177 178    def __init__(self, permission, http_exception=None):179        self.permission = permission180        self.http_exception = http_exception181        """The permission of this principal182        """183 184    @property185    def identity(self):186        """The identity of this principal187        """188        return g.identity189 190    def can(self):191        """Whether the identity has access to the permission192        """193        return self.identity.can(self.permission)194 195    def __call__(self, f):196        @wraps(f)197        def _decorated(*args, **kw):198            with self:199                rv = f(*args, **kw)200            return rv201        return _decorated202 203    def __enter__(self):204        # check the permission here205        if not self.can():206            if self.http_exception:207                abort(self.http_exception, self.permission)208            raise PermissionDenied(self.permission)209 210    def __exit__(self, *args):211        return False212 213 214class Permission(object):215    """Represents needs, any of which must be present to access a resource216 217    :param needs: The needs for this permission218    """219    def __init__(self, *needs):220        """A set of needs, any of which must be present in an identity to have221        access.222        """223 224        self.needs = set(needs)225        self.excludes = set()226 227    def _bool(self):228        return bool(self.can())229 230    def __nonzero__(self):231        """Equivalent to ``self.can()``.232        """233        return self._bool()234 235    def __bool__(self):236        """Equivalent to ``self.can()``.237        """238        return self._bool()239 240    def __and__(self, other):241        """Does the same thing as ``self.union(other)``242        """243        return self.union(other)244 245    def __or__(self, other):246        """Does the same thing as ``self.difference(other)``247        """248        return self.difference(other)249 250    def __contains__(self, other):251        """Does the same thing as ``other.issubset(self)``.252        """253        return other.issubset(self)254 255    def __repr__(self):256        return '<{0} needs={1} excludes={2}>'.format(257            self.__class__.__name__, self.needs, self.excludes258        )259 260    def require(self, http_exception=None):261        """Create a principal for this permission.262 263        The principal may be used as a context manager, or a decroator.264 265        If ``http_exception`` is passed then ``abort()`` will be called266        with the HTTP exception code. Otherwise a ``PermissionDenied``267        exception will be raised if the identity does not meet the268        requirements.269 270        :param http_exception: the HTTP exception code (403, 401 etc)271        """272        return IdentityContext(self, http_exception)273 274    def test(self, http_exception=None):275        """276        Checks if permission available and raises relevant exception277        if not. This is useful if you just want to check permission278        without wrapping everything in a require() block.279 280        This is equivalent to::281 282            with permission.require():283                pass284        """285 286        with self.require(http_exception):287            pass288 289    def reverse(self):290        """291        Returns reverse of current state (needs->excludes, excludes->needs)292        """293 294        p = Permission()295        p.needs.update(self.excludes)296        p.excludes.update(self.needs)297        return p298 299    def union(self, other):300        """Create a new permission with the requirements of the union of this301        and other.302 303        :param other: The other permission304        """305        p = Permission(*self.needs.union(other.needs))306        p.excludes.update(self.excludes.union(other.excludes))307        return p308 309    def difference(self, other):310        """Create a new permission consisting of requirements in this311        permission and not in the other.312        """313 314        p = Permission(*self.needs.difference(other.needs))315        p.excludes.update(self.excludes.difference(other.excludes))316        return p317 318    def issubset(self, other):319        """Whether this permission needs are a subset of another320 321        :param other: The other permission322        """323        return (324            self.needs.issubset(other.needs) and325            self.excludes.issubset(other.excludes)326        )327 328    def allows(self, identity):329        """Whether the identity can access this permission.330 331        :param identity: The identity332        """333        if self.needs and not self.needs.intersection(identity.provides):334            return False335 336        if self.excludes and self.excludes.intersection(identity.provides):337            return False338 339        return True340 341    def can(self):342        """Whether the required context for this permission has access343 344        This creates an identity context and tests whether it can access this345        permission346        """347        return self.require().can()348 349 350class Denial(Permission):351    """352    Shortcut class for passing excluded needs.353    """354 355    def __init__(self, *excludes):356        self.excludes = set(excludes)357        self.needs = set()358 359 360def session_identity_loader():361    if 'identity.id' in session and 'identity.auth_type' in session:362        identity = Identity(session['identity.id'],363                            session['identity.auth_type'])364        return identity365 366 367def session_identity_saver(identity):368    session['identity.id'] = identity.id369    session['identity.auth_type'] = identity.auth_type370    session.modified = True371 372 373class Principal(object):374    """Principal extension375 376    :param app: The flask application to extend377    :param use_sessions: Whether to use sessions to extract and store378                         identification.379    :param skip_static: Whether to ignore static endpoints.380    """381    def __init__(self, app=None, use_sessions=True, skip_static=False):382        self.identity_loaders = deque()383        self.identity_savers = deque()384        # XXX This will probably vanish for a better API385        self.use_sessions = use_sessions386        self.skip_static = skip_static387 388        if app is not None:389            self.init_app(app)390 391    def _init_app(self, app):392        from warnings import warn393        warn(DeprecationWarning(394            '_init_app is deprecated, use the new init_app '395            'method instead.'), stacklevel=1396        )397        self.init_app(app)398 399    def init_app(self, app):400        if hasattr(app, 'static_url_path'):401            self._static_path = app.static_url_path402        else:403            self._static_path = app.static_path404 405        app.before_request(self._on_before_request)406        identity_changed.connect(self._on_identity_changed, app)407 408        if self.use_sessions:409            self.identity_loader(session_identity_loader)410            self.identity_saver(session_identity_saver)411 412    def set_identity(self, identity):413        """Set the current identity.414 415        :param identity: The identity to set416        """417 418        self._set_thread_identity(identity)419        for saver in self.identity_savers:420            saver(identity)421 422    def identity_loader(self, f):423        """Decorator to define a function as an identity loader.424 425        An identity loader function is called before request to find any426        provided identities. The first found identity is used to load from.427 428        For example::429 430            app = Flask(__name__)431 432            principals = Principal(app)433 434            @principals.identity_loader435            def load_identity_from_weird_usecase():436                return Identity('ali')437        """438        self.identity_loaders.appendleft(f)439        return f440 441    def identity_saver(self, f):442        """Decorator to define a function as an identity saver.443 444        An identity loader saver is called when the identity is set to persist445        it for the next request.446 447        For example::448 449            app = Flask(__name__)450 451            principals = Principal(app)452 453            @principals.identity_saver454            def save_identity_to_weird_usecase(identity):455                my_special_cookie['identity'] = identity456        """457        self.identity_savers.appendleft(f)458        return f459 460    def _set_thread_identity(self, identity):461        g.identity = identity462        identity_loaded.send(current_app._get_current_object(),463                             identity=identity)464 465    def _on_identity_changed(self, app, identity):466        if self._is_static_route():467            return468 469        self.set_identity(identity)470 471    def _on_before_request(self):472        if self._is_static_route():473            return474 475        g.identity = AnonymousIdentity()476        for loader in self.identity_loaders:477            identity = loader()478            if identity is not None:479                self.set_identity(identity)480                return481 482    def _is_static_route(self):483        return (484            self.skip_static and485            request.path.startswith(self._static_path)486        )487