Add a PAMIdentityProvider for TurboGears. This utilizes the PAM python module written by Chris AtLee.

3 files changed, 182 insertions, 0 deletions

diff --git a/funcweb/funcweb/identity/__init__.py b/funcweb/funcweb/identity/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/funcweb/funcweb/identity/__init__.py
diff --git a/funcweb/funcweb/identity/pam.py b/funcweb/funcweb/identity/pam.py
new file mode 100644
index 0000000..aed5420
--- /dev/null
+++ b/funcweb/funcweb/identity/pam.py
@@ -0,0 +1,127 @@
+# (c) 2007 Chris AtLee <chris@atlee.ca>
+# Licensed under the MIT license:
+# http://www.opensource.org/licenses/mit-license.php
+"""
+PAM module for python
+
+Provides an authenticate function that will allow the caller to authenticate
+a user against the Pluggable Authentication Modules (PAM) on the system.
+
+Implemented using ctypes, so no compilation is necessary.
+"""
+__all__ = ['authenticate']
+
+from ctypes import CDLL, POINTER, Structure, CFUNCTYPE, cast, pointer, sizeof
+from ctypes import c_void_p, c_uint, c_char_p, c_char, c_int
+
+
+LIBPAM = CDLL("libpam.so")
+LIBC = CDLL("libc.so.6")
+
+CALLOC = LIBC.calloc
+CALLOC.restype = c_void_p
+CALLOC.argtypes = [c_uint, c_uint]
+
+STRDUP = LIBC.strdup
+STRDUP.argstypes = [c_char_p]
+STRDUP.restype = POINTER(c_char) # NOT c_char_p !!!!
+
+# Various constants
+PAM_PROMPT_ECHO_OFF = 1
+PAM_PROMPT_ECHO_ON = 2
+PAM_ERROR_MSG = 3
+PAM_TEXT_INFO = 4
+
+class PamHandle(Structure):
+    """wrapper class for pam_handle_t"""
+    _fields_ = [
+            ("handle", c_void_p)
+            ]
+
+    def __init__(self):
+        Structure.__init__(self)
+        self.handle = 0
+
+class PamMessage(Structure):
+    """wrapper class for pam_message structure"""
+    _fields_ = [
+            ("msg_style", c_int),
+            ("msg", c_char_p),
+            ]
+
+    def __repr__(self):
+        return "<PamMessage %i '%s'>" % (self.msg_style, self.msg)
+
+class PamResponse(Structure):
+    """wrapper class for pam_response structure"""
+    _fields_ = [
+            ("resp", c_char_p),
+            ("resp_retcode", c_int),
+            ]
+
+    def __repr__(self):
+        return "<PamResponse %i '%s'>" % (self.resp_retcode, self.resp)
+
+CONV_FUNC = CFUNCTYPE(c_int,
+        c_int, POINTER(POINTER(PamMessage)),
+               POINTER(POINTER(PamResponse)), c_void_p)
+
+class PamConv(Structure):
+    """wrapper class for pam_conv structure"""
+    _fields_ = [
+            ("conv", CONV_FUNC),
+            ("appdata_ptr", c_void_p)
+            ]
+
+PAM_START = LIBPAM.pam_start
+PAM_START.restype = c_int
+PAM_START.argtypes = [c_char_p, c_char_p, POINTER(PamConv),
+        POINTER(PamHandle)]
+
+PAM_AUTHENTICATE = LIBPAM.pam_authenticate
+PAM_AUTHENTICATE.restype = c_int
+PAM_AUTHENTICATE.argtypes = [PamHandle, c_int]
+
+def authenticate(username, password, service='login'):
+    """Returns True if the given username and password authenticate for the
+    given service.  Returns False otherwise
+
+    ``username``: the username to authenticate
+
+    ``password``: the password in plain text
+
+    ``service``: the PAM service to authenticate against.
+                 Defaults to 'login'"""
+    @CONV_FUNC
+    def my_conv(n_messages, messages, p_response, app_data):
+        """Simple conversation function that responds to any
+        prompt where the echo is off with the supplied password"""
+        # Create an array of n_messages response objects
+        addr = CALLOC(n_messages, sizeof(PamResponse))
+        p_response[0] = cast(addr, POINTER(PamResponse))
+        for i in range(n_messages):
+            if messages[i].contents.msg_style == PAM_PROMPT_ECHO_OFF:
+                pw_copy = STRDUP(password)
+                p_response.contents[i].resp = cast(pw_copy, c_char_p)
+                p_response.contents[i].resp_retcode = 0
+        return 0
+
+    # STRDUP expects byte strings
+    if isinstance(password, unicode):
+        password = str(password)
+
+    handle = PamHandle()
+    conv = PamConv(my_conv, 0)
+    retval = PAM_START(service, username, pointer(conv), pointer(handle))
+
+    if retval != 0:
+        # TODO: This is not an authentication error, something
+        # has gone wrong starting up PAM
+        return False
+
+    retval = PAM_AUTHENTICATE(handle, 0)
+    return retval == 0
+
+if __name__ == "__main__":
+    import getpass
+    print authenticate(getpass.getuser(), getpass.getpass())
diff --git a/funcweb/funcweb/identity/pamprovider.py b/funcweb/funcweb/identity/pamprovider.py
new file mode 100644
index 0000000..9f3ecf6
--- /dev/null
+++ b/funcweb/funcweb/identity/pamprovider.py
@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright © 2008  Red Hat, Inc. All rights reserved.
+#
+# This copyrighted material is made available to anyone wishing to use, modify,
+# copy, or redistribute it subject to the terms and conditions of the GNU
+# General Public License v.2.  This program is distributed in the hope that it
+# will be useful, but WITHOUT ANY WARRANTY expressed or implied, including the
+# implied warranties 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. Any Red Hat trademarks that are
+# incorporated in the source code or documentation are not subject to the GNU
+# General Public License and may only be used or replicated with the express
+# permission of Red Hat, Inc.
+#
+# Author(s): Luke Macken <lmacken@redhat.com>
+
+import logging
+
+from model import *
+from turbogears.identity.saprovider import *
+
+log = logging.getLogger(__name__)
+
+visit_identity_class = None
+
+class PAMIdentityProvider(SqlAlchemyIdentityProvider):
+    """
+        IdentityProvider that authenticates users against PAM.
+    """
+    def validate_identity(self, user_name, password, visit_key):
+        if not self.validate_password(user_name, password):
+            log.warning("Invalid password for %s" % user_name)
+            return None
+
+        log.info("Login successful for %s" % user_name)
+
+        try:
+            link = VisitIdentity.by_visit_key(visit_key)
+            #link.user_id = user.id
+            log.debug("Found visit!")
+        except Exception, e:
+            log.debug("Cannot find visit")
+            link = VisitIdentity(visit_key=visit_key, user_id=user_name)
+            print "Exception: %s" % str(e)
+
+        return SqlAlchemyIdentity(visit_key, user)
+
+    def validate_password(self,user_name, password):
+        import pam
+        log.debug("Authenticating user '%s' against PAM" % user_name)
+        assert pam
+        return pam.authenticate(user_name, password)