Add scan queue execution and listing endpoint.

Run scans in background with progress/status updates and store scan metrics in memory or PostgreSQL. Made-with: Cursor
2026-04-09 21:49:38 +10:00
parent 284794ec8d
commit 43cb339605
12 changed files with 753 additions and 23 deletions
@@ -0,0 +1,282 @@
+package scans
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"sort"
+	"time"
+)
+
+type PostgresStore struct {
+	db *sql.DB
+}
+
+func NewPostgresStore(db *sql.DB) *PostgresStore {
+	return &PostgresStore{db: db}
+}
+
+func (s *PostgresStore) CreateScan(req CreateScanRequest) (ScanJob, error) {
+	if err := validateCreateScanRequest(req); err != nil {
+		return ScanJob{}, err
+	}
+	applyDefaultOptions(&req.Options)
+
+	id, err := newID()
+	if err != nil {
+		return ScanJob{}, err
+	}
+
+	now := time.Now().UTC()
+	job := ScanJob{
+		ID:                id,
+		Name:              req.Name,
+		Status:            "queued",
+		CIDRs:             req.CIDRs,
+		ExcludeIPs:        req.ExcludeIPs,
+		SNMPCredentialsID: req.SNMPCredentialsID,
+		Options:           req.Options,
+		Progress:          0,
+		Stats:             ScanStats{},
+		CreatedAt:         now,
+	}
+
+	cidrsJSON, err := json.Marshal(job.CIDRs)
+	if err != nil {
+		return ScanJob{}, err
+	}
+	excludeJSON, err := json.Marshal(job.ExcludeIPs)
+	if err != nil {
+		return ScanJob{}, err
+	}
+	optionsJSON, err := json.Marshal(job.Options)
+	if err != nil {
+		return ScanJob{}, err
+	}
+	statsJSON, err := json.Marshal(job.Stats)
+	if err != nil {
+		return ScanJob{}, err
+	}
+
+	var snmpID any
+	if job.SNMPCredentialsID != "" {
+		snmpID = job.SNMPCredentialsID
+	}
+
+	query := `
+insert into scan_jobs (
+  id, name, status, cidrs, exclude_ips, options, progress, stats, snmp_credentials_id, created_at
+) values (
+  $1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8::jsonb, $9, $10
+)`
+
+	_, err = s.db.ExecContext(context.Background(), query,
+		job.ID,
+		job.Name,
+		job.Status,
+		string(cidrsJSON),
+		string(excludeJSON),
+		string(optionsJSON),
+		job.Progress,
+		string(statsJSON),
+		snmpID,
+		job.CreatedAt,
+	)
+	if err != nil {
+		return ScanJob{}, err
+	}
+
+	return job, nil
+}
+
+func (s *PostgresStore) GetScan(id string) (ScanJob, bool) {
+	query := `
+select id, name, status, cidrs::text, exclude_ips::text, options::text,
+       progress, stats::text, coalesce(snmp_credentials_id::text, ''),
+       created_at, started_at, finished_at
+from scan_jobs
+where id = $1
+limit 1`
+
+	var job ScanJob
+	var cidrsRaw string
+	var excludeRaw string
+	var optionsRaw string
+	var statsRaw string
+	var startedAt sql.NullTime
+	var finishedAt sql.NullTime
+
+	err := s.db.QueryRowContext(context.Background(), query, id).Scan(
+		&job.ID,
+		&job.Name,
+		&job.Status,
+		&cidrsRaw,
+		&excludeRaw,
+		&optionsRaw,
+		&job.Progress,
+		&statsRaw,
+		&job.SNMPCredentialsID,
+		&job.CreatedAt,
+		&startedAt,
+		&finishedAt,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ScanJob{}, false
+	}
+	if err != nil {
+		return ScanJob{}, false
+	}
+
+	if err = json.Unmarshal([]byte(cidrsRaw), &job.CIDRs); err != nil {
+		return ScanJob{}, false
+	}
+	if err = json.Unmarshal([]byte(excludeRaw), &job.ExcludeIPs); err != nil {
+		return ScanJob{}, false
+	}
+	if err = json.Unmarshal([]byte(optionsRaw), &job.Options); err != nil {
+		return ScanJob{}, false
+	}
+	if err = json.Unmarshal([]byte(statsRaw), &job.Stats); err != nil {
+		return ScanJob{}, false
+	}
+	if startedAt.Valid {
+		job.StartedAt = startedAt.Time
+	}
+	if finishedAt.Valid {
+		job.FinishedAt = finishedAt.Time
+	}
+
+	return job, true
+}
+
+func (s *PostgresStore) ListScans(limit int) []ScanJob {
+	if limit <= 0 {
+		limit = 50
+	}
+	query := `
+select id, name, status, cidrs::text, exclude_ips::text, options::text, progress,
+       stats::text, coalesce(snmp_credentials_id::text, ''), created_at, started_at, finished_at
+from scan_jobs
+order by created_at desc
+limit $1`
+
+	rows, err := s.db.QueryContext(context.Background(), query, limit)
+	if err != nil {
+		return nil
+	}
+	defer rows.Close()
+
+	jobs := make([]ScanJob, 0, limit)
+	for rows.Next() {
+		var job ScanJob
+		var cidrsRaw string
+		var excludeRaw string
+		var optionsRaw string
+		var statsRaw string
+		var startedAt sql.NullTime
+		var finishedAt sql.NullTime
+
+		err = rows.Scan(
+			&job.ID,
+			&job.Name,
+			&job.Status,
+			&cidrsRaw,
+			&excludeRaw,
+			&optionsRaw,
+			&job.Progress,
+			&statsRaw,
+			&job.SNMPCredentialsID,
+			&job.CreatedAt,
+			&startedAt,
+			&finishedAt,
+		)
+		if err != nil {
+			return nil
+		}
+
+		if json.Unmarshal([]byte(cidrsRaw), &job.CIDRs) != nil ||
+			json.Unmarshal([]byte(excludeRaw), &job.ExcludeIPs) != nil ||
+			json.Unmarshal([]byte(optionsRaw), &job.Options) != nil ||
+			json.Unmarshal([]byte(statsRaw), &job.Stats) != nil {
+			return nil
+		}
+		if startedAt.Valid {
+			job.StartedAt = startedAt.Time
+		}
+		if finishedAt.Valid {
+			job.FinishedAt = finishedAt.Time
+		}
+		jobs = append(jobs, job)
+	}
+
+	sort.Slice(jobs, func(i, j int) bool {
+		return jobs[i].CreatedAt.After(jobs[j].CreatedAt)
+	})
+	return jobs
+}
+
+func (s *PostgresStore) UpdateScan(id string, update ScanUpdate) (ScanJob, bool) {
+	current, ok := s.GetScan(id)
+	if !ok {
+		return ScanJob{}, false
+	}
+
+	if update.Status != "" {
+		current.Status = update.Status
+	}
+	if update.Progress >= 0 {
+		current.Progress = update.Progress
+	}
+	current.Stats = update.Stats
+	if update.StartedAt != nil {
+		current.StartedAt = *update.StartedAt
+	}
+	if update.FinishedAt != nil {
+		current.FinishedAt = *update.FinishedAt
+	}
+
+	optionsJSON, err := json.Marshal(current.Options)
+	if err != nil {
+		return ScanJob{}, false
+	}
+	statsJSON, err := json.Marshal(current.Stats)
+	if err != nil {
+		return ScanJob{}, false
+	}
+
+	var startedAt any
+	if !current.StartedAt.IsZero() {
+		startedAt = current.StartedAt
+	}
+	var finishedAt any
+	if !current.FinishedAt.IsZero() {
+		finishedAt = current.FinishedAt
+	}
+
+	query := `
+update scan_jobs
+set status = $2,
+    options = $3::jsonb,
+    progress = $4,
+    stats = $5::jsonb,
+    started_at = $6,
+    finished_at = $7
+where id = $1`
+	_, err = s.db.ExecContext(
+		context.Background(),
+		query,
+		id,
+		current.Status,
+		string(optionsJSON),
+		current.Progress,
+		string(statsJSON),
+		startedAt,
+		finishedAt,
+	)
+	if err != nil {
+		return ScanJob{}, false
+	}
+
+	return current, true
+}
@@ -0,0 +1,184 @@
+package scans
+
+import (
+	"log"
+	"net"
+	"os/exec"
+	"runtime"
+	"strconv"
+	"sync"
+	"time"
+)
+
+type Runner struct {
+	store Store
+}
+
+func NewRunner(store Store) *Runner {
+	return &Runner{store: store}
+}
+
+func (r *Runner) Start(job ScanJob) {
+	go r.run(job)
+}
+
+func (r *Runner) run(job ScanJob) {
+	started := time.Now().UTC()
+	_, _ = r.store.UpdateScan(job.ID, ScanUpdate{
+		Status:    "running",
+		Progress:  0,
+		Stats:     ScanStats{},
+		StartedAt: &started,
+	})
+
+	ips, err := expandTargets(job.CIDRs, job.ExcludeIPs)
+	if err != nil {
+		finished := time.Now().UTC()
+		_, _ = r.store.UpdateScan(job.ID, ScanUpdate{
+			Status:     "failed",
+			Progress:   100,
+			Stats:      ScanStats{},
+			FinishedAt: &finished,
+		})
+		return
+	}
+
+	total := len(ips)
+	if total == 0 {
+		finished := time.Now().UTC()
+		_, _ = r.store.UpdateScan(job.ID, ScanUpdate{
+			Status:     "done",
+			Progress:   100,
+			Stats:      ScanStats{HostsTotal: 0, HostsUp: 0},
+			FinishedAt: &finished,
+		})
+		return
+	}
+
+	maxWorkers := job.Options.MaxParallelHosts
+	if maxWorkers <= 0 {
+		maxWorkers = 64
+	}
+
+	var mu sync.Mutex
+	var done int
+	var up int
+	workCh := make(chan string)
+	wg := sync.WaitGroup{}
+
+	worker := func() {
+		defer wg.Done()
+		for ip := range workCh {
+			isUp := probeHost(ip, job.Options.PingTimeoutMS)
+
+			mu.Lock()
+			done++
+			if isUp {
+				up++
+			}
+			progress := done * 100 / total
+			stats := ScanStats{HostsTotal: total, HostsUp: up}
+			mu.Unlock()
+
+			_, _ = r.store.UpdateScan(job.ID, ScanUpdate{
+				Status:   "running",
+				Progress: progress,
+				Stats:    stats,
+			})
+		}
+	}
+
+	for i := 0; i < maxWorkers; i++ {
+		wg.Add(1)
+		go worker()
+	}
+
+	for _, ip := range ips {
+		workCh <- ip
+	}
+	close(workCh)
+	wg.Wait()
+
+	finished := time.Now().UTC()
+	_, _ = r.store.UpdateScan(job.ID, ScanUpdate{
+		Status:     "done",
+		Progress:   100,
+		Stats:      ScanStats{HostsTotal: total, HostsUp: up},
+		FinishedAt: &finished,
+	})
+}
+
+func expandTargets(cidrs []string, excludeIPs []string) ([]string, error) {
+	excluded := make(map[string]struct{}, len(excludeIPs))
+	for _, ip := range excludeIPs {
+		excluded[ip] = struct{}{}
+	}
+
+	seen := make(map[string]struct{})
+	out := make([]string, 0, 256)
+	for _, c := range cidrs {
+		ip, ipnet, err := net.ParseCIDR(c)
+		if err != nil {
+			return nil, err
+		}
+		start := ip.Mask(ipnet.Mask).To4()
+		if start == nil {
+			continue
+		}
+		for current := dupIP(start); ipnet.Contains(current); incIP(current) {
+			host := current.String()
+			if _, ok := excluded[host]; ok {
+				continue
+			}
+			if _, ok := seen[host]; ok {
+				continue
+			}
+			seen[host] = struct{}{}
+			out = append(out, host)
+		}
+	}
+	return out, nil
+}
+
+func probeHost(ip string, timeoutMS int) bool {
+	if timeoutMS <= 0 {
+		timeoutMS = 700
+	}
+	timeoutSec := strconv.Itoa(max(1, timeoutMS/1000))
+
+	var cmd *exec.Cmd
+	if runtime.GOOS == "windows" {
+		// -n 1 one packet, -w timeout in ms
+		cmd = exec.Command("ping", "-n", "1", "-w", strconv.Itoa(timeoutMS), ip)
+	} else {
+		// -c 1 one packet, -W timeout in sec on Linux/macOS
+		cmd = exec.Command("ping", "-c", "1", "-W", timeoutSec, ip)
+	}
+	if err := cmd.Run(); err != nil {
+		log.Printf("probe failed for %s: %v", ip, err)
+		return false
+	}
+	return true
+}
+
+func dupIP(ip net.IP) net.IP {
+	out := make(net.IP, len(ip))
+	copy(out, ip)
+	return out
+}
+
+func incIP(ip net.IP) {
+	for j := len(ip) - 1; j >= 0; j-- {
+		ip[j]++
+		if ip[j] > 0 {
+			break
+		}
+	}
+}
+
+func max(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"net"
+	"sort"
 	"sync"
 	"time"
 )
