feature: users can be enrolled for TOTP one-time passwords.

debian-package/src/usr/lib/vpn-router/vpnrouter_users.py

@@ -10,22 +10,31 @@ optional description and at least one member; a group cannot be empty, so the la
 cannot leave it and a user who is the last member of a group cannot be removed. Groups only
 record membership; nothing on the router reads them.
 
+A user may have a TOTP token: the user entry itself is an oathTOTPUser and oathTOTPToken whose
+oathTOTPToken points to itself, with the secret in oathSecret and the parameters of
+vpnrouter_ldap.TOTP_PARAMS. Enrolment stores nothing until a code of the new secret is
+verified: otp_new makes a secret and its otpauth URI, and otp_enrol stores the secret given a
+code of it. Nothing on the router checks the codes at login.
+
 Every operation runs in a session(), which holds the lock that serializes changes and the
 connection of the user manager account. A request that cannot be done raises Failure, whose
 status says why: 2 the request is invalid, 3 the user does not exist, 4 the user exists, 1
 anything else. Answers are dicts with the keys of the JSON answers.
 """
 
+import base64
 import contextlib
 import fcntl
 import pathlib
 import re
 import subprocess
 import sys
+import time
 
 import ldap
 import ldap.filter
 import ldap.modlist
+import pyotp
 
 import vpnrouter
 import vpnrouter_ldap
@@ -42,7 +51,12 @@ EMAIL_MAX = 254
 #: The keys of the person attributes and their LDAP attributes.
 PERSON = (('first_name', 'givenName'), ('last_name', 'sn'), ('email', 'mail'))
 #: The attributes read for a user.
-ATTRIBUTES = ['uid', 'pwdAccountLockedTime', 'givenName', 'sn', 'mail']
+ATTRIBUTES = ['uid', 'pwdAccountLockedTime', 'givenName', 'sn', 'mail', 'oathTOTPToken']
+#: The readable attributes of a TOTP token; oathSecret is never readable.
+TOKEN_ATTRIBUTES = ['oathTOTPToken', 'oathTOTPParams', 'oathSecretTime', 'oathTOTPLastTimeStep', 'oathTOTPTimeStepDrift',
+                    'oathLastLogin', 'oathLastFailure', 'oathFailureCount']
+#: A TOTP secret as otp_new makes it: 32 base32 characters, 160 bits.
+TOTP_SECRET = re.compile(r'^[A-Z2-7]{32}$')
 #: The attributes read for a group.
 GROUP_ATTRIBUTES = ['cn', 'description', 'member']
 #: The longest group description, in characters.
@@ -145,6 +159,7 @@ def summary(attributes):
         'name': name,
         'enabled': 'pwdAccountLockedTime' not in attributes,
         'certificate': certificate(name)[0].is_file(),
+        'otp': 'oathTOTPToken' in attributes,
     }
     for key, attribute in PERSON:
         values = attributes.get(attribute)
@@ -218,6 +233,49 @@ def enable(connection, name):
     return summary(entry(connection, name))
 
 
+def totp(secret):
+    return pyotp.TOTP(secret, digits=vpnrouter_ldap.TOTP_DIGITS, interval=vpnrouter_ldap.TOTP_PERIOD)
+
+
+def otp_new(connection, name):
+    """A new TOTP secret for the user and its otpauth URI, issued by the router's wan.local_fqdn; nothing is stored."""
+    entry(connection, name)
+    secret = pyotp.random_base32()
+    issuer = vpnrouter.load_config()['wan']['local_fqdn'].strip()
+    return {'name': name, 'secret': secret, 'uri': totp(secret).provisioning_uri(name=name, issuer_name=issuer or None)}
+
+
+def otp_enrol(connection, name, secret, code):
+    """Store the secret of otp_new as the user's TOTP token, replacing one the user has, when code is a code of it."""
+    current = entry(connection, name)
+    if not isinstance(secret, str) or not TOTP_SECRET.match(secret):
+        raise Failure('the secret must be 32 base32 characters, as a new secret has them', 2)
+    if not isinstance(code, str) or not re.fullmatch(rf'\d{{{vpnrouter_ldap.TOTP_DIGITS}}}', code):
+        raise Failure(f'the code must be {vpnrouter_ldap.TOTP_DIGITS} digits', 2)
+    if not totp(secret).verify(code, valid_window=vpnrouter_ldap.TOTP_WINDOW):
+        raise Failure('the code does not match the secret; check the clock of the authenticator', 2)
+    changes = [] if 'oathTOTPToken' in current else [(ldap.MOD_ADD, 'objectClass', [b'oathTOTPUser', b'oathTOTPToken'])]
+    changes += [
+        (ldap.MOD_REPLACE, 'oathTOTPToken', [dn_of(name).encode()]),
+        (ldap.MOD_REPLACE, 'oathTOTPParams', [vpnrouter_ldap.TOTP_PARAMS.encode()]),
+        (ldap.MOD_REPLACE, 'oathSecret', [base64.b32decode(secret)]),
+        (ldap.MOD_REPLACE, 'oathSecretTime', [time.strftime('%Y%m%d%H%M%SZ', time.gmtime()).encode()]),
+    ]
+    connection.modify_s(dn_of(name), changes)
+    return summary(entry(connection, name))
+
+
+def otp_remove(connection, name):
+    """Remove the user's TOTP token, when the user has one."""
+    if 'oathTOTPToken' in entry(connection, name):
+        token = connection.search_s(dn_of(name), ldap.SCOPE_BASE, attrlist=TOKEN_ATTRIBUTES)[0][1]
+        changes = [(ldap.MOD_DELETE, 'objectClass', [b'oathTOTPUser', b'oathTOTPToken']),
+                   (ldap.MOD_DELETE, 'oathSecret', None)]
+        changes += [(ldap.MOD_DELETE, attribute, None) for attribute in token]
+        connection.modify_s(dn_of(name), changes)
+    return summary(entry(connection, name))
+
+
 def remove(connection, name):
     """Remove the user from its groups and the directory, and revoke the certificate.