aswi2021jmsd-gitlab/app.py @ 03484d49

import os

from flask import Flask, redirect, request
from flask_injector import FlaskInjector
from injector import Injector
from werkzeug.exceptions import HTTPException

from src.config import configuration
from src.config.configuration import Configuration
from src.config.connection_provider import ConnectionProvider
from src.controllers.certificates_controller import CertController
from src.controllers.crl_ocsp_controller import CrlOcspController
from src.controllers.exception_handlers import handle_cryptography_exception, handle_database_exception, \
    handle_generic_exception
from src.exceptions.database_exception import DatabaseException
from src.services.cryptography import CryptographyService, CryptographyException
from src.utils.logger import Logger

app = Flask(__name__)


@app.route('/')
def index():
    return redirect("/static/index.html")


@app.route('/api/certificates', methods=["POST"])
def create_certificate(certificate_controller: CertController):
    return certificate_controller.create_certificate()


@app.route('/api/certificates', methods=["GET"])
def get_cert_list(certificate_controller: CertController):
    return certificate_controller.get_certificate_list()


@app.route('/api/certificates/<id>', methods=["GET"])
def get_cert(id, certificate_controller: CertController):
    return certificate_controller.get_certificate_by_id(id)


@app.route('/api/certificates/<id>', methods=["PATCH"])
def set_certificate_status(id, certificate_controller: CertController):
    return certificate_controller.set_certificate_status(id)


@app.route('/api/certificates/<id>', methods=["DELETE"])
def delete_certificate(id, certificate_controller: CertController):
    return certificate_controller.delete_certificate(id)


@app.route('/api/certificates/<id>/details', methods=["GET"])
def get_cert_details(id, certificate_controller: CertController):
    return certificate_controller.get_certificate_details_by_id(id)


@app.route('/api/certificates/<id>/root', methods=["GET"])
def get_cert_root(id, certificate_controller: CertController):
    return certificate_controller.get_certificate_root_by_id(id)


@app.route('/api/certificates/<id>/chain', methods=["GET"])
def get_cert_chain(id, certificate_controller: CertController):
    return certificate_controller.get_certificate_trust_chain_by_id(id)


@app.route('/api/certificates/<id>/privatekey', methods=["GET"])
def get_private_key_of_a_certificate(id, certificate_controller: CertController):
    return certificate_controller.get_private_key_of_a_certificate(id)


@app.route('/api/certificates/<id>/publickey', methods=["GET"])
def get_public_key_of_a_certificate(id, certificate_controller: CertController):
    return certificate_controller.get_public_key_of_a_certificate(id)


@app.route('/api/crl/<id>', methods=["GET"])
def get_crl_of_issuer(id, crl_ocsp_controller: CrlOcspController):
    return crl_ocsp_controller.get_crl(id)


@app.route('/api/ocsp/<id>/<path:ocsp_request>', methods=["GET"])
def get_ocsp_of_issuer_get(id, ocsp_request, crl_ocsp_controller: CrlOcspController):
    return crl_ocsp_controller.get_ocsp_from_base64(id, ocsp_request)


@app.route('/api/ocsp/<id>', methods=["POST"])
def get_ocsp_of_issuer_post(id, crl_ocsp_controller: CrlOcspController):
    return crl_ocsp_controller.get_ocsp_from_der(id, request.data)


@app.route('/api/certificates/<id>/identity', methods=["POST"])
def generate_certificate_pkcs_identity(id, certificate_controller: CertController):
    return certificate_controller.generate_certificate_pkcs_identity(id)


@app.errorhandler(CryptographyException)
def cryptography_error(e):
    return handle_cryptography_exception(e)


@app.errorhandler(DatabaseException)
def database_error(e):
    return handle_database_exception(e)


@app.errorhandler(Exception)
def generic_exception(e):
    if isinstance(e, HTTPException):
        # handle HTTPException exceptions here (MethodNotAllowed for example)
        Logger.warning(f""" HTTPException occurred: "{str(e)}" """)
        return str(e), e.code
    return handle_generic_exception(e)


def initialize_app(application) -> bool:
    """
    Initializes the application
        -   configure dependency injection
        -   check whether OpenSSL is on the system
    :param application Flask Application to be initialized.
    :return: boolean flag indicating whether initialization was successful or not
    """
    modules = [configuration.configure_env_variable, ConnectionProvider]
    injector = Injector(modules)
    FlaskInjector(app=application, modules=modules)

    config = injector.get(Configuration)
    configuration.configure_logging(config)

    Logger.info(f"Using configuration file: {config.config_file}")

    # There's a little dependency on the CryptoService, which is not a pretty thing from
    # architectural point of view. However it is only a minimal piece of code and
    # it makes sense to do it in this way instead of trying to run openssl via subprocess here
    cryptography_service = injector.get(CryptographyService)
    try:
        # if version string is returned, OpenSSL is present on the system
        Logger.info(f"Using {cryptography_service.get_openssl_version()}")
        return True
    except CryptographyException:
        # If getting the version string throws an exception the OpenSSL is not available
        print("OpenSSL was not located on the system. Application will now exit.")
        Logger.error(f"OpenSSL was not located on the system. Application will now exit.")
        return False


# app initialization must follow endpoint declaration (after all Flask decoration)
with app.app_context():
    if not initialize_app(app):
        print("Failed to initialize app, aborting...")
        Logger.error(f"Failed to initialize app, aborting...")
        exit(-1)

if __name__ == '__main__':
    app_host = "0.0.0.0"
    app_port = 5000

    # TODO better load this from config.py
    if "FLASK_HOST" in os.environ:
        app_host = os.environ["FLASK_HOST"]

    if "FLASK_PORT" in os.environ:
        app_host = os.environ["FLASK_PORT"]

    app.run(host=app_host, port=app_port)