@@ -33,14 +34,31 @@ type ScanJob struct {
 	ExcludeIPs        []string    `json:"exclude_ips"`
 	SNMPCredentialsID string      `json:"snmp_credentials_id"`
 	Options           ScanOptions `json:"options"`
+	Progress          int         `json:"progress"`
+	Stats             ScanStats   `json:"stats"`
 	CreatedAt         time.Time   `json:"created_at"`
 	StartedAt         time.Time   `json:"started_at,omitempty"`
 	FinishedAt        time.Time   `json:"finished_at,omitempty"`
 }

+type ScanStats struct {
+	HostsTotal int `json:"hosts_total"`
+	HostsUp    int `json:"hosts_up"`
+}
+
+type ScanUpdate struct {
+	Status     string
+	Progress   int
+	Stats      ScanStats
+	StartedAt  *time.Time
+	FinishedAt *time.Time
+}
+
 type Store interface {
 	CreateScan(CreateScanRequest) (ScanJob, error)
 	GetScan(id string) (ScanJob, bool)
+	ListScans(limit int) []ScanJob
+	UpdateScan(id string, update ScanUpdate) (ScanJob, bool)
 }

 type MemoryStore struct {
@@ -55,20 +73,8 @@ func NewMemoryStore() *MemoryStore {
 }

 func (s *MemoryStore) CreateScan(req CreateScanRequest) (ScanJob, error) {
-	if len(req.CIDRs) == 0 {
-		return ScanJob{}, errors.New("cidrs must not be empty")
-	}
-
-	for _, cidr := range req.CIDRs {
-		if _, _, err := net.ParseCIDR(cidr); err != nil {
-			return ScanJob{}, errors.New("invalid cidr: " + cidr)
-		}
-	}
-
-	for _, ip := range req.ExcludeIPs {
-		if net.ParseIP(ip) == nil {
-			return ScanJob{}, errors.New("invalid exclude ip: " + ip)
-		}
+	if err := validateCreateScanRequest(req); err != nil {
+		return ScanJob{}, err
 	}

 	applyDefaultOptions(&req.Options)
@@ -87,6 +93,8 @@ func (s *MemoryStore) CreateScan(req CreateScanRequest) (ScanJob, error) {
 		ExcludeIPs:        req.ExcludeIPs,
 		SNMPCredentialsID: req.SNMPCredentialsID,
 		Options:           req.Options,
+		Progress:          0,
+		Stats:             ScanStats{},
 		CreatedAt:         now,
 	}

@@ -104,6 +112,55 @@ func (s *MemoryStore) GetScan(id string) (ScanJob, bool) {
 	return job, ok
 }

+func (s *MemoryStore) ListScans(limit int) []ScanJob {
+	if limit <= 0 {
+		limit = 50
+	}
+
+	s.mu.RLock()
+	jobs := make([]ScanJob, 0, len(s.jobs))
+	for _, j := range s.jobs {
+		jobs = append(jobs, j)
+	}
+	s.mu.RUnlock()
+
+	sort.Slice(jobs, func(i, j int) bool {
+		return jobs[i].CreatedAt.After(jobs[j].CreatedAt)
+	})
+	if len(jobs) > limit {
+		jobs = jobs[:limit]
+	}
+
+	return jobs
+}
+
+func (s *MemoryStore) UpdateScan(id string, update ScanUpdate) (ScanJob, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	job, ok := s.jobs[id]
+	if !ok {
+		return ScanJob{}, false
+	}
+
+	if update.Status != "" {
+		job.Status = update.Status
+	}
+	if update.Progress >= 0 {
+		job.Progress = update.Progress
+	}
+	job.Stats = update.Stats
+	if update.StartedAt != nil {
+		job.StartedAt = *update.StartedAt
+	}
+	if update.FinishedAt != nil {
+		job.FinishedAt = *update.FinishedAt
+	}
+
+	s.jobs[id] = job
+	return job, true
+}
+
 func applyDefaultOptions(opts *ScanOptions) {
 	if opts.PingTimeoutMS <= 0 {
 		opts.PingTimeoutMS = 700
@@ -119,6 +176,26 @@ func applyDefaultOptions(opts *ScanOptions) {
 	}
 }

+func validateCreateScanRequest(req CreateScanRequest) error {
+	if len(req.CIDRs) == 0 {
+		return errors.New("cidrs must not be empty")
+	}
+
+	for _, cidr := range req.CIDRs {
+		if _, _, err := net.ParseCIDR(cidr); err != nil {
+			return errors.New("invalid cidr: " + cidr)
+		}
+	}
+
+	for _, ip := range req.ExcludeIPs {
+		if net.ParseIP(ip) == nil {
+			return errors.New("invalid exclude ip: " + ip)
+		}
+	}
+
+	return nil
+}
+
 func newID() (string, error) {
 	var b [16]byte
 	if _, err := rand.Read(b[:]); err != nil {