models.py

from __future__ import annotations
from typing import Tuple
from django.utils.translation import gettext_lazy as _
from django.utils.timezone import now
from django.conf import settings
from django.urls import reverse
from django.db import models, transaction
from nm2.lib.email import build_python_message
import backend.models as bmodels
from backend import const
from keyring import utils as kutils
import datetime
import os
from collections import namedtuple
from . import permissions

RequirementType = namedtuple(
    "RequirementType", ("tag", "sdesc", "desc", "sort_order"))

REQUIREMENT_TYPES = (
    RequirementType("intent", "Intent", _("Declaration of intent"), 0),
    RequirementType("sc_dmup", "SC/DMUP", _("SC/DFSG/DMUP agreement"), 1),
    RequirementType("advocate", "Advocate", _("Advocate"), 2),
    RequirementType("keycheck", "Keycheck", _("Key consistency checks"), 3),
    RequirementType("am_ok", "AM report", _("Application Manager report"), 4),
    RequirementType("approval", "Approval", _("Front Desk or DAM approval"), 5),
)

REQUIREMENT_TYPES_CHOICES = [(x.tag, x.desc) for x in REQUIREMENT_TYPES]

REQUIREMENT_TYPES_DICT = {x.tag: x for x in REQUIREMENT_TYPES}


class ProcessManager(models.Manager):
    def compute_requirements(self, status, applying_for):
        """
        Compute the process requirements for person applying for applying_for
        """
        if status == applying_for:
            raise RuntimeError("Invalid applying_for value {} for a person with status {}".format(
                applying_for, status))

        requirements = ["intent", "sc_dmup"]
        if applying_for == const.STATUS_DC_GA:
            if status != const.STATUS_DC:
                raise RuntimeError("Invalid applying_for value {} for a person with status {}".format(
                    applying_for, status))
            requirements.append("advocate")
        elif applying_for == const.STATUS_DM:
            if status != const.STATUS_DC:
                raise RuntimeError("Invalid applying_for value {} for a person with status {}".format(
                    applying_for, status))
            requirements.append("advocate")
            requirements.append("keycheck")
        elif applying_for == const.STATUS_DM_GA:
            if status == const.STATUS_DC_GA:
                requirements.append("advocate")
                requirements.append("keycheck")
            elif status == const.STATUS_DM:
                # No extra requirement: the declaration of intents is
                # sufficient
                pass
            else:
                raise RuntimeError("Invalid applying_for value {} for a person with status {}".format(
                    applying_for, status))
        elif applying_for in (const.STATUS_DD_U, const.STATUS_DD_NU):
            if status == const.STATUS_DD_U:
                # crossing from DD_U to DD_NU only needs intent:
                requirements = ["intent"]
            elif status == const.STATUS_DD_NU:
                requirements = ["intent", "advocate", 'am_ok']
            else:
                requirements.append("keycheck")
                requirements.append("am_ok")
                if status not in (const.STATUS_EMERITUS_DD, const.STATUS_REMOVED_DD):
                    requirements.append("advocate")
        elif applying_for in (const.STATUS_EMERITUS_DD, const.STATUS_REMOVED_DD):
            if status not in (const.STATUS_DD_NU, const.STATUS_DD_U):
                raise RuntimeError("Invalid applying_for value {} for a person with status {}".format(
                    applying_for, status))
            # Only intent is required to become emeritus or removed
            requirements = ["intent"]
        else:
            raise RuntimeError(
                "Invalid applying_for value {}".format(applying_for))
        requirements.append("approval")

        return requirements

    @transaction.atomic
    def create(self, person, applying_for, skip_requirements=False, **kw):
        """
        Create a new process and all its requirements
        """
        # Forbid pending persons to start processes
        if person.pending:
            raise RuntimeError(
                "Invalid applying_for value {} for a person whose account is still pending".format(applying_for))

        # Check that no active process of the same kind exists
        conflicts_with = [applying_for]
        if applying_for == const.STATUS_EMERITUS_DD:
            conflicts_with.append(const.STATUS_REMOVED_DD)
        if applying_for == const.STATUS_REMOVED_DD:
            conflicts_with.append(const.STATUS_EMERITUS_DD)
        if self.filter(person=person, applying_for__in=conflicts_with, closed_time__isnull=True).exists():
            raise RuntimeError("there is already an active process for {} to become {}".format(
                person, applying_for))

        # Compute requirements
        if skip_requirements:
            requirements = []
        else:
            requirements = self.compute_requirements(
                person.status, applying_for)

        # Create the new process
        res = self.model(person=person, applying_for=applying_for, **kw)
        # Save it to get an ID
        res.save(using=self._db)

        # Create the requirements
        for req in requirements:
            Requirement.objects.create(process=res, type=req)

        return res

    def in_early_stage(self):
        """
        Return processes that are in an early stage, that is, that still have
        an unapproved intent, sc_dmup or advocate requirement.
        """
        reqs = Requirement.objects.filter(type__in=("intent", "sc_dmup", "advocate"), approved_by__isnull=True).exclude(
            process__applying_for__in=(const.STATUS_EMERITUS_DD, const.STATUS_REMOVED_DD))
        return self.get_queryset().filter(
                closed_time__isnull=True, frozen_by__isnull=True,
                approved_by__isnull=True, requirements__in=reqs).distinct()


