240 lines
9.1 KiB
Python
240 lines
9.1 KiB
Python
"""Persist results and artifacts produced by the shared analysis context."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
import traceback
|
|
|
|
from django.conf import settings
|
|
from django.db import connection
|
|
from analyzer import AnalysisEngine, extract_urls, extract_hashes, __version__ as analyzer_version
|
|
from analyzer.extract_tshark import build_context_from_pcap
|
|
from analyzer.findings import Severity
|
|
from analyzer import logutil
|
|
from IoTAnalyzer.models import FindingReport, UrlRaw, HashRaw
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def analysis_log_path(filename: str) -> str | None:
|
|
"""Per-capture log written next to the upload."""
|
|
path = resolve_capture_path(filename)
|
|
if not path:
|
|
return None
|
|
return path + ".analysis.log"
|
|
|
|
|
|
def _error_summary(exc: BaseException) -> str:
|
|
"""Short, user-facing description of a failure with the failing location."""
|
|
tb = traceback.extract_tb(exc.__traceback__)
|
|
where = f" ({tb[-1].filename.rsplit('/', 1)[-1]}:{tb[-1].lineno})" if tb else ""
|
|
return f"{type(exc).__name__}: {exc}{where}"[:1000]
|
|
|
|
|
|
def _set_error(file_id: int, message: str | None) -> None:
|
|
from IoTAnalyzer.models import File
|
|
try:
|
|
File.objects.filter(id=file_id).update(error_message=message)
|
|
except Exception as exc:
|
|
logger.warning("Could not store error message for file=%s: %s", file_id, exc)
|
|
|
|
_CANDIDATE_DIRS = [
|
|
str(settings.CAPTURE_UPLOAD_DIR),
|
|
]
|
|
|
|
|
|
def resolve_capture_path(filename: str) -> str | None:
|
|
"""Resolve a stored capture name against supported upload locations."""
|
|
if not filename:
|
|
return None
|
|
if os.path.isabs(filename) and os.path.exists(filename):
|
|
return filename
|
|
for base in _CANDIDATE_DIRS:
|
|
candidate = os.path.join(base, filename)
|
|
if os.path.exists(candidate):
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _max_severity(result) -> str:
|
|
worst = Severity.INFO
|
|
for f in result.findings:
|
|
if f.severity.rank > worst.rank:
|
|
worst = f.severity
|
|
return worst.value
|
|
|
|
|
|
def collect_cloud_results(file_id: int) -> dict:
|
|
"""Return deduplicated capture artifacts and any enrichment results."""
|
|
from IoTAnalyzer.models import UrlRaw, HashRaw, UrlAnalyzed, HashAnalyzed
|
|
|
|
# Retain raw IDs so the frontend can request a selected enrichment subset.
|
|
raw_urls = list(UrlRaw.objects.filter(raw_dataset__file_id=file_id)
|
|
.values_list('id', 'url'))
|
|
raw_hashes = list(HashRaw.objects.filter(raw_dataset__file_id=file_id)
|
|
.values_list('id', 'hash_value'))
|
|
|
|
url_map = {}
|
|
for ua in (UrlAnalyzed.objects.filter(analyzed_dataset__file_id=file_id)
|
|
.prefetch_related('ports', 'tls_versions', 'cipher_suites')):
|
|
url_map[ua.url] = {
|
|
"ports": sorted({p.port for p in ua.ports.all()}),
|
|
"tls_versions": [t.tls_version for t in ua.tls_versions.all()],
|
|
"cipher_suites": [c.cipher_suite for c in ua.cipher_suites.all()],
|
|
}
|
|
|
|
hash_map = {}
|
|
for ha in (HashAnalyzed.objects.filter(analyzed_dataset__file_id=file_id)
|
|
.prefetch_related('hash_type')):
|
|
entry = {
|
|
"cracked": ha.cracked,
|
|
"cracked_value": ha.cracked_value,
|
|
"crack_time": ha.crack_time,
|
|
"hash_type": [h.hash_type for h in ha.hash_type.all()],
|
|
}
|
|
existing = hash_map.get(ha.hash_value)
|
|
if existing is None or (ha.cracked and not existing.get("cracked")):
|
|
hash_map[ha.hash_value] = entry
|
|
|
|
urls, seen = [], set()
|
|
for raw_id, u in raw_urls:
|
|
if not u or u in seen:
|
|
continue
|
|
seen.add(u)
|
|
a = url_map.get(u)
|
|
urls.append({"id": raw_id, "url": u, "analyzed": a is not None, **(a or {})})
|
|
|
|
hashes, seen_h = [], set()
|
|
for raw_id, h in raw_hashes:
|
|
if not h or h in seen_h:
|
|
continue
|
|
seen_h.add(h)
|
|
a = hash_map.get(h)
|
|
hashes.append({"id": raw_id, "hash_value": h, "analyzed": a is not None,
|
|
**(a or {"cracked": False, "cracked_value": None,
|
|
"crack_time": None, "hash_type": []})})
|
|
|
|
return {"urls": urls, "hashes": hashes}
|
|
|
|
|
|
def _persist_artifacts(ctx, raw_dataset_id: int) -> None:
|
|
"""Persist URL and hash artifacts extracted during the analysis pass."""
|
|
if raw_dataset_id is None:
|
|
return
|
|
try:
|
|
for u in extract_urls(ctx):
|
|
UrlRaw.objects.get_or_create(raw_dataset_id=raw_dataset_id, url=u.value[:200])
|
|
except Exception as exc:
|
|
logger.exception("URL persistence failed: %s", exc)
|
|
try:
|
|
for h in extract_hashes(ctx):
|
|
HashRaw.objects.get_or_create(raw_dataset_id=raw_dataset_id, hash_value=h.value)
|
|
except Exception as exc:
|
|
logger.exception("Hash persistence failed: %s", exc)
|
|
|
|
|
|
def analyze_capture(filename: str, file_id: int, raw_dataset_id: int | None = None,
|
|
packet_limit: int | None = None, device_ip: str | None = None):
|
|
"""Analyze a capture once and persist findings plus extracted artifacts.
|
|
|
|
``device_ip`` is the address entered at upload time. It focuses the report on
|
|
the device under test; when omitted it is looked up from the File record.
|
|
Raises on failure so the caller can record the reason.
|
|
"""
|
|
path = resolve_capture_path(filename)
|
|
if not path:
|
|
logger.error("Capture file not found: %s (searched: %s)", filename, _CANDIDATE_DIRS)
|
|
raise FileNotFoundError(f"Capture file {filename} not found in {_CANDIDATE_DIRS}")
|
|
|
|
if packet_limit is None:
|
|
packet_limit = getattr(settings, "CAPTURE_PACKET_LIMIT", 0)
|
|
|
|
if device_ip is None:
|
|
try:
|
|
from IoTAnalyzer.models import File
|
|
device_ip = File.objects.filter(id=file_id).values_list("ip_address", flat=True).first()
|
|
except Exception:
|
|
device_ip = None
|
|
|
|
started = time.time()
|
|
logger.info("Capture analysis start: file_id=%s path=%s device_ip=%s packet_limit=%s analyzer=%s",
|
|
file_id, path, device_ip, packet_limit or "unlimited", analyzer_version)
|
|
ctx = build_context_from_pcap(path, packet_limit=packet_limit or None, device_ip=device_ip)
|
|
if ctx.truncated:
|
|
logger.warning("Capture %s analyzed partially: %s", file_id, ctx.truncation_reason)
|
|
result = AnalysisEngine().analyze_context(ctx)
|
|
|
|
# Reanalysis replaces the previous report atomically at the file level.
|
|
payload = json.dumps(result.to_dict())
|
|
FindingReport.objects.filter(file_id=file_id).delete()
|
|
FindingReport.objects.create(
|
|
file_id=file_id,
|
|
result_json=payload,
|
|
finding_count=len(result.findings),
|
|
max_severity=_max_severity(result),
|
|
)
|
|
logger.info("Stored finding report: file_id=%s findings=%s max_severity=%s json=%d bytes",
|
|
file_id, len(result.findings), _max_severity(result), len(payload))
|
|
|
|
_persist_artifacts(ctx, raw_dataset_id)
|
|
logger.info("Capture analysis finished: file_id=%s in %.1fs", file_id, time.time() - started)
|
|
return result
|
|
|
|
|
|
def run_security_analysis(filename: str, file_id: int):
|
|
try:
|
|
return analyze_capture(filename, file_id, raw_dataset_id=None)
|
|
except Exception as exc:
|
|
logger.exception("Security analysis failed for %s: %s", filename, exc)
|
|
return None
|
|
|
|
|
|
def start_capture_analysis(file_id: int, filename: str, raw_dataset_id: int | None = None,
|
|
synchronous: bool = False) -> None:
|
|
"""Analyze a capture off the request thread and keep File.status current.
|
|
|
|
A multi-hundred-megabyte capture takes far longer than any sane HTTP
|
|
timeout, so the upload response must not wait for the analysis.
|
|
"""
|
|
if synchronous:
|
|
_run_capture_analysis(file_id, filename, raw_dataset_id, close_conn=False)
|
|
return
|
|
thread = threading.Thread(
|
|
target=_run_capture_analysis,
|
|
args=(file_id, filename, raw_dataset_id),
|
|
daemon=True,
|
|
name=f"capture-analysis-{file_id}",
|
|
)
|
|
thread.start()
|
|
|
|
|
|
def _run_capture_analysis(file_id: int, filename: str, raw_dataset_id: int | None = None,
|
|
close_conn: bool = True) -> None:
|
|
from IoTAnalyzer.models import File
|
|
|
|
status = "analysis_failed"
|
|
error = None
|
|
log_path = analysis_log_path(filename)
|
|
try:
|
|
with logutil.run_log(log_path):
|
|
_set_error(file_id, None)
|
|
result = analyze_capture(filename, file_id, raw_dataset_id)
|
|
status = "analyzed" if result is not None else "analysis_failed"
|
|
except Exception as exc:
|
|
error = _error_summary(exc)
|
|
logger.exception("Capture analysis failed for file=%s: %s", file_id, error)
|
|
finally:
|
|
try:
|
|
File.objects.filter(id=file_id).update(status=status, error_message=error)
|
|
logger.info("Capture %s completed with status %s%s", file_id, status,
|
|
f" (log: {log_path})" if log_path else "")
|
|
except Exception as exc:
|
|
logger.exception("Could not persist status for file=%s: %s", file_id, exc)
|
|
# Close only the worker thread's connection, never the request connection.
|
|
if close_conn:
|
|
connection.close()
|