19715527181

Committed 26 Nov 2025 07:41PM UTC coverage: 74.358% (+13.1%) from 61.246%

Build # 19715527181

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge 9e08c361d into 0224a24c8

Pull Request Pull Request #744: fix(defrag): improve statems, caching, logs, circuit breaking

Run Details

771 of 916 new or added lines in 23 files covered. (84.17%)

3 existing lines in 2 files now uncovered.

2404 of 3233 relevant lines covered (74.36%)

4254.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.9

/lib/supavisor/client_handler/auth.ex

defmodule Supavisor.ClientHandler.Auth do
  @moduledoc """
  Authentication logic for client connections.

  This module handles all authentication-related business logic including:
  - Secret retrieval and caching
  - Credential validation for different auth methods (MD5, SCRAM)
  - Authentication challenge preparation
  - Retry logic for failed authentications
  """

  require Logger

  alias Supavisor.{Helpers, Protocol.Server}
  alias Supavisor.ClientHandler.Auth.{MD5Secrets, PasswordSecrets, SASLSecrets}

  @type auth_method :: :password | :auth_query | :auth_query_md5
  @type auth_secrets :: {auth_method(), function()}

  ## Secret Management

  @doc """
  Retrieves authentication secrets for a user, with caching support.

  For password auth (require_user: true), returns password-based secrets.
  For auth_query, uses cache with TTL or fetches from database.
  """
  @spec get_user_secrets(Supavisor.id(), map(), String.t(), String.t()) ::
          {:ok, auth_secrets()} | {:error, term()}
  def get_user_secrets(
        _id,
        %{user: user, tenant: %{require_user: true}},
        _db_user,
        _tenant_or_alias
      ) do
    secrets = %PasswordSecrets{
      user: user.db_user,
      password: user.db_password
    }

    {:ok, {:password, fn -> secrets end}}
  end

  def get_user_secrets(id, info, db_user, tenant_or_alias) do
    fetch_fn = fn ->
      fetch_secrets_from_database(id, info, db_user)
    end

    Supavisor.SecretCache.fetch_validation_secrets(tenant_or_alias, db_user, fetch_fn)
  end

  ## Authentication Validation

  @doc """
  Validates authentication credentials based on the method.

  Supports password, auth_query, and auth_query_md5 methods.
  Returns {:ok, client_key} on success or {:error, reason} on failure.
  """
  @spec validate_credentials(auth_method(), term(), term(), term()) ::
          {:ok, binary() | nil} | {:error, :wrong_password}
  def validate_credentials(:password, _secrets, signatures, client_proof) do
    if client_proof == signatures.client,
      do: {:ok, nil},
      else: {:error, :wrong_password}
  end

  def validate_credentials(:auth_query, secrets, signatures, client_proof) do
    client_key = :crypto.exor(Base.decode64!(client_proof), signatures.client)

    if Helpers.hash(client_key) == secrets.().stored_key do
      {:ok, client_key}
    else
      {:error, :wrong_password}
    end
  end

  def validate_credentials(:auth_query_md5, server_hash, salt, client_hash) do
    expected_hash = "md5" <> Helpers.md5([server_hash, salt])

    if expected_hash == client_hash,
      do: {:ok, nil},
      else: {:error, :wrong_password}
  end

  ## Challenge Preparation

  @doc """
  Prepares authentication challenge data for SCRAM-based authentication.

  Generates the initial server message and signatures needed for the auth exchange.
  """
  @spec prepare_auth_challenge(auth_method(), function(), binary(), binary(), binary()) ::
          {binary(), map()}
  def prepare_auth_challenge(:password, secret_fn, nonce, user, channel) do
    message = Server.exchange_first_message(nonce)
    server_first_parts = Helpers.parse_server_first(message, nonce)

    {client_final_message, server_proof} =
      Helpers.get_client_final(
        :password,
        secret_fn.(),
        server_first_parts,
        nonce,
        user,
        channel
      )

    signatures = %{
      client: List.last(client_final_message),
      server: server_proof
    }

    {message, signatures}
  end

  def prepare_auth_challenge(:auth_query, secret_fn, nonce, user, channel) do
    secret = secret_fn.()
    message = Server.exchange_first_message(nonce, secret.salt)
    server_first_parts = Helpers.parse_server_first(message, nonce)

    signatures =
      Helpers.signatures(
        secret.stored_key,
        secret.server_key,
        server_first_parts,
        nonce,
        user,
        channel
      )

    {message, signatures}
  end

  ## Cache Update Logic (No Retry)

  @doc """
  Checks if secrets have changed and updates cache if needed

  This is used to detect password changes and refresh the cache (and the pool)
  for future connections
  """
  @spec check_and_update_secrets(
          auth_method(),
          term(),
          Supavisor.id(),
          map(),
          String.t(),
          String.t(),
          function()
        ) :: :ok
  def check_and_update_secrets(method, reason, client_id, info, tenant, user, current_secrets_fn) do
    if method != :password and reason == :wrong_password and
         not Supavisor.CacheRefreshLimiter.cache_refresh_limited?(client_id) do
      case fetch_secrets_from_database(client_id, info, user) do
        {:ok, {method2, secrets2}} ->
          current_secrets = current_secrets_fn.()
          new_secrets = secrets2.()

          if method != method2 or
               Map.delete(current_secrets, :client_key) != Map.delete(new_secrets, :client_key) do
            Logger.warning("ClientHandler: Update secrets")
            Supavisor.SecretCache.put_validation_secrets(tenant, user, method2, secrets2)
          end

        other ->
          Logger.error("ClientHandler: Auth secrets check error: #{inspect(other)}")
      end
    else
      Logger.debug("ClientHandler: No cache check needed")
    end

    :ok
  end

  ## Message Parsing

  @doc """
  Parses authentication response packets for different auth methods.

  Returns parsed credentials or error information.
  """
  @spec parse_auth_message(binary(), auth_method()) ::
          {:ok, term()} | {:error, term()}
  def parse_auth_message(bin, :auth_query_md5) do
    case Server.decode_pkt(bin) do
      {:ok, %{tag: :password_message, payload: {:md5, client_md5}}, _} ->
        {:ok, client_md5}

      {:error, error} ->
        {:error, {:decode_error, error}}

      other ->
        {:error, {:unexpected_message, other}}
    end
  end

  def parse_auth_message(bin, _scram_method) do
    case Server.decode_pkt(bin) do
      {:ok,
       %{
         tag: :password_message,
         payload: {:scram_sha_256, %{"n" => user, "r" => nonce, "c" => channel}}
       }, _} ->
        {:ok, {user, nonce, channel}}

      {:ok, %{tag: :password_message, payload: {:first_msg_response, %{"p" => p}}}, _} ->
        {:ok, p}

      {:error, error} ->
        {:error, {:decode_error, error}}

      other ->
        {:error, {:unexpected_message, other}}
    end
  end

  ## Authentication Context Management

  @doc """
  Creates initial authentication context for a given method and secrets.
  """
  @spec create_auth_context(auth_method(), function(), map()) :: map()
  def create_auth_context(:auth_query_md5, secrets, info) do
    salt = :crypto.strong_rand_bytes(4)

    %{
      method: :auth_query_md5,
      secrets: secrets,
      salt: salt,
      info: info,
      signatures: nil
    }
  end

  def create_auth_context(method, secrets, info) when method in [:password, :auth_query] do
    %{
      method: method,
      secrets: secrets,
      info: info,
      signatures: nil
    }
  end

  @doc """
  Updates authentication context with new signatures after challenge exchange.
  """
  @spec update_auth_context_with_signatures(map(), map()) :: map()
  def update_auth_context_with_signatures(auth_context, signatures) do
    %{auth_context | signatures: signatures}
  end

  ## Success Response Preparation

  @doc """
  Builds the final SCRAM response message for successful authentication.

  Takes the auth context and returns the complete protocol message to send.
  """
  @spec build_scram_final_response(map()) :: iodata()
  def build_scram_final_response(%{signatures: %{server: server_signature}}) do
    message = "v=#{Base.encode64(server_signature)}"
    Server.exchange_message(:final, message)
  end

  @doc """
  Prepares final secrets after successful authentication.

  Adds client_key to secrets if provided, otherwise returns original secrets.
  """
  @spec prepare_final_secrets(function(), binary() | nil) :: function()
  def prepare_final_secrets(secrets_fn, nil), do: secrets_fn

  def prepare_final_secrets(secrets_fn, client_key) do
    fn -> Map.put(secrets_fn.(), :client_key, client_key) end
  end

  ## Private Helpers

  @spec fetch_secrets_from_database(Supavisor.id(), map(), String.t()) ::
          {:ok, auth_secrets()} | {:error, term()}
  defp fetch_secrets_from_database(id, %{user: user, tenant: tenant}, db_user) do
    case Supavisor.SecretChecker.get_secrets(id) do
      {:error, :not_started} ->
        Logger.info(
          "ClientHandler: secret checker not started, starting a new database connection"
        )

        ssl_opts = build_ssl_options(tenant)

        {:ok, conn} =
          Postgrex.start_link(
            hostname: tenant.db_host,
            port: tenant.db_port,
            database: tenant.db_database,
            password: user.db_password,
            username: user.db_user,
            parameters: [application_name: "Supavisor auth_query"],
            ssl: tenant.upstream_ssl,
            socket_options: [
              Helpers.ip_version(tenant.ip_version, tenant.db_host)
            ],
            queue_target: 1_000,
            queue_interval: 5_000,
            ssl_opts: ssl_opts
          )

        try do
          Logger.debug(
            "ClientHandler: Connected to db #{tenant.db_host} #{tenant.db_port} #{tenant.db_database} #{user.db_user}"
          )

          with {:ok, secret} <- Helpers.get_user_secret(conn, tenant.auth_query, db_user) do
            auth_type =
              case secret do
                %MD5Secrets{} -> :auth_query_md5
                %SASLSecrets{} -> :auth_query
              end

            {:ok, {auth_type, fn -> secret end}}
          end
        rescue
          exception ->
            Logger.error("ClientHandler: Couldn't fetch user secrets from #{tenant.db_host}")
            reraise exception, __STACKTRACE__
        after
          Process.unlink(conn)

          spawn(fn ->
            try do
              GenServer.stop(conn, :normal, 5_000)
            catch
              :exit, _ -> Process.exit(conn, :kill)
            end
          end)
        end

      secrets ->
        secrets
    end
  end

  defp build_ssl_options(%{upstream_ssl: true, upstream_verify: :peer} = tenant) do
    [
      verify: :verify_peer,
      cacerts: [Helpers.upstream_cert(tenant.upstream_tls_ca)],
      server_name_indication: String.to_charlist(tenant.db_host),
      customize_hostname_check: [{:match_fun, fn _, _ -> true end}]
    ]
  end

  defp build_ssl_options(_tenant) do
    [verify: :verify_none]
  end
