models.py

"""
Core models of the New Member site
"""
from __future__ import annotations
from django.utils.translation import ugettext_lazy as _
from django.core.exceptions import ObjectDoesNotExist
from django.utils.timezone import now
from django.db import models
from django.conf import settings
from django.urls import reverse
from django.contrib.auth.models import BaseUserManager, PermissionsMixin
from django.forms.models import model_to_dict
from . import const
from . import permissions
from .fields import FingerprintField
from .utils import cached_property
import datetime
import urllib.request
import urllib.parse
import urllib.error
import re
import json

DM_IMPORT_DATE = getattr(settings, "DM_IMPORT_DATE", None)


class PersonManager(BaseUserManager):
    def create_user(self, email, **other_fields):
        if not email:
            raise ValueError('Users must have an email address')
        audit_author = other_fields.pop("audit_author", None)
        audit_notes = other_fields.pop("audit_notes", None)
        audit_skip = other_fields.pop("audit_skip", False)
        fpr = other_fields.pop("fpr", None)
        if "fullname" not in other_fields:
            other_fields["fullname"] = _build_fullname(
                    other_fields.get("cn"), other_fields.get("mn"), other_fields.get("sn"))
        user = self.model(
            email=self.normalize_email(email),
            **other_fields
        )
        user.save(using=self._db, audit_author=audit_author, audit_notes=audit_notes, audit_skip=audit_skip)
        if fpr:
            Fingerprint.objects.create(fpr=fpr, person=user, is_active=True,
                                       audit_author=audit_author,
                                       audit_notes=audit_notes,
                                       audit_skip=audit_skip)
        return user

    def create_superuser(self, email, **other_fields):
        other_fields["is_superuser"] = True
        other_fields["is_staff"] = True
        return self.create_user(email, **other_fields)

    def get_or_none(self, *args, **kw):
        """
        Same as get(), but returns None instead of raising DoesNotExist if the
        object cannot be found
        """
        try:
            return self.get(*args, **kw)
        except self.model.DoesNotExist:
            return None

    def get_housekeeper(self):
        """
        Return the housekeeping person, creating it if it does not exist yet
        """
        # Ensure that there is a __housekeeping__ user
        try:
            return self.get(email="nm@debian.org")
        except self.model.DoesNotExist:
            from dsa.models import LDAPFields
            res = self.create_user(
                is_staff=False,
                fullname="nm.debian.org Housekeeping Robot",
                email="nm@debian.org",
                bio="I am the robot that runs the automated tasks in the site",
                status=const.STATUS_DC,
                audit_skip=True)
            LDAPFields.objects.create(person=res, audit_skip=True)
            return res

    def get_from_other_db(
            self, other_db_name, uid=None, email=None, fpr=None, format_person=lambda x: str(x)):
        """
        Get one Person entry matching the informations that another database
        has about a person.

        One or more of uid, email, or fpr must be provided, and the
        function will ensure consistency in the results. That is, only one
        person will be returned, and it will raise an exception if the data
        provided match different Person entries in our database.

        other_db_name is the name of the database where the parameters come
        from, to use in generating exception messages.

        It returns None if nothing is matched.
        """
        candidates = []
        if uid is not None:
            p = self.get_or_none(ldap_fields__uid=uid)
            if p is not None:
                candidates.append((p, "uid", uid))
        if email is not None:
            p = self.get_or_none(email=email)
            if p is not None:
                candidates.append((p, "email", email))
        if fpr is not None:
            p = self.get_or_none(fprs__fpr=fpr)
            if p is not None:
                candidates.append((p, "fingerprint", fpr))

        # No candidates, nothing was found
        if not candidates:
            return None

        candidate = candidates[0]

        # Check for conflicts in the database
        for person, match_type, match_value in candidates[1:]:
            if candidate[0].pk != person.pk:
                raise self.model.MultipleObjectsReturned(
                    "{} has {} {}, which corresponds to two different users in our db: "
                    "{} (by {} {}) and {} (by {} {})".format(
                        other_db_name, match_type, match_value,
                        format_person(candidate[0]), candidate[1], candidate[2],
                        format_person(person), match_type, match_value))

        return candidate[0]


