caddy-opnsense-blocker/internal/service/service_test.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"git.dern.ovh/infrastructure/caddy-opnsense-blocker/internal/config"
	"git.dern.ovh/infrastructure/caddy-opnsense-blocker/internal/model"
	"git.dern.ovh/infrastructure/caddy-opnsense-blocker/internal/opnsense"
	"git.dern.ovh/infrastructure/caddy-opnsense-blocker/internal/store"
)

func TestServiceProcessesMultipleSourcesAndManualActions(t *testing.T) {
	t.Parallel()

	tempDir := t.TempDir()
	mainLogPath := filepath.Join(tempDir, "main.log")
	giteaLogPath := filepath.Join(tempDir, "gitea.log")
	if err := os.WriteFile(mainLogPath, nil, 0o600); err != nil {
		t.Fatalf("create main log: %v", err)
	}
	if err := os.WriteFile(giteaLogPath, nil, 0o600); err != nil {
		t.Fatalf("create gitea log: %v", err)
	}

	backend := newFakeOPNsenseServer(t)
	defer backend.Close()

	configPath := filepath.Join(tempDir, "config.yaml")
	payload := fmt.Sprintf(`storage:
  path: %s/blocker.db
opnsense:
  enabled: true
  base_url: %s
  api_key: key
  api_secret: secret
  ensure_alias: true
  alias:
    name: blocked-ips
profiles:
  main:
    auto_block: true
    block_unexpected_posts: true
    block_php_paths: true
    suspicious_path_prefixes:
      - /wp-login.php
  gitea:
    auto_block: false
    block_unexpected_posts: true
    allowed_post_paths:
      - /user/login
    suspicious_path_prefixes:
      - /install.php
sources:
  - name: main
    path: %s
    profile: main
    initial_position: beginning
    poll_interval: 20ms
    batch_size: 128
  - name: gitea
    path: %s
    profile: gitea
    initial_position: beginning
    poll_interval: 20ms
    batch_size: 128
`, tempDir, backend.URL, mainLogPath, giteaLogPath)
	if err := os.WriteFile(configPath, []byte(payload), 0o600); err != nil {
		t.Fatalf("write config: %v", err)
	}

	cfg, err := config.Load(configPath)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	database, err := store.Open(cfg.Storage.Path)
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer database.Close()
	svc := New(cfg, database, opnsense.NewClient(cfg.OPNsense), nil, log.New(os.Stderr, "", 0))

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = svc.Run(ctx) }()

	appendLine(t, mainLogPath, caddyJSONLine("203.0.113.10", "198.51.100.10", "example.test", "GET", "/wp-login.php", 404, "curl/8.0", time.Now().UTC()))
	appendLine(t, giteaLogPath, caddyJSONLine("203.0.113.11", "198.51.100.11", "git.example.test", "POST", "/user/login", 401, "curl/8.0", time.Now().UTC()))
	appendLine(t, giteaLogPath, caddyJSONLine("203.0.113.12", "198.51.100.12", "git.example.test", "GET", "/install.php", 404, "curl/8.0", time.Now().UTC()))

	waitFor(t, 3*time.Second, func() bool {
		overview, err := database.GetOverview(context.Background(), 10)
		return err == nil && overview.TotalEvents == 3
	})

	blockedState, found, err := database.GetIPState(context.Background(), "203.0.113.10")
	if err != nil || !found {
		t.Fatalf("load blocked state: found=%v err=%v", found, err)
	}
	if blockedState.State != model.IPStateBlocked {
		t.Fatalf("expected blocked state, got %+v", blockedState)
	}

	reviewState, found, err := database.GetIPState(context.Background(), "203.0.113.12")
	if err != nil || !found {
		t.Fatalf("load review state: found=%v err=%v", found, err)
	}
	if reviewState.State != model.IPStateReview {
		t.Fatalf("expected review state, got %+v", reviewState)
	}

	observedState, found, err := database.GetIPState(context.Background(), "203.0.113.11")
	if err != nil || !found {
		t.Fatalf("load observed state: found=%v err=%v", found, err)
	}
	if observedState.State != model.IPStateObserved {
		t.Fatalf("expected observed state, got %+v", observedState)
	}

	recentRows, err := svc.ListRecentIPs(context.Background(), time.Now().UTC().Add(-time.Hour), 10)
	if err != nil {
		t.Fatalf("list recent ips: %v", err)
	}
	blockedRow, found := findRecentIPRow(recentRows, "203.0.113.10")
	if !found {
		t.Fatalf("expected blocked IP row in recent rows: %+v", recentRows)
	}
	if blockedRow.SourceName != "main" || blockedRow.Events != 1 {
		t.Fatalf("unexpected blocked recent row: %+v", blockedRow)
	}
	if !blockedRow.Actions.CanUnblock || blockedRow.Actions.CanBlock {
		t.Fatalf("unexpected blocked recent row actions: %+v", blockedRow.Actions)
	}

	if err := svc.ForceAllow(context.Background(), "203.0.113.10", "test", "manual unblock"); err != nil {
		t.Fatalf("force allow: %v", err)
	}
	state, found, err := database.GetIPState(context.Background(), "203.0.113.10")
	if err != nil || !found {
		t.Fatalf("reload unblocked state: found=%v err=%v", found, err)
	}
	if state.ManualOverride != model.ManualOverrideForceAllow || state.State != model.IPStateAllowed {
		t.Fatalf("unexpected manual allow state: %+v", state)
	}

	backend.mu.Lock()
	defer backend.mu.Unlock()
	if _, ok := backend.ips["203.0.113.10"]; ok {
		t.Fatalf("expected IP to be removed from backend alias after manual unblock")
	}
}

