diff --git a/README.md b/README.md index 465d5c9f..31ad742b 100644 --- a/README.md +++ b/README.md @@ -144,10 +144,73 @@ Requires connecting with a user with appropriate rights. More info [here](https://docs.microsoft.com/en-us/dotnet/framework/data/adonet/sql/snapshot-isolation-in-sql-server). -## Federation Authentication +## Federated Authentication + +Azure SQL Database and Microsoft Fabric SQL endpoints accept a Microsoft Entra ID +access token instead of a username and password. Acquire the token yourself +(scope `https://database.windows.net/.default`) and pass it as `:access_token`. +Username and password are ignored, and encryption is required. + +`:access_token` accepts a string, a zero-arity function or an `{module, function, args}` +tuple. Functions and MFAs are called on every connect and return the token, +`{:ok, token}` or `{:error, reason}`, so connections opened after the token expired get +a fresh one. If the function returns `{:error, reason}`, raises, throws, exits or takes +longer than `:connect_timeout` (default 15 seconds), the connect fails with a +`Tds.Error` and the pool retries with its usual backoff. A raised exception or an +`{:error, reason}` struct is reported by its module name only (an atom reason as is, +any other reason not at all), so secrets in it never reach the logs. -This Authentication mechanism is not supported. -This functionality requires specific environment to be developed. +```elixir +config :your_app, :tds_conn, + hostname: "xxxx.datawarehouse.fabric.microsoft.com", + database: "my_warehouse", + access_token: {MyApp.Entra, :fetch_sql_token, []}, + ssl: true, + ssl_opts: [ + verify: :verify_peer, + cacerts: :public_key.cacerts_get(), + server_name_indication: ~c"xxxx.datawarehouse.fabric.microsoft.com", + customize_hostname_check: [ + match_fun: :public_key.pkix_verify_hostname_match_fun(:https) + ] + ] +``` + +When the server routes the login elsewhere (the Azure SQL gateway redirect), the driver +closes the gateway connection and connects to the routed host with the same options, +except that a pinned `server_name_indication` is replaced by the routed host name, so +the hostname check runs against the server that is actually reached. + +Only the Security Token workflow is supported: the driver does not acquire tokens +itself (no ADAL/MSAL `FEDAUTHINFO` exchange). + +### Known limitations + +A connect can block indefinitely if the server stops responding during the TLS +handshake or before it answers LOGIN7: the handshake runs with an `:infinity` timeout +and the login response is read without one. This predates federated authentication +and applies to every login. + +## Limiting response size + +The driver buffers a whole response, across every result set, before decoding it. +Set `max_response_bytes: pos_integer()` to cap that buffer (anything else but `nil` +raises `ArgumentError` at `start_link`). The cap applies to query responses only, not +to the PRELOGIN, LOGIN7 and session setup exchanges of a connect. The count covers the +bytes read for one response, TDS packet headers included; over TLS these are the +decrypted bytes. Once it goes over the limit the driver stops reading, closes the +connection (the rest of the response is still on the wire) and the query returns: + +```elixir +{:error, %Tds.ResponseTooLargeError{ + limit: 10_000_000, + received: 10_002_432, + message: "response exceeded max_response_bytes (limit 10000000 bytes)" + }} +``` + +`received` is the number of bytes read when the driver stopped. The pool opens a new +connection for the next checkout. Without the option there is no limit. ## Data representation diff --git a/lib/tds.ex b/lib/tds.ex index bf3e7d1b..cf4051d2 100644 --- a/lib/tds.ex +++ b/lib/tds.ex @@ -27,11 +27,25 @@ defmodule Tds do | {:database, String.t()} | {:username, String.t()} | {:password, String.t()} + | {:access_token, access_token()} + | {:max_response_bytes, pos_integer()} | {:timeout, timeout()} | {:connect_timeout, timeout()} | {:execution_mode, :prepare_execute | :executesql} | DBConnection.start_option() + @typedoc """ + A Microsoft Entra ID access token for federated authentication, or a + function / MFA returning `token`, `{:ok, token}` or `{:error, reason}` that is + called on every connect so pooled connections can pick up a refreshed token. + A function that fails, raises or runs past `:connect_timeout` fails the + connect, and the pool backs off and retries. + """ + @type access_token :: + String.t() + | (-> String.t() | {:ok, String.t()} | {:error, term}) + | {module(), atom(), [term()]} + @type isolation_level :: :read_uncommitted | :read_committed @@ -179,11 +193,21 @@ defmodule Tds do end defp default(opts) do + validate_max_response_bytes!(opts[:max_response_bytes]) + opts |> Keyword.put_new(:idle_timeout, @timeout) |> Keyword.put_new(:execution_mode, @execution_mode) end + defp validate_max_response_bytes!(nil), do: :ok + defp validate_max_response_bytes!(limit) when is_integer(limit) and limit > 0, do: :ok + + defp validate_max_response_bytes!(limit) do + raise ArgumentError, + "expected :max_response_bytes to be a positive integer or nil, got: #{inspect(limit)}" + end + @doc """ Returns the configured JSON library. diff --git a/lib/tds/messages.ex b/lib/tds/messages.ex index 5294e803..734f16d0 100644 --- a/lib/tds/messages.ex +++ b/lib/tds/messages.ex @@ -23,7 +23,8 @@ defmodule Tds.Messages do # responses defrecord :msg_preloginack, [:response] - defrecord :msg_loginack, [:redirect] + # features: FeatureIds the server acknowledged in FEATUREEXTACK + defrecord :msg_loginack, [:redirect, features: []] defrecord :msg_prepared, [:params] defrecord :msg_sql_result, [:columns, :rows, :row_count] defrecord :msg_result, [:set, :params, :status] @@ -79,6 +80,9 @@ defmodule Tds.Messages do {:envchange, other}, {msg, s} -> {msg, on_envchange(other, s)} + {:featureextack, features}, {msg_loginack() = msg, s} -> + {msg_loginack(msg, features: Enum.map(features, &elem(&1, 0))), s} + {:loginack, %{tds_version: version}}, {msg, s} -> Process.put(:tds_version, version) {msg, s} @@ -93,8 +97,21 @@ defmodule Tds.Messages do {:error, error}, _ -> {msg_error(error: error), s} + {:fedauthinfo, _}, {msg_error(), _s} = msg_s -> + msg_s + + {:fedauthinfo, _}, {_, s} -> + error = %{ + line_number: 0, + number: 0, + msg_text: + "Server requested FEDAUTHINFO, which is not supported. " <> + "Pass a pre-acquired :access_token instead" + } + + {msg_error(error: error), s} + _, msg -> - # FeatureExtAck should be processed here in future msg end) end diff --git a/lib/tds/protocol.ex b/lib/tds/protocol.ex index f89e1534..27990844 100644 --- a/lib/tds/protocol.ex +++ b/lib/tds/protocol.ex @@ -8,6 +8,7 @@ defmodule Tds.Protocol do use DBConnection @timeout 5_000 + @connect_timeout 15_000 @sock_opts [packet: :raw, mode: :binary, active: false] @trans_levels [ :read_uncommitted, @@ -42,7 +43,10 @@ defmodule Tds.Protocol do result: nil | list(), query: nil | String.t(), transaction: transaction, - env: env + env: env, + fed_auth_echo: boolean(), + access_token: nil | String.t(), + max_response_bytes: nil | pos_integer() } defstruct sock: nil, @@ -59,7 +63,16 @@ defmodule Tds.Protocol do savepoint: 0, collation: %Tds.Protocol.Collation{}, packetsize: 4096 - } + }, + # FEDAUTHREQUIRED value from the server's PRELOGIN response, + # echoed back in the LOGIN7 FEDAUTH feature extension + fed_auth_echo: false, + # Resolved access token, only kept until login succeeds. It is + # redacted from opts and hidden from inspect. + access_token: nil, + # :max_response_bytes, set once login is done so the cap only + # applies to query responses + max_response_bytes: nil @spec connect(opts :: Keyword.t()) :: {:ok, state :: t()} | {:error, Exception.t()} def connect(opts) do @@ -71,7 +84,13 @@ defmodule Tds.Protocol do |> Keyword.put_new(:hostname, System.get_env("MSSQLHOST") || "localhost") |> Enum.reject(fn {_k, v} -> is_nil(v) end) - s = %__MODULE__{} + with {:ok, opts, access_token} <- resolve_access_token(opts) do + connect_resolved(opts, access_token) + end + end + + defp connect_resolved(opts, access_token) do + s = %__MODULE__{access_token: access_token} case opts[:instance] do nil -> @@ -85,6 +104,146 @@ defmodule Tds.Protocol do end end + # The access token is resolved on every connect, so a function or MFA can + # hand out a fresh token when the pool (re)connects after the last one expired. + defp resolve_access_token(opts) do + case Keyword.fetch(opts, :access_token) do + :error -> + {:ok, opts, nil} + + {:ok, token} -> + timeout = Keyword.get(opts, :connect_timeout, @connect_timeout) + + with :ok <- validate_access_token_ssl(opts), + {:ok, token} <- fetch_access_token(token, timeout) do + warn_unverified_tls(opts) + {:ok, Keyword.put(opts, :access_token, :REDACTED), token} + end + end + end + + defp warn_unverified_tls(opts) do + if opts[:ssl_opts][:verify] == :verify_none do + Logger.warning( + "Tds is sending an :access_token over TLS with verify: :verify_none, " <> + "so the server certificate is not checked and the token can be intercepted" + ) + end + end + + # The token is a bearer credential, never send it over an unencrypted connection + defp validate_access_token_ssl(opts) do + if opts[:ssl] in [true, :required, :on] do + :ok + else + {:error, Tds.Error.exception(":access_token requires ssl: true (or :required)")} + end + end + + # A function or MFA runs in its own process, so a raise, throw, exit or hang + # becomes a connect error (and DBConnection backs off) instead of crashing + # the connection process. Only the exception module is reported, its + # message could carry a secret. + defp fetch_access_token(fun, timeout) when is_function(fun, 0), + do: call_access_token(fun, timeout) + + defp fetch_access_token({mod, fun, args}, timeout) + when is_atom(mod) and is_atom(fun) and is_list(args), + do: call_access_token(fn -> apply(mod, fun, args) end, timeout) + + defp fetch_access_token(token, _timeout), do: normalize_access_token(token) + + defp call_access_token(fun, timeout) do + parent = self() + tag = make_ref() + + # The result goes back as a message and the resolver exits :normal, so + # the token never travels in an exit signal to processes the function + # linked to + {pid, ref} = + spawn_monitor(fn -> + watch_parent(parent, self()) + + result = + try do + {:ok, fun.()} + rescue + exception -> {:failed, inspect(exception.__struct__)} + catch + kind, _reason -> {:failed, Atom.to_string(kind)} + end + + send(parent, {tag, result}) + end) + + receive do + {^tag, result} -> + Process.demonitor(ref, [:flush]) + + case result do + {:ok, token} -> normalize_access_token(token) + {:failed, reason} -> access_token_failed(reason) + end + + {:DOWN, ^ref, :process, ^pid, _reason} -> + access_token_failed("exit") + after + timeout -> + Process.exit(pid, :kill) + + # a result sent just before the kill arrives ahead of :DOWN, drop it + # so the token does not linger in this mailbox + receive do + {:DOWN, ^ref, :process, ^pid, _reason} -> :ok + end + + receive do + {^tag, _result} -> :ok + after + 0 -> :ok + end + + access_token_failed("timed out after #{timeout}ms") + end + end + + # Kills the resolver if the connecting process dies first, and goes away + # with the resolver otherwise + defp watch_parent(parent, resolver) do + spawn(fn -> + parent_ref = Process.monitor(parent) + resolver_ref = Process.monitor(resolver) + + receive do + {:DOWN, ^parent_ref, :process, _, _} -> Process.exit(resolver, :kill) + {:DOWN, ^resolver_ref, :process, _, _} -> :ok + end + end) + end + + defp access_token_failed(reason), + do: {:error, Tds.Error.exception("access token could not be fetched: #{reason}")} + + defp normalize_access_token(token) when is_binary(token) and token != "", do: {:ok, token} + + defp normalize_access_token({:ok, token}) when is_binary(token) and token != "", + do: {:ok, token} + + # The reason may hold credentials, only its atom or struct name is reported + defp normalize_access_token({:error, reason}) do + message = + case reason do + %module{} -> "unable to fetch access token: #{inspect(module)}" + atom when is_atom(atom) -> "unable to fetch access token: #{inspect(atom)}" + _other -> "unable to fetch access token" + end + + {:error, Tds.Error.exception(message)} + end + + defp normalize_access_token(_other), + do: {:error, Tds.Error.exception("invalid :access_token, expected a non-empty string")} + @spec disconnect(err :: Exception.t() | String.t(), state :: t()) :: :ok def disconnect(_err, %{sock: {mod, sock}} = s) do # If socket is active we flush any socket messages so the next @@ -99,8 +258,8 @@ defmodule Tds.Protocol do {:ok, _, s} -> {:ok, s} - {:disconnect, :closed, s} -> - {:disconnect, %Tds.Error{message: "Connection closed."}, s} + {:disconnect, _err, _s} = disconnect -> + disconnect {:error, err, s} -> err = @@ -363,16 +522,40 @@ defmodule Tds.Protocol do s = %{s | opts: opts} - # Initalize TCP connection with the SQL Server - case :gen_tcp.connect(host, port, sock_opts, timeout) do + # SOCKS5 proxy support + proxy_host = Keyword.get(opts, :proxy_host) + proxy_port = Keyword.get(opts, :proxy_port, 1080) + proxy_username = Keyword.get(opts, :proxy_username) + proxy_password = Keyword.get(opts, :proxy_password) + + connection_result = + if proxy_host do + socks_connect( + proxy_host, + proxy_port, + host, + port, + proxy_username, + proxy_password, + sock_opts, + timeout + ) + else + :gen_tcp.connect(host, port, sock_opts, timeout) + end + + # Initialize TCP connection with the SQL Server + case connection_result do {:ok, sock} -> with {:ok, buffers} <- :inet.getopts(sock, [:sndbuf, :recbuf, :buffer]), :ok <- :inet.setopts(sock, buffer: max_buf_size(buffers)), {:ok, s} <- send_prelogin(%{s | sock: {:gen_tcp, sock}}) do {:ok, s} else - {_error_or_disconnect, exception, _state} -> - :gen_tcp.close(sock) + {error_or_disconnect, exception, state} + when error_or_disconnect in [:error, :disconnect] -> + # PRELOGIN may have upgraded the connection to TLS. + disconnect(exception, state) {:error, exception} {:error, error} -> @@ -385,6 +568,59 @@ defmodule Tds.Protocol do end end + defp socks_connect( + proxy_host, + proxy_port, + target_host, + target_port, + username, + password, + sock_opts, + timeout + ) do + with {:ok, sock} <- + :gen_tcp.connect(String.to_charlist(proxy_host), proxy_port, sock_opts, timeout) do + # SOCKS5 greeting: version 5, support no-auth (0x00) and user/pass (0x02) + :gen_tcp.send(sock, <<5, 2, 0, 2>>) + {:ok, <<5, method>>} = :gen_tcp.recv(sock, 2) + + case method do + # User/pass auth + 2 -> + user_bin = :erlang.iolist_to_binary(username || "") + pass_bin = :erlang.iolist_to_binary(password || "") + + :gen_tcp.send( + sock, + <<1, byte_size(user_bin)>> <> user_bin <> <> <> pass_bin + ) + + {:ok, <<1, status>>} = :gen_tcp.recv(sock, 2) + if status != 0, do: raise("SOCKS5 auth failed") + + # No auth, proceed + 0 -> + :ok + + _ -> + raise "SOCKS5 unsupported method" + end + + # Connect request: ver 5, cmd 1 (connect), rsv 0, addr type 3 (domain), addr len + addr, port + host_bin = :erlang.iolist_to_binary(target_host) + + :gen_tcp.send( + sock, + <<5, 1, 0, 3, byte_size(host_bin)>> <> + host_bin <> <> + ) + + # Read full reply + {:ok, <<5, rep, 0, _atype, _rest::binary>>} = :gen_tcp.recv(sock, 0) + if rep == 0, do: {:ok, sock}, else: raise("SOCKS5 connect failed: #{rep}") + end + end + defp parse_udp( {_, 1434, <<_head::binary-3, data::binary>>}, %{opts: opts, usock: sock} = s @@ -495,6 +731,9 @@ defmodule Tds.Protocol do {:error, _, _} = err -> err + + {:disconnect, _, _} = err -> + err end end @@ -528,11 +767,16 @@ defmodule Tds.Protocol do end def login(%{opts: opts} = s) do - msg = msg_login(params: opts) + params = + opts + |> Keyword.put(:fed_auth_echo, s.fed_auth_echo) + |> Keyword.put(:access_token, s.access_token) + + msg = msg_login(params: params) case login_send(msg, %{s | state: :login}) do {:ok, s} -> - {:ok, %{s | state: :ready}} + {:ok, %{s | state: :ready, max_response_bytes: opts[:max_response_bytes]}} err -> err @@ -592,11 +836,11 @@ defmodule Tds.Protocol do {:ok, %{result: result} = s} -> {:ok, result, s} - {:error, err} -> - {:disconnect, err, s} - {:error, err, s} -> {:disconnect, err, s} + + {:disconnect, _err, _s} = disconnect -> + disconnect end end @@ -744,22 +988,42 @@ defmodule Tds.Protocol do def message( :login, - msg_loginack(redirect: %{hostname: host, port: port}), - %{opts: opts} + msg_loginack(redirect: %{hostname: host, port: port} = route), + %{opts: opts, sock: {mod, sock}} = s ) do + # The gateway leg is done with, close it before following the route + _ = mod.close(sock) + opts |> Keyword.put(:hostname, host) |> Keyword.put(:port, port) - |> connect() + # LOGIN7 names the server the gateway routed to, with its instance: + # a Fabric backend hangs up on a login that names the bare host + |> Keyword.put(:routed_server_name, routed_server_name(route)) + |> Keyword.update(:ssl_opts, [], &route_ssl_opts(&1, host)) + # the routed login reuses the token resolved for this connect + |> connect_resolved(s.access_token) + |> case do + {:ok, s} -> {:ok, s} + {:error, err} -> {:error, err, s} + end end - def message(:login, msg_loginack(), %{opts: opts} = s) do - state = %{s | opts: clean_opts(opts)} + # FEDAUTH FeatureId. Without its FEATUREEXTACK the server did not log in + # with the token. + @fed_auth_feature 0x02 - opts - |> conn_opts() - |> IO.iodata_to_binary() - |> send_query(state) + def message(:login, msg_loginack(features: features), %{opts: opts} = s) do + if is_binary(s.access_token) and @fed_auth_feature not in features do + {:error, Tds.Error.exception("server did not acknowledge federated authentication"), s} + else + state = %{s | opts: clean_opts(opts), access_token: nil} + + opts + |> conn_opts() + |> IO.iodata_to_binary() + |> send_query(state) + end end def message(:executing, msg_result(set: set), s) do @@ -806,6 +1070,22 @@ defmodule Tds.Protocol do {:ok, %{s | statement: "", state: :ready, result: result}} end + defp routed_server_name(%{hostname: host, instance: instance}) + when is_binary(instance) and instance != "", + do: host <> "\\" <> instance + + defp routed_server_name(%{hostname: host}), do: host + + # A pinned SNI names the gateway. The routed server presents its own + # certificate, so the SNI (and the hostname check derived from it) follow + # the route. + defp route_ssl_opts(ssl_opts, host) do + case ssl_opts[:server_name_indication] do + sni when sni in [nil, :disable] -> ssl_opts + _gateway -> Keyword.put(ssl_opts, :server_name_indication, to_charlist(host)) + end + end + defp mark_ready(%{state: _} = s) do %{s | state: :ready} end @@ -854,56 +1134,99 @@ defmodule Tds.Protocol do buffer |> IO.iodata_to_binary() |> decode(s) + else + {:error, reason} -> + {:disconnect, closed_error(reason), s} + + {:max_response_bytes, limit, received} -> + max_response_bytes_exceeded(limit, received, s) + + {:disconnect, _ex, _s} = disconnect -> + disconnect end end + # The rest of the response is still on the wire, so the connection can't be + # reused. Close it and let DBConnection drop it. + defp max_response_bytes_exceeded(limit, received, %{sock: {mod, sock}} = s) do + _ = mod.close(sock) + {:disconnect, Tds.ResponseTooLargeError.exception(limit: limit, received: received), s} + end + defp msg_recv(%{sock: {mod, pid}} = s) do case mod.recv(pid, 0) do {:ok, pkg} -> + received = count_received(0, pkg, s) + pkg |> next_tds_pkg([]) - |> msg_recv(s) + |> msg_recv(received, s) {:error, error} -> - {:disconnect, - %Tds.Error{ - message: "Connection failed to receive packet due #{inspect(error)}" - }, s} + {:disconnect, closed_error(error), s} end catch - {:error, error} -> {:disconnect, error, s} + {:error, error} -> {:disconnect, closed_error(error), s} + {:max_response_bytes, _limit, _received} = exceeded -> exceeded end - defp msg_recv({:done, buffer, _}, _s) do + defp msg_recv({:done, buffer, _}, _received, _s) do Enum.reverse(buffer) end - defp msg_recv({:more, buffer, more, last?}, %{sock: {mod, pid}} = s) do + defp msg_recv({:more, buffer, more, last?}, received, %{sock: {mod, pid}} = s) do take = if last?, do: more, else: 0 case mod.recv(pid, take) do {:ok, pkg} -> + received = count_received(received, pkg, s) + next_tds_pkg(pkg, buffer, more, last?) - |> msg_recv(s) + |> msg_recv(received, s) {:error, error} -> throw({:error, error}) end end - defp msg_recv({:more, buffer, unknown_pkg}, %{sock: {mod, pid}} = s) do + defp msg_recv({:more, buffer, unknown_pkg}, received, %{sock: {mod, pid}} = s) do case mod.recv(pid, 0) do {:ok, pkg} -> + received = count_received(received, pkg, s) + unknown_pkg |> Kernel.<>(pkg) |> next_tds_pkg(buffer) - |> msg_recv(s) + |> msg_recv(received, s) {:error, error} -> throw({:error, error}) end end + # Same exception as the other DBConnection drivers raise for a dead socket, + # so `rescue DBConnection.ConnectionError` keeps catching it + defp closed_error(reason) do + DBConnection.ConnectionError.exception( + "tds connection closed (the pool may have closed it after a timeout): #{inspect(reason)}" + ) + end + + # Bytes read from the socket for the current response, packet headers + # included (decrypted bytes over TLS). Stops reading once + # :max_response_bytes is exceeded. + defp count_received(received, pkg, s) do + received = received + byte_size(pkg) + + case s.max_response_bytes do + limit when is_integer(limit) and received > limit -> + throw({:max_response_bytes, limit, received}) + + _ -> + received + end + end + defp next_tds_pkg(pkg, buffer) do case pkg do <<0x04, 0x01, size::int16(), _::int32(), chunk::binary>> -> @@ -942,7 +1265,11 @@ defmodule Tds.Protocol do end defp clean_opts(opts) do - Keyword.put(opts, :password, :REDACTED) + opts = Keyword.put(opts, :password, :REDACTED) + + if Keyword.has_key?(opts, :access_token), + do: Keyword.put(opts, :access_token, :REDACTED), + else: opts end @spec conn_opts(Keyword.t()) :: list() | no_return @@ -1160,3 +1487,50 @@ defmodule Tds.Protocol do |> Enum.max() end end + +defimpl Inspect, for: Tds.Protocol do + # Protocol state ends up in crash reports and DBConnection errors, keep + # credentials out of it + @secrets [:password, :access_token, :proxy_password] + # ssl_opts: the private key, its password and the same in certs_keys + @ssl_secrets [:password, :key] + + def inspect(s, opts) do + s = %{s | opts: redact(s.opts), access_token: s.access_token && :REDACTED} + Inspect.Any.inspect(s, opts) + end + + defp redact(opts) when is_list(opts) do + Enum.map(opts, fn + {key, _value} when key in @secrets -> {key, :REDACTED} + {:ssl_opts, ssl_opts} -> {:ssl_opts, redact_ssl(ssl_opts)} + other -> other + end) + end + + defp redact(other), do: other + + defp redact_ssl(ssl_opts) when is_list(ssl_opts) do + Enum.map(ssl_opts, fn + {key, _value} when key in @ssl_secrets -> + {key, :REDACTED} + + {:certs_keys, certs_keys} when is_list(certs_keys) -> + {:certs_keys, Enum.map(certs_keys, &redact_cert_key/1)} + + other -> + other + end) + end + + defp redact_ssl(other), do: other + + defp redact_cert_key(%{} = cert_key) do + Map.new(cert_key, fn + {key, _value} when key in @ssl_secrets -> {key, :REDACTED} + other -> other + end) + end + + defp redact_cert_key(other), do: other +end diff --git a/lib/tds/protocol/login7.ex b/lib/tds/protocol/login7.ex index c5dd5f5b..1e0c8824 100644 --- a/lib/tds/protocol/login7.ex +++ b/lib/tds/protocol/login7.ex @@ -22,6 +22,13 @@ defmodule Tds.Protocol.Login7 do @default_app_name "Elixir TDS" # EN-US @language_code_id <<0x09, 0x04, 0x00, 0x00>> + # OptionFlags3 fExtension: ibExtension points to a FeatureExt block + @option_flag_3_extension <<0x10>> + # FeatureExt ids + @feature_fed_auth 0x02 + @feature_terminator 0xFF + # bFedAuthLibrary: Security Token (the client already holds an access token) + @fed_auth_library_security_token 0x01 defstruct [ # Highest TDS version used by the client @@ -58,7 +65,11 @@ defmodule Tds.Protocol.Login7 do # Hostname of the SQL server :hostname, # Database to use (defaults to user database) - :database + :database, + # Federated authentication (nil for SQL Server authentication). When set, + # a map with the access `:token` and the `:echo` flag, which mirrors the + # FEDAUTHREQUIRED option the server sent back in its PRELOGIN response. + :fed_auth ] def new(opts) do @@ -81,9 +92,24 @@ defmodule Tds.Protocol.Login7 do client_language_code_id: @language_code_id, username: opts[:username], password: opts[:password], - servername: opts[:hostname], + servername: opts[:routed_server_name] || opts[:hostname], database: Keyword.get(opts, :database, "") } + |> put_fed_auth(opts[:access_token], opts) + end + + defp put_fed_auth(login, nil, _opts), do: login + + # Federated authentication with a Microsoft Entra ID access token. + # Username and password MUST be empty and fExtension MUST be set. + defp put_fed_auth(login, token, opts) when is_binary(token) do + %{ + login + | username: "", + password: "", + option_flags_3: @option_flag_3_extension, + fed_auth: %{token: token, echo: Keyword.get(opts, :fed_auth_echo, false)} + } end def encode(%__MODULE__{} = login) do @@ -145,8 +171,10 @@ defmodule Tds.Protocol.Login7 do variable_login = variable_login <> servername current_offset = current_offset + byte_size(servername) - # Unused - offsets = offsets <> <<0::ushort(), 0::ushort()>> + # ibExtension/cbExtension is filled in below, once the length of the + # variable data is known + offsets_before_extension = offsets + offsets = <<>> # Client Int Name variable_login = variable_login <> UCS2.from_string(@clt_int_name) @@ -157,13 +185,14 @@ defmodule Tds.Protocol.Login7 do offsets = offsets <> <<0::ushort(), 0::ushort()>> # Database - variable_login = variable_login <> UCS2.from_string(login.database) + database_data = UCS2.from_string(login.database) + variable_login = variable_login <> database_data database = - if login.database == "" do - 0xAC - else - String.length(login.database) + cond do + login.database != "" -> String.length(login.database) + is_nil(login.fed_auth) -> 0xAC + true -> 0 end offsets = offsets <> <> @@ -183,7 +212,28 @@ defmodule Tds.Protocol.Login7 do # SSPI Long offsets = offsets <> <<0::dword()>> - {variable_login, offsets} + current_offset = current_offset + byte_size(database_data) + {extension, extension_data} = encode_extension(login.fed_auth, current_offset) + + {variable_login <> extension_data, offsets_before_extension <> extension <> offsets} + end + + defp encode_extension(nil, _offset), do: {<<0::ushort(), 0::ushort()>>, <<>>} + + # ibExtension points to a DWORD holding the offset of the FeatureExt block, + # which follows it directly. + defp encode_extension(fed_auth, offset) do + feature_ext = encode_feature_ext_fed_auth(fed_auth) <> <<@feature_terminator>> + + {<>, <> <> feature_ext} + end + + defp encode_feature_ext_fed_auth(%{token: token, echo: echo}) do + options = Bitwise.bsl(@fed_auth_library_security_token, 1) + if(echo, do: 1, else: 0) + token = UCS2.from_string(token) + data = <> <> token + + <<@feature_fed_auth, byte_size(data)::ulong()>> <> data end defp encode_tds_password(list) do diff --git a/lib/tds/protocol/prelogin.ex b/lib/tds/protocol/prelogin.ex index b9ea6928..799c02a7 100644 --- a/lib/tds/protocol/prelogin.ex +++ b/lib/tds/protocol/prelogin.ex @@ -38,7 +38,7 @@ defmodule Tds.Protocol.Prelogin do @instopt_token 0x02 @thread_id_token 0x03 @mars_token 0x04 - # @trace_id_token 0x05 + @trace_id_token 0x05 @fed_auth_required_token 0x06 @nonce_opt_token 0x07 @terminator_token 0xFF @@ -62,6 +62,7 @@ defmodule Tds.Protocol.Prelogin do # encode_instance(opts), encode_thread_id(opts), encode_mars(opts), + encode_trace_id(opts), encode_fed_auth_required(opts) ] @@ -156,6 +157,14 @@ defmodule Tds.Protocol.Prelogin do {@mars_token, <<0x00>>} end + # Connection id, then activity id and its sequence number, new for every + # PRELOGIN. A Fabric backend refuses a routed login without them. + defp encode_trace_id(_opts) do + conn_id = :crypto.strong_rand_bytes(16) + activity_id = :crypto.strong_rand_bytes(16) + {@trace_id_token, conn_id <> activity_id <> <<0::little-32>>} + end + defp encode_fed_auth_required(_opts) do {@fed_auth_required_token, <<0x01>>} end @@ -166,14 +175,16 @@ defmodule Tds.Protocol.Prelogin do | {:login, state()} | {:disconnect, Tds.Error.t(), state()} def decode(packet_data, %{opts: opts} = s) do - {:ok, %{encryption: encryption, instance: instance}} = + {:ok, %{encryption: encryption, instance: instance, fed_auth_required: fed_auth_required}} = packet_data |> IO.iodata_to_binary() |> decode_tokens([], s) + s = %{s | fed_auth_echo: fed_auth_required} + case {ssl?(opts), encryption, instance} do {_, _, false} -> - msg = "Connection terminated, connected instance is not '#{instance}'!" + msg = "Connection terminated, the server is not the instance the client asked for" disconnect(msg, s) # Encryption is off. Allowed server response is :off or :not_supported @@ -196,6 +207,11 @@ defmodule Tds.Protocol.Prelogin do {:required, <<@encryption_on>>, _} -> {:encrypt, s} + # The server requires encryption (Azure SQL, Fabric): the whole + # connection is encrypted, as with :on + {client, <<@encryption_required>>, _} when client in [:on, :required] -> + {:encrypt, s} + {_, _, _} -> disconnect("Server does not allow the requested encryption level.", s) end @@ -224,7 +240,7 @@ defmodule Tds.Protocol.Prelogin do tokens, s ) do - tokens = [{:encryption, offset, length} | tokens] + tokens = [{:instance, offset, length} | tokens] decode_tokens(tail, tokens, s) end @@ -272,6 +288,17 @@ defmodule Tds.Protocol.Prelogin do {:ok, decode_data(Enum.reverse(tokens), tail, %__MODULE__{})} end + # Any other option, such as TRACEID echoed back by Azure SQL, is kept only so + # decode_data/3 skips its bytes. Clients must ignore options they don't know. + defp decode_tokens( + <>, + tokens, + s + ) do + tokens = [{{:unknown, token}, offset, length} | tokens] + decode_tokens(tail, tokens, s) + end + defp decode_data([], _, result), do: result defp decode_data([{key, _, length} | tokens], bin, m) do @@ -301,9 +328,15 @@ defmodule Tds.Protocol.Prelogin do %{m | instance: data == <<0x00>>} ) + :fed_auth_required -> + decode_data( + tokens, + tail, + %{m | fed_auth_required: data != <<0x00>>} + ) + # :thread_id -> # :mars -> - # :fed_auth_required -> # :nonce_opt -> _ -> decode_data(tokens, tail, m) diff --git a/lib/tds/response_too_large_error.ex b/lib/tds/response_too_large_error.ex new file mode 100644 index 00000000..b4887fd2 --- /dev/null +++ b/lib/tds/response_too_large_error.ex @@ -0,0 +1,31 @@ +defmodule Tds.ResponseTooLargeError do + @moduledoc """ + Returned when a response grows past the `:max_response_bytes` connection + option. + + * `:limit`: the configured `:max_response_bytes` + * `:received`: bytes read for the response when the driver stopped reading, + packet headers included + + The rest of the response is left unread, so the connection is closed. + """ + + @type t :: %__MODULE__{ + limit: pos_integer(), + received: pos_integer(), + message: String.t() + } + + defexception [:limit, :received, :message] + + @impl true + def exception(opts) do + limit = Keyword.fetch!(opts, :limit) + + %__MODULE__{ + limit: limit, + received: Keyword.fetch!(opts, :received), + message: "response exceeded max_response_bytes (limit #{limit} bytes)" + } + end +end diff --git a/lib/tds/tls.ex b/lib/tds/tls.ex index e2a494dc..cdcdea68 100644 --- a/lib/tds/tls.ex +++ b/lib/tds/tls.ex @@ -95,6 +95,8 @@ defmodule Tds.Tls do # SERVER def init({socket, ssl_opts}) do + # Stop with the socket, whoever closes it + _ = Port.monitor(socket) {:ok, %__MODULE__{socket: socket, ssl_opts: ssl_opts, handshake?: true}} end @@ -210,14 +212,18 @@ defmodule Tds.Tls do {:noreply, s} end + # The transport is linked to the connection process, so it stops :normal: + # a closed socket reaches the owner as a message, never as an exit signal def handle_info({tag, _} = msg, %{owner_pid: pid} = s) when tag in [:tcp_closed, :ssl_closed] do Kernel.send(pid, msg) - {:stop, tag, s} + {:stop, :normal, s} end def handle_info({tag, _, _} = msg, %{owner_pid: pid} = s) when tag in [:tcp_error, :ssl_error] do Kernel.send(pid, msg) - {:stop, tag, s} + {:stop, :normal, s} end + + def handle_info({:DOWN, _ref, :port, _socket, _reason}, s), do: {:stop, :normal, s} end diff --git a/lib/tds/tokens.ex b/lib/tds/tokens.ex index 31e26e23..f6f8a3ed 100644 --- a/lib/tds/tokens.ex +++ b/lib/tds/tokens.ex @@ -27,6 +27,8 @@ defmodule Tds.Tokens do | :doneproc | :envchange | :error + | :featureextack + | :fedauthinfo | :info | :loginack | :order @@ -53,8 +55,8 @@ defmodule Tds.Tokens do 0xFF -> decode_doneinproc(tail, collmetadata) 0xE3 -> decode_envchange(tail, collmetadata) 0xAA -> decode_error(tail, collmetadata) - # 0xAE -> decode_featureextack(tail, collmetadata) - # 0xEE -> decode_fedauthinfo(tail, collmetadata) + 0xAE -> decode_featureextack(tail, collmetadata) + 0xEE -> decode_fedauthinfo(tail, collmetadata) 0xAB -> decode_info(tail, collmetadata) 0xAD -> decode_loginack(tail, collmetadata) 0xD2 -> decode_nbcrow(tail, collmetadata) @@ -486,6 +488,33 @@ defmodule Tds.Tokens do {{:loginack, token}, tail, collmetadata} end + # FEATUREEXTACK: one FeatureAckOpt per feature the server acknowledged, + # each FeatureId, FeatureAckDataLen (DWORD) and FeatureAckData, ending with + # the 0xFF terminator + defp decode_featureextack(bin, collmetadata, acc \\ []) + + defp decode_featureextack(<<0xFF, tail::binary>>, collmetadata, acc) do + {{:featureextack, Enum.reverse(acc)}, tail, collmetadata} + end + + defp decode_featureextack( + <>, + collmetadata, + acc + ) do + decode_featureextack(tail, collmetadata, [{feature_id, data} | acc]) + end + + # FEDAUTHINFO: sent when the client asked the server for the information + # needed to acquire a token (ADAL workflows). The token is passed up front + # here, so the payload is only skipped and reported as unsupported. + defp decode_fedauthinfo( + <>, + collmetadata + ) do + {{:fedauthinfo, data}, tail, collmetadata} + end + defp decode_column_order(tail, n, acc \\ []) defp decode_column_order(tail, n, acc) when n < 1 do diff --git a/test/connection_failure_test.exs b/test/connection_failure_test.exs new file mode 100644 index 00000000..30be6840 --- /dev/null +++ b/test/connection_failure_test.exs @@ -0,0 +1,316 @@ +Code.require_file("support/fake_tds_server.exs", __DIR__) + +defmodule ConnectionFailureTest do + # changes the global log level + use ExUnit.Case, async: false + + import ExUnit.CaptureLog + + alias Tds.Encoding.UCS2 + alias Tds.FakeServer + + @closed "tds connection closed (the pool may have closed it after a timeout): :closed" + + setup do + level = Logger.level() + Logger.configure(level: :debug) + on_exit(fn -> Logger.configure(level: level) end) + end + + defp start_conn(port, opts \\ []) do + Tds.start_link( + [ + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: false, + execution_mode: :executesql, + backoff_type: :stop, + pool_size: 1 + ] ++ opts + ) + end + + test "a socket closed mid-response returns a disconnect error" do + partial = FakeServer.packet(0, <<0::size(100)-unit(8)>>) + {_server, port} = FakeServer.start(script: [{:close, [partial]}]) + {:ok, pid} = start_conn(port) + + log = + capture_log(fn -> + assert {:error, %DBConnection.ConnectionError{message: @closed}} = + Tds.query(pid, "SELECT 1", []) + + # the pool drops the broken connection + Process.sleep(100) + end) + + refute log =~ "bad return value" + end + + test "a server that stops answering times out" do + {_server, port} = FakeServer.start(script: [:hang]) + {:ok, pid} = start_conn(port) + + log = + capture_log(fn -> + assert {:error, error} = Tds.query(pid, "SELECT 1", [], timeout: 200) + # DBConnection adds that the pool closed it + assert %DBConnection.ConnectionError{ + message: @closed <> " (the connection was closed" <> _ + } = + error + + Process.sleep(100) + end) + + assert log =~ "timed out" + refute log =~ "bad return value" + end + + test "a socket closed mid-response raises DBConnection.ConnectionError from query!" do + partial = FakeServer.packet(0, <<0::size(100)-unit(8)>>) + {_server, port} = FakeServer.start(script: [{:close, [partial]}]) + {:ok, pid} = start_conn(port) + + capture_log(fn -> + assert_raise DBConnection.ConnectionError, @closed, fn -> + Tds.query!(pid, "SELECT 1", []) + end + end) + end + + test "a close after the connection SET batch fails the connect with a ConnectionError" do + {_server, port} = FakeServer.start(set_batch: :close) + + opts = [hostname: "127.0.0.1", port: port, username: "u", password: "p", ssl: false] + assert {:error, %DBConnection.ConnectionError{message: @closed}} = Tds.Protocol.connect(opts) + end + + test "a password login needs no FEDAUTH acknowledgement" do + {_server, port} = FakeServer.start(fed_auth_ack: false) + + opts = [hostname: "127.0.0.1", port: port, username: "u", password: "p", ssl: false] + assert {:ok, _state} = Tds.Protocol.connect(opts) + end + + describe "on a socket the client already closed" do + setup do + {_server, port} = FakeServer.start(script: []) + + {:ok, s} = + Tds.Protocol.connect( + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: false + ) + + {:gen_tcp, sock} = s.sock + :ok = :gen_tcp.close(sock) + %{state: s} + end + + test "a send failure returns a ConnectionError disconnect", %{state: s} do + query = %Tds.Query{statement: "SELECT 1"} + + assert {:disconnect, %DBConnection.ConnectionError{message: @closed}, _s} = + Tds.Protocol.handle_execute(query, [], [], s) + end + + test "ping returns a ConnectionError disconnect", %{state: s} do + assert {:disconnect, %DBConnection.ConnectionError{message: @closed}, _s} = + Tds.Protocol.ping(s) + end + end + + describe "with an access token" do + @token "SECRET-TOKEN-123" + + defp token_opts(port) do + [ + hostname: "127.0.0.1", + port: port, + ssl: true, + ssl_opts: [verify: :verify_none], + access_token: fn -> @token end, + execution_mode: :executesql, + backoff_type: :stop, + pool_size: 1 + ] + end + + defp refute_token(text) do + refute text =~ @token + refute text =~ UCS2.from_string(@token) + end + + test "LOGIN7 carries the token over TLS and the state never shows it" do + {_server, port} = + FakeServer.start(tls: true, script: [{:reply, [FakeServer.packet(1, FakeServer.done())]}]) + + log = + capture_log(fn -> + assert {:ok, state} = Tds.Protocol.connect(token_opts(port)) + refute_token(inspect(state)) + assert state.access_token == nil + assert state.opts[:access_token] == :REDACTED + end) + + # FEDAUTH feature extension with the UTF-16LE token, echo clear + assert_received {:login7, login7} + token = UCS2.from_string(@token) + + assert login7 =~ + <<0x02, byte_size(token) + 5::little-32, 0x02, byte_size(token)::little-32>> <> + token <> <<0xFF>> + + refute_token(log) + end + + test "LOGIN7 echoes the server's FEDAUTHREQUIRED" do + for {required, options} <- [{1, 0x03}, {0, 0x02}] do + {_server, port} = FakeServer.start(tls: true, fed_auth_required: required) + + capture_log(fn -> assert {:ok, _state} = Tds.Protocol.connect(token_opts(port)) end) + + assert_received {:login7, login7} + token = UCS2.from_string(@token) + + assert login7 =~ + <<0x02, byte_size(token) + 5::little-32, options, byte_size(token)::little-32>> <> + token + end + end + + test "a login the server did not acknowledge as federated fails" do + {_server, port} = FakeServer.start(tls: true, fed_auth_ack: false) + + capture_log(fn -> + assert {:error, + %Tds.Error{message: "server did not acknowledge federated authentication"}} = + Tds.Protocol.connect(token_opts(port)) + end) + end + + test "a redirect sends the token to the routed server" do + {_target, target_port} = FakeServer.start(tls: true) + {_gateway, gateway_port} = FakeServer.start(tls: true, redirect: target_port) + + log = + capture_log(fn -> + assert {:ok, _state} = Tds.Protocol.connect(token_opts(gateway_port)) + end) + + # one connect, one verify_none warning, although it logs in twice + assert [_] = Regex.scan(~r/verify: :verify_none/, log) + + token = UCS2.from_string(@token) + assert_received {:login7, gateway_login7} + assert_received {:login7, routed_login7} + assert gateway_login7 =~ token + assert routed_login7 =~ token + end + + test "a live connection's state never shows the token or password" do + {_server, port} = FakeServer.start(tls: true) + opts = token_opts(port) ++ [password: "pw-123", connection_listeners: [self()]] + + log = + capture_log(fn -> + {:ok, _pool} = Tds.start_link(opts) + assert_receive {:connected, conn}, 5_000 + + # DBConnection keeps its start options in its own state, only the + # driver's part is checked + {:no_state, %{state: %Tds.Protocol{} = state}} = :sys.get_state(conn) + inspected = inspect(state, limit: :infinity) + assert inspected =~ "%Tds.Protocol{" + refute_token(inspected) + refute inspected =~ "pw-123" + end) + + refute_token(log) + end + + test "a failure mid-login does not show the token" do + {_server, port} = FakeServer.start(tls: true, login: :close) + + log = + capture_log(fn -> + assert {:error, error} = Tds.Protocol.connect(token_opts(port)) + assert %DBConnection.ConnectionError{message: @closed} = error + refute_token(inspect(error)) + end) + + refute_token(log) + end + + test "a socket closed mid-query does not show the token" do + partial = FakeServer.packet(0, <<0::size(100)-unit(8)>>) + {_server, port} = FakeServer.start(tls: true, script: [{:close, [partial]}]) + {:ok, pid} = Tds.start_link(token_opts(port)) + + log = + capture_log(fn -> + assert {:error, %DBConnection.ConnectionError{message: @closed}} = + Tds.query(pid, "SELECT 1", []) + + Process.sleep(100) + end) + + refute log =~ "bad return value" + refute_token(log) + end + + test "a query timeout does not show the token" do + {_server, port} = FakeServer.start(tls: true, script: [:hang]) + {:ok, pid} = Tds.start_link(token_opts(port)) + + log = + capture_log(fn -> + assert {:error, %DBConnection.ConnectionError{}} = + Tds.query(pid, "SELECT 1", [], timeout: 200) + + Process.sleep(100) + end) + + assert log =~ "timed out" + refute log =~ "bad return value" + refute_token(log) + end + + test "inspect never shows credentials" do + state = %Tds.Protocol{ + opts: [ + password: "pw-123", + access_token: @token, + proxy_password: "proxy-123", + ssl_opts: [ + verify: :verify_peer, + password: "keyfile-secret", + key: {:RSAPrivateKey, "der-key-123"}, + certs_keys: [ + %{certfile: "c.pem", key: {:ECPrivateKey, "ec-key-123"}, password: "ck-123"} + ] + ] + ], + access_token: @token + } + + inspected = inspect(state) + refute_token(inspected) + refute inspected =~ "pw-123" + refute inspected =~ "proxy-123" + + for secret <- ["keyfile-secret", "der-key-123", "ec-key-123", "ck-123"], + do: refute(inspected =~ secret) + + assert inspected =~ ":verify_peer" + assert inspected =~ "c.pem" + assert inspected =~ "%Tds.Protocol{" + end + end +end diff --git a/test/fabric_test.exs b/test/fabric_test.exs new file mode 100644 index 00000000..01a30ea0 --- /dev/null +++ b/test/fabric_test.exs @@ -0,0 +1,129 @@ +Code.require_file("support/fake_tds_server.exs", __DIR__) + +defmodule FabricTest do + # Microsoft Fabric warehouses answer PRELOGIN with ENCRYPT_REQ and an INSTOPT + # option, and their gateway routes the login to a `host\instance` backend. + use ExUnit.Case, async: false + + alias Tds.Encoding.UCS2 + alias Tds.FakeServer + alias Tds.Protocol.Prelogin + + @moduletag :capture_log + + # PRELOGIN response captured from a Fabric warehouse: VERSION, ENCRYPTION + # (ENCRYPT_REQ), INSTOPT (0x00, instance matches), MARS and FEDAUTHREQUIRED + @fabric_prelogin Base.decode16!( + "00001A00060100200001020021000104002200010600230001FF0C001788000003000001" + ) + + defp tls_opts(port) do + [ + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: true, + ssl_opts: [verify: :verify_none] + ] + end + + # PRELOGIN options as [{token, data}], terminator excluded + defp prelogin_options(data), do: prelogin_options(data, data) + + defp prelogin_options(<<0xFF, _::binary>>, _data), do: [] + + defp prelogin_options(<>, data), + do: [{token, binary_part(data, offset, length)} | prelogin_options(rest, data)] + + # LOGIN7 ServerName: its offset and length (in characters) follow the fixed + # 36 bytes and the HostName, UserName, Password and AppName pairs + defp login7_server_name(login7) do + <<_::binary-52, offset::little-16, length::little-16, _::binary>> = login7 + login7 |> binary_part(offset, length * 2) |> UCS2.to_string() + end + + describe "PRELOGIN response" do + test "a server that requires encryption gets a TLS connection" do + {_server, port} = FakeServer.start(tls: true, prelogin_reply: @fabric_prelogin) + + assert {:ok, _state} = Tds.Protocol.connect(tls_opts(port)) + end + + test "INSTOPT is read as the instance check, not as the encryption level" do + s = %Tds.Protocol{opts: [ssl: true]} + + # ENCRYPTION ON, then INSTOPT 0x00 (instance matches) + matching = <<0x01, 11::16, 1::16, 0x02, 12::16, 1::16, 0xFF, 0x01, 0x00>> + assert {:encrypt, _} = Prelogin.decode(matching, s) + + # INSTOPT 0x01: the server is not the instance the client asked for + other = <<0x01, 11::16, 1::16, 0x02, 12::16, 1::16, 0xFF, 0x01, 0x01>> + + assert {:disconnect, %Tds.Error{message: "Connection terminated" <> _}, _} = + Prelogin.decode(other, s) + end + + test "a server that requires encryption is refused when the client turned it off" do + s = %Tds.Protocol{opts: [ssl: false]} + + assert {:disconnect, %Tds.Error{message: "Server does not allow" <> _}, _} = + Prelogin.decode(@fabric_prelogin, s) + end + end + + describe "PRELOGIN request" do + test "carries a TRACEID with new connection and activity ids for every connection" do + encode = fn -> + <<_header::binary-8, data::binary>> = + [ssl: true] |> Prelogin.encode() |> IO.iodata_to_binary() + + data |> prelogin_options() |> List.keyfind(0x05, 0) + end + + assert {0x05, <>} = encode.() + assert {0x05, <>} = encode.() + assert conn_id != other_conn_id + assert activity_id != other_activity_id + end + end + + describe "gateway redirect" do + test "the routed LOGIN7 names the routed instance as ServerName" do + {_target, target_port} = FakeServer.start(tls: true) + + {_gateway, gateway_port} = + FakeServer.start(tls: true, redirect: target_port, route_instance: "TRKYbackend") + + assert {:ok, _state} = Tds.Protocol.connect(tls_opts(gateway_port)) + + assert_received {:login7, gateway_login7} + assert_received {:login7, routed_login7} + assert login7_server_name(gateway_login7) == "127.0.0.1" + assert login7_server_name(routed_login7) == "127.0.0.1\\TRKYbackend" + end + + test "a route without an instance keeps the host as ServerName" do + {_target, target_port} = FakeServer.start(tls: true) + {_gateway, gateway_port} = FakeServer.start(tls: true, redirect: target_port) + + assert {:ok, _state} = Tds.Protocol.connect(tls_opts(gateway_port)) + + assert_received {:login7, _gateway_login7} + assert_received {:login7, routed_login7} + assert login7_server_name(routed_login7) == "127.0.0.1" + end + + test "both legs send a TRACEID" do + {_target, target_port} = FakeServer.start(tls: true) + {_gateway, gateway_port} = FakeServer.start(tls: true, redirect: target_port) + + assert {:ok, _state} = Tds.Protocol.connect(tls_opts(gateway_port)) + + for _leg <- 1..2 do + assert_received {:prelogin, prelogin} + assert {0x05, <<_::binary-36>>} = prelogin |> prelogin_options() |> List.keyfind(0x05, 0) + end + end + end +end diff --git a/test/fed_auth_test.exs b/test/fed_auth_test.exs new file mode 100644 index 00000000..3bef46c6 --- /dev/null +++ b/test/fed_auth_test.exs @@ -0,0 +1,270 @@ +defmodule FedAuthTest.LinkedHolder do + # Stands in for a cache or HTTP client the token function starts and links + use GenServer + + def init(_) do + Process.flag(:trap_exit, true) + {:ok, nil} + end +end + +defmodule FedAuthTest do + use ExUnit.Case, async: true + + import Tds.Messages, only: [msg_error: 1, msg_loginack: 1] + + alias Tds.Protocol.Prelogin + + # PRELOGIN response: VERSION, ENCRYPTION (ENCRYPT_ON) and optionally + # FEDAUTHREQUIRED, followed by the terminator and the option data + defp prelogin_response(fed_auth_required, extra_options \\ []) do + fed_auth = if fed_auth_required, do: [{0x06, <>}], else: [] + options = [{0x00, <<15, 0, 0x07, 0xD0, 0, 0>>}, {0x01, <<0x01>>} | fed_auth] ++ extra_options + offset = 5 * length(options) + 1 + + {headers, data, _} = + Enum.reduce(options, {<<>>, <<>>, offset}, fn {token, value}, {headers, data, offset} -> + size = byte_size(value) + {headers <> <>, data <> value, offset + size} + end) + + headers <> <<0xFF>> <> data + end + + defp prelogin_options(<<0xFF, _::binary>>, _data), do: [] + + defp prelogin_options(<>, data), + do: [{token, binary_part(data, offset, length)} | prelogin_options(rest, data)] + + describe "PRELOGIN" do + test "always requests FEDAUTHREQUIRED, with or without an access token" do + # [{token, data}], with the TRACEID ids left out: they are new on every call + encode = fn opts -> + <<_header::binary-8, data::binary>> = opts |> Prelogin.encode() |> IO.iodata_to_binary() + + data + |> prelogin_options(data) + |> Enum.map(fn + {0x05, _ids} -> {0x05, :ids} + option -> option + end) + end + + assert encode.(ssl: true) == encode.(ssl: true, access_token: "tok") + + # VERSION, ENCRYPTION, THREADID, MARS, TRACEID, FEDAUTHREQUIRED + assert [0x00, 0x01, 0x03, 0x04, 0x05, 0x06] = Enum.map(encode.(ssl: true), &elem(&1, 0)) + assert {0x06, <<0x01>>} = List.keyfind(encode.(ssl: true), 0x06, 0) + end + + test "records the server FEDAUTHREQUIRED response as the echo flag" do + s = %Tds.Protocol{opts: [ssl: true]} + + assert {:encrypt, %{fed_auth_echo: true}} = Prelogin.decode(prelogin_response(1), s) + assert {:encrypt, %{fed_auth_echo: false}} = Prelogin.decode(prelogin_response(0), s) + assert {:encrypt, %{fed_auth_echo: false}} = Prelogin.decode(prelogin_response(nil), s) + end + + test "skips a TRACEID the server echoes back and options it doesn't know" do + s = %Tds.Protocol{opts: [ssl: true]} + trace_id = {0x05, :binary.copy(<<0xAB>>, 36)} + unknown = {0x42, <<1, 2, 3>>} + + assert {:encrypt, %{fed_auth_echo: true}} = + Prelogin.decode(prelogin_response(1, [trace_id, unknown]), s) + end + end + + describe "login response tokens" do + # DONE token, final, no row count + @done <<0xFD, 0::16, 0::16, 0::64>> + + test "FEATUREEXTACK is decoded and the token stream continues after it" do + ack = <<0xAE, 0x02, 0::little-32, 0x0A, 1::little-32, 0x01, 0xFF>> + + assert [{:featureextack, [{0x02, ""}, {0x0A, <<0x01>>}]}, {:done, _}] = + Tds.Tokens.decode_tokens(ack <> @done) + end + + test "the login parser records the acknowledged features" do + ack = <<0xAE, 0x02, 0::little-32, 0xFF>> + + assert {msg_loginack(features: [0x02]), %Tds.Protocol{}} = + Tds.Messages.parse(:login, ack <> @done, %Tds.Protocol{}) + end + + test "FEDAUTHINFO fails the login with a clear error" do + info = <<0xEE, 4::little-32, 1, 2, 3, 4>> + + assert {msg_error(error: %{msg_text: "Server requested FEDAUTHINFO" <> _}), _} = + Tds.Messages.parse(:login, info <> @done, %Tds.Protocol{}) + end + end + + describe "connect/1 with :access_token" do + # Nothing listens on port 1, so a resolved token ends in a TCP error + @opts [hostname: "127.0.0.1", port: 1, ssl: true, timeout: 1_000] + + test "refuses to send the token without encryption" do + token = fn -> flunk("resolved") end + without_ssl = [access_token: token] ++ Keyword.delete(@opts, :ssl) + + for opts <- [ + without_ssl | for(ssl <- [false, :not_supported, nil], do: [ssl: ssl] ++ without_ssl) + ] do + assert {:error, %Tds.Error{message: ":access_token requires ssl" <> _}} = + Tds.Protocol.connect(opts) + end + end + + test "resolves a function on every connect" do + parent = self() + token = fn -> send(parent, :token_resolved) && {:ok, "tok"} end + + assert {:error, %Tds.Error{message: "tcp connect: " <> _}} = + Tds.Protocol.connect([access_token: token] ++ @opts) + + assert_received :token_resolved + end + + test "resolves an MFA" do + opts = [access_token: {Function, :identity, ["tok"]}] ++ @opts + + assert {:error, %Tds.Error{message: "tcp connect: " <> _}} = Tds.Protocol.connect(opts) + end + + test "returns an error when the token cannot be fetched" do + opts = [access_token: fn -> {:error, :expired} end] ++ @opts + + assert {:error, %Tds.Error{message: "unable to fetch access token: :expired"}} = + Tds.Protocol.connect(opts) + + assert {:error, %Tds.Error{message: "invalid :access_token" <> _}} = + Tds.Protocol.connect([access_token: ""] ++ @opts) + end + + test "warns once when the server certificate is not verified" do + opts = [access_token: "tok", ssl_opts: [verify: :verify_none]] ++ @opts + log = ExUnit.CaptureLog.capture_log(fn -> Tds.Protocol.connect(opts) end) + + assert [_] = Regex.scan(~r/verify: :verify_none/, log) + refute log =~ "tok\"" + end + + test "does not warn when the server certificate is verified" do + for ssl_opts <- [[verify: :verify_peer], []] do + opts = [access_token: "tok", ssl_opts: ssl_opts] ++ @opts + log = ExUnit.CaptureLog.capture_log(fn -> Tds.Protocol.connect(opts) end) + + refute log =~ "verify_none" + end + end + + test "the token never reaches a process the token function linked to" do + log = + ExUnit.CaptureLog.capture_log(fn -> + token = fn -> + {:ok, _holder} = GenServer.start_link(FedAuthTest.LinkedHolder, nil) + "SECRET-TOKEN-123" + end + + Tds.Protocol.connect([access_token: token] ++ @opts) + Process.sleep(200) + end) + + refute log =~ "SECRET-TOKEN-123" + end + + test "a token function past connect_timeout is killed" do + parent = self() + + token = fn -> + send(parent, {:resolver, self()}) + Process.sleep(:infinity) + end + + assert {:error, %Tds.Error{message: "access token could not be fetched: timed out" <> _}} = + Tds.Protocol.connect([access_token: token, connect_timeout: 50] ++ @opts) + + assert_received {:resolver, resolver} + refute Process.alive?(resolver) + end + + test "the resolver stops when the connecting process dies" do + parent = self() + + token = fn -> + send(parent, {:resolver, self()}) + Process.sleep(:infinity) + end + + conn = + spawn(fn -> + Tds.Protocol.connect([access_token: token, connect_timeout: 60_000] ++ @opts) + end) + + assert_receive {:resolver, resolver}, 1_000 + ref = Process.monitor(resolver) + Process.exit(conn, :kill) + + assert_receive {:DOWN, ^ref, :process, ^resolver, _reason}, 1_000 + end + + test "an {:error, reason} names at most the reason's atom or struct, never its values" do + for {reason, message} <- [ + {%{client_secret: "S3CR3T"}, "unable to fetch access token"}, + {{:http_error, "S3CR3T"}, "unable to fetch access token"}, + {"S3CR3T", "unable to fetch access token"}, + {%RuntimeError{message: "S3CR3T"}, "unable to fetch access token: RuntimeError"}, + {:expired, "unable to fetch access token: :expired"} + ] do + opts = [access_token: fn -> {:error, reason} end] ++ @opts + assert {:error, %Tds.Error{message: ^message}} = Tds.Protocol.connect(opts) + end + end + + test "returns an error naming only the exception when the token function raises" do + opts = [access_token: fn -> raise "SECRET-BOOM" end] ++ @opts + + assert {:error, %Tds.Error{message: message}} = Tds.Protocol.connect(opts) + assert message == "access token could not be fetched: RuntimeError" + end + + test "returns an error when the token function throws or exits" do + for fun <- [fn -> throw(:boom) end, fn -> exit(:boom) end] do + assert {:error, %Tds.Error{message: "access token could not be fetched: " <> kind}} = + Tds.Protocol.connect([access_token: fun] ++ @opts) + + assert kind in ["throw", "exit"] + end + end + + test "gives up on a token function slower than connect_timeout" do + opts = [access_token: fn -> Process.sleep(5_000) end, connect_timeout: 50] ++ @opts + + assert {:error, %Tds.Error{message: message}} = Tds.Protocol.connect(opts) + assert message == "access token could not be fetched: timed out after 50ms" + end + + @tag :capture_log + test "a raising token function leaves the pool alive and backs off" do + parent = self() + + token = fn -> + send(parent, :token_called) + raise "SECRET-BOOM" + end + + {:ok, pool} = + Tds.start_link( + [access_token: token, pool_size: 1, backoff_min: 10, backoff_max: 20] ++ @opts + ) + + for _ <- 1..3, do: assert_receive(:token_called, 1_000) + assert Process.alive?(pool) + + assert {:error, %DBConnection.ConnectionError{}} = + Tds.query(pool, "SELECT 1", [], queue_target: 10, queue_interval: 10, timeout: 50) + end + end +end diff --git a/test/login7_test.exs b/test/login7_test.exs index 9dbd44f4..787e7cf0 100644 --- a/test/login7_test.exs +++ b/test/login7_test.exs @@ -39,4 +39,145 @@ defmodule Login7Test do 97, 0, 115, 0, 101, 0>> ] end + + describe "federated authentication with an access token" do + @fed_auth_login %Tds.Protocol.Login7{ + app_name: "Elixir TDS", + client_language_code_id: <<9, 4, 0, 0>>, + client_pid: <<0, 0, 3, 34>>, + client_time_zone: <<0, 0, 0, 0>>, + client_version: <<4, 0, 0, 7>>, + connection_id: <<0, 0, 0, 0>>, + database: "my_database", + hostname: "test.host.com", + option_flags_1: <<0>>, + option_flags_2: <<0>>, + option_flags_3: <<0x10>>, + packet_size: <<0, 16, 0, 0>>, + password: "", + servername: "some.host.com", + tds_version: <<4, 0, 0, 116>>, + type_flags: <<0>>, + username: "", + fed_auth: %{token: "abc", echo: true} + } + + test "new/1 without :access_token keeps SQL Server authentication" do + login = Login7.new(hostname: "h", username: "u", password: "p", fed_auth_echo: true) + + assert login.fed_auth == nil + assert login.option_flags_3 == <<0>> + assert {login.username, login.password} == {"u", "p"} + end + + test "new/1 with :access_token sets fExtension and blanks the credentials" do + login = + Login7.new( + hostname: "h", + username: "u", + password: "p", + access_token: "tok", + fed_auth_echo: true + ) + + assert login.fed_auth == %{token: "tok", echo: true} + assert login.option_flags_3 == <<0x10>> + assert {login.username, login.password} == {"", ""} + end + + test "encodes the FEDAUTH feature extension after the variable data" do + # ibExtension (196, cbExtension 4) points to a DWORD holding 200, the + # offset of the FeatureExt block: FeatureId 0x02, FeatureDataLen 11, + # options 0x03 (SecurityToken << 1 | fFedAuthEcho), token length 6, + # "abc" as UTF-16LE, terminator 0xFF + assert Login7.encode(@fed_auth_login) == + [ + <<16, 1, 0, 225, 0, 0, 1, 0, 217, 0, 0, 0, 4, 0, 0, 116, 0, 16, 0, 0, 4, 0, 0, 7, + 0, 0, 3, 34, 0, 0, 0, 0, 0, 0, 0, 16, 0, 0, 0, 0, 9, 4, 0, 0, 94, 0, 13, 0, + 120, 0, 0, 0, 120, 0, 0, 0, 120, 0, 10, 0, 140, 0, 13, 0, 196, 0, 4, 0, 166, 0, + 4, 0, 0, 0, 0, 0, 174, 0, 11, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 116, 0, 101, 0, 115, 0, 116, 0, 46, 0, 104, 0, 111, 0, + 115, 0, 116, 0, 46, 0, 99, 0, 111, 0, 109, 0, 69, 0, 108, 0, 105, 0, 120, 0, + 105, 0, 114, 0, 32, 0, 84, 0, 68, 0, 83, 0, 115, 0, 111, 0, 109, 0, 101, 0, 46, + 0, 104, 0, 111, 0, 115, 0, 116, 0, 46, 0, 99, 0, 111, 0, 109, 0, 79, 0, 68, 0, + 66, 0, 67, 0, 109, 0, 121, 0, 95, 0, 100, 0, 97, 0, 116, 0, 97, 0, 98, 0, 97, + 0, 115, 0, 101, 0, 200, 0, 0, 0, 2, 11, 0, 0, 0, 3, 6, 0, 0, 0, 97, 0, 98, 0, + 99, 0, 255>> + ] + end + + test "fFedAuthEcho is clear when the server did not send FEDAUTHREQUIRED" do + [packet] = Login7.encode(%{@fed_auth_login | fed_auth: %{token: "abc", echo: false}}) + + # options byte 0x02: SecurityToken << 1, fFedAuthEcho clear + assert binary_part(packet, byte_size(packet) - 12, 12) == + <<0x02, 6, 0, 0, 0, 97, 0, 98, 0, 99, 0, 255>> + end + + test "FEDAUTH feature matches go-mssqldb's TestSendLoginWithFeatureExt reference" do + # go-mssqldb tds_test.go, token "fedauthtoken" without echo: + # FeatureId 2, FeatureDataLen 29, options 2, token length 24, UTF-16LE token + go_mssqldb_fed_auth = + <<2, 29, 0, 0, 0, 2, 24, 0, 0, 0, 102, 0, 101, 0, 100, 0, 97, 0, 117, 0, 116, 0, 104, 0, + 116, 0, 111, 0, 107, 0, 101, 0, 110, 0>> + + login = %{@fed_auth_login | fed_auth: %{token: "fedauthtoken", echo: false}} + [packet] = Login7.encode(login) + size = byte_size(go_mssqldb_fed_auth) + 1 + + assert binary_part(packet, byte_size(packet) - size, size) == + go_mssqldb_fed_auth <> <<0xFF>> + end + + test "an empty database is sent with a zero length" do + [packet] = Login7.encode(%{@fed_auth_login | database: ""}) + + # ibDatabase / cchDatabase + assert binary_part(packet, 8 + 4 + 32 + 32, 4) == <<174, 0, 0, 0>> + end + + test "a token larger than one packet is split across LOGIN7 packets" do + token = String.duplicate("a", 3000) + packets = Login7.encode(%{@fed_auth_login | fed_auth: %{token: token, echo: false}}) + + assert [<<0x10, 0, _::binary>>, <<0x10, 1, _::binary>>] = + Enum.map(packets, &IO.iodata_to_binary/1) + + payload = + Enum.map_join(packets, fn packet -> + <<_header::binary-8, data::binary>> = IO.iodata_to_binary(packet) + data + end) + + <> = payload + assert length == byte_size(payload) + + assert :binary.part(payload, byte_size(payload) - 6_001, 6_001) == + :binary.copy(<>, 3000) <> <<0xFF>> + end + + test "a 6000 character token spans three packets with EOM only on the last" do + token = String.duplicate("a", 6000) + + packets = + %{@fed_auth_login | fed_auth: %{token: token, echo: false}} + |> Login7.encode() + |> Enum.map(&IO.iodata_to_binary/1) + + headers = + for <> <> _ <- packets, + do: {type, status, length, id} + + # 4088 bytes of data + 8 byte header per full packet, status 0x01 (EOM) last + assert [{0x10, 0, 4096, 1}, {0x10, 0, 4096, 2}, {0x10, 1, last_length, 3}] = headers + assert last_length == byte_size(List.last(packets)) + + payload = Enum.map_join(packets, fn <<_header::binary-8, data::binary>> -> data end) + <> = payload + assert length == byte_size(payload) + + assert :binary.part(payload, byte_size(payload) - 12_001, 12_001) == + :binary.copy(<>, 6000) <> <<0xFF>> + end + end end diff --git a/test/max_response_bytes_test.exs b/test/max_response_bytes_test.exs new file mode 100644 index 00000000..8e2d4cae --- /dev/null +++ b/test/max_response_bytes_test.exs @@ -0,0 +1,134 @@ +Code.require_file("support/fake_tds_server.exs", __DIR__) + +defmodule MaxResponseBytesTest do + use ExUnit.Case, async: true + + alias Tds.FakeServer + + @query %Tds.Query{statement: "SELECT 1"} + + defp connect(script, opts) do + {_server, port} = FakeServer.start(script: script) + + Tds.Protocol.connect( + [hostname: "127.0.0.1", port: port, username: "u", password: "p", ssl: false] ++ opts + ) + end + + defp junk, do: FakeServer.packet(0, <<0::size(4000)-unit(8)>>) + + test "a response under the limit decodes normally" do + <> = FakeServer.done() + script = [{:reply, [FakeServer.packet(0, first), FakeServer.packet(1, rest)]}] + {:ok, s} = connect(script, max_response_bytes: 100) + + assert {:ok, _query, %Tds.Result{num_rows: 5}, _s} = + Tds.Protocol.handle_execute(@query, [], [], s) + end + + test "without the option a large response is read in full" do + {:ok, s} = connect([{:reply, [junk(), junk(), FakeServer.packet(1, FakeServer.done())]}], []) + + # the padding is not a valid token stream, so decoding fails once it + # has all been buffered + assert {:error, %RuntimeError{message: "Unsupported Token code" <> _}, _s} = + Tds.Protocol.handle_execute(@query, [], [], s) + end + + test "a response over the limit stops reading and closes the connection" do + script = [{:reply, [junk(), junk(), junk(), FakeServer.packet(1, FakeServer.done())]}] + {:ok, s} = connect(script, max_response_bytes: 6_000) + + assert {:disconnect, + %Tds.ResponseTooLargeError{ + limit: 6_000, + received: received, + message: "response exceeded max_response_bytes (limit 6000 bytes)" + }, _s} = Tds.Protocol.handle_execute(@query, [], [], s) + + assert received > 6_000 + {:gen_tcp, sock} = s.sock + assert {:error, _closed} = :inet.peername(sock) + assert_receive {:server_done, {:error, :closed}} + end + + test "the limit does not apply to PRELOGIN, LOGIN7 or the connection SET batch" do + # every login response is longer than 10 bytes + {:ok, s} = + connect([{:reply, [FakeServer.packet(1, FakeServer.done())]}], max_response_bytes: 10) + + assert {:disconnect, %Tds.ResponseTooLargeError{limit: 10}, _s} = + Tds.Protocol.handle_execute(@query, [], [], s) + end + + describe "Tds.start_link/1" do + test "rejects anything but a positive integer or nil" do + for value <- [0, -1, 1.5, "10", :infinity] do + assert_raise ArgumentError, ~r/max_response_bytes/, fn -> + Tds.start_link(max_response_bytes: value) + end + end + end + + @tag :capture_log + test "accepts nil" do + {_server, port} = FakeServer.start(script: []) + + assert {:ok, _pool} = + Tds.start_link( + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: false, + max_response_bytes: nil, + backoff_type: :stop, + pool_size: 1 + ) + end + end + + describe "through Tds.query/3" do + defp start_pool(script) do + {_server, port} = FakeServer.start(script: script) + + {:ok, pid} = + Tds.start_link( + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: false, + execution_mode: :executesql, + max_response_bytes: 6_000, + backoff_type: :stop, + pool_size: 1 + ) + + pid + end + + @tag :capture_log + test "a response under the limit returns the result" do + pid = start_pool([{:reply, [FakeServer.packet(1, FakeServer.done())]}]) + + assert {:ok, %Tds.Result{num_rows: 5}} = Tds.query(pid, "SELECT 1", []) + end + + @tag :capture_log + test "a response over the limit returns ResponseTooLargeError and closes the connection" do + pid = + start_pool([{:reply, [junk(), junk(), junk(), FakeServer.packet(1, FakeServer.done())]}]) + + assert {:error, + %Tds.ResponseTooLargeError{ + limit: 6_000, + received: received, + message: "response exceeded max_response_bytes (limit 6000 bytes)" + }} = Tds.query(pid, "SELECT 1", []) + + assert received > 6_000 + assert_receive {:server_done, {:error, :closed}} + end + end +end diff --git a/test/support/fake_tds_server.exs b/test/support/fake_tds_server.exs new file mode 100644 index 00000000..e4595d52 --- /dev/null +++ b/test/support/fake_tds_server.exs @@ -0,0 +1,209 @@ +defmodule Tds.FakeServer do + @moduledoc false + alias Tds.Encoding.UCS2 + + # Just enough of a SQL Server to get a client through PRELOGIN (optionally + # with TLS, wrapped in PRELOGIN packets like the real thing), LOGIN7 and + # the connection SET batch, then hand each query to a script. + # + # `script` is a list of steps run for each client query, in order: + # {:reply, packets} | {:close, packets} | :hang + # and `login: :close` hangs up after reading LOGIN7, `set_batch: :close` + # after reading the connection SET batch. `fed_auth_required: 0 | 1` adds + # FEDAUTHREQUIRED to the PRELOGIN response and `fed_auth_ack: false` leaves + # the FEDAUTH FEATUREEXTACK out of the login response. `prelogin_reply:` + # replaces the whole PRELOGIN response with the given option data. The + # client's PRELOGIN is reported as {:prelogin, data}. `redirect: port` plays + # an Azure gateway: it routes the client to `route_host` (127.0.0.1 by + # default), as `host\instance` when `route_instance` is given, on that + # port and reports whether the client closes the leg (as {:gateway_leg, + # result}), hanging up itself if the client keeps it open. + # + # Pass a list of option lists to serve one connection after another. + + # DONE with the count bit set and 5 rows + def done, do: <<0xFD, 0x10::little-16, 0xC1::little-16, 5::little-64>> + + def packet(status, data) do + <<0x04, status, byte_size(data) + 8::16, 0::16, 1, 0>> <> data + end + + def start(opts) do + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + test = self() + connections = if Keyword.keyword?(opts), do: [opts], else: opts + server = spawn_link(fn -> Enum.each(connections, &serve(listen, test, &1)) end) + {server, port} + end + + defp serve(listen, test, opts) do + {:ok, sock} = :gen_tcp.accept(listen) + tls? = Keyword.get(opts, :tls, false) + + # PRELOGIN: VERSION, ENCRYPTION (ON with TLS, NOT_SUP without) and + # optionally FEDAUTHREQUIRED + send(test, {:prelogin, recv_message({:gen_tcp, sock})}) + encryption = if tls?, do: 0x01, else: 0x02 + + reply = + Keyword.get_lazy(opts, :prelogin_reply, fn -> + prelogin(encryption, opts[:fed_auth_required]) + end) + + :ok = :gen_tcp.send(sock, packet(1, reply)) + + conn = if tls?, do: {:ssl, tls_handshake(sock, test)}, else: {:gen_tcp, sock} + + login7 = recv_message(conn) + send(test, {:login7, login7}) + + login = + if port = opts[:redirect], do: {:redirect, port}, else: Keyword.get(opts, :login, :ok) + + case login do + :close -> + close(conn) + + {:redirect, port} -> + route = Keyword.get(opts, :route_host, "127.0.0.1") + route = if instance = opts[:route_instance], do: route <> "\\" <> instance, else: route + redirect(conn, route, port, test) + + :ok -> + # LOGINACK, FEATUREEXTACK (FEDAUTH, no data) and DONE + featureextack = + if Keyword.get(opts, :fed_auth_ack, true), + do: <<0xAE, 0x02, 0::little-32, 0xFF>>, + else: <<>> + + send_data(conn, packet(1, loginack() <> featureextack <> done())) + + # connection SET statements + _ = recv_message(conn) + + if opts[:set_batch] == :close do + close(conn) + else + send_data(conn, packet(1, done())) + run(conn, Keyword.get(opts, :script, []), test) + end + end + end + + defp prelogin(encryption, nil) do + <<0x00, 11::16, 6::16, 0x01, 17::16, 1::16, 0xFF, 16, 0, 0, 0, 0, 0, encryption>> + end + + defp prelogin(encryption, fed_auth_required) do + <<0x00, 16::16, 6::16, 0x01, 22::16, 1::16, 0x06, 23::16, 1::16, 0xFF, 16, 0, 0, 0, 0, 0, + encryption, fed_auth_required>> + end + + # LOGINACK with a routing ENVCHANGE to host:port + defp redirect(conn, host, port, test) do + host = UCS2.from_string(host) + routing = <<0x00, port::little-16, div(byte_size(host), 2)::little-16>> <> host + env = <<0x14, byte_size(routing)::little-16>> <> routing <> <<0, 0>> + envchange = <<0xE3, byte_size(env)::little-16>> <> env + send_data(conn, packet(1, loginack() <> envchange <> done())) + + send(test, {:gateway_leg, recv(conn, 0, 500)}) + close(conn) + end + + defp loginack do + ack = <<1, 0x74000004::32, 1, ?M, 0, 16, 0, 0, 0>> + <<0xAD, byte_size(ack)::little-16>> <> ack + end + + defp run(conn, [], test), do: send(test, {:server_done, recv(conn, 0, 1_000)}) + + defp run(conn, [step | script], test) do + _ = recv_message(conn) + + case step do + {:reply, packets} -> + Enum.each(packets, &send_data(conn, &1)) + run(conn, script, test) + + {:close, packets} -> + Enum.each(packets, &send_data(conn, &1)) + close(conn) + + :hang -> + send(test, {:server_done, recv(conn, 0, 5_000)}) + end + end + + # The client's Tds.Tls transport wraps the handshake in PRELOGIN packets; + # it is symmetric, so the server side can use it as well + # Reports the client's SNI as {:sni, hostname | :none} + defp tls_handshake(sock, test) do + %{server_config: server_config} = + :public_key.pkix_test_data(%{ + server_chain: %{root: cert_opts(), intermediates: [], peer: cert_opts()}, + client_chain: %{root: cert_opts(), intermediates: [], peer: cert_opts()} + }) + + {:ok, tls} = GenServer.start_link(Tds.Tls, {sock, []}) + :ok = :gen_tcp.controlling_process(sock, tls) + + ssl_opts = + server_config ++ + [ + versions: [:"tlsv1.2"], + active: false, + cb_info: {Tds.FakeServer.Transport, :tcp, :tcp_closed, :tcp_error} + ] + + {:ok, ssl} = :ssl.handshake(sock, ssl_opts, 5_000) + GenServer.cast(tls, :handshake_complete) + + sni = + case :ssl.connection_information(ssl, [:sni_hostname]) do + {:ok, [sni_hostname: host]} -> host + _ -> :none + end + + send(test, {:sni, sni}) + ssl + end + + defp cert_opts, do: [key: {:rsa, 2048, 65_537}, digest: :sha256] + + # Reads one client message, packet by packet, up to the EOM status bit. + # Returns {:error, reason} once the client has hung up. + def recv_message(conn, acc \\ <<>>) do + with {:ok, <<_type, status, length::16, _::binary-4>>} <- recv(conn, 8, 5_000), + {:ok, data} <- recv(conn, length - 8, 5_000) do + if status == 0, do: recv_message(conn, acc <> data), else: acc <> data + end + end + + defp recv({:gen_tcp, sock}, n, timeout), do: :gen_tcp.recv(sock, n, timeout) + defp recv({:ssl, sock}, n, timeout), do: :ssl.recv(sock, n, timeout) + + defp send_data({:gen_tcp, sock}, data), do: _ = :gen_tcp.send(sock, data) + defp send_data({:ssl, sock}, data), do: _ = :ssl.send(sock, data) + + defp close({:gen_tcp, sock}), do: :gen_tcp.close(sock) + defp close({:ssl, sock}), do: :ssl.close(sock) +end + +defmodule Tds.FakeServer.Transport do + @moduledoc false + # Tds.Tls as an :ssl server transport, which also needs port/1 + + defdelegate send(socket, data), to: Tds.Tls + defdelegate recv(socket, length), to: Tds.Tls + defdelegate recv(socket, length, timeout), to: Tds.Tls + defdelegate controlling_process(socket, pid), to: Tds.Tls + defdelegate setopts(socket, opts), to: Tds.Tls + defdelegate getopts(socket, opts), to: :inet + defdelegate peername(socket), to: :inet + defdelegate sockname(socket), to: :inet + defdelegate port(socket), to: :inet + defdelegate close(socket), to: :gen_tcp + defdelegate shutdown(socket, how), to: :gen_tcp +end diff --git a/test/tls_lifecycle_test.exs b/test/tls_lifecycle_test.exs new file mode 100644 index 00000000..0c7814b8 --- /dev/null +++ b/test/tls_lifecycle_test.exs @@ -0,0 +1,167 @@ +Code.require_file("support/fake_tds_server.exs", __DIR__) + +defmodule TlsLifecycleTest do + use ExUnit.Case, async: false + + alias Tds.FakeServer + + @moduletag :capture_log + + defp ok_reply, do: {:reply, [FakeServer.packet(1, FakeServer.done())]} + + defp tls_opts(port) do + [ + hostname: "127.0.0.1", + port: port, + username: "u", + password: "p", + ssl: true, + ssl_opts: [verify: :verify_none], + execution_mode: :executesql, + pool_size: 1 + ] + end + + # The Tds.Tls transport the calling process started for its connection + defp tls_transport do + {:links, links} = Process.info(self(), :links) + + Enum.find(links, fn pid -> + is_pid(pid) and match?({Tds.Tls, :init, _}, :proc_lib.translate_initial_call(pid)) + end) + end + + test "a server close mid-query disconnects and the same connection process reconnects" do + partial = FakeServer.packet(0, <<0::size(100)-unit(8)>>) + + {_server, port} = + FakeServer.start([ + [tls: true, script: [{:close, [partial]}]], + [tls: true, script: [ok_reply()]] + ]) + + opts = tls_opts(port) ++ [connection_listeners: [self()], backoff_min: 10, backoff_max: 50] + {:ok, pool} = Tds.start_link(opts) + assert_receive {:connected, conn}, 5_000 + ref = Process.monitor(conn) + + assert {:error, _} = Tds.query(pool, "SELECT 1", []) + assert_receive {:disconnected, ^conn}, 5_000 + assert_receive {:connected, ^conn}, 5_000 + refute_received {:DOWN, ^ref, _, _, _} + + assert {:ok, %Tds.Result{num_rows: 5}} = Tds.query(pool, "SELECT 1", []) + assert Process.alive?(conn) + end + + test "disconnect stops the TLS transport" do + {_server, port} = FakeServer.start(tls: true) + {:ok, state} = Tds.Protocol.connect(tls_opts(port)) + tls = tls_transport() + assert is_pid(tls) + ref = Process.monitor(tls) + + Tds.Protocol.disconnect(RuntimeError.exception("bye"), state) + assert_receive {:DOWN, ^ref, :process, ^tls, :normal}, 1_000 + end + + test "max_response_bytes closing the connection stops the TLS transport" do + junk = FakeServer.packet(0, <<0::size(4000)-unit(8)>>) + {_server, port} = FakeServer.start(tls: true, script: [{:reply, [junk, junk]}]) + {:ok, state} = Tds.Protocol.connect(tls_opts(port) ++ [max_response_bytes: 6_000]) + tls = tls_transport() + ref = Process.monitor(tls) + + assert {:disconnect, %Tds.ResponseTooLargeError{}, _} = + Tds.Protocol.handle_execute(%Tds.Query{statement: "SELECT 1"}, [], [], state) + + assert_receive {:DOWN, ^ref, :process, ^tls, :normal}, 1_000 + end + + test "a redirect closes the gateway leg and the routed connection outlives it" do + {_target, target_port} = FakeServer.start(tls: true, script: [ok_reply(), ok_reply()]) + {_gateway, gateway_port} = FakeServer.start(tls: true, redirect: target_port) + + opts = tls_opts(gateway_port) ++ [connection_listeners: [self()], backoff_type: :stop] + {:ok, pool} = Tds.start_link(opts) + assert_receive {:connected, conn}, 5_000 + ref = Process.monitor(conn) + + assert {:ok, %Tds.Result{num_rows: 5}} = Tds.query(pool, "SELECT 1", []) + + # the client hangs up on the gateway rather than leaving the leg open + assert_receive {:gateway_leg, {:error, :closed}}, 1_000 + Process.sleep(100) + + assert {:ok, %Tds.Result{num_rows: 5}} = Tds.query(pool, "SELECT 1", []) + refute_received {:disconnected, _} + refute_received {:DOWN, ^ref, _, _, _} + end + + test "a redirect sends the routed host as SNI" do + {_target, target_port} = FakeServer.start(tls: true) + + {_gateway, gateway_port} = + FakeServer.start(tls: true, redirect: target_port, route_host: "localhost") + + ssl_opts = [verify: :verify_none, server_name_indication: ~c"gateway.example"] + opts = Keyword.put(tls_opts(gateway_port), :ssl_opts, ssl_opts) + + assert {:ok, _state} = Tds.Protocol.connect(opts) + assert_received {:sni, ~c"gateway.example"} + assert_received {:sni, ~c"localhost"} + end + + test "failed TLS logins leave no ssl or TLS transport processes behind" do + parent = self() + + client = + spawn(fn -> + for _ <- 1..5 do + {_server, port} = FakeServer.start(tls: true, fed_auth_ack: false) + opts = tls_opts(port) ++ [access_token: "tok"] + {:error, _} = Tds.Protocol.connect(opts) + end + + send(parent, {:done, Process.info(self(), :links)}) + Process.sleep(:infinity) + end) + + assert_receive {:done, {:links, links}}, 10_000 + Process.sleep(200) + + # each ssl connection process monitors its owner + assert {:monitored_by, []} = Process.info(client, :monitored_by) + assert Enum.filter(links, &(is_pid(&1) and Process.alive?(&1))) == [] + Process.exit(client, :kill) + end + + describe "Tds.Tls on a socket event" do + setup do + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false]) + {:ok, port} = :inet.port(listen) + {:ok, sock} = :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false]) + {:ok, tls} = GenServer.start(Tds.Tls, {sock, []}) + :ok = GenServer.call(tls, {:controlling_process, self()}) + %{sock: sock, tls: tls, ref: Process.monitor(tls)} + end + + for event <- [:tcp_closed, :ssl_closed] do + test "#{event} is passed on and stops it normally", %{sock: sock, tls: tls, ref: ref} do + send(tls, {unquote(event), sock}) + + assert_receive {unquote(event), ^sock} + assert_receive {:DOWN, ^ref, :process, ^tls, :normal} + end + end + + for event <- [:tcp_error, :ssl_error] do + test "#{event} is passed on and stops it normally", %{sock: sock, tls: tls, ref: ref} do + send(tls, {unquote(event), sock, :econnreset}) + + assert_receive {unquote(event), ^sock, :econnreset} + assert_receive {:DOWN, ^ref, :process, ^tls, :normal} + end + end + end +end