feat: add remaining v2 stretch goals — Consul discovery, circuit breakers, Vault, WAF analytics

Implement all five remaining stretch goals:

1. Consul Service Discovery — behaviour + HTTP resolver, discovery source
   validation, secret-aware token handling
2. Circuit Breaker Health View — per-node state tracking via heartbeats,
   summary queries, dashboard stats, upstream group detail UI
3. Vault Integration — encrypted config, token/approle/k8s auth methods,
   transparent secret backend in resolve_references, API controller
4. WAF Event Logging Dashboard — bulk ingestion, filterable queries,
   top blocked IPs/paths, LiveView with auto-refresh, prune worker
5. WAF Anomaly Detection — statistical baselines (hourly mean/stddev),
   z-score spike/new-vector/IP-burst/rate-change detection, Oban workers,
   acknowledge/resolve/false-positive workflow, anomalies LiveView

30 new files, 14 modified files, 4 migrations, 63 new tests (1226 total).

Author: raffaelschneider

config/test.exs

@@ -57,6 +57,12 @@ config :sentinel_cp, :dns_resolver, SentinelCp.Services.DnsResolver.Mock
 # Use mock K8s resolver in tests
 config :sentinel_cp, :k8s_resolver, SentinelCp.Services.K8sResolver.Mock
 
+# Use mock Consul resolver in tests
+config :sentinel_cp, :consul_resolver, SentinelCp.Services.ConsulResolver.Mock
+
+# Use mock Vault client in tests
+config :sentinel_cp, :vault_client, SentinelCp.Secrets.VaultClient.Mock
+
 # Use mock ACME client in tests
 config :sentinel_cp, :acme_client, SentinelCp.Services.Acme.Client.Mock
 

lib/sentinel_cp/analytics.ex

@@ -5,7 +5,7 @@ defmodule SentinelCp.Analytics do
 
   import Ecto.Query, warn: false
   alias SentinelCp.Repo
-  alias SentinelCp.Analytics.{ServiceMetric, RequestLog}
+  alias SentinelCp.Analytics.{ServiceMetric, RequestLog, WafEvent, WafBaseline, WafAnomaly}
 
   ## Ingestion
 
@@ -198,6 +198,307 @@ defmodule SentinelCp.Analytics do
     {:ok, count}
   end
 
