import atexit import http import inspect import os from mailbox import FormatError 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.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 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(data: dict): if 'token' not in data: raise Exception("Missing 'token' in request body. The API doesn't support anonymous access anymore.") else: valid, user_id = True, data['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]) -> Union[list[str], None]: args = [] data = request.get_json() if not data: raise Exception("Empty request body.") user = authenticate_request(data) data['user'] = user for name in parameters: if name not in data: raise Exception(f"Missing parameter {name}") value = data[name] args.append(value) return 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, 'AccountCreate': generic_executor.account_create, 'ServerCreate': generic_executor.server_create, 'ServerDelete': generic_executor.server_delete, 'AccountDelete': generic_executor.account_delete, 'ServerRun': generic_executor.server_run, 'ServerStop': generic_executor.server_stop, 'UpdateProperties': generic_executor.update_properties, 'Command': generic_executor.run_command, } @apiBP.route('/', 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: mapped_parameters = parse_and_validate_request(parameters) 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 = form.get('token') name: str or None = form.get('name') files: ImmutableMultiDict[str, FileStorage] = request.files try: if not form: raise FormatError(0) if not files: raise FormatError(1) if not token: raise KeyError('token') if not name: raise KeyError('name') data: dict = {'token': token} user: UserRecord = authenticate_request(data) 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 FormatError 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 exit_safety() -> None: firebase_manager.set_servers_not_running() return app.register_blueprint(apiBP) if __name__ == '__main__': atexit.register(exit_safety) scheduler = BackgroundScheduler() scheduler.add_job(generic_executor.scheduled_actions, 'interval', minutes=10) scheduler.start() app.run(host='0.0.0.0', port=3000, debug=False)