Merge branch 'webmilter' into 'master'

Implement Webmilter alias resolution

See merge request !12

README.rst

 AlekSIS (School Information System) — App Postbuero (Mail server management)
-==================================================================================================
+============================================================================
 
 AlekSIS
 -------
@@ -9,7 +9,20 @@ This is an application for use with the `AlekSIS®`_ platform.
 Features
 --------
 
-The author of this app did not describe it yet.
+Postbuero provides integration with various mail server functionality, among which are:
+
+ * Management of supported mail domains
+ * Management of mail addresses (mailboxes) for persons
+
+   * Public registration for domains allowing it
+
+ * Management of mail addresses (aliases) for groups
+
+   * Including support for members, owners, and guardians
+
+ * `WebMilter`_ support for Postfix
+
+   * Alias resolution for persons and groups
 
 Licence
 -------
@@ -17,6 +30,7 @@ Licence
 ::
 
   Copyright © 2020 Tom Teichler <tom.teichler@teckids.org>
+  Copyright © 2022 Tom Teichler <tom.teichler@teckids.org>
 
   Licenced under the EUPL, version 1.2 or later
 
@@ -33,5 +47,6 @@ by Teckids e.V. Please refer to the `trademark policy`_ for hints on using the t
 AlekSIS®.
 
 .. _AlekSIS®: https://edugit.org/AlekSIS/official/AlekSIS
+.. _WebMilter: https://docs.bergblau.io/concepts/webmilter/
 .. _European Union Public Licence: https://eupl.eu/
 .. _trademark policy: https://aleksis.org/pages/about

aleksis/apps/postbuero/apps.py

+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
 from aleksis.core.util.apps import AppConfig
 
+from django.apps import apps
+from django.db.models import functions
+
 
 class PostBueroConfig(AppConfig):
     name = "aleksis.apps.postbuero"
@@ -9,4 +15,33 @@ class PostBueroConfig(AppConfig):
         "Repository": "https://edugit.org/AlekSIS/Onboarding/AlekSIS-App-Postbuero",
     }
     licence = "EUPL-1.2+"
-    copyright_info = (([2021], "Tom Teichler", "tom.teichler@teckids.org"),)
+    copyright_info = (
+        ([2021], "Jonathan Weth", "dev@jonathanweth.de"),
+        ([2021], "Tom Teichler", "tom.teichler@teckids.org"),
+        ([2022], "Dominik George", "dominik.george@teckids.org"),
+    )
+
+    @classmethod
+    def get_all_scopes(cls) -> dict[str, str]:
+        """Return all OAuth scopes and their descriptions for this app."""
+        MailDomain = apps.get_model("postbuero", "MailDomain")
+        scopes = {}
+
+        label_prefix_webmilter = _("Use WebMilter APIs for domain")
+        scopes_webmilter = dict(
+            MailDomain.objects.annotate(
+                scope=functions.Concat(
+                    models.Value(f"{MailDomain.SCOPE_PREFIX_WEBMILTER}_"),
+                    models.F("domain"),
+                    output_field=models.CharField(),
+                ),
+                label=functions.Concat(
+                    models.Value(f"{label_prefix_webmilter}: "), models.F("domain")
+                ),
+            )
+            .values_list("scope", "label")
+            .distinct()
+        )
+        scopes.update(scopes_webmilter)
+
+        return scopes

aleksis/apps/postbuero/migrations/0005_group_address.py

+# Generated by Django 3.2.15 on 2022-08-10 16:29
+
+import aleksis.core.managers
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contenttypes', '0002_remove_content_type_name'),
+        ('sites', '0002_alter_domain_unique'),
+        ('core', '0041_update_gender_choices'),
+        ('postbuero', '0004_domain_is_public'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='MailAlias',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('extended_data', models.JSONField(default=dict, editable=False)),
+                ('local_part', models.CharField(max_length=64, verbose_name='Local part')),
+                ('domain', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='postbuero.maildomain', verbose_name='Domain')),
+                ('polymorphic_ctype', models.ForeignKey(editable=False, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='polymorphic_postbuero.mailalias_set+', to='contenttypes.contenttype')),
+                ('site', models.ForeignKey(default=1, editable=False, on_delete=django.db.models.deletion.CASCADE, to='sites.site')),
+            ],
+            options={
+                'verbose_name': 'Mail alias',
+                'verbose_name_plural': 'Mail alias',
+            },
+            managers=[
+                ('objects', aleksis.core.managers.PolymorphicCurrentSiteManager()),
+            ],
+        ),
+        migrations.CreateModel(
+            name='MailAliasForPerson',
+            fields=[
+                ('mailalias_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='postbuero.mailalias')),
+                ('person', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='local_mail_aliases', to='core.person', verbose_name='Person')),
+            ],
+            options={
+                'verbose_name': 'Mail alias for a person',
+                'verbose_name_plural': 'Mail aliases for persons',
+            },
+            bases=('postbuero.mailalias',),
+            managers=[
+                ('objects', aleksis.core.managers.PolymorphicCurrentSiteManager()),
+            ],
+        ),
+        migrations.CreateModel(
+            name='MailAliasForGroup',
+            fields=[
+                ('mailalias_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='postbuero.mailalias')),
+                ('group', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='local_mail_aliases', to='core.group', verbose_name='Group')),
+            ],
+            options={
+                'verbose_name': 'Mail alias for a group',
+                'verbose_name_plural': 'Mail aliases for groups',
+            },
+            bases=('postbuero.mailalias',),
+            managers=[
+                ('objects', aleksis.core.managers.PolymorphicCurrentSiteManager()),
+            ],
+        ),
+        migrations.AddConstraint(
+            model_name='mailalias',
+            constraint=models.UniqueConstraint(fields=('local_part', 'domain'), name='unique_alias_per_domain'),
+        ),
+    ]

