oplik0/cherrydoor

View on GitHub
cherrydoor/app.py

Summary

Maintainability
A
1 hr
Test Coverage
"""Everything connected to setup of the app."""

__author__ = "opliko"
__license__ = "MIT"
__version__ = "0.8.b0"
__status__ = "Prototype"

import asyncio
import base64
import os
from hashlib import sha256
from typing import Any, Dict

import aiohttp_csrf
import sentry_sdk
from aiohttp import web
from aiohttp_jinja2 import get_env
from aiohttp_jinja2 import setup as setup_jinja2
from aiohttp_rest_api.loader import (
    get_openapi_documentation,
    load_and_connect_all_endpoints_from_folder,
)
from aiohttp_rest_api.redoc import setup_redoc
from aiohttp_security import setup as setup_security
from aiohttp_session import setup as setup_session
from aiohttp_session_mongo import MongoStorage
from jinja2 import PackageLoader, contextfunction
from sentry_sdk.integrations.aiohttp import AioHttpIntegration

from cherrydoor.__version__ import __version__
from cherrydoor.api import (
    openapi_contact,
    openapi_description,
    openapi_overrides,
    redoc_options,
    redoc_routes,
)
from cherrydoor.api_tokens import ApiTokens
from cherrydoor.auth import AuthorizationPolicy, SessionIdentityPolicy
from cherrydoor.config import load_config
from cherrydoor.database import init_db, setup_db
from cherrydoor.secure import set_secure_headers
from cherrydoor.secure import setup as secure_setup
from cherrydoor.views import routes as views
from cherrydoor.socketio import sio, setup_socket_tasks

CSRF_FIELD_NAME = "_csrf_token"
CSRF_SESSION_NAME = "csrf_token"
CSRF_HEADER_NAME = "csrf_token"


def setup_app(loop=asyncio.get_event_loop(), config=load_config()[0]):
    """Create the app and initiate all services (database, security, etc).

    Parameters
    ----------
    loop : asyncio.EventLoop
        The event loop to use.
    config : AttrDict
        a dictionary with configuration values
    Returns
    -------
    app : web.Application
        The application instance.
    """
    if config.get("sentry_dsn", None):
        sentry_sdk.init(
            dsn=config["sentry_dsn"],
            integrations=[AioHttpIntegration()],
            release=f"cherrydoor@{__version__}",
        )
    # create app
    app = web.Application(loop=loop)
    # make config accessible through the app
    app["config"] = config
    # setup database and add it to the app
    db = init_db(config, loop)
    app["db"] = db
    # create a token generator/validator and add make it accessible through the app
    api_tokens = ApiTokens(app, config.get("secret_key", ""))
    app["api_tokens"] = api_tokens

    app.on_startup.append(setup_db)
    # set up aiohttp-session with aiohttp-session-mongo for storage
    setup_session(
        app,
        MongoStorage(
            db["sessions"],
            max_age=None,
            cookie_name="session_id",
        ),
    )
    # set up aiohttp-security
    setup_security(
        app,
        SessionIdentityPolicy("uid", config.get("max_session_age", 31536000)),
        AuthorizationPolicy(app),
    )
    # set up secure.py
    secure_setup(app)
    app.middlewares.append(set_secure_headers)

    csrf_policy = aiohttp_csrf.policy.FormAndHeaderPolicy(
        CSRF_HEADER_NAME, CSRF_FIELD_NAME
    )
    csrf_storage = aiohttp_csrf.storage.SessionStorage(CSRF_SESSION_NAME)
    aiohttp_csrf.setup(app, policy=csrf_policy, storage=csrf_storage)
    # app.middlewares.append(aiohttp_csrf.csrf_middleware)

    load_and_connect_all_endpoints_from_folder(
        path=f"{os.path.dirname(os.path.realpath(__file__))}/api",
        app=app,
        version_prefix="api/v1",
    )
    redoc_url = "/api/v1/docs"
    setup_redoc(
        app,
        redoc_url=redoc_url,
        description=openapi_description,
        title="Cherrydoor API",
        page_title="Cherrydocs",
        openapi_info=get_openapi_documentation(overrides=openapi_overrides),
        redoc_options=redoc_options,
        contact=openapi_contact,
    )
    app["redoc_url"] = redoc_url
    app.router.add_routes(redoc_routes)
    setup_static_routes(app)

    jinja2_loader = PackageLoader("cherrydoor", "templates")
    setup_jinja2(app, loader=jinja2_loader, auto_reload=True)
    get_env(app).globals["sri"] = sri_for
    get_env(app).globals["csrf_field_name"] = CSRF_FIELD_NAME
    get_env(app).filters["vue"] = vue
    setup_routes(app)
    sio.attach(app)
    app.on_startup.append(setup_socket_tasks)

    return app


def setup_static_routes(app):
    """Set the mapping of system static items path to /static url.

    Parameters
    ----------
    app : web.Application
        The application instance.
    """
    app.router.add_static(
        "/static/",
        path=f"{os.path.dirname(os.path.realpath(__file__))}/static",
        name="static",
        append_version=True,
        follow_symlinks=True,
    )
    app["static_root_url"] = "/static"


def setup_routes(app):
    """Set all the routes for the app using RouteTableDef.

    Notes
    -----
    Currently imported files with routes:
        - views.py

    Parameters
    ----------
    app : web.Application
        The application instance.
    """
    app.add_routes(views)


@contextfunction
def sri_for(
    context: Dict[str, Any], static_file_path: str  # pylint: disable=unused-argument
) -> str:
    """
    Return a hash in the format used by the SRI HTML attribute.

    Parameters:
        context (dict): The template context.
        static_file_path (str): The path to the static file.
    Returns:
        sri (str): The SRI hash in sha256 format.
    """
    input_file = (
        f"{os.path.dirname(os.path.realpath(__file__))}/static/{static_file_path}"
    )
    sha256_hasher = sha256()
    with open(input_file, "rb") as f:
        while True:
            data = f.read(65536)
            if not data:
                break
            sha256_hasher.update(data)
    sha256_hash = sha256_hasher.digest()
    hash_base64 = base64.b64encode(sha256_hash).decode()
    return f"sha256-{hash_base64}"


def vue(item):
    """Filter out vue templates.

    For example: {{ "message.text" | vue }} will be transformed to just {{ "message.text" }} in HTML

    Parameters:
        item (str): The text to filter.
    Returns:
        item (str): Text that jinja2 will render properly.
    """
    return f"{{{{ {item} }}}}"