Merge pull request #773 from ae-utbm/taiste

SAS, Eboutic, Antispam, psycopg

.gitignore

@@ -1,4 +1,4 @@
-db.sqlite3
+*.sqlite3
 *.log
 *.pyc
 *.mo

antispam/admin.py

@@ -0,0 +1,10 @@
+from django.contrib import admin
+
+from antispam.models import ToxicDomain
+
+
+@admin.register(ToxicDomain)
+class ToxicDomainAdmin(admin.ModelAdmin):
+    list_display = ("domain", "is_externally_managed", "created")
+    search_fields = ("domain", "is_externally_managed", "created")
+    list_filter = ("is_externally_managed",)

antispam/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class AntispamConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    verbose_name = "antispam"
+    name = "antispam"

antispam/forms.py

@@ -0,0 +1,18 @@
+import re
+
+from django import forms
+from django.core.validators import EmailValidator
+from django.utils.translation import gettext_lazy as _
+
+from antispam.models import ToxicDomain
+
+
+class AntiSpamEmailField(forms.EmailField):
+    """An email field that email addresses with a known toxic domain."""
+
+    def run_validators(self, value: str):
+        super().run_validators(value)
+        # Domain part should exist since email validation is guaranteed to run first
+        domain = re.search(EmailValidator.domain_regex, value)
+        if ToxicDomain.objects.filter(domain=domain[0]).exists():
+            raise forms.ValidationError(_("Email domain is not allowed."))

antispam/management/commands/update_spam_database.py

@@ -0,0 +1,69 @@
+import requests
+from django.conf import settings
+from django.core.management import BaseCommand
+from django.db.models import Max
+from django.utils import timezone
+
+from antispam.models import ToxicDomain
+
+
+class Command(BaseCommand):
+    """Update blocked ips/mails database"""
+
+    help = "Update blocked ips/mails database"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--force", action="store_true", help="Force re-creation even if up to date"
+        )
+
+    def _should_update(self, *, force: bool = False) -> bool:
+        if force:
+            return True
+        oldest = ToxicDomain.objects.filter(is_externally_managed=True).aggregate(
+            res=Max("created")
+        )["res"]
+        return not (oldest and timezone.now() < (oldest + timezone.timedelta(days=1)))
+
+    def _download_domains(self, providers: list[str]) -> set[str]:
+        domains = set()
+        for provider in providers:
+            res = requests.get(provider)
+            if not res.ok:
+                self.stderr.write(
+                    f"Source {provider} responded with code {res.status_code}"
+                )
+                continue
+            domains |= set(res.content.decode().splitlines())
+        return domains
+
+    def _update_domains(self, domains: set[str]):
+        # Cleanup database
+        ToxicDomain.objects.filter(is_externally_managed=True).delete()
+
+        # Create database
+        ToxicDomain.objects.bulk_create(
+            [
+                ToxicDomain(domain=domain, is_externally_managed=True)
+                for domain in domains
+            ],
+            ignore_conflicts=True,
+        )
+        self.stdout.write("Domain database updated")
+
+    def handle(self, *args, **options):
+        if not self._should_update(force=options["force"]):
+            self.stdout.write("Domain database is up to date")
+            return
+        self.stdout.write("Updating domain database")
+
+        domains = self._download_domains(settings.TOXIC_DOMAINS_PROVIDERS)
+
+        if not domains:
+            self.stderr.write(
+                "No domains could be fetched from settings.TOXIC_DOMAINS_PROVIDERS. "
+                "Please, have a look at your settings."
+            )
+            return
+
+        self._update_domains(domains)

antispam/migrations/0001_initial.py

@@ -0,0 +1,35 @@
+# Generated by Django 4.2.14 on 2024-08-03 23:05
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="ToxicDomain",
+            fields=[
+                (
+                    "domain",
+                    models.URLField(
+                        max_length=253,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="domain",
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                (
+                    "is_externally_managed",
+                    models.BooleanField(
+                        default=False,
+                        help_text="True if kept up-to-date using external toxic domain providers, else False",
+                        verbose_name="is externally managed",
+                    ),
+                ),
+            ],
+        ),
+    ]

antispam/models.py

@@ -0,0 +1,19 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+
+class ToxicDomain(models.Model):
+    """Domain marked as spam in public databases"""
+
+    domain = models.URLField(_("domain"), max_length=253, primary_key=True)
+    created = models.DateTimeField(auto_now_add=True)
+    is_externally_managed = models.BooleanField(
+        _("is externally managed"),
+        default=False,
+        help_text=_(
+            "True if kept up-to-date using external toxic domain providers, else False"
+        ),
+    )
+
+    def __str__(self) -> str:
+        return self.domain

core/apps.py

@@ -20,8 +20,7 @@
 # Place - Suite 330, Boston, MA 02111-1307, USA.
 #
 #
-
-import sys
+import logging
 
 from django.apps import AppConfig
 from django.core.cache import cache
@@ -41,7 +40,7 @@ def ready(self):
         def clear_cached_memberships(**kwargs):
             Forum._club_memberships = {}
 