func TestServiceBackgroundInvestigationEnrichesRecentIPs(t *testing.T) {
	t.Parallel()

	tempDir := t.TempDir()
	logPath := filepath.Join(tempDir, "access.log")
	if err := os.WriteFile(logPath, nil, 0o600); err != nil {
		t.Fatalf("create log: %v", err)
	}

	configPath := filepath.Join(tempDir, "config.yaml")
	payload := fmt.Sprintf(`storage:
  path: %s/blocker.db
investigation:
  enabled: true
  refresh_after: 24h
  timeout: 500ms
  background_workers: 1
  background_poll_interval: 50ms
  background_lookback: 24h
  background_batch_size: 32
profiles:
  main:
    auto_block: false
    block_unexpected_posts: true
    suspicious_path_prefixes:
      - /wp-login.php
sources:
  - name: main
    path: %s
    profile: main
    initial_position: beginning
    poll_interval: 20ms
    batch_size: 128
`, tempDir, logPath)
	if err := os.WriteFile(configPath, []byte(payload), 0o600); err != nil {
		t.Fatalf("write config: %v", err)
	}

	cfg, err := config.Load(configPath)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	database, err := store.Open(cfg.Storage.Path)
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer database.Close()
	investigator := &fakeInvestigator{}
	svc := New(cfg, database, nil, investigator, log.New(os.Stderr, "", 0))

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = svc.Run(ctx) }()

	appendLine(t, logPath, caddyJSONLine("203.0.113.33", "198.51.100.33", "example.test", "GET", "/wp-login.php", 404, "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", time.Now().UTC()))

	waitFor(t, 3*time.Second, func() bool {
		recentRows, err := svc.ListRecentIPs(context.Background(), time.Now().UTC().Add(-time.Hour), 10)
		if err != nil {
			return false
		}
		row, found := findRecentIPRow(recentRows, "203.0.113.33")
		return found && row.Bot != nil && row.Bot.Name == "Googlebot"
	})

	recentRows, err := svc.ListRecentIPs(context.Background(), time.Now().UTC().Add(-time.Hour), 10)
	if err != nil {
		t.Fatalf("list recent ips: %v", err)
	}
	row, found := findRecentIPRow(recentRows, "203.0.113.33")
	if !found {
		t.Fatalf("expected recent row for investigated ip: %+v", recentRows)
	}
	if row.Bot == nil || row.Bot.Name != "Googlebot" {
		t.Fatalf("expected background investigation bot on recent row, got %+v", row)
	}
	if investigator.callCount() == 0 {
		t.Fatalf("expected background investigator to be called")
	}
}

func TestServiceDoesNotRefreshCachedInvestigationAutomatically(t *testing.T) {
	t.Parallel()

	tempDir := t.TempDir()
	configPath := filepath.Join(tempDir, "config.yaml")
	payload := fmt.Sprintf(`storage:
  path: %s/blocker.db
investigation:
  enabled: true
  refresh_after: 1ns
  timeout: 500ms
  background_workers: 1
  background_poll_interval: 50ms
profiles:
  main:
    auto_block: false
sources:
  - name: main
    path: %s/access.log
    profile: main
`, tempDir, tempDir)
	if err := os.WriteFile(configPath, []byte(payload), 0o600); err != nil {
		t.Fatalf("write config: %v", err)
	}

	cfg, err := config.Load(configPath)
	if err != nil {
		t.Fatalf("load config: %v", err)
	}
	database, err := store.Open(cfg.Storage.Path)
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer database.Close()
	investigator := &fakeInvestigator{}
	svc := New(cfg, database, nil, investigator, log.New(os.Stderr, "", 0))

	ctx := context.Background()
	event := &model.Event{
		SourceName:      "main",
		ProfileName:     "main",
		OccurredAt:      time.Now().UTC(),
		RemoteIP:        "198.51.100.10",
		ClientIP:        "203.0.113.44",
		Host:            "example.test",
		Method:          "GET",
		URI:             "/cached",
		Path:            "/cached",
		Status:          404,
		UserAgent:       "test-agent/1.0",
		Decision:        model.DecisionActionReview,
		DecisionReason:  "review",
		DecisionReasons: []string{"review"},
		RawJSON:         `{"status":404}`,
	}
	if err := database.RecordEvent(ctx, event); err != nil {
		t.Fatalf("record event: %v", err)
	}
	if err := database.SaveInvestigation(ctx, model.IPInvestigation{
		IP:        "203.0.113.44",
		UpdatedAt: time.Now().UTC().Add(-48 * time.Hour),
		Bot: &model.BotMatch{
			Name:     "CachedBot",
			Verified: true,
		},
	}); err != nil {
		t.Fatalf("save investigation: %v", err)
	}

	loaded, err := svc.refreshInvestigation(ctx, "203.0.113.44", false)
	if err != nil {
		t.Fatalf("refresh investigation: %v", err)
	}
	if loaded == nil || loaded.Bot == nil || loaded.Bot.Name != "CachedBot" {
		t.Fatalf("expected cached investigation, got %+v", loaded)
	}
	if investigator.callCount() != 0 {
		t.Fatalf("expected cached investigation to skip external refresh, got %d calls", investigator.callCount())
	}
}