class Process(models.Model):
    person = models.ForeignKey(bmodels.Person, related_name="processes", on_delete=models.CASCADE)
    applying_for = models.CharField("target status", max_length=20, null=False, choices=[
                                    x[1:3] for x in const.ALL_STATUS])
    started = models.DateTimeField(
        default=now, verbose_name='process started')
    frozen_by = models.ForeignKey(
            bmodels.Person, related_name="+", blank=True, null=True,
            on_delete=models.PROTECT,
            help_text=_("Person who froze this process for review, or NULL if it is still being worked on"))
    frozen_time = models.DateTimeField(
            null=True, blank=True,
            help_text=_("Date the process was frozen for review, or NULL if it is still being worked on"))
    approved_by = models.ForeignKey(
            bmodels.Person, related_name="+", blank=True, null=True,
            on_delete=models.PROTECT,
            help_text=_("Person who reviewed this process and considered it complete, or NULL if not yet reviewed"))
    approved_time = models.DateTimeField(
            null=True, blank=True,
            help_text=_("Date the process was reviewed and considered complete, or NULL if not yet reviewed"))
    closed_by = models.ForeignKey(
            bmodels.Person, related_name="+", blank=True, null=True,
            help_text=_("Person who closed this process, or NULL if still open"), on_delete=models.PROTECT)
    closed_time = models.DateTimeField(null=True, blank=True, help_text=_(
        "Date the process was closed, or NULL if still open"))
    fd_comment = models.TextField(
        "Front Desk comments", blank=True, default="")
    rt_request = models.TextField("RT request text", blank=True, default="")
    rt_ticket = models.IntegerField("RT request ticket", null=True, blank=True)
    hide_until = models.DateTimeField(null=True, blank=True, help_text=_(
        "Hide this process from the AM dashboard until the given date"))

    objects = ProcessManager()

    def __str__(self):
        return "{} to become {}".format(self.person, self.applying_for)

    def get_absolute_url(self):
        return reverse("process_show", args=[self.pk])

    def get_admin_url(self):
        return reverse("admin:process_process_change", args=[self.pk])

    @property
    def frozen(self):
        return self.frozen_by is not None

    @property
    def approved(self):
        return self.approved_by is not None

    @property
    def closed(self):
        return self.closed_by is not None

    @property
    def a_link(self):
        from django.utils.safestring import mark_safe
        from django.utils.html import conditional_escape
        return mark_safe("<a href='{}'>→ {}</a>".format(
            conditional_escape(self.get_absolute_url()),
            conditional_escape(const.ALL_STATUS_DESCS[self.applying_for])))

    @property
    def can_advocate_self(self):
        return self.applying_for == const.STATUS_DM_GA and self.person.status == const.STATUS_DM

    @property
    def current_am_assignment(self):
        """
        Return the current Application Manager assignment for this process, or
        None if there is none.
        """
        try:
            return self.ams.select_related("am", "am__person").get(unassigned_by__isnull=True)
        except AMAssignment.DoesNotExist:
            return None

    def has_dam_approval(self) -> bool:
        """
        Returns True if a DAM member has approved the process
        """

        # This first block is for backward compatibility. Any process should
        # have an approval requirement now, but older ones don't.
        approved_by = self.approved_by
        if approved_by:
            am = approved_by.am_or_none
            if am is not None and am.is_dam:
                return True

        # TODO: This should probably be .get() instead of .filter().first(), to get an error if we have two
        appr = self.requirements.filter(type="approval").first()

        if appr is None:
            return False

        for statement in appr.statements.all():
            am = statement.fpr.person.am_or_none
            if am is None:
                continue

            if am.is_dam:
                return True

        return False

    def latest_approval_activities(self) -> Tuple[datetime.datetime, datetime.datetime]:
        """
        Compute the latest approval statement and the latest approval comment
        posterior to this approval statement. Used to determine if a FD
        approval activates a RT ticket when one hasn't been sent yet.
        """

        latest_comment_date = None
        first_next_statement_date = None

        # TODO: This should probably be .get() instead of .filter().first(), to get an error if we have two
        appr = self.requirements.filter(type="approval").first()
        if appr is None or not appr.approved_by:
            return (None, None)

        latest_log_comment = self.log.filter(
            changed_by__status__in=[
                const.STATUS_DD_U, const.STATUS_DD_NU
            ],
            action="",
        ).order_by('-logdate').first()

        if latest_log_comment:
            latest_comment_date = latest_log_comment.logdate
            latest_statement = appr.statements.filter(
                uploaded_time__gt=latest_comment_date
            ).order_by('uploaded_time').first()

            if latest_statement:
                first_next_statement_date = latest_statement.uploaded_time
        else:
            for statement in appr.statements.order_by('uploaded_time'):
                am = statement.fpr.person.am_or_none
                if am is None:
                    continue
                elif am.is_dam or am.is_fd:
                    first_next_statement_date = statement.uploaded_time

        return (latest_comment_date, first_next_statement_date)

    def permissions_of(self, visitor):
        """
        Compute which ProcessVisitorPermissions \a visitor has over this process
        """
        if visitor.is_authenticated:
            return permissions.ProcessVisitorPermissions(self, visitor)
        else:
            return permissions.ProcessVisitorPermissions(self, None)

    def add_log(self, changed_by, logtext, is_public=False, action="", logdate=None):
        """
        Add a log entry for this process
        """
        if logdate is None:
            logdate = now()
        return Log.objects.create(
                changed_by=changed_by, process=self, is_public=is_public,
                logtext=logtext, action=action, logdate=logdate)

    def get_statements_as_mbox(self, for_user):

        # Generating mailboxes in python2 is surprisingly difficult and painful.
        # A lot of this code has been put together thanks to:
        # http://wordeology.com/computer/how-to-send-good-unicode-email-with-python.html
        import mailbox
        import tempfile

        with tempfile.NamedTemporaryFile(mode="wb+") as outfile:
            mbox = mailbox.mbox(path=outfile.name, create=True)

            for req in self.requirements.all():
                perms = req.permissions_of(for_user)
                if "req_view" not in perms:
                    continue
                for stm in req.statements.all():
                    msg = build_python_message(
                        stm.uploaded_by,
                        subject="Signed statement for " + req.get_type_display(),
                        date=stm.uploaded_time,
                        body=stm.statement,
                        factory=mailbox.Message)
                    mbox.add(msg)

            mbox.close()

            outfile.seek(0)
            return outfile.read()

    @property
    def archive_email(self):
        return "archive-{}@nm.debian.org".format(self.pk)

    @property
    def mailbox_file(self):
        """
        The pathname of the archival mailbox, or None if it does not exist
        """
        PROCESS_MAILBOX_DIR = getattr(
            settings, "PROCESS_MAILBOX_DIR", "/srv/nm.debian.org/mbox/processes/")
        fname = os.path.join(PROCESS_MAILBOX_DIR,
                             "process-{}.mbox".format(self.pk))
        if os.path.exists(fname):
            return fname
        if os.path.exists(fname + ".gz"):
            return fname + ".gz"
        if os.path.exists(fname + ".xz"):
            return fname + ".xz"
        return None

    @property
    def mailbox_mtime(self):
        """
        The mtime of the archival mailbox, or None if it does not exist
        """
        fname = self.mailbox_file
        if fname is None:
            return None
        return datetime.datetime.fromtimestamp(os.path.getmtime(fname))


