add gaiia webhook listener with per-org signature verification

Stage 3 of Gaiia integration: webhook event processing for real-time
incremental updates from Gaiia. HMAC-SHA256 signature verification
using per-organization secrets stored in integration credentials.
Handles account, billing subscription, and inventory item events.

lib/towerops/gaiia/webhooks.ex

@@ -0,0 +1,105 @@
+defmodule Towerops.Gaiia.Webhooks do
+  @moduledoc """
+  Processes webhook events from Gaiia.
+
+  Handles incremental updates for accounts, billing subscriptions,
+  and inventory items. Unknown events are logged and ignored for
+  forward-compatibility.
+  """
+
+  alias Towerops.Gaiia
+
+  require Logger
+
+  @doc """
+  Verify an HMAC-SHA256 signature against the raw request body.
+
+  Returns `true` if the signature matches, `false` otherwise.
+  Uses constant-time comparison to prevent timing attacks.
+  """
+  @spec verify_signature(binary(), binary(), binary()) :: boolean()
+  def verify_signature(body, signature, secret) do
+    expected = :hmac |> :crypto.mac(:sha256, secret, body) |> Base.encode16(case: :lower)
+    Plug.Crypto.secure_compare(expected, signature)
+  end
+
+  @doc """
+  Process a webhook event from Gaiia.
+
+  Returns `:ok` for all events, including unknown ones (forward-compatible).
+  """
+  @spec process_event(String.t(), String.t(), map()) :: :ok
+  def process_event(organization_id, event, payload) do
+    case event do
+      "account." <> _ -> handle_account_event(organization_id, event, payload)
+      "billing_subscription." <> _ -> handle_subscription_event(organization_id, event, payload)
+      "inventory_item." <> _ -> handle_inventory_event(organization_id, event, payload)
+      _ -> Logger.info("Ignoring unknown Gaiia webhook event: #{event}")
+    end
+
+    :ok
+  end
+
+  defp handle_account_event(organization_id, event, %{"data" => data}) do
+    gaiia_id = data["id"]
+
+    case event do
+      "account.deleted" ->
+        upsert_account(organization_id, gaiia_id, data, %{status: "deleted"})
+
+      _ ->
+        upsert_account(organization_id, gaiia_id, data, %{})
+    end
+  end
+
+  defp handle_subscription_event(organization_id, _event, %{"data" => data}) do
+    attrs = %{
+      gaiia_id: data["id"],
+      account_gaiia_id: data["account_id"],
+      status: data["status"],
+      product_name: data["product_name"],
+      mrr_amount: parse_decimal(data["mrr_amount"]),
+      currency: data["currency"]
+    }
+
+    Gaiia.upsert_billing_subscription(organization_id, drop_nils(attrs))
+  end
+
+  defp handle_inventory_event(organization_id, _event, %{"data" => data}) do
+    attrs = %{
+      gaiia_id: data["id"],
+      name: data["name"],
+      ip_address: data["ip_address"],
+      assigned_account_gaiia_id: data["assigned_account_id"],
+      assigned_network_site_gaiia_id: data["assigned_network_site_id"]
+    }
+
+    Gaiia.upsert_inventory_item(organization_id, drop_nils(attrs))
+  end
+
+  defp upsert_account(organization_id, gaiia_id, data, overrides) do
+    attrs =
+      %{
+        gaiia_id: gaiia_id,
+        readable_id: data["readable_id"],
+        name: data["name"],
+        status: data["status"],
+        account_type: data["account_type"],
+        address: data["address"]
+      }
+      |> Map.merge(overrides)
+      |> drop_nils()
+
+    Gaiia.upsert_account(organization_id, attrs)
+  end
+
+  defp parse_decimal(nil), do: nil
+  defp parse_decimal(val) when is_binary(val), do: Decimal.new(val)
+  defp parse_decimal(val) when is_number(val), do: Decimal.from_float(val / 1)
+
+  defp drop_nils(map) do
+    map
+    |> Enum.reject(fn {_k, v} -> is_nil(v) end)
+    |> Map.new()
+  end
+end

