567 lines
25 KiB
Python
567 lines
25 KiB
Python
"""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 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
|
|
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. slapd's otp overlay then takes the password followed by a code at every bind.
|
|
|
|
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. 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
|
|
status says why: 2 the request is invalid, 3 the user does not exist, 4 the user exists, 5 a
|
|
login failed, 1 anything else. Answers are dicts with the keys of the JSON answers.
|
|
"""
|
|
|
|
import base64
|
|
import calendar
|
|
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
|
|
|
|
#: A user name is one lowercase DNS label, as it is also the common name of the certificate.
|
|
NAME = re.compile(r'^[a-z][a-z0-9-]{0,30}[a-z0-9]$|^[a-z]$')
|
|
PASSWORD_MIN = 8
|
|
PASSWORD_MAX = 128
|
|
#: The longest first or last name, in characters.
|
|
PERSON_NAME_MAX = 64
|
|
#: An e-mail address: one @, no spaces, and a dot in the domain.
|
|
EMAIL = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$')
|
|
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', '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']
|
|
#: 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.
|
|
DESCRIPTION_MAX = 256
|
|
#: The value of pwdAccountLockedTime that disables an account until it is enabled.
|
|
LOCKED = b'000001010000Z'
|
|
#: The group whose members manage the router through the web service.
|
|
ADMIN_GROUP = 'admins'
|
|
#: The digits of a TOTP code given to login().
|
|
CODE = re.compile(r'^\d{6}$')
|
|
LOCK_FILE = pathlib.Path('/run/vpn-router/users.lock')
|
|
|
|
|
|
class Failure(Exception):
|
|
"""A request that cannot be done; status is 2 invalid, 3 no such user, 4 the user exists, 5 failed login, 1 other."""
|
|
|
|
def __init__(self, message, status=1):
|
|
super().__init__(message)
|
|
self.status = status
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def session():
|
|
"""The connection of the user manager account, used under the lock that serializes changes."""
|
|
LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(LOCK_FILE, 'w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
connection = vpnrouter_ldap.manager()
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.unbind_s()
|
|
|
|
|
|
def check_name(name):
|
|
if not isinstance(name, str) or not NAME.match(name):
|
|
raise Failure('a user name is 1 to 32 lowercase letters, digits and hyphens, starting with a letter', 2)
|
|
|
|
|
|
def check_password(value):
|
|
if not isinstance(value, str) or '\n' in value or '\r' in value:
|
|
raise Failure('the password must be a string of one line', 2)
|
|
if not PASSWORD_MIN <= len(value) <= PASSWORD_MAX:
|
|
raise Failure(f'the password must have {PASSWORD_MIN} to {PASSWORD_MAX} characters', 2)
|
|
|
|
|
|
def person(values, required):
|
|
"""The person attributes of values ({first_name, last_name, email}), checked, as {LDAP attribute: value}.
|
|
|
|
A key whose value is None is not given; with required, all three must be given.
|
|
"""
|
|
found = {}
|
|
for key, attribute in PERSON:
|
|
value = values.get(key)
|
|
if value is None:
|
|
if required:
|
|
raise Failure(f'{key} is required', 2)
|
|
continue
|
|
if not isinstance(value, str):
|
|
raise Failure(f'{key} must be a string', 2)
|
|
value = value.strip()
|
|
if key == 'email':
|
|
if len(value) > EMAIL_MAX or not EMAIL.match(value):
|
|
raise Failure(f'{value!r} is not an e-mail address', 2)
|
|
elif not value or len(value) > PERSON_NAME_MAX or any(not character.isprintable() for character in value):
|
|
raise Failure(f'the {key.replace("_", " ")} must have 1 to {PERSON_NAME_MAX} printable characters', 2)
|
|
found[attribute] = value
|
|
return found
|
|
|
|
|
|
def dn_of(name):
|
|
return f'uid={name},{vpnrouter_ldap.USERS}'
|
|
|
|
|
|
def certificate(name):
|
|
return vpnrouter.USER_CERT_DIR / f'{name}_cert.pem', vpnrouter.USER_CERT_DIR / f'{name}_key.pem'
|
|
|
|
|
|
def entry(connection, name):
|
|
"""The attributes of the user's entry; Failure with status 3 when there is none."""
|
|
check_name(name)
|
|
try:
|
|
return connection.search_s(dn_of(name), ldap.SCOPE_BASE, attrlist=ATTRIBUTES)[0][1]
|
|
except ldap.NO_SUCH_OBJECT:
|
|
raise Failure(f'user {name} does not exist', 3)
|
|
|
|
|
|
def simple_ca(*args):
|
|
result = subprocess.run([sys.executable, str(vpnrouter.SIMPLE_CA), *args], capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise Failure(f'simple-ca {args[0]} failed: {result.stderr.strip() or result.stdout.strip()}')
|
|
|
|
|
|
def publish_crl():
|
|
"""Give strongSwan the CRL after a revocation; OpenVPN reads it at every connection."""
|
|
if vpnrouter.install_file(vpnrouter.PKI_DIR / 'crl.pem', vpnrouter.CRL_DEST, 0o644):
|
|
subprocess.run(['swanctl', '--load-creds'], capture_output=True, text=True)
|
|
|
|
|
|
def locked_since(value):
|
|
"""The time a pwdAccountLockedTime value names, in seconds since the epoch, or 0 when it names none."""
|
|
try:
|
|
return calendar.timegm(time.strptime(value[:14].decode(), '%Y%m%d%H%M%S'))
|
|
except ValueError:
|
|
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(),
|
|
'otp': 'oathTOTPToken' in attributes,
|
|
}
|
|
for key, attribute in PERSON:
|
|
values = attributes.get(attribute)
|
|
user[key] = values[0].decode() if values else None
|
|
return user
|
|
|
|
|
|
def common_name(attributes, name):
|
|
"""The cn of an entry: the first and the last name, or the user name without them."""
|
|
full = ' '.join(value for value in (attributes.get('givenName'), attributes.get('sn')) if value)
|
|
return full or name
|
|
|
|
|
|
def list_users(connection):
|
|
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: 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=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 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')
|
|
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)
|
|
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.
|
|
|
|
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)
|
|
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()]
|
|
changes.append((ldap.MOD_REPLACE, 'cn', [common_name(names, name).encode()]))
|
|
connection.modify_s(dn_of(name), changes)
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def passwd(connection, name, password):
|
|
check_password(password)
|
|
entry(connection, name)
|
|
connection.passwd_s(dn_of(name), None, password)
|
|
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])])
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def enable(connection, name):
|
|
if entry(connection, name).get('pwdAccountLockedTime') == [LOCKED]:
|
|
connection.modify_s(dn_of(name), [(ldap.MOD_DELETE, 'pwdAccountLockedTime', None)])
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def unlock(connection, name):
|
|
"""End a lockout after failed binds and clear the failure count; a disabled user stays disabled."""
|
|
entry(connection, name)
|
|
current = connection.search_s(dn_of(name), ldap.SCOPE_BASE, attrlist=['pwdAccountLockedTime', 'pwdFailureTime'])[0][1]
|
|
changes = [(ldap.MOD_DELETE, 'pwdFailureTime', None)] if 'pwdFailureTime' in current else []
|
|
if 'pwdAccountLockedTime' in current and current['pwdAccountLockedTime'] != [LOCKED]:
|
|
changes.append((ldap.MOD_DELETE, 'pwdAccountLockedTime', None))
|
|
if changes:
|
|
connection.modify_s(dn_of(name), changes)
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def login(name, password, code):
|
|
"""Check the user's password, and the TOTP code of an enrolled user, with a bind as the user.
|
|
|
|
A wrong name, password or code, a locked or a disabled user give the same Failure, status 5.
|
|
"""
|
|
refused = Failure('the name, the password or the code is wrong', 5)
|
|
if not isinstance(name, str) or not NAME.match(name) or not isinstance(password, str) or not password:
|
|
raise refused
|
|
if code is not None and code != '' and (not isinstance(code, str) or not CODE.match(code)):
|
|
raise refused
|
|
connection = ldap.initialize(vpnrouter_ldap.URI)
|
|
try:
|
|
connection.simple_bind_s(dn_of(name), password + (code or ''))
|
|
except ldap.INVALID_CREDENTIALS:
|
|
raise refused
|
|
finally:
|
|
connection.unbind_s()
|
|
|
|
|
|
def is_admin(connection, name):
|
|
"""Whether the user exists and is a member of ADMIN_GROUP."""
|
|
try:
|
|
entry(connection, name)
|
|
except Failure:
|
|
return False
|
|
return ADMIN_GROUP in groups_of(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.
|
|
|
|
A user who is the last member of a group is not removed.
|
|
"""
|
|
entry(connection, name)
|
|
groups = groups_of(connection, name)
|
|
alone = sorted(group for group, members in groups.items() if members == [name])
|
|
if alone:
|
|
raise Failure(f'user {name} is the last member of {", ".join(alone)}; remove the group first', 2)
|
|
for group in groups:
|
|
connection.modify_s(group_dn(group), [(ldap.MOD_DELETE, 'member', [dn_of(name).encode()])])
|
|
cert, key = certificate(name)
|
|
if cert.is_file():
|
|
simple_ca('revoke-cert', '--ca-dir', str(vpnrouter.PKI_DIR), str(cert))
|
|
simple_ca('make-crl', '--ca-dir', str(vpnrouter.PKI_DIR), '--days', str(vpnrouter.CRL_DAYS))
|
|
publish_crl()
|
|
cert.unlink(missing_ok=True)
|
|
key.unlink(missing_ok=True)
|
|
connection.delete_s(dn_of(name))
|
|
return {'name': name, 'removed': True}
|
|
|
|
|
|
def profile(connection, name):
|
|
"""The OpenVPN client profile of the user, with the CA, the certificate and the key inline."""
|
|
entry(connection, name)
|
|
cfg = vpnrouter.load_config()
|
|
if not cfg['openvpn']['enabled']:
|
|
raise Failure('OpenVPN is not enabled', 2)
|
|
cert, key = certificate(name)
|
|
if not cert.is_file() or not key.is_file():
|
|
raise Failure(f'user {name} has no certificate')
|
|
fqdn = cfg['wan']['local_fqdn'].strip()
|
|
text = '\n'.join([
|
|
'client',
|
|
'dev tun',
|
|
f'proto {cfg["openvpn"]["protocol"]}',
|
|
f'remote {fqdn} {cfg["openvpn"]["port"]}',
|
|
'nobind',
|
|
'persist-key',
|
|
'persist-tun',
|
|
'remote-cert-tls server',
|
|
f'verify-x509-name {fqdn} name',
|
|
'auth-user-pass',
|
|
'auth-nocache',
|
|
'verb 3',
|
|
'<ca>', extract_pem(vpnrouter.CA_CERT_DEST.read_text(encoding='utf-8')), '</ca>',
|
|
'<cert>', extract_pem(cert.read_text(encoding='utf-8')), '</cert>',
|
|
'<key>', key.read_text(encoding='utf-8').strip(), '</key>',
|
|
'',
|
|
])
|
|
return {'name': name, 'profile': text}
|
|
|
|
|
|
def group_dn(name):
|
|
return f'cn={name},{vpnrouter_ldap.GROUPS}'
|
|
|
|
|
|
def check_group_name(name):
|
|
if not isinstance(name, str) or not NAME.match(name):
|
|
raise Failure('a group name is 1 to 32 lowercase letters, digits and hyphens, starting with a letter', 2)
|
|
|
|
|
|
def check_description(value):
|
|
"""The description, stripped; '' means none."""
|
|
if not isinstance(value, str):
|
|
raise Failure('the description must be a string', 2)
|
|
value = value.strip()
|
|
if len(value) > DESCRIPTION_MAX or any(not character.isprintable() for character in value):
|
|
raise Failure(f'the description must have at most {DESCRIPTION_MAX} printable characters', 2)
|
|
return value
|
|
|
|
|
|
def member_names(attributes):
|
|
"""The user names of a group's member DNs, sorted."""
|
|
names = []
|
|
for value in attributes.get('member', []):
|
|
dn = value.decode()
|
|
if dn.startswith('uid=') and dn.endswith(f',{vpnrouter_ldap.USERS}'):
|
|
names.append(dn[4:-len(f',{vpnrouter_ldap.USERS}')])
|
|
return sorted(names)
|
|
|
|
|
|
def group_summary(attributes):
|
|
"""A group as the answers show it."""
|
|
values = attributes.get('description')
|
|
return {
|
|
'name': attributes['cn'][0].decode(),
|
|
'description': values[0].decode() if values else '',
|
|
'members': member_names(attributes),
|
|
}
|
|
|
|
|
|
def group_entry(connection, name):
|
|
"""The attributes of the group's entry; Failure with status 3 when there is none."""
|
|
check_group_name(name)
|
|
try:
|
|
return connection.search_s(group_dn(name), ldap.SCOPE_BASE, attrlist=GROUP_ATTRIBUTES)[0][1]
|
|
except ldap.NO_SUCH_OBJECT:
|
|
raise Failure(f'group {name} does not exist', 3)
|
|
|
|
|
|
def groups_of(connection, name):
|
|
"""The groups the user is a member of, as {group name: its member names}."""
|
|
found = connection.search_s(vpnrouter_ldap.GROUPS, ldap.SCOPE_ONELEVEL,
|
|
f'(&(objectClass=groupOfNames)(member={ldap.filter.escape_filter_chars(dn_of(name))}))',
|
|
GROUP_ATTRIBUTES)
|
|
return {attributes['cn'][0].decode(): member_names(attributes) for _, attributes in found}
|
|
|
|
|
|
def list_groups(connection):
|
|
found = connection.search_s(vpnrouter_ldap.GROUPS, ldap.SCOPE_ONELEVEL, '(objectClass=groupOfNames)', GROUP_ATTRIBUTES)
|
|
return {'groups': sorted((group_summary(attributes) for _, attributes in found), key=lambda group: group['name'])}
|
|
|
|
|
|
def group_add(connection, name, description, members):
|
|
"""Add a group with a description ('' for none) and at least one member, each an existing user."""
|
|
check_group_name(name)
|
|
description = check_description(description if description is not None else '')
|
|
if not isinstance(members, list) or not members:
|
|
raise Failure('a group needs at least one member', 2)
|
|
for member in members:
|
|
entry(connection, member)
|
|
record = {'objectClass': [b'groupOfNames'], 'cn': [name.encode()],
|
|
'member': [dn_of(member).encode() for member in sorted(set(members))]}
|
|
if description:
|
|
record['description'] = [description.encode()]
|
|
try:
|
|
connection.add_s(group_dn(name), ldap.modlist.addModlist(record))
|
|
except ldap.ALREADY_EXISTS:
|
|
raise Failure(f'group {name} exists', 4)
|
|
return group_summary(group_entry(connection, name))
|
|
|
|
|
|
def group_update(connection, name, new_name=None, description=None):
|
|
"""Rename the group or change its description ('' removes it); at least one must be given."""
|
|
if new_name is None and description is None:
|
|
raise Failure('give a new name or a description', 2)
|
|
group_entry(connection, name)
|
|
if description is not None:
|
|
description = check_description(description)
|
|
change = (ldap.MOD_REPLACE, 'description', [description.encode()]) if description else \
|
|
(ldap.MOD_REPLACE, 'description', [])
|
|
connection.modify_s(group_dn(name), [change])
|
|
if new_name is not None and new_name != name:
|
|
check_group_name(new_name)
|
|
try:
|
|
connection.rename_s(group_dn(name), f'cn={new_name}', delold=1)
|
|
except ldap.ALREADY_EXISTS:
|
|
raise Failure(f'group {new_name} exists', 4)
|
|
name = new_name
|
|
return group_summary(group_entry(connection, name))
|
|
|
|
|
|
def group_remove(connection, name):
|
|
group_entry(connection, name)
|
|
connection.delete_s(group_dn(name))
|
|
return {'name': name, 'removed': True}
|
|
|
|
|
|
def group_member_add(connection, name, user):
|
|
attributes = group_entry(connection, name)
|
|
entry(connection, user)
|
|
if user not in member_names(attributes):
|
|
connection.modify_s(group_dn(name), [(ldap.MOD_ADD, 'member', [dn_of(user).encode()])])
|
|
return group_summary(group_entry(connection, name))
|
|
|
|
|
|
def group_member_remove(connection, name, user):
|
|
"""Take the user out of the group; the last member cannot leave."""
|
|
check_name(user)
|
|
members = member_names(group_entry(connection, name))
|
|
if user not in members:
|
|
raise Failure(f'user {user} is not a member of group {name}', 3)
|
|
if members == [user]:
|
|
raise Failure(f'user {user} is the last member of group {name}; remove the group instead', 2)
|
|
connection.modify_s(group_dn(name), [(ldap.MOD_DELETE, 'member', [dn_of(user).encode()])])
|
|
return group_summary(group_entry(connection, name))
|
|
|
|
|
|
def extract_pem(text):
|
|
"""The PEM blocks of the certificates in a file that also holds their text form."""
|
|
begin, end = '-----BEGIN CERTIFICATE-----', '-----END CERTIFICATE-----'
|
|
blocks = []
|
|
start = text.find(begin)
|
|
while start != -1:
|
|
stop = text.index(end, start) + len(end)
|
|
blocks.append(text[start:stop])
|
|
start = text.find(begin, stop)
|
|
return '\n'.join(blocks)
|