def _build_fullname(cn, mn, sn):
    if not mn:
        if not sn:
            return cn
        else:
            return "{} {}".format(cn, sn)
    else:
        if not sn:
            return "{} {}".format(cn, mn)
        else:
            return "{} {} {}".format(cn, mn, sn)


class Person(PermissionsMixin, models.Model):
    """
    A person (DM, DD, AM, applicant, FD member, DAM, anything)
    """
    class Meta:
        db_table = "person"
        ordering = ("fullname",)

    objects = PersonManager()

    # Standard Django user fields
    last_login = models.DateTimeField(_('last login'), default=now)
    date_joined = models.DateTimeField(_('date joined'), default=now)
    is_staff = models.BooleanField(default=False)
    # is_active = True
    fullname = models.CharField(_("full name"), max_length=255)

    #  enrico> For people like Wookey, do you prefer we use only cn or only sn?
    #          "sn" is used currently, and "cn" has a dash, but rather than
    #          cargo-culting that in the new NM double check it with you
    # @sgran> cn would be more usual
    # @sgran> cn is the "whole name" and you can split it up into givenName + sn if you like
    #  phil> Except that in Debian LDAP it isn't.
    #  enrico> sgran: ok. should I use 'cn' for potential new cases then?
    # @sgran> phil: indeed
    # @sgran> but if we keep doing it the other way, we'll never be in a position to change
    # @sgran> enrico: please
    #  enrico> sgran: ack

    # Most user fields mirror Debian LDAP fields

    # First/Given name, or only name in case of only one name
    email = models.EmailField(_("email address"), null=False, unique=True)
    bio = models.TextField(_("short biography"), blank=True, null=False, default="",
                           help_text=_("Please enter here a short biographical information"))

    # Membership status
    status = models.CharField(_("current status in the project"), max_length=20, null=False,
                              choices=[(x.tag, x.ldesc) for x in const.ALL_STATUS])
    status_changed = models.DateTimeField(_("when the status last changed"), null=False, default=now)
    fd_comment = models.TextField(_("Front Desk comments"), null=False, blank=True, default="")
    # null=True because we currently do not have the info for old entries
    created = models.DateTimeField(_("Person record created"), null=True, default=now)
    expires = models.DateField(
            _("Expiration date for the account"), null=True, blank=True, default=None,
            help_text=_("This person will be deleted after this date if the status is still {} and"
                        " no Process has started").format(const.STATUS_DC))
    pending = models.CharField(_("Nonce used to confirm this pending record"), max_length=255, unique=False, blank=True)
    last_vote = models.DateField(null=True, blank=True, help_text=_("date of the last vote done with this uid"))

    status_description = models.CharField(
            max_length=128, null=False, blank=True,
            help_text=_("Override for the status description for when the default is not enough"))

    def get_full_name(self):
        return self.fullname

    def get_short_name(self):
        return self.ldap_fields.cn

    def get_username(self):
        return self.email

    @property
    def is_anonymous(self):
        return False

    @property
    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def set_password(self, raw_password):
        pass

    def check_password(self, raw_password):
        return False

    def set_unusable_password(self):
        pass

    def has_usable_password(self):
        return False

    @property
    def fingerprint(self):
        """
        Return the Fingerprint associated to this person, or None if there is
        none
        """
        # If there is more than one active fingerprint, return a random one.
        # This should not happen, and a nightly maintenance task will warn if
        # it happens.
        for f in self.fprs.filter(is_active=True):
            return f
        return None

    @property
    def fpr(self):
        """
        Return the current fingerprint for this Person
        """
        f = self.fingerprint
        if f is not None:
            return f.fpr
        return None

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ["cn", "status"]

    @property
    def person(self):
        """
        Allow to call foo.person to get a Person record, regardless if foo is a Person or an AM
        """
        return self

    @cached_property
    def perms(self):
        """
        Get permission tags for this user
        """
        res = set()
        is_dd = self.status in (const.STATUS_DD_U, const.STATUS_DD_NU)

        if is_dd:
            res.add("dd")
            am = self.am_or_none
            if am:
                if am.is_am:
                    res.add("am")
                if am.is_admin:
                    res.add("am")
                    res.add("admin")
            else:
                res.add("am_candidate")

        return frozenset(res)

    @property
    def is_dd(self):
        return "dd" in self.perms

    @property
    def is_am(self):
        return "am" in self.perms

    @property
    def is_admin(self):
        return "admin" in self.perms

    def can_become_am(self):
        """
        Check if the person can become an AM
        """
        return "am_candidate" in self.perms

    @property
    def am_or_none(self):
        try:
            return self.am
        except AM.DoesNotExist:
            return None

    @property
    def changed_before_data_import(self):
        return DM_IMPORT_DATE is not None and self.status in (
                const.STATUS_DM, const.STATUS_DM_GA) and self.status_changed <= DM_IMPORT_DATE

    def permissions_of(self, visitor):
        """
        Compute which PersonVisitorPermissions the given person has over this person
        """
        return permissions.PersonVisitorPermissions(self, visitor)

    @property
    def preferred_email(self):
        """
        Return uid@debian.org if the person is a DD, else return the email
        field.
        """
        if self.status in (const.STATUS_DD_U, const.STATUS_DD_NU):
            return "{}@debian.org".format(self.ldap_fields.uid)
        else:
            return self.email

    def __str__(self):
        return "{} <{}>".format(self.fullname, self.email)

    def __repr__(self):
        return "{} <{}> [uid:{}, status:{}]".format(
                self.fullname, self.email, self.get_ldap_uid(), self.status)

    def get_absolute_url(self):
        return reverse("person", kwargs=dict(key=self.lookup_key))

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

    def get_picture_url(self):
        """
        Return a URL to a picture for the person, if available, else None
        """
        for identity in self.identities.all():
            if identity.picture:
                return identity.picture
        return None

    def get_ldap_uid(self):
        try:
            return self.ldap_fields.uid
        except ObjectDoesNotExist:
            return 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(self.lookup_key)))

    def get_ddpo_url(self):
        return "http://qa.debian.org/developer.php?{}".format(urllib.parse.urlencode(dict(login=self.preferred_email)))

    def get_portfolio_url(self):
        parms = dict(
            email=self.preferred_email,
            name=self.fullname,
            gpgfp="",
            username="",
            nonddemail=self.email,
            wikihomepage="",
            forumsid=""
        )
        if self.fpr:
            parms["gpgfp"] = self.fpr
        if self.get_ldap_uid():
            parms["username"] = self.get_ldap_uid()
        return "http://portfolio.debian.net/result?" + urllib.parse.urlencode(parms)

    def get_contributors_url(self):
        from signon.models import Identity
        if self.is_dd:
            return "https://contributors.debian.org/contributor/{}@debian".format(self.ldap_fields.uid)

        try:
            debsso = self.identities.get(issuer="debsso")
            if debsso.subject.endswith("@users.alioth.debian.org"):
                return "https://contributors.debian.org/contributor/{}@alioth".format(debsso.subject[:-24])
        except Identity.DoesNotExist:
            pass

        return None

    _new_status_table = {
        const.STATUS_DC: [const.STATUS_DC_GA, const.STATUS_DM, const.STATUS_DD_U, const.STATUS_DD_NU],
        const.STATUS_DC_GA: [const.STATUS_DM_GA, const.STATUS_DD_U, const.STATUS_DD_NU],
        const.STATUS_DM: [const.STATUS_DM_GA, const.STATUS_DD_NU, const.STATUS_DD_U],
        const.STATUS_DM_GA: [const.STATUS_DD_NU, const.STATUS_DD_U],
        const.STATUS_DD_NU: [const.STATUS_DD_U, const.STATUS_EMERITUS_DD],
        const.STATUS_DD_U: [const.STATUS_DD_NU, const.STATUS_EMERITUS_DD],
        const.STATUS_EMERITUS_DD: [const.STATUS_DD_U, const.STATUS_DD_NU],
        const.STATUS_REMOVED_DD: [const.STATUS_DD_U, const.STATUS_DD_NU],
    }

    @property
    def possible_new_statuses(self):
        """
        Return a list of possible new statuses that can be requested for the
        person
        """
        if self.pending:
            return []

        statuses = list(self._new_status_table.get(self.status, []))

        # Compute statuses one is already applying for in active processes
        blacklist = []
        if statuses:
            import process.models as pmodels
            for proc in pmodels.Process.objects.filter(person=self, closed_time__isnull=True):
                blacklist.append(proc.applying_for)
        if const.STATUS_DD_U in blacklist:
            blacklist.append(const.STATUS_DD_NU)
        if const.STATUS_DD_NU in blacklist:
            blacklist.append(const.STATUS_DD_U)
        if const.STATUS_EMERITUS_DD in blacklist:
            blacklist.append(const.STATUS_REMOVED_DD)
            blacklist.append(const.STATUS_DD_U)
            blacklist.append(const.STATUS_DD_NU)
        if const.STATUS_REMOVED_DD in blacklist:
            blacklist.append(const.STATUS_EMERITUS_DD)
            blacklist.append(const.STATUS_DD_U)
            blacklist.append(const.STATUS_DD_NU)

        for status in blacklist:
            try:
                statuses.remove(status)
            except ValueError:
                pass

        return statuses

    def make_pending(self, days_valid=30):
        """
        Make this person a pending person.

        It does not automatically save the Person.
        """
        from django.utils.crypto import get_random_string
        self.pending = get_random_string(length=12,
                                         allowed_chars='abcdefghijklmnopqrstuvwxyz'
                                         'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789')
        self.expires = now().date() + datetime.timedelta(days=days_valid)

    def save(self, *args, **kw):
        """
        Save, and add an entry to the Person audit log.

        Extra arguments that can be passed:

            audit_author: Person instance of the person doing the change
            audit_notes: free form text annotations for this change
            audit_skip: skip audit logging, used only for tests

        """
        # Extract our own arguments, so that they are not passed to django
        author = kw.pop("audit_author", None)
        notes = kw.pop("audit_notes", "")
        audit_skip = kw.pop("audit_skip", False)

        if audit_skip:
            changes = None
        else:
            # Get the previous version of the Person object, so that PersonAuditLog
            # can compute differences
            if self.pk:
                old_person = Person.objects.get(pk=self.pk)
            else:
                old_person = None

            changes = Person.diff(old_person, self)
            if changes and not author:
                raise RuntimeError("Cannot save a Person instance without providing Author information")

        # Perform the save; if we are creating a new person, this will also
        # fill in the id/pk field, so that PersonAuditLog can link to us
        super(Person, self).save(*args, **kw)

        # Finally, create the audit log entry
        if changes:
            PersonAuditLog.objects.create(
                    person=self, author=author, notes=notes, changes=PersonAuditLog.serialize_changes(changes))

    @classmethod
    def diff(cls, old_person: "Person", new_person: "Person"):
        """
        Compute the changes between two different instances of a Person model
        """
        exclude = ["last_login", "date_joined", "groups", "user_permissions"]
        changes = {}
        if old_person is None:
            for k, nv in list(model_to_dict(new_person, exclude=exclude).items()):
                changes[k] = [None, nv]
        else:
            old = model_to_dict(old_person, exclude=exclude)
            new = model_to_dict(new_person, exclude=exclude)
            for k, nv in list(new.items()):
                ov = old.get(k, None)
                # Also ignore changes like None -> ""
                if ov != nv and (ov or nv):
                    changes[k] = [ov, nv]
        return changes

    @property
    def lookup_key(self):
        """
        Return a key that can be used to look up this person in the database
        using Person.lookup.

        Currently, this is the uid if available, else the email.
        """
        uid = self.get_ldap_uid()
        if uid:
            return uid
        elif self.email:
            return self.email
        else:
            return self.fpr

    @classmethod
    def lookup(cls, key):
        try:
            if "@" in key:
                return cls.objects.get(email=key)
            elif re.match(r"^[0-9A-Fa-f]{32,40}$", key):
                return cls.objects.get(fpr=key.upper())
            else:
                return cls.objects.get(ldap_fields__uid=key)
        except cls.DoesNotExist:
            return None

    @classmethod
    def lookup_by_email(cls, addr):
        """
        Return the person corresponding to an email address, or None if no such
        person has been found.
        """
        try:
            return cls.objects.get(email=addr)
        except cls.DoesNotExist:
            pass
        if not addr.endswith("@debian.org"):
            return None
        try:
            return cls.objects.get(uid=addr[:-11])
        except cls.DoesNotExist:
            return None

    @classmethod
    def lookup_or_404(cls, key):
        from django.http import Http404
        res = cls.lookup(key)
        if res is not None:
            return res
        raise Http404