lib/towerops_web/controllers/api/v1/gaiia_webhook_controller.ex

@@ -0,0 +1,99 @@
+defmodule ToweropsWeb.Api.V1.GaiiaWebhookController do
+  @moduledoc """
+  Webhook endpoint for receiving real-time updates from Gaiia.
+
+  Each organization has its own webhook secret stored in the integration
+  credentials. The signature is verified using HMAC-SHA256 against the
+  raw request body.
+
+  Route: POST /api/v1/webhooks/gaiia/:organization_id
+  """
+  use ToweropsWeb, :controller
+
+  alias Towerops.Gaiia.Webhooks
+  alias Towerops.Integrations
+
+  def create(conn, %{"organization_id" => organization_id}) do
+    with {:ok, integration} <- get_gaiia_integration(organization_id),
+         {:ok, secret} <- get_webhook_secret(integration),
+         {:ok, raw_body} <- get_raw_body(conn),
+         :ok <- verify_signature(conn, raw_body, secret),
+         {:ok, event} <- get_event(conn.body_params) do
+      data = Map.get(conn.body_params, "data", %{})
+      payload = %{"entity_id" => Map.get(data, "id"), "data" => data}
+
+      Webhooks.process_event(organization_id, event, payload)
+
+      json(conn, %{status: "ok"})
+    end
+  end
+
+  defp get_gaiia_integration(organization_id) do
+    case Integrations.get_integration(organization_id, "gaiia") do
+      {:ok, integration} ->
+        {:ok, integration}
+
+      {:error, :not_found} ->
+        {:error, :not_found}
+    end
+  end
+
+  defp get_webhook_secret(integration) do
+    case get_in(integration.credentials, ["webhook_secret"]) do
+      nil -> {:error, :no_secret}
+      secret -> {:ok, secret}
+    end
+  end
+
+  defp get_raw_body(conn) do
+    case conn.private[:raw_body] do
+      nil -> {:error, :no_body}
+      body -> {:ok, body}
+    end
+  end
+
+  defp verify_signature(conn, raw_body, secret) do
+    case get_req_header(conn, "x-gaiia-signature") do
+      [signature] ->
+        if Webhooks.verify_signature(raw_body, signature, secret) do
+          :ok
+        else
+          {:error, :invalid_signature}
+        end
+
+      _ ->
+        {:error, :missing_signature}
+    end
+  end
+
+  defp get_event(%{"event" => event}) when is_binary(event), do: {:ok, event}
+  defp get_event(_), do: {:error, :missing_event}
+
+  # Override action/2 to handle with clause failures
+  def action(conn, _opts) do
+    case apply(__MODULE__, action_name(conn), [conn, conn.params]) do
+      {:error, :not_found} ->
+        conn |> put_status(:not_found) |> json(%{error: "Gaiia integration not found"})
+
+      {:error, :no_secret} ->
+        conn
+        |> put_status(:internal_server_error)
+        |> json(%{error: "Webhook secret not configured"})
+
+      {:error, :missing_signature} ->
+        conn |> put_status(:unauthorized) |> json(%{error: "Missing signature header"})
+
+      {:error, :invalid_signature} ->
+        conn |> put_status(:unauthorized) |> json(%{error: "Invalid webhook signature"})
+
+      {:error, :no_body} ->
+        conn |> put_status(:bad_request) |> json(%{error: "Missing request body"})
+
+      {:error, :missing_event} ->
+        conn |> put_status(:bad_request) |> json(%{error: "Missing event field"})
+
+      other ->
+        other
+    end
+  end
+end

lib/towerops_web/endpoint.ex

@@ -60,6 +60,7 @@ defmodule ToweropsWeb.Endpoint do
     json_decoder: Phoenix.json_library()
 
   # Custom body reader that skips parsing for protobuf content type
