import json import socket import platform import subprocess import ipaddress import threading import logging import requests as http_requests from django.views.generic import TemplateView, CreateView, UpdateView, DeleteView, ListView, DetailView, View from django.shortcuts import get_object_or_404, redirect from django.urls import reverse_lazy, reverse from django.http import JsonResponse from .models import ScanProfile, ScanRun, ScanFinding from .forms import ScanProfileForm from .scanner import run_scan from .notifications import send_test_telegram logger = logging.getLogger(__name__) SEVERITY_ORDER = ['critical', 'warning', 'info', 'ok'] def _worst_severity(summary: dict) -> str: for s in SEVERITY_ORDER: if summary.get(s, 0) > 0: return s return 'ok' def _detect_gateway() -> str | None: try: if platform.system() == 'Linux': r = subprocess.run(['ip', 'route', 'show', 'default'], capture_output=True, text=True, timeout=5) for line in r.stdout.splitlines(): if 'default' in line and 'via' in line: parts = line.split() return parts[parts.index('via') + 1] else: r = subprocess.run(['netstat', '-rn'], capture_output=True, text=True, timeout=5) for line in r.stdout.splitlines(): if line.startswith('default') or line.startswith('0.0.0.0'): parts = line.split() if len(parts) >= 2: return parts[1] except Exception: pass return None def _detect_local_ip() -> str | None: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return None def _detect_subnet_mask(local_ip: str) -> str | None: """Try to get the real subnet mask from the OS, fallback to /24.""" try: if platform.system() == 'Linux': r = subprocess.run(['ip', 'addr', 'show'], capture_output=True, text=True, timeout=5) for line in r.stdout.splitlines(): line = line.strip() if line.startswith('inet ') and local_ip in line: cidr_part = line.split()[1] net = ipaddress.IPv4Network(cidr_part, strict=False) return str(net) else: r = subprocess.run(['ifconfig'], capture_output=True, text=True, timeout=5) lines = r.stdout.splitlines() for i, line in enumerate(lines): if local_ip in line: for detail in lines[i:i + 3]: if 'netmask' in detail.lower(): parts = detail.split() try: mask_idx = [p.lower() for p in parts].index('netmask') mask = parts[mask_idx + 1] # macOS outputs hex netmask like 0xffffff00 if mask.startswith('0x'): mask = socket.inet_ntoa(int(mask, 16).to_bytes(4, 'big')) net = ipaddress.IPv4Network(f'{local_ip}/{mask}', strict=False) return str(net) except (ValueError, IndexError): pass except Exception: pass # fallback to /24 try: net = ipaddress.IPv4Network(f'{local_ip}/24', strict=False) return str(net) except Exception: return None def _detect_public_ip() -> str | None: for url in ['https://api.ipify.org', 'https://icanhazip.com', 'https://checkip.amazonaws.com']: try: resp = http_requests.get(url, timeout=5) ip = resp.text.strip() ipaddress.ip_address(ip) # validate return ip except Exception: continue return None class DashboardView(TemplateView): template_name = 'netscan/dashboard.html' def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) profiles = ScanProfile.objects.all() profile_data = [] for p in profiles: last_run = p.runs.first() worst = _worst_severity(last_run.summary) if last_run else None profile_data.append({ 'profile': p, 'last_run': last_run, 'worst_severity': worst, }) ctx['profile_data'] = profile_data return ctx class ProfileCreateView(CreateView): model = ScanProfile form_class = ScanProfileForm template_name = 'netscan/profile_form.html' success_url = reverse_lazy('netscan-dashboard') def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx['form_title'] = 'Create Scan Profile' return ctx class ProfileUpdateView(UpdateView): model = ScanProfile form_class = ScanProfileForm template_name = 'netscan/profile_form.html' success_url = reverse_lazy('netscan-dashboard') def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx['form_title'] = f'Edit: {self.object.name}' return ctx class ProfileDeleteView(DeleteView): model = ScanProfile template_name = 'netscan/profile_confirm_delete.html' success_url = reverse_lazy('netscan-dashboard') class ScanRunListView(ListView): template_name = 'netscan/run_list.html' context_object_name = 'runs' paginate_by = 20 def get_queryset(self): self.profile = get_object_or_404(ScanProfile, pk=self.kwargs['pk']) return ScanRun.objects.filter(profile=self.profile) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx['profile'] = self.profile return ctx class ScanRunDetailView(DetailView): model = ScanRun template_name = 'netscan/run_detail.html' context_object_name = 'run' def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) findings = self.object.findings.all() ctx['critical_findings'] = findings.filter(severity='critical') ctx['warning_findings'] = findings.filter(severity='warning') ctx['ok_findings'] = findings.filter(severity__in=['ok', 'info']) return ctx class TriggerScanView(View): def post(self, request, pk): profile = get_object_or_404(ScanProfile, pk=pk) t = threading.Thread(target=run_scan, args=[profile.pk, 'manual'], daemon=True) t.start() return redirect(reverse('netscan-run-list', kwargs={'pk': profile.pk})) class TestTelegramView(View): def post(self, request, pk): profile = get_object_or_404(ScanProfile, pk=pk) result = send_test_telegram(profile) return JsonResponse(result) class DetectNetworkView(View): def get(self, request): local_ip = _detect_local_ip() data = { 'gateway_ip': _detect_gateway(), 'local_ip': local_ip, 'network_cidr': _detect_subnet_mask(local_ip) if local_ip else None, 'public_ip': _detect_public_ip(), } return JsonResponse(data)