mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
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
|