class FingerprintManager(BaseUserManager):
    def create(self, **fields):
        audit_author = fields.pop("audit_author", None)
        audit_notes = fields.pop("audit_notes", None)
        audit_skip = fields.pop("audit_skip", False)
        fields["fpr"] = fields["fpr"].replace(" ", "")
        res = self.model(**fields)
        res.save(using=self._db, audit_author=audit_author, audit_notes=audit_notes, audit_skip=audit_skip)
        return res


class Fingerprint(models.Model):
    """
    A fingerprint for a person
    """
    class Meta:
        db_table = "fingerprints"

    objects = FingerprintManager()

    person = models.ForeignKey(Person, related_name="fprs", on_delete=models.CASCADE)
    fpr = FingerprintField(verbose_name=_("OpenPGP key fingerprint"), max_length=40, unique=True)
    is_active = models.BooleanField(default=False, help_text=_("whether this key is curently in use"))
    last_upload = models.DateField(
            null=True, blank=True, help_text=_("date of the last ftp-master upload done with this key"))

    def __str__(self):
        return self.fpr

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

    def save(self, *args, **kw):
        """
        Save, and add an entry to the Person audit log.

        Extra arguments that can be passed:

            audit_author: Person instance of the person doing the change
            audit_notes: free form text annotations for this change
            audit_skip: skip audit logging, used only for tests

        """
        # Extract our own arguments, so that they are not passed to django
        author = kw.pop("audit_author", None)
        notes = kw.pop("audit_notes", "")
        audit_skip = kw.pop("audit_skip", False)

        if audit_skip:
            changes = None
        else:
            # Get the previous version of the Fingerprint object, so that
            # PersonAuditLog can compute differences
            if self.pk:
                existing_fingerprint = Fingerprint.objects.get(pk=self.pk)
            else:
                existing_fingerprint = None

            changes = PersonAuditLog.diff_fingerprint(existing_fingerprint, self)
            if changes and not author:
                raise RuntimeError("Cannot save a Fingerprint instance without providing Author information")

        # Perform the save; if we are creating a new person, this will also
        # fill in the id/pk field, so that PersonAuditLog can link to us
        super(Fingerprint, self).save(*args, **kw)

        # Finally, create the audit log entry
        if changes:
            if existing_fingerprint is not None and existing_fingerprint.person.pk != self.person.pk:
                PersonAuditLog.objects.create(
                        person=existing_fingerprint.person, author=author, notes=notes,
                        changes=PersonAuditLog.serialize_changes(changes))
            PersonAuditLog.objects.create(
                    person=self.person, author=author, notes=notes, changes=PersonAuditLog.serialize_changes(changes))

        # If we are saving an active fingerprint, make all others inactive
        if self.is_active:
            for fpr in Fingerprint.objects.filter(person=self.person, is_active=True).exclude(pk=self.pk):
                fpr.is_active = False
                fpr.save(audit_notes=notes, audit_author=author, audit_skip=audit_skip)


