re2o/logs/models.py

# -*- mode: python; coding: utf-8 -*-
# Re2o est un logiciel d'administration développé initiallement au rezometz. Il
# se veut agnostique au réseau considéré, de manière à être installable en
# quelques clics.
#
# Copyright © 2020  Jean-Romain Garnier
#
# 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.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""logs.models
The models definitions for the logs app
"""
from reversion.models import Version, Revision
from django.utils.translation import ugettext_lazy as _
from django.contrib.auth.models import Group
from django.db.models import Q
from django.apps import apps
from netaddr import EUI
from macaddress.fields import default_dialect

from machines.models import IpList
from machines.models import Interface
from machines.models import Machine
from machines.models import MachineType
from users.models import User
from users.models import Adherent
from users.models import Club
from topologie.models import Room
from topologie.models import Port

from .forms import classes_for_action_type


def make_version_filter(key, value):
    """
    Builds a filter for a Version object to filter by argument in its
    serialized_date
    :param key: str, The argument's key
    :param value: str or int, The argument's value
    :returns: A Q filter
    """
    # The lookup is done in a json string, so it has to be formated
    # based on the value's type (to add " or not)
    if type(value) is str:
        formatted_value = "\"{}\"".format(value)
    else:
        formatted_value = str(value)

    return (
        Q(serialized_data__contains='\"{}\": {},'.format(key, formatted_value))
        | Q(serialized_data__contains='\"{}\": {}}}'.format(key, formatted_value))
    )


############################
#  Machine history search  #
############################

class MachineHistorySearchEvent:
    def __init__(self, user, machine, interface, start=None, end=None):
        """Initialise an instance of MachineHistorySearchEvent.

        Args:
            user: User, the user owning the machine at the time of the event.
            machine: Version, the machine version related to the interface.
            interface: Version, the interface targeted by this event.
            start: datetime, the date at which this version was created
                (default: None).
            end: datetime, the date at which this version was replace by a new
                one (default: None).
        """
        self.user = user
        self.machine = machine
        self.interface = interface
        self.ipv4 = IpList.objects.get(id=interface.field_dict["ipv4_id"]).ipv4
        self.mac = self.interface.field_dict["mac_address"]
        self.start_date = start
        self.end_date = end
        self.comment = interface.revision.get_comment() or None

    def is_similar(self, elt2):
        """Check whether two events are similar enough to be merged.

        Args:
            elt2: MachineHistorySearchEvent, the event to compare with self.

        Returns:
            A boolean, True if the events can be merged and False otherwise.
        """
        return (
            elt2 is not None
            and self.user.id == elt2.user.id
            and self.ipv4 == elt2.ipv4
            and self.machine.field_dict["id"] == elt2.machine.field_dict["id"]
            and self.interface.field_dict["id"] == elt2.interface.field_dict["id"]
        )

    def __repr__(self):
        return "{} ({} - ): from {} to {} ({})".format(
            self.machine,
            self.mac,
            self.ipv4,
            self.start_date,
            self.end_date,
            self.comment or "No comment"
        )


class MachineHistorySearch:
    def __init__(self):
        self.events = []
        self._last_evt = None

    def get(self, search, params):
        """Get the events in machine histories related to the search.

        Args:
            search: the IP or MAC address used in the search.
            params: the dictionary built by the search view.

        Returns:
            A list of MachineHistorySearchEvent in reverse chronological order.
        """
        self.start = params.get("s", None)
        self.end = params.get("e", None)
        search_type = params.get("t", 0)

        self.events = []
        if search_type == "ip":
            try:
                return self._get_by_ip(search)[::-1]
            except:
                pass
        elif search_type == "mac":
            try:
                search = EUI(search, dialect=default_dialect())
                return self._get_by_mac(search)[::-1]
            except:
                pass

        return []

    def _add_revision(self, user, machine, interface):
        """Add a new revision to the chronological order.

        Args:
            user: User, the user owning the maching at the time of the event.
            machine: Version, the machine version related to the interface.
            interface: Version, the interface targeted by this event.
        """
        evt = MachineHistorySearchEvent(user, machine, interface)
        evt.start_date = interface.revision.date_created

        # Try not to recreate events if it's unnecessary
        if evt.is_similar(self._last_evt):
            return

        # Mark the end of validity of the last element
        if self._last_evt and not self._last_evt.end_date:
            self._last_evt.end_date = evt.start_date

            # If the event ends before the given date, remove it
            if self.start and evt.start_date.date() < self.start:
                self._last_evt = None
                self.events.pop()

        # Make sure the new event starts before the given end date
        if self.end and evt.start_date.date() > self.end:
            return

        # Save the new element
        self.events.append(evt)
        self._last_evt = evt

    def _get_interfaces_for_ip(self, ip):
        """Get the Version objects of interfaces with the given IP
        address.

        Args:
            ip: the string corresponding to the IP address.

        Returns:
            An iterable object with the Version objects of interfaces with the
            given IP address.
        """
        # TODO: What if ip list was deleted?
        try:
            ip_id = IpList.objects.get(ipv4=ip).id
        except IpList.DoesNotExist:
            return []

        return (
            Version.objects.get_for_model(Interface)
            .filter(make_version_filter("ipv4", ip_id))
            .order_by("revision__date_created")
        )

    def _get_interfaces_for_mac(self, mac):
        """Get the Version objects of interfaces with the given MAC
        address.

        Args:
            mac: the string corresponding to the MAC address.

        Returns:
            An iterable object with the Version objects of interfaces with the
            given MAC address.
        """
        return (
            Version.objects.get_for_model(Interface)
            .filter(make_version_filter("mac_address", str(mac)))
            .order_by("revision__date_created")
        )

    def _get_machines_for_interface(self, interface):
        """Get the Version objects of machines with the given interface.

        Args:
            interface: Version, the interface used to find machines.

        Returns:
            An iterable object with the Version objects of machines to which
            the given interface was assigned.
        """
        machine_id = interface.field_dict["machine_id"]
        return (
            Version.objects.get_for_model(Machine)
            .filter(make_version_filter("pk", machine_id))
            .order_by("revision__date_created")
        )

    def _get_user_for_machine(self, machine):
        """Get the User instance owning the given machine.

        Args:
            machine: Version, the machine used to find its owner.

        Returns:
            The User instance of the owner of the given machine.
        """
        # TODO: What if user was deleted?
        user_id = machine.field_dict["user_id"]
        return User.objects.get(id=user_id)

    def _get_by_ip(self, ip):
        """Get events related to the given IP address.

        Args:
            ip: the string corresponding to the IP address.

        Returns:
            A list of MachineHistorySearchEvent related to the given IP
            address.
        """
        interfaces = self._get_interfaces_for_ip(ip)

        for interface in interfaces:
            machines = self._get_machines_for_interface(interface)

            for machine in machines:
                user = self._get_user_for_machine(machine)
                self._add_revision(user, machine, interface)

        return self.events

    def _get_by_mac(self, mac):
        """Get events related to the given MAC address.

        Args:
            mac: the string corresponding to the MAC address.

        Returns:
            A list of MachineHistorySearchEvent related to the given MAC
            address.
        """
        interfaces = self._get_interfaces_for_mac(mac)

        for interface in interfaces:
            machines = self._get_machines_for_interface(interface)

            for machine in machines:
                user = self._get_user_for_machine(machine)
                self._add_revision(user, machine, interface)

        return self.events


############################
#  Generic history classes #
############################

class RelatedHistory:
    def __init__(self, version):
        """Initialise an instance of RelatedHistory.

        Args:
            version: Version, the version related to the history.
        """
        self.version = version
        self.app_name = version.content_type.app_label
        self.model_name = version.content_type.model
        self.object_id = version.object_id
        self.name = version.object_repr

        if self.model_name:
            self.name = "{}: {}".format(self.model_name.title(), self.name)

    def __eq__(self, other):
        return (
            self.model_name == other.model_name
            and self.object_id == other.object_id
        )

    def __hash__(self):
        return hash((self.model_name, self.object_id))


class HistoryEvent:
    def __init__(self, version, previous_version=None, edited_fields=None):
        """Initialise an instance of HistoryEvent.

        Args:
            version: Version, the version of the object for this event.
            previous_version: Version, the version of the object before this
                event (default: None).
            edited_fields: list, The list of modified fields by this event
                (default: None).
        """
        self.version = version
        self.previous_version = previous_version
        self.edited_fields = edited_fields or []
        self.date = version.revision.date_created
        self.performed_by = version.revision.user
        self.comment = version.revision.get_comment() or None

    def _repr(self, name, value):
        """Get the appropriate representation of the given field.

        Args:
            name: the name of the field
            value: the value of the field

        Returns:
            The string corresponding to the appropriate representation of the
            given field.
        """
        if value is None:
            return _("None")

        return value

    def edits(self, hide=[]):
        """Get the list of the changes performed during this event.

        Args:
            hide: the list of fields for which not to show details (default:
            []).

        Returns:
            The list of fields edited by the event to display.
        """
        edits = []

        for field in self.edited_fields:
            if field in hide:
                # Don't show sensitive information
                edits.append((field, None, None))
            else:
                edits.append((
                    field,
                    self._repr(field, self.previous_version.field_dict[field]),
                    self._repr(field, self.version.field_dict[field])
                ))

        return edits


class History:
    def __init__(self):
        self.name = None
        self.events = []
        self.related = []  # For example, a machine has a list of its interfaces
        self._last_version = None
        self.event_type = HistoryEvent

    def get(self, instance_id, model):
        """Get the list of history events of the given object.

        Args:
            instance_id: int, the id of the instance to lookup.
            model: class, the type of object to lookup.

        Returns:
            A list of HistoryEvent, in reverse chronological order, related to
            the given object or None if no version was found.
        """
        self.events = []

        # Get all the versions for this instance, with the oldest first
        self._last_version = None
        interface_versions = (
            Version.objects.get_for_model(model)
            .filter(make_version_filter("pk", instance_id))
            .order_by("revision__date_created")
        )

        for version in interface_versions:
            self._add_revision(version)

        # Return None if interface_versions was empty
        if self._last_version is None:
            return None

        self.name = self._last_version.object_repr
        return self.events[::-1]

    def _compute_diff(self, v1, v2, ignoring=[]):
        """Find the edited fields between two versions.

        Args:
            v1: Version to compare.
            v2: Version to compare.
            ignoring: a list of fields to ignore.

        Returns:
            The list of field names in v1 that are different from the ones in
            v2.
        """
        fields = []

        for key in v1.field_dict.keys():
            if key not in ignoring and v1.field_dict[key] != v2.field_dict[key]:
                fields.append(key)

        return fields

    def _add_revision(self, version):
        """Add a new revision to the chronological order.

        Args:
            version: Version, the version of the interface for this event.
        """
        diff = None
        if self._last_version is not None:
            diff = self._compute_diff(version, self._last_version)

        # Ignore "empty" events
        # but always keep the first event
        if not diff and self._last_version:
            self._last_version = version
            return

        evt = self.event_type(version, self._last_version, diff)
        self.events.append(evt)
        self._last_version = version


############################
#     Revision history     #
############################

class VersionAction(HistoryEvent):
    def __init__(self, version):
        self.version = version

    def name(self):
        return self.version._object_cache or self.version.object_repr

    def application(self):
        return self.version.content_type.app_label

    def model_name(self):
        return self.version.content_type.model

    def object_id(self):
        return self.version.object_id

    def object_type(self):
        return apps.get_model(self.application(), self.model_name())

    def edits(self, hide=["password", "pwd_ntlm", "gpg_fingerprint"]):
        """Get the list of the changes performed during this event.

        Args:
            hide: the list of fields for which not to show details (default:
            ["password", "pwd_ntlm", "gpg_fingerprint"]).

        Returns:
            The list of fields edited by the event to display.
        """
        self.previous_version = self._previous_version()

        if self.previous_version is None:
            return None, None, None

        self.edited_fields = self._compute_diff(self.version, self.previous_version)
        return super(VersionAction, self).edits(hide)

    def _previous_version(self):
        """Get the previous version of self.

        Returns:
            The Version corresponding to the previous version of self, or None
            in case of exception.
        """
        model = self.object_type()
        try:
            query = (
                make_version_filter("pk", self.object_id())
                & Q(
                    revision__date_created__lt=self.version.revision.date_created
                )
            )
            return (Version.objects.get_for_model(model)
                    .filter(query)
                    .order_by("-revision__date_created")[0])
        except Exception:
            return None

    def _compute_diff(self, v1, v2, ignoring=["pwd_ntlm"]):
        """Find the edited fields between two versions.

        Args:
            v1: Version to compare.
            v2: Version to compare.
            ignoring: a list of fields to ignore (default: ["pwd_ntlm"]).

        Returns:
            The list of field names in v1 that are different from the ones in
            v2.
        """
        fields = []

        for key in v1.field_dict.keys():
            if key not in ignoring and v1.field_dict[key] != v2.field_dict[key]:
                fields.append(key)

        return fields


class RevisionAction:
    """A Revision may group multiple Version objects together."""

    def __init__(self, revision):
        self.performed_by = revision.user
        self.revision = revision
        self.versions = [VersionAction(v) for v in revision.version_set.all()]

    def id(self):
        return self.revision.id

    def date_created(self):
        return self.revision.date_created

    def comment(self):
        return self.revision.get_comment()


class ActionsSearch:
    def get(self, params):
        """Get the Revision objects corresponding to the search.

        Args:
            params: dictionary built by the search view.

        Returns:
            The QuerySet of Revision objects corresponding to the search.
        """
        user = params.get("u", None)
        start = params.get("s", None)
        end = params.get("e", None)
        action_types = params.get("t", None)

        query = Q()

        if user:
            query &= Q(user__pseudo=user)

        if start:
            query &= Q(date_created__gte=start)

        if end:
            query &= Q(date_created__lte=end)

        action_models = self.models_for_action_types(action_types)
        if action_models:
            query &= Q(version__content_type__model__in=action_models)

        return (
            Revision.objects.all()
            .filter(query)
            .select_related("user")
            .prefetch_related("version_set__object")
        )

    def models_for_action_types(self, action_types):
        if action_types is None:
            return None

        classes = []
        for action_type in action_types:
            c = classes_for_action_type(action_type)

            # Selecting "all" removes the filter
            if c is None:
                return None

            classes += list(map(str.lower, c))

        return classes


############################
#  Class-specific history  #
############################

class UserHistoryEvent(HistoryEvent):
    def _repr(self, name, value):
        """Get the appropriate representation of the given field.

        Args:
            name: the name of the field
            value: the value of the field

        Returns:
            The string corresponding to the appropriate representation of the
            given field.
        """
        if name == "groups":
            if len(value) == 0:
                # Removed all the user's groups
                return _("None")

            # value is a list of ints
            groups = []
            for gid in value:
                # Try to get the group name, if it's not deleted
                try:
                    groups.append(Group.objects.get(id=gid).name)
                except Group.DoesNotExist:
                    # TODO: Find the group name in the versions?
                    groups.append("{} ({})".format(_("Deleted"), gid))

            return ", ".join(groups)
        elif name == "state":
            if value is not None:
                return User.STATES[value][1]
            else:
                return _("Unknown")
        elif name == "email_state":
            if value is not None:
                return User.EMAIL_STATES[value][1]
            else:
                return _("Unknown")
        elif name == "room_id" and value is not None:
            # Try to get the room name, if it's not deleted
            try:
                return Room.objects.get(id=value)
            except Room.DoesNotExist:
                # TODO: Find the room name in the versions?
                return "{} ({})".format(_("Deleted"), value)
        elif name == "members" or name == "administrators":
            if len(value) == 0:
                # Removed all the club's members
                return _("None")

            # value is a list of ints
            users = []
            for uid in value:
                # Try to get the user's name, if theyr're not deleted
                try:
                    users.append(User.objects.get(id=uid).pseudo)
                except User.DoesNotExist:
                    # TODO: Find the user's name in the versions?
                    users.append("{} ({})".format(_("Deleted"), uid))

            return ", ".join(users)

        return super(UserHistoryEvent, self)._repr(name, value)

    def edits(self, hide=["password", "pwd_ntlm", "gpg_fingerprint"]):
        """Get the list of the changes performed during this event.

        Args:
            hide: the list of fields for which not to show details (default:
            ["password", "pwd_ntlm", "gpg_fingerprint"]).

        Returns:
            The list of fields edited by the event to display.
        """
        return super(UserHistoryEvent, self).edits(hide)

    def __eq__(self, other):
        return (
            self.edited_fields == other.edited_fields
            and self.date == other.date
            and self.performed_by == other.performed_by
            and self.comment == other.comment
        )

    def __hash__(self):
        return hash((frozenset(self.edited_fields), self.date, self.performed_by, self.comment))

    def __repr__(self):
        return "{} edited fields {} ({})".format(
            self.performed_by,
            self.edited_fields or "nothing",
            self.comment or "No comment"
        )


class UserHistory(History):
    def __init__(self):
        super(UserHistory, self).__init__()
        self.event_type = UserHistoryEvent

    def get(self, user_id, model):
        """Get the the list of UserHistoryEvent related to the object.

        Args:
            user_id: int, the id of the user to lookup.

        Returns:
            The list of UserHistoryEvent, in reverse chronological order,
            related to the object, or None if nothing was found.
        """
        self.events = []

        # Try to find an Adherent object
        # If it exists, its id will be the same as the user's
        adherents = (
            Version.objects.get_for_model(Adherent)
            .filter(make_version_filter("pk", user_id))
        )
        try:
            obj = adherents[0]
            model = Adherent
        except IndexError:
            obj = None

        # Fallback on a Club
        if obj is None:
            clubs = (
                Version.objects.get_for_model(Club)
                .filter(make_version_filter("pk", user_id))
            )

            try:
                obj = clubs[0]
                model = Club
            except IndexError:
                obj = None

        # If nothing was found, abort
        if obj is None:
            return None

        # Add in "related" elements the list of objects
        # that were once owned by this user
        self.related = (
            Version.objects.all()
            .filter(make_version_filter("user", user_id))
            .order_by("content_type__model")
        )
        self.related = [RelatedHistory(v) for v in self.related]
        self.related = list(dict.fromkeys(self.related))

        # Get all the versions for this user, with the oldest first
        self._last_version = None
        user_versions = (
            Version.objects.get_for_model(User)
            .filter(make_version_filter("pk", user_id))
            .order_by("revision__date_created")
        )

        for version in user_versions:
            self._add_revision(version)

        # Update name
        self.name = self._last_version.field_dict["pseudo"]

        # Do the same thing for the Adherent of Club
        self._last_version = None
        obj_versions = (
            Version.objects.get_for_model(model)
            .filter(make_version_filter("pk", user_id))
            .order_by("revision__date_created")
        )

        for version in obj_versions:
            self._add_revision(version)

        # Remove duplicates and sort
        self.events = list(dict.fromkeys(self.events))
        return sorted(
            self.events,
            key=lambda e: e.date,
            reverse=True
        )

    def _add_revision(self, version):
        """Add a new revision to the chronological order.

        Args:
            version: Version, the version of the user for this event.
        """
        diff = None
        if self._last_version is not None:
            diff = self._compute_diff(
                version,
                self._last_version,
                ignoring=["last_login", "pwd_ntlm", "email_change_date"]
            )

        # Ignore "empty" events like login
        # but always keep the first event
        if not diff and self._last_version:
            self._last_version = version
            return

        evt = UserHistoryEvent(version, self._last_version, diff)
        self.events.append(evt)
        self._last_version = version


class MachineHistoryEvent(HistoryEvent):
    def _repr(self, name, value):
        """Return the appropriate representation of the given field.

        Args:
            name: the name of the field
            value: the value of the field

        Returns:
            The string corresponding to the appropriate representation of the
            given field.
        """
        if name == "user_id":
            try:
                return User.objects.get(id=value).pseudo
            except User.DoesNotExist:
                return "{} ({})".format(_("Deleted"), value)

        return super(MachineHistoryEvent, self)._repr(name, value)


class MachineHistory(History):
    def __init__(self):
        super(MachineHistory, self).__init__()
        self.event_type = MachineHistoryEvent

    def get(self, machine_id, model):
        """Get the the list of MachineHistoryEvent related to the object.

        Args:
            machine_id: int, the id of the machine to lookup.

        Returns:
            The list of MachineHistoryEvent, in reverse chronological order,
            related to the object.
        """
        self.related = (
            Version.objects.get_for_model(Interface)
            .filter(make_version_filter("machine", machine_id))
            .order_by("content_type__model")
        )

        # Create RelatedHistory objects and remove duplicates
        self.related = [RelatedHistory(v) for v in self.related]
        self.related = list(dict.fromkeys(self.related))

        return super(MachineHistory, self).get(machine_id, Machine)


class InterfaceHistoryEvent(HistoryEvent):
    def _repr(self, name, value):
        """Get the appropriate representation of the given field.

        Args:
            name: the name of the field
            value: the value of the field

        Returns:
            The string corresponding to the appropriate representation of the
            given field.
        """
        if name == "ipv4_id" and value is not None:
            try:
                return IpList.objects.get(id=value)
            except IpList.DoesNotExist:
                return "{} ({})".format(_("Deleted"), value)
        elif name == "machine_type_id":
            try:
                return MachineType.objects.get(id=value).name
            except MachineType.DoesNotExist:
                return "{} ({})".format(_("Deleted"), value)
        elif name == "machine_id":
            try:
                return Machine.objects.get(id=value).get_name() or _("No name")
            except Machine.DoesNotExist:
                return "{} ({})".format(_("Deleted"), value)
        elif name == "port_lists":
            if len(value) == 0:
                return _("None")

            ports = []
            for pid in value:
                try:
                    ports.append(Port.objects.get(id=pid).pretty_name())
                except Group.DoesNotExist:
                    ports.append("{} ({})".format(_("Deleted"), pid))

        return super(InterfaceHistoryEvent, self)._repr(name, value)


class InterfaceHistory(History):
    def __init__(self):
        super(InterfaceHistory, self).__init__()
        self.event_type = InterfaceHistoryEvent

    def get(self, interface_id, model):
        """Get the the list of InterfaceHistoryEvent related to the object.

        Args:
            interface_id: int, the id of the interface to lookup.

        Returns:
            The list of InterfaceHistoryEvent, in reverse chronological order,
            related to the object.
        """
        return super(InterfaceHistory, self).get(interface_id, Interface)


############################
#    History auto-detect   #
############################

HISTORY_CLASS_MAPPING = {
    User: UserHistory,
    Machine: MachineHistory,
    Interface: InterfaceHistory,
    "default": History
}


def get_history_class(model):
    """Get the most appropriate History subclass to represent the given model's
    history.

    Args:
        model: the class for which to get the history.

    Returns:
        The most appropriate History subclass for the given model's history,
        or History if no other was found.
    """
    try:
        return HISTORY_CLASS_MAPPING[model]()
    except KeyError:
        return HISTORY_CLASS_MAPPING["default"]()