-        print("Connecting signals!", file=sys.stderr)
+        logging.getLogger("django").info("Connecting signals!")
         request_started.connect(
             clear_cached_memberships,
             weak=False,

core/lookups.py

@@ -21,19 +21,12 @@
 from core.models import Group, SithFile, User
 from core.views.site import search_user
 from counter.models import Counter, Customer, Product
-
-
-def check_token(request):
-    return (
-        "counter_token" in request.session.keys()
-        and request.session["counter_token"]
-        and Counter.objects.filter(token=request.session["counter_token"]).exists()
-    )
+from counter.utils import is_logged_in_counter
 
 
 class RightManagedLookupChannel(LookupChannel):
     def check_auth(self, request):
-        if not request.user.was_subscribed and not check_token(request):
+        if not request.user.was_subscribed and not is_logged_in_counter(request):
             raise PermissionDenied
 
 

core/management/commands/install_xapian.py

@@ -48,20 +48,24 @@ def _desired_version(self) -> str:
 
     def handle(self, *args, force: bool, **options):
         if not os.environ.get("VIRTUAL_ENV", None):
-            print("No virtual environment detected, this command can't be used")
+            self.stdout.write(
+                "No virtual environment detected, this command can't be used"
+            )
             return
 
         desired = self._desired_version()
         if desired == self._current_version():
             if not force:
-                print(
+                self.stdout.write(
                     f"Version {desired} is already installed, use --force to re-install"
                 )
                 return
-            print(f"Version {desired} is already installed, re-installing")
-        print(f"Installing xapian version {desired} at {os.environ['VIRTUAL_ENV']}")
+            self.stdout.write(f"Version {desired} is already installed, re-installing")
+        self.stdout.write(
+            f"Installing xapian version {desired} at {os.environ['VIRTUAL_ENV']}"
+        )
         subprocess.run(
             [str(Path(__file__).parent / "install_xapian.sh"), desired],
             env=dict(os.environ),
         ).check_returncode()
-        print("Installation success")
+        self.stdout.write("Installation success")

core/management/commands/markdown.py

@@ -35,4 +35,4 @@ def handle(self, *args, **options):
         root_path = settings.BASE_DIR
         with open(root_path / "core/fixtures/SYNTAX.md", "r") as md:
             result = markdown(md.read())
-        print(result, end="")
+        self.stdout.write(result)

core/models.py

@@ -24,6 +24,7 @@
 from __future__ import annotations
 
 import importlib
+import logging
 import os
 import unicodedata
 from datetime import date, timedelta
@@ -981,7 +982,7 @@ def is_owned_by(self, user):
             return True
         if self.is_in_sas and user.is_in_group(pk=settings.SITH_GROUP_SAS_ADMIN_ID):
             return True
-        return user.id == self.owner.id
+        return user.id == self.owner_id
 
     def can_be_viewed_by(self, user):
         if hasattr(self, "profile_of"):
@@ -1085,19 +1086,15 @@ def _repair_fs(self):
             # file storage
             parent_path = "." + self.parent.get_full_path()
             parent_full_path = settings.MEDIA_ROOT + parent_path
-            print("Parent full path: %s" % parent_full_path)
             os.makedirs(parent_full_path, exist_ok=True)
             old_path = self.file.name  # Should be relative: "./users/skia/bleh.jpg"
             new_path = "." + self.get_full_path()
-            print("Old path: %s " % old_path)
-            print("New path: %s " % new_path)
             try:
                 # Make this atomic, so that a FS problem rolls back the DB change
                 with transaction.atomic():
                     # Set the new filesystem path
                     self.file.name = new_path
                     self.save()
-                    print("New file path: %s " % self.file.path)
                     # Really move at the FS level
                     if os.path.exists(parent_full_path):
                         os.rename(
@@ -1108,25 +1105,22 @@ def _repair_fs(self):
                         # problem, and that can be solved with a simple shell
                         # command: `find . -type d -empty -delete`
             except Exception as e:
-                print("This file likely had a problem. Here is the exception:")
-                print(repr(e))
-            print("-" * 80)
+                logging.error(e)
 
     def _check_path_consistence(self):
         file_path = str(self.file)
         file_full_path = settings.MEDIA_ROOT + file_path
         db_path = ".%s" % self.get_full_path()
         if not os.path.exists(file_full_path):
-            print("%s: WARNING: real file does not exists!" % self.id)
-            print("file path: %s" % file_path, end="")
-            print("  db path: %s" % db_path)
+            print("%s: WARNING: real file does not exists!" % self.id)  # noqa T201
+            print("file path: %s" % file_path, end="")  # noqa T201
+            print("  db path: %s" % db_path)  # noqa T201
             return False
         if file_path != db_path:
-            print("%s: " % self.id, end="")
-            print("file path: %s" % file_path, end="")
-            print("  db path: %s" % db_path)
+            print("%s: " % self.id, end="")  # noqa T201
+            print("file path: %s" % file_path, end="")  # noqa T201
+            print("  db path: %s" % db_path)  # noqa T201
             return False
-        print("%s OK (%s)" % (self.id, file_path))
         return True
 
     def _check_fs(self):
@@ -1137,11 +1131,9 @@ def _check_fs(self):
         else:
             self._check_path_consistence()
 
-    def __getattribute__(self, attr):
-        if attr == "is_file":
-            return not self.is_folder
-        else:
-            return super().__getattribute__(attr)
+    @property
+    def is_file(self):
+        return not self.is_folder
 
     @cached_property
     def as_picture(self):