107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .config import DEFAULT_SUBNET
|
|
from .db import init_db
|
|
from .scanner import discover_hosts, scan_host
|
|
from .service import (
|
|
complete_scan,
|
|
create_scan,
|
|
fetch_device,
|
|
fetch_devices,
|
|
fetch_scans,
|
|
mark_missing_devices,
|
|
scan_state,
|
|
upsert_host,
|
|
)
|
|
|
|
app = FastAPI(title="NetTrak")
|
|
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
|
|
|
|
|
@app.on_event("startup")
|
|
def startup() -> None:
|
|
init_db()
|
|
|
|
|
|
@app.get("/")
|
|
def home() -> FileResponse:
|
|
return FileResponse("app/static/index.html")
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict[str, Any]:
|
|
progress = scan_state.snapshot()
|
|
return {
|
|
"status": "ok",
|
|
"scan_running": progress["running"],
|
|
"current_scan_id": progress["scan_id"],
|
|
"scan_progress": progress,
|
|
}
|
|
|
|
|
|
@app.get("/api/devices")
|
|
def api_devices() -> list[dict]:
|
|
return fetch_devices()
|
|
|
|
|
|
@app.get("/api/devices/{device_id}")
|
|
def api_device(device_id: int) -> dict:
|
|
device = fetch_device(device_id)
|
|
if not device:
|
|
raise HTTPException(status_code=404, detail="Device not found")
|
|
return device
|
|
|
|
|
|
@app.get("/api/scans")
|
|
def api_scans(limit: int = 20) -> list[dict]:
|
|
return fetch_scans(limit=limit)
|
|
|
|
|
|
@app.post("/api/scans/run")
|
|
def run_scan(subnet: str | None = None) -> dict[str, Any]:
|
|
subnet = subnet or DEFAULT_SUBNET
|
|
scan_id = create_scan(subnet)
|
|
if not scan_state.start(scan_id, subnet):
|
|
complete_scan(scan_id, "cancelled", 0, notes="Another scan was already running")
|
|
raise HTTPException(status_code=409, detail="Scan already running")
|
|
|
|
def worker() -> None:
|
|
host_count = 0
|
|
try:
|
|
discovered = discover_hosts(subnet)
|
|
scan_state.set_total_hosts(len(discovered))
|
|
|
|
for idx, host in enumerate(discovered, start=1):
|
|
scan_state.set_current_host(host["ip"])
|
|
detailed = scan_host(host["ip"])
|
|
if not detailed:
|
|
scan_state.update_progress(idx, host_count)
|
|
continue
|
|
host_count += 1
|
|
if not detailed.hostname and host.get("hostname"):
|
|
detailed.hostname = host["hostname"]
|
|
if not detailed.mac and host.get("mac"):
|
|
detailed.mac = host["mac"]
|
|
if not detailed.vendor and host.get("vendor"):
|
|
detailed.vendor = host["vendor"]
|
|
|
|
upsert_host(scan_id, detailed)
|
|
scan_state.update_progress(idx, host_count)
|
|
|
|
mark_missing_devices(scan_id)
|
|
complete_scan(scan_id, "completed", host_count)
|
|
except Exception as exc:
|
|
complete_scan(scan_id, "failed", host_count, notes=str(exc))
|
|
finally:
|
|
scan_state.finish()
|
|
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
return {"status": "started", "scan_id": scan_id, "subnet": subnet}
|