1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
# Copyright (c) 2016 Catalyst IT Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import logging
from osc_lib.command import command
from osc_lib.i18n import _
from adjutantclient import client as adjutant_client
LOG = logging.getLogger(__name__)
def _list_tokens(client, filters={}):
tokens = client.tokens.list(filters=filters)
headers = ['Token', 'Task', 'Expires on', 'Created on']
rows = [[token.token, token.task, token.expires,
token.created_on] for token in tokens]
return headers, rows
class TokenList(command.Lister):
"""Lists adjutant tokens. """
def get_parser(self, prog_name):
parser = super(TokenList, self).get_parser(prog_name)
parser.add_argument(
'--filters', metavar='<filters>',
required=False,
help=_('JSON containing filters for tokens.'),
default={})
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.admin_logic
return _list_tokens(client, parsed_args.filters)
class TokenShow(command.ShowOne):
"""Show details of one token."""
def get_parser(self, prog_name):
parser = super(TokenShow, self).get_parser(prog_name)
parser.add_argument(
'token', metavar='<token_id>',
help=_("The token."))
parser.add_argument(
'--bypass-url', metavar='<bypass-url>', default=None,
help=_('Bypasss URL for unauthenticated access to the endpoint.'))
return parser
def take_action(self, parsed_args):
if not parsed_args.bypass_url:
self.app.client_manager._auth_required = True
self.app.client_manager.setup_auth()
client = self.app.client_manager.admin_logic
else:
client = adjutant_client.Client("1", parsed_args.bypass_url)
token = client.tokens.get(parsed_args.token)
return zip(*(token.to_dict()).items())
class TokenSubmit(command.Command):
"""Submit token data."""
def get_parser(self, prog_name):
parser = super(TokenSubmit, self).get_parser(prog_name)
parser.add_argument(
'token', metavar='<token_id>', help=_('The token.'))
parser.add_argument(
'data', metavar='<token_data>',
help=_('Submission data for the token. Must be valid json.'))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.admin_logic
resp = client.tokens.submit(
parsed_args.token, json.loads(parsed_args.data))
print('Success', ' '.join(resp.notes))
class TokenClear(command.Lister):
"""Clear Expired tokens, admin only."""
def take_action(self, parsed_args):
client = self.app.client_manager.admin_logic
resp = client.tokens.clear_expired()
print('Success. ' + ' '.join(resp.json()['notes']))
return _list_tokens(client)
|
