links/NETSCAN_PLAN.md

NetScan — Home Network Security Scanner

A new Django sub-app (netscan) added to the Links project that runs configurable, scheduled security checks against your home network, stores results in SQLite, sends Telegram alerts for critical findings, and surfaces everything through a Tailwind UI inside the existing mini-apps section.

---

Context & Constraints

• Codebase: Django + APScheduler + Tailwind + SQLite (PostgreSQL in prod via env var)
• Deployment: Same Docker image as the Links app, deployed in K3s at 192.168.1.2 (server-3)
• Why in-cluster: The scanner must run inside the LAN to probe 192.168.1.x addresses (router, cameras). Running on the cluster node satisfies this automatically.
• No new dependencies except requests (already present). All checks use stdlib socket, ssl, subprocess, struct.
• No new JS framework — pure Django templates + Tailwind, matching the rest of the app.

---

Database Models (netscan/models.py)

ScanProfile

Field	Type	Notes
name	CharField	Human label e.g. "Home Network"
enabled	BooleanField	Controls scheduler
schedule_interval	IntegerField	1 / 3 / 7 / 30 days (choices)
gateway_ip	GenericIPAddressField	e.g. 192.168.1.1
public_ip	GenericIPAddressField	e.g. 14.137.198.99
network_cidr	CharField	e.g. 192.168.1.0/24
auth_provider_host	CharField	e.g. pass.junv.cc (for ingress check)
domains	JSONField	List of public hostnames to check
cameras	JSONField	List of camera IPs to probe
telegram_bot_token	CharField	blank/null; stored encrypted in env preferred
telegram_chat_id	CharField	blank/null
notify_on_severity	CharField	"warning" or "critical" (default "critical")
last_run_at	DateTimeField	null
created_at	DateTimeField	auto

ScanRun

Field	Type	Notes
profile	FK → ScanProfile	cascade delete
started_at	DateTimeField
finished_at	DateTimeField	null
status	CharField	pending / running / success / failed
summary	JSONField	{ok:N, info:N, warning:N, critical:N}
triggered_by	CharField	"scheduler" or "manual"

ScanFinding

Field	Type	Notes
run	FK → ScanRun	cascade delete
check_name	CharField	e.g. "router_ports", "tls_expiry"
severity	CharField	ok / info / warning / critical
title	CharField	Short human summary
detail	TextField	Full explanation
raw	JSONField	Raw probe output for debugging

---

Check Modules (netscan/checks/)

Each module exposes a single function run(profile) -> list[Finding]. All use only stdlib + requests.

checks/base.py

@dataclass
class Finding:
    check_name: str
    severity: str        # ok | info | warning | critical
    title: str
    detail: str
    raw: dict

checks/router.py — Gateway security

• TCP connect-probe gateway IP on ports: 22, 23, 53, 80, 139, 443, 445, 8080, 8443
• Fetch HTTP headers from port 80 to identify plain-HTTP admin (Server: httpd)
• Flag: SMB open (139/445) → warning
• Flag: SSH closed → ok; SSH open from WAN → warning
• Flag: plain-HTTP admin (no HTTPS on 443) → warning
• Flag: unknown port 8080 open → info

checks/dns.py — Open resolver (WAN DNS exposure)

• Send raw DNS query for google.com A to profile.public_ip:53 over UDP using socket + struct
• Parse the response: if NOERROR + answer section returned → recursion available
• Flag: open recursive resolver → critical
• Note in detail: "Verify from off-LAN; NAT hairpin may cause false positive"

checks/ingress.py — Public domain auth verification

• For each domain in profile.domains:
  • requests.get(f"https://{domain}/", allow_redirects=True, timeout=8)
  • Check: does the redirect chain pass through profile.auth_provider_host?
  • Check: does the final URL (after all redirects) contain the auth provider host?
  • If final URL is the app itself (not the auth host) → critical (auth bypassed)
  • If redirect goes through auth provider → ok
  • If connection refused / DNS fails → warning

