feat: jans-cli use test client (ref: #1283) (#1285)

* feat: jans-cli use test client (ref: #1283)

* feat: jans-linux-setup add option --cli-test-client (ref: #1283)

* feat: jans-cli code smells

* feat: jans-cli use test client only if arg -use-test-client provided

jans-cli/cli/config_cli.py

@@ -121,6 +121,9 @@ def join(self):
 parser.add_argument("--key-password", help="Password for SSL Key file")
 parser.add_argument("-noverify", help="Ignore verifying the SSL certificate", action='store_true', default=True)
 
+parser.add_argument("-use-test-client", help="Use test client without device authorization", action='store_true')
+
+
 parser.add_argument("--patch-add", help="Colon delimited key:value pair for add patch operation. For example loggingLevel:DEBUG")
 parser.add_argument("--patch-replace", help="Colon delimited key:value pair for replace patch operation. For example loggingLevel:DEBUG")
 parser.add_argument("--patch-remove", help="Key for remove patch operation. For example imgLocation")
@@ -131,6 +134,7 @@ def join(self):
 
 ################## end of arguments #################
 
+test_client = args.use_test_client
 
 
 if args.plugins:
@@ -159,12 +163,21 @@ def write_config():
     if config_ini_fn.exists():
         config.read_string(config_ini_fn.read_text())
         host = config['DEFAULT']['jans_host']
-        client_id = config['DEFAULT']['jca_client_id']
-        if config['DEFAULT'].get('jca_client_secret'):
-            client_secret = config['DEFAULT']['jca_client_secret']
-        elif config['DEFAULT'].get('jca_client_secret_enc'):
-            client_secret_enc = config['DEFAULT']['jca_client_secret_enc']
+
+        if 'jca_test_client_id' in config['DEFAULT'] and test_client:
+            client_id = config['DEFAULT']['jca_test_client_id']
+            secret_key_str = 'jca_test_client_secret'
+        else:
+            client_id = config['DEFAULT']['jca_client_id']
+            secret_key_str = 'jca_client_secret'
+
+        secret_enc_key_str = secret_key_str + '_enc'
+        if config['DEFAULT'].get(secret_key_str):
+            client_secret = config['DEFAULT'][secret_key_str]
+        elif config['DEFAULT'].get(secret_enc_key_str):
+            client_secret_enc = config['DEFAULT'][secret_enc_key_str]
             client_secret = encode_decode(client_secret_enc, decode=True)
+
         debug = config['DEFAULT'].get('debug')
         debug_log_file = config['DEFAULT'].get('debug_log_file')
     else:
@@ -257,36 +270,26 @@ def __contains__(self, child):
 
 class JCA_CLI:
 
-    def __init__(self, host, client_id, client_secret, access_token):
+    def __init__(self, host, client_id, client_secret, access_token, test_client=False):
         self.host = host
         self.client_id = client_id
         self.client_secret = client_secret
+        self.use_test_client = test_client
+
         self.swagger_configuration = swagger_client.Configuration()
         self.swagger_configuration.host = 'https://{}'.format(self.host)
         self.access_token = access_token or config['DEFAULT'].get('access_token')
 
-        for plugin_s in config['DEFAULT'].get(my_op_mode + '_plugins', '').split(','):
-            plugin = plugin_s.strip()
-            if plugin:
-                plugins.append(plugin)
+        self.set_user()
+        self.plugins()
 
         if not self.access_token and config['DEFAULT'].get('access_token_enc'):
             self.access_token = encode_decode(config['DEFAULT']['access_token_enc'], decode=True)
 
-
         if my_op_mode == 'scim':
             self.swagger_configuration.host += '/jans-scim/restv1/v2'
 
-        if args.noverify:
-            self.swagger_configuration.verify_ssl = False
-        else:
-            self.swagger_configuration.verify_ssl = True
-
-        if args.config_api_mtls_client_cert:
-            self.swagger_configuration.cert_file = args.config_api_mtls_client_cert
-
-        if args.config_api_mtls_client_key:
-            self.swagger_configuration.key_file = args.config_api_mtls_client_key
+        self.ssl_settings()
 
         self.swagger_configuration.debug = debug
         if self.swagger_configuration.debug:
@@ -298,6 +301,46 @@ def __init__(self, host, client_id, client_secret, access_token):
         self.make_menu()
         self.current_menu = self.menu
 
+
+    def set_user(self):
+        self.auth_username = None
+        self.auth_password = None
+        self.askuser = get_bool(config['DEFAULT'].get('askuser'))
+
+        if self.askuser:
+            if args.username:
+                self.auth_username = args.username
+            if args.password:
+                self.auth_password = args.password
+            elif args.j:
+                if os.path.isfile(args.j):
+                    with open(args.j) as reader:
+                        self.auth_password = reader.read()
+                else:
+                    print(args.j, "does not exist. Exiting ...")
+                    sys.exit()
+            if not (self.auth_username and self.auth_password):
+                print("I need username and password. Exiting ...")
+                sys.exit()
+
+    def plugins(self):
+        for plugin_s in config['DEFAULT'].get(my_op_mode + '_plugins', '').split(','):
+            plugin = plugin_s.strip()
+            if plugin:
+                plugins.append(plugin)
+
+    def ssl_settings(self):
+        if args.noverify:
+            self.swagger_configuration.verify_ssl = False
+        else:
+            self.swagger_configuration.verify_ssl = True
+
+        if args.config_api_mtls_client_cert:
+            self.swagger_configuration.cert_file = args.config_api_mtls_client_cert
+
+        if args.config_api_mtls_client_key:
+            self.swagger_configuration.key_file = args.config_api_mtls_client_key
+
     def drop_to_shell(self, mylocals):
         locals_ = locals()
         locals_.update(mylocals)
@@ -341,7 +384,7 @@ def check_connection(self):
 
 
     def check_access_token(self):
-        if not self.access_token:
+        if not self.access_token :
             print(self.colored_text("Access token was not found.", warning_color))
             return
 
@@ -410,6 +453,39 @@ def get_json_from_response(self, response):
                 pass
         return js_data
 
+    def get_scoped_access_token(self, scope):
+        sys.stderr.write("Getting access token for scope {}\n".format(scope))
+        rest = self.get_rest_client()
+        headers = urllib3.make_headers(basic_auth='{}:{}'.format(self.client_id, self.client_secret))
+        url = 'https://{}/jans-auth/restv1/token'.format(self.host)
+        headers['Content-Type'] = 'application/x-www-form-urlencoded'
+        if self.askuser:
+            post_params = {"grant_type": "password", "scope": scope, "username": self.auth_username,
+                           "password": self.auth_password}
+        else:
+            post_params = {"grant_type": "client_credentials", "scope": scope}
+
+        response = rest.POST(
+            url,
+            headers=headers,
+            post_params=post_params
+        )
+
+        try:
+            data = json.loads(response.data)
+            if 'access_token' in data:
+                self.swagger_configuration.access_token = data['access_token']
+            else:
+                sys.stderr.write("Error while getting access token")
+                sys.stderr.write(data)
+                sys.stderr.write('\n')
+        except Exception as e:
+            print("Error while getting access token")
+            sys.stderr.write(response.data)
+            sys.stderr.write(e)
+            sys.stderr.write('\n')
+
+
     def get_jwt_access_token(self):
 
         rest = self.get_rest_client()
@@ -523,10 +599,14 @@ def get_jwt_access_token(self):
 
 
     def get_access_token(self, scope):
-        self.check_access_token()
-        if not self.access_token:
+        if self.use_test_client:
+            self.get_scoped_access_token(scope)
+        elif not self.access_token:
+            self.check_access_token()
             self.get_jwt_access_token()
-        self.swagger_configuration.access_token = self.access_token
+
+        if not self.use_test_client:
+            self.swagger_configuration.access_token = self.access_token
 
     def print_exception(self, e):
         error_printed = False
@@ -2025,7 +2105,7 @@ def runApp(self):
 
 def main():
 
-    cli_object = JCA_CLI(host, client_id, client_secret, access_token)
+    cli_object = JCA_CLI(host, client_id, client_secret, access_token, test_client)
 
     try:
         if not access_token:

jans-linux-setup/jans_setup/setup_app/installers/config_api.py

@@ -21,7 +21,6 @@ class ConfigApiInstaller(JettyInstaller):
                 (os.path.join(Config.dist_jans_dir, 'jans-config-api.war'), os.path.join(base.current_app.app_info['JANS_MAVEN'], 'maven/io/jans/jans-config-api-server/{0}/jans-config-api-server-{0}.war').format(base.current_app.app_info['ox_version'])),
                 (os.path.join(Config.dist_jans_dir, 'scim-plugin.jar'), os.path.join(base.current_app.app_info['JANS_MAVEN'], 'maven/io/jans/jans-config-api/plugins/scim-plugin/{0}/scim-plugin-{0}-distribution.jar').format(base.current_app.app_info['ox_version'])),
                 (os.path.join(Config.dist_jans_dir, 'facter'), 'https://raw.githubusercontent.com/GluuFederation/gluu-snap/master/facter/facter'),
-                #(os.path.join(Config.data_dir, 'jans-config-api-swagger.yaml'), 'https://raw.githubusercontent.com/JanssenProject/jans/main/jans-config-api/docs/jans-config-api-swagger.yaml'),
                 (os.path.join(Config.dist_jans_dir, 'user-mgt-plugin.jar'), os.path.join(base.current_app.app_info['JANS_MAVEN'], 'maven/io/jans/jans-config-api/plugins/user-mgt-plugin/{0}/user-mgt-plugin-{0}-distribution.jar').format(base.current_app.app_info['ox_version'])),
                 ]
 

jans-linux-setup/jans_setup/setup_app/installers/jans_cli.py

@@ -89,6 +89,9 @@ def configure(self, options={}):
         if Config.install_config_api:
             config['DEFAULT']['jca_client_id'] = Config.role_based_client_id
             config['DEFAULT']['jca_client_secret_enc'] = Config.role_based_client_encoded_pw
+            if base.argsp.cli_test_client:
+                config['DEFAULT']['jca_test_client_id'] = Config.jca_client_id
+                config['DEFAULT']['jca_test_client_secret_enc'] = Config.jca_client_encoded_pw
 
         if Config.get('install_scim_server'):
             config['DEFAULT']['scim_client_id'] = Config.scim_client_id

jans-linux-setup/jans_setup/setup_app/utils/arg_parser.py

@@ -32,7 +32,6 @@
 parser.add_argument('-state', help="State field used for generating X.509 certificates")
 parser.add_argument('-country', help="Two letters country coude used for generating X.509 certificates")
 
-
 parser.add_argument('-rdbm-user', help="RDBM username")
 parser.add_argument('-rdbm-password', help="RDBM password")
 parser.add_argument('-rdbm-port', help="RDBM port")
@@ -54,6 +53,7 @@
 parser.add_argument('-setup-branch', help="Jannsen setup github branch", default='main')
 
 parser.add_argument('--disable-config-api-security', help="Turn off oauth2 security validation for jans-config-api", action='store_true')
+parser.add_argument('--cli-test-client', help="Use config api test client for CLI", action='store_true')
 
 if PROFILE != OPENBANKING_PROFILE: