def create_app()

in airflow-core/src/airflow/utils/serve_logs.py [0:0]

• 91 lines of code
• 15 McCabe index (conditional complexity)

def create_app():
    flask_app = Flask(__name__, static_folder=None)
    leeway = conf.getint("webserver", "log_request_clock_grace", fallback=30)
    log_directory = os.path.expanduser(conf.get("logging", "BASE_LOG_FOLDER"))
    log_config_class = conf.get("logging", "logging_config_class")
    if log_config_class:
        logger.info("Detected user-defined logging config. Attempting to load %s", log_config_class)
        try:
            logging_config = import_string(log_config_class)
            try:
                base_log_folder = logging_config["handlers"]["task"]["base_log_folder"]
            except KeyError:
                base_log_folder = None
            if base_log_folder is not None:
                log_directory = base_log_folder
                logger.info(
                    "Successfully imported user-defined logging config. Flask App will serve log from %s",
                    log_directory,
                )
            else:
                logger.warning(
                    "User-defined logging config does not specify 'base_log_folder'. "
                    "Flask App will use default log directory %s",
                    base_log_folder,
                )
        except Exception as e:
            raise ImportError(f"Unable to load {log_config_class} due to error: {e}")
    signer = JWTValidator(
        issuer=None,
        secret_key=get_signing_key("webserver", "secret_key"),
        algorithm="HS512",
        leeway=leeway,
        audience="task-instance-logs",
    )

    # Prevent direct access to the logs port
    @flask_app.before_request
    def validate_pre_signed_url():
        try:
            auth = request.headers.get("Authorization")
            if auth is None:
                logger.warning("The Authorization header is missing: %s.", request.headers)
                abort(403)
            payload = signer.validated_claims(auth)
            token_filename = payload.get("filename")
            request_filename = request.view_args["filename"]
            if token_filename is None:
                logger.warning("The payload does not contain 'filename' key: %s.", payload)
                abort(403)
            if token_filename != request_filename:
                logger.warning(
                    "The payload log_relative_path key is different than the one in token:"
                    "Request path: %s. Token path: %s.",
                    request_filename,
                    token_filename,
                )
                abort(403)
        except HTTPException:
            raise
        except InvalidAudienceError:
            logger.warning("Invalid audience for the request", exc_info=True)
            abort(403)
        except InvalidSignatureError:
            logger.warning("The signature of the request was wrong", exc_info=True)
            abort(403)
        except ImmatureSignatureError:
            logger.warning("The signature of the request was sent from the future", exc_info=True)
            abort(403)
        except ExpiredSignatureError:
            logger.warning(
                "The signature of the request has expired. Make sure that all components "
                "in your system have synchronized clocks. "
                "See more at %s",
                get_docs_url("configurations-ref.html#secret-key"),
                exc_info=True,
            )
            abort(403)
        except InvalidIssuedAtError:
            logger.warning(
                "The request was issues in the future. Make sure that all components "
                "in your system have synchronized clocks. "
                "See more at %s",
                get_docs_url("configurations-ref.html#secret-key"),
                exc_info=True,
            )
            abort(403)
        except Exception:
            logger.warning("Unknown error", exc_info=True)
            abort(403)

    @flask_app.route("/log/<path:filename>")
    def serve_logs_view(filename):
        return send_from_directory(log_directory, filename, mimetype="application/json", as_attachment=False)

    return flask_app