class Requirement(models.Model):
    process = models.ForeignKey(Process, related_name="requirements", on_delete=models.CASCADE)
    type = models.CharField(verbose_name=_(
        "Requirement type"), max_length=16, choices=REQUIREMENT_TYPES_CHOICES)
    approved_by = models.ForeignKey(bmodels.Person, null=True, blank=True, help_text=_(
        "Set to the person who reviewed and approved this requirement"), on_delete=models.PROTECT)
    approved_time = models.DateTimeField(
        null=True, blank=True, help_text=_("When the requirement has been approved"))

    class Meta:
        unique_together = ("process", "type")
        ordering = ["type"]

    def __str__(self):
        return f"{self.process.pk}: {self.type_desc}"

    @property
    def type_desc(self):
        res = REQUIREMENT_TYPES_DICT.get(self.type, None)
        if res is None:
            return self.type
        return res.desc

    @property
    def type_sdesc(self):
        res = REQUIREMENT_TYPES_DICT.get(self.type, None)
        if res is None:
            return self.type
        return res.sdesc

    def get_absolute_url(self):
        return reverse("process_req_" + self.type, args=[self.process_id])

    def get_admin_url(self):
        return reverse("admin:process_requirement_change", args=[self.pk])

    @property
    def a_link(self):
        from django.utils.safestring import mark_safe
        from django.utils.html import conditional_escape
        return mark_safe("<a href='{}'>{}</a>".format(
            conditional_escape(self.get_absolute_url()),
            conditional_escape(REQUIREMENT_TYPES_DICT[self.type].desc)))

    def permissions_of(self, visitor):
        """
        Compute which permissions \a visitor has over this requirement
        """
        if visitor.is_authenticated:
            return permissions.RequirementVisitorPermissions(self, visitor)
        else:
            return permissions.RequirementVisitorPermissions(self, None)

    def add_log(self, changed_by, logtext, is_public=False, action="", logdate=None):
        """
        Add a log entry for this requirement
        """
        if logdate is None:
            logdate = now()
        return Log.objects.create(
            changed_by=changed_by, process=self.process,
            requirement=self, is_public=is_public, logtext=logtext,
            action=action, logdate=logdate)

    def compute_status(self):
        """
        Return a dict describing the status of this requirement.

        The dict can contain:
        {
            "satisfied": bool,
            "notes": [ ("class", "text") ],
        }
        """
        meth = getattr(self, "compute_status_" + self.type, None)
        if meth is None:
            return {}
        return meth()

    def _compute_warnings_own_statement(self, notes):
        """
        Check that the statement is signed with the current active key of the
        process' person
        """
        satisfied = False
        for s in self.statements.all().select_related("uploaded_by"):
            if s.uploaded_by != self.process.person:
                notes.append(("warn", _("statement of intent uploaded by {} instead of the applicant").format(
                    s.uploaded_by.lookup_key)))
            if not s.fpr:
                notes.append(("warn", _("statement of intent not signed")))
            elif s.fpr.person != self.process.person:
                notes.append(("warn", _("statement of intent signed by {} instead of the applicant").format(
                    s.fpr.person.lookup_key)))
            elif not s.fpr.is_active:
                notes.append(
                    ("warn", _("statement of intent signed with key {} instead of the current active key").format(
                        s.fpr.fpr)))
            satisfied = True
        return satisfied

    def compute_status_intent(self):
        notes = []
        satisfied = self._compute_warnings_own_statement(notes)
        return {
            "satisfied": satisfied,
            "notes": notes,
        }

    def compute_status_sc_dmup(self):
        notes = []
        satisfied = self._compute_warnings_own_statement(notes)
        return {
            "satisfied": satisfied,
            "notes": notes,
        }

    def compute_status_advocate(self):
        notes = []
        satisfied_count = 0
        can_advocate_self = self.process.can_advocate_self
        for s in self.statements.all().select_related("uploaded_by"):
            if not can_advocate_self and s.uploaded_by == self.process.person:
                notes.append(("warn", _("statement signed by the applicant")))
            else:
                satisfied_count += 1
        if self.process.applying_for in (const.STATUS_DD_U, const.STATUS_DD_NU):
            if satisfied_count == 1:
                notes.append(
                    ("warn", _("if possible, have more than 1 advocate")))
        return {
            "satisfied": satisfied_count > 0,
            "notes": notes,
        }

    def compute_status_am_ok(self):
        # Compute the latest AM
        latest_am = self.process.current_am_assignment
        if latest_am is None:
            try:
                latest_am = self.process.ams.order_by("-unassigned_time")[0]
            except IndexError:
                latest_am = None
        notes = []
        satisfied = False
        for s in self.statements.all().select_related("uploaded_by"):
            if latest_am is None:
                notes.append(("warn", "statement of intent signed by {} but no AMs have been assigned".format(
                    s.uploaded_by.lookup_key)))
            elif s.uploaded_by != latest_am.am.person:
                notes.append(("warn", "statement of intent signed by {} instead of {} as the last assigned AM".format(
                    s.uploaded_by.lookup_key, latest_am.am.person.lookup_key)))
            satisfied = True
        return {
            "satisfied": satisfied,
            "notes": notes,
        }

    def compute_status_keycheck(self):
        notes = []
        satisfied = True
        keycheck_results = None

        if not self.process.person.fpr:
            notes.append(("error", "no key is configured for {}".format(
                self.process.person.lookup_key)))
            satisfied = False
        else:
            from keyring.models import Key
            try:
                key = Key.objects.get_or_download(self.process.person.fpr)
            except RuntimeError as e:
                key = None
                notes.append(("error", "cannot run keycheck: " + str(e)))
                satisfied = False

            if key is not None:
                try:
                    keycheck = key.keycheck()
                except RuntimeError as e:
                    notes.append(("error", "cannot run keycheck: " + str(e)))
                    satisfied = False
                else:
                    uids = []
                    has_good_uid = False

                    # The requirement in terms of web of trust to consider the keycheck
                    # as ok depends on whether the applicant is applying for DM or for
                    # DD.
                    #
                    # Note that this is not really elegant to have such constants hard
                    # coded in the code, it should probably go in backend/const.py
                    sigs_ok_req = 2
                    if self.process.applying_for in [const.STATUS_DM, const.STATUS_DM_GA]:
                        sigs_ok_req = 1

                    for ku in keycheck.uids:
                        uids.append({
                            "name": ku.uid.name.replace("@", ", "),
                            "remarks": " ".join(sorted(ku.errors)) if ku.errors else "ok",
                            "sigs_ok": ku.sigs_ok,
                            "sigs_no_key": len(ku.sigs_no_key),
                            "sigs_bad": len(ku.sigs_bad)
                        })
                        if not ku.errors and len(ku.sigs_ok) >= sigs_ok_req:
                            has_good_uid = True

                    if not has_good_uid:
                        notes.append(
                            ("warn", "no UID found that fully satisfies requirements"))
                        satisfied = False

                    keycheck_results = {
                        "main": {
                            "remarks": " ".join(sorted(keycheck.errors)) if keycheck.errors else "ok",
                        },
                        "uids": uids,
                        "updated": key.check_sigs_updated,
                    }

                    if keycheck.errors:
                        notes.append(("warn", "key has issues " +
                                      keycheck_results["main"]["remarks"]))
                        satisfied = False

        return {
            "satisfied": satisfied,
            "notes": notes,
            "keycheck": keycheck_results,
        }

    def compute_status_approval(self):
        # Compute the latest AM
        notes = []
        have_one_statement = False
        satisfied = False
        for s in self.statements.all().select_related("uploaded_by"):
            have_one_statement = True
            if not s.fpr:
                notes.append(("warn", _("statement uploaded by {} is not signed and therefore invalid").format(
                    s.upload_by.lookup_key)))
                continue
            elif not s.fpr.is_active:
                notes.append(("warn", _("statement uploaded by {} is not signed by their active key"
                                        " and therefore is invalid").format(
                    s.fpr.person.lookup_key)))
                continue

            am = s.fpr.person.am_or_none
            if am is None or (am is not None and not (am.is_fd or am.is_dam)):
                notes.append(("warn", _("statement uploaded by {}"
                                        " who has no right to file an approval requirement").format(
                    s.fpr.person.lookup_key)))
                continue

            satisfied = True

        if have_one_statement and not satisfied:
            notes.append(("warn", _("all statements here are invalid.")))
        return {
            "satisfied": satisfied,
            "notes": notes,
        }