+  ## WAF Events
+
+  @doc """
+  Bulk inserts WAF event records from a node push.
+  """
+  def ingest_waf_events(events_list) when is_list(events_list) do
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    entries =
+      Enum.map(events_list, fn attrs ->
+        %{
+          id: Ecto.UUID.generate(),
+          project_id: attrs["project_id"],
+          service_id: attrs["service_id"],
+          node_id: attrs["node_id"],
+          timestamp: parse_datetime_usec(attrs["timestamp"]),
+          rule_type: attrs["rule_type"],
+          rule_id: attrs["rule_id"],
+          action: attrs["action"],
+          severity: attrs["severity"],
+          client_ip: attrs["client_ip"],
+          method: attrs["method"],
+          path: attrs["path"],
+          matched_data: attrs["matched_data"],
+          user_agent: attrs["user_agent"],
+          geo_country: attrs["geo_country"],
+          request_headers: attrs["request_headers"] || %{},
+          metadata: attrs["metadata"] || %{},
+          inserted_at: now
+        }
+      end)
+
+    {count, _} = Repo.insert_all(WafEvent, entries)
+
+    # Broadcast & emit events
+    project_ids = entries |> Enum.map(& &1.project_id) |> Enum.uniq()
+
+    for pid <- project_ids do
+      Phoenix.PubSub.broadcast(SentinelCp.PubSub, "waf:#{pid}", {:waf_event, pid})
+    end
+
+    blocked_count = Enum.count(entries, &(&1.action == "blocked"))
+
+    if blocked_count > 0 do
+      for pid <- project_ids do
+        SentinelCp.Events.emit("security.waf_blocked", %{count: blocked_count},
+          project_id: pid
+        )
+      end
+    end
+
+    {:ok, count}
+  end
+
+  @doc """
+  Lists WAF events for a project with optional filters and pagination.
+  """
+  def list_waf_events(project_id, opts \\ []) do
+    limit = Keyword.get(opts, :limit, 50)
+    offset = Keyword.get(opts, :offset, 0)
+    time_range = Keyword.get(opts, :time_range, 24)
+
+    {start_time, end_time} = resolve_time_range(time_range)
+
+    query =
+      from(e in WafEvent,
+        where: e.project_id == ^project_id,
+        where: e.timestamp >= ^start_time,
+        where: e.timestamp <= ^end_time,
+        order_by: [desc: e.timestamp],
+        limit: ^limit,
+        offset: ^offset
+      )
+
+    query =
+      Enum.reduce(opts, query, fn
+        {:rule_type, rt}, q when is_binary(rt) -> where(q, [e], e.rule_type == ^rt)
+        {:action, a}, q when is_binary(a) -> where(q, [e], e.action == ^a)
+        {:severity, s}, q when is_binary(s) -> where(q, [e], e.severity == ^s)
+        {:client_ip, ip}, q when is_binary(ip) -> where(q, [e], e.client_ip == ^ip)
+        _, q -> q
+      end)
+
+    Repo.all(query)
+  end
+
+  @doc """
+  Returns WAF event statistics for a project within a time range.
+  """
+  def get_waf_event_stats(project_id, time_range_hours \\ 24) do
+    {start_time, end_time} = resolve_time_range(time_range_hours)
+
+    stats =
+      from(e in WafEvent,
+        where: e.project_id == ^project_id,
+        where: e.timestamp >= ^start_time,
+        where: e.timestamp <= ^end_time,
+        select: %{
+          total: count(e.id),
+          blocked: count(fragment("CASE WHEN ? = 'blocked' THEN 1 END", e.action)),
+          logged: count(fragment("CASE WHEN ? = 'logged' THEN 1 END", e.action)),
+          challenged: count(fragment("CASE WHEN ? = 'challenged' THEN 1 END", e.action)),
+          unique_ips: count(e.client_ip, :distinct)
+        }
+      )
+      |> Repo.one()
+
+    stats || %{total: 0, blocked: 0, logged: 0, challenged: 0, unique_ips: 0}
+  end
+
+  @doc """
+  Returns the top blocked client IPs for a project.
+  """
+  def get_top_blocked_ips(project_id, time_range_hours \\ 24, limit \\ 10) do
+    {start_time, end_time} = resolve_time_range(time_range_hours)
+
+    from(e in WafEvent,
+      where: e.project_id == ^project_id,
+      where: e.action == "blocked",
+      where: e.timestamp >= ^start_time,
+      where: e.timestamp <= ^end_time,
+      where: not is_nil(e.client_ip),
+      group_by: e.client_ip,
+      select: {e.client_ip, count(e.id)},
+      order_by: [desc: count(e.id)],
+      limit: ^limit
+    )
+    |> Repo.all()
+  end
+
+  @doc """
+  Returns the top blocked paths for a project.
+  """
+  def get_top_blocked_paths(project_id, time_range_hours \\ 24, limit \\ 10) do
+    {start_time, end_time} = resolve_time_range(time_range_hours)
+
+    from(e in WafEvent,
+      where: e.project_id == ^project_id,
+      where: e.action == "blocked",
+      where: e.timestamp >= ^start_time,
+      where: e.timestamp <= ^end_time,
+      where: not is_nil(e.path),
+      group_by: e.path,
+      select: {e.path, count(e.id)},
+      order_by: [desc: count(e.id)],
+      limit: ^limit
+    )
+    |> Repo.all()
+  end
+
+  @doc """
+  Returns time-series WAF event counts grouped by rule_type.
+  """
+  def get_waf_time_series(project_id, time_range_hours \\ 24, bucket_minutes \\ 60) do
+    {start_time, end_time} = resolve_time_range(time_range_hours)
+    bucket_seconds = bucket_minutes * 60
+
+    from(e in WafEvent,
+      where: e.project_id == ^project_id,
+      where: e.timestamp >= ^start_time,
+      where: e.timestamp <= ^end_time,
+      group_by: [
+        fragment("(strftime('%s', ?) / ? * ?)", e.timestamp, ^bucket_seconds, ^bucket_seconds),
+        e.rule_type
+      ],
+      select: %{
+        bucket: fragment("datetime((strftime('%s', ?) / ? * ?), 'unixepoch')", e.timestamp, ^bucket_seconds, ^bucket_seconds),
+        rule_type: e.rule_type,
+        count: count(e.id)
+      },
+      order_by: [asc: fragment("(strftime('%s', ?) / ? * ?)", e.timestamp, ^bucket_seconds, ^bucket_seconds)]
+    )
+    |> Repo.all()
+  end
+
+  @doc """
+  Deletes WAF events older than the retention period.
+  Returns the number of deleted records.
+  """
+  def prune_old_waf_events(retention_days \\ 30) do
+    cutoff = DateTime.utc_now() |> DateTime.add(-retention_days * 86_400, :second)
+
+    {count, _} =
+      from(e in WafEvent, where: e.timestamp < ^cutoff)
+      |> Repo.delete_all()
+
+    {:ok, count}
+  end
+
+  ## WAF Baselines
+
+  @doc """
+  Gets WAF baselines for a project.
+  """
+  def get_waf_baselines(project_id) do
+    from(b in WafBaseline,
+      where: b.project_id == ^project_id,
+      order_by: [asc: b.metric_type]
+    )
+    |> Repo.all()
+  end
+
+  @doc """
+  Upserts a WAF baseline.
+  """
+  def upsert_waf_baseline(attrs) do
+    %WafBaseline{}
+    |> WafBaseline.changeset(attrs)
+    |> Repo.insert(
+      on_conflict: {:replace, [:mean, :stddev, :sample_count, :last_computed_at, :updated_at]},
+      conflict_target: [:project_id, :service_id, :metric_type, :period]
+    )
+  end
+
+  ## WAF Anomalies
+
+  @doc """
+  Lists WAF anomalies for a project with optional status filter.
+  """
+  def list_waf_anomalies(project_id, opts \\ []) do
+    query =
+      from(a in WafAnomaly,
+        where: a.project_id == ^project_id,
+        order_by: [desc: a.detected_at],
+        limit: ^Keyword.get(opts, :limit, 100)
+      )
+
+    query =
+      case Keyword.get(opts, :status) do
+        nil -> query
+        status -> where(query, [a], a.status == ^status)
+      end
+
+    Repo.all(query)
+  end
+
+  @doc """
+  Gets a single WAF anomaly by ID.
+  """
+  def get_waf_anomaly(id), do: Repo.get(WafAnomaly, id)
+
+  @doc """
+  Creates a WAF anomaly.
+  """
+  def create_waf_anomaly(attrs) do
+    %WafAnomaly{}
+    |> WafAnomaly.create_changeset(attrs)
+    |> Repo.insert()
+  end
+
+  @doc """
+  Acknowledges a WAF anomaly.
+  """
+  def acknowledge_anomaly(id, user_id) do
+    case get_waf_anomaly(id) do
+      nil -> {:error, :not_found}
+      anomaly -> anomaly |> WafAnomaly.acknowledge_changeset(user_id) |> Repo.update()
+    end
+  end
+
+  @doc """
+  Resolves a WAF anomaly.
+  """
+  def resolve_anomaly(id) do
+    case get_waf_anomaly(id) do
+      nil -> {:error, :not_found}
+      anomaly -> anomaly |> WafAnomaly.resolve_changeset() |> Repo.update()
+    end
+  end
+
+  @doc """
+  Marks a WAF anomaly as a false positive.
+  """
+  def mark_false_positive(id, user_id) do
+    case get_waf_anomaly(id) do
+      nil -> {:error, :not_found}
+      anomaly -> anomaly |> WafAnomaly.false_positive_changeset(user_id) |> Repo.update()
+    end
+  end
+
+  @doc """
+  Returns anomaly statistics for a project.
+  """
+  def get_anomaly_stats(project_id) do
+    stats =
+      from(a in WafAnomaly,
+        where: a.project_id == ^project_id,
+        group_by: a.status,
+        select: {a.status, count(a.id)}
+      )
+      |> Repo.all()
+      |> Map.new()
+
+    %{
+      active: Map.get(stats, "active", 0),
+      acknowledged: Map.get(stats, "acknowledged", 0),
+      resolved: Map.get(stats, "resolved", 0),
+      false_positive: Map.get(stats, "false_positive", 0)
+    }
+  end
+
   ## Private
 
   defp resolve_time_range(hours) when is_integer(hours) do