class PersonAuditLog(models.Model):
    person = models.ForeignKey(Person, related_name="audit_log", on_delete=models.CASCADE)
    logdate = models.DateTimeField(null=False, auto_now_add=True)
    author = models.ForeignKey(Person, related_name="+", null=False, on_delete=models.CASCADE)
    notes = models.TextField(null=False, default="")
    changes = models.TextField(null=False, default="{}")

    def __str__(self):
        return "{:%Y-%m-%d %H:%S}:{}: {}:{}".format(
                self.logdate, self.person.lookup_key, self.author.lookup_key, self.notes)

    @classmethod
    def diff_fingerprint(cls, existing_fpr, new_fpr):
        """
        Compute the changes between two different instances of a Fingerprint model
        """
        exclude = []
        changes = {}
        if existing_fpr is None:
            for k, nv in list(model_to_dict(new_fpr, exclude=exclude).items()):
                changes["fpr:{}:{}".format(new_fpr.fpr, k)] = [None, nv]
        else:
            old = model_to_dict(existing_fpr, exclude=exclude)
            new = model_to_dict(new_fpr, exclude=exclude)
            for k, nv in list(new.items()):
                ov = old.get(k, None)
                # Also ignore changes like None -> ""
                if ov != nv and (ov or nv):
                    changes["fpr:{}:{}".format(existing_fpr.fpr, k)] = [ov, nv]
        return changes

    @classmethod
    def serialize_changes(cls, changes):
        class Serializer(json.JSONEncoder):
            def default(self, o):
                if isinstance(o, datetime.datetime):
                    return o.strftime("%Y-%m-%d %H:%M:%S")
                elif isinstance(o, datetime.date):
                    return o.strftime("%Y-%m-%d")
                else:
                    return json.JSONEncoder.default(self, o)
        return json.dumps(changes, cls=Serializer)


