Using SO_PEERCRED inside of twisted/klein on a unix domain socket.

klein_peer_credentials.py

import struct
from socket import SOL_SOCKET, socket
import grp
import pwd
from klein.app import Klein


class MyApp(object):
    """
    An example Klein application to test unix domain socket peer credential checking
    """

    app = Klein()

    SO_PEERCRED = 17

    @staticmethod
    def get_user_groups(unix_socket):
        if isinstance(unix_socket, socket):
            credentials = unix_socket.getsockopt(SOL_SOCKET, MyApp.SO_PEERCRED, struct.calcsize('3i'))
            user_groups = []
            try:
                print 'Get Credentials'
                pid, uid, gid = struct.unpack('3i', credentials)
            except Exception as e:
                return e.message

            try:
                print 'Get use details'
                user = pwd.getpwuid(uid)
            except KeyError as err:
                error_message = u''.join([u'failed to retrieve user details for user ',
                                          u'connected to socket: %s']) % str(err)
                return error_message

            try:
                print 'Get primary group'
                user_primary_group = grp.getgrgid(gid)
            except KeyError as err:
                error_message = u''.join([u'failed to retrieve primary group details ',
                                          u'for user connected to socket: %s']) % str(err)
                return error_message

            user_groups.append(user_primary_group.gr_name)

            try:
                print 'Get extended groups'
                for entry in grp.getgrall():
                    print 'Checking group: %s' % entry.gr_name
                    if user.pw_name in entry.gr_mem:
                        print 'Found user %s in group %s' % (user.pw_name, entry.gr_name)
                        user_groups.append(entry.gr_name)
            except Exception as err: # pylint: disable=broad-except
                error_message = u''.join([u'failed to retrieve secondary group details ',
                                          u'for user connected to socket: %s']) % str(err)
                return error_message

            return user_groups
        else:
            return None

    @app.route("/test")
    def get_peer_cred(self, request):
        unix_socket = request.channel.transport.getHandle()
        return '%s' % self.get_user_groups(unix_socket)

myapp = MyApp()
resource = myapp.app.resource