221 lines
8.2 KiB
Python
221 lines
8.2 KiB
Python
#
|
|
# Copyright © 2021,2022 IsardVDI S.L.
|
|
# Copyright © 2022 Evilham <contact@evilham.com>
|
|
#
|
|
# This file is part of DD
|
|
#
|
|
# DD is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or (at your
|
|
# option) any later version.
|
|
#
|
|
# DD is distributed in the hope that it will be useful, but WITHOUT ANY
|
|
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
|
|
# FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
|
|
# details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with DD. If not, see <https://www.gnu.org/licenses/>.
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
import logging as log
|
|
import os
|
|
import os.path
|
|
import secrets
|
|
import traceback
|
|
from typing import TYPE_CHECKING, Any, Callable, Dict
|
|
|
|
import yaml
|
|
from cerberus import Validator
|
|
from flask import Flask, Response, jsonify, render_template, send_from_directory
|
|
|
|
from admin.lib.api_exceptions import Error
|
|
from admin.views.decorators import OptionalJsonResponse
|
|
from admin.views.ApiViews import setup_api_views
|
|
from admin.views.AppViews import setup_app_views
|
|
from admin.views.LoginViews import setup_login_views
|
|
from admin.views.WebViews import setup_web_views
|
|
from admin.views.WpViews import setup_wp_views
|
|
from admin.auth.authentication import setup_auth
|
|
|
|
if TYPE_CHECKING:
|
|
from admin.lib.admin import Admin
|
|
from admin.lib.postup import Postup
|
|
|
|
|
|
class AdminValidator(Validator): # type: ignore # cerberus type hints MIA
|
|
# TODO: What's the point of this class?
|
|
None
|
|
# def _normalize_default_setter_genid(self, document):
|
|
# return _parse_string(document["name"])
|
|
|
|
# def _normalize_default_setter_genidlower(self, document):
|
|
# return _parse_string(document["name"]).lower()
|
|
|
|
# def _normalize_default_setter_gengroupid(self, document):
|
|
# return _parse_string(
|
|
# document["parent_category"] + "-" + document["uid"]
|
|
# ).lower()
|
|
|
|
|
|
class AdminFlaskApp(Flask):
|
|
"""
|
|
Subclass Flask app to ease customisation and type checking.
|
|
|
|
In order for an instance of this class to be useful,
|
|
the setup method should be called after instantiating.
|
|
"""
|
|
|
|
admin: "Admin"
|
|
data_dir: str
|
|
custom_dir: str
|
|
ready: bool = False
|
|
|
|
def __init__(self, *args: Any, **kwargs: Any):
|
|
super().__init__(*args, **kwargs)
|
|
self.url_map.strict_slashes = False
|
|
self._load_config()
|
|
# Minor setup tasks
|
|
self._load_validators()
|
|
self._setup_routes()
|
|
setup_api_views(self)
|
|
setup_app_views(self)
|
|
setup_login_views(self)
|
|
setup_web_views(self)
|
|
setup_wp_views(self)
|
|
setup_auth(self)
|
|
|
|
@property
|
|
def legal_path(self) -> str:
|
|
return os.path.join(self.root_path, "static/templates/pages/legal/")
|
|
|
|
@property
|
|
def avatars_path(self) -> str:
|
|
return os.path.join(self.custom_dir, "avatars/")
|
|
|
|
@property
|
|
def secrets_dir(self) -> str:
|
|
return os.path.join(self.data_dir, "secrets")
|
|
|
|
def setup(self) -> None:
|
|
"""
|
|
Perform setup tasks that might do network
|
|
"""
|
|
from admin.lib.postup import Postup
|
|
Postup(self)
|
|
# This must happen after Postup since it, e.g. fetches moodle secrets
|
|
from admin.lib.admin import Admin
|
|
self.admin = Admin(self)
|
|
|
|
def json_route(self, rule: str, **options: Any) -> Callable[..., OptionalJsonResponse]:
|
|
return self.route(rule, **options) # type: ignore # mypy issue #7187
|
|
|
|
def _load_validators(self, purge_unknown: bool = True) -> Dict[str, Validator]:
|
|
validators = {}
|
|
schema_path = os.path.join(self.root_path, "schemas")
|
|
for schema_filename in os.listdir(schema_path):
|
|
try:
|
|
with open(os.path.join(schema_path, schema_filename)) as file:
|
|
schema_yml = file.read()
|
|
schema = yaml.load(schema_yml, Loader=yaml.FullLoader)
|
|
validators[schema_filename.split(".")[0]] = AdminValidator(
|
|
schema, purge_unknown=purge_unknown
|
|
)
|
|
except IsADirectoryError:
|
|
None
|
|
return validators
|
|
|
|
def _load_config(self) -> None:
|
|
try:
|
|
self.data_dir = os.environ.get("DATA_FOLDER", ".")
|
|
self.custom_dir = os.environ.get("CUSTOM_FOLDER", ".")
|
|
# Handle secrets like Flask's session key
|
|
secret_key_file = os.path.join(self.secrets_dir, "secret_key")
|
|
if not os.path.exists(self.secrets_dir):
|
|
os.mkdir(self.secrets_dir, mode=0o700)
|
|
if not os.path.exists(secret_key_file):
|
|
# Generate as needed
|
|
# https://flask.palletsprojects.com/en/2.1.x/config/#SECRET_KEY
|
|
with os.fdopen(
|
|
os.open(secret_key_file, os.O_WRONLY | os.O_CREAT, 0o600), "w"
|
|
) as f:
|
|
f.write(secrets.token_hex())
|
|
self.secret_key = open(secret_key_file, "r").read()
|
|
|
|
# Move on with settings from the environment
|
|
self.config.update({
|
|
"DOMAIN": os.environ["DOMAIN"],
|
|
"KEYCLOAK_POSTGRES_USER": os.environ["KEYCLOAK_DB_USER"],
|
|
"KEYCLOAK_POSTGRES_PASSWORD": os.environ["KEYCLOAK_DB_PASSWORD"],
|
|
"MOODLE_POSTGRES_USER": os.environ["MOODLE_POSTGRES_USER"],
|
|
"MOODLE_POSTGRES_PASSWORD": os.environ["MOODLE_POSTGRES_PASSWORD"],
|
|
"NEXTCLOUD_POSTGRES_USER": os.environ["NEXTCLOUD_POSTGRES_USER"],
|
|
"NEXTCLOUD_POSTGRES_PASSWORD": os.environ["NEXTCLOUD_POSTGRES_PASSWORD"],
|
|
"VERIFY": os.environ["VERIFY"] == "true",
|
|
"API_SECRET": os.environ.get("API_SECRET"),
|
|
})
|
|
except Exception as e:
|
|
log.error(traceback.format_exc())
|
|
raise
|
|
|
|
def _setup_routes(self) -> None:
|
|
"""
|
|
Setup routes to Serve static files
|
|
"""
|
|
|
|
@self.route("/build/<path:path>")
|
|
def send_build(path: str) -> Response:
|
|
return send_from_directory(
|
|
os.path.join(self.root_path, "node_modules/gentelella/build"), path
|
|
)
|
|
|
|
@self.route("/vendors/<path:path>")
|
|
def send_vendors(path: str) -> Response:
|
|
return send_from_directory(
|
|
os.path.join(self.root_path, "node_modules/gentelella/vendors"), path
|
|
)
|
|
|
|
@self.route("/node_modules/<path:path>")
|
|
def send_nodes(path: str) -> Response:
|
|
return send_from_directory(
|
|
os.path.join(self.root_path, "node_modules"), path
|
|
)
|
|
|
|
@self.route("/templates/<path:path>")
|
|
def send_templates(path: str) -> Response:
|
|
return send_from_directory(os.path.join(self.root_path, "templates"), path)
|
|
|
|
# @self.route('/templates/<path:path>')
|
|
# def send_templates(path):
|
|
# return send_from_directory(os.path.join(self.root_path, 'static/templates'), path)
|
|
|
|
@self.route("/static/<path:path>")
|
|
def send_static_js(path: str) -> Response:
|
|
return send_from_directory(os.path.join(self.root_path, "static"), path)
|
|
|
|
@self.route("/avatars/<path:path>")
|
|
def send_avatars_img(path: str) -> Response:
|
|
return send_from_directory(
|
|
os.path.join(self.root_path, "../avatars/master-avatars"), path
|
|
)
|
|
|
|
@self.route("/custom/<path:path>")
|
|
def send_custom(path: str) -> Response:
|
|
return send_from_directory(self.custom_dir, path)
|
|
|
|
# @self.errorhandler(404)
|
|
# def not_found_error(error):
|
|
# return render_template('page_404.html'), 404
|
|
|
|
# @self.errorhandler(500)
|
|
# def internal_error(error):
|
|
# return render_template('page_500.html'), 500
|
|
|
|
@self.errorhandler(Error)
|
|
def handle_user_error(ex : Error) -> Response:
|
|
response : Response = jsonify(ex.error)
|
|
response.status_code = ex.status_code
|
|
response.headers.extend(ex.content_type) # type: ignore # werkzeug type hint MIA
|
|
return response
|