Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
auth.add_argument('--conf.client_id', metavar='TEXT')
auth.add_argument('--conf.client_secret', metavar='TEXT')
auth.add_argument(
'--conf.scope', choices=['read', 'write'], default='write'
)
if client.help:
self.print_help(parser)
raise SystemExit()
parsed = parser.parse_known_args()[0]
kwargs = {
'client_id': getattr(parsed, 'conf.client_id', None),
'client_secret': getattr(parsed, 'conf.client_secret', None),
'scope': getattr(parsed, 'conf.scope', None),
}
try:
token = api.Api().get_oauth2_token(**kwargs)
except Exception as e:
self.print_help(parser)
cprint(
'Error retrieving an OAuth2.0 token ({}).'.format(e.__class__),
'red'
)
else:
fmt = client.get_config('format')
if fmt == 'human':
print('export TOWER_TOKEN={}'.format(token))
else:
print(to_str(FORMATTERS[fmt]({'token': token}, '.')).strip())
def get(self, **kwargs):
v2 = api.Api(connection=self.conn).get().current_version.get()
return getattr(v2, self.resource).get(**kwargs)
config.credentials = utils.load_credentials(
akit_args.credential_file)
else:
config.credentials = utils.PseudoNamespace({
'default': {
'username': os.getenv('AWXKIT_USER', 'admin'),
'password': os.getenv('AWXKIT_USER_PASSWORD', 'password')
}
})
if akit_args.project_file != utils.not_provided:
config.project_urls = utils.load_projects(
akit_args.project_file)
global root
root = api.Api()
if uses_sessions(root.connection):
config.use_sessions = True
root.load_session().get()
else:
root.load_authtoken().get()
if 'v2' in root.available_versions:
global v2
v2 = root.available_versions.v2.get()
rc = 0
if akit_args.akit_script:
try:
exec(open(akit_args.akit_script).read(), globals())
except Exception as e:
exc = e
config.credentials = utils.PseudoNamespace({
'default': {
'username': self.get_config('username'),
'password': self.get_config('password'),
}
})
_, remainder = self.parser.parse_known_args()
if remainder and remainder[0] == 'config':
# the config command is special; it doesn't require
# API connectivity
return
# ...otherwise, set up a awxkit connection because we're
# likely about to do some requests to /api/v2/
self.root = api.Api()
try:
self.fetch_version_root()
except RequestException:
# If we can't reach the API root (this usually means that the
# hostname is wrong, or the credentials are wrong)
if self.help:
# ...but the user specified -h...
known, unknown = self.parser.parse_known_args(self.argv)
if len(unknown) == 1 and os.path.basename(unknown[0]) == 'awx':
return
raise