feat: add lf-endpoint-redact-pii (#2756)

* feat: add lf-endpoint-redact-pii

* chore: version bump

* chore: formatting

* fix: correctly forward :redact_pii option

* chore: formatting

* chore: fix failing tests

* chore: more test fixes

Author: Ziinc

VERSION

@@ -1 +1 @@
-1.21.1
+1.22.0

docs/docs.logflare.com/docs/concepts/endpoints.md

@@ -176,7 +176,7 @@ Endpoints are unsecure by default. However, you can generate [access tokens](/co
 
 Logflare endpoints support automatic redaction of personally identifiable information (PII) from query results to help protect sensitive data. When enabled, the PII redaction feature will automatically replace IP addresses in query result values with "REDACTED".
 
-PII redaction can be enabled when checking the "Redact PII from query results" checkbox when configuring the endpoint.
+PII redaction can be enabled when checking the "Redact PII from query results" checkbox when configuring the endpoint. Override per request with `LF-ENDPOINT-REDACT-PII: true|false`; if omitted, the endpoint setting is used.
 
 Currently, PII redaction targets:
 

lib/logflare/endpoints.ex

@@ -344,14 +344,15 @@ defmodule Logflare.Endpoints do
   Runs a cached query.
   """
   @spec run_cached_query(query :: Query.t(), params :: map()) :: run_query_return()
-  def run_cached_query(%Query{} = query, params \\ %{}) when is_map(params) do
+  def run_cached_query(%Query{} = query, params \\ %{}, opts \\ [])
+      when is_map(params) and is_list(opts) do
     if query.cache_duration_seconds > 0 do
       query
-      |> Resolver.resolve(params)
+      |> Resolver.resolve(params, opts)
       |> ResultsCache.query()
     else
       # execute the query directly
-      run_query(query, params)
+      run_query(query, params, opts)
     end
   end
 
@@ -422,14 +423,16 @@ defmodule Logflare.Endpoints do
           {final_query, declared_params, input_params, endpoint_query}
         end
 
+      redact_pii = Keyword.get(opts, :redact_pii, endpoint_query.redact_pii)
+
       case adaptor.execute_query(backend, query_args, opts) do
         {:ok, rows} when is_list(rows) ->
-          redacted_rows = PiiRedactor.redact_query_result(rows, endpoint_query.redact_pii)
+          redacted_rows = PiiRedactor.redact_query_result(rows, redact_pii)
           {:ok, %{rows: redacted_rows}}
 
         {:ok, %{rows: rows} = result} ->
           # Pass through the full result map with all metadata, but redact PII in rows
-          redacted_rows = PiiRedactor.redact_query_result(rows, endpoint_query.redact_pii)
+          redacted_rows = PiiRedactor.redact_query_result(rows, redact_pii)
           {:ok, %{result | rows: redacted_rows}}
 
         {:error, error} ->

lib/logflare/endpoints/resolver.ex

@@ -21,7 +21,7 @@ defmodule Logflare.Endpoints.Resolver do
   Starts up or performs a lookup for an Endpoint.Cache process.
   Returns the resolved pid.
   """
-  def resolve(%Logflare.Endpoints.Query{id: id} = query, params) do
+  def resolve(%Logflare.Endpoints.Query{id: id} = query, params, opts) do
     attributes = %{
       "endpoint.id" => id,
       "endpoint.token" => query.token,
@@ -40,12 +40,12 @@ defmodule Logflare.Endpoints.Resolver do
         OpenTelemetry.Tracer.with_span "logflare.endpoints.results_cache.create", %{
           attributes: attributes
         } do
-          spec = {ResultsCache, {query, params}}
+          spec = {ResultsCache, {query, params, opts}}
           Logger.debug("Starting up Endpoint.Cache for Endpoint.Query id=#{id}", endpoint_id: id)
 
           via =
             {:via, PartitionSupervisor,
-             {Logflare.Endpoints.ResultsCache.PartitionSupervisor, {id, params}}}
+             {Logflare.Endpoints.ResultsCache.PartitionSupervisor, {id, params, opts}}}
 
           case DynamicSupervisor.start_child(via, spec) do
             {:ok, pid} ->

lib/logflare/endpoints/results_cache.ex

@@ -12,6 +12,7 @@ defmodule Logflare.Endpoints.ResultsCache do
 
   defstruct query_tasks: [],
             params: %{},
+            opts: [],
             cached_result: nil,
             shutdown_timer: nil,
             refresh_timer: nil,
@@ -24,15 +25,16 @@ defmodule Logflare.Endpoints.ResultsCache do
           endpoint_query_token: String.t(),
           query_tasks: list(%Task{}),
           params: map(),
+          opts: list(),
           cached_result: binary(),
           shutdown_timer: reference(),
           refresh_timer: reference(),
           parsed_labels: map()
         }
 
-  def start_link({query, params}) do
+  def start_link({query, params, opts}) do
     name = name(query.id, params)
-    GenServer.start_link(__MODULE__, {query, params}, name: name, hibernate_after: 5_000)
+    GenServer.start_link(__MODULE__, {query, params, opts}, name: name, hibernate_after: 5_000)
   end
 
   @doc """
@@ -86,7 +88,7 @@ defmodule Logflare.Endpoints.ResultsCache do
     GenServer.call(cache, :invalidate)
   end
 
-  def init({query, params}) do
+  def init({query, params, opts}) do
     endpoints = endpoints_part(query.id)
     :syn.join(endpoints, query.id, self())
 
@@ -97,6 +99,7 @@ defmodule Logflare.Endpoints.ResultsCache do
         endpoint_query_id: query.id,
         endpoint_query_token: query.token,
         params: params,
+        opts: opts,
         shutdown_timer: timer,
         parsed_labels: query.parsed_labels
       }
@@ -188,7 +191,7 @@ defmodule Logflare.Endpoints.ResultsCache do
       Endpoints.Cache.get_mapped_query_by_token(state.endpoint_query_token)
       |> Map.put(:parsed_labels, state.parsed_labels)
 
-    Logflare.Endpoints.run_query(query, state.params)
+    Logflare.Endpoints.run_query(query, state.params, state.opts)
     |> Tuple.append(query)
   end
 

lib/logflare_web/controllers/endpoints_controller.ex

@@ -28,7 +28,8 @@ defmodule LogflareWeb.EndpointsController do
       "Content-Length",
       "X-Requested-With",
       "X-API-Key",
-      "LF-ENDPOINT-LABELS"
+      "LF-ENDPOINT-LABELS",
+      "LF-ENDPOINT-REDACT-PII"
     ],
     methods: ["GET", "POST", "OPTIONS"],
     send_preflight_response?: true
@@ -66,8 +67,16 @@ defmodule LogflareWeb.EndpointsController do
       end
 
     parsed_labels = Endpoints.parse_labels(endpoint_query.labels, header_str, params)
+    override = get_req_header(conn, "lf-endpoint-redact-pii") |> List.first()
 
-    case Endpoints.run_cached_query(%{endpoint_query | parsed_labels: parsed_labels}, params) do
+    redact_pii =
+      if override, do: String.downcase(override) == "true", else: endpoint_query.redact_pii
+
+    case Endpoints.run_cached_query(
+           %{endpoint_query | parsed_labels: parsed_labels},
+           params,
+           redact_pii: redact_pii
+         ) do
       {:ok, result} ->
         Logger.debug("Endpoint cache result, #{inspect(result, pretty: true)}")
         render(conn, "query.json", result: result.rows)

lib/logflare_web/live/endpoints/actions/show.html.heex

@@ -101,5 +101,14 @@
       -H 'Content-Type: application/json; charset=utf-8' <span :if={length(@declared_params) > 0}>\</span>
       <span :if={length(@declared_params) > 0}>-G <%= Enum.map(@declared_params, fn p -> "-d \"#{p}=VALUE\"" end) |> Enum.join(" ") %></span>
     </code>
+
+    <code class="tw-whitespace-pre-line">
+      # With per-request PII redaction
+      curl "<%= "https://api.logflare.app" <> ~p"/api/endpoints/query/#{@show_endpoint.token}" %>" \
+      -H 'X-API-KEY: YOUR-ACCESS-TOKEN' \
+      -H 'LF-ENDPOINT-REDACT-PII: true' \
+      -H 'Content-Type: application/json; charset=utf-8' <span :if={length(@declared_params) > 0}>\</span>
+      <span :if={length(@declared_params) > 0}>-G <%= Enum.map(@declared_params, fn p -> "-d \"#{p}=VALUE\"" end) |> Enum.join(" ") %></span>
+    </code>
   </div>
 </section>

test/logflare/endpoints/cache_test.exs

@@ -38,7 +38,7 @@ defmodule Logflare.Endpoints.CacheTest do
       end)
 
       # Start cache process
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should hit backend
@@ -54,7 +54,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:error, :timeout}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       assert {:error, %{"message" => :timeout}} = Endpoints.run_cached_query(endpoint)
@@ -70,7 +70,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response(test_response)}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should succeed
@@ -94,7 +94,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:error, TestUtils.gen_bq_error("BQ Error")}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       assert {:error, %{"message" => "BQ Error"}} = Endpoints.run_cached_query(endpoint)
@@ -113,7 +113,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response(test_response)}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should succeed
@@ -136,7 +136,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response(test_response)}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should succeed
@@ -161,7 +161,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response(test_response)}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should return first test response
@@ -202,7 +202,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response()}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
       assert {:ok, %{rows: [_]}} = Endpoints.run_cached_query(endpoint)
 
@@ -222,7 +222,7 @@ defmodule Logflare.Endpoints.CacheTest do
         {:ok, TestUtils.gen_bq_response(test_response)}
       end)
 
-      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      {:ok, cache_pid} = start_supervised({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert Process.alive?(cache_pid)
 
       # First query should succeed

test/logflare/endpoints_test.exs

@@ -314,7 +314,7 @@ defmodule Logflare.EndpointsTest do
           cache_duration_seconds: 4
         )
 
-      _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert {:ok, %{rows: [%{"testing" => _}]}} = Endpoints.run_cached_query(endpoint)
       # 2nd query should hit local cache
       assert {:ok, %{rows: [%{"testing" => _}]}} = Endpoints.run_cached_query(endpoint)
@@ -334,7 +334,7 @@ defmodule Logflare.EndpointsTest do
           cache_duration_seconds: 1
         )
 
-      _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+      _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
       assert {:ok, %{rows: [%{"testing" => _}]}} = Endpoints.run_cached_query(endpoint)
       # 2nd query should hit local cache
       assert {:ok, %{rows: [%{"testing" => _}]}} = Endpoints.run_cached_query(endpoint)
@@ -375,7 +375,7 @@ defmodule Logflare.EndpointsTest do
 
         user = insert(:user)
         endpoint = insert(:endpoint, user: user, query: "select current_datetime() as testing")
-        cache_pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+        cache_pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
         assert {:ok, %{rows: [%{"testing" => _}]}} = Endpoints.run_cached_query(endpoint)
 
         params =
@@ -454,7 +454,7 @@ defmodule Logflare.EndpointsTest do
              }
            } = Endpoints.calculate_endpoint_metrics(endpoint)
 
-    _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}}})
+    _pid = start_supervised!({Logflare.Endpoints.ResultsCache, {endpoint, %{}, []}})
 
     assert %_{
              metrics: %Query.Metrics{

test/logflare_web/controllers/endpoints_controller_test.exs

@@ -576,6 +576,40 @@ defmodule LogflareWeb.EndpointsControllerTest do
       refute conn.halted
     end
 
+    test "redacts IP addresses when header is set", %{conn: init_conn, user: user} do
+      endpoint = insert(:endpoint, user: user, enable_auth: false, redact_pii: false)
+
+      GoogleApi.BigQuery.V2.Api.Jobs
+      |> expect(:bigquery_jobs_query, fn _conn, _proj_id, _opts ->
+        bq_response =
+          TestUtils.gen_bq_response([
+            %{"ip_address" => "192.168.1.1", "event_message" => "User 10.0.0.1 connected ::1"}
+          ])
+
+        {:ok, bq_response}
+      end)
+
+      conn =
+        init_conn
+        |> put_req_header("lf-endpoint-redact-pii", "true")
+        |> get(~p"/endpoints/query/#{endpoint.token}")
+
+      response =
+        conn
+        |> json_response(200)
+        |> assert_schema("EndpointQuery")
+
+      assert [
+               %{
+                 "ip_address" => "REDACTED",
+                 "event_message" => "User REDACTED connected REDACTED"
+               }
+             ] = response.result
+
+      refute response.error
+      refute conn.halted
+    end
+
     test "does not redact IP addresses when redact_pii is disabled", %{
       conn: init_conn,
       user: user