/usr/share/pyshared/social_auth/db/base.py is in python-django-social-auth 0.7.23-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 | """Models mixins for Social Auth"""
import base64
import time
import re
from datetime import datetime, timedelta
from openid.association import Association as OIDAssociation
# django.contrib.auth and mongoengine.django.auth regex to validate usernames
# '^[\w@.+-_]+$', we use the opposite to clean invalid characters
CLEAN_USERNAME_REGEX = re.compile(r'[^\w.@+-_]+', re.UNICODE)
class UserSocialAuthMixin(object):
user = ''
provider = ''
def __unicode__(self):
"""Return associated user unicode representation"""
return u'%s - %s' % (unicode(self.user), self.provider.title())
def get_backend(self):
# Make import here to avoid recursive imports :-/
from social_auth.backends import get_backends
return get_backends().get(self.provider)
@property
def tokens(self):
"""Return access_token stored in extra_data or None"""
backend = self.get_backend()
if backend:
return backend.AUTH_BACKEND.tokens(self)
else:
return {}
def refresh_token(self):
data = self.extra_data
if 'refresh_token' in data or 'access_token' in data:
backend = self.get_backend()
if hasattr(backend, 'refresh_token'):
token = data.get('refresh_token') or data.get('access_token')
response = backend.refresh_token(token)
self.extra_data.update(
backend.AUTH_BACKEND.extra_data(self.user, self.uid,
response)
)
self.save()
def expiration_datetime(self):
"""Return provider session live seconds. Returns a timedelta ready to
use with session.set_expiry().
If provider returns a timestamp instead of session seconds to live, the
timedelta is inferred from current time (using UTC timezone). None is
returned if there's no value stored or it's invalid.
"""
if self.extra_data and 'expires' in self.extra_data:
try:
expires = int(self.extra_data['expires'])
except (ValueError, TypeError):
return None
now = datetime.utcnow()
# Detect if expires is a timestamp
if expires > time.mktime(now.timetuple()):
# expires is a datetime
return datetime.fromtimestamp(expires) - now
else:
# expires is a timedelta
return timedelta(seconds=expires)
@classmethod
def user_model(cls):
raise NotImplementedError('Implement in subclass')
@classmethod
def username_max_length(cls):
raise NotImplementedError('Implement in subclass')
@classmethod
def email_max_length(cls):
raise NotImplementedError('Implement in subclass')
@classmethod
def clean_username(cls, value):
return CLEAN_USERNAME_REGEX.sub('', value)
@classmethod
def allowed_to_disconnect(cls, user, backend_name, association_id=None):
if association_id is not None:
qs = cls.objects.exclude(id=association_id)
else:
qs = cls.objects.exclude(provider=backend_name)
qs = qs.filter(user=user)
if hasattr(user, 'has_usable_password'):
valid_password = user.has_usable_password()
else:
valid_password = True
return valid_password or qs.count() > 0
@classmethod
def user_username(cls, user):
if hasattr(user, 'USERNAME_FIELD'):
# Django 1.5 custom user model, 'username' is just for internal
# use, doesn't imply that the model should have an username field
field_name = user.USERNAME_FIELD
else:
field_name = 'username'
return getattr(user, field_name)
@classmethod
def username_field(cls, values):
user_model = cls.user_model()
if hasattr(user_model, 'USERNAME_FIELD'):
# Django 1.5 custom user model, 'username' is just for internal
# use, doesn't imply that the model should have an username field
values[user_model.USERNAME_FIELD] = values.pop('username')
return values
@classmethod
def simple_user_exists(cls, *args, **kwargs):
"""
Return True/False if a User instance exists with the given arguments.
Arguments are directly passed to filter() manager method.
TODO: consider how to ensure case-insensitive email matching
"""
kwargs = cls.username_field(kwargs)
return cls.user_model().objects.filter(*args, **kwargs).count() > 0
@classmethod
def create_user(cls, *args, **kwargs):
kwargs = cls.username_field(kwargs)
return cls.user_model().objects.create_user(*args, **kwargs)
@classmethod
def get_user(cls, pk):
try:
return cls.user_model().objects.get(pk=pk)
except cls.user_model().DoesNotExist:
return None
@classmethod
def get_user_by_email(cls, email):
"Case insensitive search"
return cls.user_model().objects.get(email__iexact=email)
@classmethod
def resolve_user_or_id(cls, user_or_id):
if isinstance(user_or_id, cls.user_model()):
return user_or_id
return cls.user_model().objects.get(pk=user_or_id)
@classmethod
def get_social_auth(cls, provider, uid):
if not isinstance(uid, basestring):
uid = str(uid)
try:
return cls.objects.get(provider=provider, uid=uid)
except cls.DoesNotExist:
return None
@classmethod
def get_social_auth_for_user(cls, user):
return user.social_auth.all()
@classmethod
def create_social_auth(cls, user, uid, provider):
if not isinstance(uid, basestring):
uid = str(uid)
return cls.objects.create(user=user, uid=uid, provider=provider)
@classmethod
def store_association(cls, server_url, association):
from social_auth.models import Association
args = {'server_url': server_url, 'handle': association.handle}
try:
assoc = Association.objects.get(**args)
except Association.DoesNotExist:
assoc = Association(**args)
assoc.secret = base64.encodestring(association.secret)
assoc.issued = association.issued
assoc.lifetime = association.lifetime
assoc.assoc_type = association.assoc_type
assoc.save()
@classmethod
def remove_association(cls, server_url, handle):
from social_auth.models import Association
assocs = list(Association.objects.filter(
server_url=server_url, handle=handle))
assocs_exist = len(assocs) > 0
for assoc in assocs:
assoc.delete()
return assocs_exist
@classmethod
def get_oid_associations(cls, server_url, handle=None):
from social_auth.models import Association
args = {'server_url': server_url}
if handle is not None:
args['handle'] = handle
return sorted([
(assoc.id,
OIDAssociation(assoc.handle,
base64.decodestring(assoc.secret),
assoc.issued,
assoc.lifetime,
assoc.assoc_type))
for assoc in Association.objects.filter(**args)
], key=lambda x: x[1].issued, reverse=True)
@classmethod
def delete_associations(cls, ids_to_delete):
from social_auth.models import Association
Association.objects.filter(pk__in=ids_to_delete).delete()
@classmethod
def use_nonce(cls, server_url, timestamp, salt):
from social_auth.models import Nonce
return Nonce.objects.get_or_create(server_url=server_url,
timestamp=timestamp,
salt=salt)[1]
class NonceMixin(object):
"""One use numbers"""
server_url = ''
timestamp = 0
salt = ''
def __unicode__(self):
"""Unicode representation"""
return self.server_url
class AssociationMixin(object):
"""OpenId account association"""
server_url = ''
handle = ''
secret = ''
issued = 0
lifetime = 0
assoc_type = ''
def __unicode__(self):
"""Unicode representation"""
return '%s %s' % (self.handle, self.issued)
|