checks/cameras.py — RTSP unauthenticated access

• For each IP in profile.cameras:
  1. TCP connect to port 554 — if closed → info (skip)
  2. Send OPTIONS rtsp://{ip}/ RTSP/1.0\r\nCSeq: 1\r\n\r\n over raw socket
  3. Parse response status line
  4. If OPTIONS returns 200, send DESCRIBE rtsp://{ip}/ RTSP/1.0\r\nCSeq: 2\r\nAccept: application/sdp\r\n\r\n
  5. If DESCRIBE returns 200 (stream accessible with no creds) → critical
  6. If DESCRIBE returns 401/403 → ok (auth required)
  7. If OPTIONS returns 404 / no common path accessible → ok

checks/tls.py — TLS certificate validity

• For each domain in profile.domains:
  • ssl.get_server_certificate((domain, 443)) + parse notAfter
  • Days to expiry < 0 → critical (expired)
  • Days to expiry < 14 → critical
  • Days to expiry < 30 → warning
  • Otherwise → ok
  • Also flag if cert CN/SAN doesn't match the domain → warning

checks/ports.py — Public IP port exposure

• TCP connect-probe profile.public_ip on: 22, 23, 25, 53, 80, 443, 3306, 5432, 6379, 8080, 8443
• Flag presence of each open port with canned risk descriptions:
  • 80/443 → info (expected for web services)
  • 22 → warning (SSH exposed to internet)
  • 23 → critical (Telnet exposed)
  • 53 → warning (DNS — run dns check to confirm resolver)
  • Database ports (3306/5432/6379) → critical
  • 8443/8080 → warning (alt web ports)

---

Scanner Orchestrator (netscan/scanner.py)

