src/ssh_keys/views.py

   1 from datetime import datetime, timedelta
   2 import re
   3 from django.contrib.auth.mixins import LoginRequiredMixin
   4 from django.contrib import messages
   5 from django.db import IntegrityError
   6 from django.db.models import Q
   7 from django import http
   8 from django.shortcuts import get_object_or_404
   9 from django.utils.timezone import now
  10 from django.utils.translation import gettext as _
  11 from django.views.decorators.csrf import csrf_exempt
  12 from django.views.generic import ListView, CreateView, DeleteView
  13 from services.models import Service
  14 from .models import SSHKey
  15 from .utils import parse_log_line
  16
  17
  18 class SSHKeysView(LoginRequiredMixin, ListView):
  19     def get_queryset(self):
  20         return SSHKey.objects.filter(user=self.request.user)
  21
  22
  23 class AddSSHKeyView(LoginRequiredMixin, CreateView):
  24     fields = ['key']
  25     model = SSHKey
  26     success_url = '/ssh/'
  27     template_name = 'ssh_keys/sshkey_add.html'
  28
  29     def form_valid(self, form):
  30         form.instance.user = self.request.user
  31         try:
  32             return super().form_valid(form)
  33         except ValueError as e:
  34             messages.add_message(self.request, messages.ERROR, e)
  35         except IntegrityError:
  36             messages.add_message(self.request, messages.ERROR, _("Key already in the database."))
  37         return http.HttpResponseRedirect(self.success_url)
  38
  39
  40
  41 class DeleteSSHKeyView(LoginRequiredMixin, DeleteView):
  42     success_url = '/ssh/'
  43
  44     def get_queryset(self):
  45         return SSHKey.objects.filter(user=self.request.user)
  46
  47
  48 @csrf_exempt
  49 def ssh_keys_seen(request):
  50     key = request.GET.get('key')
  51     service = get_object_or_404(Service, key=key)
  52     n = now()
  53
  54     logline_re = r'^(?P<time>\w{3}\s+\d+\s+\d\d:\d\d:\d\d).*: Accepted publickey for .* from .* port .* ssh2: (?P<algo>\w+) (?P<hash>[a-f0-9:]+)$'
  55
  56     last_seen = {}
  57     for line in request.body.decode('latin1').split('\n'):
  58         if not line.strip():
  59             continue
  60         data = parse_log_line(line)
  61         if data is None:
  62             continue
  63         dt = data['datetime']
  64         algo = data['algo']
  65         if 'md5' in data:
  66             hash_type = 'md5'
  67             hash_value = data['md5']
  68         else:
  69             hash_type = 'sha256'
  70             hash_value = data['sha256']
  71         key = algo, hash_type, hash_value
  72         last_seen[key] = max(last_seen.get(key, dt), dt)
  73
  74     for key, dt in last_seen.items():
  75         algo, hash_type, hash_value = key
  76         SSHKey.objects.filter(
  77             Q(last_seen_at=None) | Q(last_seen_at__lt=dt),
  78             algorithm=algo,
  79             **{f'{hash_type}_hash': hash_value}
  80         ).update(
  81             last_seen_at=dt
  82         )
  83
  84     return http.HttpResponse('ok')
  85