class AMAssignment(models.Model):
    """
    AM assignment on a process
    """
    process = models.ForeignKey(Process, related_name="ams", on_delete=models.CASCADE)
    am = models.ForeignKey(bmodels.AM, related_name="+", on_delete=models.PROTECT)
    paused = models.BooleanField(default=False, help_text=_(
        "Whether this process is paused and the AM is free to take another applicant in the meantime"))
    assigned_by = models.ForeignKey(
        bmodels.Person, related_name="+", help_text=_("Person who did the assignment"), on_delete=models.PROTECT)
    assigned_time = models.DateTimeField(
        help_text=_("When the assignment happened"))
    unassigned_by = models.ForeignKey(bmodels.Person, related_name="+",
                                      blank=True, null=True, on_delete=models.PROTECT,
                                      help_text=_("Person who did the unassignment"))
    unassigned_time = models.DateTimeField(
        blank=True, null=True, help_text=_("When the unassignment happened"))

    class Meta:
        ordering = ["-assigned_by"]

    def get_admin_url(self):
        return reverse("admin:process_amassignment_change", args=[self.pk])


class Statement(models.Model):
    """
    A signed statement
    """
    requirement = models.ForeignKey(Requirement, related_name="statements", on_delete=models.CASCADE)
    fpr = models.ForeignKey(bmodels.Fingerprint, related_name="+",
                            null=True, on_delete=models.PROTECT,
                            help_text=_("Fingerprint used to verify the statement"))
    statement = models.TextField(
        verbose_name=_("Signed statement"), blank=True)
    uploaded_by = models.ForeignKey(
        bmodels.Person, related_name="+", help_text=_("Person who uploaded the statement"), on_delete=models.PROTECT)
    uploaded_time = models.DateTimeField(
        help_text=_("When the statement has been uploaded"))

    def __str__(self):
        return "{}:{}".format(self.fpr, self.requirement)

    def get_key(self):
        from keyring.models import Key
        return Key.objects.get_or_download(self.fpr.fpr)

    @property
    def statement_clean(self):
        """
        Return the statement without the OpenPGP wrapping
        """
        return kutils.cleaned_text(self.statement)

    @property
    def rfc3156(self):
        """
        If the statement is an email, parse it as rfc3156, else return None
        """
        return kutils.rfc3156(self.statement)


class Log(models.Model):
    """
    A log entry about anything that happened during a process
    """
    changed_by = models.ForeignKey(bmodels.Person, related_name="+", null=True, on_delete=models.CASCADE)
    process = models.ForeignKey(Process, related_name="log", on_delete=models.CASCADE)
    requirement = models.ForeignKey(
        Requirement, related_name="log", null=True, blank=True, on_delete=models.CASCADE)
    is_public = models.BooleanField(default=False)
    logdate = models.DateTimeField(default=now)
    action = models.CharField(max_length=16, blank=True, help_text=_(
        "Action performed with this log entry, if any"))
    logtext = models.TextField(blank=True, default="")

    class Meta:
        ordering = ["-logdate"]

    def __str__(self):
        return "{}: {}".format(self.logdate, self.logtext)

    @property
    def previous(self):
        """
        Return the previous log entry for this process.
        """
        try:
            return Log.objects.filter(logdate__lt=self.logdate, process=self.process).order_by("-logdate")[0]
        except IndexError:
            return None