type fakeOPNsenseServer struct {
	*httptest.Server
	mu          sync.Mutex
	aliasUUID   string
	aliasExists bool
	ips         map[string]struct{}
}

func newFakeOPNsenseServer(t *testing.T) *fakeOPNsenseServer {
	t.Helper()
	backend := &fakeOPNsenseServer{ips: map[string]struct{}{}}
	backend.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		username, password, ok := r.BasicAuth()
		if !ok || username != "key" || password != "secret" {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		backend.mu.Lock()
		defer backend.mu.Unlock()

		switch {
		case r.Method == http.MethodGet && r.URL.Path == "/api/firewall/alias/get_alias_u_u_i_d/blocked-ips":
			if backend.aliasExists {
				_ = json.NewEncoder(w).Encode(map[string]any{"uuid": backend.aliasUUID})
			} else {
				_ = json.NewEncoder(w).Encode(map[string]any{"uuid": ""})
			}
		case r.Method == http.MethodPost && r.URL.Path == "/api/firewall/alias/add_item":
			backend.aliasExists = true
			backend.aliasUUID = "uuid-1"
			_ = json.NewEncoder(w).Encode(map[string]any{"status": "ok"})
		case r.Method == http.MethodPost && r.URL.Path == "/api/firewall/alias/set_item/uuid-1":
			_ = json.NewEncoder(w).Encode(map[string]any{"status": "ok"})
		case r.Method == http.MethodPost && r.URL.Path == "/api/firewall/alias/reconfigure":
			_ = json.NewEncoder(w).Encode(map[string]any{"status": "ok"})
		case r.Method == http.MethodGet && r.URL.Path == "/api/firewall/alias_util/list/blocked-ips":
			rows := make([]map[string]string, 0, len(backend.ips))
			for ip := range backend.ips {
				rows = append(rows, map[string]string{"ip": ip})
			}
			_ = json.NewEncoder(w).Encode(map[string]any{"rows": rows})
		case r.Method == http.MethodPost && r.URL.Path == "/api/firewall/alias_util/add/blocked-ips":
			var payload map[string]string
			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			backend.ips[payload["address"]] = struct{}{}
			_ = json.NewEncoder(w).Encode(map[string]any{"status": "done"})
		case r.Method == http.MethodPost && r.URL.Path == "/api/firewall/alias_util/delete/blocked-ips":
			var payload map[string]string
			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			delete(backend.ips, payload["address"])
			_ = json.NewEncoder(w).Encode(map[string]any{"status": "done"})
		default:
			http.Error(w, "not found", http.StatusNotFound)
		}
	}))
	return backend
}

func appendLine(t *testing.T, path string, line string) {
	t.Helper()
	file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		t.Fatalf("open log file for append: %v", err)
	}
	defer file.Close()
	if _, err := file.WriteString(line + "\n"); err != nil {
		t.Fatalf("append log line: %v", err)
	}
}

func caddyJSONLine(clientIP string, remoteIP string, host string, method string, uri string, status int, userAgent string, occurredAt time.Time) string {
	return fmt.Sprintf(`{"ts":%q,"status":%d,"request":{"remote_ip":%q,"client_ip":%q,"host":%q,"method":%q,"uri":%q,"headers":{"User-Agent":[%q]}}}`,
		occurredAt.UTC().Format(time.RFC3339Nano),
		status,
		remoteIP,
		clientIP,
		host,
		method,
		uri,
		userAgent,
	)
}

func waitFor(t *testing.T, timeout time.Duration, condition func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if condition() {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("condition was not met within %s", timeout)
}

func findRecentIPRow(items []model.RecentIPRow, ip string) (model.RecentIPRow, bool) {
	for _, item := range items {
		if item.IP == ip {
			return item, true
		}
	}
	return model.RecentIPRow{}, false
}

type fakeInvestigator struct {
	mu    sync.Mutex
	count int
}

func (f *fakeInvestigator) Investigate(_ context.Context, ip string, _ []string) (model.IPInvestigation, error) {
	f.mu.Lock()
	f.count++
	f.mu.Unlock()
	return model.IPInvestigation{
		IP:        ip,
		UpdatedAt: time.Now().UTC(),
		Bot: &model.BotMatch{
			ProviderID: "google_official",
			Name:       "Googlebot",
			Method:     "test",
			Verified:   true,
		},
	}, nil
}

func (f *fakeInvestigator) callCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.count
}