def run_scan(profile_id: int) -> int:
    """
    Runs 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=...)

    all_findings = []
    check_modules = [router, dns, ingress, cameras, tls, ports]
    for mod in check_modules:
        try:
            findings = mod.run(profile)
            all_findings.extend(findings)
        except Exception as e:
            # Wrap uncaught errors as a warning finding so run still completes
            all_findings.append(Finding(check_name=mod.__name__, severity='warning',
                                        title='Check errored', detail=str(e), raw={}))

    # Persist findings
    ScanFinding.objects.bulk_create([...])

    # Update run summary
    summary = Counter(f.severity for f in all_findings)
    run.summary = dict(summary)
    run.status = 'success'
    run.finished_at = now()
    run.save()

    # Telegram notification
    if profile.telegram_bot_token and profile.telegram_chat_id:
        notify_telegram(profile, run, all_findings)

    # Update profile.last_run_at
    profile.last_run_at = now()
    profile.save(update_fields=['last_run_at'])

    return run.pk

---

Telegram Notifications (netscan/notifications.py)

def notify_telegram(profile, run, findings):
    """
    Sends a Telegram message if any finding meets or exceeds notify_on_severity.
    Uses the Bot API sendMessage endpoint directly via requests (no library needed).
    """
    threshold_order = ['ok', 'info', 'warning', 'critical']
    threshold_idx = threshold_order.index(profile.notify_on_severity)

    flagged = [f for f in findings
               if threshold_order.index(f.severity) >= threshold_idx]
    if not flagged:
        return

    lines = [f"🔒 *NetScan Alert* — {profile.name}",
             f"Run #{run.pk} finished at {run.finished_at:%Y-%m-%d %H:%M}",
             f"Summary: {run.summary}",
             ""]
    for f in flagged[:10]:   # cap at 10 to stay under TG message limit
        icon = {'critical': '🔴', 'warning': '🟡', 'ok': '🟢', 'info': 'ℹ️'}[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"
    requests.post(url, json={
        "chat_id": profile.telegram_chat_id,
        "text": text,
        "parse_mode": "Markdown"
    }, timeout=10)

---

APScheduler Integration (netscan/tasks.py + netscan/apps.py)

tasks.py

from core.scheduler import scheduler
from apscheduler.triggers.interval import IntervalTrigger

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],
        replace_existing=True
    )

def unschedule_profile(profile):
    job_id = f'netscan_profile_{profile.pk}'
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)

apps.py

class NetscanConfig(AppConfig):
    name = 'netscan'

    def ready(self):
        from netscan.tasks import schedule_profile
        from netscan.models import ScanProfile
        for profile in ScanProfile.objects.filter(enabled=True):
            schedule_profile(profile)

signals.py

@receiver(post_save, sender=ScanProfile)
def reschedule_on_save(sender, instance, **kwargs):
    if instance.enabled:
        schedule_profile(instance)
    else:
        unschedule_profile(instance)

---

Views (netscan/views.py)

All views use LoginRequiredMixin.

View	URL	Notes
DashboardView	/ui/netscan/	List profiles, worst-severity badge per profile, last run time, Run Now + Edit buttons
ProfileCreateView	/ui/netscan/profile/new/	ModelForm
ProfileUpdateView	/ui/netscan/profile/<pk>/edit/	ModelForm
ProfileDeleteView	/ui/netscan/profile/<pk>/delete/	Confirm page
ScanRunListView	/ui/netscan/profile/<pk>/runs/	Paginated history, status + severity count cols
ScanRunDetailView	/ui/netscan/run/<pk>/	Findings grouped by severity, collapsible raw JSON
TriggerScanView	/ui/netscan/profile/<pk>/trigger/	POST-only; spawns Thread(target=run_scan, args=[pk]), redirects to run list
TestTelegramView	/ui/netscan/profile/<pk>/test-telegram/	POST-only; sends a test message, returns JSON

---

Templates (netscan/templates/netscan/)

All extend base.html, use Tailwind classes matching the existing app.

dashboard.html

• Grid of profile cards (matches mini_apps card style)
• Each card: name, schedule chip (e.g. "Every 7 days"), last run timestamp, worst-severity badge (🔴/🟡/🟢), finding count breakdown
• "Run Now" button (POST to trigger URL), "Edit" link, "History" link
• Empty state with "Create your first scan profile" CTA

profile_form.html

• Fields: Name, Schedule (dropdown: 1/3/7/30 days), Gateway IP, Public IP, Network CIDR, Auth Provider Host, Domains (textarea, one per line), Camera IPs (textarea, one per line), Telegram Bot Token, Telegram Chat ID, Notify on Severity (dropdown: warning/critical), Enabled checkbox
• "Test Telegram" button (JS fetch to TestTelegramView, shows inline success/error)

run_list.html

• Table: Started, Duration, Triggered by, Status badge, 🔴 Critical, 🟡 Warning, 🟢 OK counts, View link
• Pagination

run_detail.html

• Header: profile name, run timestamp, status, summary badges, "Re-run" button
• Three collapsible sections: Critical findings, Warnings, OK/Info
• Each finding: title, detail text; expandable "Raw" disclosure showing JSON
• Back to history link

---

URL Wiring

core/urls.py — add:

path('ui/netscan/', include('netscan.urls')),

netscan/urls.py:

urlpatterns = [
    path('', DashboardView.as_view(), name='netscan-dashboard'),
    path('profile/new/', ProfileCreateView.as_view(), name='netscan-profile-create'),
    path('profile/<int:pk>/edit/', ProfileUpdateView.as_view(), name='netscan-profile-edit'),
    path('profile/<int:pk>/delete/', ProfileDeleteView.as_view(), name='netscan-profile-delete'),
    path('profile/<int:pk>/runs/', ScanRunListView.as_view(), name='netscan-run-list'),
    path('profile/<int:pk>/trigger/', TriggerScanView.as_view(), name='netscan-trigger'),
    path('profile/<int:pk>/test-telegram/', TestTelegramView.as_view(), name='netscan-test-telegram'),
    path('run/<int:pk>/', ScanRunDetailView.as_view(), name='netscan-run-detail'),
]

core/settings.py — add to INSTALLED_APPS:

'netscan',

links/mini_apps_views.py — add to mini_apps list:

{
    'name': 'NetScan',
    'description': 'Scheduled home network security scanner. Checks router exposure, DNS, TLS certs, public ingress auth, and camera access.',
    'url': 'netscan-dashboard',
    'thumbnail': 'https://images.unsplash.com/photo-1558494949-ef010cbdcc31?w=400&h=300&fit=crop',
    'icon': 'fas fa-shield-alt',
    'color': '#e74c3c'
},

---

K3s Deployment Changes (k8s/manifest.yaml)

The scanner runs inside the same Pod as the Links app — no new container or service needed. The existing Deployment only needs two new env vars (or they can be set per-profile in the DB):

# Optional: global fallback Telegram credentials
- name: NETSCAN_TELEGRAM_BOT_TOKEN
  valueFrom:
    secretKeyRef:
      name: netscan-credentials
      key: telegram_bot_token
- name: NETSCAN_TELEGRAM_CHAT_ID
  valueFrom:
    secretKeyRef:
      name: netscan-credentials
      key: telegram_chat_id

Create the secret:

kubectl create secret generic netscan-credentials \
  --from-literal=telegram_bot_token=YOUR_TOKEN \
  --from-literal=telegram_chat_id=YOUR_CHAT_ID \
  -n home-apps

The Pod already runs inside the LAN (on server-3 at 192.168.1.2), so it can reach 192.168.1.1 (router), 192.168.1.70–250 (cameras), and the public IP.

---

File Checklist

File	Action
netscan/__init__.py	create (empty)
netscan/apps.py	create
netscan/models.py	create
netscan/admin.py	create (register all 3 models)
netscan/checks/__init__.py	create (empty)
netscan/checks/base.py	create
netscan/checks/router.py	create
netscan/checks/dns.py	create
netscan/checks/ingress.py	create
netscan/checks/cameras.py	create
netscan/checks/tls.py	create
netscan/checks/ports.py	create
netscan/scanner.py	create
netscan/notifications.py	create
netscan/tasks.py	create
netscan/signals.py	create
netscan/forms.py	create
netscan/views.py	create
netscan/urls.py	create
netscan/migrations/0001_initial.py	create (via makemigrations)
netscan/templates/netscan/dashboard.html	create
netscan/templates/netscan/profile_form.html	create
netscan/templates/netscan/run_list.html	create
netscan/templates/netscan/run_detail.html	create
core/settings.py	add 'netscan' to INSTALLED_APPS
core/urls.py	add path('ui/netscan/', include('netscan.urls'))
links/mini_apps_views.py	add NetScan entry to mini_apps list
k8s/manifest.yaml	add env vars for Telegram credentials

---

Verification Steps

1. python manage.py makemigrations netscan && python manage.py migrate — no errors
2. python manage.py runserver — navigate to /ui/netscan/ — dashboard loads
3. Create a ScanProfile via the form, ensure all fields save
4. Click "Run Now" — ScanRun + ScanFinding rows appear in DB; run_detail page shows them
5. Set schedule to 1 day, save — scheduler.get_jobs() shows a netscan_profile_N job
6. Toggle profile disabled — job is removed from scheduler
7. If Telegram configured, click "Test Telegram" — message appears in chat
8. After a scheduled run: Telegram alert arrives for any critical/warning findings
9. All views return 302 to login when accessed without auth
10. kubectl apply -f k8s/manifest.yaml — Pod starts cleanly with new env vars

---

Implementation Order (for LLM session in links workspace)

1. Phase 1 (parallel): models + migration + app registration + all check modules (no inter-dependencies)
2. Phase 2: scanner.py + notifications.py (depends on models + checks)
3. Phase 3: tasks.py + apps.py + signals.py (depends on scanner)
4. Phase 4: views.py + urls.py + forms.py (depends on models)
5. Phase 5: all 4 templates (depends on views)
6. Phase 6: wire into core/settings, core/urls, mini_apps_views, k8s/manifest