auth.py
75 lines
| 2.3 KiB
| text/x-python
|
PythonLexer
r190 | from functools import wraps | |||
r52 | from pylons import session, url | |||
from pylons.controllers.util import abort, redirect | ||||
|
r64 | from pylons_app.model import meta | ||
r197 | from pylons_app.model.db import Users | |||
r190 | from sqlalchemy.exc import OperationalError | |||
|
r64 | from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound | ||
r190 | import crypt | |||
import logging | ||||
log = logging.getLogger(__name__) | ||||
|
r41 | |||
|
r64 | def get_crypt_password(password): | ||
r190 | """ | |||
Cryptographic function used for password hashing | ||||
@param password: password to hash | ||||
""" | ||||
|
r64 | return crypt.crypt(password, '6a') | ||
|
r46 | |||
|
r41 | def authfunc(environ, username, password): | ||
|
r64 | sa = meta.Session | ||
password_crypt = get_crypt_password(password) | ||||
r42 | try: | |||
|
r64 | user = sa.query(Users).filter(Users.username == username).one() | ||
except (NoResultFound, MultipleResultsFound, OperationalError) as e: | ||||
r42 | log.error(e) | |||
|
r64 | user = None | ||
if user: | ||||
if user.active: | ||||
if user.username == username and user.password == password_crypt: | ||||
|
r41 | log.info('user %s authenticated correctly', username) | ||
return True | ||||
else: | ||||
log.error('user %s is disabled', username) | ||||
return False | ||||
r190 | class AuthUser(object): | |||
""" | ||||
A simple object that handles a mercurial username for authentication | ||||
""" | ||||
r199 | username = 'None' | |||
r190 | is_authenticated = False | |||
is_admin = False | ||||
permissions = set() | ||||
group = set() | ||||
def __init__(self): | ||||
pass | ||||
#=============================================================================== | ||||
# DECORATORS | ||||
#=============================================================================== | ||||
class LoginRequired(object): | ||||
""" | ||||
Must be logged in to execute this function else redirect to login page | ||||
""" | ||||
def __init__(self): | ||||
pass | ||||
def __call__(self, func): | ||||
@wraps(func) | ||||
def _wrapper(*fargs, **fkwargs): | ||||
r199 | user = session.get('hg_app_user', AuthUser()) | |||
log.info('Checking login required for user:%s', user.username) | ||||
r190 | if user.is_authenticated: | |||
log.info('user %s is authenticated', user.username) | ||||
func(*fargs) | ||||
else: | ||||
logging.info('user %s not authenticated', user.username) | ||||
r199 | logging.info('redirecting to login page') | |||
r190 | return redirect(url('login_home')) | |||
r52 | ||||
r190 | return _wrapper | |||