path: root/pynslcd/pam.py

# pam.py - functions authentication, authorisation and session handling
#
# Copyright (C) 2010-2019 Arthur de Jong
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301 USA

import logging
import random
import socket
import time

import ldap
from ldap.controls.ppolicy import PasswordPolicyControl, PasswordPolicyError
from ldap.filter import escape_filter_chars

import cfg
import common
import constants
import passwd
import search
import shadow


random = random.SystemRandom()


def authenticate(binddn, password):
    # open a new connection
    conn = search.Connection()
    # bind using the specified credentials
    serverctrls = []
    if cfg.pam_authc_ppolicy:
        serverctrls.append(PasswordPolicyControl())
    res, data, msgid, ctrls = conn.simple_bind_s(binddn, password, serverctrls=serverctrls)
    # go over bind result server controls
    for ctrl in ctrls:
        if ctrl.controlType == PasswordPolicyControl.controlType:
            # found a password policy control
            logging.debug(
                'PasswordPolicyControl found: error=%s (%s), '
                'timeBeforeExpiration=%s, graceAuthNsRemaining=%s',
                'None' if ctrl.error is None else PasswordPolicyError(ctrl.error).prettyPrint(),
                ctrl.error, ctrl.timeBeforeExpiration, ctrl.graceAuthNsRemaining)
            if ctrl.error == 0:  # passwordExpired
                return (
                    conn, constants.NSLCD_PAM_AUTHTOK_EXPIRED,
                    PasswordPolicyError(ctrl.error).prettyPrint())
            elif ctrl.error == 1:  # accountLocked
                return (
                    conn, constants.NSLCD_PAM_ACCT_EXPIRED,
                    PasswordPolicyError(ctrl.error).prettyPrint())
            elif ctrl.error == 2:  # changeAfterReset
                return (
                    conn, constants.NSLCD_PAM_NEW_AUTHTOK_REQD,
                    'Password change is needed after reset')
            elif ctrl.error:
                return (
                    conn, constants.NSLCD_PAM_PERM_DENIED,
                    PasswordPolicyError(ctrl.error).prettyPrint())
            elif ctrl.timeBeforeExpiration is not None:
                return (
                    conn, constants.NSLCD_PAM_NEW_AUTHTOK_REQD,
                    'Password will expire in %d seconds' % ctrl.timeBeforeExpiration)
            elif ctrl.graceAuthNsRemaining is not None:
                return (
                    conn, constants.NSLCD_PAM_NEW_AUTHTOK_REQD,
                    'Password expired, %d grace logins left' % ctrl.graceAuthNsRemaining)
    # perform search for own object (just to do any kind of search)
    results = search.LDAPSearch(
        conn, base=binddn, scope=ldap.SCOPE_BASE,
        filter='(objectClass=*)', attributes=['dn'])
    for entry in results:
        if entry[0] == binddn:
            return conn, constants.NSLCD_PAM_SUCCESS, ''
    # if our DN wasn't found raise an error to signal bind failure
    raise ldap.NO_SUCH_OBJECT()


def pwmod(conn, userdn, oldpassword, newpassword):
    # perform request without old password
    try:
        conn.passwd_s(userdn, None, newpassword)
    except ldap.LDAPError:
        # retry with old password
        if oldpassword:
            conn.passwd_s(userdn, oldpassword, newpassword)
        else:
            raise


