#!/usr/bin/python3 """Manage the groups of the router's directory from the command line, with vpnrouter_users. A group has a name, an optional description and at least one member, each a user of the directory. The answer is text for people on standard output, and a failure is a message on standard error. With --json the answer is JSON on standard output, and a failure is {"error": ...} there too. The exit status says what failed: 2 the request is invalid, 3 the group or the user does not exist, 4 the group exists, 1 anything else. -h or --help shows the commands and their options. vpn-router-group [--json] list vpn-router-group [--json] add --member U [--member U ...] [--description D] vpn-router-group [--json] update [--name N] [--description D] vpn-router-group [--json] remove vpn-router-group [--json] member-add vpn-router-group [--json] member-remove """ import argparse import json import sys sys.path.insert(0, '/usr/lib/vpn-router') import ldap import vpnrouter_users from vpnrouter_users import Failure #: The help line of each command. HELP = { 'list': 'list the groups with their members', 'add': 'add a group with at least one member', 'update': 'rename the group or change its description', 'remove': 'remove the group; its members are kept', 'member-add': 'add a user to the group', 'member-remove': 'take a user out of the group; the last member cannot leave', } class Parser(argparse.ArgumentParser): """An argument parser whose errors are Failures with status 2, so that --json answers them as JSON.""" def error(self, message): raise Failure(f'{message}; see {self.prog} --help', 2) def parse(argv): parser = Parser(prog='vpn-router-group', description='Manage the groups of the router.') parser.add_argument('--json', action='store_true', help='answer in JSON') commands = parser.add_subparsers(dest='command', required=True, metavar='command', parser_class=Parser) for command, text in HELP.items(): sub = commands.add_parser(command, help=text, description=text[0].upper() + text[1:] + '.') if command != 'list': sub.add_argument('name', help='the group name') if command == 'add': sub.add_argument('--member', dest='members', action='append', required=True, metavar='USER', help='a member; give it once for each member') sub.add_argument('--description', metavar='TEXT', help='the description') if command == 'update': sub.add_argument('--name', dest='new_name', metavar='NAME', help='the new group name') sub.add_argument('--description', metavar='TEXT', help="the description; '' removes it") if command in ('member-add', 'member-remove'): sub.add_argument('user', help='the user name') return parser.parse_args(argv) def run(connection, options): """The answer of the command the options name.""" command = options.command if command == 'list': return vpnrouter_users.list_groups(connection) if command == 'add': return vpnrouter_users.group_add(connection, options.name, options.description, options.members) if command == 'update': return vpnrouter_users.group_update(connection, options.name, options.new_name, options.description) if command == 'remove': return vpnrouter_users.group_remove(connection, options.name) if command == 'member-add': return vpnrouter_users.group_member_add(connection, options.name, options.user) return vpnrouter_users.group_member_remove(connection, options.name, options.user) def text_of(answer): """The answer as text for people.""" if answer.get('removed'): return f'group {answer["name"]} removed' groups = answer['groups'] if 'groups' in answer else [answer] if not groups: return 'no groups' head = ('GROUP', 'MEMBERS', 'DESCRIPTION') rows = [(group['name'], ', '.join(group['members']), group['description'] or '-') for group in groups] widths = [max(len(row[column]) for row in (head, *rows)) for column in range(len(head))] return '\n'.join(' '.join(cell.ljust(width) for cell, width in zip(row, widths)).rstrip() for row in (head, *rows)) def main(argv): # A request that cannot be parsed is answered in JSON when --json is among the arguments. as_json = '--json' in argv try: options = parse(argv) as_json = options.json with vpnrouter_users.session() as connection: answer = run(connection, options) except Failure as failure: message, status = str(failure), failure.status except (ldap.LDAPError, OSError) as exc: message, status = f'the user directory failed: {exc}', 1 else: print(json.dumps(answer) if as_json else text_of(answer)) return 0 if as_json: print(json.dumps({'error': message})) else: print(f'vpn-router-group: {message}', file=sys.stderr) return status if __name__ == '__main__': sys.exit(main(sys.argv[1:]))