482 lines
16 KiB
Python
482 lines
16 KiB
Python
"""REST endpoints exposed by IoT PreTester 4.6 Demo."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import ObjectDoesNotExist, ValidationError
|
|
from django.core.files.storage import FileSystemStorage
|
|
from django.http import HttpResponse
|
|
from django.db.models.deletion import ProtectedError
|
|
from django.shortcuts import get_object_or_404
|
|
from rest_framework import serializers, status
|
|
from rest_framework.permissions import AllowAny, IsAdminUser
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from .analyze.security_analysis import analysis_log_path, collect_cloud_results
|
|
from .db_create import (
|
|
create_capture,
|
|
create_device_type,
|
|
create_manufacturer,
|
|
create_new_device,
|
|
)
|
|
from .db_read import (
|
|
get_all_device_types,
|
|
get_all_devices,
|
|
get_all_manufacturers,
|
|
get_device_by_id,
|
|
get_files_by_device,
|
|
)
|
|
from .models import (
|
|
DeepAnalysisJob,
|
|
Device,
|
|
DeviceType,
|
|
File,
|
|
FindingReport,
|
|
Manufacturer,
|
|
)
|
|
from .user_funcs import (
|
|
delete_user,
|
|
list_users,
|
|
login_user,
|
|
logout_user,
|
|
register_user,
|
|
set_password,
|
|
)
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _validation_error(exc: ValidationError) -> Response:
|
|
messages = getattr(exc, "messages", [str(exc)])
|
|
return Response({"error": messages[0]}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
def _tool_version(name: str, args=("-v",)) -> str | None:
|
|
"""First line of a tool's version output, or None when it is missing."""
|
|
path = shutil.which(name)
|
|
if not path:
|
|
return None
|
|
try:
|
|
out = subprocess.run([path, *args], capture_output=True, text=True, timeout=15)
|
|
first = (out.stdout or out.stderr or "").strip().splitlines()
|
|
return first[0][:120] if first else path
|
|
except Exception as exc:
|
|
return f"{path} (version check failed: {exc})"
|
|
|
|
|
|
class HealthView(APIView):
|
|
"""Service status including the external tools the analysis depends on."""
|
|
|
|
permission_classes = [AllowAny]
|
|
|
|
def get(self, request):
|
|
from analyzer import __version__ as analyzer_version
|
|
|
|
tshark = _tool_version(os.environ.get("TSHARK_PATH") or "tshark")
|
|
upload_dir = str(settings.CAPTURE_UPLOAD_DIR)
|
|
data = {
|
|
"status": "ok" if tshark else "degraded",
|
|
"service": "IoT PreTester 4.6 Demo",
|
|
"analyzer_version": analyzer_version,
|
|
"tools": {
|
|
"tshark": tshark,
|
|
"hashcat": _tool_version("hashcat", ("--version",)),
|
|
"nmap": _tool_version("nmap", ("--version",)),
|
|
},
|
|
"upload_dir": upload_dir,
|
|
"upload_dir_writable": os.access(upload_dir, os.W_OK),
|
|
"log_dir": str(getattr(settings, "LOG_DIR", "")),
|
|
"log_level": getattr(settings, "LOG_LEVEL", "INFO"),
|
|
"packet_limit": settings.CAPTURE_PACKET_LIMIT or "unlimited",
|
|
}
|
|
if not tshark:
|
|
data["error"] = "tshark is not installed; capture analysis cannot run."
|
|
logger.error("Health check: tshark not found on PATH")
|
|
return Response(data)
|
|
|
|
|
|
class LoginView(APIView):
|
|
permission_classes = [AllowAny]
|
|
|
|
def post(self, request):
|
|
username = request.data.get("username")
|
|
password = request.data.get("password")
|
|
if not username or not password:
|
|
return Response(
|
|
[False, {"message": "Username and password are required."}],
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
return Response(login_user(request, username, password))
|
|
|
|
|
|
class LogoutView(APIView):
|
|
def post(self, request):
|
|
return Response(logout_user(request))
|
|
|
|
|
|
class UserSerializer(serializers.Serializer):
|
|
username = serializers.CharField(max_length=150)
|
|
password = serializers.CharField(write_only=True)
|
|
role = serializers.ChoiceField(choices=["admin", "user"])
|
|
|
|
|
|
class RegisterUserView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
serializer = UserSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
result = register_user(**serializer.validated_data)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
return Response(result, status=status.HTTP_201_CREATED)
|
|
|
|
|
|
class ListUsersView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def get(self, request):
|
|
return Response(list_users())
|
|
|
|
|
|
class DeleteUserView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
try:
|
|
user_id = int(request.data.get("id"))
|
|
return Response(delete_user(request, user_id))
|
|
except (TypeError, ValueError):
|
|
return Response({"error": "A valid user ID is required."}, status=400)
|
|
except ObjectDoesNotExist:
|
|
return Response({"error": "The user does not exist."}, status=404)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
|
|
|
|
class ResetPasswordView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
user_id = request.data.get("user_id")
|
|
password = request.data.get("new_password")
|
|
if not isinstance(password, str) or not password:
|
|
return Response({"error": "A new password is required."}, status=400)
|
|
try:
|
|
return Response(set_password(int(user_id), password))
|
|
except (TypeError, ValueError):
|
|
return Response({"error": "A valid user ID is required."}, status=400)
|
|
except ObjectDoesNotExist:
|
|
return Response({"error": "The user does not exist."}, status=404)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
|
|
|
|
class CreateManufacturerView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
try:
|
|
item = create_manufacturer(request.data.get("name"))
|
|
return Response(item.to_json(), status=status.HTTP_201_CREATED)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
|
|
|
|
class CreateDeviceTypeView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
try:
|
|
item = create_device_type(request.data.get("device_type"))
|
|
return Response(item.to_json(), status=status.HTTP_201_CREATED)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
|
|
|
|
class DeviceSerializer(serializers.Serializer):
|
|
device_type_id = serializers.IntegerField()
|
|
device_name = serializers.CharField(max_length=100)
|
|
serial = serializers.CharField(max_length=100)
|
|
manufacturer_id = serializers.IntegerField()
|
|
comment = serializers.CharField(max_length=255, required=False, allow_blank=True)
|
|
|
|
|
|
class CreateDeviceView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def post(self, request):
|
|
serializer = DeviceSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
device = create_new_device(**serializer.validated_data)
|
|
return Response(device.to_json(), status=status.HTTP_201_CREATED)
|
|
except ObjectDoesNotExist:
|
|
return Response({"error": "The manufacturer or device type does not exist."}, status=400)
|
|
except ValidationError as exc:
|
|
return _validation_error(exc)
|
|
|
|
|
|
class CaptureSerializer(serializers.Serializer):
|
|
capture = serializers.FileField(write_only=True)
|
|
device_id = serializers.IntegerField()
|
|
firmware_version = serializers.CharField(max_length=100)
|
|
ip_address = serializers.IPAddressField()
|
|
|
|
def validate_capture(self, capture):
|
|
if capture.size <= 0:
|
|
raise serializers.ValidationError("The capture is empty.")
|
|
if capture.size > 512 * 1024 * 1024:
|
|
raise serializers.ValidationError("The capture exceeds the 512 MB demo limit.")
|
|
if Path(capture.name).suffix.lower() not in {".pcap", ".pcapng"}:
|
|
raise serializers.ValidationError("Only PCAP and PCAPNG captures are supported.")
|
|
return capture
|
|
|
|
|
|
class CreateCaptureView(APIView):
|
|
def post(self, request):
|
|
serializer = CaptureSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
values = dict(serializer.validated_data)
|
|
uploaded_capture = values.pop("capture")
|
|
suffix = Path(uploaded_capture.name).suffix.lower()
|
|
stem = re.sub(r"[^A-Za-z0-9._-]+", "_", Path(uploaded_capture.name).stem)[:80]
|
|
stored_name = f"{stem or 'capture'}-{uuid4().hex[:12]}{suffix}"
|
|
storage = FileSystemStorage(location=settings.CAPTURE_UPLOAD_DIR)
|
|
stored_name = storage.save(stored_name, uploaded_capture)
|
|
try:
|
|
capture = create_capture(filename=stored_name, **values)
|
|
except ObjectDoesNotExist:
|
|
storage.delete(stored_name)
|
|
return Response({"error": "The selected device does not exist."}, status=400)
|
|
except ValidationError as exc:
|
|
storage.delete(stored_name)
|
|
return _validation_error(exc)
|
|
except Exception:
|
|
storage.delete(stored_name)
|
|
raise
|
|
return Response(
|
|
{"status": "success", "data": capture.to_json()},
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
|
|
|
|
class ManufacturersView(APIView):
|
|
def get(self, request):
|
|
return Response(get_all_manufacturers())
|
|
|
|
|
|
class DeviceTypesView(APIView):
|
|
def get(self, request):
|
|
return Response(get_all_device_types())
|
|
|
|
|
|
class DevicesView(APIView):
|
|
def get(self, request):
|
|
return Response(get_all_devices())
|
|
|
|
|
|
class DeviceDetailView(APIView):
|
|
def get(self, request, device_id):
|
|
get_object_or_404(Device, id=device_id)
|
|
return Response(get_device_by_id(device_id))
|
|
|
|
|
|
class DeviceCapturesView(APIView):
|
|
def get(self, request, device_id):
|
|
get_object_or_404(Device, id=device_id)
|
|
return Response(get_files_by_device(device_id))
|
|
|
|
|
|
class DeleteDeviceView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def delete(self, request, device_id):
|
|
device = get_object_or_404(Device, id=device_id)
|
|
device.delete()
|
|
return Response({"status": "success"})
|
|
|
|
|
|
class DeleteManufacturerView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def delete(self, request, manufacturer_id):
|
|
item = get_object_or_404(Manufacturer, id=manufacturer_id)
|
|
try:
|
|
item.delete()
|
|
except ProtectedError:
|
|
return Response(
|
|
{"error": "The manufacturer is still used by a product."},
|
|
status=status.HTTP_409_CONFLICT,
|
|
)
|
|
return Response({"status": "success"})
|
|
|
|
|
|
class DeleteDeviceTypeView(APIView):
|
|
permission_classes = [IsAdminUser]
|
|
|
|
def delete(self, request, device_type_id):
|
|
item = get_object_or_404(DeviceType, id=device_type_id)
|
|
try:
|
|
item.delete()
|
|
except ProtectedError:
|
|
return Response(
|
|
{"error": "The device type is still used by a product."},
|
|
status=status.HTTP_409_CONFLICT,
|
|
)
|
|
return Response({"status": "success"})
|
|
|
|
|
|
# Capture status -> payload shown while no finding report exists yet.
|
|
_PENDING_CAPTURE_STATES = {
|
|
"analyzing": ("analyzing", "The security analysis is still running."),
|
|
"pending": ("analyzing", "The security analysis is queued."),
|
|
"analysis_failed": ("failed", "The security analysis failed. The error and the analysis log are shown below."),
|
|
}
|
|
|
|
|
|
class FindingsView(APIView):
|
|
def get(self, request, file_id):
|
|
capture = get_object_or_404(File, id=file_id)
|
|
report = FindingReport.objects.filter(file_id=file_id).first()
|
|
if report is None:
|
|
state, message = _PENDING_CAPTURE_STATES.get(
|
|
capture.status,
|
|
("empty", "No security analysis is available for this capture."),
|
|
)
|
|
response = Response(
|
|
{
|
|
"status": state,
|
|
"capture_status": capture.status,
|
|
"message": message,
|
|
"error": capture.error_message,
|
|
"log_available": _log_exists(capture),
|
|
"summary": {
|
|
"total": 0,
|
|
"by_severity": {},
|
|
"by_category": {},
|
|
"highlights": [],
|
|
"top": [],
|
|
},
|
|
"findings": [],
|
|
}
|
|
)
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
payload = json.loads(report.result_json)
|
|
payload["status"] = "ready"
|
|
payload["capture_status"] = capture.status
|
|
payload["error"] = capture.error_message
|
|
payload["log_available"] = _log_exists(capture)
|
|
response = Response(payload)
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
def _log_exists(capture: File) -> bool:
|
|
path = analysis_log_path(capture.filename)
|
|
return bool(path and os.path.exists(path))
|
|
|
|
|
|
class AnalysisLogView(APIView):
|
|
"""Tail of the per-capture analysis log for troubleshooting."""
|
|
|
|
def get(self, request, file_id):
|
|
from analyzer.logutil import tail
|
|
|
|
capture = get_object_or_404(File, id=file_id)
|
|
path = analysis_log_path(capture.filename)
|
|
try:
|
|
lines = max(20, min(int(request.GET.get("lines", 400)), 5000))
|
|
except (TypeError, ValueError):
|
|
lines = 400
|
|
if not path or not os.path.exists(path):
|
|
return Response(
|
|
{"file_id": capture.id, "status": capture.status, "error": capture.error_message,
|
|
"log": "", "available": False,
|
|
"message": "No analysis log exists for this capture yet."},
|
|
)
|
|
response = Response({
|
|
"file_id": capture.id, "status": capture.status, "error": capture.error_message,
|
|
"available": True, "path": os.path.basename(path), "log": tail(path, lines),
|
|
})
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
class CloudResultsView(APIView):
|
|
def get(self, request, file_id):
|
|
get_object_or_404(File, id=file_id)
|
|
response = Response(collect_cloud_results(file_id))
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
class StartAnalysisView(APIView):
|
|
def post(self, request, file_id):
|
|
get_object_or_404(File, id=file_id)
|
|
try:
|
|
url_ids = [int(value) for value in (request.data.get("url_ids") or [])]
|
|
hash_ids = [int(value) for value in (request.data.get("hash_ids") or [])]
|
|
except (TypeError, ValueError):
|
|
return Response({"error": "Artifact IDs must be integers."}, status=400)
|
|
if not url_ids and not hash_ids:
|
|
return Response({"error": "Select at least one artifact."}, status=400)
|
|
|
|
from .analyze.deep_analysis import start_deep_analysis
|
|
|
|
start_deep_analysis(file_id, url_ids=url_ids, hash_ids=hash_ids)
|
|
return Response(
|
|
{"status": "started", "url_count": len(url_ids), "hash_count": len(hash_ids)},
|
|
status=status.HTTP_202_ACCEPTED,
|
|
)
|
|
|
|
|
|
class AnalysisStatusView(APIView):
|
|
def get(self, request, file_id):
|
|
get_object_or_404(File, id=file_id)
|
|
job = DeepAnalysisJob.objects.filter(file_id=file_id).first()
|
|
data = job.to_json() if job else {
|
|
"state": "pending",
|
|
"url_total": 0,
|
|
"url_done": 0,
|
|
"hash_total": 0,
|
|
"hash_done": 0,
|
|
"cracked": 0,
|
|
"progress_done": 0,
|
|
"progress_total": 0,
|
|
"percent": 0,
|
|
"message": "Not started.",
|
|
}
|
|
response = Response(data)
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
class ReportView(APIView):
|
|
def get(self, request, file_id, fmt):
|
|
from analyzer import FORMATS, render, result_from_dict
|
|
|
|
fmt = fmt.lower()
|
|
if fmt not in FORMATS:
|
|
return Response({"error": f"Allowed formats: {', '.join(FORMATS)}"}, status=400)
|
|
report = get_object_or_404(FindingReport, file_id=file_id)
|
|
result = result_from_dict(json.loads(report.result_json))
|
|
content = render(result, fmt, collect_cloud_results(file_id))
|
|
content_types = {"md": "text/markdown", "txt": "text/plain", "json": "application/json"}
|
|
response = HttpResponse(content, content_type=content_types[fmt])
|
|
response["Content-Disposition"] = (
|
|
f'attachment; filename="iot-pretester-4.6-demo-report-{file_id}.{fmt}"'
|
|
)
|
|
return response
|