aleksis/apps/postbuero/models.py

 from django.db import models
 from django.utils.translation import gettext_lazy as _
 
-from aleksis.core.mixins import ExtensibleModel
-from aleksis.core.models import Person
+from aleksis.core.mixins import ExtensibleModel, ExtensiblePolymorphicModel
+from aleksis.core.models import Group, Person
 
 
 class MailDomain(ExtensibleModel):
 
+    SCOPE_PREFIX_WEBMILTER = "maildomain"
+
     domain = models.CharField(verbose_name=_("Domain"), max_length=255)
 
     is_public = models.BooleanField(verbose_name=_("Public usable"), default=True)
@@ -14,6 +16,11 @@ class MailDomain(ExtensibleModel):
     def __str__(self) -> str:
         return self.domain
 
+    @property
+    def scope_webmilter(self) -> str:
+        """Return OAuth2 scope name to use WebMilter API."""
+        return f"{self.SCOPE_PREFIX_WEBMILTER}_{self.domain}"
+
     class Meta:
         permissions = (("can_use_domain", _("Can use domain")),)
 
@@ -40,3 +47,74 @@ class MailAddress(ExtensibleModel):
                 fields=["local_part", "domain"], name="unique_local_part_per_domain"
             )
         ]
+
+
+class MailAlias(ExtensiblePolymorphicModel):
+    domain = models.ForeignKey(MailDomain, verbose_name=_("Domain"), on_delete=models.CASCADE)
+    local_part = models.CharField(verbose_name=_("Local part"), max_length=64)
+
+    def __str__(self) -> str:
+        return f"{self.local_part}@{self.domain}"
+
+    def resolve(self, **kwargs):
+        raise NotImplementedError("You must use the concrete model to resovle the alias.")
+
+    class Meta:
+        verbose_name = _("Mail alias")
+        verbose_name_plural = _("Mail alias")
+        constraints = [
+            models.UniqueConstraint(fields=["local_part", "domain"], name="unique_alias_per_domain")
+        ]
+
+
+class MailAliasForPerson(MailAlias):
+    person = models.ForeignKey(
+        Person,
+        verbose_name=_("Person"),
+        null=True,
+        on_delete=models.SET_NULL,
+        related_name="local_mail_aliases",
+    )
+
+    def resolve(self, guardians: bool = False, **kwargs):
+        """Resolve alias to the e-mail address of this person or its guardians."""
+        if not self.person:
+            return []
+
+        if guardians:
+            return list(self.person.guardians.all().values_list("email", flat=True))
+
+        return [self.person.email]
+
+    class Meta:
+        verbose_name = _("Mail alias for a person")
+        verbose_name_plural = _("Mail aliases for persons")
+
+
+class MailAliasForGroup(MailAlias):
+    group = models.ForeignKey(
+        Group,
+        verbose_name=_("Group"),
+        null=True,
+        on_delete=models.SET_NULL,
+        related_name="local_mail_aliases",
+    )
+
+    def resolve(self, guardians: bool = False, owners: bool = False, **kwargs):
+        """Resolve alias to the e-mail addresses of this group's members, owners, or their guardians."""
+        if not self.group:
+            return []
+
+        if owners:
+            pq = self.group.owners.all()
+        else:
+            pq = self.group.members.all()
+
+        if guardians:
+            pq = Person.objects.filter(children__in=pq)
+
+        return list(pq.values_list("email", flat=True))
+
+    class Meta:
+        verbose_name = _("Mail alias for a group")
+        verbose_name_plural = _("Mail aliases for groups")

