servii-backend/app.py

import argparse
import http
import inspect
import os
from typing import Union

from apscheduler.schedulers.background import BackgroundScheduler
from firebase_admin.auth import UserNotFoundError, UserRecord
from flask import (Blueprint, Flask, Response, jsonify, request)
from flask_cors import CORS
from werkzeug import run_simple
from werkzeug.datastructures import ImmutableMultiDict, FileStorage
from werkzeug.utils import secure_filename

import file_manager
import firebase_manager
import generic_executor

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1000 * 1000 * 1000  #15.28MB~
cors = CORS(app, origins="*")
apiBP = Blueprint('apiBP', 'BPapi')


def generic_response_maker(status_code: http.HTTPStatus, _message: str = None) -> tuple[Response, int]:
    if _message is not None:
        return jsonify({'message': _message}), status_code.value
    match status_code:
        case http.HTTPStatus.CREATED:
            message = jsonify({'message': 'Creation successful.'})
        case http.HTTPStatus.INTERNAL_SERVER_ERROR:
            message = jsonify({'message': 'Internal Server Error.'})
        case http.HTTPStatus.NO_CONTENT:
            message = jsonify({'message': 'Deletion successful'})
        case http.HTTPStatus.ACCEPTED:
            message = jsonify({'message': 'Action successful.'})
        case http.HTTPStatus.BAD_REQUEST:
            message = jsonify({'message': 'Bad Request.'})
        case http.HTTPStatus.NOT_FOUND:
            message = jsonify({'message': 'Server not found.'})
        case http.HTTPStatus.UNSUPPORTED_MEDIA_TYPE:
            message = jsonify({'message': 'Unsupported Media Type / No JSON payload'})
        case http.HTTPStatus.OK:
            message = jsonify({'message': 'Success.'})
        case http.HTTPStatus.METHOD_NOT_ALLOWED:
            message = jsonify({'message': 'This API call does not exist.'})
        case _:
            message = jsonify({'message': 'Could not process request.'})
    return message, status_code.value


'''
valid, user_id = firebase_manager.verify_jwt_token(data['token']) 
TODO : replace 53 by the given statement.
'''


def authenticate_request(token: str):
    valid, user_id = firebase_manager.verify_jwt_token(token)
    if not valid:
        raise Exception("Invalid JWT token.")
    else:
        user = firebase_manager.get_user_from_id(user_id)
        if not user:
            raise Exception("User not found.")
        if not user.email_verified:
            raise Exception("Your google account isn't verified yet.")
    return user


def parse_and_validate_request(parameters: list[str], token: str) -> Union[list[str], None]:
    fn_args = []
    data = request.get_json()
    if not data:
        raise Exception("Empty request body.")
    user = authenticate_request(token)
    data['user'] = user
    for name in parameters:
        if name not in data:
            raise Exception(f"Missing parameter {name}")
        value = data[name]
        fn_args.append(value)
    return fn_args


route_handlers = {
    'SetSubdomain': generic_executor.set_subdomain,
    'FetchServers': generic_executor.fetch_servers,
    'FetchLogs': generic_executor.fetch_logs,
    'FetchHistory': generic_executor.fetch_history,
    'FetchPlayersStatus': generic_executor.fetch_players_status,
    'FetchDirContent': generic_executor.fetch_dir_content,
    'AccountCreate': generic_executor.account_create,
    'AccountDelete': generic_executor.account_delete,
    'ServerCreate': generic_executor.server_create,
    'ServerDelete': generic_executor.server_delete,
    'ServerRun': generic_executor.server_run,
    'ServerStop': generic_executor.server_stop,
    'UpdateProperties': generic_executor.update_properties,
    'Command': generic_executor.run_command,
}


@apiBP.route('/<path:path>', methods=['POST'])
def dynamic_route_handler(path):
    if path not in route_handlers:
        return generic_response_maker(http.HTTPStatus.METHOD_NOT_ALLOWED)

    route_fn = route_handlers[path]
    parameters = []
    sig = inspect.signature(route_fn)
    for param in sig.parameters.values():
        parameters.append(param.name)
    try:
        token: str or None = request.headers.get('SST')
        if not token:
            return generic_response_maker(http.HTTPStatus.BAD_REQUEST, "No provided token.")
        mapped_parameters = parse_and_validate_request(parameters, token)
        if mapped_parameters is None:
            return generic_response_maker(http.HTTPStatus.BAD_REQUEST)
        status, message = route_fn(*mapped_parameters)
        if isinstance(message, list):
            return jsonify(message), http.HTTPStatus.OK
        return generic_response_maker(status, message if message else None)
    except Exception as e:
        return generic_response_maker(http.HTTPStatus.BAD_REQUEST, str(e))


# [!] This route is specific and has to remain out of the dynamic route handler.
@apiBP.route('/Upload', methods=['POST'])
def upload():
    form = request.form
    token: str or None = request.headers.get('SST')
    name: str or None = form.get('name')
    files: ImmutableMultiDict[str, FileStorage] = request.files

    try:
        if not form:
            raise SyntaxError(0)
        if not files:
            raise SyntaxError(1)
        if not token:
            raise KeyError('token')
        if not name:
            raise KeyError('name')

        user: UserRecord = authenticate_request(token)
        user_id: str = user.uid

        for _, file in files.items():
            filename = file.filename
            internal_path = file_manager.get_path_from_extension(filename)
            if internal_path is None:
                continue
            filename = secure_filename(filename)
            file.save(os.path.join(f"users/{user_id}/{name}/{internal_path}", filename))

    except SyntaxError as e:
        match e:
            case 0:
                return generic_response_maker(http.HTTPStatus.BAD_REQUEST, "No FormData found in the payload.")
            case 1:
                return generic_response_maker(http.HTTPStatus.NOT_ACCEPTABLE, "No file(s) were uploaded.")

    except FileNotFoundError:
        return generic_response_maker(http.HTTPStatus.CONFLICT, "Please launch the server at least once.")

    except KeyError as e:
        return f"Missing parameter {e} in FormData.", http.HTTPStatus.BAD_REQUEST

    except UserNotFoundError as e:
        return str(e), http.HTTPStatus.BAD_REQUEST

    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return generic_response_maker(http.HTTPStatus.BAD_REQUEST, f"{type(e).__name__}, {str(e)}")

    return generic_response_maker(http.HTTPStatus.OK, "Successfully uploaded files !")


def api_cleanup() -> None:
    firebase_manager.set_servers_not_running()
    return


app.register_blueprint(apiBP)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Background Scheduler")
    (parser.add_argument
     ('--servers-interval', type=int, default=10, help="Interval to check for idle servers. (minutes)"))
    (parser.add_argument
     ('--modpacks-interval', type=int, default=48, help="Interval to check for modpack updates. (hours)"))
    args = parser.parse_args()

    scheduler = BackgroundScheduler()
    scheduler.add_job(generic_executor.scheduled_actions_short, 'interval', minutes=args.servers_interval)
    scheduler.add_job(generic_executor.scheduled_actions_long, 'interval', hours=args.modpacks_interval)
    scheduler.start()

    run_simple('0.0.0.0', 3000, app, use_debugger=False, use_reloader=False)
    #api_cleanup()