path: root/mga-advisor.py

# This Python file uses the following encoding: utf-8
import os
import re
import sys
from pathlib import Path
from subprocess import run
from textwrap import wrap

from PySide6.QtCore import QDate, QDir, QFile, QSettings, QTimer, Qt
from PySide6.QtGui import QCursor
from PySide6.QtUiTools import QUiLoader
from PySide6.QtWidgets import (
    QApplication,
    QComboBox,
    QDialog,
    QFileDialog,
    QHBoxLayout,
    QLineEdit,
    QMessageBox,
    QPushButton,
    QTextEdit,
    QVBoxLayout,
    QWidget,
)
import requests
import yaml
from yaml.representer import SafeRepresenter

BASE_URL = "https://bugs.mageia.org"
DUP_WORD_RE = re.compile(r'\b([-_\d\w]+)\s+\1(\s|$)')


class folded_str(str):
    pass


class literal_str(str):
    pass


class LineDialog(QDialog):
    def __init__(self, title, init="", parent=None):
        super().__init__(parent)
        self.setWindowTitle(title)
        self.name_ql = QLineEdit()
        self.name_ql.setText(init)
        apply_bt = QPushButton("Apply")
        layout = QHBoxLayout()
        layout.addWidget(self.name_ql)
        layout.addWidget(apply_bt)
        self.setLayout(layout)
        apply_bt.clicked.connect(self.apply)

    def apply(self):
        self.accept()


class SrcDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("Add source")
        self.name_ql = QLineEdit()
        self.release = QComboBox()
        self.release.addItems(["9", "8"])
        self.repo = QComboBox()
        self.repo.addItems(["core", "nonfree", "tainted"])
        apply_bt = QPushButton("Apply")
        layout = QHBoxLayout()
        layout.addWidget(self.release)
        layout.addWidget(self.repo)
        layout.addWidget(self.name_ql)
        layout.addWidget(apply_bt)
        self.setLayout(layout)
        apply_bt.clicked.connect(self.apply)

    def apply(self):
        self.accept()