def update_lastchange(conns, userdn):
    """Try to update the shadowLastChange attribute of the entry."""
    attribute = shadow.attmap['shadowLastChange']
    if str(attribute) == '"${shadowLastChange:--1}"':
        attribute = 'shadowLastChange'
    if not attribute or '$' in str(attribute):
        raise ValueError('shadowLastChange has unsupported mapping')
    # build the value for the new attribute
    if attribute.lower() == 'pwdlastset':
        # for AD we use another timestamp */
        value = '%d000000000' % (int(time.time()) // 100 + (134774 * 864))
    else:
        # time in days since Jan 1, 1970
        value = '%d' % (int(time.time()) // (60 * 60 * 24))
    # perform the modification, return at first success
    for conn in conns:
        try:
            conn.modify_s(userdn, [(ldap.MOD_REPLACE, attribute, [value.encode('utf-8')])])
            return
        except ldap.LDAPError:
            pass  # ignore error and try next connection


class PAMRequest(common.Request):

    def validate(self, parameters):
        """Check the username for validity and fill in the DN if needed."""
        # check username for validity
        common.validate_name(parameters['username'])
        # look up user DN
        entry = passwd.uid2entry(self.conn, parameters['username'])
        if not entry:
            # FIXME: we should close the stream with an empty response here
            raise ValueError('%r: user not found' % parameters['username'])
        # save the DN
        parameters['userdn'] = entry[0]
        # get the "real" username
        value = passwd.attmap.get_rdn_value(entry[0], 'uid')
        if not value:
            # get the username from the uid attribute
            values = entry[1]['uid']
            if not values or not values[0]:
                logging.warning('%s: is missing a %s attribute', entry[0], passwd.attmap['uid'])
            value = values[0]
        # check the username
        if value and not common.is_valid_name(value):
            raise ValueError('%s: has invalid %s attribute', entry[0], passwd.attmap['uid'])
        # check if the username is different and update it if needed
        if value != parameters['username']:
            logging.info('username changed from %r to %r', parameters['username'], value)
            parameters['username'] = value


class PAMAuthenticationRequest(PAMRequest):

    action = constants.NSLCD_ACTION_PAM_AUTHC

    def read_parameters(self, fp):
        return dict(username=fp.read_string(),
                    service=fp.read_string(),
                    ruser=fp.read_string(),
                    rhost=fp.read_string(),
                    tty=fp.read_string(),
                    password=fp.read_string())
        # TODO: log call with parameters

    def write(self, username, authc=constants.NSLCD_PAM_SUCCESS,
              authz=constants.NSLCD_PAM_SUCCESS, msg=''):
        self.fp.write_int32(constants.NSLCD_RESULT_BEGIN)
        self.fp.write_int32(authc)
        self.fp.write_string(username)
        self.fp.write_int32(authz)
        self.fp.write_string(msg)
        self.fp.write_int32(constants.NSLCD_RESULT_END)

    def handle_request(self, parameters):
        # if the username is blank and rootpwmoddn is configured, try to
        # authenticate as administrator, otherwise validate request as usual
        if not parameters['username'] and cfg.rootpwmoddn:
            # authenticate as rootpwmoddn
            binddn = cfg.rootpwmoddn
            # if the caller is root we will allow the use of rootpwmodpw
            if not parameters['password'] and self.calleruid == 0 and cfg.rootpwmodpw:
                password = cfg.rootpwmodpw
            elif parameters['password']:
                password = parameters['password']
            else:
                raise ValueError('password missing')
        else:
            self.validate(parameters)
            binddn = parameters['userdn']
            password = parameters['password']
        # try authentication
        try:
            conn, authz, msg = authenticate(binddn, password)
        except ldap.INVALID_CREDENTIALS as e:
            try:
                msg = e[0]['desc']
            except Exception:
                msg = str(e)
            logging.debug('bind failed: %s', msg)
            self.write(parameters['username'], authc=constants.NSLCD_PAM_AUTH_ERR, msg=msg)
            return
        if authz != constants.NSLCD_PAM_SUCCESS:
            logging.warning('%s: %s: %s', binddn, parameters['username'], msg)
        else:
            logging.debug('bind successful')
        # FIXME: perform shadow attribute checks with check_shadow()
        self.write(parameters['username'], authz=authz, msg=msg)


class PAMAuthorisationRequest(PAMRequest):

    action = constants.NSLCD_ACTION_PAM_AUTHZ

    def read_parameters(self, fp):
        return dict(username=fp.read_string(),
                    service=fp.read_string(),
                    ruser=fp.read_string(),
                    rhost=fp.read_string(),
                    tty=fp.read_string())
        # TODO: log call with parameters

    def write(self, authz=constants.NSLCD_PAM_SUCCESS, msg=''):
        self.fp.write_int32(constants.NSLCD_RESULT_BEGIN)
        self.fp.write_int32(authz)
        self.fp.write_string(msg)
        self.fp.write_int32(constants.NSLCD_RESULT_END)

    def check_authz_search(self, parameters):
        if not cfg.pam_authz_searches:
            return
        # escape all parameters
        variables = dict((k, escape_filter_chars(v)) for k, v in parameters.items())
        variables.update(
            hostname=escape_filter_chars(socket.gethostname()),
            fqdn=escape_filter_chars(socket.getfqdn()),
            dn=variables['userdn'],
            uid=variables['username'])
        # go over all authz searches
        for x in cfg.pam_authz_searches:
            filter = x.value(variables)
            logging.debug('trying pam_authz_search "%s"', filter)
            srch = search.LDAPSearch(self.conn, filter=filter, attributes=('dn', ))
            try:
                dn, values = srch.items().next()
            except StopIteration:
                logging.error('pam_authz_search "%s" found no matches', filter)
                raise
            logging.debug('pam_authz_search found "%s"', dn)

    def handle_request(self, parameters):
        # fill in any missing userdn, etc.
        self.validate(parameters)
        # check authorisation search
        try:
            self.check_authz_search(parameters)
        except StopIteration:
            self.write(constants.NSLCD_PAM_PERM_DENIED,
                       'LDAP authorisation check failed')
            return
        # all tests passed, return OK response
        self.write()


class PAMPasswordModificationRequest(PAMRequest):

    action = constants.NSLCD_ACTION_PAM_PWMOD

    def read_parameters(self, fp):
        return dict(username=fp.read_string(),
                    service=fp.read_string(),
                    ruser=fp.read_string(),
                    rhost=fp.read_string(),
                    tty=fp.read_string(),
                    asroot=fp.read_int32(),
                    oldpassword=fp.read_string(),
                    newpassword=fp.read_string())
        # TODO: log call with parameters

    def write(self, rc=constants.NSLCD_PAM_SUCCESS, msg=''):
        self.fp.write_int32(constants.NSLCD_RESULT_BEGIN)
        self.fp.write_int32(rc)
        self.fp.write_string(msg)
        self.fp.write_int32(constants.NSLCD_RESULT_END)

    def handle_request(self, parameters):
        # fill in any missing userdn, etc.
        self.validate(parameters)
        # check if pam_password_prohibit_message is set
        if cfg.pam_password_prohibit_message:
            self.write(constants.NSLCD_PAM_PERM_DENIED,
                       cfg.pam_password_prohibit_message)
            return
        # check if the the user passed the rootpwmoddn
        if parameters['asroot']:
            binddn = cfg.rootpwmoddn
            # check if rootpwmodpw should be used
            if not parameters['oldpassword'] and self.calleruid == 0 and cfg.rootpwmodpw:
                password = cfg.rootpwmodpw
            elif parameters['oldpassword']:
                password = parameters['oldpassword']
            else:
                raise ValueError('password missing')
        else:
            binddn = parameters['userdn']
            password = parameters['oldpassword']
            # TODO: check if shadow properties allow password change
        # perform password modification
        try:
            conn, authz, msg = authenticate(binddn, password)
            pwmod(conn, parameters['userdn'], parameters['oldpassword'], parameters['newpassword'])
            # try to update lastchange with normal or user connection
            update_lastchange((self.conn, conn), parameters['userdn'])
        except ldap.INVALID_CREDENTIALS as e:
            try:
                msg = e[0]['desc']
            except Exception:
                msg = str(e)
            logging.debug('pwmod failed: %s', msg)
            self.write(constants.NSLCD_PAM_PERM_DENIED, msg)
            return
        logging.debug('pwmod successful')
        self.write()


SESSION_ID_LENGTH = 25
SESSION_ID_ALPHABET = (
    "ABCDEFGHIJKLMNOPQRSTUVWXYZ" +
    "abcdefghijklmnopqrstuvwxyz" +
    "01234567890"
)


def generate_session_id():
    return ''.join(
        random.choice(SESSION_ID_ALPHABET)
        for i in range(SESSION_ID_LENGTH)
    )


class PAMSessionOpenRequest(PAMRequest):

    action = constants.NSLCD_ACTION_PAM_SESS_O

    def read_parameters(self, fp):
        return dict(username=fp.read_string(),
                    service=fp.read_string(),
                    ruser=fp.read_string(),
                    rhost=fp.read_string(),
                    tty=fp.read_string())
        # TODO: log call with parameters

    def write(self, sessionid):
        self.fp.write_int32(constants.NSLCD_RESULT_BEGIN)
        self.fp.write_string(sessionid)
        self.fp.write_int32(constants.NSLCD_RESULT_END)

    def handle_request(self, parameters):
        # generate a session id
        session_id = generate_session_id()
        self.write(session_id)


class PAMSessionCloseRequest(PAMRequest):

    action = constants.NSLCD_ACTION_PAM_SESS_C

    def read_parameters(self, fp):
        return dict(username=fp.read_string(),
                    service=fp.read_string(),
                    ruser=fp.read_string(),
                    rhost=fp.read_string(),
                    tty=fp.read_string(),
                    session_id=fp.read_string())
        # TODO: log call with parameters

    def write(self):
        self.fp.write_int32(constants.NSLCD_RESULT_BEGIN)
        self.fp.write_int32(constants.NSLCD_RESULT_END)

    def handle_request(self, parameters):
        self.write()