auth.py
178 lines
| 5.8 KiB
| text/x-python
|
PythonLexer
r252 | #!/usr/bin/env python | |||
# encoding: utf-8 | ||||
# authentication and permission libraries | ||||
# Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com> | ||||
# This program is free software; you can redistribute it and/or | ||||
# modify it under the terms of the GNU General Public License | ||||
# as published by the Free Software Foundation; version 2 | ||||
# of the License or (at your opinion) any later version of the license. | ||||
# | ||||
# This program is distributed in the hope that it will be useful, | ||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of | ||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||||
# GNU General Public License for more details. | ||||
# | ||||
# You should have received a copy of the GNU General Public License | ||||
# along with this program; if not, write to the Free Software | ||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, | ||||
# MA 02110-1301, USA. | ||||
""" | ||||
Created on April 4, 2010 | ||||
@author: marcink | ||||
""" | ||||
r190 | from functools import wraps | |||
r239 | from pylons import session, url, app_globals as g | |||
r52 | from pylons.controllers.util import abort, redirect | |||
|
r64 | from pylons_app.model import meta | ||
r234 | from pylons_app.model.db import User | |||
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: | |||
r234 | user = sa.query(User).filter(User.username == username).one() | |||
|
r64 | 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 | ||||
r239 | ||||
def set_available_permissions(config): | ||||
""" | ||||
This function will propagate pylons globals with all available defined | ||||
permission given in db. We don't wannt to check each time from db for new | ||||
permissions since adding a new permission also requires application restart | ||||
ie. to decorate new views with the newly created permission | ||||
@param config: | ||||
""" | ||||
from pylons_app.model.meta import Session | ||||
from pylons_app.model.db import Permission | ||||
logging.info('getting information about all available permissions') | ||||
sa = Session() | ||||
all_perms = sa.query(Permission).all() | ||||
config['pylons.app_globals'].available_permissions = [x.permission_name for x in all_perms] | ||||
r190 | #=============================================================================== | |||
# 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 | |||
r239 | ||||
class PermsDecorator(object): | ||||
def __init__(self, *perms): | ||||
available_perms = g.available_permissions | ||||
for perm in perms: | ||||
if perm not in available_perms: | ||||
raise Exception("'%s' permission in not defined" % perm) | ||||
self.required_perms = set(perms) | ||||
self.user_perms = set([])#propagate this list from somewhere. | ||||
def __call__(self, func): | ||||
@wraps(func) | ||||
def _wrapper(*args, **kwargs): | ||||
logging.info('checking %s permissions %s for %s', | ||||
self.__class__.__name__[-3:], self.required_perms, func.__name__) | ||||
if self.check_permissions(): | ||||
logging.info('Permission granted for %s', func.__name__) | ||||
return func(*args, **kwargs) | ||||
else: | ||||
logging.warning('Permission denied for %s', func.__name__) | ||||
#redirect with forbidden ret code | ||||
return redirect(url('access_denied'), 403) | ||||
return _wrapper | ||||
def check_permissions(self): | ||||
""" | ||||
Dummy function for overiding | ||||
""" | ||||
raise Exception('You have to write this function in child class') | ||||
class CheckPermissionAll(PermsDecorator): | ||||
""" | ||||
Checks for access permission for all given predicates. All of them have to | ||||
be meet in order to fulfill the request | ||||
""" | ||||
def check_permissions(self): | ||||
if self.required_perms.issubset(self.user_perms): | ||||
return True | ||||
return False | ||||
class CheckPermissionAny(PermsDecorator): | ||||
""" | ||||
Checks for access permission for any of given predicates. In order to | ||||
fulfill the request any of predicates must be meet | ||||
""" | ||||
def check_permissions(self): | ||||
if self.required_perms.intersection(self.user_perms): | ||||
return True | ||||
return False | ||||