# -*- coding: UTF-8 -* # pylint: disable=invalid-name import base64 import datetime import json import os import socket import sys import time import traceback import logging from logging.handlers import SMTPHandler, WatchedFileHandler from threading import Thread import warnings from flask import current_app, g, request from flask import Flask from flask import abort, flash, has_request_context from flask import render_template from flask.logging import default_handler from flask_caching import Cache from flask_json import FlaskJSON, json_response from flask_login import LoginManager, current_user from flask_mail import Mail from flask_migrate import Migrate from flask_sqlalchemy import SQLAlchemy from jinja2 import select_autoescape import sqlalchemy as sa import werkzeug.debug from wtforms.fields import HiddenField from flask_cas import CAS from app.scodoc.sco_exceptions import ( AccessDenied, ScoBugCatcher, ScoException, ScoGenError, ScoInvalidCSRF, ScoPDFFormatError, ScoValueError, APIInvalidParams, ) from app.scodoc.sco_vdi import ApoEtapeVDI from config import DevConfig import sco_version db = SQLAlchemy() migrate = Migrate(compare_type=True) login = LoginManager() login.login_view = "auth.login" login.login_message = "Identifiez-vous pour accéder à cette page." mail = Mail() CACHE_TYPE = os.environ.get("CACHE_TYPE") cache = Cache( config={ # see https://flask-caching.readthedocs.io/en/latest/index.html#configuring-flask-caching "CACHE_TYPE": CACHE_TYPE or "RedisCache", # by default, never expire: "CACHE_DEFAULT_TIMEOUT": os.environ.get("CACHE_DEFAULT_TIMEOUT") or 0, } ) def handle_sco_value_error(exc): "page d'erreur avec message" return render_template("sco_value_error.j2", exc=exc), 404 def handle_access_denied(exc): return render_template("error_access_denied.j2", exc=exc), 403 def handle_invalid_csrf(exc): """Form submit with invalid CSRF token""" # logout user and go back to login page with an error message from app import auth auth.logic.logout() return render_template("error_csrf.j2", exc=exc), 404 # def handle_pdf_format_error(exc): # return "ay ay ay" handle_pdf_format_error = handle_sco_value_error def internal_server_error(exc): """Bugs scodoc, erreurs 500""" # note that we set the 500 status explicitly from app.scodoc import sco_utils as scu # Invalide tous les caches log("internal_server_error: clearing caches") clear_scodoc_cache() return ( render_template( "error_500.j2", SCOVERSION=sco_version.SCOVERSION, date=datetime.datetime.now().isoformat(), exc=exc, traceback_str_base64=base64.urlsafe_b64encode( traceback.format_exc().encode(scu.SCO_ENCODING) ).decode(scu.SCO_ENCODING), request_url=request.url, scu=scu, ), 500, ) def handle_sco_bug(exc): """Un bug, en général rare, sur lequel les dev cherchent des informations pour le corriger. """ if current_app.config["TESTING"] or current_app.config["DEBUG"]: raise ScoException # for development servers only else: Thread( target=_async_dump, args=(current_app._get_current_object(), request.url) ).start() return internal_server_error(exc) def _async_dump(app, request_url: str): from app.scodoc.sco_dump_db import sco_dump_and_send_db with app.app_context(): ndb.open_db_connection() try: sco_dump_and_send_db("ScoBugCatcher", request_url=request_url) except ScoValueError: pass def handle_invalid_usage(error): response = json_response(data_=error.to_dict()) response.status_code = error.status_code return response # JSON ENCODING # used by some internal finctions # the API is now using flask_json, NOT THIS ENCODER class ScoDocJSONEncoder(json.JSONEncoder): def default(self, o): # pylint: disable=E0202 if isinstance(o, (datetime.date, datetime.datetime)): return o.isoformat() elif isinstance(o, ApoEtapeVDI): return str(o) else: return json.JSONEncoder.default(self, o) def render_raw_html(template_filename: str, **args) -> str: """Load and render an HTML file _without_ using Flask Necessary for 503 error message, when DB is down and Flask may be broken. """ template_path = os.path.join( current_app.config["SCODOC_DIR"], "app", current_app.template_folder, template_filename, ) with open(template_path) as f: txt = f.read().format(**args) return txt def postgresql_server_error(e): """Erreur de connection au serveur postgresql (voir notesdb.open_db_connection)""" return render_raw_html("error_503.j2", SCOVERSION=sco_version.SCOVERSION), 503 class LogRequestFormatter(logging.Formatter): """Ajoute URL et remote_addr for logging""" def format(self, record): if has_request_context(): record.url = request.url record.remote_addr = request.remote_addr else: record.url = None record.remote_addr = None record.sco_user = current_user if has_request_context(): record.sco_admin_mail = current_app.config["SCODOC_ADMIN_MAIL"] else: record.sco_admin_mail = "(pas de requête)" return super().format(record) class LogExceptionFormatter(logging.Formatter): """Formatteur pour les exceptions: ajoute détails""" def format(self, record): if has_request_context(): record.url = request.url record.remote_addr = request.environ.get( "HTTP_X_FORWARDED_FOR", request.remote_addr ) record.http_referrer = request.referrer record.http_method = request.method if request.method == "GET": record.http_params = str(request.args) else: # rep = reprlib.Repr() # abbrège record.http_params = str(request.form)[:2048] else: record.url = None record.remote_addr = None record.http_referrer = None record.http_method = None record.http_params = None record.sco_user = current_user if has_request_context(): record.sco_admin_mail = current_app.config["SCODOC_ADMIN_MAIL"] else: record.sco_admin_mail = "(pas de requête)" return super().format(record) class ScoSMTPHandler(SMTPHandler): def getSubject(self, record: logging.LogRecord) -> str: stack_summary = traceback.extract_tb(record.exc_info[2]) frame_summary = stack_summary[-1] subject = f"ScoExc({sco_version.SCOVERSION}): {record.exc_info[0].__name__} in {frame_summary.name} {frame_summary.filename}" return subject class ReverseProxied(object): """Adaptateur wsgi qui nous permet d'avoir toutes les URL calculées en https sauf quand on est en dev. La variable HTTP_X_FORWARDED_PROTO est positionnée par notre config nginx""" def __init__(self, app): self.app = app def __call__(self, environ, start_response): scheme = environ.get("HTTP_X_FORWARDED_PROTO") if scheme: environ["wsgi.url_scheme"] = scheme # ou forcer à https ici ? return self.app(environ, start_response) def create_app(config_class=DevConfig): app = Flask(__name__, static_url_path="/ScoDoc/static", static_folder="static") app.config.from_object(config_class) from app.auth import cas CAS(app, url_prefix="/cas", configuration_function=cas.set_cas_configuration) app.wsgi_app = ReverseProxied(app.wsgi_app) app_json = FlaskJSON(app) @app_json.encoder def scodoc_json_encoder(o): "Overide default date encoding (RFC 822) and use ISO" if isinstance(o, (datetime.date, datetime.datetime)): return o.isoformat() # Pour conserver l'ordre des objets dans les JSON: # e.g. l'ordre des UE dans les bulletins app.json.sort_keys = False # Evite de logguer toutes les requetes dans notre log logging.getLogger("werkzeug").disabled = True app.logger.setLevel(app.config["LOG_LEVEL"]) if app.config["TESTING"] or app.config["DEBUG"]: # S'arrête sur tous les warnings, sauf # flask_sqlalchemy/query (pb deprecation du model.get()) warnings.filterwarnings("error", module="flask_sqlalchemy/query") # warnings.filterwarnings("ignore", module="json/provider.py") xxx sans effet en test if app.config["DEBUG"]: # comme on a désactivé ci-dessus les logs de werkzeug, # on affiche nous même le PIN en mode debug: print( f""" * Debugger is active! * Debugger PIN: {werkzeug.debug.get_pin_and_cookie_name(app)[0]} """ ) # Vérifie/crée lien sym pour les URL statiques link_filename = f"{app.root_path}/static/links/{sco_version.SCOVERSION}" if not os.path.exists(link_filename): app.logger.info(f"creating symlink {link_filename}") os.symlink("..", link_filename) db.init_app(app) migrate.init_app(app, db) login.init_app(app) mail.init_app(app) app.extensions["mail"].debug = 0 # disable copy of mails to stderr cache.init_app(app) sco_cache.CACHE = cache if CACHE_TYPE: # non default app.logger.info(f"CACHE_TYPE={CACHE_TYPE}") app.register_error_handler(ScoGenError, handle_sco_value_error) app.register_error_handler(ScoValueError, handle_sco_value_error) app.register_error_handler(ScoBugCatcher, handle_sco_bug) app.register_error_handler(ScoInvalidCSRF, handle_invalid_csrf) app.register_error_handler(ScoPDFFormatError, handle_pdf_format_error) app.register_error_handler(AccessDenied, handle_access_denied) app.register_error_handler(500, internal_server_error) app.register_error_handler(503, postgresql_server_error) app.register_error_handler(APIInvalidParams, handle_invalid_usage) # Add some globals # previously in Flask-Bootstrap: app.jinja_env.globals["bootstrap_is_hidden_field"] = lambda field: isinstance( field, HiddenField ) from app.auth import bp as auth_bp app.register_blueprint(auth_bp, url_prefix="/auth") from app.entreprises import bp as entreprises_bp app.register_blueprint(entreprises_bp, url_prefix="/ScoDoc/entreprises") from app.views import scodoc_bp from app.views import scolar_bp from app.views import notes_bp from app.views import users_bp from app.views import absences_bp from app.views import assiduites_bp from app.api import api_bp from app.api import api_web_bp # Enable autoescaping of all templates, including .j2 app.jinja_env.autoescape = select_autoescape(default_for_string=True, default=True) # https://scodoc.fr/ScoDoc app.register_blueprint(scodoc_bp) # https://scodoc.fr/ScoDoc/RT/Scolarite/... app.register_blueprint(scolar_bp, url_prefix="/ScoDoc/<scodoc_dept>/Scolarite") # https://scodoc.fr/ScoDoc/RT/Scolarite/Notes/... app.register_blueprint(notes_bp, url_prefix="/ScoDoc/<scodoc_dept>/Scolarite/Notes") # https://scodoc.fr/ScoDoc/RT/Scolarite/Users/... app.register_blueprint(users_bp, url_prefix="/ScoDoc/<scodoc_dept>/Scolarite/Users") # https://scodoc.fr/ScoDoc/RT/Scolarite/Absences/... app.register_blueprint( absences_bp, url_prefix="/ScoDoc/<scodoc_dept>/Scolarite/Absences" ) app.register_blueprint( assiduites_bp, url_prefix="/ScoDoc/<scodoc_dept>/Scolarite/Assiduites" ) app.register_blueprint(api_bp, url_prefix="/ScoDoc/api") app.register_blueprint(api_web_bp, url_prefix="/ScoDoc/<scodoc_dept>/api") scodoc_log_formatter = LogRequestFormatter( "[%(asctime)s] %(sco_user)s@%(remote_addr)s requested %(url)s\n" "%(levelname)s: %(message)s" ) # les champs additionnels sont définis dans LogRequestFormatter scodoc_exc_formatter = LogExceptionFormatter( "[%(asctime)s] %(sco_user)s@%(remote_addr)s requested %(url)s\n" "%(levelname)s: %(message)s\n" "Referrer: %(http_referrer)s\n" "Method: %(http_method)s\n" "Params: %(http_params)s\n" "Admin mail: %(sco_admin_mail)s\n" ) if not app.testing: if not app.debug: # --- Config logs pour PRODUCTION # On supprime le logguer par défaut qui va vers stderr et pollue les logs systemes app.logger.removeHandler(default_handler) # --- Mail des messages ERROR et CRITICAL if app.config["MAIL_SERVER"]: auth = None if app.config["MAIL_USERNAME"] or app.config["MAIL_PASSWORD"]: auth = (app.config["MAIL_USERNAME"], app.config["MAIL_PASSWORD"]) secure = None if app.config["MAIL_USE_TLS"]: secure = () host_name = socket.gethostname() mail_handler = ScoSMTPHandler( mailhost=(app.config["MAIL_SERVER"], app.config["MAIL_PORT"]), fromaddr=app.config["SCODOC_MAIL_FROM"], toaddrs=["exception@scodoc.org"], subject="ScoDoc Exception", # unused see ScoSMTPHandler credentials=auth, secure=secure, ) mail_handler.setFormatter(scodoc_exc_formatter) mail_handler.setLevel(logging.ERROR) app.logger.addHandler(mail_handler) else: # Pour logs en DEV uniquement: default_handler.setFormatter(scodoc_log_formatter) # Config logs pour DEV et PRODUCTION # Configuration des logs (actifs aussi en mode development) # usually /opt/scodoc-data/log/scodoc.log: # rotated by logrotate file_handler = WatchedFileHandler( app.config["SCODOC_LOG_FILE"], encoding="utf-8" ) file_handler.setFormatter(scodoc_log_formatter) file_handler.setLevel(logging.INFO) app.logger.addHandler(file_handler) # Log pour les erreurs (exceptions) uniquement: # usually /opt/scodoc-data/log/scodoc_exc.log file_handler = WatchedFileHandler( app.config["SCODOC_ERR_FILE"], encoding="utf-8" ) file_handler.setFormatter(scodoc_exc_formatter) file_handler.setLevel(logging.ERROR) app.logger.addHandler(file_handler) # app.logger.setLevel(logging.INFO) app.logger.info(f"{sco_version.SCONAME} {sco_version.SCOVERSION} startup") app.logger.info( f"create_app({config_class.__name__}, {config_class.SQLALCHEMY_DATABASE_URI})" ) # ---- INITIALISATION SPECIFIQUES A SCODOC from app.scodoc import sco_bulletins_generator from app.scodoc.sco_bulletins_legacy import BulletinGeneratorLegacy from app.scodoc.sco_bulletins_standard import BulletinGeneratorStandard from app.but.bulletin_but_pdf import BulletinGeneratorStandardBUT from app.scodoc.sco_bulletins_ucac import BulletinGeneratorUCAC # l'ordre est important, le premier sera le "défaut" pour les nouveaux départements. sco_bulletins_generator.register_bulletin_class(BulletinGeneratorStandard) sco_bulletins_generator.register_bulletin_class(BulletinGeneratorStandardBUT) sco_bulletins_generator.register_bulletin_class(BulletinGeneratorLegacy) sco_bulletins_generator.register_bulletin_class(BulletinGeneratorUCAC) if app.testing or app.debug: from app.scodoc.sco_bulletins_example import BulletinGeneratorExample sco_bulletins_generator.register_bulletin_class(BulletinGeneratorExample) from app.auth.cas import set_cas_configuration with app.app_context(): try: set_cas_configuration(app) except sa.exc.ProgrammingError: # Si la base n'a pas été upgradée (arrive durrant l'install) # il se peut que la table scodoc_site_config n'existe pas encore. pass return app def set_sco_dept(scodoc_dept: str, open_cnx=True): """Set global g object to given dept and open db connection if needed""" # Check that dept exists try: dept = Departement.query.filter_by(acronym=scodoc_dept).first() except sa.exc.OperationalError: abort(503) if not dept: raise ScoValueError(f"Invalid dept: {scodoc_dept}") g.scodoc_dept = scodoc_dept # l'acronyme g.scodoc_dept_id = dept.id # l'id if open_cnx and not hasattr(g, "db_conn"): ndb.open_db_connection() if not hasattr(g, "stored_get_formsemestre"): g.stored_get_formsemestre = {} def user_db_init(): """Initialize the users database. Check that basic roles and admin user exist. """ from app.auth.models import User, Role current_app.logger.info("Init User's db") # Create roles: Role.reset_standard_roles_permissions() current_app.logger.info("created initial roles") # Ensure that admin exists admin_mail = current_app.config.get("SCODOC_ADMIN_MAIL") if admin_mail: admin_user_name = current_app.config["SCODOC_ADMIN_LOGIN"] user = User.query.filter_by(user_name=admin_user_name).first() if not user: user = User(user_name=admin_user_name, email=admin_mail) try: db.session.add(user) db.session.commit() except: db.session.rollback() raise current_app.logger.info( "created initial admin user, login: {u.user_name}, email: {u.email}".format( u=user ) ) def sco_db_insert_constants(): """Initialize Sco database: insert some constants (modalités, ...).""" from app import models current_app.logger.info("Init Sco db") # Modalités: models.FormationModalite.insert_modalites() def initialize_scodoc_database(erase=False, create_all=False): """Initialize the database for unit tests Starts from an existing database and create all necessary SQL tables and functions. If erase is True, _erase_ all database content. """ # - ERASE (the truncation sql function has been defined above) if erase: truncate_database() # - Create all tables if create_all: # managed by migrations, except for TESTS db.create_all() # - Insert initial roles and create super-admin user user_db_init() # - Insert some constant values (modalites, ...) sco_db_insert_constants() # - Flush cache clear_scodoc_cache() def truncate_database(): """Erase content of all tables (including users !) from the current database. """ # use a stored SQL function, see createtables.sql try: db.session.execute(sa.text("SELECT truncate_tables('scodoc');")) db.session.commit() except: db.session.rollback() raise # Remet les compteurs (séquences sql) à zéro db.session.execute( sa.text( """ CREATE OR REPLACE FUNCTION reset_sequences(username IN VARCHAR) RETURNS void AS $$ DECLARE statements CURSOR FOR SELECT sequence_name FROM information_schema.sequences ORDER BY sequence_name ; BEGIN FOR stmt IN statements LOOP EXECUTE 'ALTER SEQUENCE ' || quote_ident(stmt.sequence_name) || ' RESTART;'; END LOOP; END; $$ LANGUAGE plpgsql; SELECT reset_sequences('scodoc'); """ ) ) db.session.commit() def clear_scodoc_cache(): """Clear ScoDoc cache This cache (currently Redis) is persistent between invocation and it may be necessary to clear it during developement or tests. """ # attaque directement redis, court-circuite ScoDoc: import redis r = redis.Redis() r.flushall() # Also clear local caches: sco_preferences.clear_base_preferences() # --------- Logging def log(msg: str): """log a message. If Flask app, use configured logger, else stderr. """ try: dept = getattr(g, "scodoc_dept", "") msg = f" ({dept}) {msg}" except RuntimeError: # Flask Working outside of application context. pass if current_app and not current_app.config["DEBUG"]: current_app.logger.info(msg) else: sys.stdout.flush() sys.stderr.write( f"""[{time.strftime("%a %b %d %H:%M:%S %Y")}] scodoc: {msg}\n""" ) sys.stderr.flush() # Debug: log call stack def log_call_stack(): log("Call stack:\n" + "\n".join(x.strip() for x in traceback.format_stack()[:-1])) # Alarms by email: def send_scodoc_alarm(subject, txt): from app import email sender = email.get_from_addr() email.send_email(subject, sender, ["exception@scodoc.org"], txt) from app.models import Departement from app.scodoc import notesdb as ndb, sco_preferences from app.scodoc import sco_cache def scodoc_flash_status_messages(): """Should be called on each page: flash messages indicating specific ScoDoc status""" email_test_mode_address = sco_preferences.get_preference("email_test_mode_address") if email_test_mode_address: flash( f"Mode test: mails redirigés vers {email_test_mode_address}", category="warning", ) def critical_error(msg): """Handle a critical error: flush all caches, display message to the user""" import app.scodoc.sco_utils as scu log(f"\n*** CRITICAL ERROR: {msg}") send_scodoc_alarm(f"CRITICAL ERROR: {msg}", msg) clear_scodoc_cache() raise ScoValueError( f""" Une erreur est survenue. Si le problème persiste, merci de contacter le support ScoDoc via {scu.SCO_DISCORD_ASSISTANCE} {msg} """ )