Add network scan

This commit is contained in:
2026-03-21 16:41:14 +11:00
parent 4d73ed9ac9
commit 551fdf5cbb
37 changed files with 2564 additions and 1 deletions
View File
+21
View File
@@ -0,0 +1,21 @@
from django.contrib import admin
from .models import ScanProfile, ScanRun, ScanFinding
@admin.register(ScanProfile)
class ScanProfileAdmin(admin.ModelAdmin):
list_display = ['name', 'enabled', 'schedule_interval', 'gateway_ip', 'public_ip', 'last_run_at', 'created_at']
list_filter = ['enabled', 'schedule_interval']
@admin.register(ScanRun)
class ScanRunAdmin(admin.ModelAdmin):
list_display = ['profile', 'status', 'triggered_by', 'started_at', 'finished_at']
list_filter = ['status', 'triggered_by', 'profile']
readonly_fields = ['started_at', 'finished_at', 'summary']
@admin.register(ScanFinding)
class ScanFindingAdmin(admin.ModelAdmin):
list_display = ['run', 'check_name', 'severity', 'title']
list_filter = ['severity', 'check_name']
+20
View File
@@ -0,0 +1,20 @@
from django.apps import AppConfig
import logging
logger = logging.getLogger(__name__)
class NetscanConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'netscan'
def ready(self):
import netscan.signals # noqa: F401
try:
from netscan.tasks import schedule_profile
from netscan.models import ScanProfile
for profile in ScanProfile.objects.filter(enabled=True):
schedule_profile(profile)
logger.info(f'Scheduled netscan profile: {profile.name}')
except Exception as e:
logger.warning(f'Could not schedule netscan profiles on startup: {e}')
View File
+10
View File
@@ -0,0 +1,10 @@
from dataclasses import dataclass, field
@dataclass
class Finding:
check_name: str
severity: str # ok | info | warning | critical
title: str
detail: str
raw: dict = field(default_factory=dict)
+117
View File
@@ -0,0 +1,117 @@
import socket
import logging
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'camera_rtsp'
TIMEOUT = 5
RTSP_PORT = 554
def _tcp_open(host: str, port: int) -> bool:
try:
with socket.create_connection((host, port), timeout=TIMEOUT):
return True
except (OSError, socket.timeout):
return False
def _rtsp_request(host: str, method: str, seq: int, extra_headers: str = '') -> str:
try:
with socket.create_connection((host, RTSP_PORT), timeout=TIMEOUT) as s:
request = (
f'{method} rtsp://{host}/ RTSP/1.0\r\n'
f'CSeq: {seq}\r\n'
f'{extra_headers}'
'\r\n'
)
s.sendall(request.encode())
response = s.recv(4096).decode('utf-8', errors='replace')
return response
except Exception as e:
return f'ERROR: {e}'
def _parse_rtsp_status(response: str) -> int:
"""Extract HTTP-style status code from RTSP response."""
try:
first_line = response.splitlines()[0]
return int(first_line.split()[1])
except (IndexError, ValueError):
return 0
def run(profile) -> list:
findings = []
cameras = profile.cameras or []
if not cameras:
return [Finding(
check_name=CHECK_NAME,
severity='info',
title='No camera IPs configured',
detail='Add camera IPs to the scan profile to enable RTSP unauthenticated access check.',
raw={},
)]
for ip in cameras:
raw = {'camera_ip': ip}
if not _tcp_open(ip, RTSP_PORT):
findings.append(Finding(
check_name=CHECK_NAME,
severity='info',
title=f'{ip}: RTSP port 554 closed',
detail=f'Port 554 is not open on {ip}. Camera may be offline or not using RTSP.',
raw=raw,
))
continue
options_resp = _rtsp_request(ip, 'OPTIONS', 1)
raw['options_response'] = options_resp[:500]
options_status = _parse_rtsp_status(options_resp)
if options_status == 0:
findings.append(Finding(
check_name=CHECK_NAME,
severity='info',
title=f'{ip}: RTSP OPTIONS failed',
detail=f'Got unexpected RTSP OPTIONS response from {ip}.',
raw=raw,
))
continue
describe_resp = _rtsp_request(ip, 'DESCRIBE', 2, 'Accept: application/sdp\r\n')
raw['describe_response'] = describe_resp[:500]
describe_status = _parse_rtsp_status(describe_resp)
if describe_status == 200:
findings.append(Finding(
check_name=CHECK_NAME,
severity='critical',
title=f'{ip}: RTSP stream accessible without credentials',
detail=(
f'Camera at {ip} returned 200 to DESCRIBE without authentication. '
'Live stream may be publicly accessible on the LAN.'
),
raw=raw,
))
elif describe_status in (401, 403):
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title=f'{ip}: RTSP requires authentication',
detail=f'Camera at {ip} returned {describe_status} to DESCRIBE — auth is enforced.',
raw=raw,
))
else:
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title=f'{ip}: RTSP DESCRIBE returned {describe_status}',
detail=f'Camera at {ip} responded with status {describe_status} — no unauthenticated stream detected.',
raw=raw,
))
return findings
+86
View File
@@ -0,0 +1,86 @@
import socket
import struct
import logging
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'dns_resolver'
TIMEOUT = 5
def _build_dns_query(domain: str) -> bytes:
"""Build a minimal DNS A query packet."""
header = struct.pack('>HHHHHH', 0xAAAA, 0x0100, 1, 0, 0, 0)
parts = domain.encode().split(b'.')
question = b''.join(bytes([len(p)]) + p for p in parts) + b'\x00'
question += struct.pack('>HH', 1, 1) # type A, class IN
return header + question
def _parse_dns_response(data: bytes) -> dict:
"""Return basic info from a DNS response header."""
if len(data) < 12:
return {'error': 'response too short'}
txid, flags, qdcount, ancount, nscount, arcount = struct.unpack('>HHHHHH', data[:12])
rcode = flags & 0x000F
return {
'txid': txid,
'flags': flags,
'rcode': rcode,
'ancount': ancount,
}
def run(profile) -> list:
host = profile.public_ip
raw = {'public_ip': host}
query = _build_dns_query('google.com')
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(TIMEOUT)
sock.sendto(query, (host, 53))
data, _ = sock.recvfrom(512)
sock.close()
parsed = _parse_dns_response(data)
raw['response'] = parsed
if parsed.get('rcode') == 0 and parsed.get('ancount', 0) > 0:
return [Finding(
check_name=CHECK_NAME,
severity='critical',
title=f'Open DNS resolver detected on {host}:53',
detail=(
'Your public IP responds to recursive DNS queries from external hosts. '
'This can be abused for DNS amplification attacks. '
'Note: NAT hairpin may cause false positive — verify from off-LAN.'
),
raw=raw,
)]
else:
return [Finding(
check_name=CHECK_NAME,
severity='ok',
title='DNS port 53 does not appear to be an open resolver',
detail=f'DNS query to {host}:53 returned rcode={parsed.get("rcode")} with {parsed.get("ancount", 0)} answers.',
raw=raw,
)]
except socket.timeout:
return [Finding(
check_name=CHECK_NAME,
severity='ok',
title='DNS port 53 timed out (not an open resolver)',
detail=f'No response from {host}:53 within {TIMEOUT}s. Port is likely closed or filtered.',
raw={**raw, 'error': 'timeout'},
)]
except Exception as e:
logger.warning(f'DNS check error: {e}')
return [Finding(
check_name=CHECK_NAME,
severity='info',
title='DNS resolver check failed',
detail=f'Could not complete DNS probe to {host}:53 — {e}',
raw={**raw, 'error': str(e)},
)]
+93
View File
@@ -0,0 +1,93 @@
import logging
import requests
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'ingress_auth'
TIMEOUT = 10
def run(profile) -> list:
findings = []
domains = profile.domains or []
auth_host = profile.auth_provider_host
if not domains:
return [Finding(
check_name=CHECK_NAME,
severity='info',
title='No domains configured for ingress check',
detail='Add domains to the scan profile to enable ingress auth verification.',
raw={},
)]
for domain in domains:
raw = {'domain': domain, 'auth_provider_host': auth_host}
try:
resp = requests.get(f'https://{domain}/', allow_redirects=True, timeout=TIMEOUT,
headers={'User-Agent': 'NetScan/1.0'})
redirect_chain = [r.url for r in resp.history] + [resp.url]
raw['redirect_chain'] = redirect_chain
raw['final_url'] = resp.url
raw['status_code'] = resp.status_code
if auth_host:
passed_through_auth = any(auth_host in url for url in redirect_chain[:-1])
final_is_auth = auth_host in resp.url
if passed_through_auth or final_is_auth:
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title=f'{domain}: auth provider in redirect chain',
detail=f'Request passed through {auth_host} as expected.',
raw=raw,
))
else:
findings.append(Finding(
check_name=CHECK_NAME,
severity='critical',
title=f'{domain}: auth provider NOT in redirect chain',
detail=(
f'Expected redirect through {auth_host} but final URL is {resp.url}. '
'Authentication may be bypassed.'
),
raw=raw,
))
else:
findings.append(Finding(
check_name=CHECK_NAME,
severity='info',
title=f'{domain}: reachable (no auth provider configured)',
detail=f'Domain reached with status {resp.status_code}. Set auth_provider_host to verify auth.',
raw=raw,
))
except requests.exceptions.ConnectionError as e:
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: connection refused',
detail=f'Could not connect to https://{domain}/ — {e}',
raw={**raw, 'error': str(e)},
))
except requests.exceptions.Timeout:
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: request timed out',
detail=f'Request to https://{domain}/ timed out after {TIMEOUT}s.',
raw={**raw, 'error': 'timeout'},
))
except Exception as e:
logger.warning(f'Ingress check error for {domain}: {e}')
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: check error',
detail=str(e),
raw={**raw, 'error': str(e)},
))
return findings
+58
View File
@@ -0,0 +1,58 @@
import socket
import logging
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'public_ports'
TIMEOUT = 3
PORT_INFO = {
22: ('warning', 'SSH', 'SSH exposed to the internet. Ensure key-only auth and restrict access.'),
23: ('critical', 'Telnet', 'Telnet (cleartext) is exposed to the internet. Disable immediately.'),
25: ('warning', 'SMTP', 'SMTP port exposed. Could be used for spam relay if misconfigured.'),
53: ('warning', 'DNS', 'DNS port open. Run the DNS resolver check to confirm if recursive queries are allowed.'),
80: ('info', 'HTTP', 'HTTP port open. Expected for public web services.'),
443: ('info', 'HTTPS', 'HTTPS port open. Expected for public web services.'),
3306: ('critical', 'MySQL', 'MySQL database port exposed to the internet. Restrict access immediately.'),
5432: ('critical', 'Postgres','PostgreSQL database port exposed to the internet. Restrict access immediately.'),
6379: ('critical', 'Redis', 'Redis port exposed to the internet. Redis has no auth by default — critical risk.'),
8080: ('warning', 'HTTP-alt','Alternate HTTP port 8080 is open. Verify this is intentional.'),
8443: ('warning', 'HTTPS-alt','Alternate HTTPS port 8443 is open. Verify this is intentional.'),
}
def _tcp_open(host: str, port: int) -> bool:
try:
with socket.create_connection((host, port), timeout=TIMEOUT):
return True
except (OSError, socket.timeout):
return False
def run(profile) -> list:
findings = []
host = profile.public_ip
raw = {'public_ip': host, 'open_ports': []}
for port, (severity, label, detail) in PORT_INFO.items():
if _tcp_open(host, port):
raw['open_ports'].append(port)
findings.append(Finding(
check_name=CHECK_NAME,
severity=severity,
title=f'Port {port} ({label}) open on public IP {host}',
detail=detail,
raw={'public_ip': host, 'port': port},
))
if not findings:
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title=f'No high-risk ports open on public IP {host}',
detail='All probed ports are closed or filtered.',
raw=raw,
))
return findings
+117
View File
@@ -0,0 +1,117 @@
import socket
import logging
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'router_ports'
PORTS_TO_PROBE = [22, 23, 53, 80, 139, 443, 445, 8080, 8443]
TIMEOUT = 3
def _tcp_open(host: str, port: int) -> bool:
try:
with socket.create_connection((host, port), timeout=TIMEOUT):
return True
except (OSError, socket.timeout):
return False
def _fetch_http_headers(host: str, port: int = 80) -> dict:
try:
with socket.create_connection((host, port), timeout=TIMEOUT) as s:
s.sendall(f'HEAD / HTTP/1.0\r\nHost: {host}\r\n\r\n'.encode())
resp = s.recv(4096).decode('utf-8', errors='replace')
headers = {}
for line in resp.splitlines()[1:]:
if ':' in line:
k, _, v = line.partition(':')
headers[k.strip().lower()] = v.strip()
return headers
except Exception:
return {}
def run(profile) -> list:
findings = []
host = profile.gateway_ip
open_ports = {}
for port in PORTS_TO_PROBE:
open_ports[port] = _tcp_open(host, port)
raw = {'gateway_ip': host, 'open_ports': {str(p): v for p, v in open_ports.items()}}
# SMB exposure
if open_ports.get(139) or open_ports.get(445):
smb_ports = [p for p in [139, 445] if open_ports.get(p)]
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'SMB ports open on gateway ({", ".join(str(p) for p in smb_ports)})',
detail='Windows file sharing (SMB) is accessible on the gateway. This could expose network shares.',
raw=raw,
))
# Telnet
if open_ports.get(23):
findings.append(Finding(
check_name=CHECK_NAME,
severity='critical',
title='Telnet port 23 open on gateway',
detail='Telnet transmits credentials in plaintext. Disable telnet and use SSH instead.',
raw=raw,
))
# SSH open on gateway
if open_ports.get(22):
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title='SSH port 22 open on gateway',
detail='SSH is accessible on the gateway. Ensure key-only auth is enforced and access is restricted.',
raw=raw,
))
# Plain HTTP admin (port 80 open, port 443 closed)
if open_ports.get(80) and not open_ports.get(443):
headers = _fetch_http_headers(host, 80)
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title='Gateway admin over plain HTTP (no HTTPS)',
detail=f'Port 80 is open but 443 is closed. Admin interface may be served unencrypted. Server header: {headers.get("server", "unknown")}',
raw={**raw, 'http_headers': headers},
))
# Unknown port 8080
if open_ports.get(8080):
findings.append(Finding(
check_name=CHECK_NAME,
severity='info',
title='Port 8080 open on gateway',
detail='An alternate HTTP service is running on port 8080. Verify this is intentional.',
raw=raw,
))
# Port 8443 open
if open_ports.get(8443):
findings.append(Finding(
check_name=CHECK_NAME,
severity='info',
title='Port 8443 open on gateway',
detail='An alternate HTTPS service is running on port 8443. Verify this is intentional.',
raw=raw,
))
if not findings:
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title='Gateway port scan looks clean',
detail=f'No high-risk ports found open on {host}.',
raw=raw,
))
return findings
+131
View File
@@ -0,0 +1,131 @@
import ssl
import socket
import logging
from datetime import datetime, timezone
from .base import Finding
logger = logging.getLogger(__name__)
CHECK_NAME = 'tls_expiry'
TIMEOUT = 10
def _get_cert_info(domain: str) -> dict:
ctx = ssl.create_default_context()
try:
with ctx.wrap_socket(socket.create_connection((domain, 443), timeout=TIMEOUT),
server_hostname=domain) as s:
cert = s.getpeercert()
return {'cert': cert, 'error': None}
except ssl.SSLCertVerificationError as e:
return {'cert': None, 'error': f'SSL verification failed: {e}'}
except Exception as e:
return {'cert': None, 'error': str(e)}
def _days_until_expiry(not_after: str) -> int:
expiry = datetime.strptime(not_after, '%b %d %H:%M:%S %Y %Z').replace(tzinfo=timezone.utc)
now = datetime.now(tz=timezone.utc)
return (expiry - now).days
def _cert_covers_domain(cert: dict, domain: str) -> bool:
san_list = [v for t, v in cert.get('subjectAltName', []) if t == 'DNS']
for san in san_list:
if san == domain:
return True
if san.startswith('*.') and domain.endswith(san[1:]):
return True
if not san_list:
cn = dict(x[0] for x in cert.get('subject', [])).get('commonName', '')
return cn == domain
return False
def run(profile) -> list:
findings = []
domains = profile.domains or []
if not domains:
return [Finding(
check_name=CHECK_NAME,
severity='info',
title='No domains configured for TLS check',
detail='Add domains to the scan profile to enable TLS certificate checks.',
raw={},
)]
for domain in domains:
raw = {'domain': domain}
info = _get_cert_info(domain)
if info['error']:
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: TLS check failed',
detail=info['error'],
raw={**raw, 'error': info['error']},
))
continue
cert = info['cert']
not_after = cert.get('notAfter', '')
raw['not_after'] = not_after
try:
days = _days_until_expiry(not_after)
raw['days_to_expiry'] = days
if days < 0:
findings.append(Finding(
check_name=CHECK_NAME,
severity='critical',
title=f'{domain}: TLS certificate EXPIRED {abs(days)} days ago',
detail=f'Certificate expired on {not_after}. Renew immediately.',
raw=raw,
))
elif days < 14:
findings.append(Finding(
check_name=CHECK_NAME,
severity='critical',
title=f'{domain}: TLS certificate expires in {days} days',
detail=f'Certificate will expire on {not_after}. Renew urgently.',
raw=raw,
))
elif days < 30:
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: TLS certificate expires in {days} days',
detail=f'Certificate expires on {not_after}. Plan renewal soon.',
raw=raw,
))
else:
findings.append(Finding(
check_name=CHECK_NAME,
severity='ok',
title=f'{domain}: TLS certificate valid for {days} more days',
detail=f'Certificate expires on {not_after}.',
raw=raw,
))
except ValueError as e:
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: could not parse certificate expiry',
detail=str(e),
raw=raw,
))
continue
if not _cert_covers_domain(cert, domain):
findings.append(Finding(
check_name=CHECK_NAME,
severity='warning',
title=f'{domain}: certificate CN/SAN does not match domain',
detail=f'The TLS certificate does not include {domain} in its names.',
raw=raw,
))
return findings
+39
View File
@@ -0,0 +1,39 @@
import base64
import hashlib
from cryptography.fernet import Fernet
from django.conf import settings
from django.db import models
def _get_fernet():
"""Derive a Fernet key from Django's SECRET_KEY."""
raw_key = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
return Fernet(base64.urlsafe_b64encode(raw_key))
class EncryptedCharField(models.TextField):
"""
Stores values encrypted at rest using Fernet symmetric encryption.
The encryption key is derived from Django's SECRET_KEY so no extra
secrets management is required — if the SECRET_KEY is set, values
are protected.
From the application's perspective this behaves like a plain text field:
you read/write plaintext; encryption/decryption happens transparently.
"""
def from_db_value(self, value, expression, connection):
if not value:
return value
try:
return _get_fernet().decrypt(value.encode()).decode()
except Exception:
# Gracefully return raw value if decryption fails
# (e.g. migrating plaintext rows that were saved before encryption)
return value
def get_prep_value(self, value):
if not value:
return value
return _get_fernet().encrypt(value.encode()).decode()
+73
View File
@@ -0,0 +1,73 @@
from django import forms
from .models import ScanProfile
_INPUT = (
'w-full border border-gray-300 rounded-lg px-3 py-2 text-sm text-gray-700 '
'focus:outline-none focus:ring-2 focus:ring-red-500 focus:border-transparent'
)
_TEXTAREA = _INPUT + ' resize-none'
class ScanProfileForm(forms.ModelForm):
domains_text = forms.CharField(
widget=forms.Textarea(attrs={
'rows': 4,
'placeholder': 'to.junv.cc\ngo.junv.cc',
'class': _TEXTAREA,
}),
required=False,
label='Domains (one per line)',
help_text='Public hostnames to check for TLS and auth.',
)
cameras_text = forms.CharField(
widget=forms.Textarea(attrs={
'rows': 3,
'placeholder': '192.168.1.70\n192.168.1.71',
'class': _TEXTAREA,
}),
required=False,
label='Camera IPs (one per line)',
help_text='Local IP addresses of cameras to probe for unauthenticated RTSP.',
)
class Meta:
model = ScanProfile
fields = [
'name', 'enabled', 'schedule_interval',
'gateway_ip', 'public_ip', 'network_cidr',
'auth_provider_host',
'telegram_bot_token', 'telegram_chat_id', 'notify_on_severity',
]
widgets = {
'name': forms.TextInput(attrs={'class': _INPUT}),
'schedule_interval': forms.Select(attrs={'class': _INPUT}),
'gateway_ip': forms.TextInput(attrs={'class': _INPUT}),
'public_ip': forms.TextInput(attrs={'class': _INPUT}),
'network_cidr': forms.TextInput(attrs={'class': _INPUT}),
'auth_provider_host': forms.TextInput(attrs={'class': _INPUT}),
'telegram_bot_token': forms.PasswordInput(render_value=True, attrs={'class': _INPUT}),
'telegram_chat_id': forms.TextInput(attrs={'class': _INPUT}),
'notify_on_severity': forms.Select(attrs={'class': _INPUT}),
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if self.instance and self.instance.pk:
self.fields['domains_text'].initial = '\n'.join(self.instance.domains or [])
self.fields['cameras_text'].initial = '\n'.join(self.instance.cameras or [])
def clean_domains_text(self):
raw = self.cleaned_data.get('domains_text', '')
return [line.strip() for line in raw.splitlines() if line.strip()]
def clean_cameras_text(self):
raw = self.cleaned_data.get('cameras_text', '')
return [line.strip() for line in raw.splitlines() if line.strip()]
def save(self, commit=True):
instance = super().save(commit=False)
instance.domains = self.cleaned_data['domains_text']
instance.cameras = self.cleaned_data['cameras_text']
if commit:
instance.save()
return instance
+63
View File
@@ -0,0 +1,63 @@
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='ScanProfile',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=200)),
('enabled', models.BooleanField(default=True)),
('schedule_interval', models.IntegerField(choices=[(1, 'Every 1 day'), (3, 'Every 3 days'), (7, 'Every 7 days'), (30, 'Every 30 days')], default=7)),
('gateway_ip', models.GenericIPAddressField(help_text='e.g. 192.168.1.1')),
('public_ip', models.GenericIPAddressField(help_text='Your public/WAN IP address')),
('network_cidr', models.CharField(blank=True, help_text='e.g. 192.168.1.0/24', max_length=50)),
('auth_provider_host', models.CharField(blank=True, help_text='e.g. pass.junv.cc', max_length=255)),
('domains', models.JSONField(blank=True, default=list, help_text='List of public hostnames to check')),
('cameras', models.JSONField(blank=True, default=list, help_text='List of camera IPs to probe')),
('telegram_bot_token', models.CharField(blank=True, max_length=255)),
('telegram_chat_id', models.CharField(blank=True, max_length=100)),
('notify_on_severity', models.CharField(choices=[('warning', 'Warning and above'), ('critical', 'Critical only')], default='critical', max_length=20)),
('last_run_at', models.DateTimeField(blank=True, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
],
),
migrations.CreateModel(
name='ScanRun',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('started_at', models.DateTimeField(auto_now_add=True)),
('finished_at', models.DateTimeField(blank=True, null=True)),
('status', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('success', 'Success'), ('failed', 'Failed')], default='pending', max_length=20)),
('summary', models.JSONField(default=dict)),
('triggered_by', models.CharField(default='manual', max_length=20)),
('profile', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='runs', to='netscan.scanprofile')),
],
options={
'ordering': ['-started_at'],
},
),
migrations.CreateModel(
name='ScanFinding',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('check_name', models.CharField(max_length=100)),
('severity', models.CharField(choices=[('ok', 'OK'), ('info', 'Info'), ('warning', 'Warning'), ('critical', 'Critical')], max_length=20)),
('title', models.CharField(max_length=255)),
('detail', models.TextField()),
('raw', models.JSONField(default=dict)),
('run', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='findings', to='netscan.scanrun')),
],
options={
'ordering': ['severity', 'check_name'],
},
),
]
@@ -0,0 +1,22 @@
from django.db import migrations
import netscan.fields
class Migration(migrations.Migration):
dependencies = [
('netscan', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='scanprofile',
name='telegram_bot_token',
field=netscan.fields.EncryptedCharField(blank=True),
),
migrations.AlterField(
model_name='scanprofile',
name='telegram_chat_id',
field=netscan.fields.EncryptedCharField(blank=True),
),
]
View File
+84
View File
@@ -0,0 +1,84 @@
from django.db import models
from .fields import EncryptedCharField
class ScanProfile(models.Model):
INTERVAL_CHOICES = [
(1, 'Every 1 day'),
(3, 'Every 3 days'),
(7, 'Every 7 days'),
(30, 'Every 30 days'),
]
SEVERITY_CHOICES = [
('warning', 'Warning and above'),
('critical', 'Critical only'),
]
name = models.CharField(max_length=200)
enabled = models.BooleanField(default=True)
schedule_interval = models.IntegerField(choices=INTERVAL_CHOICES, default=7)
gateway_ip = models.GenericIPAddressField(help_text='e.g. 192.168.1.1')
public_ip = models.GenericIPAddressField(help_text='Your public/WAN IP address')
network_cidr = models.CharField(max_length=50, blank=True, help_text='e.g. 192.168.1.0/24')
auth_provider_host = models.CharField(max_length=255, blank=True, help_text='e.g. pass.junv.cc')
domains = models.JSONField(default=list, blank=True, help_text='List of public hostnames to check')
cameras = models.JSONField(default=list, blank=True, help_text='List of camera IPs to probe')
telegram_bot_token = EncryptedCharField(blank=True)
telegram_chat_id = EncryptedCharField(blank=True)
notify_on_severity = models.CharField(max_length=20, choices=SEVERITY_CHOICES, default='critical')
last_run_at = models.DateTimeField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return self.name
class ScanRun(models.Model):
STATUS_CHOICES = [
('pending', 'Pending'),
('running', 'Running'),
('success', 'Success'),
('failed', 'Failed'),
]
profile = models.ForeignKey(ScanProfile, on_delete=models.CASCADE, related_name='runs')
started_at = models.DateTimeField(auto_now_add=True)
finished_at = models.DateTimeField(null=True, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
summary = models.JSONField(default=dict)
triggered_by = models.CharField(max_length=20, default='manual')
class Meta:
ordering = ['-started_at']
def __str__(self):
return f'{self.profile.name} run #{self.pk} ({self.status})'
@property
def duration_seconds(self):
if self.finished_at and self.started_at:
return int((self.finished_at - self.started_at).total_seconds())
return None
class ScanFinding(models.Model):
SEVERITY_CHOICES = [
('ok', 'OK'),
('info', 'Info'),
('warning', 'Warning'),
('critical', 'Critical'),
]
run = models.ForeignKey(ScanRun, on_delete=models.CASCADE, related_name='findings')
check_name = models.CharField(max_length=100)
severity = models.CharField(max_length=20, choices=SEVERITY_CHOICES)
title = models.CharField(max_length=255)
detail = models.TextField()
raw = models.JSONField(default=dict)
class Meta:
ordering = ['severity', 'check_name']
def __str__(self):
return f'[{self.severity.upper()}] {self.title}'
+60
View File
@@ -0,0 +1,60 @@
import logging
import requests
logger = logging.getLogger(__name__)
SEVERITY_ORDER = ['ok', 'info', 'warning', 'critical']
SEVERITY_ICONS = {'critical': '🔴', 'warning': '🟡', 'ok': '🟢', 'info': ''}
def notify_telegram(profile, run, findings):
"""
Send a Telegram message if any finding meets or exceeds notify_on_severity.
"""
threshold_idx = SEVERITY_ORDER.index(profile.notify_on_severity)
flagged = [f for f in findings if SEVERITY_ORDER.index(f.severity) >= threshold_idx]
if not flagged:
return
finished_str = run.finished_at.strftime('%Y-%m-%d %H:%M') if run.finished_at else 'unknown'
lines = [
f'🔒 *NetScan Alert* — {profile.name}',
f'Run \\#{run.pk} finished at {finished_str}',
f'Summary: {run.summary}',
'',
]
for f in flagged[:10]:
icon = SEVERITY_ICONS.get(f.severity, '')
lines.append(f'{icon} *{f.title}*\n {f.detail[:120]}')
if len(flagged) > 10:
lines.append(f'_...and {len(flagged) - 10} more findings_')
text = '\n'.join(lines)
url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage'
resp = requests.post(url, json={
'chat_id': profile.telegram_chat_id,
'text': text,
'parse_mode': 'Markdown',
}, timeout=10)
resp.raise_for_status()
logger.info(f'Telegram notification sent for run #{run.pk}')
def send_test_telegram(profile) -> dict:
"""Send a test message. Returns {'ok': True} or {'ok': False, 'error': str}."""
if not profile.telegram_bot_token or not profile.telegram_chat_id:
return {'ok': False, 'error': 'Telegram bot token or chat ID not configured.'}
try:
url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage'
resp = requests.post(url, json={
'chat_id': profile.telegram_chat_id,
'text': f'✅ *NetScan test message* from profile _{profile.name}_. Notifications are working.',
'parse_mode': 'Markdown',
}, timeout=10)
resp.raise_for_status()
return {'ok': True}
except Exception as e:
return {'ok': False, 'error': str(e)}
+74
View File
@@ -0,0 +1,74 @@
import logging
from collections import Counter
from django.utils.timezone import now
from .models import ScanProfile, ScanRun, ScanFinding
from .checks import router, dns, ingress, cameras, tls, ports
from .checks.base import Finding
from .notifications import notify_telegram
logger = logging.getLogger(__name__)
CHECK_MODULES = [router, dns, ingress, cameras, tls, ports]
def run_scan(profile_id: int, triggered_by: str = 'scheduler') -> int:
"""
Run all checks for the given profile. Returns the ScanRun PK.
Called by APScheduler jobs and TriggerScanView.
"""
profile = ScanProfile.objects.get(pk=profile_id)
run = ScanRun.objects.create(
profile=profile,
status='running',
triggered_by=triggered_by,
)
logger.info(f'Starting netscan run #{run.pk} for profile "{profile.name}" (triggered_by={triggered_by})')
all_findings: list[Finding] = []
for mod in CHECK_MODULES:
mod_name = mod.__name__.split('.')[-1]
try:
findings = mod.run(profile)
all_findings.extend(findings)
logger.debug(f' {mod_name}: {len(findings)} findings')
except Exception as e:
logger.exception(f' {mod_name}: uncaught exception')
all_findings.append(Finding(
check_name=mod_name,
severity='warning',
title=f'{mod_name}: check errored',
detail=str(e),
raw={'exception': str(e)},
))
ScanFinding.objects.bulk_create([
ScanFinding(
run=run,
check_name=f.check_name,
severity=f.severity,
title=f.title,
detail=f.detail,
raw=f.raw,
)
for f in all_findings
])
summary = dict(Counter(f.severity for f in all_findings))
run.summary = summary
run.status = 'success'
run.finished_at = now()
run.save()
profile.last_run_at = now()
profile.save(update_fields=['last_run_at'])
if profile.telegram_bot_token and profile.telegram_chat_id:
try:
notify_telegram(profile, run, all_findings)
except Exception as e:
logger.warning(f'Telegram notification failed: {e}')
logger.info(f'Finished netscan run #{run.pk}: {summary}')
return run.pk
+17
View File
@@ -0,0 +1,17 @@
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
@receiver(post_save, sender='netscan.ScanProfile')
def reschedule_on_save(sender, instance, **kwargs):
from netscan.tasks import schedule_profile, unschedule_profile
if instance.enabled:
schedule_profile(instance)
else:
unschedule_profile(instance)
@receiver(post_delete, sender='netscan.ScanProfile')
def unschedule_on_delete(sender, instance, **kwargs):
from netscan.tasks import unschedule_profile
unschedule_profile(instance)
+25
View File
@@ -0,0 +1,25 @@
import logging
from apscheduler.triggers.interval import IntervalTrigger
from core.scheduler import scheduler
from netscan.scanner import run_scan
logger = logging.getLogger(__name__)
def schedule_profile(profile):
job_id = f'netscan_profile_{profile.pk}'
scheduler.add_job(
run_scan,
trigger=IntervalTrigger(days=profile.schedule_interval),
id=job_id,
args=[profile.pk, 'scheduler'],
replace_existing=True,
)
logger.info(f'Scheduled netscan job {job_id} every {profile.schedule_interval} day(s)')
def unschedule_profile(profile):
job_id = f'netscan_profile_{profile.pk}'
if scheduler.get_job(job_id):
scheduler.remove_job(job_id)
logger.info(f'Removed netscan job {job_id}')
+133
View File
@@ -0,0 +1,133 @@
{% extends 'base.html' %}
{% block content %}
<div class="max-w-7xl mx-auto px-4 sm:px-6 lg:px-8 py-8">
<div class="bg-white shadow-md rounded-lg overflow-hidden">
<!-- Header -->
<div class="px-4 py-5 sm:px-6 bg-gray-50 border-b border-gray-200 flex items-center justify-between">
<div>
<h1 class="text-2xl font-bold text-gray-900 flex items-center">
<i class="fas fa-shield-alt mr-3 text-red-500"></i>
NetScan
</h1>
<p class="text-sm text-gray-500 mt-0.5">Home network security scanner</p>
</div>
<a href="{% url 'netscan-profile-create' %}"
class="inline-flex items-center px-4 py-2 bg-red-600 text-white text-sm font-medium rounded-md hover:bg-red-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-red-500">
<i class="fas fa-plus mr-2"></i>
New Profile
</a>
</div>
{% if not profile_data %}
<!-- Empty state -->
<div class="text-center py-20">
<div class="w-16 h-16 bg-red-100 rounded-full flex items-center justify-center mx-auto mb-4">
<i class="fas fa-shield-alt text-red-500 text-2xl"></i>
</div>
<h2 class="text-lg font-semibold text-gray-700 mb-2">No scan profiles yet</h2>
<p class="text-gray-500 text-sm mb-6">Create your first scan profile to start monitoring your home network.</p>
<a href="{% url 'netscan-profile-create' %}"
class="inline-flex items-center px-5 py-2.5 bg-red-600 text-white text-sm font-medium rounded-md hover:bg-red-700">
<i class="fas fa-plus mr-2"></i>
Create your first scan profile
</a>
</div>
{% else %}
<!-- Profiles table -->
<table class="min-w-full text-sm">
<thead class="bg-gray-50 border-b border-gray-100">
<tr>
<th class="px-4 py-3 text-left text-xs font-semibold text-gray-500 uppercase tracking-wide">Profile</th>
<th class="px-4 py-3 text-left text-xs font-semibold text-gray-500 uppercase tracking-wide hidden sm:table-cell">Network</th>
<th class="px-4 py-3 text-left text-xs font-semibold text-gray-500 uppercase tracking-wide hidden md:table-cell">Last Run</th>
<th class="px-4 py-3 text-left text-xs font-semibold text-gray-500 uppercase tracking-wide hidden sm:table-cell">Findings</th>
<th class="px-4 py-3 text-right text-xs font-semibold text-gray-500 uppercase tracking-wide">Actions</th>
</tr>
</thead>
<tbody class="divide-y divide-gray-100">
{% for item in profile_data %}
{% with p=item.profile run=item.last_run worst=item.worst_severity %}
<tr class="hover:bg-gray-50 transition-colors">
<!-- Profile name + badges -->
<td class="px-4 py-4">
<div class="font-semibold text-gray-900 flex items-center gap-2">
{% if worst %}
{% if worst == 'critical' %}🔴{% elif worst == 'warning' %}🟡{% else %}🟢{% endif %}
{% endif %}
{{ p.name }}
</div>
<div class="flex flex-wrap gap-1 mt-1">
<span class="inline-flex items-center px-2 py-0.5 rounded text-xs font-medium
{% if p.enabled %}bg-green-100 text-green-700{% else %}bg-gray-100 text-gray-500{% endif %}">
{% if p.enabled %}Enabled{% else %}Disabled{% endif %}
</span>
<span class="inline-flex items-center px-2 py-0.5 rounded text-xs font-medium bg-blue-50 text-blue-600">
Every {{ p.schedule_interval }} day{{ p.schedule_interval|pluralize }}
</span>
</div>
</td>
<!-- Network info -->
<td class="px-4 py-4 hidden sm:table-cell">
<div class="text-gray-700 font-mono text-xs">{{ p.gateway_ip }}</div>
<div class="text-gray-400 font-mono text-xs">{{ p.public_ip }}</div>
</td>
<!-- Last run -->
<td class="px-4 py-4 hidden md:table-cell text-gray-500 text-xs">
{% if p.last_run_at %}{{ p.last_run_at|date:"M d, H:i" }}{% else %}<span class="text-gray-400">Never</span>{% endif %}
</td>
<!-- Findings -->
<td class="px-4 py-4 hidden sm:table-cell">
{% if run %}
<div class="flex items-center gap-2">
{% if run.summary.critical %}<span class="text-red-600 font-medium text-xs">🔴{{ run.summary.critical }}</span>{% endif %}
{% if run.summary.warning %}<span class="text-yellow-600 font-medium text-xs">🟡{{ run.summary.warning }}</span>{% endif %}
{% if run.summary.ok %}<span class="text-green-600 font-medium text-xs">🟢{{ run.summary.ok }}</span>{% endif %}
</div>
{% else %}
<span class="text-gray-300 text-xs"></span>
{% endif %}
</td>
<!-- Actions -->
<td class="px-4 py-4 text-right">
<div class="flex items-center justify-end gap-2">
<form method="post" action="{% url 'netscan-trigger' p.pk %}">
{% csrf_token %}
<button type="submit"
class="inline-flex items-center px-3 py-1.5 bg-red-600 text-white text-xs rounded-md hover:bg-red-700 font-medium">
<i class="fas fa-play mr-1.5"></i> Run
</button>
</form>
<a href="{% url 'netscan-run-list' p.pk %}"
class="inline-flex items-center px-3 py-1.5 bg-gray-100 text-gray-700 text-xs rounded-md hover:bg-gray-200">
<i class="fas fa-history mr-1.5"></i> History
</a>
<a href="{% url 'netscan-profile-edit' p.pk %}"
class="inline-flex items-center px-3 py-1.5 bg-gray-100 text-gray-700 text-xs rounded-md hover:bg-gray-200">
<i class="fas fa-edit mr-1.5"></i> Edit
</a>
<a href="{% url 'netscan-profile-delete' p.pk %}"
class="inline-flex items-center px-2 py-1.5 bg-red-50 text-red-500 text-xs rounded-md hover:bg-red-100">
<i class="fas fa-trash"></i>
</a>
</div>
</td>
</tr>
{% endwith %}
{% endfor %}
</tbody>
</table>
{% endif %}
</div>
</div>
{% endblock %}
@@ -0,0 +1,29 @@
{% extends 'base.html' %}
{% block content %}
<div class="max-w-lg mx-auto px-4 py-12">
<div class="bg-white rounded-2xl shadow-sm border border-gray-200 p-8 text-center">
<div class="w-16 h-16 bg-red-100 rounded-full flex items-center justify-center mx-auto mb-4">
<i class="fas fa-trash text-red-600 text-xl"></i>
</div>
<h1 class="text-xl font-bold text-gray-900 mb-2">Delete Scan Profile</h1>
<p class="text-gray-600 mb-6">
Are you sure you want to delete <strong>{{ object.name }}</strong>?
All scan runs and findings for this profile will be permanently deleted.
</p>
<form method="post">
{% csrf_token %}
<div class="flex justify-center gap-3">
<button type="submit"
class="px-6 py-2.5 bg-red-600 text-white rounded-lg hover:bg-red-700 font-medium">
Delete
</button>
<a href="{% url 'netscan-dashboard' %}"
class="px-6 py-2.5 bg-gray-100 text-gray-700 rounded-lg hover:bg-gray-200 font-medium">
Cancel
</a>
</div>
</form>
</div>
</div>
{% endblock %}
+234
View File
@@ -0,0 +1,234 @@
{% extends 'base.html' %}
{% block content %}
<div class="max-w-4xl mx-auto px-4 sm:px-6 lg:px-8 py-8">
<div class="bg-white shadow-md rounded-lg overflow-hidden">
<!-- Page header -->
<div class="px-4 py-5 sm:px-6 bg-gray-50 border-b border-gray-200 flex items-center justify-between">
<h1 class="text-2xl font-bold text-gray-900 flex items-center">
<i class="fas fa-shield-alt mr-3 text-red-500"></i>
{{ form_title }}
</h1>
<a href="{% url 'netscan-dashboard' %}" class="text-sm text-gray-500 hover:text-gray-700">
<i class="fas fa-arrow-left mr-1"></i> Back
</a>
</div>
<form method="post" id="profile-form" class="px-4 py-5 sm:p-6 space-y-8">
{% csrf_token %}
<!-- General -->
<div>
<h2 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-4">General</h2>
<div class="grid grid-cols-1 sm:grid-cols-2 gap-4">
<!-- Name (full width) -->
<div class="sm:col-span-2">
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_name">Name</label>
{{ form.name }}
{% for error in form.name.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<!-- Enabled -->
<div class="flex items-center gap-2">
{{ form.enabled }}
<label class="text-sm font-medium text-gray-700" for="id_enabled">Enabled</label>
</div>
<!-- Schedule interval -->
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_schedule_interval">Scan interval</label>
{{ form.schedule_interval }}
{% for error in form.schedule_interval.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
</div>
</div>
<hr class="border-gray-100">
<!-- Network -->
<div>
<div class="flex items-center justify-between mb-4">
<h2 class="text-sm font-semibold text-gray-500 uppercase tracking-wide">Network</h2>
<button type="button" id="auto-detect-btn"
class="inline-flex items-center px-3 py-1.5 bg-blue-50 text-blue-700 rounded-lg hover:bg-blue-100 text-sm font-medium border border-blue-200">
<i class="fas fa-magic mr-2"></i>
Auto-detect
</button>
</div>
<div class="grid grid-cols-1 sm:grid-cols-2 gap-4">
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_gateway_ip">Gateway IP</label>
{{ form.gateway_ip }}
{% for error in form.gateway_ip.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_public_ip">Public IP</label>
{{ form.public_ip }}
{% for error in form.public_ip.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_network_cidr">Network CIDR</label>
{{ form.network_cidr }}
{% if form.network_cidr.help_text %}<p class="text-xs text-gray-400 mt-1">{{ form.network_cidr.help_text }}</p>{% endif %}
{% for error in form.network_cidr.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_auth_provider_host">Auth provider host</label>
{{ form.auth_provider_host }}
{% if form.auth_provider_host.help_text %}<p class="text-xs text-gray-400 mt-1">{{ form.auth_provider_host.help_text }}</p>{% endif %}
{% for error in form.auth_provider_host.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_domains_text">Domains</label>
{{ form.domains_text }}
{% if form.domains_text.help_text %}<p class="text-xs text-gray-400 mt-1">{{ form.domains_text.help_text }}</p>{% endif %}
{% for error in form.domains_text.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_cameras_text">Camera IPs</label>
{{ form.cameras_text }}
{% if form.cameras_text.help_text %}<p class="text-xs text-gray-400 mt-1">{{ form.cameras_text.help_text }}</p>{% endif %}
{% for error in form.cameras_text.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
</div>
</div>
<hr class="border-gray-100">
<!-- Telegram Notifications -->
<div>
<h2 class="text-sm font-semibold text-gray-500 uppercase tracking-wide mb-4">Telegram Notifications</h2>
<div class="grid grid-cols-1 sm:grid-cols-2 gap-4">
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_telegram_bot_token">Bot token</label>
{{ form.telegram_bot_token }}
{% for error in form.telegram_bot_token.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_telegram_chat_id">Chat ID</label>
{{ form.telegram_chat_id }}
{% for error in form.telegram_chat_id.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
<div>
<label class="block text-sm font-medium text-gray-700 mb-1" for="id_notify_on_severity">Notify on severity</label>
{{ form.notify_on_severity }}
{% for error in form.notify_on_severity.errors %}<p class="text-xs text-red-600 mt-1">{{ error }}</p>{% endfor %}
</div>
</div>
{% if object.pk %}
<div class="mt-4">
<button type="button" id="test-telegram-btn"
class="inline-flex items-center px-4 py-2 bg-blue-50 text-blue-700 rounded-lg hover:bg-blue-100 text-sm font-medium border border-blue-200">
<i class="fas fa-paper-plane mr-2"></i>
Test Telegram
</button>
<span id="test-telegram-result" class="ml-3 text-sm hidden"></span>
</div>
{% endif %}
</div>
<!-- Actions -->
<div class="flex justify-end gap-3 pt-2 border-t border-gray-100">
<a href="{% url 'netscan-dashboard' %}"
class="inline-flex justify-center py-2 px-4 border border-transparent shadow-sm text-sm font-medium rounded-md text-gray-700 bg-gray-200 hover:bg-gray-300 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-gray-500">
Cancel
</a>
<button type="submit"
class="inline-flex justify-center py-2 px-4 border border-transparent shadow-sm text-sm font-medium rounded-md text-white bg-red-600 hover:bg-red-700 focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-red-500">
Save Profile
</button>
</div>
</form>
</div>
</div>
{% endblock %}
{% block extra_js %}
<script>
document.getElementById('auto-detect-btn').addEventListener('click', function () {
const btn = this;
btn.disabled = true;
btn.innerHTML = '<i class="fas fa-spinner fa-spin mr-2"></i>Detecting...';
fetch("{% url 'netscan-detect-network' %}")
.then(r => r.json())
.then(data => {
if (data.gateway_ip) document.getElementById('id_gateway_ip').value = data.gateway_ip;
if (data.public_ip) document.getElementById('id_public_ip').value = data.public_ip;
if (data.network_cidr) document.getElementById('id_network_cidr').value = data.network_cidr;
btn.innerHTML = '<i class="fas fa-check mr-2"></i>Detected!';
btn.classList.replace('text-blue-700', 'text-green-700');
btn.classList.replace('bg-blue-50', 'bg-green-50');
btn.classList.replace('border-blue-200', 'border-green-200');
setTimeout(() => {
btn.disabled = false;
btn.innerHTML = '<i class="fas fa-magic mr-2"></i>Auto-detect';
btn.classList.replace('text-green-700', 'text-blue-700');
btn.classList.replace('bg-green-50', 'bg-blue-50');
btn.classList.replace('border-green-200', 'border-blue-200');
}, 3000);
})
.catch(() => {
btn.innerHTML = '<i class="fas fa-exclamation-triangle mr-2"></i>Failed';
setTimeout(() => {
btn.disabled = false;
btn.innerHTML = '<i class="fas fa-magic mr-2"></i>Auto-detect';
}, 3000);
});
});
</script>
{% if object.pk %}
<script>
document.getElementById('test-telegram-btn').addEventListener('click', function () {
const btn = this;
const result = document.getElementById('test-telegram-result');
btn.disabled = true;
btn.innerHTML = '<i class="fas fa-spinner fa-spin mr-2"></i>Sending...';
result.className = 'ml-3 text-sm hidden';
fetch("{% url 'netscan-test-telegram' object.pk %}", {
method: 'POST',
headers: {'X-CSRFToken': '{{ csrf_token }}'},
})
.then(r => r.json())
.then(data => {
result.classList.remove('hidden');
if (data.ok) {
result.className = 'ml-3 text-sm text-green-600';
result.textContent = '✓ Test message sent!';
} else {
result.className = 'ml-3 text-sm text-red-600';
result.textContent = '✗ ' + (data.error || 'Failed');
}
})
.catch(() => {
result.classList.remove('hidden');
result.className = 'ml-3 text-sm text-red-600';
result.textContent = '✗ Network error';
})
.finally(() => {
btn.disabled = false;
btn.innerHTML = '<i class="fas fa-paper-plane mr-2"></i>Test Telegram';
});
});
</script>
{% endif %}
{% endblock %}
+147
View File
@@ -0,0 +1,147 @@
{% extends 'base.html' %}
{% block content %}
<div class="max-w-4xl mx-auto px-4 py-8">
<!-- Breadcrumb -->
<div class="flex items-center gap-2 text-sm text-gray-400 mb-4">
<a href="{% url 'netscan-dashboard' %}" class="hover:text-gray-600">NetScan</a>
<span>/</span>
<a href="{% url 'netscan-run-list' run.profile.pk %}" class="hover:text-gray-600">{{ run.profile.name }}</a>
<span>/</span>
<span class="text-gray-600">Run #{{ run.pk }}</span>
</div>
<!-- Header -->
<div class="bg-white rounded-2xl shadow-sm border border-gray-200 p-6 mb-6">
<div class="flex items-start justify-between flex-wrap gap-4">
<div>
<h1 class="text-xl font-bold text-gray-900">{{ run.profile.name }} — Run #{{ run.pk }}</h1>
<p class="text-gray-500 text-sm mt-1">
Started {{ run.started_at|date:"N j, Y H:i:s" }}
{% if run.duration_seconds is not None %}· {{ run.duration_seconds }}s{% endif %}
· Triggered by <strong>{{ run.triggered_by }}</strong>
</p>
</div>
<div class="flex items-center gap-3">
<span class="inline-flex items-center px-3 py-1 rounded-full text-sm font-medium
{% if run.status == 'success' %}bg-green-100 text-green-700
{% elif run.status == 'failed' %}bg-red-100 text-red-700
{% elif run.status == 'running' %}bg-blue-100 text-blue-700
{% else %}bg-gray-100 text-gray-600{% endif %}">
{{ run.status }}
</span>
<form method="post" action="{% url 'netscan-trigger' run.profile.pk %}">
{% csrf_token %}
<button type="submit"
class="inline-flex items-center px-3 py-1.5 bg-red-600 text-white rounded-lg hover:bg-red-700 font-medium text-sm">
<i class="fas fa-redo mr-1.5"></i> Re-run
</button>
</form>
</div>
</div>
<!-- Summary badges -->
<div class="flex flex-wrap gap-3 mt-4 pt-4 border-t border-gray-100">
{% if run.summary.critical %}
<div class="flex items-center gap-1.5 bg-red-50 text-red-700 px-3 py-1.5 rounded-lg text-sm font-semibold">
🔴 {{ run.summary.critical }} Critical
</div>
{% endif %}
{% if run.summary.warning %}
<div class="flex items-center gap-1.5 bg-yellow-50 text-yellow-700 px-3 py-1.5 rounded-lg text-sm font-semibold">
🟡 {{ run.summary.warning }} Warning
</div>
{% endif %}
{% if run.summary.ok %}
<div class="flex items-center gap-1.5 bg-green-50 text-green-700 px-3 py-1.5 rounded-lg text-sm font-semibold">
🟢 {{ run.summary.ok }} OK
</div>
{% endif %}
{% if run.summary.info %}
<div class="flex items-center gap-1.5 bg-blue-50 text-blue-700 px-3 py-1.5 rounded-lg text-sm font-semibold">
️ {{ run.summary.info }} Info
</div>
{% endif %}
</div>
</div>
<!-- Critical -->
{% if critical_findings %}
<details class="mb-4 open" open>
<summary class="flex items-center gap-2 cursor-pointer bg-red-50 border border-red-200 rounded-xl px-5 py-3 font-semibold text-red-800 select-none">
🔴 Critical Findings ({{ critical_findings|length }})
</summary>
<div class="mt-2 space-y-3">
{% for f in critical_findings %}
<div class="bg-white border border-red-200 rounded-xl p-4">
<div class="font-medium text-gray-900 mb-1">{{ f.title }}</div>
<p class="text-sm text-gray-600">{{ f.detail }}</p>
{% if f.raw %}
<details class="mt-2">
<summary class="text-xs text-gray-400 cursor-pointer hover:text-gray-600">View raw data</summary>
<pre class="mt-2 bg-gray-50 rounded-lg p-3 text-xs overflow-auto text-gray-700">{{ f.raw|pprint }}</pre>
</details>
{% endif %}
</div>
{% endfor %}
</div>
</details>
{% endif %}
<!-- Warnings -->
{% if warning_findings %}
<details class="mb-4">
<summary class="flex items-center gap-2 cursor-pointer bg-yellow-50 border border-yellow-200 rounded-xl px-5 py-3 font-semibold text-yellow-800 select-none">
🟡 Warnings ({{ warning_findings|length }})
</summary>
<div class="mt-2 space-y-3">
{% for f in warning_findings %}
<div class="bg-white border border-yellow-200 rounded-xl p-4">
<div class="font-medium text-gray-900 mb-1">{{ f.title }}</div>
<p class="text-sm text-gray-600">{{ f.detail }}</p>
{% if f.raw %}
<details class="mt-2">
<summary class="text-xs text-gray-400 cursor-pointer hover:text-gray-600">View raw data</summary>
<pre class="mt-2 bg-gray-50 rounded-lg p-3 text-xs overflow-auto text-gray-700">{{ f.raw|pprint }}</pre>
</details>
{% endif %}
</div>
{% endfor %}
</div>
</details>
{% endif %}
<!-- OK / Info -->
{% if ok_findings %}
<details class="mb-4">
<summary class="flex items-center gap-2 cursor-pointer bg-green-50 border border-green-200 rounded-xl px-5 py-3 font-semibold text-green-800 select-none">
🟢 OK / Info ({{ ok_findings|length }})
</summary>
<div class="mt-2 space-y-3">
{% for f in ok_findings %}
<div class="bg-white border border-gray-200 rounded-xl p-4">
<div class="flex items-center gap-2 font-medium text-gray-900 mb-1">
{% if f.severity == 'info' %}️{% else %}🟢{% endif %}
{{ f.title }}
</div>
<p class="text-sm text-gray-600">{{ f.detail }}</p>
{% if f.raw %}
<details class="mt-2">
<summary class="text-xs text-gray-400 cursor-pointer hover:text-gray-600">View raw data</summary>
<pre class="mt-2 bg-gray-50 rounded-lg p-3 text-xs overflow-auto text-gray-700">{{ f.raw|pprint }}</pre>
</details>
{% endif %}
</div>
{% endfor %}
</div>
</details>
{% endif %}
{% if not critical_findings and not warning_findings and not ok_findings %}
<div class="text-center py-12 text-gray-400">
<i class="fas fa-hourglass-half text-3xl mb-2"></i>
<p>No findings recorded yet — the scan may still be running.</p>
</div>
{% endif %}
</div>
{% endblock %}
+98
View File
@@ -0,0 +1,98 @@
{% extends 'base.html' %}
{% block content %}
<div class="max-w-5xl mx-auto px-4 py-8">
<div class="flex items-center justify-between mb-6">
<div>
<a href="{% url 'netscan-dashboard' %}" class="text-gray-400 hover:text-gray-600 text-sm">
<i class="fas fa-arrow-left mr-1"></i> NetScan
</a>
<h1 class="text-2xl font-bold text-gray-900 mt-1">{{ profile.name }} — Scan History</h1>
</div>
<form method="post" action="{% url 'netscan-trigger' profile.pk %}">
{% csrf_token %}
<button type="submit"
class="inline-flex items-center px-4 py-2 bg-red-600 text-white rounded-lg hover:bg-red-700 font-medium text-sm">
<i class="fas fa-play mr-2"></i> Run Now
</button>
</form>
</div>
{% if runs %}
<div class="bg-white rounded-2xl shadow-sm border border-gray-200 overflow-hidden">
<table class="min-w-full text-sm">
<thead class="bg-gray-50 border-b border-gray-100">
<tr>
<th class="px-4 py-3 text-left font-semibold text-gray-600">Started</th>
<th class="px-4 py-3 text-left font-semibold text-gray-600">Duration</th>
<th class="px-4 py-3 text-left font-semibold text-gray-600">Triggered by</th>
<th class="px-4 py-3 text-left font-semibold text-gray-600">Status</th>
<th class="px-4 py-3 text-left font-semibold text-gray-600">Findings</th>
<th class="px-4 py-3"></th>
</tr>
</thead>
<tbody class="divide-y divide-gray-50">
{% for run in runs %}
<tr class="hover:bg-gray-50">
<td class="px-4 py-3 text-gray-800 font-mono text-xs">{{ run.started_at|date:"M d, H:i:s" }}</td>
<td class="px-4 py-3 text-gray-500">
{% if run.duration_seconds is not None %}{{ run.duration_seconds }}s{% else %}—{% endif %}
</td>
<td class="px-4 py-3">
<span class="inline-flex items-center px-2 py-0.5 rounded text-xs font-medium
{% if run.triggered_by == 'manual' %}bg-blue-50 text-blue-700{% else %}bg-gray-100 text-gray-600{% endif %}">
{{ run.triggered_by }}
</span>
</td>
<td class="px-4 py-3">
<span class="inline-flex items-center px-2 py-0.5 rounded text-xs font-medium
{% if run.status == 'success' %}bg-green-100 text-green-700
{% elif run.status == 'failed' %}bg-red-100 text-red-700
{% elif run.status == 'running' %}bg-blue-100 text-blue-700
{% else %}bg-gray-100 text-gray-600{% endif %}">
{{ run.status }}
</span>
</td>
<td class="px-4 py-3">
<span class="flex items-center gap-2">
{% if run.summary.critical %}<span class="text-red-600 font-medium">🔴{{ run.summary.critical }}</span>{% endif %}
{% if run.summary.warning %}<span class="text-yellow-600 font-medium">🟡{{ run.summary.warning }}</span>{% endif %}
{% if run.summary.ok %}<span class="text-green-600 font-medium">🟢{{ run.summary.ok }}</span>{% endif %}
{% if run.summary.info %}<span class="text-blue-600 font-medium">️{{ run.summary.info }}</span>{% endif %}
</span>
</td>
<td class="px-4 py-3">
<a href="{% url 'netscan-run-detail' run.pk %}"
class="text-blue-600 hover:text-blue-800 text-xs font-medium">View →</a>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<!-- Pagination -->
{% if is_paginated %}
<div class="flex justify-center mt-6 gap-2">
{% if page_obj.has_previous %}
<a href="?page={{ page_obj.previous_page_number }}"
class="px-3 py-1.5 bg-white border border-gray-200 rounded-lg text-sm hover:bg-gray-50">← Prev</a>
{% endif %}
<span class="px-3 py-1.5 text-sm text-gray-600">
Page {{ page_obj.number }} of {{ page_obj.paginator.num_pages }}
</span>
{% if page_obj.has_next %}
<a href="?page={{ page_obj.next_page_number }}"
class="px-3 py-1.5 bg-white border border-gray-200 rounded-lg text-sm hover:bg-gray-50">Next →</a>
{% endif %}
</div>
{% endif %}
{% else %}
<div class="text-center py-16 text-gray-400">
<i class="fas fa-history text-4xl mb-3"></i>
<p>No scan runs yet. Click <strong>Run Now</strong> to start.</p>
</div>
{% endif %}
</div>
{% endblock %}
+14
View File
@@ -0,0 +1,14 @@
from django.urls import path
from . import views
urlpatterns = [
path('', views.DashboardView.as_view(), name='netscan-dashboard'),
path('profile/new/', views.ProfileCreateView.as_view(), name='netscan-profile-create'),
path('profile/<int:pk>/edit/', views.ProfileUpdateView.as_view(), name='netscan-profile-edit'),
path('profile/<int:pk>/delete/', views.ProfileDeleteView.as_view(), name='netscan-profile-delete'),
path('profile/<int:pk>/runs/', views.ScanRunListView.as_view(), name='netscan-run-list'),
path('profile/<int:pk>/trigger/', views.TriggerScanView.as_view(), name='netscan-trigger'),
path('profile/<int:pk>/test-telegram/', views.TestTelegramView.as_view(), name='netscan-test-telegram'),
path('detect-network/', views.DetectNetworkView.as_view(), name='netscan-detect-network'),
path('run/<int:pk>/', views.ScanRunDetailView.as_view(), name='netscan-run-detail'),
]
+215
View File
@@ -0,0 +1,215 @@
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)