[WebConsole] Sync to github. (#845)

Co-authored-by: root <[email protected]>

web_console_v2/README.md

@@ -2,7 +2,7 @@
 
 ### Docker
 ```shell
-docker build -t fedlearner_webconsole_v2 .
+docker build --memory 4G --tag fedlearner_webconsole_v2 .
 docker run --rm -it -p 1989:1989 -p 1990:1990 fedlearner_webconsole_v2
 ```
 Then visiting http://localhost:1989/ for the UI.

web_console_v2/api/.gitignore

@@ -8,3 +8,5 @@ fedlearner_webconsole/proto/*.pyi
 # Coverage generated
 .coverage_html_report/
 .coverage*
+
+root.log.*

web_console_v2/api/command.py

@@ -15,8 +15,11 @@
 # coding: utf-8
 from config import Config
 from fedlearner_webconsole.app import create_app
-from fedlearner_webconsole.db import db
+from fedlearner_webconsole.db import db_handler as db
 from fedlearner_webconsole.initial_db import initial_db
+from flask_migrate import Migrate
+
+from fedlearner_webconsole.utils.hooks import pre_start_hook
 
 
 class CliConfig(Config):
@@ -25,7 +28,10 @@ class CliConfig(Config):
     START_COMPOSER = False
 
 
+pre_start_hook()
+migrate = Migrate()
 app = create_app(CliConfig())
+migrate.init_app(app, db)
 
 
 @app.cli.command('create-initial-data')
@@ -36,4 +42,3 @@ def create_initial_data():
 @app.cli.command('create-db')
 def create_db():
     db.create_all()
-    db.session.commit()

web_console_v2/api/config.py

@@ -20,8 +20,6 @@
 from fedlearner_webconsole.db import get_database_uri
 from envs import Envs
 
-BASE_DIR = os.path.abspath(os.path.dirname(__file__))
-
 
 class Config(object):
     SQLALCHEMY_DATABASE_URI = get_database_uri()

web_console_v2/api/envs.py

@@ -6,9 +6,9 @@
 
 class Envs(object):
     TZ = pytz.timezone(os.environ.get('TZ', 'UTC'))
-    HDFS_SERVER = os.environ.get('HDFS_SERVER', None)
     ES_HOST = os.environ.get('ES_HOST',
                              'fedlearner-stack-elasticsearch-client')
+    ES_READ_HOST = os.environ.get('ES_READ_HOST', ES_HOST)
     ES_PORT = os.environ.get('ES_PORT', 9200)
     ES_USERNAME = os.environ.get('ES_USERNAME', 'elastic')
     ES_PASSWORD = os.environ.get('ES_PASSWORD', 'Fedlearner123')
@@ -38,6 +38,16 @@ class Envs(object):
     GRPC_CLIENT_TIMEOUT = os.environ.get('GRPC_CLIENT_TIMEOUT', 5)
     # storage filesystem
     STORAGE_ROOT = os.getenv('STORAGE_ROOT', '/data')
+    # BASE_DIR
+    BASE_DIR = os.path.abspath(os.path.dirname(__file__))
+    # spark on k8s image url
+    SPARKAPP_IMAGE_URL = os.getenv('SPARKAPP_IMAGE_URL', None)
+    SPARKAPP_FILES_PATH = os.getenv('SPARKAPP_FILES_PATH', None)
+    SPARKAPP_VOLUMES = os.getenv('SPARKAPP_VOLUMES', None)
+    SPARKAPP_VOLUME_MOUNTS = os.getenv('SPARKAPP_VOLUME_MOUNTS', None)
+
+    # Hooks
+    PRE_START_HOOK = os.environ.get('PRE_START_HOOK', None)
 
 
 class Features(object):

web_console_v2/api/fedlearner_webconsole/app.py

@@ -14,21 +14,18 @@
 
 # coding: utf-8
 # pylint: disable=wrong-import-position, global-statement
-import importlib
 import logging
 import logging.config
 import os
 import traceback
 
 from http import HTTPStatus
 from flask import Flask, jsonify
-from flask_migrate import Migrate
 from flask_restful import Api
 from flask_jwt_extended import JWTManager
 from envs import Envs
 from fedlearner_webconsole.utils import metrics
 
-migrate = Migrate()
 jwt = JWTManager()
 
 from fedlearner_webconsole.auth.apis import initialize_auth_apis
@@ -113,6 +110,8 @@ def user_lookup_callback(jwt_header, jwt_data):
 
 @jwt.token_in_blocklist_loader
 def check_if_token_invalid(jwt_header, jwt_data):
+    del jwt_header  # unused by check_if_token_invalid
+
     jti = jwt_data['jti']
     session = Session.query.filter_by(jti=jti).first()
     return session is None
@@ -122,18 +121,9 @@ def create_app(config):
     # format logging
     logging.config.dictConfig(LOGGING_CONFIG)
 
-    before_hook_path = os.getenv('FEDLEARNER_WEBCONSOLE_BEFORE_APP_START')
-    if before_hook_path:
-        module_path, func_name = before_hook_path.split(':')
-        module = importlib.import_module(module_path)
-        # Dynamically run the function
-        getattr(module, func_name)()
-
     app = Flask('fedlearner_webconsole')
     app.config.from_object(config)
 
-    db.init_app(app)
-    migrate.init_app(app, db)
     jwt.init_app(app)
 
     # Error handlers
@@ -142,6 +132,9 @@ def create_app(config):
     app.register_error_handler(WebConsoleApiException, make_response)
     app.register_error_handler(Exception, _handle_uncaught_exception)
 
+    # TODO(wangsen.0914): This will be removed sooner!
+    db.init_app(app)
+
     api = Api(prefix='/api/v2')
     initialize_auth_apis(api)
     initialize_project_apis(api)
@@ -152,7 +145,7 @@ def create_app(config):
     initialize_setting_apis(api)
     initialize_mmgr_apis(api)
     initialize_sparkapps_apis(api)
-    if os.environ.get('FLASK_ENV') != 'production':
+    if os.environ.get('FLASK_ENV') != 'production' or Envs.DEBUG:
         initialize_debug_apis(api)
     # A hack that use our customized error handlers
     # Ref: https://github.com/flask-restful/flask-restful/issues/280

web_console_v2/api/fedlearner_webconsole/auth/apis.py

@@ -14,12 +14,15 @@
 
 # coding: utf-8
 # pylint: disable=cyclic-import
+import re
 import datetime
 from http import HTTPStatus
 from flask import request
 from flask_restful import Resource, reqparse
 from flask_jwt_extended.utils import get_current_user
 from flask_jwt_extended import create_access_token, decode_token, get_jwt
+
+from fedlearner_webconsole.utils.base64 import base64decode
 from fedlearner_webconsole.utils.decorators import jwt_required
 
 from fedlearner_webconsole.utils.decorators import admin_required
@@ -32,6 +35,28 @@
                                               UnauthorizedException,
                                               NoAccessException)
 
+# rule: password must have a letter, a num and a special character
+PASSWORD_FORMAT_L = re.compile(r'.*[A-Za-z]')
+PASSWORD_FORMAT_N = re.compile(r'.*[0-9]')
+PASSWORD_FORMAT_S = re.compile(r'.*[`!@#$%^&*()\-_=+|{}\[\];:\'\",<.>/?~]')
+
+
+def check_password_format(password: str):
+    if not 8 <= len(password) <= 20:
+        raise InvalidArgumentException(
+            'Password is not legal: 8 <= length <= 20')
+    required_chars = []
+    if PASSWORD_FORMAT_L.match(password) is None:
+        required_chars.append('a letter')
+    if PASSWORD_FORMAT_N.match(password) is None:
+        required_chars.append('a num')
+    if PASSWORD_FORMAT_S.match(password) is None:
+        required_chars.append('a special character')
+    if required_chars:
+        tip = ', '.join(required_chars)
+        raise InvalidArgumentException(
+            f'Password is not legal: must have {tip}.')
+
 
 class SigninApi(Resource):
     def post(self):
@@ -44,11 +69,11 @@ def post(self):
                             help='password is empty')
         data = parser.parse_args()
         username = data['username']
-        password = data['password']
+        password = base64decode(data['password'])
         user = User.query.filter_by(username=username).filter_by(
             state=State.ACTIVE).first()
         if user is None:
-            raise NotFoundException()
+            raise NotFoundException(f'Failed to find user: {username}')
         if not user.verify_password(password):
             raise UnauthorizedException('Invalid password')
         token = create_access_token(identity=username)
@@ -61,11 +86,11 @@ def post(self):
         db.session.commit()
 
         return {
-            'data': {
-                'user': user.to_dict(),
-                'access_token': token
-            }
-        }, HTTPStatus.OK
+                   'data': {
+                       'user': user.to_dict(),
+                       'access_token': token
+                   }
+               }, HTTPStatus.OK
 
     @jwt_required()
     def delete(self):
@@ -105,11 +130,13 @@ def post(self):
 
         data = parser.parse_args()
         username = data['username']
-        password = data['password']
+        password = base64decode(data['password'])
         role = data['role']
         name = data['name']
         email = data['email']
 
+        check_password_format(password)
+
         if User.query.filter_by(username=username).first() is not None:
             raise ResourceConflictException(
                 'user {} already exists'.format(username))
@@ -129,7 +156,8 @@ class UserApi(Resource):
     def _find_user(self, user_id) -> User:
         user = User.query.filter_by(id=user_id).first()
         if user is None or user.state == State.DELETED:
-            raise NotFoundException()
+            raise NotFoundException(
+                f'Failed to find user_id: {user_id}')
         return user
 
     def _check_current_user(self, user_id, msg):
@@ -158,7 +186,9 @@ def patch(self, user_id):
             if k not in mutable_attrs:
                 raise InvalidArgumentException(f'cannot edit {k} attribute!')
             if k == 'password':
-                user.set_password(v)
+                password = base64decode(v)
+                check_password_format(password)
+                user.set_password(password)
             else:
                 setattr(user, k, v)
 

web_console_v2/api/fedlearner_webconsole/composer/composer.py

@@ -407,4 +407,4 @@ def _build_pipeline(name: str, items: List[IItem],
 
 
 composer = Composer(config=ComposerConfig(
-    runner_fn=global_runner_fn, name='scheduler for fedlearner webconsole'))
+    runner_fn=global_runner_fn(), name='scheduler for fedlearner webconsole'))

web_console_v2/api/fedlearner_webconsole/composer/interface.py

@@ -22,10 +22,13 @@
 from fedlearner_webconsole.composer.models import Context, RunnerStatus
 
 
+# NOTE: remember to register new item in `global_runner_fn` \
+# which defined in `runner.py`
 class ItemType(enum.Enum):
-    TASK = 'task'
+    TASK = 'task'  # test only
     MEMORY = 'memory'
     WORKFLOW_CRON_JOB = 'workflow_cron_job'
+    DATA_PIPELINE = 'data_pipeline'
 
 
 # item interface

web_console_v2/api/fedlearner_webconsole/composer/runner.py

@@ -16,12 +16,14 @@
 import datetime
 import logging
 import random
+import sys
 import time
 from typing import Tuple
 
 from fedlearner_webconsole.composer.interface import IItem, IRunner, ItemType
 from fedlearner_webconsole.composer.models import Context, RunnerStatus, \
     SchedulerRunner
+from fedlearner_webconsole.dataset.data_pipeline import DataPipelineRunner
 from fedlearner_webconsole.db import get_session
 from fedlearner_webconsole.workflow.cronjob import WorkflowCronJob
 
@@ -77,8 +79,16 @@ def result(self, context: Context) -> Tuple[RunnerStatus, dict]:
         return RunnerStatus.DONE, {}
 
 
-# register runner_fn
-global_runner_fn = {
-    ItemType.MEMORY.value: MemoryRunner,
-    ItemType.WORKFLOW_CRON_JOB.value: WorkflowCronJob,
-}
+def global_runner_fn():
+    # register runner_fn
+    runner_fn = {
+        ItemType.MEMORY.value: MemoryRunner,
+        ItemType.WORKFLOW_CRON_JOB.value: WorkflowCronJob,
+        ItemType.DATA_PIPELINE.value: DataPipelineRunner,
+    }
+    for item in ItemType:
+        if item.value in runner_fn or item == ItemType.TASK:
+            continue
+        logging.error(f'failed to find item, {item.value}')
+        sys.exit(-1)
+    return runner_fn

web_console_v2/api/fedlearner_webconsole/composer/runner_cache.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 # coding: utf-8
+import logging
 import threading
 
 from fedlearner_webconsole.composer.interface import IRunner
@@ -37,6 +38,11 @@ def find_runner(self, runner_id: int, runner_name: str) -> IRunner:
             if obj:
                 return obj
             item_type, item_id = runner_name.rsplit('_', 1)
+            if item_type not in self.runner_fn:
+                logging.error(
+                    f'failed to find item_type {item_type} in runner_fn, '
+                    f'please register it in global_runner_fn')
+                raise ValueError(f'unknown item_type {item_type} in runner')
             obj = self.runner_fn[item_type](int(item_id))
             self._cache[key] = obj
             return obj