🗃️ GIS API: database migrations - some logic to manage token request

5 files changed, 203 insertions, 3 deletions

diff --git a/example_project/settings.py b/example_project/settings.py
index 5c1a039d3..cc6cafebd 100644
--- a/example_project/settings.py
+++ b/example_project/settings.py
@@ -286,7 +286,8 @@ ISHTAR_DEFAULT_YEAR = 1900
 ISHTAR_MUSEUM_GAM = False  # France - AlimGAM export of exhibitions
 # exclude business days from deadline calculation
 ISHTAR_FILE_EXCLUDE_BUSSINESS_DAYS = True
-
+# timeout for request token (GIS connector)
+ISHTAR_REQUEST_TOKEN_TIMEOUT = 60*10  # 10 minutes
 ISHTAR_SLUGS = {
     "document-publisher": ["publisher"],
 }
diff --git a/ishtar_common/admin.py b/ishtar_common/admin.py
index 8d419af7c..f3c59f144 100644
--- a/ishtar_common/admin.py
+++ b/ishtar_common/admin.py
@@ -636,6 +636,7 @@ class IshtarSiteProfileAdmin(admin.ModelAdmin):
             "classes": ("collapse",),
             "fields": (
                 "experimental_feature",
+                "gis_connector",
                 "calculate_weight_on_full",
                 "locate_warehouses",
                 "use_town_for_geo",
@@ -2865,6 +2866,18 @@ class DocumentTemplateAdmin(admin.ModelAdmin):
 admin_site.register(models.DocumentTemplate, DocumentTemplateAdmin)
 
 
+@admin.register(models_rest.UserRequestToken, site=admin_site)
+class UserRequestTokenAdmin(admin.ModelAdmin):
+    list_display = ("user", "access_type", "created")
+    readonly_fields = ("key",)
+
+
+@admin.register(models_rest.UserToken, site=admin_site)
+class UserTokenAdmin(admin.ModelAdmin):
+    list_display = ("user", "access_type", "name", "last_ip", "last_access")
+    readonly_fields = ("key", "last_ip", "last_access")
+
+
 class ApiUserAdmin(admin.ModelAdmin):
     list_display = ("user_ptr", "ip")
 
diff --git a/ishtar_common/migrations/0264_userrequesttoken_usertoken.py b/ishtar_common/migrations/0264_userrequesttoken_usertoken.py
new file mode 100644
index 000000000..2a861e4b7
--- /dev/null
+++ b/ishtar_common/migrations/0264_userrequesttoken_usertoken.py
@@ -0,0 +1,64 @@
+# Generated by Django 4.2.19 on 2025-05-09 08:25
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('auth', '0012_alter_user_first_name_max_length'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('ishtar_common', '0263_add_timezone_django_v4'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UserRequestToken',
+            fields=[
+                ('created', models.DateTimeField(auto_now_add=True, verbose_name='Created')),
+                ('key', models.CharField(max_length=6, unique=True, verbose_name='Key')),
+                ('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, primary_key=True, related_name='user_request_token', serialize=False, to=settings.AUTH_USER_MODEL, verbose_name='User')),
+                ('access_type', models.CharField(choices=[('R', 'GIS - read'), ('W', 'GIS - read/write')], default='R', max_length=1, verbose_name='Access type')),
+                ('name', models.TextField(verbose_name='Name')),
+                ('limit_date', models.DateField(blank=True, null=True, verbose_name='Limit date')),
+            ],
+            options={
+                'verbose_name': 'API - GIS - Request token',
+                'verbose_name_plural': 'API - GIS - Request tokens',
+            },
+        ),
+        migrations.AlterModelOptions(
+            name='importertype',
+            options={'ordering': ('name',), 'permissions': (('view_gis_importer', 'Can export to QGIS'), ('view_own_gis_importer', 'Can export own to QGIS'), ('change_gis_importer', 'Can import from QGIS'), ('change_own_gis_importer', 'Can import own to QGIS')), 'verbose_name': 'Importer - Type', 'verbose_name_plural': 'Importer - Types'},
+        ),
+        migrations.AddField(
+            model_name='ishtarsiteprofile',
+            name='gis_connector',
+            field=models.BooleanField(default=False, verbose_name='GIS connector'),
+        ),
+        migrations.AlterField(
+            model_name='importertype',
+            name='type',
+            field=models.CharField(choices=[('tab', 'Table'), ('gis', 'GIS'), ('qgs', 'QGIS')], default='tab', max_length=3, verbose_name='Type'),
+        ),
+        migrations.CreateModel(
+            name='UserToken',
+            fields=[
+                ('key', models.CharField(max_length=40, primary_key=True, serialize=False, verbose_name='Key')),
+                ('name', models.TextField(verbose_name='Name')),
+                ('created', models.DateTimeField(auto_now_add=True, verbose_name='Created')),
+                ('access_type', models.CharField(choices=[('R', 'GIS - read'), ('W', 'GIS - read/write')], default='R', max_length=1, verbose_name='Access type')),
+                ('limit_date', models.DateField(blank=True, null=True, verbose_name='Limit date')),
+                ('last_ip', models.GenericIPAddressField(blank=True, null=True, verbose_name='Last IP')),
+                ('last_access', models.DateField(auto_now_add=True, verbose_name='Last access date')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='user_token', to=settings.AUTH_USER_MODEL, verbose_name='User')),
+            ],
+            options={
+                'verbose_name': 'API - GIS - Token',
+                'verbose_name_plural': 'API - GIS - Tokens',
+                'ordering': ('user', 'name', 'limit_date'),
+            },
+        ),
+    ]
diff --git a/ishtar_common/models.py b/ishtar_common/models.py
index da4a98ce3..c3b16e405 100644
--- a/ishtar_common/models.py
+++ b/ishtar_common/models.py
@@ -1165,6 +1165,7 @@ class IshtarSiteProfile(models.Model, Cached):
     preventive_operator = models.BooleanField(
         _("Preventive operator module"), default=False
     )
+    gis_connector = models.BooleanField(_("GIS connector"), default=False)
     underwater = models.BooleanField(_("Underwater module"), default=False)
     no_context_button = models.ForeignKey(
         "archaeological_context_records.ContextRecord",
diff --git a/ishtar_common/models_rest.py b/ishtar_common/models_rest.py
index 9ed9abd3e..e26968daf 100644
--- a/ishtar_common/models_rest.py
+++ b/ishtar_common/models_rest.py
@@ -8,13 +8,16 @@ from django.contrib.contenttypes.models import ContentType
 from django.contrib.gis.db import models
 from django.contrib.postgres.fields import ArrayField
 from django.core.files import File
+from django.utils import timezone
 from django.utils.text import slugify
 
 from django.apps import apps
 from django.template import loader
 
 from ishtar_common.models_common import BaseSheetFilter
-from ishtar_common.utils import gettext_lazy as _
+from ishtar_common.utils import get_current_profile, gettext_lazy as _
+
+from rest_framework.authtoken.models import Token
 
 UnoCalc = None
 ITALIC = None
@@ -26,9 +29,127 @@ if settings.USE_LIBREOFFICE:
         pass
 
 
+API_ACCESS_TYPES = (
+    ('R', _("GIS - read")),
+    ('W', _("GIS - read/write")),
+)
+
+API_ACCESS_TYPES_DICT = dict(API_ACCESS_TYPES)
+
+
+class UserRequestToken(models.Model):
+    """
+    Temporary key to request an API key
+    """
+    created = models.DateTimeField(_("Created"), auto_now_add=True)
+    key = models.CharField(_("Key"), max_length=6, unique=True)
+    user = models.OneToOneField(
+        User, primary_key=True, related_name="user_request_token",
+        on_delete=models.CASCADE,
+        verbose_name=_("User")
+    )  # only one request at a time
+    access_type = models.CharField(_("Access type"), default="R",
+                                   choices=API_ACCESS_TYPES, max_length=1)
+    name = models.TextField(_("Name"))
+    limit_date = models.DateField(_("Limit date"), blank=True, null=True)
+
+    class Meta:
+        verbose_name = _("API - GIS - Request token")
+        verbose_name_plural = _("API - GIS - Request tokens")
+    ADMIN_SECTION = _("API")
+
+    @classmethod
+    def clean_keys(cls):
+        profile = get_current_profile()
+        if not profile.gis_connector:
+            cls.objects.all().delete()
+            return
+        timeout = timezone.now() - datetime.timedelta(
+            seconds=settings.ISHTAR_REQUEST_TOKEN_TIMEOUT)
+        cls.objects.filter(created__lte=timeout).delete()
+
+    def generate_token(self, app_key, from_ip="127.0.0.1"):
+        """
+        Create API token if not timeouted.
+        app_key is the QGIS ID given by the QGIS auth db
+        """
+        profile = get_current_profile()
+        timeout = timezone.now() - self.created
+        if not profile.gis_connector or timeout.seconds > settings.ISHTAR_REQUEST_TOKEN_TIMEOUT or (
+                self.limit_date and self.limit_date < datetime.today()):
+            self.delete()
+            return
+        app_key = f"{app_key[:7]:0<7}"  # limit to 7 char and fill with 0
+        key = app_key + Token.generate_key()[7:]  # integrate QGIS ID with api key
+        # this is only intended to prevent bad practice such as
+        # sharing different API key with several devices and users
+        # or naive key compromission. A saisoned user can bypass with
+        # a modified QGIS auth db or plugin.
+        token = UserToken(
+            user=self.user, access_type=self.access_type,
+            limit_date=self.limit_date, last_ip=from_ip,
+            key=key, name=self.name
+        )
+        token.save()
+        self.delete()
+        return token
+
+    def save(self, *args, **kwargs):
+        while not self.key:
+            key = Token.generate_key()[:6]
+            # very unlikely but could occurs
+            if UserRequestToken.objects.filter(key=key).count():
+                continue
+            self.key = key
+        return super().save(*args, **kwargs)
+
+
+class UserToken(models.Model):
+    """
+    GIS token used for QGIS access.
+    Token is an aggregation of a standard API key and QGIS auth DB key (specific for
+    one QGIS installation)
+    """
+    key = models.CharField(_("Key"), max_length=40, primary_key=True)
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL, related_name='user_token',
+        on_delete=models.CASCADE, verbose_name=_("User")
+    )
+    name = models.TextField(_("Name"))
+    created = models.DateTimeField(_("Created"), auto_now_add=True)
+    access_type = models.CharField(_("Access type"), default="R",
+                                   choices=API_ACCESS_TYPES, max_length=1)
+    limit_date = models.DateField(_("Limit date"), blank=True, null=True)
+    last_ip = models.GenericIPAddressField(verbose_name=_("Last IP"), blank=True,
+                                           null=True)
+    last_access = models.DateField(_("Last access date"), auto_now_add=True)
+
+    class Meta:
+        verbose_name = _("API - GIS - Token")
+        verbose_name_plural = _("API - GIS - Tokens")
+        ordering = ("user", "name", "limit_date")
+    ADMIN_SECTION = _("API")
+
+    @property
+    def access_type_label(self):
+        if self.access_type in API_ACCESS_TYPES_DICT:
+            return API_ACCESS_TYPES_DICT[self.access_type]
+        return ""
+
+    def save(self, *args, **kwargs):
+        if not self.key:
+            self.key = Token.generate_key()
+        return super().save(*args, **kwargs)
+
+    def __str__(self):
+        access = self.access_type if self.access_type in API_ACCESS_TYPES_DICT else 'R'
+        return f"{self.user.username}-{API_ACCESS_TYPES_DICT[access]}"
+
+
 class ApiUser(models.Model):
     user_ptr = models.OneToOneField(
-        User, primary_key=True, related_name="apiuser", on_delete=models.CASCADE, verbose_name=_("User")
+        User, primary_key=True, related_name="apiuser", on_delete=models.CASCADE,
+        verbose_name=_("User")
     )
     ip = models.GenericIPAddressField(verbose_name=_("IP"))