# -*- encoding: utf-8 -*- """ The application entrypoint. """ from aiohttp import web import base64 from cryptography import fernet import logging import logging.handlers import os import typing as T import click from aiohttp_session import setup, session_middleware from aiohttp_session.cookie_storage import EncryptedCookieStorage from bot_z.async_operator import AsyncOperator from api.rest import routes, add_static_routes from api.conf import read_conf def setup_log(level: str, syslog: bool) -> logging.Logger: alog = logging.getLogger("api") alog.setLevel(os.environ.get("BOTZ_LOGLEVEL", level)) h = logging.StreamHandler() f = logging.Formatter("%(levelname)s [%(name)s] -> %(message)s") h.setFormatter(f) alog.addHandler(h) if syslog: sh = logging.handlers.SysLogHandler() alog.addHandler(sh) return alog def init_secret() -> bytes: fernet_key = fernet.Fernet.generate_key() return base64.urlsafe_b64decode(fernet_key) def setup_session(app: web.Application, secure: bool): secret = init_secret() setup( app, EncryptedCookieStorage( secret_key=secret, cookie_name="BOTZ_SESSION", httponly=False, secure=secure ), ) def run( address: T.Optional[T.Text], port: T.Optional[int], conf_path: T.Optional[T.Text] ) -> None: """Application entrypoint.""" conf = read_conf(conf_path) # This closure is needed to intercept the to-be-prepared response # and add the right CORS headers async def on_prepare_cors(request, response): response.headers["Access-Control-Allow-Origin"] = conf["http"].get("cors_allow") response.headers["Access-Control-Allow-Credentials"] = "true" alog = setup_log(conf["log"]["level"], conf["log"]["syslog"]) alog.debug("conf %s", conf) app = web.Application(logger=alog) app["base_uri"] = conf["base_uri"] app["debug"] = conf["debug"] app["headless"] = conf["headless"] if conf["http"].get("cors_allow"): app.on_response_prepare.append(on_prepare_cors) setup_session(app, conf["http"]["cookie_secure"]) add_static_routes(alog) app.add_routes(routes) addr = [] if address is not None: addr = [address] elif conf["http"].get("bind_addr"): addr.extend(conf["http"]["bind_addr"]) else: addr = ["127.0.0.1"] if port is None: port = conf["http"]["port"] alog.debug("Starting app with: addr -> %s, port -> %d", addr, port) web.run_app(app, host=addr, port=port) @click.command() @click.option( "-a", "--address", type=click.STRING, help="Address to bind the server to." ) @click.option("-p", "--port", type=click.INT, help="Port to bind to.") @click.option( "-c", "--conf", type=click.Path(exists=False), help="A path to a configuration file.", ) def cli( address: T.Optional[T.Text], port: T.Optional[int] = None, conf: T.Optional[T.Text] = None, ) -> None: run(address, port, conf)