221 lines
9.4 KiB
Python
221 lines
9.4 KiB
Python
"""The router's user directory in the local slapd.
|
|
|
|
The package keeps its own database, dc=vpn-router,dc=local in /var/lib/ldap/vpn-router,
|
|
next to the one slapd's installation creates. Root manages it through the local socket
|
|
with SASL EXTERNAL: the database's root DN is root's peer-credential DN, so there is no
|
|
administrator password. Users and groups are changed by the user-manager service account,
|
|
which the ACLs allow to write ou=users and ou=groups and to set, never read, passwords and
|
|
TOTP secrets.
|
|
|
|
The password policy locks an account for LOCKOUT_SECONDS after LOCKOUT_FAILURES failed binds;
|
|
a successful bind clears the count. slapd's otp overlay is configured on the database, so a
|
|
bind of a user with a TOTP token takes the password followed by a current code.
|
|
cn=totp,ou=policies holds the TOTP parameters the token of every user points to.
|
|
"""
|
|
|
|
import os
|
|
import pathlib
|
|
import pwd
|
|
import re
|
|
import secrets
|
|
|
|
import ldap
|
|
import ldap.modlist
|
|
import ldap.sasl
|
|
|
|
URI = 'ldapi:///'
|
|
BASE = 'dc=vpn-router,dc=local'
|
|
USERS = f'ou=users,{BASE}'
|
|
GROUPS = f'ou=groups,{BASE}'
|
|
SERVICES = f'ou=services,{BASE}'
|
|
POLICIES = f'ou=policies,{BASE}'
|
|
DEFAULT_POLICY = f'cn=default,{POLICIES}'
|
|
TOTP_PARAMS = f'cn=totp,{POLICIES}'
|
|
MANAGER = f'cn=user-manager,{SERVICES}'
|
|
ROOT = 'gidNumber=0+uidNumber=0,cn=peercred,cn=external,cn=auth'
|
|
|
|
DB_DIRECTORY = pathlib.Path('/var/lib/ldap/vpn-router')
|
|
SECRET_DIR = pathlib.Path('/etc/vpn-router/ldap')
|
|
MANAGER_SECRET = SECRET_DIR / 'user-manager.secret'
|
|
SLAPD_DEFAULTS = pathlib.Path('/etc/default/slapd')
|
|
SLAPD_SERVICES = 'SLAPD_SERVICES="ldapi:///"'
|
|
|
|
MODULES = ('argon2', 'ppolicy', 'otp')
|
|
PASSWORD_HASH = '{ARGON2}'
|
|
|
|
#: Failed binds that lock an account, within LOCKOUT_SECONDS, and the seconds it stays locked.
|
|
LOCKOUT_FAILURES = 5
|
|
LOCKOUT_SECONDS = 900
|
|
|
|
#: The overlays of the database, in the order they are configured.
|
|
OVERLAYS = ('ppolicy', 'otp')
|
|
|
|
#: The TOTP parameters: digits, the OID of HMAC-SHA1, seconds per time step, and the time steps accepted before and after now.
|
|
TOTP_DIGITS = 6
|
|
TOTP_ALGORITHM = '1.2.840.113549.2.7'
|
|
TOTP_PERIOD = 30
|
|
TOTP_WINDOW = 1
|
|
|
|
#: The ACLs of the database, in order.
|
|
ACCESS = (
|
|
f'to attrs=userPassword by dn.exact="{MANAGER}" =wx by self =wx by anonymous auth by * none',
|
|
f'to attrs=oathSecret by dn.exact="{MANAGER}" =w by * none',
|
|
f'to dn.subtree="{USERS}" by dn.exact="{MANAGER}" manage by * read',
|
|
f'to dn.subtree="{GROUPS}" by dn.exact="{MANAGER}" manage by * read',
|
|
'to * by * read',
|
|
)
|
|
|
|
#: The index of an ordered cn=config value: {0}, {1}, ...
|
|
ORDER = re.compile(rb'^\{\d+\}')
|
|
|
|
|
|
def unordered(values):
|
|
"""Values of an ordered cn=config attribute without their {n} prefixes."""
|
|
return [ORDER.sub(b'', value).decode() for value in values]
|
|
|
|
|
|
def root():
|
|
"""A connection to slapd as root, through the local socket."""
|
|
connection = ldap.initialize(URI)
|
|
connection.sasl_interactive_bind_s('', ldap.sasl.external())
|
|
return connection
|
|
|
|
|
|
def manager():
|
|
"""A connection to slapd as the user-manager service account."""
|
|
connection = ldap.initialize(URI)
|
|
connection.simple_bind_s(MANAGER, MANAGER_SECRET.read_text(encoding='utf-8'))
|
|
return connection
|
|
|
|
|
|
def ensure_services():
|
|
"""Make slapd listen on the local socket only. Returns True when the setting changed."""
|
|
text = SLAPD_DEFAULTS.read_text(encoding='utf-8')
|
|
lines = text.splitlines(keepends=True)
|
|
updated = ''.join(SLAPD_SERVICES + '\n' if line.startswith('SLAPD_SERVICES=') else line for line in lines)
|
|
if not any(line.startswith('SLAPD_SERVICES=') for line in lines):
|
|
updated = text.rstrip('\n') + '\n' + SLAPD_SERVICES + '\n'
|
|
if updated == text:
|
|
return False
|
|
SLAPD_DEFAULTS.write_text(updated, encoding='utf-8')
|
|
return True
|
|
|
|
|
|
def _ensure_modules(connection):
|
|
dn, attributes = connection.search_s('cn=config', ldap.SCOPE_ONELEVEL, '(objectClass=olcModuleList)', ['olcModuleLoad'])[0]
|
|
loaded = unordered(attributes.get('olcModuleLoad', []))
|
|
missing = [name for name in MODULES if name not in loaded and f'{name}.la' not in loaded]
|
|
if missing:
|
|
connection.modify_s(dn, [(ldap.MOD_ADD, 'olcModuleLoad', [name.encode() for name in missing])])
|
|
|
|
|
|
def _ensure_password_hash(connection):
|
|
dn = 'olcDatabase={-1}frontend,cn=config'
|
|
current = connection.search_s(dn, ldap.SCOPE_BASE, attrlist=['olcPasswordHash'])[0][1].get('olcPasswordHash', [])
|
|
if current != [PASSWORD_HASH.encode()]:
|
|
connection.modify_s(dn, [(ldap.MOD_REPLACE, 'olcPasswordHash', [PASSWORD_HASH.encode()])])
|
|
|
|
|
|
def _ensure_database(connection):
|
|
"""The DN of the package's database in cn=config, created when missing; its ACLs are kept as ACCESS."""
|
|
found = connection.search_s('cn=config', ldap.SCOPE_ONELEVEL, f'(olcSuffix={BASE})', ['olcAccess'])
|
|
if not found:
|
|
DB_DIRECTORY.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
openldap = pwd.getpwnam('openldap')
|
|
os.chown(DB_DIRECTORY, openldap.pw_uid, openldap.pw_gid)
|
|
connection.add_s('olcDatabase=mdb,cn=config', ldap.modlist.addModlist({
|
|
'objectClass': [b'olcDatabaseConfig', b'olcMdbConfig'],
|
|
'olcDatabase': [b'mdb'],
|
|
'olcSuffix': [BASE.encode()],
|
|
'olcDbDirectory': [str(DB_DIRECTORY).encode()],
|
|
'olcRootDN': [ROOT.encode()],
|
|
'olcDbIndex': [b'objectClass eq', b'uid eq'],
|
|
'olcAccess': [f'{{{index}}}{rule}'.encode() for index, rule in enumerate(ACCESS)],
|
|
}))
|
|
found = connection.search_s('cn=config', ldap.SCOPE_ONELEVEL, f'(olcSuffix={BASE})', ['olcAccess'])
|
|
dn, attributes = found[0]
|
|
if unordered(attributes.get('olcAccess', [])) != list(ACCESS):
|
|
connection.modify_s(dn, [(ldap.MOD_REPLACE, 'olcAccess', [f'{{{index}}}{rule}'.encode() for index, rule in enumerate(ACCESS)])])
|
|
return dn
|
|
|
|
|
|
def _ensure_overlays(connection, database):
|
|
"""Add the overlays of OVERLAYS the database lacks; the otp overlay has no settings of its own."""
|
|
found = connection.search_s(database, ldap.SCOPE_ONELEVEL, '(objectClass=olcOverlayConfig)', ['olcOverlay'])
|
|
present = {ORDER.sub(b'', attributes['olcOverlay'][0]).decode() for _, attributes in found}
|
|
for name in OVERLAYS:
|
|
if name in present:
|
|
continue
|
|
record = {'objectClass': [b'olcOverlayConfig'], 'olcOverlay': [name.encode()]}
|
|
if name == 'ppolicy':
|
|
record = {'objectClass': [b'olcOverlayConfig', b'olcPPolicyConfig'], 'olcOverlay': [b'ppolicy'],
|
|
'olcPPolicyDefault': [DEFAULT_POLICY.encode()]}
|
|
connection.add_s(f'olcOverlay={name},{database}', ldap.modlist.addModlist(record))
|
|
|
|
|
|
def _ensure_entry(connection, dn, attributes):
|
|
"""Add the entry when it is missing, else set the given attributes that differ."""
|
|
try:
|
|
current = connection.search_s(dn, ldap.SCOPE_BASE, attrlist=list(attributes))[0][1]
|
|
except ldap.NO_SUCH_OBJECT:
|
|
connection.add_s(dn, ldap.modlist.addModlist(attributes))
|
|
return
|
|
changes = [(ldap.MOD_REPLACE, name, values) for name, values in attributes.items()
|
|
if name != 'objectClass' and sorted(current.get(name, [])) != sorted(values)]
|
|
if changes:
|
|
connection.modify_s(dn, changes)
|
|
|
|
|
|
def _ensure_manager(connection):
|
|
SECRET_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(SECRET_DIR, 0o700)
|
|
created = False
|
|
try:
|
|
connection.search_s(MANAGER, ldap.SCOPE_BASE, attrlist=['1.1'])
|
|
except ldap.NO_SUCH_OBJECT:
|
|
connection.add_s(MANAGER, ldap.modlist.addModlist({
|
|
'objectClass': [b'organizationalRole', b'simpleSecurityObject'],
|
|
'cn': [b'user-manager'],
|
|
'userPassword': [b'{CRYPT}!'],
|
|
}))
|
|
created = True
|
|
if created or not MANAGER_SECRET.is_file():
|
|
password = secrets.token_urlsafe(32)
|
|
connection.passwd_s(MANAGER, None, password)
|
|
MANAGER_SECRET.touch(mode=0o600)
|
|
MANAGER_SECRET.chmod(0o600)
|
|
MANAGER_SECRET.write_text(password, encoding='utf-8')
|
|
|
|
|
|
def provision():
|
|
"""Create or bring up to date the modules, database, ACLs, password policy, entries and service account."""
|
|
connection = root()
|
|
try:
|
|
_ensure_modules(connection)
|
|
_ensure_password_hash(connection)
|
|
database = _ensure_database(connection)
|
|
_ensure_overlays(connection, database)
|
|
_ensure_entry(connection, BASE, {'objectClass': [b'dcObject', b'organization'], 'dc': [b'vpn-router'], 'o': [b'vpn-router']})
|
|
for unit in (USERS, GROUPS, SERVICES, POLICIES):
|
|
_ensure_entry(connection, unit, {'objectClass': [b'organizationalUnit'], 'ou': [unit.split(',')[0][3:].encode()]})
|
|
_ensure_entry(connection, DEFAULT_POLICY, {
|
|
'objectClass': [b'organizationalRole', b'pwdPolicy'],
|
|
'cn': [b'default'],
|
|
'pwdAttribute': [b'userPassword'],
|
|
'pwdLockout': [b'TRUE'],
|
|
'pwdMaxFailure': [str(LOCKOUT_FAILURES).encode()],
|
|
'pwdLockoutDuration': [str(LOCKOUT_SECONDS).encode()],
|
|
'pwdFailureCountInterval': [str(LOCKOUT_SECONDS).encode()],
|
|
})
|
|
_ensure_entry(connection, TOTP_PARAMS, {
|
|
'objectClass': [b'organizationalRole', b'oathTOTPParams'],
|
|
'cn': [b'totp'],
|
|
'oathOTPLength': [str(TOTP_DIGITS).encode()],
|
|
'oathHMACAlgorithm': [TOTP_ALGORITHM.encode()],
|
|
'oathTOTPTimeStepPeriod': [str(TOTP_PERIOD).encode()],
|
|
'oathTOTPTimeStepWindow': [str(TOTP_WINDOW).encode()],
|
|
})
|
|
_ensure_manager(connection)
|
|
finally:
|
|
connection.unbind_s()
|