turbogears.identity.saprovider

1 import logging 2 3 from turbogears import config, identity 4 from turbogears.database import bind_metadata, session 5 from turbogears.util import load_class 6 7 from sqlalchemy.orm import class_mapper 8 try: 9 from sqlalchemy.exc import IntegrityError 10 except ImportError: # SQLAlchemy < 0.5 11 from sqlalchemy.exceptions import IntegrityError 12 13 14 log = logging.getLogger("turbogears.identity.saprovider") 15 16 17 # Global class references -- 18 # these will be set when the provider is initialized. 19 user_class = None 20 group_class = None 21 permission_class = None 22 visit_class = None

23 24 25 -class SqlAlchemyIdentity(object):

26 """Identity that uses a model from a database (via SQLAlchemy).""" 27

28 - def __init__(self, visit_key=None, user=None):

29 self.visit_key = visit_key 30 if user: 31 self._user = user 32 if visit_key is not None: 33 self.login()

34 35 @property

36 - def user(self):

37 """Get user instance for this identity.""" 38 try: 39 return self._user 40 except AttributeError: 41 # User hasn't already been set 42 pass 43 # Attempt to load the user. After this code executes, there *will* be 44 # a _user attribute, even if the value is None. 45 visit = self.visit_link 46 self._user = visit and session.query(user_class).get(visit.user_id) 47 return self._user

48 49 @property

50 - def user_name(self):

51 """Get user name of this identity.""" 52 if not self.user: 53 return None 54 return self.user.user_name

55 56 @property

57 - def user_id(self):

58 """Get user id of this identity.""" 59 if not self.user: 60 return None 61 return self.user.user_id

62 63 @property

64 - def anonymous(self):

65 """Return true if not logged in.""" 66 return not self.user

67 68 @property

69 - def permissions(self):

70 """Get set of permission names of this identity.""" 71 try: 72 return self._permissions 73 except AttributeError: 74 # Permissions haven't been computed yet 75 pass 76 if not self.user: 77 self._permissions = frozenset() 78 else: 79 self._permissions = frozenset( 80 p.permission_name for p in self.user.permissions) 81 return self._permissions

82 83 @property

84 - def groups(self):

85 """Get set of group names of this identity.""" 86 try: 87 return self._groups 88 except AttributeError: 89 # Groups haven't been computed yet 90 pass 91 if not self.user: 92 self._groups = frozenset() 93 else: 94 self._groups = frozenset(g.group_name for g in self.user.groups) 95 return self._groups

96 97 @property

98 - def group_ids(self):

99 """Get set of group IDs of this identity.""" 100 try: 101 return self._group_ids 102 except AttributeError: 103 # Groups haven't been computed yet 104 pass 105 if not self.user: 106 self._group_ids = frozenset() 107 else: 108 self._group_ids = frozenset(g.group_id for g in self.user.groups) 109 return self._group_ids

110 111 @property

112 - def visit_link(self):

113 """Get the visit link to this identity.""" 114 if self.visit_key is None: 115 return None 116 return session.query(visit_class).filter_by( 117 visit_key=self.visit_key).first()

118 119 @property

120 - def login_url(self):

121 """Get the URL for the login page.""" 122 return identity.get_failure_url()

123

125 """Set the link between this identity and the visit.""" 126 visit = self.visit_link 127 if not visit: 128 visit = visit_class() 129 visit.visit_key = self.visit_key 130 session.add(visit) 131 visit.user_id = self._user.user_id 132 try: 133 session.flush() 134 except IntegrityError: 135 visit = self.visit_link 136 if not visit: 137 raise 138 visit.user_id = self._user.user_id 139 session.flush()

140

141 - def logout(self):

142 """Remove the link between this identity and the visit.""" 143 visit = self.visit_link 144 if visit: 145 session.delete(visit) 146 session.flush() 147 # Clear the current identity 148 identity.set_current_identity(SqlAlchemyIdentity())

149

150 151 -class SqlAlchemyIdentityProvider(object):

152 """IdentityProvider that uses a model from a database (via SQLAlchemy).""" 153

154 - def __init__(self):

155 super(SqlAlchemyIdentityProvider, self).__init__() 156 glob_ns = globals() 157 158 for classname in ('user', 'group', 'permission', 'visit'): 159 default_classname = '.TG_' + (classname == 'visit' 160 and 'VisitIdentity' or classname.capitalize()) 161 class_path = config.get("identity.saprovider.model.%s" % classname, 162 __name__ + default_classname) 163 class_ = load_class(class_path) 164 if class_: 165 log.info('Successfully loaded "%s".', class_path) 166 glob_ns['%s_class' % classname] = class_ 167 else: 168 log.error('Could not load class "%s". Check ' 169 'identity.saprovider.model.%s setting', class_path, classname)

170

171 - def encrypt_password(self, password):

172 # Default encryption algorithm is to use plain text passwords 173 algorithm = config.get("identity.saprovider.encryption_algorithm", None) 174 return identity.encrypt_pw_with_algorithm(algorithm, password)

175

176 - def create_provider_model(self):

177 """Create the database tables if they don't already exist.""" 178 bind_metadata() 179 class_mapper(user_class).local_table.create(checkfirst=True) 180 class_mapper(group_class).local_table.create(checkfirst=True) 181 class_mapper(permission_class).local_table.create(checkfirst=True) 182 class_mapper(visit_class).local_table.create(checkfirst=True)

183

184 - def validate_identity(self, user_name, password, visit_key):

185 """Validate the identity represented by user_name using the password. 186 187 Must return either None if the credentials weren't valid or an object 188 with the following properties: 189 user_name: original user name 190 user: a provider dependent object (TG_User or similar) 191 groups: a set of group names 192 permissions: a set of permission names 193 194 """ 195 user = session.query(user_class).filter_by( 196 user_name=user_name).first() 197 198 if not user: 199 log.warning("No such user: %s", user_name) 200 return None 201 202 if not self.validate_password(user, user_name, password): 203 log.info("Passwords don't match for user: %s", user_name) 204 return None 205 206 log.info("Associating user (%s) with visit (%s)", 207 user_name, visit_key) 208 return SqlAlchemyIdentity(visit_key, user)

209

210 - def validate_password(self, user, user_name, password):

211 """Check the user_name and password against existing credentials. 212 213 Note: user_name is not used here, but is required by external 214 password validation schemes that might override this method. 215 If you use SqlAlchemyIdentityProvider, but want to check the passwords 216 against an external source (i.e. PAM, LDAP, Windows domain, etc), 217 subclass SqlAlchemyIdentityProvider, and override this method. 218 219 """ 220 return user.password == self.encrypt_password(password)

221

222 - def load_identity(self, visit_key):

223 """Lookup the principal represented by user_name. 224 225 Return None if there is no principal for the given user ID. 226 227 Must return an object with the following properties: 228 user_name: original user name 229 user: a provider dependent object (TG_User or similar) 230 groups: a set of group names 231 permissions: a set of permission names 232 233 """ 234 return SqlAlchemyIdentity(visit_key)

235

236 - def anonymous_identity(self):

237 """Return anonymous identity. 238 239 Must return an object with the following properties: 240 user_name: original user name 241 user: a provider dependent object (TG_User or similar) 242 groups: a set of group names 243 permissions: a set of permission names 244 245 """ 246 return SqlAlchemyIdentity()

247

248 - def authenticated_identity(self, user):

249 """Construct Identity object for users with no visit_key.""" 250 return SqlAlchemyIdentity(user=user)

251

Source Code for Module turbogears.identity.saprovider