Update Mailcow API 2024-08a

.gitignore

@@ -136,11 +136,13 @@ htdocs/media/images/*
 htdocs/static
 /output/
 /test/output/
+/squire/logs
 
 # Secrets
 /squire/local_settings.py
 /squire/mailcowconfig.json
 /squire/secret_key.txt
+/squire/config
 
 # Disables tracking of membership Card file, need to be added manually
 /membership_file/static/images/MembershipCard.jpg

core/admin_status/config.py

@@ -0,0 +1,16 @@
+from django.urls import path
+from core.admin_status.views import LogFileView
+from core.status_collective import AdminStatusBaseConfig
+
+
+class LogConfig(AdminStatusBaseConfig):
+    url_keyword = "logs"
+    name = "Logs"
+    icon_class = "fas fa-bug"
+    url_name = "logs"
+    order_value = 2  # Value determining the order of the tabs on the admin status page
+
+    def get_urls(self):
+        return [
+            path("log", LogFileView.as_view(config=self), name="logs"),
+        ]

core/admin_status/views.py

@@ -0,0 +1,83 @@
+import os
+import re
+from django.conf import settings
+from django.views.generic import TemplateView
+
+from django.utils.safestring import SafeText
+from django.utils.html import escape
+
+from core.status_collective import AdminStatusViewMixin
+
+
+class LogFileView(AdminStatusViewMixin, TemplateView):
+    """
+    A simple page that allows viewing the content of rotating log files in `/squire/logs/squire.log.<number>`
+    """
+
+    tags = {
+        "&lt;[a-z0-9_\-:\s]*&gt;": "text-danger",
+        "[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}(,[0-9]{3})?": "far fa-clock text-muted",
+        re.escape("[debug]"): "text-secondary font-weight-bold",
+        re.escape("[info]"): "text-info font-weight-bold",
+        re.escape("[warning]"): "text-warning font-weight-bold",
+        re.escape("[error]"): "text-danger font-weight-bold",
+        re.escape("JSONParseError"): "font-weight-bold",
+        re.escape("(mailcow_api)"): "far fa-envelope text-primary",
+    }
+
+    template_name = "core/admin_status/log.html"
+    log_location = os.path.join(settings.BASE_DIR, "squire", "logs", "squire.log")
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._logfile1 = None
+        self._logfile2 = None
+        self._logfile1_name = None
+        self._logfile2_name = None
+
+        # Open both logfiles, reverse the log's lines (most recent at the top),
+        #   and make sure the most recent rotating logfile appears first.
+        try:
+            with open(self.log_location, "r") as fl:
+                data = "".join(reversed(fl.readlines()))
+                # The log format is known; the first 23 chars are an ISO-timestamp
+                log1_peek = data[:23]
+                self._logfile1 = self._format_logfile(data)
+                self._logfile1_name = os.path.join("squire", "logs", "squire.log")
+        except OSError:
+            pass
+        try:
+            with open(self.log_location + ".1", "r") as fl:
+                data = "".join(reversed(fl.readlines()))
+                if data[:23] > log1_peek:
+                    self._logfile2 = self._logfile1
+                    self._logfile1 = self._format_logfile(data)
+                    self._logfile1_name = os.path.join("squire", "logs", "squire.log.1")
+                    self._logfile2_name = os.path.join("squire", "logs", "squire.log")
+                else:
+                    self._logfile2 = self._format_logfile(data)
+                    self._logfile2_name = os.path.join("squire", "logs", "squire.log.1")
+        except OSError:
+            pass
+
+    def _format_logfile(self, data: str) -> SafeText:
+        """Formats a logfile by adding 'syntax' highlighting to specific keywords defined in this class's `tags`."""
+        text = escape(data)
+        for string, classes in self.tags.items():
+            text = SafeText(
+                re.sub(
+                    rf"(?i)(\b|(?!\w))({string})(\b|(?!\w))",
+                    rf"\g<1><span class='{classes}'>\g<2></span>\g<3>",
+                    text,
+                )
+            )
+        text = SafeText(re.sub("(\r\n|\r|\n)", "<br>", text))
+        return text
+
+    def get_context_data(self, **kwargs):
+        context = super().get_context_data(**kwargs)
+        context["log1"] = self._logfile1
+        context["log2"] = self._logfile2
+        context["log1_name"] = self._logfile1_name
+        context["log2_name"] = self._logfile2_name
+        return context

core/templates/core/admin_status/log.html

@@ -0,0 +1,46 @@
+
+{% extends 'core/base.html' %}
+
+{% load static %}
+{% load bootstrap_tabs %}
+
+{% block title %}
+  Logging
+{% endblock title %}
+
+{% block css %}
+    {{ block.super }}
+    <link rel="stylesheet" href="{% static 'css/bootstrap_tabs.css' %}">
+{% endblock css %}
+
+{% block content %}
+    {% bootstrap_tabs tabs %}
+    <h1>Logs</h1>
+    <p>Logging created by various modules. Logs are written to two rotating files sizing (at most) 5MB each.</p>
+    {% if log1 is not None %}
+        <h4>{{ log1_name }}</h4>
+        {% if not log1 %}
+            <p class="font-italic">No logging written.</p>
+        {% else %}
+            <samp class="text-break">{{ log1 }}</samp>
+        {% endif %}
+        <hr>
+    {% endif %}
+    {% if log2 is not None %}
+        <h4>{{ log2_name }}</h4>
+        {% if not log2 %}
+            <p class="font-italic">No logging written.</p>
+        {% else %}
+            <samp class="text-break">{{ log2 }}</samp>
+        {% endif %}
+        <hr>
+    {% endif %}
+    {% if log1 is None and log2 is None %}
+        <p class="font-italic">No logging written.</p>
+    {% endif %}
+{% endblock content %}
+
+{% block js_bottom %}
+    {{ block.super }}
+    <script src="{% static "js/activate_bootstrap_tooltip.js" %}"></script>
+{% endblock %}

mailcow_integration/api/client.py

@@ -2,7 +2,7 @@
 import json
 import logging
 import requests
-from typing import Generator, List, Union
+from typing import Generator, List, Optional, Union
 
 from mailcow_integration.api.exceptions import *
 from mailcow_integration.api.interface.alias import AliasType, MailcowAlias
@@ -98,12 +98,12 @@ def _make_request(
     ################
     # ALIASES
     ################
-    def get_alias_all(self) -> Generator[MailcowAlias, None, None]:
+    def get_alias_all(self) -> Generator[Optional[MailcowAlias], None, None]:
         """Gets a list of all email aliases"""
         content = self._make_request(f"get/alias/all")
         return map(lambda alias: MailcowAlias.from_json(alias), content)
 
-    def get_alias(self, id: int) -> MailcowAlias:
+    def get_alias(self, id: int) -> Optional[MailcowAlias]:
         """Gets an email alias with a specific id"""
         content = self._make_request(f"get/alias/{id}")
         return MailcowAlias.from_json(content)
@@ -176,20 +176,20 @@ def delete_aliases(self, aliases: List[MailcowAlias]) -> dict:
     ################
     # MAILBOXES
     ################
-    def get_mailbox_all(self) -> Generator[MailcowMailbox, None, None]:
+    def get_mailbox_all(self) -> Generator[Optional[MailcowMailbox], None, None]:
         """Gets a list of all mailboxes"""
         content = self._make_request(f"get/mailbox/all")
         return map(lambda mailbox: MailcowMailbox.from_json(mailbox), content)
 
     ################
     # RSPAMD SETTINGS (undocumented API)
     ################
-    def get_rspamd_setting_all(self) -> Generator[RspamdSettings, None, None]:
+    def get_rspamd_setting_all(self) -> Generator[Optional[RspamdSettings], None, None]:
         """Gets all Rspamd settings maps"""
         content = self._make_request("get/rsetting/all")
         return map(lambda rspamdsetting: RspamdSettings.from_json(rspamdsetting), content)
 
-    def get_rspamd_setting(self, id: int) -> RspamdSettings:
+    def get_rspamd_setting(self, id: int) -> Optional[RspamdSettings]:
         """Gets an Rspamd settings map with a specific id"""
         content = self._make_request(f"get/rsetting/{id}")
         return RspamdSettings.from_json(content)

mailcow_integration/api/interface/alias.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Optional, List
+from typing import Optional, List, Set
 
 from datetime import datetime
 from dataclasses import dataclass
@@ -38,6 +38,11 @@
     created: Optional[datetime] = None
     modified: Optional[datetime] = None
 
+    _cleanable_bools = ("active", "is_catch_all", "sogo_visible")
+    _cleanable_ints = ("id", "active_int", "sogo_visible_int")
+    _cleanable_strings = ("in_primary_domain", "domain")
+    _cleanable_datetimes = ("created",)
+
     def __post_init__(self):
         if self.active_int is None:
             self.active_int = int(self.active)
@@ -54,17 +59,43 @@
         return AliasType.NORMAL
 
     @classmethod
-    def from_json(cls, json: dict) -> "MailcowAlias":
-        json.update(
-            {
-                "goto": json["goto"].split(","),
-                "active": bool(json["active"]),
-                "is_catch_all": bool(json["is_catch_all"]),
-                "public_comment": json["public_comment"] or "",
-                "private_comment": json["private_comment"] or "",
-                "sogo_visible": bool(json["sogo_visible"]),
-                "created": datetime.fromisoformat(json["created"]),
-                "modified": datetime.fromisoformat(json["modified"]) if json["modified"] is not None else None,
-            }
-        )
-        return cls(**json)
+    def clean(cls, json: dict, extra_keys: Set[str] = None) -> dict:
+        address = json.get("address", None)
+        if address is None:
+            cls._issue_warning("address", address)
+            raise AttributeError(f"address was not provided when creating {cls.__name__}")
+
+        # Goto-addresses are comma-separated strings
+        goto = json.get("goto", None)
+        if goto is None or not isinstance(goto, str):
+            cls._issue_warning("goto", goto, "list")
+            raise AttributeError(f"goto was not provided or had an invalid value when creating {cls.__name__}")
+
+        new_json = {
+            "address": address,
+            "goto": goto.split(","),
+        }
+
+        # Modified can be returned as None
+        if "modified" not in json:
+            cls._issue_warning("modified", None, "ISO-datetime (or None)")
+        else:
+            modified = json.get("modified")
+            if modified is not None:
+                modified = cls._parse_as_dt("modified", json)
+            new_json["modified"] = modified
+
+        # Same for the private/public comment
+        if "public_comment" not in json:
+            cls._issue_warning("public_comment", None, "string (or None)")
+        else:
+            new_json["public_comment"] = str(json.get("public_comment") or "")
+
+        if "private_comment" not in json:
+            cls._issue_warning("private_comment", None, "string (or None)")
+        else:
+            new_json["private_comment"] = str(json.get("private_comment") or "")
+
+        extra_keys = extra_keys or set()
+        new_json.update(**super().clean(json, extra_keys=new_json.keys() | extra_keys))
+        return new_json