/
secrets.py
164 lines (146 loc) · 4.93 KB
/
secrets.py
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
# -*- coding: utf-8 -*-
#
# This file is part of REANA.
# Copyright (C) 2019 CERN.
#
# REANA is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""REANA client workflow related commands."""
import logging
import sys
import traceback
import click
from reana_client.cli.utils import (add_access_token_options, check_connection,
NotRequiredIf)
from reana_client.config import ERROR_MESSAGES
from reana_commons.errors import (REANASecretAlreadyExists,
REANASecretDoesNotExist)
from reana_commons.utils import click_table_printer
from reana_client.utils import parse_secret_from_path, \
parse_secret_from_literal
@click.group(help='Secret management commands')
@click.pass_context
def secrets_group(ctx):
"""Top level wrapper for secrets management."""
logging.debug(ctx.info_name)
@secrets_group.command()
@click.option(
'--env',
multiple=True,
cls=NotRequiredIf,
not_required_if='file',
help='Secrets to be uploaded from literal string.'
'e.g. PASSWORD=password123')
@click.option(
'--file',
multiple=True,
cls=NotRequiredIf,
type=click.Path(exists=True, file_okay=True, dir_okay=False,
readable=True),
not_required_if='env',
help='Secrets to be uploaded from file.')
@click.option(
'--overwrite',
is_flag=True,
default=False,
help='Overwrite the secret if already present')
@add_access_token_options
@check_connection
def secrets_add(env, file, overwrite, access_token): # noqa: D301
"""Add secrets from literal string or from file.
Examples: \n
\t $ reana-client secrets-add --env PASSWORD=password \n
\t $ reana-client secrets-add --file ~/.keytab \n
\t $ reana-client secrets-add --env USER=reanauser \n
\t --env PASSWORD=password \n
\t --file ~/.keytab
"""
from reana_client.api.client import add_secrets
secrets_ = {}
for literal in env:
secret = parse_secret_from_literal(literal)
secrets_.update(secret)
for path in file:
secret = parse_secret_from_path(path)
secrets_.update(secret)
try:
add_secrets(secrets_, overwrite, access_token)
except REANASecretAlreadyExists as e:
logging.debug(str(e), exc_info=True)
click.echo(
click.style(
'One of the secrets already exists. No secrets were added. '
'If you want to overwrite it use --overwrite option.',
fg='red'),
err=True)
sys.exit(1)
except Exception as e:
logging.debug(str(e), exc_info=True)
click.echo(
click.style(
'Something went wrong while uploading secrets',
fg='red'),
err=True)
else:
click.echo(
click.style('Secrets {} were successfully uploaded.'.format(
', '.join(secrets_.keys())),
fg='green')
)
@secrets_group.command()
@add_access_token_options
@check_connection
@click.argument('secrets', type=str, nargs=-1)
def secrets_delete(secrets, access_token): # noqa: D301
"""Delete user secrets by name.
Examples: \n
\t $ reana-client secrets-delete PASSWORD
"""
from reana_client.api.client import delete_secrets
try:
deleted_secrets = delete_secrets(secrets, access_token)
except REANASecretDoesNotExist as e:
logging.debug(str(e), exc_info=True)
click.echo(
click.style(
str('Secrets {} do not exist. Nothing was deleted'
.format(e.missing_secrets_list)
),
fg='red'),
err=True)
except Exception as e:
logging.debug(str(e), exc_info=True)
click.echo(
click.style(
'Something went wrong while deleting secrets',
fg='red'),
err=True)
else:
click.echo(
click.style('Secrets {} were successfully deleted.'.format(
', '.join(deleted_secrets)),
fg='green'))
@secrets_group.command()
@add_access_token_options
@check_connection
def secrets_list(access_token): # noqa: D301
"""List user secrets.
Examples: \n
\t $ reana-client secrets-list
"""
from reana_client.api.client import list_secrets
try:
secrets = list_secrets(access_token)
headers = ['name', 'type']
data = []
for secret_ in secrets:
data.append(list(map(str, [secret_['name'],
secret_['type']])))
click_table_printer(headers, headers, data)
except Exception as e:
logging.debug(str(e), exc_info=True)
click.echo(
click.style(
'Something went wrong while listing secrets',
fg='red'),
err=True)