aiohttp-security/demo/database_auth/db_auth.py

63 lines
2.1 KiB
Python
Raw Normal View History

2015-11-21 07:37:44 +00:00
import sqlalchemy as sa
2015-07-08 17:30:24 +00:00
2016-02-01 17:25:14 +00:00
from aiohttp_security.abc import AbstractAuthorizationPolicy
from passlib.hash import sha256_crypt
2015-07-08 17:30:24 +00:00
2015-11-21 07:37:44 +00:00
from . import db
2015-07-08 17:30:24 +00:00
2015-11-21 06:45:08 +00:00
class DBAuthorizationPolicy(AbstractAuthorizationPolicy):
2015-11-26 18:09:00 +00:00
def __init__(self, dbengine):
self.dbengine = dbengine
2015-07-08 17:30:24 +00:00
def authorized_userid(self, identity):
2017-12-13 14:51:46 +00:00
async with self.dbengine as conn:
where = sa.and_(db.users.c.login == identity,
sa.not_(db.users.c.disabled))
query = db.users.count().where(where)
2017-12-13 14:51:46 +00:00
ret = await conn.scalar(query)
2015-11-21 07:37:44 +00:00
if ret:
return identity
else:
return None
2017-12-13 14:51:46 +00:00
async def permits(self, identity, permission, context=None):
if identity is None:
return False
2017-12-13 14:51:46 +00:00
async with self.dbengine as conn:
where = sa.and_(db.users.c.login == identity,
sa.not_(db.users.c.disabled))
query = db.users.select().where(where)
2017-12-13 14:51:46 +00:00
ret = await conn.execute(query)
user = await ret.fetchone()
if user is not None:
user_id = user[0]
is_superuser = user[3]
if is_superuser:
return True
where = db.permissions.c.user_id == user_id
query = db.permissions.select().where(where)
2017-12-13 14:51:46 +00:00
ret = await conn.execute(query)
result = await ret.fetchall()
if ret is not None:
for record in result:
if record.perm_name == permission:
return True
return False
2017-12-13 14:51:46 +00:00
async def check_credentials(db_engine, username, password):
async with db_engine as conn:
where = sa.and_(db.users.c.login == username,
sa.not_(db.users.c.disabled))
query = db.users.select().where(where)
2017-12-13 14:51:46 +00:00
ret = await conn.execute(query)
user = await ret.fetchone()
if user is not None:
hash = user[2]
return sha256_crypt.verify(password, hash)
return False