aleksis/apps/postbuero/urls.py

 from django.urls import include, path
 
-from rest_framework import routers
-
 from . import views
 
-router = routers.DefaultRouter()
-router.register(r"mails", views.MailAddressViewSet, basename="MailAddress")
-
 urlpatterns = [
     path("mails/manage/", views.ManageMail.as_view(), name="manage_mail"),
     path("mail_domains/", views.MailDomainListView.as_view(), name="mail_domains"),
     path("mail_domains/create/", views.MailDomainCreateView.as_view(), name="create_mail_domain"),
     path("mail_domains/<int:pk>/", views.MailDomainEditView.as_view(), name="edit_mail_domain"),
-    path("api/", include(router.urls)),
+    path(
+        "api/webmilter/alias/<str:address>/",
+        views.WebMilterAliasView.as_view(),
+        name="webmilter_resolveAliasByAddress",
+    ),
+    path(
+        "api/webmilter/domain/<str:domain>/alias/<str:local>/",
+        views.WebMilterAliasView.as_view(),
+        name="webmilter_resolveAliasByDomainAndLocalpart",
+    ),
 ]

aleksis/apps/postbuero/views.py

 from django.conf import settings
+from django.core.exceptions import BadRequest
 from django.http import HttpRequest, HttpResponse
 from django.shortcuts import render
 from django.urls import reverse_lazy
@@ -9,18 +10,20 @@ from django.views.generic.base import View
 
 import reversion
 from django_tables2 import SingleTableView
-from rest_framework import viewsets
+from oauth2_provider.views.mixins import ScopedResourceMixin
+from rest_framework.generics import RetrieveAPIView
+from rest_framework.response import Response
 from rules.contrib.views import PermissionRequiredMixin
 
 from aleksis.core.mixins import AdvancedCreateView, AdvancedEditView
 from aleksis.core.models import Activity
 from aleksis.core.util import messages
+from aleksis.core.util.auth_helpers import ClientProtectedResourceMixin
 from aleksis.core.util.core_helpers import get_site_preferences
 from aleksis.core.util.email import send_email
 
 from .forms import MailAddForm, MailDomainForm
-from .models import MailAddress, MailDomain
-from .serializers import MailAddressSerializer
+from .models import MailAddress, MailAlias, MailDomain
 from .tables import MailDomainTable
 
 
@@ -115,8 +118,56 @@ class MailDomainEditView(PermissionRequiredMixin, AdvancedEditView):
     success_message = _("The mail domain has been saved.")
 
 
-class MailAddressViewSet(viewsets.ModelViewSet):
-    serializer_class = MailAddressSerializer
+class WebMilterDomainMixin(ScopedResourceMixin, ClientProtectedResourceMixin):
+    """Base view for WebMilter domain operations."""
 
-    def get_queryset(self):
-        return MailAddress.objects.all()
+    def _get_address_parts(self):
+        if "local" in self.kwargs:
+            local_part, domain = self.kwargs["local"], self.kwargs["domain"]
+        elif "address" in self.kwargs:
+            if "@" not in self.kwargs["address"]:
+                raise BadRequest(f"E-mail address {self.kwargs['address']} is malformed")
+            local_part, domain = self.kwargs["address"].split("@")
+
+        return local_part, domain
+
+    def _get_domain(self):
+        """Get domain object by either address or domain only."""
+        local_part, domain = self._get_address_parts()
+
+        return MailDomain.objects.get(domain=domain)
+
+    def get_scopes(self, *args, **kwargs) -> list[str]:
+        """Return the scope needed to access the domain."""
+        return [self._get_domain().scope_webmilter]
+
+
+class WebMilterAliasView(WebMilterDomainMixin, RetrieveAPIView):
+    """View to resolve an alias address using WebMilter."""
+
+    def get_object(self):
+        local_part, domain = self._get_address_parts()
+        local_part = local_part.split("+")[0]
+
+        return MailAlias.objects.get(domain__domain=domain, local_part=local_part)
+
+    def _resolve_args_from_local_part(self):
+        local_part, domain = self._get_address_parts()
+
+        args = {}
+
+        if "+" not in local_part:
+            return args
+
+        extension = local_part.split("+")[1]
+        mods = list(extension)
+        if "g" in mods:
+            args["guardians"] = True
+        if "o" in mods:
+            args["owners"] = True
+
+        return args
+
+    def retrieve(self, request, *args, **kwargs):
+        args = self._resolve_args_from_local_part()
+        return Response(self.get_object().resolve(**args))