<s> from __future__ import with_statement <EOL> from contextlib import contextmanager <EOL> from datetime import datetime <EOL> from UserDict import DictMixin <EOL> import bcrypt <EOL> from pyramid . location import lineage <EOL> from pyramid . security import view_execution_permitted <EOL> from six import string_types <EOL> from sqlalchemy import Boolean , bindparam <EOL> from sqlalchemy import Column <EOL> from sqlalchemy import DateTime <EOL> from sqlalchemy import func <EOL> from sqlalchemy import Integer <EOL> from sqlalchemy import Unicode <EOL> from sqlalchemy . orm . exc import NoResultFound <EOL> from sqlalchemy . sql . expression import and_ <EOL> from sqlalchemy . sql . expression import or_ <EOL> from zope . deprecation . deprecation import deprecated <EOL> from kotti import Base <EOL> from kotti import DBSession <EOL> from kotti import get_settings <EOL> from kotti . sqla import bakery <EOL> from kotti . sqla import JsonType <EOL> from kotti . sqla import MutationList <EOL> from kotti . util import _ <EOL> from kotti . util import request_cache <EOL> from kotti . util import DontCache <EOL> def get_principals ( ) : <EOL> return get_settings ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ( ) <EOL> @ request_cache ( lambda request : None ) <EOL> def get_user ( request ) : <EOL> userid = request . unauthenticated_userid <EOL> return get_principals ( ) . get ( userid ) <EOL> def has_permission ( permission , context , request ) : <EOL> """<STR_LIT>""" <EOL> return request . has_permission ( permission , context ) <EOL> deprecated ( u'<STR_LIT>' , <EOL> u"<STR_LIT>" <EOL> u"<STR_LIT>" <EOL> u"<STR_LIT>" ) <EOL> class Principal ( Base ) : <EOL> """<STR_LIT>""" <EOL> id = Column ( Integer , primary_key = True ) <EOL> name = Column ( Unicode ( <NUM_LIT:100> ) , unique = True ) <EOL> password = Column ( Unicode ( <NUM_LIT:100> ) ) <EOL> active = Column ( Boolean ) <EOL> confirm_token = Column ( Unicode ( <NUM_LIT:100> ) ) <EOL> title = Column ( Unicode ( <NUM_LIT:100> ) , nullable = False ) <EOL> email = Column ( Unicode ( <NUM_LIT:100> ) , unique = True ) <EOL> groups = Column ( MutationList . as_mutable ( JsonType ) , nullable = False ) <EOL> creation_date = Column ( DateTime ( ) , nullable = False ) <EOL> last_login_date = Column ( DateTime ( ) ) <EOL> __tablename__ = '<STR_LIT>' <EOL> __mapper_args__ = dict ( <EOL> order_by = name , <EOL> ) <EOL> def __init__ ( self , name , password = None , active = True , confirm_token = None , <EOL> title = u"<STR_LIT>" , email = None , groups = None ) : <EOL> self . name = name <EOL> if password is not None : <EOL> password = get_principals ( ) . hash_password ( password ) <EOL> self . password = password <EOL> self . active = active <EOL> self . confirm_token = confirm_token <EOL> self . title = title <EOL> self . email = email <EOL> if groups is None : <EOL> groups = [ ] <EOL> self . groups = groups <EOL> self . creation_date = datetime . now ( ) <EOL> self . last_login_date = None <EOL> def __repr__ ( self ) : <EOL> return u'<STR_LIT>' . format ( self . name ) <EOL> class AbstractPrincipals ( object ) : <EOL> """<STR_LIT>""" <EOL> def __getitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> def __setitem__ ( self , name , principal ) : <EOL> """<STR_LIT>""" <EOL> def __delitem__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> def keys ( self ) : <EOL> """<STR_LIT>""" <EOL> def search ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> def hash_password ( self , password ) : <EOL> """<STR_LIT>""" <EOL> def validate_password ( self , clear , hashed ) : <EOL> """<STR_LIT>""" <EOL> ROLES = { <EOL> u'<STR_LIT>' : Principal ( u'<STR_LIT>' , title = _ ( u'<STR_LIT>' ) ) , <EOL> u'<STR_LIT>' : Principal ( u'<STR_LIT>' , title = _ ( u'<STR_LIT>' ) ) , <EOL> u'<STR_LIT>' : Principal ( u'<STR_LIT>' , title = _ ( u'<STR_LIT>' ) ) , <EOL> u'<STR_LIT>' : Principal ( u'<STR_LIT>' , title = _ ( u'<STR_LIT>' ) ) , <EOL> } <EOL> _DEFAULT_ROLES = ROLES . copy ( ) <EOL> SHARING_ROLES = [ u'<STR_LIT>' , u'<STR_LIT>' , u'<STR_LIT>' ] <EOL> USER_MANAGEMENT_ROLES = SHARING_ROLES + [ '<STR_LIT>' ] <EOL> _DEFAULT_SHARING_ROLES = SHARING_ROLES [ : ] <EOL> _DEFAULT_USER_MANAGEMENT_ROLES = USER_MANAGEMENT_ROLES [ : ] <EOL> SITE_ACL = [ <EOL> [ '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ] , <EOL> ] <EOL> def set_roles ( roles_dict ) : <EOL> ROLES . clear ( ) <EOL> ROLES . update ( roles_dict ) <EOL> def set_sharing_roles ( role_names ) : <EOL> SHARING_ROLES [ : ] = role_names <EOL> def set_user_management_roles ( role_names ) : <EOL> USER_MANAGEMENT_ROLES [ : ] = role_names <EOL> def reset_roles ( ) : <EOL> ROLES . clear ( ) <EOL> ROLES . update ( _DEFAULT_ROLES ) <EOL> def reset_sharing_roles ( ) : <EOL> SHARING_ROLES [ : ] = _DEFAULT_SHARING_ROLES <EOL> def reset_user_management_roles ( ) : <EOL> USER_MANAGEMENT_ROLES [ : ] = _DEFAULT_USER_MANAGEMENT_ROLES <EOL> def reset ( ) : <EOL> reset_roles ( ) <EOL> reset_sharing_roles ( ) <EOL> reset_user_management_roles ( ) <EOL> class PersistentACLMixin ( object ) : <EOL> def _get_acl ( self ) : <EOL> if self . _acl is None : <EOL> raise AttributeError ( '<STR_LIT>' ) <EOL> return self . _acl <EOL> def _set_acl ( self , value ) : <EOL> self . _acl = value <EOL> def _del_acl ( self ) : <EOL> self . _acl = None <EOL> __acl__ = property ( _get_acl , _set_acl , _del_acl ) <EOL> def _cachekey_list_groups_raw ( name , context ) : <EOL> context_id = context is not None and getattr ( context , '<STR_LIT:id>' , id ( context ) ) <EOL> return name , context_id <EOL> @ request_cache ( _cachekey_list_groups_raw ) <EOL> def list_groups_raw ( name , context ) : <EOL> """<STR_LIT>""" <EOL> from kotti . resources import Node <EOL> if isinstance ( context , Node ) : <EOL> return set ( <EOL> r . group_name for r in context . local_groups <EOL> if r . principal_name == name <EOL> ) <EOL> return set ( ) <EOL> def list_groups ( name , context = None ) : <EOL> """<STR_LIT>""" <EOL> return list_groups_ext ( name , context ) [ <NUM_LIT:0> ] <EOL> def _cachekey_list_groups_ext ( name , context = None , _seen = None , _inherited = None ) : <EOL> if _seen is not None or _inherited is not None : <EOL> raise DontCache <EOL> else : <EOL> context_id = getattr ( context , '<STR_LIT:id>' , id ( context ) ) <EOL> return unicode ( name ) , context_id <EOL> @ request_cache ( _cachekey_list_groups_ext ) <EOL> def list_groups_ext ( name , context = None , _seen = None , _inherited = None ) : <EOL> name = unicode ( name ) <EOL> groups = set ( ) <EOL> recursing = _inherited is not None <EOL> _inherited = _inherited or set ( ) <EOL> principal = get_principals ( ) . get ( name ) <EOL> if principal is not None : <EOL> groups . update ( principal . groups ) <EOL> if context is not None or ( context is None and _seen is not None ) : <EOL> _inherited . update ( principal . groups ) <EOL> if _seen is None : <EOL> _seen = { name } <EOL> if context is not None : <EOL> items = lineage ( context ) <EOL> for idx , item in enumerate ( items ) : <EOL> group_names = [ i for i in list_groups_raw ( name , item ) <EOL> if i not in _seen ] <EOL> groups . update ( group_names ) <EOL> if recursing or idx != <NUM_LIT:0> : <EOL> _inherited . update ( group_names ) <EOL> new_groups = groups - _seen <EOL> _seen . update ( new_groups ) <EOL> for group_name in new_groups : <EOL> g , i = list_groups_ext ( <EOL> group_name , context , _seen = _seen , _inherited = _inherited ) <EOL> groups . update ( g ) <EOL> _inherited . update ( i ) <EOL> return list ( groups ) , list ( _inherited ) <EOL> def set_groups ( name , context , groups_to_set = ( ) ) : <EOL> """<STR_LIT>""" <EOL> from kotti . resources import LocalGroup <EOL> name = unicode ( name ) <EOL> context . local_groups = [ <EOL> lg for lg in context . local_groups <EOL> if lg . principal_name != name <EOL> ] + [ <EOL> LocalGroup ( context , name , unicode ( group_name ) ) <EOL> for group_name in groups_to_set <EOL> ] <EOL> def list_groups_callback ( name , request ) : <EOL> """<STR_LIT>""" <EOL> if not is_user ( name ) : <EOL> return None <EOL> if name in get_principals ( ) : <EOL> context = request . environ . get ( <EOL> '<STR_LIT>' , getattr ( request , '<STR_LIT>' , None ) ) <EOL> if context is None : <EOL> from kotti . resources import get_root <EOL> context = get_root ( request ) <EOL> return list_groups ( name , context ) <EOL> @ contextmanager <EOL> def authz_context ( context , request ) : <EOL> before = request . environ . pop ( '<STR_LIT>' , None ) <EOL> request . environ [ '<STR_LIT>' ] = context <EOL> try : <EOL> yield <EOL> finally : <EOL> del request . environ [ '<STR_LIT>' ] <EOL> if before is not None : <EOL> request . environ [ '<STR_LIT>' ] = before <EOL> @ contextmanager <EOL> def request_method ( request , method ) : <EOL> before = request . method <EOL> request . method = method <EOL> try : <EOL> yield <EOL> finally : <EOL> request . method = before <EOL> def view_permitted ( context , request , name = '<STR_LIT>' , method = '<STR_LIT:GET>' ) : <EOL> with authz_context ( context , request ) : <EOL> with request_method ( request , method ) : <EOL> return view_execution_permitted ( context , request , name ) <EOL> def principals_with_local_roles ( context , inherit = True ) : <EOL> """<STR_LIT>""" <EOL> principals = set ( ) <EOL> items = [ context ] <EOL> if inherit : <EOL> items = lineage ( context ) <EOL> for item in items : <EOL> principals . update ( <EOL> r . principal_name for r in item . local_groups <EOL> if not r . principal_name . startswith ( '<STR_LIT>' ) <EOL> ) <EOL> return list ( principals ) <EOL> def map_principals_with_local_roles ( context ) : <EOL> principals = get_principals ( ) <EOL> value = [ ] <EOL> for principal_name in principals_with_local_roles ( context ) : <EOL> try : <EOL> principal = principals [ principal_name ] <EOL> except KeyError : <EOL> continue <EOL> else : <EOL> all , inherited = list_groups_ext ( principal_name , context ) <EOL> value . append ( ( principal , ( all , inherited ) ) ) <EOL> return sorted ( value , key = lambda t : t [ <NUM_LIT:0> ] . name ) <EOL> def is_user ( principal ) : <EOL> if not isinstance ( principal , string_types ) : <EOL> principal = principal . name <EOL> return '<STR_LIT::>' not in principal <EOL> class Principals ( DictMixin ) : <EOL> """<STR_LIT>""" <EOL> factory = Principal <EOL> @ classmethod <EOL> def _principal_by_name ( cls , name ) : <EOL> query = bakery ( lambda session : session . query ( cls . factory ) . filter ( <EOL> cls . factory . name == bindparam ( '<STR_LIT:name>' ) ) ) <EOL> return query ( DBSession ( ) ) . params ( name = name ) . one ( ) <EOL> @ request_cache ( lambda self , name : unicode ( name ) ) <EOL> def __getitem__ ( self , name ) : <EOL> name = unicode ( name ) <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> raise KeyError ( name ) <EOL> try : <EOL> return self . _principal_by_name ( name ) <EOL> except NoResultFound : <EOL> raise KeyError ( name ) <EOL> def __setitem__ ( self , name , principal ) : <EOL> name = unicode ( name ) <EOL> if isinstance ( principal , dict ) : <EOL> principal = self . factory ( ** principal ) <EOL> DBSession . add ( principal ) <EOL> def __delitem__ ( self , name ) : <EOL> name = unicode ( name ) <EOL> try : <EOL> principal = self . _principal_by_name ( name ) <EOL> DBSession . delete ( principal ) <EOL> except NoResultFound : <EOL> raise KeyError ( name ) <EOL> def iterkeys ( self ) : <EOL> for ( principal_name , ) in DBSession . query ( self . factory . name ) : <EOL> yield principal_name <EOL> def keys ( self ) : <EOL> return list ( self . iterkeys ( ) ) <EOL> def search ( self , match = '<STR_LIT>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not kwargs : <EOL> return [ ] <EOL> filters = [ ] <EOL> for key , value in kwargs . items ( ) : <EOL> col = getattr ( self . factory , key ) <EOL> if isinstance ( value , string_types ) and '<STR_LIT:*>' in value : <EOL> value = value . replace ( '<STR_LIT:*>' , '<STR_LIT:%>' ) . lower ( ) <EOL> filters . append ( func . lower ( col ) . like ( value ) ) <EOL> else : <EOL> filters . append ( col == value ) <EOL> query = DBSession . query ( self . factory ) <EOL> if match == '<STR_LIT>' : <EOL> query = query . filter ( or_ ( * filters ) ) <EOL> elif match == '<STR_LIT:all>' : <EOL> query = query . filter ( and_ ( * filters ) ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return query <EOL> log_rounds = <NUM_LIT:10> <EOL> def hash_password ( self , password , hashed = None ) : <EOL> if hashed is None : <EOL> hashed = bcrypt . gensalt ( self . log_rounds ) <EOL> return unicode ( <EOL> bcrypt . hashpw ( password . encode ( '<STR_LIT:utf-8>' ) , hashed . encode ( '<STR_LIT:utf-8>' ) ) ) <EOL> def validate_password ( self , clear , hashed ) : <EOL> try : <EOL> return self . hash_password ( clear , hashed ) == hashed <EOL> except ValueError : <EOL> return False <EOL> def principals_factory ( ) : <EOL> return Principals ( ) </s>
