feature: the OpenID Provider has the client credentials and device authorization grants, public clients, and service accounts as resources with roles and scopes in JWT access tokens.
Release / release (push) Successful in 17s

This commit is contained in:
slawek committed 2026-10-30 01:01:45 +01:00
1 parent 3ed79c7563
commit 3c4c5a8c32
15 files changed
+1572 -240

No files matched your search

+121
View File
@@ -0,0 +1,121 @@
#!/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:]))
+26 -8
View File
@@ -13,14 +13,17 @@ user exists, 1 anything else. -h or --help shows the commands and their options.
vpn-router-user [--json] add <name> [--email E] (a Manager-only account; password on stdin)
vpn-router-user [--json] add <name> --service [--display-name N] [--description D] [--email E]
[--redirect-uri U ...] [--post-logout-redirect-uri U ...] [--allowed-group G ...]
[--grant G ...] [--public-client yes|no] [--client-scope R:S ...] [--resource-scope S ...]
(a service account; password on stdin)
vpn-router-user [--json] update <name> [--first-name F] [--last-name L] [--email E]
vpn-router-user [--json] update <name> [--display-name N] [--description D] [--email E]
[--redirect-uri U ...] [--post-logout-redirect-uri U ...] [--allowed-group G ...]
[--grant G ...] [--public-client yes|no] [--client-scope R:S ...] [--resource-scope S ...]
(a service account)
A service account with a redirect URI is a client of the OpenID Provider; each of the three options may be repeated,
and replaces all values of its kind; one '' removes them.
A service account with a redirect URI or a grant (client_credentials, device_code) is a client of the OpenID Provider,
and one with a resource scope a resource; each of the repeatable options replaces all values of its kind, and one ''
removes them. vpn-router-role manages the roles of a resource.
vpn-router-user [--json] passwd <name> (password on stdin)
vpn-router-user [--json] disable <name>
vpn-router-user [--json] enable <name>
@@ -80,6 +83,9 @@ CLIENT_OPTIONS = (
('redirect_uris', 'redirect-uri', 'a redirect URI of the OpenID Provider client'),
('post_logout_redirect_uris', 'post-logout-redirect-uri', 'a post-logout redirect URI of the OpenID Provider client'),
('allowed_groups', 'allowed-group', 'a group whose members may sign in to the OpenID Provider client'),
('grants', 'grant', 'a grant the client may use besides the authorization code: client_credentials or device_code'),
('client_scopes', 'client-scope', 'a scope of a resource the client may ask for, <resource>:<scope>'),
('resource_scopes', 'resource-scope', 'a scope the service account defines as a resource'),
)
@@ -104,6 +110,8 @@ def parse(argv):
for key, option, text in CLIENT_OPTIONS:
sub.add_argument(f'--{option}', dest=key, action='append', metavar='VALUE',
help=f"{text}, of a service account; repeat it for more, '' for none")
sub.add_argument('--public-client', choices=('yes', 'no'),
help='whether the client of a service account is public, without a secret')
if command == 'add':
sub.add_argument('--service', action='store_true', help='add a service account, which cannot sign in to the web service')
if command == 'otp-enrol':
@@ -131,6 +139,8 @@ def run(connection, options, secret):
given = getattr(options, key, None)
if given is not None:
values[key] = [value for value in given if value]
if getattr(options, 'public_client', None) is not None:
values['public_client'] = options.public_client == 'yes'
if command == 'add':
return vpnrouter_users.add(connection, options.name, secret, values, service=options.service)
if command == 'update':
@@ -157,12 +167,20 @@ def console_text(user):
def client_text(user):
"""The OpenID Provider client settings of a service account as a cell: -, or the numbers of its URIs and its allowed
groups."""
if not user.get('redirect_uris'):
return '-'
groups = ','.join(user['allowed_groups']) or 'all users'
return f'{len(user["redirect_uris"])} redirect, {len(user["post_logout_redirect_uris"])} logout, {groups}'
"""The OpenID Provider settings of a service account as a cell: -, or the numbers of its URIs, its allowed groups,
grants, public, its client scopes and its resource scopes."""
parts = []
if user.get('redirect_uris'):
groups = ','.join(user['allowed_groups']) or 'all users'
parts.append(f'{len(user["redirect_uris"])} redirect, {len(user["post_logout_redirect_uris"])} logout, {groups}')
parts += user.get('grants') or []
if user.get('public_client'):
parts.append('public')
if user.get('client_scopes'):
parts.append('asks ' + ','.join(user['client_scopes']))
if user.get('resource_scopes'):
parts.append('defines ' + ','.join(user['resource_scopes']))
return '; '.join(parts) or '-'
def text_of(answer):