chezbetty.models.User.from_id() - Code Metrics - Inspection of "add support for adding purchases manually" - um-cseg/chez-betty - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ad73e3...2c5cfd )

unknown

created 2015-12-04 22:14 UTC

chezbetty.models.User.from_id() A

↳ Parent: chezbetty.models.User

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	3
rs	10

from pyramid.security import authenticated_userid
import hashlib
import binascii
import os
from .model import *
from . import account
from . import event
from chezbetty import utility

import ldap3
import random
import string

class InvalidUserException(Exception):
    pass


class LDAPLookup(object):
    """class that allows lookup of an individual in the UM directory
    based on Michigan ID number, uniqname, MCARD"""

    SERVER = None
    USERNAME = None
    BASE_DN = "ou=People,dc=umich,dc=edu"
    PASSWORD = None
    ATTRIBUTES = ["uid", "entityid", "displayName"]

    def __init__(self):
        self.__conn = None

    def __connect(self):
        if not self.__conn:
            s = ldap3.Server(self.SERVER, port=636, use_ssl=True, get_info=ldap3.GET_ALL_INFO)
            self.__conn = ldap3.Connection(s, auto_bind=True,
                    user=self.USERNAME, password=self.PASSWORD,
                    client_strategy=ldap3.STRATEGY_SYNC,
                    authentication=ldap3.AUTH_SIMPLE
            )


    def __lookup(self, k, v):
        self.__connect()
        query = "(%s=%s)" % (k, v)
        try:
            self.__conn.search(self.BASE_DN,
                    query,
                    ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
                    attributes=self.ATTRIBUTES
            )
        except:
            # sometimes our connections time out
            self.__conn = None
            self.__connect()
            self.__conn.search(self.BASE_DN,
                    query,
                    ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
                    attributes=self.ATTRIBUTES
            )

        if len(self.__conn.response) == 0:
            raise InvalidUserException()
        return {
            "umid":self.__conn.response[0]["attributes"]["entityid"],
            "uniqname":self.__conn.response[0]["attributes"]["uid"][0],
            "name":self.__conn.response[0]["attributes"]["displayName"][0]
        }

    def lookup_umid(self, umid):
        return self.__lookup("entityid", umid)

    def lookup_uniqname(self, uniqname):
        return self.__lookup("uid", uniqname)