end

1	defmodule Supavisor.ClientHandler.Auth do
2	@moduledoc """
3	Authentication logic for client connections.
4
5	This module handles all authentication-related business logic including:
6	- Secret retrieval and caching
7	- Credential validation for different auth methods (MD5, SCRAM)
8	- Authentication challenge preparation
9	- Retry logic for failed authentications
10	"""
11
12	require Logger
13
14	alias Supavisor.{Helpers, Protocol.Server}
15	alias Supavisor.ClientHandler.Auth.{MD5Secrets, PasswordSecrets, SASLSecrets}
16
17	@type auth_method :: :password \| :auth_query \| :auth_query_md5
18	@type auth_secrets :: {auth_method(), function()}
19
20	## Secret Management
21
22	@doc """
23	Retrieves authentication secrets for a user, with caching support.
24
25	For password auth (require_user: true), returns password-based secrets.
26	For auth_query, uses cache with TTL or fetches from database.
27	"""
28	@spec get_user_secrets(Supavisor.id(), map(), String.t(), String.t()) ::
29	{:ok, auth_secrets()} \| {:error, term()}
30	def get_user_secrets(
31	_id,
32	%{user: user, tenant: %{require_user: true}},
33	_db_user,
34	_tenant_or_alias
35	) do
36	secrets = %PasswordSecrets{	52✔
37	user: user.db_user,	52✔
38	password: user.db_password	52✔
39	}
40
41	{:ok, {:password, fn -> secrets end}}	165✔
42	end
43
44	def get_user_secrets(id, info, db_user, tenant_or_alias) do
45	fetch_fn = fn ->	19✔
46	fetch_secrets_from_database(id, info, db_user)	15✔
47	end
48
49	Supavisor.SecretCache.fetch_validation_secrets(tenant_or_alias, db_user, fetch_fn)	19✔
50	end
51
52	## Authentication Validation
53
54	@doc """
55	Validates authentication credentials based on the method.
56
57	Supports password, auth_query, and auth_query_md5 methods.
58	Returns {:ok, client_key} on success or {:error, reason} on failure.
59	"""
60	@spec validate_credentials(auth_method(), term(), term(), term()) ::
61	{:ok, binary() \| nil} \| {:error, :wrong_password}
62	def validate_credentials(:password, _secrets, signatures, client_proof) do
63	if client_proof == signatures.client,	51✔
64	do: {:ok, nil},
65	else: {:error, :wrong_password}
66	end
67
68	def validate_credentials(:auth_query, secrets, signatures, client_proof) do
69	client_key = :crypto.exor(Base.decode64!(client_proof), signatures.client)	19✔
70
71	if Helpers.hash(client_key) == secrets.().stored_key do	19✔
72	{:ok, client_key}
73	else
74	{:error, :wrong_password}
75	end
76	end
77
78	def validate_credentials(:auth_query_md5, server_hash, salt, client_hash) do
NEW 79	expected_hash = "md5" <> Helpers.md5([server_hash, salt])	×
80
NEW 81	if expected_hash == client_hash,	×
82	do: {:ok, nil},
83	else: {:error, :wrong_password}
84	end
85
86	## Challenge Preparation
87
88	@doc """
89	Prepares authentication challenge data for SCRAM-based authentication.
90
91	Generates the initial server message and signatures needed for the auth exchange.
92	"""
93	@spec prepare_auth_challenge(auth_method(), function(), binary(), binary(), binary()) ::
94	{binary(), map()}
95	def prepare_auth_challenge(:password, secret_fn, nonce, user, channel) do
96	message = Server.exchange_first_message(nonce)	51✔
97	server_first_parts = Helpers.parse_server_first(message, nonce)	51✔
98
99	{client_final_message, server_proof} =	51✔
100	Helpers.get_client_final(
101	:password,
102	secret_fn.(),
103	server_first_parts,
104	nonce,
105	user,
106	channel
107	)
108
109	signatures = %{	51✔
110	client: List.last(client_final_message),
111	server: server_proof
112	}
113
114	{message, signatures}
115	end
116
117	def prepare_auth_challenge(:auth_query, secret_fn, nonce, user, channel) do
118	secret = secret_fn.()	19✔
119	message = Server.exchange_first_message(nonce, secret.salt)	19✔
120	server_first_parts = Helpers.parse_server_first(message, nonce)	19✔
121
122	signatures =	19✔
123	Helpers.signatures(
124	secret.stored_key,	19✔
125	secret.server_key,	19✔
126	server_first_parts,
127	nonce,
128	user,
129	channel
130	)
131
132	{message, signatures}
133	end
134
135	## Cache Update Logic (No Retry)
136
137	@doc """
138	Checks if secrets have changed and updates cache if needed
139
140	This is used to detect password changes and refresh the cache (and the pool)
141	for future connections
142	"""
143	@spec check_and_update_secrets(
144	auth_method(),
145	term(),
146	Supavisor.id(),
147	map(),
148	String.t(),
149	String.t(),
150	function()
151	) :: :ok
152	def check_and_update_secrets(method, reason, client_id, info, tenant, user, current_secrets_fn) do
153	if method != :password and reason == :wrong_password and	13✔
154	not Supavisor.CacheRefreshLimiter.cache_refresh_limited?(client_id) do	1✔
155	case fetch_secrets_from_database(client_id, info, user) do	1✔
156	{:ok, {method2, secrets2}} ->
157	current_secrets = current_secrets_fn.()	1✔
158	new_secrets = secrets2.()	1✔
159
160	if method != method2 or	1✔
161	Map.delete(current_secrets, :client_key) != Map.delete(new_secrets, :client_key) do	1✔
162	Logger.warning("ClientHandler: Update secrets")	1✔
163	Supavisor.SecretCache.put_validation_secrets(tenant, user, method2, secrets2)	1✔
164	end
165
166	other ->
NEW 167	Logger.error("ClientHandler: Auth secrets check error: #{inspect(other)}")	×
168	end
169	else
170	Logger.debug("ClientHandler: No cache check needed")	12✔
171	end
172
173	:ok
174	end
175
176	## Message Parsing
177
178	@doc """
179	Parses authentication response packets for different auth methods.
180
181	Returns parsed credentials or error information.
182	"""
183	@spec parse_auth_message(binary(), auth_method()) ::
184	{:ok, term()} \| {:error, term()}
185	def parse_auth_message(bin, :auth_query_md5) do
NEW 186	case Server.decode_pkt(bin) do	×
NEW 187	{:ok, %{tag: :password_message, payload: {:md5, client_md5}}, _} ->	×
188	{:ok, client_md5}
189
NEW 190	{:error, error} ->	×
191	{:error, {:decode_error, error}}
192
NEW 193	other ->	×
194	{:error, {:unexpected_message, other}}
195	end
196	end
197
198	def parse_auth_message(bin, _scram_method) do
199	case Server.decode_pkt(bin) do	140✔
200	{:ok,
201	%{
202	tag: :password_message,
203	payload: {:scram_sha_256, %{"n" => user, "r" => nonce, "c" => channel}}
204	}, _} ->	70✔
205	{:ok, {user, nonce, channel}}
206
207	{:ok, %{tag: :password_message, payload: {:first_msg_response, %{"p" => p}}}, _} ->	70✔
208	{:ok, p}
209
NEW 210	{:error, error} ->	×
211	{:error, {:decode_error, error}}
212
NEW 213	other ->	×
214	{:error, {:unexpected_message, other}}
215	end
216	end
217
218	## Authentication Context Management
219
220	@doc """
221	Creates initial authentication context for a given method and secrets.
222	"""
223	@spec create_auth_context(auth_method(), function(), map()) :: map()
224	def create_auth_context(:auth_query_md5, secrets, info) do
NEW 225	salt = :crypto.strong_rand_bytes(4)	×
226
NEW 227	%{	×
228	method: :auth_query_md5,
229	secrets: secrets,
230	salt: salt,
231	info: info,
232	signatures: nil
233	}
234	end
235
236	def create_auth_context(method, secrets, info) when method in [:password, :auth_query] do
237	%{	70✔
238	method: method,
239	secrets: secrets,
240	info: info,
241	signatures: nil
242	}
243	end
244
245	@doc """
246	Updates authentication context with new signatures after challenge exchange.
247	"""
248	@spec update_auth_context_with_signatures(map(), map()) :: map()
249	def update_auth_context_with_signatures(auth_context, signatures) do
250	%{auth_context \| signatures: signatures}	70✔
251	end
252
253	## Success Response Preparation
254
255	@doc """
256	Builds the final SCRAM response message for successful authentication.
257
258	Takes the auth context and returns the complete protocol message to send.
259	"""
260	@spec build_scram_final_response(map()) :: iodata()
261	def build_scram_final_response(%{signatures: %{server: server_signature}}) do
262	message = "v=#{Base.encode64(server_signature)}"	57✔
263	Server.exchange_message(:final, message)	57✔
264	end
265
266	@doc """
267	Prepares final secrets after successful authentication.
268
269	Adds client_key to secrets if provided, otherwise returns original secrets.
270	"""
271	@spec prepare_final_secrets(function(), binary() \| nil) :: function()
272	def prepare_final_secrets(secrets_fn, nil), do: secrets_fn	39✔
273
274	def prepare_final_secrets(secrets_fn, client_key) do
275	fn -> Map.put(secrets_fn.(), :client_key, client_key) end	18✔
276	end
277
278	## Private Helpers
279
280	@spec fetch_secrets_from_database(Supavisor.id(), map(), String.t()) ::
281	{:ok, auth_secrets()} \| {:error, term()}
282	defp fetch_secrets_from_database(id, %{user: user, tenant: tenant}, db_user) do
283	case Supavisor.SecretChecker.get_secrets(id) do	16✔
284	{:error, :not_started} ->
285	Logger.info(	15✔
286	"ClientHandler: secret checker not started, starting a new database connection"
287	)
288
289	ssl_opts = build_ssl_options(tenant)	15✔
290
291	{:ok, conn} =	15✔
292	Postgrex.start_link(
293	hostname: tenant.db_host,	15✔
294	port: tenant.db_port,	15✔
295	database: tenant.db_database,	15✔
296	password: user.db_password,	15✔
297	username: user.db_user,	15✔
298	parameters: [application_name: "Supavisor auth_query"],
299	ssl: tenant.upstream_ssl,	15✔
300	socket_options: [
301	Helpers.ip_version(tenant.ip_version, tenant.db_host)	15✔
302	],
303	queue_target: 1_000,
304	queue_interval: 5_000,
305	ssl_opts: ssl_opts
306	)
307
308	try do	15✔
309	Logger.debug(	15✔
310	"ClientHandler: Connected to db #{tenant.db_host} #{tenant.db_port} #{tenant.db_database} #{user.db_user}"	15✔
311	)
312
313	with {:ok, secret} <- Helpers.get_user_secret(conn, tenant.auth_query, db_user) do	15✔
314	auth_type =	15✔
315	case secret do
NEW 316	%MD5Secrets{} -> :auth_query_md5	×
317	%SASLSecrets{} -> :auth_query	15✔
318	end
319
320	{:ok, {auth_type, fn -> secret end}}	126✔
321	end
322	rescue
NEW 323	exception ->	×
NEW 324	Logger.error("ClientHandler: Couldn't fetch user secrets from #{tenant.db_host}")	×
NEW 325	reraise exception, __STACKTRACE__	×
326	after
327	Process.unlink(conn)	15✔
328
329	spawn(fn ->	15✔
330	try do	15✔
331	GenServer.stop(conn, :normal, 5_000)	15✔
332	catch
NEW 333	:exit, _ -> Process.exit(conn, :kill)	×
334	end
335	end)
336	end
337
338	secrets ->
339	secrets	1✔
340	end
341	end
342
NEW 343	defp build_ssl_options(%{upstream_ssl: true, upstream_verify: :peer} = tenant) do	×
344	[
345	verify: :verify_peer,
NEW 346	cacerts: [Helpers.upstream_cert(tenant.upstream_tls_ca)],	×
NEW 347	server_name_indication: String.to_charlist(tenant.db_host),	×
NEW 348	customize_hostname_check: [{:match_fun, fn _, _ -> true end}]	×
349	]
350	end
351
352	defp build_ssl_options(_tenant) do	15✔
353	[verify: :verify_none]
354	end
355	end

supabase / supavisor / 19715527181

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous