mirror of
https://github.com/deunix-educ/Fail2banMqttActionBanishment.git
synced 2026-08-24 03:11:58 +02:00
98 lines
4.4 KiB
Python
98 lines
4.4 KiB
Python
import datetime
|
|
|
|
from django.contrib import admin, messages
|
|
from django.core.mail import send_mail
|
|
from django.db.models import QuerySet
|
|
from django.http import HttpRequest
|
|
from django.utils import timezone
|
|
|
|
from .enrollment import issue_join_token, join_command
|
|
from .models import BanEvent, EnrollmentRequest, JoinToken, NodeRegistry
|
|
from .views import NODE_OFFLINE_THRESHOLD_SECONDS
|
|
|
|
|
|
@admin.register(BanEvent)
|
|
class BanEventAdmin(admin.ModelAdmin):
|
|
list_display = ('received_at', 'node', 'jail_name', 'action', 'ip_address', 'bantime')
|
|
list_filter = ('jail_name', 'action', 'node')
|
|
search_fields = ('ip_address', 'node', 'jail_name')
|
|
ordering = ('-received_at',)
|
|
|
|
|
|
@admin.register(NodeRegistry)
|
|
class NodeRegistryAdmin(admin.ModelAdmin):
|
|
"""Seule UI d'édition d'alias/pays par noeud — pas de page dédiée.
|
|
node_id reste en premier et hors list_editable (Django exige que la
|
|
première colonne serve de lien vers la page de modification)."""
|
|
list_display = ('node_id', 'alias', 'country', 'dashboard_url', 'is_online', 'last_seen', 'first_seen')
|
|
list_editable = ('alias', 'country', 'dashboard_url')
|
|
search_fields = ('node_id', 'alias')
|
|
ordering = ('node_id',)
|
|
|
|
@admin.display(boolean=True, description='En ligne')
|
|
def is_online(self, obj: NodeRegistry) -> bool:
|
|
threshold = timezone.now() - datetime.timedelta(seconds=NODE_OFFLINE_THRESHOLD_SECONDS)
|
|
return obj.last_seen >= threshold
|
|
|
|
|
|
@admin.register(JoinToken)
|
|
class JoinTokenAdmin(admin.ModelAdmin):
|
|
"""Visibilité des jetons émis (manage.py create_join_token) — jamais
|
|
créés/édités ici, token_hash n'a d'ailleurs aucun intérêt à être
|
|
affiché (c'est un hash, le jeton en clair n'est montré qu'une fois,
|
|
en ligne de commande, jamais persisté)."""
|
|
list_display = ('node_name', 'created_at', 'expires_at', 'used_at', 'status')
|
|
search_fields = ('node_name',)
|
|
ordering = ('-created_at',)
|
|
|
|
@admin.display(description='Statut')
|
|
def status(self, obj: JoinToken) -> str:
|
|
if obj.used_at:
|
|
return 'Utilisé'
|
|
if timezone.now() > obj.expires_at:
|
|
return 'Expiré'
|
|
return 'Actif'
|
|
|
|
|
|
@admin.register(EnrollmentRequest)
|
|
class EnrollmentRequestAdmin(admin.ModelAdmin):
|
|
"""Demandes soumises via la page publique /communaute/ — jamais de
|
|
jeton émis/envoyé sans passer par l'action approve_and_send_token
|
|
ci-dessous, déclenchée à la main."""
|
|
list_display = ('email', 'node_name', 'country', 'status', 'created_at', 'processed_at')
|
|
list_filter = ('status',)
|
|
search_fields = ('email', 'node_name')
|
|
ordering = ('-created_at',)
|
|
actions = ['approve_and_send_token', 'reject']
|
|
|
|
@admin.action(description="Approuver et envoyer le jeton d'inscription par email")
|
|
def approve_and_send_token(self, request: HttpRequest, queryset: QuerySet[EnrollmentRequest]) -> None:
|
|
approved = 0
|
|
for enrollment in queryset.filter(status=EnrollmentRequest.STATUS_PENDING):
|
|
token, _join_token = issue_join_token(enrollment.node_name)
|
|
send_mail(
|
|
subject='Fail2banActionBanisher — inscription acceptée',
|
|
message=(
|
|
f"Votre demande pour rejoindre la communauté (noeud '{enrollment.node_name}') "
|
|
'a été acceptée.\n\n'
|
|
'Sur votre serveur, après avoir cloné le dépôt et configuré .env :\n\n'
|
|
f' {join_command(enrollment.node_name, token)}\n'
|
|
' sudo make install\n\n'
|
|
'Ce jeton est à usage unique et expire dans 60 minutes.'
|
|
),
|
|
from_email=None, # DEFAULT_FROM_EMAIL
|
|
recipient_list=[enrollment.email],
|
|
)
|
|
enrollment.status = EnrollmentRequest.STATUS_APPROVED
|
|
enrollment.processed_at = timezone.now()
|
|
enrollment.save(update_fields=['status', 'processed_at'])
|
|
approved += 1
|
|
self.message_user(request, f'{approved} demande(s) approuvée(s), jeton envoyé par email.', messages.SUCCESS)
|
|
|
|
@admin.action(description='Rejeter')
|
|
def reject(self, request: HttpRequest, queryset: QuerySet[EnrollmentRequest]) -> None:
|
|
updated = queryset.filter(status=EnrollmentRequest.STATUS_PENDING).update(
|
|
status=EnrollmentRequest.STATUS_REJECTED, processed_at=timezone.now(),
|
|
)
|
|
self.message_user(request, f'{updated} demande(s) rejetée(s).', messages.SUCCESS)
|