89 lines
4.7 KiB
Python
89 lines
4.7 KiB
Python
"""Atomic hardware identity registration for intake clients."""
|
|
import re
|
|
import uuid
|
|
from django.db import transaction
|
|
from django.db.models import Q
|
|
from rest_framework import serializers
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from .api import SecureAPIResponseMixin, StrictSerializer, DeviceSerializer
|
|
from .models import Device, DeviceIdentity, NumberSequence, ChoiceValue, AuditLog
|
|
from .query import COMPUTER_RELATIONS
|
|
from .services import add_audit
|
|
|
|
|
|
def canonical_hwid(value):
|
|
value = value.strip().lower()
|
|
if value.startswith('dmi-uuid:'):
|
|
try:
|
|
token = uuid.UUID(value[9:])
|
|
except ValueError:
|
|
raise serializers.ValidationError('Ungültige Hardware-UUID.')
|
|
if token.int in (0, (1 << 128) - 1):
|
|
raise serializers.ValidationError('Platzhalter-UUID ist nicht erlaubt.')
|
|
return 'dmi-uuid:' + str(token)
|
|
if re.fullmatch(r'dmi-serial-sha256:[0-9a-f]{64}', value):
|
|
return value
|
|
raise serializers.ValidationError('Verifizierbare Hardware-Kennung erforderlich.')
|
|
|
|
|
|
class IdentitySerializer(StrictSerializer):
|
|
hwid = serializers.CharField(max_length=100)
|
|
aliases = serializers.ListField(child=serializers.CharField(max_length=100), max_length=2, required=False, default=list)
|
|
device_kind = serializers.ChoiceField(choices=['Laptop', 'Desktop', 'AiO'])
|
|
model = serializers.CharField(max_length=200, required=False, default='')
|
|
|
|
def validate_aliases(self, values):
|
|
return list(dict.fromkeys(canonical_hwid(v) for v in values))
|
|
|
|
def validate_hwid(self, value):
|
|
return canonical_hwid(value)
|
|
|
|
|
|
class DeviceResolveView(SecureAPIResponseMixin, APIView):
|
|
@transaction.atomic
|
|
def post(self, request):
|
|
data = IdentitySerializer(data=request.data)
|
|
data.is_valid(raise_exception=True)
|
|
values = data.validated_data
|
|
hwid = values['hwid']
|
|
# Serialize allocation and identity lookup across clients and processes.
|
|
NumberSequence.objects.select_for_update().get_or_create(name='device')
|
|
keys = list(dict.fromkeys([hwid, *values['aliases']]))
|
|
# At most one value per source. Differing UUIDs/serials need review.
|
|
if len({k.split(':')[0] for k in keys}) != len(keys):
|
|
return Response({'detail': 'Widersprüchliche Hardware-Kennungen.'}, status=409)
|
|
ids = set(DeviceIdentity.objects.filter(hwid__in=keys).values_list('device_id', flat=True))
|
|
query = Q()
|
|
for relation in COMPUTER_RELATIONS:
|
|
for key in keys:
|
|
query |= Q(**{f'{relation}__hwid__iexact': key})
|
|
if key.startswith('dmi-uuid:'):
|
|
query |= Q(**{f'{relation}__hwid__iexact': key[9:]})
|
|
ids.update(Device.objects.filter(query).values_list('pk', flat=True))
|
|
if len(ids) > 1:
|
|
return Response({'detail': 'Hardware-Kennungen sind mehreren Geräten zugeordnet.'}, status=409)
|
|
device = Device.objects.get(pk=next(iter(ids))) if ids else None
|
|
if device is not None:
|
|
if device.archived_at:
|
|
return Response({'detail': 'Gerät ist archiviert; zuerst im Inventory prüfen.'}, status=409)
|
|
if values['model'] and device.model and values['model'].casefold() != device.model.strip().casefold():
|
|
return Response({'detail': 'Hardware-Kennung vorhanden, aber Modell stimmt nicht überein.'}, status=409)
|
|
stored = list(device.hardware_identities.values_list('hwid', flat=True))
|
|
for key in keys:
|
|
if any(old.split(':')[0] == key.split(':')[0] and old != key for old in stored):
|
|
return Response({'detail': 'UUID oder Seriennummer hat sich geändert; Gerät manuell prüfen.'}, status=409)
|
|
for key in keys:
|
|
DeviceIdentity.objects.get_or_create(hwid=key, defaults={'device': device})
|
|
return Response(DeviceSerializer(device).data)
|
|
status = ChoiceValue.objects.filter(kind='status', active=True, label__iexact='Eingang').first()
|
|
kind = ChoiceValue.objects.filter(kind='device_type', active=True, label__iexact=values['device_kind']).first()
|
|
if not status or not kind:
|
|
raise serializers.ValidationError('Aktiver Status Eingang und passender Gerätetyp müssen konfiguriert sein.')
|
|
device = Device.objects.create(status=status, device_type=kind, hwid=hwid, model=values['model'])
|
|
for key in keys:
|
|
DeviceIdentity.objects.create(hwid=key, device=device)
|
|
add_audit(request.user, device, AuditLog.Action.CREATE,
|
|
new=f'Gerät #{device.number} per Hardware-Kennung', api_client=request.auth.name)
|
|
return Response(DeviceSerializer(device).data, status=201)
|