class Widget(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.load_ui()
        self.setWindowTitle("Mageia advisor")
        self.settings = QSettings("Mageia", "advisor")

    def load_ui(self):
        loader = QUiLoader()
        path = Path(__file__).resolve().parent / "form.ui"
        ui_file = QFile(path)
        ui_file.open(QFile.ReadOnly)
        self.ui = loader.load(ui_file, self)
        ui_file.close()
        self.ui.subject_le.textChanged.connect(self.check_subject)
        self.ui.description_te.textChanged.connect(self.check_description)
        self.ui.retrieve_pb.clicked.connect(self.retrieve)
        self.ui.add_cve_pb.clicked.connect(self.add_cve)
        self.ui.remove_cve_pb.clicked.connect(self.remove_cve)
        self.ui.add_src_pb.clicked.connect(self.add_src)
        self.ui.remove_src_pb.clicked.connect(self.remove_src)
        self.ui.add_ref_pb.clicked.connect(self.add_reference)
        self.ui.remove_ref_pb.clicked.connect(self.remove_reference)
        self.ui.export_pb.clicked.connect(self.export)
        self.ui.configuration_pb.clicked.connect(self.select_path)
        self.ui.cancel_pb.clicked.connect(self.cancel)
        self.ui.preview_pb.clicked.connect(self.preview)
        self.ui.bug_le.editingFinished.connect(self.valid_number)
        self.ui.bug_le.returnPressed.connect(self.retrieve)
        self.subj_timer = QTimer()
        self.subj_timer.setSingleShot(True)
        self.subj_timer.timeout.connect(self.check_subject_now)
        self.desc_timer = QTimer()
        self.desc_timer.setSingleShot(True)
        self.desc_timer.timeout.connect(self.check_description_now)

    def retrieve(self):
        """
        Retrieve CVEs, URLs and Source package name from the bug report given by its number
        """
        if not self.valid_number():
            return
        QApplication.setOverrideCursor(QCursor(Qt.WaitCursor))
        self.clean()
        self.clean_status()
        self.ui.status.setText("Loading...")
        self.ui.status.repaint()
        url = os.path.join(BASE_URL, "rest/bug", self.ui.bug_le.text()) + "?include_fields=cf_rpmpkg,cf_cve,url,component"
        headers = {'Accept': 'application/json'}
        r = requests.get(url, headers=headers)
        if r.status_code == 200 and r.json()["faults"] == []:
            desc = ""
            for pkg in re.split(';|,| ', r.json()['bugs'][0]['cf_rpmpkg']):
                pkg = pkg.strip()
                if pkg == "":
                    continue
                analyze = re.search(r"([\w\-\+_]+)-\d", pkg)
                if analyze is not None:
                    pkg = analyze.group(1)
                sources = self.src_populate(pkg)
                for source in sources:
                    suffix = ".mga" + source["mga_release"]
                    if source["repo"] in ("tainted", "nonfree"):
                        suffix += "." + source["repo"]
                    self.ui.list_src.addItem(
                        " ".join([source["mga_release"],
                                  source["repo"],
                                  source["package"] +
                                  "-" + source["version"] +
                                  "-" + source["release"] +
                                  suffix
                                 ]))
                desc += f" {pkg}"
            for cve in re.split(';|,| ', r.json()['bugs'][0]['cf_cve']):
                cve = cve.strip()
                if cve != "":
                    self.ui.list_cve.addItem(cve)
            self.ui.list_ref.addItem(os.path.join(BASE_URL, f"show_bug.cgi?id={self.ui.bug_le.text()}"))
            for url in re.split(';|,| ', r.json()['bugs'][0]['url']):
                url = url.strip()
                if url != "":
                    self.ui.list_ref.addItem(url)
            if "component" in r.json()['bugs'][0]:
                if r.json()['bugs'][0]["component"] == "Security":
                    self.ui.security_rb.setChecked(True)
                    self.ui.subject_le.setText(f"Updated{desc} packages fix security vulnerabilities")
                else:
                    self.ui.bugfix_rb.setChecked(True)
                    self.ui.subject_le.setText(f"Updated{desc} packages fix ")
            self.clean_status()
        else:
            self.ui.status.setText("No info retrieved")
            QTimer.singleShot(5000, self.clean_status)
        QApplication.restoreOverrideCursor()

    def clean_status(self):
        self.ui.status.setText("")
        self.ui.status.setStyleSheet('color: black')

    def set_error(self, err):
        self.ui.status.setText(err)
        self.ui.status.setStyleSheet('color: red')
        self.ui.status.repaint()

    def clean(self):
        # Delete all fields except bug number
        self.ui.list_ref.clear()
        self.ui.list_src.clear()
        self.ui.subject_le.setText("")
        self.ui.list_cve.clear()
        self.ui.description_te.clear()
        self.ui.list_ref.repaint()
        self.ui.list_src.repaint()
        self.ui.subject_le.repaint()
        self.ui.list_cve.repaint()
        self.ui.description_te.repaint()

    def check_subject(self):
        # Start or restart a short timer to avoid checking on every keystroke
        self.subj_timer.start(500)

    def check_subject_now(self):
        if dup := DUP_WORD_RE.search(self.ui.subject_le.text()):
            self.set_error(f'Subject contains a duplicate word "{dup.group(1)}"')
        else:
            self.clean_status()

    def check_description(self):
        # Start or restart a short timer to avoid checking on every keystroke
        self.desc_timer.start(500)

    def check_description_now(self):
        if dup := DUP_WORD_RE.search(self.ui.description_te.toPlainText()):
            self.set_error(f'Description contains a duplicate word "{dup.group(1)}"')
        else:
            self.clean_status()

    def add_src(self):
        dl = SrcDialog()
        dl.exec()
        newsrc = " ".join([dl.release.currentText(),
                           dl.repo.currentText(),
                           self.sanitize_line(dl.name_ql.text())])
        existing = [self.ui.list_src.item(i).text() for i in range(self.ui.list_src.count())]
        if newsrc in existing:
            self.set_error(f"Package {newsrc} is a duplicate!")
        self.ui.list_src.addItem(newsrc)

    def remove_src(self):
        self.ui.list_src.takeItem(self.ui.list_src.currentRow())
        self.clean_status()

    def add_reference(self):
        dl = LineDialog("Add reference")
        dl.exec()

        newref = self.sanitize_line(dl.name_ql.text())
        existing = [self.ui.list_ref.item(i).text() for i in range(self.ui.list_ref.count())]
        if newref in existing:
            self.set_error(f"Reference {newref} is a duplicate!")
        self.ui.list_ref.addItem(newref)

    def remove_reference(self):
        self.ui.list_ref.takeItem(self.ui.list_ref.currentRow())
        self.clean_status()

    def add_cve(self):
        init_value = f"CVE-{QDate.currentDate().year()}-"
        dl = LineDialog("Add CVE", init=init_value)
        dl.exec()

        newcve = self.sanitize_line(dl.name_ql.text())
        existing = {self.ui.list_cve.item(i).text() for i in range(self.ui.list_cve.count())}
        if newcve in existing:
            self.set_error(f"CVE {newcve} is a duplicate!")
        self.ui.list_cve.addItem(self.sanitize_line(dl.name_ql.text()))

    def remove_cve(self):
        self.ui.list_cve.takeItem(self.ui.list_cve.currentRow())
        self.clean_status()

    def adv_text(self):
        # https://stackoverflow.com/questions/74955147/how-to-add-literal-string-pattern-in-pyyaml-with-chomping-indicator
        def change_style(style, representer):
            def new_representer(dumper, data):
                scalar = representer(dumper, data)
                scalar.style = style
                return scalar
            return new_representer
        represent_folded_str = change_style('>', SafeRepresenter.represent_str)
        represent_literal_str = change_style('|', SafeRepresenter.represent_str)
        yaml.add_representer(folded_str, represent_folded_str)
        yaml.add_representer(literal_str, represent_literal_str)

        data = {}
        if self.ui.bugfix_rb.isChecked():
            data['type'] = 'bugfix'
        if self.ui.security_rb.isChecked():
            data['type'] = 'security'
        if self.ui.subject_le.text() != "":
            subject = self.sanitize_line(self.ui.subject_le.text())
            data['subject'] = folded_str(subject) if len(subject) > 68 else subject
        if self.ui.list_cve.count():
            data['CVE'] = [self.ui.list_cve.item(i).text() for i in range(self.ui.list_cve.count())]
        srcs = {}
        if self.ui.list_src.count() != 0:
            for n in range(self.ui.list_src.count()):
                release, repo, name = self.ui.list_src.item(n).text().split(" ")
                rel = int(release)
                if rel not in srcs:
                    srcs[rel] = {}
                    srcs[rel][repo] = []
                else:
                    if repo not in srcs[rel]:
                        srcs[rel][repo] = []
                srcs[rel][repo].append(self.sanitize_line(name))
            data['src'] = srcs
        if len(self.ui.description_te.toPlainText()) != 0:
            paragraphs = self.ui.description_te.toPlainText().split("\n")
            lines_desc = (line for paragraph in paragraphs for line in wrap(paragraph, width=72, break_on_hyphens=False))
            data['description'] = literal_str("\n".join(lines_desc)+"\n")
        if self.ui.list_ref.count():
            data['references'] = [self.ui.list_ref.item(i).text() for i in range(self.ui.list_ref.count())]
        return yaml.dump(data, default_flow_style=False, sort_keys=False, width=75, allow_unicode=True)

    def export(self):
        if not self.check_data():
            return
        default_path = self.settings.value("Default/path", "/mageia-advisories/advisories")
        if QDir().mkpath(QDir().homePath() + default_path) and self.ui.bug_le.text() != "":
            filename = os.path.join(QDir().homePath() + default_path, f"{self.ui.bug_le.text()}.adv")
            if os.path.exists(filename):
                response = QMessageBox.question(self, 'File exists', f'The file {filename} already exists. Do you want to overwrite it ?', QMessageBox.Yes | QMessageBox.No)
                if response == QMessageBox.No:
                    return
            with open(filename, 'w') as f:
                f.write(self.adv_text())
            QMessageBox.information(self, 'Success', f'The file {filename} has been written!')

    def select_path(self):
        filepath = QFileDialog.getExistingDirectory(self, "Select default directory for advisories repository", self.settings.value("Default/path"))
        if filepath:
            filepath = filepath[len(QDir().homePath()):]
            self.settings.setValue('Default/path', filepath)

    def check_data(self):
        anomalies = ""
        if self.ui.bug_le.text() == "":
            anomalies = "\n".join([anomalies, "- no bug number given"])
        if self.ui.list_src.count() == 0:
            anomalies = "\n".join([anomalies, "- no source given"])
        if anomalies != "":
            message = "There is anomalies:\n" + anomalies + "\nDo you want to continue?"
            mb = QMessageBox(QMessageBox.Warning, "Anomalies", message, QMessageBox.Yes | QMessageBox.No)
            returncode = mb.exec()
            if returncode != QMessageBox.Yes:
                return False
        return True

    def cancel(self):
        self.close()

    def preview(self):
        dl = QDialog()
        dl.setWindowTitle("Advisory preview")
        te = QTextEdit()
        te.setPlainText(self.adv_text())
        te.setReadOnly(True)
        te.setMinimumSize(600, 0)
        ok_bt = QPushButton("OK")
        layout = QVBoxLayout()
        layout.addWidget(te)
        layout.addWidget(ok_bt)
        dl.setLayout(layout)
        ok_bt.clicked.connect(dl.close)
        dl.exec()

    def sanitize_line(self, line):
        if len(line) != 0:
            return line.splitlines()[0].strip()
        return ""

    def valid_number(self):
        if self.ui.bug_le.text() == "":
            mb = QMessageBox(QMessageBox.Warning, "Retrieve info", "Provide a bug number first", QMessageBox.Ok)
            mb.exec()
            return False
        self.ui.bug_le.setText(re.sub(r'\D', '', self.ui.bug_le.text()))
        return True

    def src_populate(self, package):
        # retrieve information with repo, release from package name
        cmd = ["mgarepo", "rpmlog"]
        sources = []
        for mga_release in range(9, 10):
            source = {}
            p = run(cmd + [f"{mga_release}/{package}"], capture_output=True, text=True)
            if p.returncode == 0:
                # example to parse : * Tue Aug 15 2023 squidf <squidf> 116.0.5845.96-1.mga9.tainted
                line1 = p.stdout.split('\n')[0]
                analyze = re.search(r"^\*.*\s(\w*[:\.\w]+)-([\.\d]+)", line1)
                if analyze is not None:
                    # get version, remove epoch
                    version = analyze.group(1).split(":")[-1]
                    release = analyze.group(2)[:-1]
                    source["mga_release"] = str(mga_release)
                    source['package'] = package
                    source["version"] = version
                    source["release"] = release
                first = True
                for line in p.stdout.split('\n'):
                    # Skip the first one which is badly formatted by mgarepo rpmlog
                    if first:
                        first = False
                    else:
                        analyze = re.search(r"^\*.*\s(\w*[:\.\d]+)-([\.\d]+).mga\d*(.*)", line)
                        if analyze is not None:
                            source["repo"] = analyze.group(3)[1:] if analyze.group(3) else "core"
                            sources.append(source)
                            break
        return sources


def main():
    app = QApplication(sys.argv)
    widget = Widget()
    widget.show()
    return app.exec()


if __name__ == "__main__":
    main()