feature: Manager-only accounts, and the Router Manager's first administrator.
Release / release (push) Successful in 20s

This commit is contained in:
slawek committed 2026-10-24 11:56:58 +02:00
1 parent b1e6d9cd0b
commit af3127d1e8
9 files changed
+259 -47

No files matched your search

@@ -1,9 +1,13 @@
"""The users and groups of the router's directory, for vpn-router-user, vpn-router-group and the web service.
A user is an entry in the router's directory, with a first name, a last name and an e-mail
address, whose password OpenVPN checks, and a client certificate of the router CA with the
user name as its common name, which OpenVPN and IKEv2 accept. Users exist whether or not a
P2S system is enabled.
A user is an entry in ou=users of the router's directory, of one of two kinds, fixed when it is
added. A VPN user is an inetOrgPerson with a first name, a last name and an e-mail address, whose
password OpenVPN checks, and a client certificate of the router CA with the user name as its
common name, which OpenVPN and IKEv2 accept. A Manager-only account is an account,
simpleSecurityObject and vpnRouterAccount (the package's schema) with a name, a password and
optionally an e-mail address, no names, no certificate and so no VPN access; it is added when
neither the first nor the last name is given, and can sign in to the web service. Users exist
whether or not a P2S system is enabled.
A group is a groupOfNames in ou=groups with a name, which follows the rules of user names, an
optional description and at least one member; a group cannot be empty, so the last member
@@ -19,8 +23,8 @@ code of it. slapd's otp overlay then takes the password followed by a code at ev
A user is disabled when pwdAccountLockedTime holds LOCKED, and locked when the password
policy has set it after failed binds, for vpnrouter_ldap.LOCKOUT_SECONDS; unlock ends that.
login() checks a user's password and code with a bind. Members of ADMIN_GROUP manage the
router through the web service. Changing the password, disabling, unlocking and removing a
user revoke its refresh tokens.
router through the web service. The module knows only the directory; the callers revoke a
user's refresh tokens after the operations of vpnrouter_auth.REVOKING.
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
@@ -44,7 +48,6 @@ import ldap.modlist
import pyotp
import vpnrouter
import vpnrouter_auth
import vpnrouter_ldap
#: A user name is one lowercase DNS label, as it is also the common name of the certificate.
@@ -59,7 +62,10 @@ 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', 'oathTOTPToken']
ATTRIBUTES = ['uid', 'objectClass', 'pwdAccountLockedTime', 'givenName', 'sn', 'mail', 'oathTOTPToken']
#: The object classes of a VPN user and of a Manager-only account.
VPN_USER_CLASSES = [b'inetOrgPerson']
MANAGER_ONLY_CLASSES = [b'account', b'simpleSecurityObject', b'vpnRouterAccount']
#: The readable attributes of a TOTP token; oathSecret is never readable.
TOKEN_ATTRIBUTES = ['oathTOTPToken', 'oathTOTPParams', 'oathSecretTime', 'oathTOTPLastTimeStep', 'oathTOTPTimeStepDrift',
'oathLastLogin', 'oathLastFailure', 'oathFailureCount']
@@ -172,12 +178,18 @@ def locked_since(value):
return 0
def manager_only(attributes):
"""Whether the entry is a Manager-only account."""
return b'account' in (value.lower() for value in attributes.get('objectClass', []))
def summary(attributes):
"""A user as the answers show it; a person attribute the entry does not have is None."""
name = attributes['uid'][0].decode()
lock = attributes.get('pwdAccountLockedTime', [None])[0]
user = {
'name': name,
'manager_only': manager_only(attributes),
'enabled': lock != LOCKED,
'locked': lock not in (None, LOCKED) and locked_since(lock) > time.time() - vpnrouter_ldap.LOCKOUT_SECONDS,
'certificate': certificate(name)[0].is_file(),
@@ -196,38 +208,62 @@ def common_name(attributes, name):
def list_users(connection):
found = connection.search_s(vpnrouter_ldap.USERS, ldap.SCOPE_ONELEVEL, '(objectClass=inetOrgPerson)', ATTRIBUTES)
found = connection.search_s(vpnrouter_ldap.USERS, ldap.SCOPE_ONELEVEL, '(|(objectClass=inetOrgPerson)(objectClass=account))',
ATTRIBUTES)
return {'users': sorted((summary(attributes) for _, attributes in found), key=lambda user: user['name'])}
def add(connection, name, password, values):
"""Add a user with the person attributes of values, all three required, and issue the certificate."""
"""Add a user: a VPN user with its certificate when values gives all three person attributes, a Manager-only account
with at most the e-mail address when it gives neither name."""
check_name(name)
details = person(values, required=True)
details = person(values, required=False)
vpn_user = 'givenName' in details or 'sn' in details
if vpn_user and len(details) != len(PERSON):
raise Failure('give first_name, last_name and email for a VPN user, or at most email for a Manager-only account', 2)
check_password(password)
if not (vpnrouter.PKI_DIR / 'ca_key.pem').is_file():
if vpn_user and not (vpnrouter.PKI_DIR / 'ca_key.pem').is_file():
raise Failure('the router CA does not exist; vpn-router-setup creates it once a platform is chosen')
record = {'objectClass': [b'inetOrgPerson'], 'uid': [name.encode()],
'cn': [common_name(details, name).encode()]}
if vpn_user:
record = {'objectClass': VPN_USER_CLASSES, 'uid': [name.encode()], 'cn': [common_name(details, name).encode()]}
else:
# simpleSecurityObject requires a userPassword; passwd_s sets the real one at once.
record = {'objectClass': MANAGER_ONLY_CLASSES, 'uid': [name.encode()], 'userPassword': [b'{CRYPT}!']}
record.update({attribute: [value.encode()] for attribute, value in details.items()})
try:
connection.add_s(dn_of(name), ldap.modlist.addModlist(record))
except ldap.ALREADY_EXISTS:
raise Failure(f'user {name} exists', 4)
connection.passwd_s(dn_of(name), None, password)
vpnrouter.USER_CERT_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
for path in certificate(name):
path.unlink(missing_ok=True)
simple_ca('make-cert', '--ca-dir', str(vpnrouter.PKI_DIR), '--cert-dir', str(vpnrouter.USER_CERT_DIR), name)
if vpn_user:
vpnrouter.USER_CERT_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
for path in certificate(name):
path.unlink(missing_ok=True)
simple_ca('make-cert', '--ca-dir', str(vpnrouter.PKI_DIR), '--cert-dir', str(vpnrouter.USER_CERT_DIR), name)
return summary(entry(connection, name))
def update(connection, name, values):
"""Change the person attributes values gives; the cn follows the first and the last name."""
"""Change the person attributes values gives; the cn follows the first and the last name.
A Manager-only account has only the e-mail address, which '' removes.
"""
current = entry(connection, name)
if manager_only(current):
if values.get('first_name') is not None or values.get('last_name') is not None:
raise Failure(f'user {name} is a Manager-only account, which has no first or last name', 2)
email = values.get('email')
if email is None:
raise Failure('give email', 2)
if isinstance(email, str) and email.strip() == '':
if 'mail' in current:
connection.modify_s(dn_of(name), [(ldap.MOD_DELETE, 'mail', None)])
else:
connection.modify_s(dn_of(name), [(ldap.MOD_REPLACE, 'mail', [person({'email': email}, required=False)['mail'].encode()])])
return summary(entry(connection, name))
details = person(values, required=False)
if not details:
raise Failure('give at least one of first_name, last_name and email', 2)
current = entry(connection, name)
names = {attribute: current[attribute][0].decode() for attribute in ('givenName', 'sn') if current.get(attribute)}
names.update({attribute: details[attribute] for attribute in ('givenName', 'sn') if attribute in details})
changes = [(ldap.MOD_REPLACE, attribute, [value.encode()]) for attribute, value in details.items()]
@@ -240,14 +276,12 @@ def passwd(connection, name, password):
check_password(password)
entry(connection, name)
connection.passwd_s(dn_of(name), None, password)
vpnrouter_auth.revoke(vpnrouter.load_config(), name)
return summary(entry(connection, name))
def disable(connection, name):
if entry(connection, name).get('pwdAccountLockedTime') != [LOCKED]:
connection.modify_s(dn_of(name), [(ldap.MOD_REPLACE, 'pwdAccountLockedTime', [LOCKED])])
vpnrouter_auth.revoke(vpnrouter.load_config(), name)
return summary(entry(connection, name))
@@ -266,7 +300,6 @@ def unlock(connection, name):
changes.append((ldap.MOD_DELETE, 'pwdAccountLockedTime', None))
if changes:
connection.modify_s(dn_of(name), changes)
vpnrouter_auth.revoke(vpnrouter.load_config(), name)
return summary(entry(connection, name))
@@ -361,7 +394,6 @@ def remove(connection, name):
cert.unlink(missing_ok=True)
key.unlink(missing_ok=True)
connection.delete_s(dn_of(name))
vpnrouter_auth.revoke(vpnrouter.load_config(), name)
return {'name': name, 'removed': True}