fix: Apply RLS rules to Channel controller

docker-compose.dbs.yml

@@ -8,7 +8,16 @@ services:
       - "5432:5432"
     volumes:
       - ./dev/postgres:/docker-entrypoint-initdb.d/
-    command: postgres -c config_file=/etc/postgresql/postgresql.conf 
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
+    environment:
+      POSTGRES_HOST: /var/run/postgresql
+      POSTGRES_PASSWORD: postgres
+  tenant_db:
+    image: supabase/postgres:14.1.0.105
+    container_name: tenant-db
+    ports:
+      - "5433:5432"
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql
       POSTGRES_PASSWORD: postgres

docker-compose.yml

@@ -12,7 +12,15 @@ services:
     environment:
       POSTGRES_HOST: /var/run/postgresql
       POSTGRES_PASSWORD: postgres
-
+  tenant_db:
+    image: supabase/postgres:14.1.0.105
+    container_name: tenant-db
+    ports:
+      - "5433:5432"
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
+    environment:
+      POSTGRES_HOST: /var/run/postgresql
+      POSTGRES_PASSWORD: postgres
   realtime:
     depends_on:
       - db

lib/realtime/helpers.ex

@@ -126,7 +126,7 @@ defmodule Realtime.Helpers do
   Checks if the Tenant CDC extension information is properly configured and that we're able to query against the tenant database.
   """
   @spec check_tenant_connection(Tenant.t(), binary()) :: {:error, atom()} | {:ok, pid()}
-  def check_tenant_connection(nil, _), do: {:error, :tenant_not_found}
+  def check_tenant_connection(nil, _, _), do: {:error, :tenant_not_found}
 
   def check_tenant_connection(tenant, application_name) do
     tenant

lib/realtime/tenants/authorization.ex

@@ -0,0 +1,109 @@
+defmodule Realtime.Tenants.Authorization do
+  @moduledoc """
+  Runs validations based on RLS policies to set permissions for a given connection.
+
+  It will assign the a new key to a socket or a conn with the following:
+  * read - a boolean indicating whether the connection has read permissions
+  """
+  require Logger
+  defstruct [:channel_name, :headers, :jwt, :claims, :role]
+
+  defmodule Permissions do
+    defstruct read: false
+
+    @type t :: %__MODULE__{
+            :read => boolean()
+          }
+  end
+
+  @type t :: %__MODULE__{
+          :channel_name => binary() | nil,
+          :claims => map(),
+          :headers => keyword({binary(), binary()}),
+          :jwt => map(),
+          :role => binary()
+        }
+  @doc """
+  Builds a new authorization params struct.
+  """
+  def build_authorization_params(%{
+        channel_name: channel_name,
+        headers: headers,
+        jwt: jwt,
+        claims: claims,
+        role: role
+      }) do
+    %__MODULE__{
+      channel_name: channel_name,
+      headers: headers,
+      jwt: jwt,
+      claims: claims,
+      role: role
+    }
+  end
+
+  @spec get_authorizations(Phoenix.Socket.t() | Plug.Conn.t(), DBConnection.t(), __MODULE__.t()) ::
+          {:ok, Phoenix.Socket.t() | Plug.Conn.t()} | {:error, :unauthorized}
+  @doc """
+  Runs validations based on RLS policies to set permissions for a given connection (either Phoenix.Socket or Plug.Conn).
+  """
+  def get_authorizations(%Phoenix.Socket{} = socket, db_conn, params) do
+    case get_permissions_for_connection(db_conn, params) do
+      {:ok, permissions} -> {:ok, Phoenix.Socket.assign(socket, :permissions, permissions)}
+      _ -> {:error, :unauthorized}
+    end
+  end
+
+  def get_authorizations(%Plug.Conn{} = conn, db_conn, params) do
+    case get_permissions_for_connection(db_conn, params) do
+      {:ok, permissions} -> {:ok, Plug.Conn.assign(conn, :permissions, permissions)}
+      _ -> {:error, :unauthorized}
+    end
+  end
+
+  defp get_permissions_for_connection(conn, params) do
+    %__MODULE__{
+      channel_name: channel_name,
+      headers: headers,
+      jwt: jwt,
+      claims: claims,
+      role: role
+    } = params
+
+    sub = Map.get(claims, :sub)
+    claims = Jason.encode!(claims)
+    headers = headers |> Map.new() |> Jason.encode!()
+
+    Postgrex.transaction(conn, fn conn ->
+      Postgrex.query(
+        conn,
+        """
+        SELECT
+         set_config('role', $1, true),
+         set_config('realtime.channel_name', $2, true),
+         set_config('request.jwt.claim.role', $3, true),
+         set_config('request.jwt', $4, true),
+         set_config('request.jwt.claim.sub', $5, true),
+         set_config('request.jwt.claims', $6, true),
+         set_config('request.headers', $7, true)
+        """,
+        [role, channel_name, role, jwt, sub, claims, headers]
+      )
+
+      case Postgrex.query(conn, "SELECT name from realtime.channels", [], mode: :savepoint) do
+        {:ok, %{num_rows: 0}} ->
+          {:ok, %Permissions{read: false}}
+
+        {:ok, _} ->
+          {:ok, %Permissions{read: true}}
+
+        {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege}}} ->
+          {:ok, %Permissions{read: false}}
+
+        {:error, error} ->
+          Logger.error("Error getting permissions for connection: #{inspect(error)}")
+          {:error, error}
+      end
+    end)
+  end
+end

lib/realtime/tenants/migrations.ex

@@ -6,6 +6,7 @@ defmodule Realtime.Tenants.Migrations do
   use GenServer, restart: :transient
 
   require Logger
+
   alias Realtime.Repo
 
   alias Realtime.Tenants.Migrations.{
@@ -39,7 +40,10 @@ defmodule Realtime.Tenants.Migrations do
     ConvertCommitTimestampToUtc,
     OutputFullRecordWhenUnchangedToast,
     CreateListChangesFunction,
-    CreateChannels
+    CreateChannels,
+    SetRequiredGrants,
+    CreateRlsHelperFunctions,
+    EnableChannelsRls
   }
 
   alias Realtime.Helpers, as: H
@@ -76,7 +80,10 @@ defmodule Realtime.Tenants.Migrations do
     {20_230_228_184_745, ConvertCommitTimestampToUtc},
     {20_230_308_225_145, OutputFullRecordWhenUnchangedToast},
     {20_230_328_144_023, CreateListChangesFunction},
-    {20_231_018_144_023, CreateChannels}
+    {20_231_018_144_023, CreateChannels},
+    {20_231_204_144_023, SetRequiredGrants},
+    {20_231_204_144_024, CreateRlsHelperFunctions},
+    {20_231_204_144_025, EnableChannelsRls}
   ]
 
   @spec start_link(GenServer.options()) :: GenServer.on_start()

lib/realtime/tenants/repo/migrations/20231204144023_set_required_grants.ex

@@ -0,0 +1,23 @@
+defmodule Realtime.Tenants.Migrations.SetRequiredGrants do
+  @moduledoc false
+
+  use Ecto.Migration
+
+  def change do
+    execute("""
+    GRANT USAGE ON SCHEMA realtime TO postgres, anon, authenticated, service_role
+    """)
+
+    execute("""
+    GRANT SELECT ON ALL TABLES IN SCHEMA realtime TO postgres, anon, authenticated, service_role
+    """)
+
+    execute("""
+    GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA realtime TO postgres, anon, authenticated, service_role
+    """)
+
+    execute("""
+    GRANT USAGE ON ALL SEQUENCES IN SCHEMA realtime TO postgres, anon, authenticated, service_role
+    """)
+  end
+end

lib/realtime/tenants/repo/migrations/20231204144024_create_rls_helper_functions.ex

@@ -0,0 +1,13 @@
+defmodule Realtime.Tenants.Migrations.CreateRlsHelperFunctions do
+  @moduledoc false
+
+  use Ecto.Migration
+
+  def change do
+    execute("""
+    create or replace function realtime.channel_name() returns text as $$
+    select nullif(current_setting('realtime.channel_name', true), '')::text;
+    $$ language sql stable;
+    """)
+  end
+end

lib/realtime/tenants/repo/migrations/20231204144025_enable_channels_rls.ex

@@ -0,0 +1,9 @@
+defmodule Realtime.Tenants.Migrations.EnableChannelsRls do
+  @moduledoc false
+
+  use Ecto.Migration
+
+  def change do
+    execute("ALTER TABLE realtime.channels ENABLE row level security")
+  end
+end

lib/realtime_web/controllers/channels_controller.ex

@@ -3,12 +3,14 @@ defmodule RealtimeWeb.ChannelsController do
   use OpenApiSpex.ControllerSpecs
 
   alias Realtime.Channels
+  alias Realtime.Tenants.Authorization.Permissions
   alias Realtime.Tenants.Connect
   alias RealtimeWeb.OpenApiSchemas.ChannelParams
   alias RealtimeWeb.OpenApiSchemas.ChannelResponse
   alias RealtimeWeb.OpenApiSchemas.ChannelResponseList
   alias RealtimeWeb.OpenApiSchemas.EmptyResponse
   alias RealtimeWeb.OpenApiSchemas.NotFoundResponse
+  alias RealtimeWeb.OpenApiSchemas.UnauthorizedResponse
 
   action_fallback(RealtimeWeb.FallbackController)
 
@@ -25,17 +27,23 @@ defmodule RealtimeWeb.ChannelsController do
       ]
     ],
     responses: %{
-      200 => ChannelResponseList.response()
+      200 => ChannelResponseList.response(),
+      401 => UnauthorizedResponse.response()
     }
   )
 
-  def index(%{assigns: %{tenant: tenant}} = conn, _params) do
+  def index(
+        %{assigns: %{tenant: tenant, permissions: {:ok, %Permissions{read: true}}}} = conn,
+        _params
+      ) do
     with {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id),
          {:ok, channels} <- Channels.list_channels(db_conn) do
       json(conn, channels)
     end
   end
 
+  def index(_conn, _params), do: {:error, :unauthorized}
+
   operation(:show,
     summary: "Show user channel",
     parameters: [
@@ -57,11 +65,14 @@ defmodule RealtimeWeb.ChannelsController do
     ],
     responses: %{
       200 => ChannelResponse.response(),
+      401 => UnauthorizedResponse.response(),
       404 => NotFoundResponse.response()
     }
   )
 
-  def show(%{assigns: %{tenant: tenant}} = conn, %{"id" => id}) do
+  def show(%{assigns: %{tenant: tenant, permissions: {:ok, %Permissions{read: true}}}} = conn, %{
+        "id" => id
+      }) do
     with {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id),
          {:ok, channel} when not is_nil(channel) <- Channels.get_channel_by_id(id, db_conn) do
       json(conn, channel)
@@ -71,6 +82,8 @@ defmodule RealtimeWeb.ChannelsController do
     end
   end
 
+  def show(_conn, _params), do: {:error, :unauthorized}
+
   operation(:create,
     summary: "Create user channel",
     parameters: [

lib/realtime_web/controllers/fallback_controller.ex

@@ -20,6 +20,13 @@ defmodule RealtimeWeb.FallbackController do
     |> render("error.json", message: "Not found")
   end
 
+  def call(conn, {:error, :unauthorized}) do
+    conn
+    |> put_status(:unauthorized)
+    |> put_view(RealtimeWeb.ErrorView)
+    |> render("error.json", message: "Unauthorized")
+  end
+
   def call(conn, {:error, status, message}) when is_atom(status) and is_binary(message) do
     conn
     |> put_status(status)

lib/realtime_web/open_api_schemas.ex

@@ -380,6 +380,20 @@ defmodule RealtimeWeb.OpenApiSchemas do
     def response(), do: {"Not Found", "application/json", __MODULE__}
   end
 
+  defmodule UnauthorizedResponse do
+    @moduledoc false
+    require OpenApiSpex
+
+    OpenApiSpex.schema(%{
+      type: :object,
+      properties: %{
+        error: %Schema{type: :string, default: "unauthorized"}
+      }
+    })
+
+    def response(), do: {"Unauthorized", "application/json", __MODULE__}
+  end
+
   defmodule UnprocessableEntityResponse do
     @moduledoc false
     require OpenApiSpex

lib/realtime_web/plugs/auth_tenant.ex

@@ -18,11 +18,18 @@ defmodule RealtimeWeb.AuthTenant do
     with %Tenant{jwt_secret: jwt_secret} <- tenant,
          token when is_binary(token) <- access_token(conn),
          jwt_secret_dec <- decrypt!(jwt_secret, secure_key),
-         {:ok, _claims} <- ChannelsAuthorization.authorize_conn(token, jwt_secret_dec) do
+         {:ok, claims} <- ChannelsAuthorization.authorize_conn(token, jwt_secret_dec) do
       Logger.metadata(external_id: tenant.external_id, project: tenant.external_id)
+
       conn
+      |> assign(:claims, claims)
+      |> assign(:jwt, token)
+      |> assign(:role, claims["role"])
     else
-      _ -> unauthorized(conn)
+      _ ->
+        conn
+        |> unauthorized()
+        |> halt()
     end
   end