226 lines
8.1 KiB
Python
Executable File
226 lines
8.1 KiB
Python
Executable File
#!/usr/bin/python3
|
|
"""Manage the users of the router's directory, whether or not a P2S system is enabled.
|
|
|
|
A user is an entry in the router's directory, 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. Passwords are read from the first line of standard input, never from the
|
|
command line. The answer is JSON on standard output; on failure it is {"error": ...} and
|
|
the exit status says why: 2 the request is invalid, 3 the user does not exist, 4 the user
|
|
exists, 1 anything else.
|
|
|
|
vpn-router-user list
|
|
vpn-router-user add <name> (password on stdin)
|
|
vpn-router-user passwd <name> (password on stdin)
|
|
vpn-router-user disable <name>
|
|
vpn-router-user enable <name>
|
|
vpn-router-user remove <name>
|
|
vpn-router-user profile <name>
|
|
"""
|
|
|
|
import fcntl
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
|
|
sys.path.insert(0, '/usr/lib/vpn-router')
|
|
|
|
import ldap
|
|
import ldap.modlist
|
|
|
|
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 value of pwdAccountLockedTime that locks an account until it is unlocked.
|
|
LOCKED = b'000001010000Z'
|
|
LOCK_FILE = pathlib.Path('/run/vpn-router/users.lock')
|
|
|
|
|
|
class Failure(Exception):
|
|
"""A request that cannot be done; status is the exit status."""
|
|
|
|
def __init__(self, message, status=1):
|
|
super().__init__(message)
|
|
self.status = status
|
|
|
|
|
|
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."""
|
|
try:
|
|
return connection.search_s(dn_of(name), ldap.SCOPE_BASE, attrlist=['uid', 'pwdAccountLockedTime'])[0][1]
|
|
except ldap.NO_SUCH_OBJECT:
|
|
raise Failure(f'user {name} does not exist', 3)
|
|
|
|
|
|
def password():
|
|
"""The password from the first line of standard input."""
|
|
value = sys.stdin.readline().rstrip('\n')
|
|
if not PASSWORD_MIN <= len(value) <= PASSWORD_MAX:
|
|
raise Failure(f'the password must have {PASSWORD_MIN} to {PASSWORD_MAX} characters', 2)
|
|
return value
|
|
|
|
|
|
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 summary(attributes):
|
|
name = attributes['uid'][0].decode()
|
|
return {
|
|
'name': name,
|
|
'enabled': 'pwdAccountLockedTime' not in attributes,
|
|
'certificate': certificate(name)[0].is_file(),
|
|
}
|
|
|
|
|
|
def list_users(connection):
|
|
found = connection.search_s(vpnrouter_ldap.USERS, ldap.SCOPE_ONELEVEL, '(objectClass=inetOrgPerson)',
|
|
['uid', 'pwdAccountLockedTime'])
|
|
return {'users': sorted((summary(attributes) for _, attributes in found), key=lambda user: user['name'])}
|
|
|
|
|
|
def add(connection, name):
|
|
secret = password()
|
|
if 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')
|
|
try:
|
|
connection.add_s(dn_of(name), ldap.modlist.addModlist({
|
|
'objectClass': [b'inetOrgPerson'], 'uid': [name.encode()], 'cn': [name.encode()], 'sn': [name.encode()],
|
|
}))
|
|
except ldap.ALREADY_EXISTS:
|
|
raise Failure(f'user {name} exists', 4)
|
|
connection.passwd_s(dn_of(name), None, secret)
|
|
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 passwd(connection, name):
|
|
secret = password()
|
|
entry(connection, name)
|
|
connection.passwd_s(dn_of(name), None, secret)
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def disable(connection, name):
|
|
if 'pwdAccountLockedTime' not in entry(connection, name):
|
|
connection.modify_s(dn_of(name), [(ldap.MOD_REPLACE, 'pwdAccountLockedTime', [LOCKED])])
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def enable(connection, name):
|
|
if 'pwdAccountLockedTime' in entry(connection, name):
|
|
connection.modify_s(dn_of(name), [(ldap.MOD_DELETE, 'pwdAccountLockedTime', None)])
|
|
return summary(entry(connection, name))
|
|
|
|
|
|
def remove(connection, name):
|
|
entry(connection, name)
|
|
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 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)
|
|
|
|
|
|
COMMANDS = {'list': list_users, 'add': add, 'passwd': passwd, 'disable': disable, 'enable': enable,
|
|
'remove': remove, 'profile': profile}
|
|
|
|
|
|
def main(argv):
|
|
try:
|
|
if not argv or argv[0] not in COMMANDS or len(argv) != (1 if argv[0] == 'list' else 2):
|
|
raise Failure('usage: vpn-router-user list | add|passwd|disable|enable|remove|profile <name>', 2)
|
|
if len(argv) == 2 and not NAME.match(argv[1]):
|
|
raise Failure('a user name is 1 to 32 lowercase letters, digits and hyphens, starting with a letter', 2)
|
|
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:
|
|
answer = COMMANDS[argv[0]](connection, *argv[1:])
|
|
finally:
|
|
connection.unbind_s()
|
|
except Failure as failure:
|
|
print(json.dumps({'error': str(failure)}))
|
|
return failure.status
|
|
except (ldap.LDAPError, OSError) as exc:
|
|
print(json.dumps({'error': f'the user directory failed: {exc}'}))
|
|
return 1
|
|
print(json.dumps(answer))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main(sys.argv[1:]))
|