ScoDoc-Front/app/scodoc/sco_dump_db.py

# -*- mode: python -*-
# -*- coding: utf-8 -*-

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2023 Emmanuel Viennet.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#   Emmanuel Viennet      emmanuel.viennet@viennet.net
#
##############################################################################

"""Dump base de données pour debug et support technique

Le principe est le suivant:
 1- S'il existe une base en cours d'anonymisation, s'arrête et affiche un msg
 d'erreur à l'utilisateur, qui peut décider de la supprimer.

 2- ScoDoc lance un script qui duplique la base (la copie de SCORT devient ANORT)
     -  (si elle existe deja, s'arrête)
createdb -E UTF-8 ANORT
pg_dump SCORT | psql ANORT


 3- ScoDoc lance le script d'anonymisation config/anonymize_db.py qui:
     - vide ou anonymise certaines colonnes
     - dump cette base modifiée
     - supprime cette base.

 4- La copie dump anonymisé est uploadée.


"""
import base64
import fcntl
import os
import subprocess

import requests

from flask import g, request
from flask_login import current_user

import app.scodoc.notesdb as ndb
import app.scodoc.sco_utils as scu
from app import log
from app.scodoc.sco_exceptions import ScoValueError
import sco_version

SCO_DUMP_LOCK = "/tmp/scodump.lock"


def sco_dump_and_send_db(
    message: str = "", request_url: str = "", traceback_str_base64: str = ""
):
    """Dump base de données et l'envoie anonymisée pour debug"""
    traceback_str = base64.urlsafe_b64decode(traceback_str_base64).decode(
        scu.SCO_ENCODING
    )
    # get current (dept) DB name:
    cursor = ndb.SimpleQuery("SELECT current_database()", {})
    db_name = cursor.fetchone()[0]
    ano_db_name = "ANO" + db_name
    # Lock
    try:
        x = open(SCO_DUMP_LOCK, "w+")
        fcntl.flock(x, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (IOError, OSError) as e:
        raise ScoValueError(
            "Un envoi de la base {db_name} est déjà en cours, re-essayer plus tard"
        ) from e

    try:
        # Drop if exists
        _drop_ano_db(ano_db_name)

        # Duplicate database
        _duplicate_db(db_name, ano_db_name)

        # Anonymisation
        anonymize_db(ano_db_name)

        # Send
        r = _send_db(ano_db_name, message, request_url, traceback_str=traceback_str)
        code = r.status_code

    finally:
        # Drop anonymized database
        # XXX _drop_ano_db(ano_db_name)
        # Remove lock
        fcntl.flock(x, fcntl.LOCK_UN)

    log("sco_dump_and_send_db: done.")

    return code


def _duplicate_db(db_name, ano_db_name):
    """Create new database, and copy old one into"""
    cmd = ["createdb", "-E", "UTF-8", ano_db_name]
    log(f"sco_dump_and_send_db/_duplicate_db: {cmd}")
    try:
        _ = subprocess.check_output(cmd)
    except subprocess.CalledProcessError as e:
        log(f"sco_dump_and_send_db: exception createdb {e}")
        raise ScoValueError(
            f"erreur lors de la creation de la base {ano_db_name}"
        ) from e

    cmd = f"pg_dump {db_name} | psql {ano_db_name}"
    log("sco_dump_and_send_db/_duplicate_db: {}".format(cmd))
    try:
        _ = subprocess.check_output(cmd, shell=1)
    except subprocess.CalledProcessError as e:
        log("sco_dump_and_send_db: exception {}".format(e))
        raise ScoValueError(
            f"erreur lors de la duplication de la base {db_name} vers {ano_db_name}"
        ) from e


def anonymize_db(ano_db_name):
    """Anonymize a ScoDoc database"""
    cmd = os.path.join(scu.SCO_TOOLS_DIR, "anonymize_db.py")
    log(f"anonymize_db: {cmd}")
    try:
        _ = subprocess.check_output([cmd, ano_db_name])
    except subprocess.CalledProcessError as e:
        log(f"sco_dump_and_send_db: exception in anonymisation: {e}")
        raise ScoValueError(
            f"erreur lors de l'anonymisation de la base {ano_db_name}"
        ) from e


def _get_scodoc_serial():
    try:
        with open(
            os.path.join(scu.SCODOC_VERSION_DIR, "scodoc.sn"), encoding=scu.SCO_ENCODING
        ) as f:
            return int(f.read())
    except:
        return 0


def _send_db(
    ano_db_name: str, message: str = "", request_url: str = "", traceback_str: str = ""
):
    """Dump this (anonymized) database and send it to tech support"""
    log(f"dumping anonymized database {ano_db_name}")
    try:
        dump = subprocess.check_output(
            f"pg_dump --format=custom {ano_db_name}", shell=1
        )
    except subprocess.CalledProcessError as e:
        log(f"sco_dump_and_send_db: exception in anonymisation: {e}")
        raise ScoValueError(
            f"erreur lors de l'anonymisation de la base {ano_db_name}"
        ) from e
    log(f"traceback_str={traceback_str}")
    log("uploading anonymized dump...")
    files = {"file": (ano_db_name + ".dump", dump)}
    try:
        r = requests.post(
            scu.SCO_DUMP_UP_URL,
            files=files,
            data={
                "dept_name": getattr(g, "scodoc_dept", "-"),
                "message": message or "",
                "request_url": request_url or request.url,
                "serial": _get_scodoc_serial(),
                "sco_user": str(current_user),
                "sent_by": f'"{current_user.get_nomcomplet()}" <{current_user.email}>',
                "sco_version": sco_version.SCOVERSION,
                "sco_fullversion": scu.get_scodoc_version(),
                "traceback_str": traceback_str,
            },
        )
    except requests.exceptions.ConnectionError as exc:
        log("ConnectionError: Impossible de joindre le serveur d'assistance")
        raise ScoValueError(
            """
        Impossible de joindre le serveur d'assistance (scodoc.org).
        Veuillez contacter le service informatique de votre établissement pour
        corriger la configuration de ScoDoc. Dans la plupart des cas, il
        s'agit d'un proxy mal configuré.
        """
        ) from exc
    return r


def _drop_ano_db(ano_db_name):
    """drop temp database if it exists"""
    existing_databases = [
        s.split("|")[0].strip()
        for s in subprocess.check_output(["psql", "-l"])
        .decode(scu.SCO_ENCODING)
        .split("\n")[3:]
    ]
    if ano_db_name not in existing_databases:
        log("_drop_ano_db: no temp db, nothing to drop")
        return
    cmd = ["dropdb", ano_db_name]
    log("sco_dump_and_send_db: {}".format(cmd))
    try:
        _ = subprocess.check_output(cmd)
    except subprocess.CalledProcessError as e:
        log("sco_dump_and_send_db: exception dropdb {}".format(e))
        raise ScoValueError(
            "erreur lors de la suppression de la base {}".format(ano_db_name)
        )