class User(account.Account):
    __tablename__ = 'users'
    __mapper_args__ = {'polymorphic_identity': 'user'}

    id        = Column(Integer, ForeignKey("accounts.id"), primary_key=True)
    uniqname  = Column(String(8), nullable=False, unique=True)
    umid      = Column(String(8), unique=True)
    _password = Column("password", String(255))
    _salt     = Column("salt", String(255))
    enabled   = Column(Boolean, nullable=False, default=True)
    role      = Column(Enum("user", "serviceaccount", "manager", "administrator", name="user_type"),
                       nullable=False, default="user")

    administrative_events = relationship(event.Event, foreign_keys=[event.Event.user_id], backref="admin")
    events_deleted        = relationship(event.Event, foreign_keys=[event.Event.deleted_user_id], backref="deleted_user")
    __ldap = LDAPLookup()

    def __init__(self, uniqname, umid, name):
        self.enabled = True
        self.uniqname = uniqname
        self.umid = umid
        self.name = name
        self.balance = 0.0

    def __str__(self):
        return "<User: id {}, uniqname {}, umid {}, name {}, balance {}>".\
                format(self.id, self.uniqname, self.umid, self.name, self.balance)

    @classmethod
    def from_id(cls, id):
        return DBSession.query(cls).filter(cls.id == id).one()

    @classmethod
    def from_uniqname(cls, uniqname, local_only=False):
        u = DBSession.query(cls).filter(cls.uniqname == uniqname).first()
        if not u and not local_only:
            u = cls(**cls.__ldap.lookup_uniqname(uniqname))
            DBSession.add(u)
        return u

    @classmethod
    def from_umid(cls, umid, create_if_never_seen=False):
        u = DBSession.query(cls).filter(cls.umid == umid).first()
        if not u:
            if create_if_never_seen:
                u = cls(**cls.__ldap.lookup_umid(umid))
                DBSession.add(u)
            else:
                raise InvalidUserException()
        return u

    @classmethod
    def from_fuzzy(cls, search_str):
        return DBSession.query(cls)\
                        .filter(or_(
                            cls.uniqname.ilike('%{}%'.format(search_str)),
                            cls.umid.ilike('%{}%'.format(search_str)),
                            cls.name.ilike('%{}%'.format(search_str))
                        )).all()

    @classmethod
    def all(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .order_by(cls.name)\
                        .all()

    @classmethod
    def count(cls):
        return DBSession.query(func.count(cls.id).label('c')).one().c

    @classmethod
    def get_admins(cls):
        return DBSession.query(cls)\
                        .filter(cls.enabled)\
                        .filter(cls.role=='administrator').all()

    @classmethod
    def get_users_total(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .one().total_balance or Decimal(0.0)

    # Sum the total amount of money in user accounts that we are holding for
    # users. This is different from just getting the total because it doesn't
    # count users with negative balances
    @classmethod
    def get_amount_held(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .filter(User.balance>0)\
                        .one().total_balance or Decimal(0.0)

    @classmethod
    def get_amount_owed(cls):
        return DBSession.query(func.sum(User.balance).label("total_balance"))\
                        .filter(User.balance<0)\
                        .one().total_balance or Decimal(0.0)

    @classmethod
    def get_user_count_cumulative(cls):
        rows = DBSession.query(cls.created_at)\
                        .order_by(cls.created_at)\
                        .all()
        return utility.timeseries_cumulative(rows)

    def __make_salt(self):
        return binascii.b2a_base64(open("/dev/urandom", "rb").read(32))[:-3].decode("ascii")

    @hybrid_property
    def password(self):
        return self._password

    @password.setter
    def password(self, password):
        if password == '':
            # Use this to clear the password so the user can't login
            self._password = None
        else:
            self._salt = self.__make_salt()
            salted = (self._salt + password).encode('utf-8')
            self._password = hashlib.sha256(salted).hexdigest()

    def random_password(self):
        password = ''.join(random.choice(string.ascii_letters + string.digits) for x in range(6))
        self._salt = self.__make_salt()
        salted = (self._salt + password).encode('utf-8')
        self._password = hashlib.sha256(salted).hexdigest()
        return password

    def check_password(self, cand):
        if not self._salt:
            return False
        salted = (self._salt + cand).encode('utf-8')
        c = hashlib.sha256(salted).hexdigest()
        return c == self._password

    @property
    def has_password(self):
        return self._password != None


def get_user(request):
    login = authenticated_userid(request)
    if not login:
        return None
    return DBSession.query(User).filter(User.uniqname == login).one()


# This is in a stupid place due to circular input problems
@property
def __user_from_foreign_key(self):
    return User.from_id(self.user_id)
event.Event.user = __user_from_foreign_key


def groupfinder(userid, request):
    user = User.from_uniqname(userid)
    if user.role == "user":
        return ["user",]
    elif user.role == "manager":
        return ["user","manager"]
    elif user.role == "administrator":
        return ["user","manager","admin","serviceaccount"]
    elif user.role == "serviceaccount":
        return ["serviceaccount"]


1			from pyramid.security import authenticated_userid
2			import hashlib
3			import binascii
4			import os
5			from .model import *
6			from . import account
7			from . import event
8			from chezbetty import utility
9
10			import ldap3
11			import random
12			import string
13
14			class InvalidUserException(Exception):
15			pass
16
17
18			class LDAPLookup(object):
19			"""class that allows lookup of an individual in the UM directory
20			based on Michigan ID number, uniqname, MCARD"""
21
22			SERVER = None
23			USERNAME = None
24			BASE_DN = "ou=People,dc=umich,dc=edu"
25			PASSWORD = None
26			ATTRIBUTES = ["uid", "entityid", "displayName"]
27
28			def __init__(self):
29			self.__conn = None
30
31			def __connect(self):
32			if not self.__conn:
33			s = ldap3.Server(self.SERVER, port=636, use_ssl=True, get_info=ldap3.GET_ALL_INFO)
34			self.__conn = ldap3.Connection(s, auto_bind=True,
35			user=self.USERNAME, password=self.PASSWORD,
36			client_strategy=ldap3.STRATEGY_SYNC,
37			authentication=ldap3.AUTH_SIMPLE
38			)
39
40
41			def __lookup(self, k, v):
42			self.__connect()
43			query = "(%s=%s)" % (k, v)
44			try:
45			self.__conn.search(self.BASE_DN,
46			query,
47			ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
48			attributes=self.ATTRIBUTES
49			)
50			except:
51			# sometimes our connections time out
52			self.__conn = None
53			self.__connect()
54			self.__conn.search(self.BASE_DN,
55			query,
56			ldap3.SEARCH_SCOPE_WHOLE_SUBTREE,
57			attributes=self.ATTRIBUTES
58			)
59
60			if len(self.__conn.response) == 0:
61			raise InvalidUserException()
62			return {
63			"umid":self.__conn.response[0]["attributes"]["entityid"],
64			"uniqname":self.__conn.response[0]["attributes"]["uid"][0],
65			"name":self.__conn.response[0]["attributes"]["displayName"][0]
66			}
67
68			def lookup_umid(self, umid):
69			return self.__lookup("entityid", umid)
70
71			def lookup_uniqname(self, uniqname):
72			return self.__lookup("uid", uniqname)
73
74
75			class User(account.Account):
76			__tablename__ = 'users'
77			__mapper_args__ = {'polymorphic_identity': 'user'}
78
79			id = Column(Integer, ForeignKey("accounts.id"), primary_key=True)
80			uniqname = Column(String(8), nullable=False, unique=True)
81			umid = Column(String(8), unique=True)
82			_password = Column("password", String(255))
83			_salt = Column("salt", String(255))
84			enabled = Column(Boolean, nullable=False, default=True)
85			role = Column(Enum("user", "serviceaccount", "manager", "administrator", name="user_type"),
86			nullable=False, default="user")
87
88			administrative_events = relationship(event.Event, foreign_keys=[event.Event.user_id], backref="admin")
89			events_deleted = relationship(event.Event, foreign_keys=[event.Event.deleted_user_id], backref="deleted_user")
90			__ldap = LDAPLookup()
91
92			def __init__(self, uniqname, umid, name):
93			self.enabled = True
94			self.uniqname = uniqname
95			self.umid = umid
96			self.name = name
97			self.balance = 0.0
98
99			def __str__(self):
100			return "<User: id {}, uniqname {}, umid {}, name {}, balance {}>".\
101			format(self.id, self.uniqname, self.umid, self.name, self.balance)
102
103			@classmethod
104			def from_id(cls, id):
105			return DBSession.query(cls).filter(cls.id == id).one()
106
107			@classmethod
108			def from_uniqname(cls, uniqname, local_only=False):
109			u = DBSession.query(cls).filter(cls.uniqname == uniqname).first()
110			if not u and not local_only:
111			u = cls(**cls.__ldap.lookup_uniqname(uniqname))
112			DBSession.add(u)
113			return u
114
115			@classmethod
116			def from_umid(cls, umid, create_if_never_seen=False):
117			u = DBSession.query(cls).filter(cls.umid == umid).first()
118			if not u:
119			if create_if_never_seen:
120			u = cls(**cls.__ldap.lookup_umid(umid))
121			DBSession.add(u)
122			else:
123			raise InvalidUserException()
124			return u
125
126			@classmethod
127			def from_fuzzy(cls, search_str):
128			return DBSession.query(cls)\
129			.filter(or_(
130			cls.uniqname.ilike('%{}%'.format(search_str)),
131			cls.umid.ilike('%{}%'.format(search_str)),
132			cls.name.ilike('%{}%'.format(search_str))
133			)).all()
134
135			@classmethod
136			def all(cls):
137			return DBSession.query(cls)\
138			.filter(cls.enabled)\
139			.order_by(cls.name)\
140			.all()
141
142			@classmethod
143			def count(cls):
144			return DBSession.query(func.count(cls.id).label('c')).one().c
145
146			@classmethod
147			def get_admins(cls):
148			return DBSession.query(cls)\
149			.filter(cls.enabled)\
150			.filter(cls.role=='administrator').all()
151
152			@classmethod
153			def get_users_total(cls):
154			return DBSession.query(func.sum(User.balance).label("total_balance"))\
155			.one().total_balance or Decimal(0.0)
156
157			# Sum the total amount of money in user accounts that we are holding for
158			# users. This is different from just getting the total because it doesn't
159			# count users with negative balances
160			@classmethod
161			def get_amount_held(cls):
162			return DBSession.query(func.sum(User.balance).label("total_balance"))\
163			.filter(User.balance>0)\
164			.one().total_balance or Decimal(0.0)
165
166			@classmethod
167			def get_amount_owed(cls):
168			return DBSession.query(func.sum(User.balance).label("total_balance"))\
169			.filter(User.balance<0)\
170			.one().total_balance or Decimal(0.0)
171
172			@classmethod
173			def get_user_count_cumulative(cls):
174			rows = DBSession.query(cls.created_at)\
175			.order_by(cls.created_at)\
176			.all()
177			return utility.timeseries_cumulative(rows)
178
179			def __make_salt(self):
180			return binascii.b2a_base64(open("/dev/urandom", "rb").read(32))[:-3].decode("ascii")
181
182			@hybrid_property
183			def password(self):
184			return self._password
185
186			@password.setter
187			def password(self, password):
188			if password == '':
189			# Use this to clear the password so the user can't login
190			self._password = None
191			else:
192			self._salt = self.__make_salt()
193			salted = (self._salt + password).encode('utf-8')
194			self._password = hashlib.sha256(salted).hexdigest()
195
196			def random_password(self):
197			password = ''.join(random.choice(string.ascii_letters + string.digits) for x in range(6))
198			self._salt = self.__make_salt()
199			salted = (self._salt + password).encode('utf-8')
200			self._password = hashlib.sha256(salted).hexdigest()
201			return password
202
203			def check_password(self, cand):
204			if not self._salt:
205			return False
206			salted = (self._salt + cand).encode('utf-8')
207			c = hashlib.sha256(salted).hexdigest()
208			return c == self._password
209
210			@property
211			def has_password(self):
212			return self._password != None
213
214
215			def get_user(request):
216			login = authenticated_userid(request)
217			if not login:
218			return None
219			return DBSession.query(User).filter(User.uniqname == login).one()
220
221
222			# This is in a stupid place due to circular input problems
223			@property
224			def __user_from_foreign_key(self):
225			return User.from_id(self.user_id)
226			event.Event.user = __user_from_foreign_key
227
228
229			def groupfinder(userid, request):
230			user = User.from_uniqname(userid)
231			if user.role == "user":
232			return ["user",]
233			elif user.role == "manager":
234			return ["user","manager"]
235			elif user.role == "administrator":
236			return ["user","manager","admin","serviceaccount"]
237			elif user.role == "serviceaccount":
238			return ["serviceaccount"]
239

um-cseg / chez-betty

Push — master ( ad73e3...2c5cfd )

chezbetty.models.User.from_id() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like