DaStra3003
1052 lines
34 KiB
Python
1052 lines
34 KiB
Python
import logging
|
|
from datetime import datetime
|
|
|
|
from django.core.exceptions import ValidationError, ObjectDoesNotExist
|
|
from django.db import IntegrityError, DatabaseError, transaction
|
|
|
|
from django.http import JsonResponse
|
|
from .analyze.analyze_pcap import analyze_pcap_file
|
|
|
|
from .models import *
|
|
from .exception_handling import handle_exceptions
|
|
from .db_validate import *
|
|
from rest_framework import serializers, status
|
|
from rest_framework.views import APIView
|
|
from rest_framework.response import Response
|
|
|
|
#Logger configuration
|
|
logger = logging.getLogger(__name__)
|
|
|
|
### CREATE - functions ###
|
|
# Create manufacturer
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_manufacturer(name: str):
|
|
# Input validation
|
|
if not name or not isinstance(name, str): # Checks for None, empty strings and non-strings
|
|
raise ValidationError("Ungültiger Herstellername")
|
|
|
|
# Manufacturer must not consist only of whitespace
|
|
name = name.strip()
|
|
if not name: # Checks again after stripping whitespace
|
|
raise ValidationError("Herstellername darf nicht leer sein")
|
|
|
|
# Check for duplicate
|
|
if Manufacturer.objects.filter(name=name).exists():
|
|
raise ValidationError(f"Hersteller '{name}' existiert bereits")
|
|
|
|
# Create manufacturer
|
|
manufacturer = Manufacturer.objects.create(
|
|
name=name,
|
|
created=datetime.now()
|
|
)
|
|
|
|
# Return manufacturer object as JSON
|
|
logging.info(f"Hersteller erfolgreich erstellt: {name} (ID: {manufacturer.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Hersteller wurde erfolgreich erstellt',
|
|
'data': manufacturer.to_json()
|
|
}, status=201)
|
|
|
|
# Create device type
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_device_type(device_type: str):
|
|
|
|
# Input validation
|
|
if not device_type or not isinstance(device_type, str):
|
|
raise ValidationError("Ungültiger Gerätetyp")
|
|
|
|
# Manufacturer must not consist only of whitespace
|
|
device_type = device_type.strip()
|
|
if not device_type: # Checks again after stripping whitespace
|
|
raise ValidationError("Gerätetyp darf nicht leer sein")
|
|
|
|
# Check for duplicate
|
|
if DeviceType.objects.filter(device_type=device_type).exists():
|
|
raise ValidationError(f"Gerätetyp '{device_type}' existiert bereits")
|
|
|
|
# Create type
|
|
device_type_obj = DeviceType.objects.create(
|
|
device_type=device_type
|
|
)
|
|
|
|
logging.info(f"Gerätetyp erfolgreich erstellt: {device_type} (ID: {device_type_obj.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Gerätetyp wurde erfolgreich erstellt',
|
|
'data': device_type_obj.to_json()
|
|
}, status=201)
|
|
|
|
# Create product
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_product(manufacturer_id: int, device_type_id: int, product_name: str):
|
|
|
|
# Input validation
|
|
if not product_name or not isinstance(product_name, str):
|
|
raise ValidationError("Ungültiger Produktname")
|
|
|
|
# Product name must not be empty
|
|
product_name = product_name.strip()
|
|
if not product_name:
|
|
raise ValidationError("Produktname darf nicht leer sein")
|
|
|
|
# Check manufacturer and device type
|
|
manufacturer = Manufacturer.objects.get(id=manufacturer_id)
|
|
device_type = DeviceType.objects.get(id=device_type_id)
|
|
|
|
# Check for duplicate
|
|
if Product.objects.filter(
|
|
manufacturer=manufacturer,
|
|
device_type=device_type,
|
|
product_name=product_name
|
|
).exists():
|
|
raise ValidationError(f"Produkt '{product_name}' existiert bereits für diesen Hersteller und Gerätetyp")
|
|
|
|
# Create product
|
|
product = Product.objects.create(
|
|
manufacturer=manufacturer,
|
|
device_type=device_type,
|
|
product_name=product_name,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Produkt erfolgreich erstellt: {product_name} (ID: {product.id}, Hersteller: {manufacturer.name}, Typ: {device_type.device_type})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Produkt wurde erfolgreich erstellt',
|
|
'data': product.to_json()
|
|
}, status=201)
|
|
|
|
# Create firmware
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_firmware(product_id: int, version: str):
|
|
|
|
# Input validation
|
|
if not version or not isinstance(version, str):
|
|
raise ValidationError("Ungültige Firmware-Version")
|
|
|
|
# Version must not be empty
|
|
version = version.strip()
|
|
if not version:
|
|
raise ValidationError("Firmware-Version darf nicht leer sein")
|
|
|
|
# Check product
|
|
product = Product.objects.get(id=product_id)
|
|
|
|
# Check for duplicate
|
|
if Firmware.objects.filter(
|
|
product=product,
|
|
version=version
|
|
).exists():
|
|
raise ValidationError(f"Firmware-Version '{version}' existiert bereits für dieses Produkt")
|
|
|
|
# Create firmware
|
|
firmware = Firmware.objects.create(
|
|
product=product,
|
|
version=version,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Firmware erfolgreich erstellt: Version {version} (ID: {firmware.id}, Produkt: {product.product_name})")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Firmware wurde erfolgreich erstellt',
|
|
'data': firmware.to_json()
|
|
}, status=201)
|
|
|
|
#Create device
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_new_device(manufacturer_id:int, device_type_id: int, device_name: str, serial: str, comment: str = None):
|
|
|
|
# Input validation
|
|
if not manufacturer_id or not isinstance(manufacturer_id, int):
|
|
raise ValidationError("Ungültige Manufacturer ID")
|
|
|
|
# Input validation
|
|
if not device_type_id or not isinstance(device_type_id, int):
|
|
raise ValidationError("Ungültige Device ID")
|
|
|
|
# Input validation
|
|
if not serial or not isinstance(serial, str):
|
|
raise ValidationError("Ungültige Seriennummer")
|
|
|
|
# Serial number must not be empty
|
|
serial = serial.strip()
|
|
if not serial:
|
|
raise ValidationError("Seriennummer darf nicht leer sein")
|
|
|
|
# Validate device name
|
|
if not device_name or not isinstance(device_name, str):
|
|
raise ValidationError("Ungültiger Gerätename")
|
|
|
|
device_name = device_name.strip()
|
|
if not device_name:
|
|
raise ValidationError("Gerätename darf nicht leer sein")
|
|
|
|
|
|
manufacturer = Manufacturer.objects.get(id=manufacturer_id)
|
|
device_type = DeviceType.objects.get(id=device_type_id)
|
|
|
|
# Find or create product
|
|
try:
|
|
product = Product.objects.get(
|
|
product_name=device_name,
|
|
manufacturer=manufacturer,
|
|
device_type=device_type
|
|
)
|
|
logging.info(f"Existierendes Produkt gefunden: {product.product_name}")
|
|
except Product.DoesNotExist:
|
|
product = Product.objects.create(
|
|
product_name=device_name,
|
|
manufacturer=manufacturer,
|
|
device_type=device_type
|
|
)
|
|
logging.info(f"Neues Produkt erstellt: {product.product_name}")
|
|
|
|
# Check for duplicate
|
|
if Device.objects.filter(
|
|
product=product,
|
|
serial=serial
|
|
).exists():
|
|
raise ValidationError(f"Gerät mit Seriennummer '{serial}' existiert bereits für dieses Produkt")
|
|
|
|
# Create device
|
|
device = Device.objects.create(
|
|
product=product,
|
|
serial=serial,
|
|
comment=comment if comment else '',
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Gerät erfolgreich erstellt: Seriennummer {serial} (ID: {device.id}, Produkt: {product.product_name})")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Gerät wurde erfolgreich erstellt',
|
|
'data': {
|
|
'device': device.to_json(),
|
|
'product_id': product.id
|
|
}
|
|
}, status=201)
|
|
|
|
# Create file
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_file(device_id: int, firmware_id: int, filename: str, ip_address: str, status: str, upload_time: datetime = None):
|
|
|
|
# Input validation
|
|
if not filename or not isinstance(filename, str):
|
|
raise ValidationError("Ungültiger Dateiname")
|
|
|
|
# Filename must not be empty
|
|
filename = filename.strip()
|
|
if not filename:
|
|
raise ValidationError("Dateiname darf nicht leer sein")
|
|
|
|
if not ip_address or not isinstance(ip_address, str):
|
|
raise ValidationError("Ungültige IP-Adresse")
|
|
|
|
# IP address must not be empty
|
|
ip_address = ip_address.strip()
|
|
if not ip_address:
|
|
raise ValidationError("IP-Adresse darf nicht leer sein")
|
|
|
|
if not status or not isinstance(status, str):
|
|
raise ValidationError("Ungültiger Status")
|
|
|
|
if not re.match(r'^(\d{1,3}\.){3}\d{1,3}$', ip_address):
|
|
raise ValidationError("Ungültige IP-Adresse")
|
|
|
|
# Check device
|
|
device = Device.objects.get(id=device_id) # Raises ObjectDoesNotExist
|
|
|
|
# Check firmware
|
|
firmware = Firmware.objects.get(id=firmware_id) # Raises ObjectDoesNotExist
|
|
|
|
# Check for duplicate
|
|
if File.objects.filter(
|
|
device=device,
|
|
firmware=firmware,
|
|
filename=filename,
|
|
ip_address=ip_address
|
|
).exists():
|
|
raise ValidationError(f"Datei '{filename}' existiert bereits für dieses Gerät und diese Firmware")
|
|
|
|
# Create file
|
|
file = File.objects.create(
|
|
device=device,
|
|
firmware=firmware,
|
|
filename=filename,
|
|
ip_address=ip_address,
|
|
status=status,
|
|
upload_time=upload_time if upload_time else datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Datei erfolgreich erstellt: {filename} (ID: {file.id}, Gerät: {device.serial}, Firmware: {firmware.version})")
|
|
|
|
#Link firmware with device
|
|
create_firmware_to_device(device_id, firmware_id)
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Datei wurde erfolgreich erstellt',
|
|
'data': file.to_json()
|
|
}, status=201)
|
|
|
|
# Create file
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_firmware_to_device(device_id: int, firmware_id: int):
|
|
|
|
# Check device
|
|
device = Device.objects.get(id=device_id) # Raises ObjectDoesNotExist
|
|
|
|
# Check device
|
|
firmware = Firmware.objects.get(id=firmware_id) # Raises ObjectDoesNotExist
|
|
|
|
# Check for duplicate
|
|
if FirmwareDevices.objects.filter(
|
|
device=device_id,
|
|
firmware=firmware_id,
|
|
).exists():
|
|
raise ValidationError(f"Firmware '{firmware}' existiert bereits!")
|
|
|
|
# Create file
|
|
firmware_device=FirmwareDevices.objects.create(
|
|
device=device,
|
|
firmware=firmware,
|
|
created=datetime.now())
|
|
|
|
logging.info(
|
|
f"Firmware erfolgreich mit device verknüpft: {firmware} (Gerät: {device})")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Firmware wurde erfolgreich verknüpft',
|
|
'data': firmware_device.to_json()
|
|
}, status=201)
|
|
|
|
|
|
# REPLACE the existing create_file_with_firmware_version function in db_create.py
|
|
# with this extended version:
|
|
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_file_with_firmware_version(device_id: int, firmware_version: str, filename: str, ip_address: str,
|
|
status: str):
|
|
# Input validation (stays the same)
|
|
if not firmware_version or not isinstance(firmware_version, str):
|
|
raise ValidationError("Ungültige Firmware-Version")
|
|
firmware_version = firmware_version.strip()
|
|
if not firmware_version:
|
|
raise ValidationError("Firmware-Version darf nicht leer sein")
|
|
|
|
if not filename or not isinstance(filename, str):
|
|
raise ValidationError("Ungültiger Dateiname")
|
|
filename = filename.strip()
|
|
if not filename:
|
|
raise ValidationError("Dateiname darf nicht leer sein")
|
|
|
|
if not ip_address or not isinstance(ip_address, str):
|
|
raise ValidationError("Ungültige IP-Adresse")
|
|
ip_address = ip_address.strip()
|
|
if not ip_address:
|
|
raise ValidationError("IP-Adresse darf nicht leer sein")
|
|
|
|
if not status or not isinstance(status, str):
|
|
raise ValidationError("Ungültiger Status")
|
|
|
|
if not re.match(r'^(\d{1,3}\.){3}\d{1,3}$', ip_address):
|
|
raise ValidationError("Ungültige IP-Adresse")
|
|
|
|
# Get device
|
|
device = Device.objects.get(id=device_id)
|
|
|
|
# Get or create firmware
|
|
try:
|
|
firmware = Firmware.objects.get(version=firmware_version, product=device.product)
|
|
except Firmware.DoesNotExist:
|
|
firmware = Firmware.objects.create(
|
|
version=firmware_version,
|
|
product=device.product
|
|
)
|
|
|
|
# Check for duplicates
|
|
if File.objects.filter(
|
|
device=device,
|
|
firmware=firmware,
|
|
filename=filename,
|
|
ip_address=ip_address
|
|
).exists():
|
|
raise ValidationError(f"Datei '{filename}' existiert bereits für dieses Gerät und diese Firmware")
|
|
|
|
# Create file
|
|
file = File.objects.create(
|
|
device=device,
|
|
firmware=firmware,
|
|
filename=filename,
|
|
ip_address=ip_address,
|
|
status=status,
|
|
upload_time=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Datei erfolgreich erstellt: {filename} (ID: {file.id}, Gerät: {device.serial}, Firmware: {firmware.version})")
|
|
|
|
# Link firmware with device
|
|
create_firmware_to_device(device.id, firmware.id)
|
|
|
|
# Create raw dataset
|
|
response = create_raw_dataset(file.id)
|
|
if not isinstance(response, JsonResponse):
|
|
raise ValueError("Expected JsonResponse from create_raw_dataset")
|
|
|
|
# Decode and parse the JsonResponse content
|
|
raw_dataset_data = json.loads(response.content.decode('utf-8'))
|
|
if raw_dataset_data['status'] != 'success':
|
|
raise ValueError(f"Failed to create raw dataset: {raw_dataset_data.get('message', 'Unknown error')}")
|
|
|
|
raw_dataset_id = raw_dataset_data['data']['id']
|
|
|
|
# *** NEW LOGIC: Distinguish between PCAP and firmware ***
|
|
file_extension = filename.lower().split('.')[-1]
|
|
|
|
if file_extension in ['pcap', 'pcapng', 'cap']:
|
|
# PCAP analysis as before
|
|
analyze_pcap_file(filename, raw_dataset_id)
|
|
elif file_extension == 'bin':
|
|
# Firmware analysis for .bin files
|
|
from .analyze.analyze_firmware import analyze_firmware_file
|
|
analyze_firmware_file(filename, raw_dataset_id)
|
|
else:
|
|
logging.warning(f"Unbekannter Dateityp: {file_extension}. Keine automatische Analyse.")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Datei wurde erfolgreich erstellt',
|
|
'data': file.to_json()
|
|
}, status=201)
|
|
|
|
# Create tool
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_tool(tool_name: str, version: str):
|
|
|
|
# Input validation
|
|
if not tool_name or not isinstance(tool_name, str):
|
|
raise ValidationError("Ungültiger Tool-Name")
|
|
|
|
tool_name = tool_name.strip()
|
|
if not tool_name:
|
|
raise ValidationError("Tool-Name darf nicht leer sein")
|
|
|
|
if not version or not isinstance(version, str):
|
|
raise ValidationError("Ungültige Version")
|
|
|
|
version = version.strip()
|
|
if not version:
|
|
raise ValidationError("Version darf nicht leer sein")
|
|
|
|
# Check for duplicate
|
|
if Tool.objects.filter(
|
|
tool_name=tool_name,
|
|
version=version
|
|
).exists():
|
|
raise ValidationError(f"Tool '{tool_name}' mit Version '{version}' existiert bereits")
|
|
|
|
# Create tool
|
|
tool = Tool.objects.create(
|
|
tool_name=tool_name,
|
|
version=version,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Tool erfolgreich erstellt: {tool_name} Version {version} (ID: {tool.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Tool wurde erfolgreich erstellt',
|
|
'data': tool.to_json()
|
|
}, status=201)
|
|
|
|
# Create raw dataset
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_raw_dataset(file_id: int):
|
|
|
|
# Check file
|
|
file = File.objects.get(id=file_id)
|
|
# Not necessary, since File.objects.get() already raises an ObjectDoesNotExist exception
|
|
"""if not file.exists():
|
|
raise ValidationError(f"File mit Id '{file_id}' existiert nicht")"""
|
|
|
|
# Create raw dataset
|
|
raw_dataset = RawDataset.objects.create(
|
|
file=file,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Raw Dataset erfolgreich erstellt für File: {file.filename} (ID: {raw_dataset.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Raw Dataset wurde erfolgreich erstellt',
|
|
'data': raw_dataset.to_json()
|
|
}, status=201)
|
|
|
|
# Create raw_url entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_raw_url(raw_dataset_id: int, url: str):
|
|
|
|
# Input validation
|
|
if not url or not isinstance(url, str):
|
|
raise ValidationError("Ungültige URL")
|
|
|
|
url = url.strip()
|
|
if not url:
|
|
raise ValidationError("URL darf nicht leer sein")
|
|
|
|
# Use raw dataset ID directly
|
|
raw_url = UrlRaw.objects.create(
|
|
raw_dataset_id=raw_dataset_id,
|
|
url=url,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Raw URL erfolgreich erstellt: {url} (ID: {raw_url.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Raw URL wurde erfolgreich erstellt',
|
|
'data': raw_url.to_json()
|
|
}, status=201)
|
|
|
|
# Create raw_dns entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_raw_dns(raw_dataset_id: int, dns_hostname: str):
|
|
|
|
# Input validation
|
|
if not dns_hostname or not isinstance(dns_hostname, str):
|
|
raise ValidationError("Ungültiger DNS Hostname")
|
|
dns_hostname = dns_hostname.strip()
|
|
if not dns_hostname:
|
|
raise ValidationError("DNS Hostname darf nicht leer sein")
|
|
|
|
# Check for duplicate by ID
|
|
if DnsRaw.objects.filter(
|
|
raw_dataset_id=raw_dataset_id,
|
|
dns_hostname=dns_hostname
|
|
).exists():
|
|
raise ValidationError(f"DNS Hostname '{dns_hostname}' existiert bereits")
|
|
|
|
# Create raw DNS with ID
|
|
raw_dns = DnsRaw.objects.create(
|
|
raw_dataset_id=raw_dataset_id,
|
|
dns_hostname=dns_hostname,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Raw DNS erfolgreich erstellt: {dns_hostname} (ID: {raw_dns.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Raw DNS wurde erfolgreich erstellt',
|
|
'data': raw_dns.to_json()
|
|
}, status=201)
|
|
|
|
# Create raw_hash entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_raw_hash(raw_dataset_id: int, hash_value: str):
|
|
|
|
# Input validation
|
|
if not hash_value or not isinstance(hash_value, str):
|
|
raise ValidationError("Ungültiger Hash-Wert")
|
|
|
|
hash_value = hash_value.strip()
|
|
if not hash_value:
|
|
raise ValidationError("Hash-Wert darf nicht leer sein")
|
|
|
|
# Use raw dataset ID directly
|
|
raw_hash = HashRaw.objects.create(
|
|
raw_dataset_id=raw_dataset_id,
|
|
hash_value=hash_value,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Raw Hash erfolgreich erstellt: {hash_value} (ID: {raw_hash.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Raw Hash wurde erfolgreich erstellt',
|
|
'data': raw_hash.to_json()
|
|
}, status=201)
|
|
|
|
# Create analyzed dataset
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_analyzed_dataset(file_id: int):
|
|
|
|
# Check file
|
|
file = File.objects.get(id=file_id)
|
|
|
|
# Create analyzed dataset
|
|
analyzed_dataset = AnalyzedDataset.objects.create(
|
|
file=file,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Analyzed Dataset erfolgreich erstellt für File: {file.filename}")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Analyzed Dataset wurde erfolgreich erstellt',
|
|
'data': analyzed_dataset.to_json()
|
|
}, status=201)
|
|
|
|
# Analyzed Data - Create port
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_port(port_number: int):
|
|
# Input validation
|
|
if not isinstance(port_number, int):
|
|
raise ValidationError("Port muss eine Ganzzahl sein")
|
|
|
|
# Create port or return existing one
|
|
port, created = Port.objects.get_or_create(
|
|
port=port_number, # Note: field name is 'port', not 'port_number'
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
|
|
logging.info(f"Port {'erstellt' if created else 'gefunden'}: {port_number}")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': f"Port wurde {'erstellt' if created else 'gefunden'}",
|
|
'data': port.to_json()
|
|
}, status=201 if created else 200)
|
|
|
|
# Analyzed Data - Create TLS version
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_tls_version(version: str):
|
|
# Input validation
|
|
if not version or not isinstance(version, str):
|
|
raise ValidationError("Ungültige TLS-Version")
|
|
|
|
version = version.strip()
|
|
if not version:
|
|
raise ValidationError("TLS-Version darf nicht leer sein")
|
|
|
|
# Create TLS version or return existing one
|
|
tls_version, created = TlsVersion.objects.get_or_create(
|
|
tls_version=version, # Note: field name is 'tls_version'
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
|
|
logging.info(f"TLS-Version {'erstellt' if created else 'gefunden'}: {version}")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': f"TLS-Version wurde {'erstellt' if created else 'gefunden'}",
|
|
'data': tls_version.to_json()
|
|
}, status=201 if created else 200)
|
|
|
|
# Analyzed Data - Create cipher suite
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_cipher_suite(name: str):
|
|
# Input validation
|
|
if not name or not isinstance(name, str):
|
|
raise ValidationError("Ungültige Cipher Suite")
|
|
|
|
name = name.strip()
|
|
if not name:
|
|
raise ValidationError("Cipher Suite darf nicht leer sein")
|
|
|
|
# Create cipher suite or return existing one
|
|
cipher_suite, created = CipherSuite.objects.get_or_create(
|
|
cipher_suite=name, # Note: field name is 'cipher_suite'
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
|
|
logging.info(f"Cipher Suite {'erstellt' if created else 'gefunden'}: {name}")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': f"Cipher Suite wurde {'erstellt' if created else 'gefunden'}",
|
|
'data': cipher_suite.to_json()
|
|
}, status=201 if created else 200)
|
|
|
|
# Analyzed Data - Add data (ports, TLS version and cipher suites) to analyzed_URL
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def add_features_to_url(url_id: int, ports: list = None, tls_versions: list = None, cipher_suites: list = None):
|
|
url = UrlAnalyzed.objects.get(id=url_id)
|
|
|
|
if ports:
|
|
for port_number in ports:
|
|
port, _ = Port.objects.get_or_create(
|
|
port=port_number,
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
url.ports.add(port)
|
|
|
|
if tls_versions:
|
|
for version in tls_versions:
|
|
tls, _ = TlsVersion.objects.get_or_create(
|
|
tls_version=version,
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
url.tls_versions.add(tls)
|
|
|
|
if cipher_suites:
|
|
for suite in cipher_suites:
|
|
cipher, _ = CipherSuite.objects.get_or_create(
|
|
cipher_suite=suite,
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
url.cipher_suites.add(cipher)
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Features wurden zur URL hinzugefügt',
|
|
'data': url.to_json()
|
|
}, status=201)
|
|
|
|
# Create analyzed_url entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_analyzed_url(analyzed_dataset_id: int, tool_id: int, url: str,
|
|
ports: list = None, tls_versions: list = None, cipher_suites: list = None):
|
|
# Input validation
|
|
if not url or not isinstance(url, str):
|
|
raise ValidationError("Ungültige URL")
|
|
|
|
url = url.strip()
|
|
if not url:
|
|
raise ValidationError("URL darf nicht leer sein")
|
|
|
|
# Check for duplicate
|
|
if UrlAnalyzed.objects.filter(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
url=url
|
|
).exists():
|
|
raise ValidationError(f"URL '{url}' existiert bereits für dieses Analyzed Dataset")
|
|
|
|
# Create URL
|
|
analyzed_url = UrlAnalyzed.objects.create(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
tool_id=tool_id,
|
|
url=url,
|
|
created=datetime.now()
|
|
)
|
|
|
|
# Add features
|
|
if any([ports, tls_versions, cipher_suites]):
|
|
add_features_to_url(analyzed_url.id, ports, tls_versions, cipher_suites)
|
|
|
|
logging.info(f"Analyzed URL erfolgreich erstellt: {url} (ID: {analyzed_url.id})")
|
|
|
|
# Return updated URL
|
|
analyzed_url.refresh_from_db()
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Analyzed URL wurde erfolgreich erstellt',
|
|
'data': analyzed_url.to_json()
|
|
}, status=201)
|
|
|
|
# Create analyzed_dns entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_analyzed_dns(analyzed_dataset_id: int, tool_id: int, dns_hostname: str, record_type: str):
|
|
|
|
# Input validation
|
|
if not dns_hostname or not isinstance(dns_hostname, str):
|
|
raise ValidationError("Ungültiger DNS Hostname")
|
|
dns_hostname = dns_hostname.strip()
|
|
if not dns_hostname:
|
|
raise ValidationError("DNS Hostname darf nicht leer sein")
|
|
|
|
if not record_type or not isinstance(record_type, str):
|
|
raise ValidationError("Ungültiger Record Type")
|
|
record_type = record_type.strip()
|
|
if not record_type:
|
|
raise ValidationError("Record Type darf nicht leer sein")
|
|
|
|
# Check for duplicate by IDs
|
|
if DnsAnalyzed.objects.filter(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
dns_hostname=dns_hostname,
|
|
record_type=record_type
|
|
).exists():
|
|
raise ValidationError(
|
|
f"DNS '{dns_hostname}' mit Record Type '{record_type}' existiert bereits für dieses Analyzed Dataset")
|
|
|
|
# Create analyzed DNS with IDs
|
|
analyzed_dns = DnsAnalyzed.objects.create(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
tool_id=tool_id,
|
|
dns_hostname=dns_hostname,
|
|
record_type=record_type,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(
|
|
f"Analyzed DNS erfolgreich erstellt: {dns_hostname} (ID: {analyzed_dns.id}, Analyzed Dataset ID: {analyzed_dataset_id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Analyzed DNS wurde erfolgreich erstellt',
|
|
'data': analyzed_dns.to_json()
|
|
}, status=201)
|
|
|
|
# Create analyzed_hash entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_analyzed_hash(analyzed_dataset_id: int, tool_id: int, hash_type: str, hash_value: str, cracked: bool = False,
|
|
cracked_value: str = None, crack_time: str = None):
|
|
# Input validation
|
|
if not hash_type or not isinstance(hash_type, str):
|
|
raise ValidationError("Ungültiger Hash-Typ")
|
|
hash_type = hash_type.strip()
|
|
if not hash_type:
|
|
raise ValidationError("Hash-Typ darf nicht leer sein")
|
|
|
|
if not hash_value or not isinstance(hash_value, str):
|
|
raise ValidationError("Ungültiger Hash-Wert")
|
|
hash_value = hash_value.strip()
|
|
if not hash_value:
|
|
raise ValidationError("Hash-Wert darf nicht leer sein")
|
|
|
|
if cracked and not cracked_value:
|
|
raise ValidationError("Bei gecracktem Hash muss ein Wert angegeben werden")
|
|
|
|
# Check for duplicate or fetch existing object for update
|
|
existing_hash = HashAnalyzed.objects.filter(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
hash_value=hash_value
|
|
).first()
|
|
|
|
if existing_hash:
|
|
# Update existing hash
|
|
if cracked:
|
|
existing_hash.cracked = True
|
|
existing_hash.cracked_value = cracked_value
|
|
existing_hash.crack_time = crack_time
|
|
existing_hash.save()
|
|
|
|
logging.info(f"Analyzed Hash aktualisiert: {hash_value} (ID: {existing_hash.id})")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Analyzed Hash wurde erfolgreich aktualisiert',
|
|
'data': existing_hash.to_json()
|
|
}, status=200)
|
|
|
|
# Create HashType or find existing one
|
|
hash_type_obj, created = HashType.objects.get_or_create(
|
|
hash_type=hash_type,
|
|
defaults={'created': datetime.now()}
|
|
)
|
|
|
|
# Create analyzed hash
|
|
analyzed_hash = HashAnalyzed.objects.create(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
tool_id=tool_id,
|
|
hash_value=hash_value,
|
|
cracked=cracked,
|
|
cracked_value=cracked_value if cracked else None,
|
|
crack_time=crack_time,
|
|
created=datetime.now()
|
|
)
|
|
|
|
# Add HashType to the many-to-many relationship
|
|
analyzed_hash.hash_type.add(hash_type_obj)
|
|
|
|
logging.info(
|
|
f"Analyzed Hash erfolgreich erstellt: {hash_value} (ID: {analyzed_hash.id}, "
|
|
f"Hash-Typ: {hash_type} ({hash_type_obj.id}), "
|
|
f"Analyzed Dataset ID: {analyzed_dataset_id})")
|
|
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Analyzed Hash wurde erfolgreich erstellt',
|
|
'data': analyzed_hash.to_json()
|
|
}, status=201)
|
|
|
|
|
|
### Firmware analysis CREATE functions ###
|
|
|
|
# Create BinwalkRaw entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_binwalk_raw(raw_dataset_id: int, offset: str, description: str, file_path: str = None):
|
|
"""
|
|
Creates a Binwalk raw data entry.
|
|
"""
|
|
# Input validation
|
|
if not offset or not isinstance(offset, str):
|
|
raise ValidationError("Ungültiger Offset")
|
|
|
|
if not description or not isinstance(description, str):
|
|
raise ValidationError("Ungültige Beschreibung")
|
|
|
|
offset = offset.strip()
|
|
description = description.strip()
|
|
|
|
if not offset or not description:
|
|
raise ValidationError("Offset und Beschreibung dürfen nicht leer sein")
|
|
|
|
# Create Binwalk Raw
|
|
binwalk_raw = BinwalkRaw.objects.create(
|
|
raw_dataset_id=raw_dataset_id,
|
|
offset=offset,
|
|
description=description,
|
|
file_path=file_path,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Binwalk Raw erfolgreich erstellt: {offset} (ID: {binwalk_raw.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Binwalk Raw wurde erfolgreich erstellt',
|
|
'data': binwalk_raw.to_json()
|
|
}, status=201)
|
|
|
|
|
|
# Create StringsRaw entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_strings_raw(raw_dataset_id: int, string_value: str, string_type: str = 'generic'):
|
|
"""
|
|
Creates a Strings raw data entry.
|
|
"""
|
|
# Input validation
|
|
if not string_value or not isinstance(string_value, str):
|
|
raise ValidationError("Ungültiger String-Wert")
|
|
|
|
string_value = string_value.strip()
|
|
if not string_value:
|
|
raise ValidationError("String-Wert darf nicht leer sein")
|
|
|
|
if len(string_value) > 1000:
|
|
raise ValidationError("String-Wert zu lang (max 1000 Zeichen)")
|
|
|
|
# Create Strings Raw
|
|
strings_raw = StringsRaw.objects.create(
|
|
raw_dataset_id=raw_dataset_id,
|
|
string_value=string_value,
|
|
string_type=string_type,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Strings Raw erfolgreich erstellt: {string_type} (ID: {strings_raw.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Strings Raw wurde erfolgreich erstellt',
|
|
'data': strings_raw.to_json()
|
|
}, status=201)
|
|
|
|
|
|
# Create BinwalkAnalyzed entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_binwalk_analyzed(analyzed_dataset_id: int, tool_id: int, component_type: str,
|
|
offset: str, size: str = None, entropy: float = None):
|
|
"""
|
|
Creates an analyzed Binwalk entry.
|
|
"""
|
|
# Input validation
|
|
if not component_type or not isinstance(component_type, str):
|
|
raise ValidationError("Ungültiger Komponententyp")
|
|
|
|
if not offset or not isinstance(offset, str):
|
|
raise ValidationError("Ungültiger Offset")
|
|
|
|
component_type = component_type.strip()
|
|
offset = offset.strip()
|
|
|
|
if not component_type or not offset:
|
|
raise ValidationError("Komponententyp und Offset dürfen nicht leer sein")
|
|
|
|
# Create Binwalk Analyzed
|
|
binwalk_analyzed = BinwalkAnalyzed.objects.create(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
tool_id=tool_id,
|
|
component_type=component_type,
|
|
offset=offset,
|
|
size=size,
|
|
entropy=entropy,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Binwalk Analyzed erfolgreich erstellt: {component_type} (ID: {binwalk_analyzed.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Binwalk Analyzed wurde erfolgreich erstellt',
|
|
'data': binwalk_analyzed.to_json()
|
|
}, status=201)
|
|
|
|
|
|
# Create StringsAnalyzed entry
|
|
@handle_exceptions
|
|
@transaction.atomic
|
|
def create_strings_analyzed(analyzed_dataset_id: int, tool_id: int, string_value: str,
|
|
category: str, severity: str = 'info'):
|
|
"""
|
|
Creates an analyzed Strings entry.
|
|
"""
|
|
# Input validation
|
|
if not string_value or not isinstance(string_value, str):
|
|
raise ValidationError("Ungültiger String-Wert")
|
|
|
|
if not category or not isinstance(category, str):
|
|
raise ValidationError("Ungültige Kategorie")
|
|
|
|
string_value = string_value.strip()
|
|
category = category.strip()
|
|
|
|
if not string_value or not category:
|
|
raise ValidationError("String-Wert und Kategorie dürfen nicht leer sein")
|
|
|
|
# Validate severity
|
|
valid_severities = ['critical', 'high', 'medium', 'low', 'info']
|
|
if severity not in valid_severities:
|
|
raise ValidationError(f"Ungültiger Severity-Wert. Erlaubt: {', '.join(valid_severities)}")
|
|
|
|
# Create Strings Analyzed
|
|
strings_analyzed = StringsAnalyzed.objects.create(
|
|
analyzed_dataset_id=analyzed_dataset_id,
|
|
tool_id=tool_id,
|
|
string_value=string_value[:1000], # Limit 1000 characters
|
|
category=category,
|
|
severity=severity,
|
|
created=datetime.now()
|
|
)
|
|
|
|
logging.info(f"Strings Analyzed erfolgreich erstellt: {category} (ID: {strings_analyzed.id})")
|
|
return JsonResponse({
|
|
'status': 'success',
|
|
'message': 'Strings Analyzed wurde erfolgreich erstellt',
|
|
'data': strings_analyzed.to_json()
|
|
}, status=201)
|
|
|
|
# Add import statements at the top of db_create.py:
|
|
# from .models import BinwalkRaw, StringsRaw, BinwalkAnalyzed, StringsAnalyzed
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|