class AM(models.Model):
    """
    Extra info for people who are or have been AMs, FD members, or DAMs
    """
    class Meta:
        db_table = "am"

    person = models.OneToOneField(Person, related_name="am", on_delete=models.CASCADE)
    slots = models.IntegerField(null=False, default=1)
    is_am = models.BooleanField(_("Active AM"), null=False, default=True)
    is_fd = models.BooleanField(_("FD member"), null=False, default=False)
    is_dam = models.BooleanField(_("DAM"), null=False, default=False)
    # Automatically computed as true if any applicant was approved in the last
    # 6 months
    is_am_ctte = models.BooleanField(_("NM CTTE member"), null=False, default=False)
    # null=True because we currently do not have the info for old entries
    created = models.DateTimeField(_("AM record created"), null=True, default=now)
    fd_comment = models.TextField(_("Front Desk comments"), null=False, blank=True, default="")

    def __str__(self):
        return "%s %c%c%c" % (
            str(self.person),
            "a" if self.is_am else "-",
            "f" if self.is_fd else "-",
            "d" if self.is_dam else "-",
        )

    def __repr__(self):
        return "%s %c%c%c slots:%d" % (
            repr(self.person),
            "a" if self.is_am else "-",
            "f" if self.is_fd else "-",
            "d" if self.is_dam else "-",
            self.slots)

    def get_absolute_url(self):
        return reverse("person", kwargs=dict(key=self.person.lookup_key))

    @property
    def is_admin(self):
        return self.is_fd or self.is_dam

    @classmethod
    def list_available(cls, free_only=False):
        """
        Get a list of active AMs with free slots, ordered by uid.

        Each AM is annotated with stats_active, stats_held and stats_free, with
        the number of NMs, held NMs and free slots.
        """
        import process.models as pmodels

        ams = {}
        for am in AM.objects.all():
            am.proc_active = []
            am.proc_held = []
            ams[am] = am

        for p in pmodels.AMAssignment.objects.filter(
                unassigned_by__isnull=True, process__frozen_by__isnull=True,
                process__approved_by__isnull=True,
                process__closed_time__isnull=True).select_related("am"):
            am = ams[p.am]
            if p.paused:
                am.proc_held.append(p)
            else:
                am.proc_active.append(p)

        res = []
        for am in list(ams.values()):
            am.stats_active = len(am.proc_active)
            am.stats_held = len(am.proc_held)
            am.stats_free = am.slots - am.stats_active

            if free_only and am.stats_free <= 0:
                continue

            res.append(am)
        res.sort(key=lambda x: (-x.stats_free, x.stats_active))
        return res

    @property
    def lookup_key(self):
        """
        Return a key that can be used to look up this manager in the database
        using AM.lookup.

        Currently, this is the lookup key of the person.
        """
        return self.person.lookup_key

    @classmethod
    def lookup(cls, key):
        p = Person.lookup(key)
        if p is None:
            return None
        return p.am_or_none

    @classmethod
    def lookup_or_404(cls, key):
        from django.http import Http404
        res = cls.lookup(key)
        if res is not None:
            return res
        raise Http404