X-Git-Url: https://git.stderr.nl/gitweb?p=matthijs%2Fprojects%2Fwipi.git;a=blobdiff_plain;f=conf%2Fauth%2Fphpbb.py;fp=conf%2Fauth%2Fphpbb.py;h=3c1f80686a746a0b2d1cea319798a2cfed8b495c;hp=0000000000000000000000000000000000000000;hb=f4d058f5d7a2f194dba31314ca7b3f327d12d13b;hpb=1ff01c4c02fb90c83980721072a5eb982b59f321 diff --git a/conf/auth/phpbb.py b/conf/auth/phpbb.py new file mode 100644 index 0000000..3c1f806 --- /dev/null +++ b/conf/auth/phpbb.py @@ -0,0 +1,246 @@ +# -*- coding: iso-8859-1 -*- +""" + MoinMoin - auth plugin doing a check against MySQL db + + @copyright: 2008 Matthijs Kooijman + @license: GNU GPL, see COPYING for details. +""" + +import MySQLdb +import md5 +from MoinMoin import user +from MoinMoin.auth import BaseAuth, ContinueLogin +from MoinMoin.datastruct.backends import LazyGroupsBackend, LazyGroup +from MoinMoin import log +logging = log.getLogger(__name__) + + +class PhpbbGroupsBackend(LazyGroupsBackend): + class PhpbbGroup(LazyGroup): + pass + + + def __init__(self, request, auth): + super(LazyGroupsBackend, self).__init__(request) + + self.auth = auth + self.request = request + + def __iter__(self): + return self.list_query("SELECT group_name \ + FROM `%sgroups` \ + WHERE group_single_user = 0" + % self.auth.phpbb_prefix) + + def __contains__(self, group_name): + return self.single_query("SELECT EXISTS ( \ + SELECT * \ + FROM `%sgroups` \ + WHERE group_single_user = 0 \ + AND group_name=%%s)" % self.auth.phpbb_prefix, + group_name) + + def __getitem__(self, group_name): + return self.PhpbbGroup(self.request, group_name, self) + + def _iter_group_members(self, group_name): + return self.list_query ("SELECT username \ + FROM `%susers` as u, `%suser_group` as ug, `%sgroups` as g \ + WHERE u.user_id = ug.user_id AND ug.group_id = g.group_id \ + AND ug.user_pending = 0 AND g.group_single_user = 0 \ + AND g.group_name = %%s" + % (self.auth.phpbb_prefix, self.auth.phpbb_prefix, self.auth.phpbb_prefix), + group_name) + + def _group_has_member(self, group_name, member): + return self.single_query ("SELECT EXISTS( \ + SELECT * \ + FROM `%susers` as u, `%suser_group` as ug, `%sgroups` as g \ + WHERE u.user_id = ug.user_id AND ug.group_id = g.group_id \ + AND ug.user_pending = 0 AND g.group_single_user = 0 \ + AND g.group_name = %%s AND u.username = %%s)" + % (self.auth.phpbb_prefix, self.auth.phpbb_prefix, self.auth.phpbb_prefix), + (group_name, member)) + + def groups_with_member(self, member): + return self.list_query ("SELECT g.group_name \ + FROM `%susers` as u, `%suser_group` as ug, `%sgroups` as g \ + WHERE u.user_id = ug.user_id AND ug.group_id = g.group_id \ + AND ug.user_pending = 0 AND g.group_single_user = 0 \ + AND u.username = %%s" + % (self.auth.phpbb_prefix, self.auth.phpbb_prefix, self.auth.phpbb_prefix), + member) + + def single_query(self, *args): + """ + Runs an SQL query, that returns single row with a single column. + Returns just that single result. + """ + conn = None + cursor = None + try: + conn = self.auth.connect(self.request) + cursor = conn.cursor() + cursor.execute(*args) + + return cursor.fetchone()[0] + finally: + if cursor: + cursor.close() + if conn: + conn.close() + + def list_query(self, *args): + """ + Runs an SQL query, that returns any number of single-column rows. + Returns the results as a list of single values + """ + conn = None + cursor = None + try: + conn = self.auth.connect(self.request) + cursor = conn.cursor() + cursor.execute(*args) + + for row in cursor: + yield row[0] + finally: + if cursor: + cursor.close() + if conn: + conn.close() + +class PhpbbAuth(BaseAuth): + logout_possible = True + login_inputs = ['username', 'password'] + + def __init__(self, name='phpbb', dbhost=None, dbuser=None, dbpass=None, dbname=None, dbport=None, phpbb_prefix='', hint=None): + """ + Authenticate using credentials from a phpbb database + + The name parameter should be unique among all authentication methods. + + The hint parameter is a snippet of HTML that is displayed below the login form. + """ + self.dbhost = dbhost + self.dbuser = dbuser + self.dbpass = dbpass + self.dbname = dbname + self.dbport = dbport + self.phpbb_prefix = phpbb_prefix + self.name = name + self.hint = hint + + # Create a "constructor" to create a phpbb_groups instance, while + # passing ourselves to it. + self.groups_backend = lambda config, request: PhpbbGroupsBackend(request, self) + + def check_login(self, request, username, password): + """ Checks the given username password combination. Returns the + corresponding emailaddress, or False if authentication failed. + """ + conn = self.connect(request) + + if not conn: + return False + + # Get some data. Note that we interpolate the prefix ourselves, since + # letting the mysql library do it only works with values (it adds '' + # automatically). Note also that this allows possible SQL injection + # through the phpbb_prefix variable, but that should be a trusted + # value anyway. + cursor = conn.cursor () + cursor.execute ("SELECT user_password,user_email FROM `%susers` WHERE username=%%s" % self.phpbb_prefix, username) + + # No data? No login. + if (cursor.rowcount == 0): + conn.close() + return False + + # Check password + row = cursor.fetchone() + conn.close() + + if (md5.new(password).hexdigest() == row[0]): + return row[1] + else: + return False + + def connect(self, request): + # This code was shamelessly stolen from + # django.db.backends.mysql.base.cursor + kwargs = { + 'charset': 'utf8', + 'use_unicode': False, + } + if self.dbuser: + kwargs['user'] = self.dbuser + if self.dbname: + kwargs['db'] = self.dbname + if self.dbpass: + kwargs['passwd'] = self.dbpass + if self.dbhost.startswith('/'): + kwargs['unix_socket'] = self.dbhost + elif self.dbhost: + kwargs['host'] = self.dbhost + if self.dbport: + kwargs['port'] = int(self.dbport) + + # End stolen code + + try: + conn = MySQLdb.connect (**kwargs) + except: + import sys + import traceback + info = sys.exc_info() + logging.error("phpbb_login: authentication failed due to exception connecting to DB, traceback follows...") + logging.error(''.join(traceback.format_exception(*info))) + return False + + return conn + + def login(self, request, user_obj, **kw): + try: + username = kw.get('username') + password = kw.get('password') + + logging.debug("phpbb_login: Trying to log in, username=%r " % (username)) + + # simply continue if something else already logged in + # successfully + if user_obj and user_obj.valid: + return ContinueLogin(user_obj) + + # Deny empty username or passwords + if not username or not password: + return ContinueLogin(user_obj) + + email = self.check_login(request, username, password) + + # Login incorrect + if (not email): + logging.debug("phpbb_login: authentication failed for %s" % (username)) + return ContinueLogin(user_obj) + + logging.debug("phpbb_login: authenticated %s (email %s)" % (username, email)) + + u = user.User(request, auth_username=username, auth_method=self.name, auth_attribs=('name', 'password', 'email')) + u.email = email + #u.remember_me = 0 # 0 enforces cookie_lifetime config param + u.create_or_update(True) + + return ContinueLogin(u) + except: + import sys + import traceback + info = sys.exc_info() + logging.error("phpbb_login: authentication failed due to unexpected exception, traceback follows...") + logging.error(''.join(traceback.format_exception(*info))) + return ContinueLogin(user_obj) + + def login_hint(self, request): + """ Return a snippet of HTML that is displayed with the login form. """ + return self.hint + +# vim: set sw=4 expandtab sts=4:vim