2 files changed, 325 insertions, 0 deletions
diff --git a/tests/scripts/test081-totp b/tests/scripts/test081-totp
new file mode 100755
index 0000000..2c7a21c
--- /dev/null
+++ b/tests/scripts/test081-totp
@@ -0,0 +1,143 @@
+#!/bin/sh
+# $OpenLDAP$
+## This work is part of OpenLDAP Software <http://www.openldap.org/>.
+##
+## Copyright 2016-2021 Ondřej Kuzník, Symas Corp.
+## Copyright 2021-2022 The OpenLDAP Foundation.
+## All rights reserved.
+##
+## Redistribution and use in source and binary forms, with or without
+## modification, are permitted only as authorized by the OpenLDAP
+## Public License.
+##
+## A copy of this license is available in the file LICENSE in the
+## top-level directory of the distribution or, alternatively, at
+## <http://www.OpenLDAP.org/license.html>.
+
+echo "running defines.sh"
+. $SRCDIR/scripts/defines.sh
+
+if test $OTP = otpno; then
+    echo "OTP overlay not available, test skipped"
+    exit 0
+fi
+
+for python in python3 python2 python2.7 python27 python ""; do
+    if test x"$python" = x; then
+        echo "Useable Python environment not found, skipping test"
+        exit 0
+    fi
+
+    "$python" "$0".py --check >>$TESTOUT 2>&1
+    RC=$?
+    case $RC in
+    0)
+        break;;
+    1)
+        echo "$python is missing some required modules, skipping"
+        python=""
+        continue;;
+    127)
+        ;;
+    esac
+done
+
+export URI1 MANAGERDN PASSWD BABSDN BJORNSDN
+
+OTP_DATA=$DATADIR/otp/totp.ldif
+
+mkdir -p $TESTDIR $DBDIR1
+
+echo "Running slapadd to build slapd database..."
+. $CONFFILTER $BACKEND < $CONF > $ADDCONF
+$SLAPADD -f $ADDCONF -l $LDIFORDERED
+RC=$?
+if test $RC != 0 ; then
+    echo "slapadd failed ($RC)!"
+    exit $RC
+fi
+
+mkdir $TESTDIR/confdir
+. $CONFFILTER $BACKEND < $CONF > $CONF1
+
+$SLAPPASSWD -g -n >$CONFIGPWF
+echo "database config" >>$CONF1
+echo "rootpw `$SLAPPASSWD -T $CONFIGPWF`" >>$CONF1
+
+echo "Starting slapd on TCP/IP port $PORT1..."
+$SLAPD -f $CONF1 -F $TESTDIR/confdir -h $URI1 -d $LVL > $LOG1 2>&1 &
+PID=$!
+if test $WAIT != 0 ; then
+    echo PID $PID
+    read foo
+fi
+KILLPIDS="$PID"
+
+sleep $SLEEP0
+
+for i in 0 1 2 3 4 5; do
+    $LDAPSEARCH -s base -b "$MONITOR" -H $URI1 \
+        'objectclass=*' > /dev/null 2>&1
+    RC=$?
+    if test $RC = 0 ; then
+        break
+    fi
+    echo "Waiting ${SLEEP1} seconds for slapd to start..."
+    sleep ${SLEEP1}
+done
+
+if [ "$OTP" = otpmod ]; then
+$LDAPADD -D cn=config -H $URI1 -y $CONFIGPWF \
+    >> $TESTOUT 2>&1 <<EOMOD
+dn: cn=module,cn=config
+objectClass: olcModuleList
+cn: module
+olcModulePath: $TESTWD/../servers/slapd/overlays
+olcModuleLoad: otp.la
+EOMOD
+RC=$?
+if test $RC != 0 ; then
+    echo "ldapmodify failed ($RC)!"
+    test $KILLSERVERS != no && kill -HUP $KILLPIDS
+    exit $RC
+fi
+fi
+
+echo "Loading test otp configuration..."
+$LDAPMODIFY -v -D cn=config -H $URI1 -y $CONFIGPWF \
+    >> $TESTOUT 2>&1 <<EOMOD
+dn: olcOverlay={0}otp,olcDatabase={1}$BACKEND,cn=config
+changetype: add
+objectClass: olcOverlayConfig
+EOMOD
+RC=$?
+if test $RC != 0 ; then
+    echo "ldapmodify failed ($RC)!"
+    test $KILLSERVERS != no && kill -HUP $KILLPIDS
+    exit $RC
+fi
+
+echo "Provisioning tokens and configuration..."
+$LDAPMODIFY -D "$MANAGERDN" -H $URI1 -w $PASSWD \
+    >> $TESTOUT 2>&1 < $OTP_DATA
+RC=$?
+if test $RC != 0 ; then
+    echo "ldapmodify failed ($RC)!"
+    test $KILLSERVERS != no && kill -HUP $KILLPIDS
+    exit $RC
+fi
+
+"$python" "$0".py
+RC=$?
+
+test $KILLSERVERS != no && kill -HUP $KILLPIDS
+
+if test $RC != 0 ; then
+    echo "Test failed ($RC)!"
+else
+    echo ">>>>> Test succeeded"
+fi
+
+test $KILLSERVERS != no && wait
+
+exit $RC
diff --git a/tests/scripts/test081-totp.py b/tests/scripts/test081-totp.py
new file mode 100755
index 0000000..aeedaf2
--- /dev/null
+++ b/tests/scripts/test081-totp.py
@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+# $OpenLDAP$
+## This work is part of OpenLDAP Software <http://www.openldap.org/>.
+##
+## Copyright 2016-2021 Ondřej Kuzník, Symas Corp.
+## Copyright 2021-2022 The OpenLDAP Foundation.
+## All rights reserved.
+##
+## Redistribution and use in source and binary forms, with or without
+## modification, are permitted only as authorized by the OpenLDAP
+## Public License.
+##
+## A copy of this license is available in the file LICENSE in the
+## top-level directory of the distribution or, alternatively, at
+## <http://www.OpenLDAP.org/license.html>.
+
+from __future__ import print_function
+
+import hashlib
+import hmac
+import os
+import struct
+import sys
+import time
+
+import ldap
+from ldap.cidict import cidict as CIDict
+from ldap.ldapobject import LDAPObject
+
+if len(sys.argv) > 1 and sys.argv[1] == "--check":
+    raise SystemExit(0)
+
+
+def get_digits(h, digits):
+    offset = h[19] & 15
+    number = struct.unpack(">I", h[offset:offset+4])[0] & 0x7fffffff
+    number %= (10 ** digits)
+    return ("%0*d" % (digits, number)).encode()
+
+
+def get_hotp_token(secret, interval_no):
+    msg = struct.pack(">Q", interval_no)
+    h = hmac.new(secret, msg, hashlib.sha1).digest()
+    return get_digits(bytearray(h), 6)
+
+
+def get_interval(period=30):
+    return int(time.time() // period)
+
+
+def get_token_for(connection, dn, typ="totp"):
+    result = connection.search_s(dn, ldap.SCOPE_BASE)
+    dn, attrs = result[0]
+    attrs = CIDict(attrs)
+
+    tokendn = attrs['oath'+typ+'token'][0].decode()
+
+    result = connection.search_s(tokendn, ldap.SCOPE_BASE)
+    dn, attrs = result[0]
+    attrs = CIDict(attrs)
+
+    return dn, attrs
+
+
+def main():
+    uri = os.environ["URI1"]
+
+    managerdn = os.environ['MANAGERDN']
+    passwd = os.environ['PASSWD']
+
+    babsdn = os.environ['BABSDN']
+    babspw = b"bjensen"
+
+    bjornsdn = os.environ['BJORNSDN']
+    bjornspw = b"bjorn"
+
+    connection = LDAPObject(uri)
+
+    start = time.time()
+    connection.bind_s(managerdn, passwd)
+    end = time.time()
+
+    if end - start > 1:
+        print("It takes more than a second to connect and bind, "
+              "skipping potentially unstable test", file=sys.stderr)
+        raise SystemExit(0)
+
+    dn, token_entry = get_token_for(connection, babsdn)
+
+    paramsdn = token_entry['oathTOTPParams'][0].decode()
+    result = connection.search_s(paramsdn, ldap.SCOPE_BASE)
+    _, attrs = result[0]
+    params = CIDict(attrs)
+
+    secret = token_entry['oathSecret'][0]
+    period = int(params['oathTOTPTimeStepPeriod'][0].decode())
+
+    bind_conn = LDAPObject(uri)
+
+    interval_no = get_interval(period)
+    token = get_hotp_token(secret, interval_no-3)
+
+    print("Testing old tokens are not useable")
+    bind_conn.bind_s(babsdn, babspw+token)
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        pass
+    else:
+        raise SystemExit("Bind with an old token should have failed")
+
+    interval_no = get_interval(period)
+    token = get_hotp_token(secret, interval_no)
+
+    print("Testing token can only be used once")
+    bind_conn.bind_s(babsdn, babspw+token)
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        pass
+    else:
+        raise SystemExit("Bind with a reused token should have failed")
+
+    token = get_hotp_token(secret, interval_no+1)
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        raise SystemExit("Bind should have succeeded")
+
+    dn, token_entry = get_token_for(connection, babsdn)
+    last = int(token_entry['oathTOTPLastTimeStep'][0].decode())
+    if last != interval_no+1:
+        SystemExit("Unexpected counter value %d (expected %d)" %
+                   (last, interval_no+1))
+
+    print("Resetting counter and testing secret sharing between accounts")
+    connection.modify_s(dn, [(ldap.MOD_REPLACE, 'oathTOTPLastTimeStep', [])])
+
+    interval_no = get_interval(period)
+    token = get_hotp_token(secret, interval_no)
+
+    try:
+        bind_conn.bind_s(bjornsdn, bjornspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        raise SystemExit("Bind should have succeeded")
+
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        pass
+    else:
+        raise SystemExit("Bind with a reused token should have failed")
+
+    print("Testing token is retired even with a wrong password")
+    connection.modify_s(dn, [(ldap.MOD_REPLACE, 'oathTOTPLastTimeStep', [])])
+
+    interval_no = get_interval(period)
+    token = get_hotp_token(secret, interval_no)
+
+    try:
+        bind_conn.bind_s(babsdn, b"not the password"+token)
+    except ldap.INVALID_CREDENTIALS:
+        pass
+    else:
+        raise SystemExit("Bind with an incorrect password should have failed")
+
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        pass
+    else:
+        raise SystemExit("Bind with a reused token should have failed")
+
+    token = get_hotp_token(secret, interval_no+1)
+    try:
+        bind_conn.bind_s(babsdn, babspw+token)
+    except ldap.INVALID_CREDENTIALS:
+        raise SystemExit("Bind should have succeeded")
+
+
+if __name__ == "__main__":
+    sys.exit(main())