122 lines
5.4 KiB
Python
Executable File
122 lines
5.4 KiB
Python
Executable File
#!/usr/bin/python3
|
|
"""Manage the roles of the resources of the router's OpenID Provider from the command line, with vpnrouter_users.
|
|
|
|
A resource is a service account; a role of it has a name, an optional description, and the groups and accounts it is
|
|
assigned to, whose tokens for the resource carry it in their roles claim. 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 resource, the role, the
|
|
group or the account does not exist, 4 the role exists, 1 anything else. -h or --help shows the commands and their
|
|
options.
|
|
|
|
vpn-router-role [--json] list [<resource>]
|
|
vpn-router-role [--json] add <resource> <role> [--description D]
|
|
vpn-router-role [--json] remove <resource> <role>
|
|
vpn-router-role [--json] member-add <resource> <role> (--group G | --account A)
|
|
vpn-router-role [--json] member-remove <resource> <role> (--group G | --account A)
|
|
"""
|
|
|
|
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 roles of every resource, or of one, with the groups and accounts they are assigned to',
|
|
'add': 'add a role to a resource',
|
|
'remove': 'remove a role of a resource',
|
|
'member-add': 'assign the role to a group or an account',
|
|
'member-remove': 'take the role from a group or an account',
|
|
}
|
|
|
|
|
|
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-role', description='Manage the roles of the resources of the OpenID Provider.')
|
|
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('resource', nargs='?', help='the service account whose roles are listed')
|
|
continue
|
|
sub.add_argument('resource', help='the service account the role belongs to')
|
|
sub.add_argument('role', help='the role name')
|
|
if command == 'add':
|
|
sub.add_argument('--description', metavar='TEXT', help='the description')
|
|
if command in ('member-add', 'member-remove'):
|
|
member = sub.add_mutually_exclusive_group(required=True)
|
|
member.add_argument('--group', metavar='GROUP', help='the group')
|
|
member.add_argument('--account', metavar='USER', help='the user or service account')
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def run(connection, options):
|
|
"""The answer of the command the options name."""
|
|
command = options.command
|
|
if command == 'list':
|
|
if options.resource is not None:
|
|
vpnrouter_users.check_resource(connection, options.resource)
|
|
return vpnrouter_users.list_roles(connection, options.resource)
|
|
if command == 'add':
|
|
role = vpnrouter_users.role_add(connection, options.resource, options.role, options.description)
|
|
elif command == 'remove':
|
|
return vpnrouter_users.role_remove(connection, options.resource, options.role)
|
|
elif command == 'member-add':
|
|
role = vpnrouter_users.role_member_add(connection, options.resource, options.role, options.group, options.account)
|
|
else:
|
|
role = vpnrouter_users.role_member_remove(connection, options.resource, options.role, options.group, options.account)
|
|
return {'resource': options.resource, **role}
|
|
|
|
|
|
def text_of(answer):
|
|
"""The answer as text for people."""
|
|
if answer.get('removed'):
|
|
return f'role {answer["name"]} of {answer["resource"]} removed'
|
|
roles = answer['roles'] if 'roles' in answer else [answer]
|
|
if not roles:
|
|
return 'no roles'
|
|
head = ('RESOURCE', 'ROLE', 'GROUPS', 'ACCOUNTS', 'DESCRIPTION')
|
|
rows = [(role['resource'], role['name'], ', '.join(role['groups']) or '-', ', '.join(role['accounts']) or '-',
|
|
role['description'] or '-') for role in roles]
|
|
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-role: {message}', file=sys.stderr)
|
|
return status
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main(sys.argv[1:]))
|