+  # and caches raw body for webhook signature verification.
   defmodule BodyReader do
     @moduledoc false
     def read_body(conn, opts) do
@@ -69,8 +70,14 @@ defmodule ToweropsWeb.Endpoint do
           {:ok, "", conn}
 
         _ ->
-          # Use default body reader for other content types
+          case Plug.Conn.read_body(conn, opts) do
-          Plug.Conn.read_body(conn, opts)
+            {:ok, body, conn} ->
+              conn = update_in(conn.private[:raw_body], &((&1 || "") <> body))
+              {:ok, body, conn}
+
+            other ->
+              other
+          end
       end
     end
   end

lib/towerops_web/router.ex

@@ -136,6 +136,13 @@ defmodule ToweropsWeb.Router do
     post "/agent-release", AgentReleaseWebhookController, :create
   end
 
+  # Gaiia webhook routes (per-organization secret verification in controller)
+  scope "/api/v1/webhooks", V1 do
+    pipe_through [:api, :rate_limit_api]
+
+    post "/gaiia/:organization_id", GaiiaWebhookController, :create
+  end
+
   # Admin API routes (requires superuser API token)
   scope "/admin/api", V1 do
     pipe_through :api_v1

test/towerops/gaiia/webhooks_test.exs

@@ -0,0 +1,274 @@
+defmodule Towerops.Gaiia.WebhooksTest do
+  use Towerops.DataCase, async: true
+
+  import Towerops.AccountsFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Gaiia
+  alias Towerops.Gaiia.Webhooks
+
+  setup do
+    user = user_fixture()
+    org = organization_fixture(user.id)
+    %{org: org}
+  end
+
+  describe "process_event/3 - account events" do
+    test "account.created inserts a new account", %{org: org} do
+      payload = %{
+        "entity_id" => "gaiia-acct-1",
+        "data" => %{
+          "id" => "gaiia-acct-1",
+          "readable_id" => "ACCT-001",
+          "name" => "John Smith",
+          "status" => "active",
+          "account_type" => "residential"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "account.created", payload)
+      assert account = Gaiia.get_account(org.id, "gaiia-acct-1")
+      assert account.name == "John Smith"
+      assert account.status == "active"
+    end
+
+    test "account.status_updated updates existing account status", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_account(org.id, %{
+          gaiia_id: "gaiia-acct-1",
+          name: "John Smith",
+          status: "active"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-acct-1",
+        "data" => %{
+          "id" => "gaiia-acct-1",
+          "name" => "John Smith",
+          "status" => "suspended"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "account.status_updated", payload)
+      assert account = Gaiia.get_account(org.id, "gaiia-acct-1")
+      assert account.status == "suspended"
+    end
+
+    test "account.physical_address_updated updates address", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_account(org.id, %{
+          gaiia_id: "gaiia-acct-1",
+          name: "John Smith",
+          status: "active"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-acct-1",
+        "data" => %{
+          "id" => "gaiia-acct-1",
+          "name" => "John Smith",
+          "status" => "active",
+          "address" => %{
+            "street" => "123 Main St",
+            "city" => "Springfield"
+          }
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "account.physical_address_updated", payload)
+      assert account = Gaiia.get_account(org.id, "gaiia-acct-1")
+      assert account.address["street"] == "123 Main St"
+    end
+
+    test "account.deleted removes or marks account inactive", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_account(org.id, %{
+          gaiia_id: "gaiia-acct-1",
+          name: "John Smith",
+          status: "active"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-acct-1",
+        "data" => %{"id" => "gaiia-acct-1"}
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "account.deleted", payload)
+      assert account = Gaiia.get_account(org.id, "gaiia-acct-1")
+      assert account.status == "deleted"
+    end
+  end
+
+  describe "process_event/3 - billing subscription events" do
+    test "billing_subscription.activated upserts subscription", %{org: org} do
+      payload = %{
+        "entity_id" => "gaiia-sub-1",
+        "data" => %{
+          "id" => "gaiia-sub-1",
+          "account_id" => "gaiia-acct-1",
+          "status" => "active",
+          "product_name" => "100 Mbps Plan",
+          "mrr_amount" => "79.99",
+          "currency" => "USD"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "billing_subscription.activated", payload)
+      assert sub = Gaiia.get_billing_subscription(org.id, "gaiia-sub-1")
+      assert sub.status == "active"
+      assert sub.product_name == "100 Mbps Plan"
+      assert Decimal.equal?(sub.mrr_amount, Decimal.new("79.99"))
+    end
+
+    test "billing_subscription.suspended updates status", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_billing_subscription(org.id, %{
+          gaiia_id: "gaiia-sub-1",
+          account_gaiia_id: "gaiia-acct-1",
+          status: "active",
+          product_name: "100 Mbps Plan"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-sub-1",
+        "data" => %{
+          "id" => "gaiia-sub-1",
+          "account_id" => "gaiia-acct-1",
+          "status" => "suspended",
+          "product_name" => "100 Mbps Plan"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "billing_subscription.suspended", payload)
+      assert sub = Gaiia.get_billing_subscription(org.id, "gaiia-sub-1")
+      assert sub.status == "suspended"
+    end
+
+    test "billing_subscription.unassigned updates status", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_billing_subscription(org.id, %{
+          gaiia_id: "gaiia-sub-1",
+          account_gaiia_id: "gaiia-acct-1",
+          status: "active",
+          product_name: "100 Mbps Plan"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-sub-1",
+        "data" => %{
+          "id" => "gaiia-sub-1",
+          "account_id" => "gaiia-acct-1",
+          "status" => "unassigned",
+          "product_name" => "100 Mbps Plan"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "billing_subscription.unassigned", payload)
+      assert sub = Gaiia.get_billing_subscription(org.id, "gaiia-sub-1")
+      assert sub.status == "unassigned"
+    end
+  end
+
+  describe "process_event/3 - inventory item events" do
+    test "inventory_item.assigned updates assignment fields", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_inventory_item(org.id, %{
+          gaiia_id: "gaiia-inv-1",
+          name: "Router Alpha"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-inv-1",
+        "data" => %{
+          "id" => "gaiia-inv-1",
+          "name" => "Router Alpha",
+          "assigned_account_id" => "gaiia-acct-1",
+          "assigned_network_site_id" => "gaiia-site-1"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "inventory_item.assigned", payload)
+      assert item = Gaiia.get_inventory_item(org.id, "gaiia-inv-1")
+      assert item.assigned_account_gaiia_id == "gaiia-acct-1"
+      assert item.assigned_network_site_gaiia_id == "gaiia-site-1"
+    end
+
+    test "inventory_item.ip_address_assigned updates IP", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_inventory_item(org.id, %{
+          gaiia_id: "gaiia-inv-1",
+          name: "Router Alpha"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-inv-1",
+        "data" => %{
+          "id" => "gaiia-inv-1",
+          "name" => "Router Alpha",
+          "ip_address" => "10.0.0.5"
+        }
+      }
+
+      assert :ok = Webhooks.process_event(org.id, "inventory_item.ip_address_assigned", payload)
+      assert item = Gaiia.get_inventory_item(org.id, "gaiia-inv-1")
+      assert item.ip_address == "10.0.0.5"
+    end
+
+    test "inventory_item.ip_address_unassigned clears IP", %{org: org} do
+      {:ok, _} =
+        Gaiia.upsert_inventory_item(org.id, %{
+          gaiia_id: "gaiia-inv-1",
+          name: "Router Alpha",
+          ip_address: "10.0.0.5"
+        })
+
+      payload = %{
+        "entity_id" => "gaiia-inv-1",
+        "data" => %{
+          "id" => "gaiia-inv-1",
+          "name" => "Router Alpha",
+          "ip_address" => nil
+        }
+      }
+
+      assert :ok =
+               Webhooks.process_event(org.id, "inventory_item.ip_address_unassigned", payload)
+
+      assert item = Gaiia.get_inventory_item(org.id, "gaiia-inv-1")
+      assert is_nil(item.ip_address)
+    end
+  end
+
+  describe "process_event/3 - unknown events" do
+    test "unknown events are ignored gracefully", %{org: org} do
+      payload = %{"entity_id" => "whatever", "data" => %{}}
+      assert :ok = Webhooks.process_event(org.id, "some.unknown.event", payload)
+    end
+  end
+
+  describe "verify_signature/2" do
+    test "returns true for valid HMAC-SHA256 signature" do
+      secret = "test-webhook-secret"
+      body = ~s({"event":"account.created"})
+      signature = :hmac |> :crypto.mac(:sha256, secret, body) |> Base.encode16(case: :lower)
+
+      assert Webhooks.verify_signature(body, signature, secret) == true
+    end
+
+    test "returns false for invalid signature" do
+      secret = "test-webhook-secret"
+      body = ~s({"event":"account.created"})
+
+      assert Webhooks.verify_signature(body, "invalid-signature", secret) == false
+    end
+
+    test "returns false for tampered body" do
+      secret = "test-webhook-secret"
+      body = ~s({"event":"account.created"})
+      signature = :hmac |> :crypto.mac(:sha256, secret, body) |> Base.encode16(case: :lower)
+      tampered_body = ~s({"event":"account.deleted"})
+
+      assert Webhooks.verify_signature(tampered_body, signature, secret) == false
+    end
+  end
+end

test/towerops_web/controllers/api/v1/gaiia_webhook_controller_test.exs

@@ -0,0 +1,161 @@
+defmodule ToweropsWeb.Api.V1.GaiiaWebhookControllerTest do
+  use ToweropsWeb.ConnCase, async: true
+
+  import Towerops.AccountsFixtures
+  import Towerops.OrganizationsFixtures
+
+  alias Towerops.Gaiia
+  alias Towerops.Integrations
+
+  @webhook_secret "test-gaiia-webhook-secret-abc123"
+
+  setup do
+    user = user_fixture()
+    org = organization_fixture(user.id)
+
+    {:ok, integration} =
+      Integrations.create_integration(org.id, %{
+        provider: "gaiia",
+        enabled: true,
+        credentials: %{
+          "api_key" => "test-api-key",
+          "webhook_secret" => @webhook_secret
+        }
+      })
+
+    %{org: org, integration: integration}
+  end
+
+  defp sign_payload(body, secret) do
+    :hmac |> :crypto.mac(:sha256, secret, body) |> Base.encode16(case: :lower)
+  end
+
+  defp post_webhook(conn, org_id, event, payload, secret \\ @webhook_secret) do
+    body = Jason.encode!(%{"event" => event, "data" => payload})
+    signature = sign_payload(body, secret)
+
+    conn
+    |> put_req_header("content-type", "application/json")
+    |> put_req_header("x-gaiia-signature", signature)
+    |> post(~p"/api/v1/webhooks/gaiia/#{org_id}", body)
+  end
+
+  describe "authentication" do
+    test "returns 401 without signature header", %{conn: conn, org: org} do
+      conn =
+        conn
+        |> put_req_header("content-type", "application/json")
+        |> post(~p"/api/v1/webhooks/gaiia/#{org.id}", Jason.encode!(%{"event" => "test"}))
+
+      assert json_response(conn, 401)["error"] =~ "Missing"
+    end
+
+    test "returns 401 with invalid signature", %{conn: conn, org: org} do
+      body = Jason.encode!(%{"event" => "account.created", "data" => %{}})
+
+      conn =
+        conn
+        |> put_req_header("content-type", "application/json")
+        |> put_req_header("x-gaiia-signature", "invalid-signature")
+        |> post(~p"/api/v1/webhooks/gaiia/#{org.id}", body)
+
+      assert json_response(conn, 401)["error"] =~ "Invalid"
+    end
+
+    test "returns 404 when organization has no gaiia integration", %{conn: conn} do
+      other_user = user_fixture()
+      other_org = organization_fixture(other_user.id)
+
+      body = Jason.encode!(%{"event" => "account.created", "data" => %{}})
+      signature = sign_payload(body, "some-secret")
+
+      conn =
+        conn
+        |> put_req_header("content-type", "application/json")
+        |> put_req_header("x-gaiia-signature", signature)
+        |> post(~p"/api/v1/webhooks/gaiia/#{other_org.id}", body)
+
+      assert json_response(conn, 404)["error"] =~ "not found"
+    end
+
+    test "returns 401 with wrong secret for organization", %{conn: conn, org: org} do
+      conn = post_webhook(conn, org.id, "account.created", %{}, "wrong-secret")
+      assert json_response(conn, 401)["error"] =~ "Invalid"
+    end
+  end
+
+  describe "event processing" do
+    test "processes account.created event", %{conn: conn, org: org} do
+      payload = %{
+        "id" => "gaiia-acct-1",
+        "readable_id" => "ACCT-001",
+        "name" => "Jane Doe",
+        "status" => "active",
+        "account_type" => "residential"
+      }
+
+      conn = post_webhook(conn, org.id, "account.created", payload)
+      assert json_response(conn, 200)["status"] == "ok"
+
+      assert account = Gaiia.get_account(org.id, "gaiia-acct-1")
+      assert account.name == "Jane Doe"
+    end
+
+    test "processes billing_subscription.activated event", %{conn: conn, org: org} do
+      payload = %{
+        "id" => "gaiia-sub-1",
+        "account_id" => "gaiia-acct-1",
+        "status" => "active",
+        "product_name" => "50 Mbps Plan",
+        "mrr_amount" => "49.99",
+        "currency" => "USD"
+      }
+
+      conn = post_webhook(conn, org.id, "billing_subscription.activated", payload)
+      assert json_response(conn, 200)["status"] == "ok"
+
+      assert sub = Gaiia.get_billing_subscription(org.id, "gaiia-sub-1")
+      assert sub.product_name == "50 Mbps Plan"
+    end
+
+    test "processes inventory_item.ip_address_assigned event", %{conn: conn, org: org} do
+      {:ok, _} =
+        Gaiia.upsert_inventory_item(org.id, %{
+          gaiia_id: "gaiia-inv-1",
+          name: "AP One"
+        })
+
+      payload = %{
+        "id" => "gaiia-inv-1",
+        "name" => "AP One",
+        "ip_address" => "10.0.0.50"
+      }
+
+      conn = post_webhook(conn, org.id, "inventory_item.ip_address_assigned", payload)
+      assert json_response(conn, 200)["status"] == "ok"
+
+      assert item = Gaiia.get_inventory_item(org.id, "gaiia-inv-1")
+      assert item.ip_address == "10.0.0.50"
+    end
+
+    test "returns ok for unknown events", %{conn: conn, org: org} do
+      conn = post_webhook(conn, org.id, "some.future.event", %{})
+      assert json_response(conn, 200)["status"] == "ok"
+    end
+  end
+
+  describe "error handling" do
+    test "returns 400 when body is missing event field", %{conn: conn, org: org} do
+      body = Jason.encode!(%{"data" => %{}})
+      signature = sign_payload(body, @webhook_secret)
+
+      conn =
+        conn
+        |> put_req_header("content-type", "application/json")
+        |> put_req_header("x-gaiia-signature", signature)
+        |> post(~p"/api/v1/webhooks/gaiia/#{org.id}", body)
+
+      assert json_response(conn, 400)["error"] =~ "Missing"
+    end
+  end
+end