Quick Code Sample on Encrypting PII with Commanded for GDPR/CCPA Compliance

1 - PII Encryption with Elixir, Commanded, Vault.md

This code is extracted from one of my private projects as an example of how to implement encryption of PII in event streams using two keys: a master key for each "data subject" that is stored in Vault and never transported to the systems that process the PII, and a key unique to each event that is stored (itself encrypted) with the event.

To be clear, the key that is stored with the data is encrypted by another key that is not stored with the data. The idea is that each "data subject" has an encryption key that is stored in Vault (external). When you encrypt data, the library will:

1. create a new AES 256 encryption key
2. use that key to encrypt the data for that event
3. encrypt that AES key itself using a Vault transit endpoint
4. store the encrypted AES key on the event It works that way for a few reasons:
5. it provides implicit key rotation for newly encrypted data
6. it's far more efficient to encrypt the larger data with the local AES 256 key and then just send that smaller key to Vault for encryption, and it means not having to send the PII itself to another system for encryption
7. one of the keys needed for decryption is never present on the machine used to encrypt/decrypt the actual PII datah

LICENSE

Copyright 2020 John Wilger [email protected]

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

email_change_confirmed.ex

defmodule TeamWebsite.Users.Events.EmailChangeConfirmed do
  @moduledoc """
  Emitted after a user confirms an email address change.
  """

  use TypedStruct

  @schema_version 1

  @derive Jason.Encoder
  typedstruct do
    field :schema_version, non_neg_integer(), default: @schema_version

    field :encrypted, boolean(), default: false
    field :encryption_key, binary()

    field :user_id, binary(), enforce: true
    field :team_id, binary(), enforce: true
    field :email, binary() | String.t(), enforce: true
  end

  use TeamWebsite.PIIEncryption.Encryptable

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encrypted?(data), do: data.encrypted

  @impl TeamWebsite.PIIEncryption.Encryptable
  def data_subject_id(data), do: data.user_id

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encryption_key(data), do: data.encryption_key

  @impl TeamWebsite.PIIEncryption.Encryptable
  def get_encrypted_field_map(data) do
    data
    |> Map.from_struct()
    |> Map.take([:email])
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_encrypted_fields(data, encrypted_data, encrypted_key) do
    data
    |> Map.from_struct()
    |> Map.merge(encrypted_data)
    |> Map.put(:encrypted, true)
    |> Map.put(:encryption_key, encrypted_key)
    |> (&struct!(__MODULE__, &1)).()
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_decrypted_fields(data, decrypted_data) do
    data
    |> Map.from_struct()
    |> Map.merge(decrypted_data)
    |> Map.put(:encrypted, false)
    |> Map.put(:encryption_key, nil)
    |> (&struct!(__MODULE__, &1)).()
  end

  defimpl Commanded.Event.Upcaster, for: __MODULE__ do
    def upcast(event, _metadata) do
      {:ok, event} = TeamWebsite.PIIEncryption.decrypt(event)
      event
    end
  end
end

email_changed.ex

defmodule TeamWebsite.Users.Events.EmailChanged do
  @moduledoc """
  Emitted when a User's email address changes.
  """
  use TypedStruct

  @schema_version 1

  @derive Jason.Encoder
  typedstruct do
    field :schema_version, non_neg_integer(), default: @schema_version

    field :encrypted, boolean(), default: false
    field :encryption_key, binary()

    field :user_id, binary(), enforce: true
    field :team_id, binary(), enforce: true
    field :confirmation_code, String.t(), enforce: true
    field :email, binary(), enforce: true
  end

  use TeamWebsite.PIIEncryption.Encryptable

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encrypted?(data), do: data.encrypted

  @impl TeamWebsite.PIIEncryption.Encryptable
  def data_subject_id(data), do: data.user_id

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encryption_key(data), do: data.encryption_key

  @impl TeamWebsite.PIIEncryption.Encryptable
  def get_encrypted_field_map(data) do
    data
    |> Map.from_struct()
    |> Map.take([:email])
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_encrypted_fields(data, encrypted_data, encrypted_key) do
    data
    |> Map.from_struct()
    |> Map.merge(encrypted_data)
    |> Map.put(:encrypted, true)
    |> Map.put(:encryption_key, encrypted_key)
    |> (&struct!(__MODULE__, &1)).()
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_decrypted_fields(data, decrypted_data) do
    data
    |> Map.from_struct()
    |> Map.merge(decrypted_data)
    |> Map.put(:encrypted, false)
    |> Map.put(:encryption_key, nil)
    |> (&struct!(__MODULE__, &1)).()
  end

  defimpl Commanded.Event.Upcaster, for: __MODULE__ do
    def upcast(event, _metadata) do
      {:ok, event} = TeamWebsite.PIIEncryption.decrypt(event)
      event
    end
  end
end

encryptable.ex

defmodule TeamWebsite.PIIEncryption.Encryptable do
  @moduledoc """
  Enables PII encryption for data structures

  Add `use TeamWebsite.PIIEncryption.Encryptable` to your module and implement the necessary
  callbacks to ensure that any PII in your data structure is properly encrypted. When you `use`
  this module, it will create the necessary implementation of the `TeamWebsite.PIIEncryption`
  protocol for the your module.

  This is intended in particular to be used with Event modules, so that PII is always stored in
  an encrypted format in within the event store. For example, to encrypt a `UserAddressChanged`
  event that includes a mix of PII and non-PII data, you might define the event as follows:

      defmodule UserAddressChanged do
        use TypedStruct
        use TeamWebsite.PIIEncryption.Encryptable

        @derive Jason.Encoder
        typedstruct do
          # The same struct is used to represent both encrypted and unencrypted data. When
          # encrypted, we set `encrypted` to true and set the value of the encryption key. This
          # is done for you assuming you have implemented the required callbacks.
          field :encrypted, boolean(), default: false
          field :encryption_key, binary()

          # These fields do not contain PII and do not need to be encrypted.
          field :user_id, binary()
          field :effective_date, Date.t()

          # These fields do contain PII and require encryption.
          field :street_1, String.t() | binary()
          field :street_2, String.t() | binary()
          field :city, String.t() | binary()
          field :state, String.t() | binary()
          field :zipcode, String.t() | binary()
        end

        @impl TeamWebsite.PIIEncryption.Encryptable
        def encrypted?(data), do: data.encrypted

        @impl TeamWebsite.PIIEncryption.Encryptable
        def data_subject_id(data), do: data.id

        @impl TeamWebsite.PIIEncryption.Encryptable
        def encryption_key(data), do: data.owner_encryption_key

        @impl TeamWebsite.PIIEncryption.Encryptable
        def get_encrypted_field_map(data) do
          data
          |> Map.from_struct()
          |> Map.take([:street_1, :street_2, :city, :state, :zipcode])
        end

        @impl TeamWebsite.PIIEncryption.Encryptable
        def apply_encrypted_fields(data, encrypted_data, encrypted_key) do
          data
          |> Map.from_struct()
          |> Map.merge(encrypted_data)
          |> Map.put(:encrypted, true)
          |> Map.put(:owner_encryption_key, encrypted_key)
          |> (&struct!(__MODULE__, &1)).()
        end

        @impl TeamWebsite.PIIEncryption.Encryptable
        def apply_decrypted_fields(data, decrypted_data) do
          data
          |> Map.from_struct()
          |> Map.merge(decrypted_data)
          |> Map.put(:encrypted, false)
          |> Map.put(:owner_encryption_key, nil)
          |> (&struct!(__MODULE__, &1)).()
        end

        # For events, we need to implement a bit of a hack using the `Commanded.Event.Upcaster`
        # protocol in order to allow event consumers to read the event data without having to
        # concern themselves with trying to decrypt it. (Encryption takes place when the event
        # stored by the event store.)
        defimpl Commanded.Event.Upcaster, for: __MODULE__ do
          def upcast(event, _metadata) do
            {:ok, event} = TeamWebsite.PIIEncryption.decrypt(event)
            event
          end
        end
      end
  """
  defmacro __using__(_) do
    quote do
      @behaviour TeamWebsite.PIIEncryption.Encryptable

      defimpl TeamWebsite.PIIEncryption, for: __MODULE__ do
        alias TeamWebsite.PIIEncryption

        def encrypt(data, _key) do
          if encrypted?(data) do
            {:ok, data, nil}
          else
            {:ok, encrypted_data, {_, encrypted_key}} =
              PIIEncryption.encrypt(encryptable_data(data), build_key(data))

            apply_encrypted_fields(data, encrypted_data, encrypted_key)
          end
        end

        def decrypt(data, _key) do
          if encrypted?(data) do
            {:ok, decrypted_data} = PIIEncryption.decrypt(encryptable_data(data), build_key(data))
            apply_decrypted_fields(data, decrypted_data)
          else
            {:ok, data}
          end
        end

        defp encrypted?(data) do
          apply(data.__struct__, :encrypted?, [data])
        end

        defp encryptable_data(data) do
          apply(data.__struct__, :get_encrypted_field_map, [data])
        end

        defp build_key(data) do
          [data_subject_id(data), encryption_key(data)]
          |> Enum.filter(fn x -> !is_nil(x) end)
          |> List.to_tuple()
        end

        defp data_subject_id(data) do
          apply(data.__struct__, :data_subject_id, [data])
        end

        defp encryption_key(data) do
          apply(data.__struct__, :encryption_key, [data])
        end

        defp apply_encrypted_fields(data, encrypted_data, encrypted_key) do
          result =
            apply(data.__struct__, :apply_encrypted_fields, [
              data,
              encrypted_data,
              encrypted_key
            ])

          {:ok, result, nil}
        end

        defp apply_decrypted_fields(data, decrypted_data) do
          result = apply(data.__struct__, :apply_decrypted_fields, [data, decrypted_data])
          {:ok, result}
        end
      end
    end
  end

  @doc """
  Callback that tells the protocol how to identify the data subject. A "data subject" the person
  or organization that owns the data and to whom rights over the data are granted. In this case,
  the data subject is the User who is telling us what their address is.
  """
  @callback data_subject_id(data :: struct()) :: binary()

  @doc """
  Callback that tells the protocol what encryption key was used if the event is currently
  encrypted.
  """
  @callback encryption_key(data :: struct()) :: binary() | nil

  @doc """
  Callback that tells the protocol how to determine if the struct is already encrypted or not
  """
  @callback encrypted?(data :: struct()) :: boolean()

  @doc """
  Callback that returns a map of the encryptable data and its values. The values will either be
  the encrypted or the unencrypted data depending on the current state of the event.
  """
  @callback get_encrypted_field_map(data :: struct()) :: map()

  @doc """
  Callback that tells the protocol how to apply the new, encrypted values as well as the
  encryption key and mark the event as having been encrypted.
  """
  @callback apply_encrypted_fields(
              data :: struct(),
              encrypted_fields :: map(),
              encrypted_key :: binary()
            ) :: data :: struct()

  @doc """
  Callback that tells the protocol how to apply decrypted values and mark the event as *not* being
  encrypted.
  """
  @callback apply_decrypted_fields(
              data :: struct(),
              decrypted_fields :: map()
            ) :: data :: struct()
end

event_encrypting_adapter.ex

defmodule TeamWebsite.PIIEncryption.EventEncryptingAdapter do
  @moduledoc """
  Wraps the actual event store adapter to ensure that events/snapshots are encrypted prior to
  storage.

  Aggregates, Events, and Process Managers may all provide an implementation for the `TeamWebsite.PIIEncryption.Encryptable`
  protocol. If they do, then this module will ensure that their data is encrypted prior to actually
  storing the data in the event store.

  Events must *additionally* provide an implementation of `Commanded.Event.Upcaster` in order to
  decrypt the event for any subscribers. The minimum such upcaster can be defined inside the event
  module as:

      defimpl Commanded.Event.Upcaster, for: __MODULE__ do
        def upcast(event, _metadata) do
          {:ok, event} = TeamWebsite.PIIEncryption.decrypt(event)
          event
        end
      end

  This is not automatically added by the `TeamWebsite.PIIEncryption.Encryptable` module, because:

    * Not everything that uses that protocol is an event that needs an upcaster, and
    * Only one upcaster implementation can be defined for an event module, and we don't want to 
      conflict with any upcaster that is defined to handle event versioning.
  """

  @behaviour Commanded.EventStore.Adapter

  alias Commanded.EventStore.Adapters.Extreme, as: EventStore
  alias TeamWebsite.PIIEncryption

  @impl Commanded.EventStore.Adapter
  def append_to_stream(adapter_meta, stream_uuid, expected_version, events) do
    events =
      Enum.map(events, fn event_data ->
        Map.put(event_data, :data, encrypt(event_data.data))
      end)

    EventStore.append_to_stream(adapter_meta, stream_uuid, expected_version, events)
  end

  @impl Commanded.EventStore.Adapter
  def read_snapshot(adapter_meta, source_uuid) do
    with {:ok, snapshot_data} <- EventStore.read_snapshot(adapter_meta, source_uuid) do
      snapshot_data = decrypt(snapshot_data)
      {:ok, snapshot_data}
    end
  end

  @impl Commanded.EventStore.Adapter
  def record_snapshot(adapter_meta, snapshot) do
    snapshot = Map.put(snapshot, :data, encrypt(snapshot.data))
    EventStore.record_snapshot(adapter_meta, snapshot)
  end

  defp encrypt(data) do
    {:ok, data, _} = PIIEncryption.encrypt(data)
    data
  end

  defp decrypt(data) do
    {:ok, data} = PIIEncryption.decrypt(data)
    data
  end

  ################################################################################
  # Functions delegated directly to `Commanded.EventStore.Adapters.Extreme`
  ################################################################################

  # N.B. Although we could hook into `stream_forward/4` to decrypt events, there is no easy way to
  # do the same for events that are sent to subscribers (`subscribe_to/5`), which are how most
  # things will read the events in practice. Instead we are forced to (ab)use the
  # `Commanded.Event.Upcaster` protocol for each encryptable event module.
  defdelegate stream_forward(
                adapter_meta,
                stream_uuid,
                start_version \\ 0,
                read_batch_size \\ 1_000
              ),
              to: EventStore

  defdelegate delete_snapshot(adapter_meta, source_uuid), to: EventStore
  defdelegate child_spec(application, config), to: EventStore
  defdelegate subscribe(adapter_meta, stream_uuid), to: EventStore

  defdelegate subscribe_to(adapter_meta, stream_uuid, subscription_name, subscriber, start_from),
    to: EventStore

  defdelegate ack_event(adapter_meta, subscription, event), to: EventStore
  defdelegate unsubscribe(adapter_meta, subscription), to: EventStore
  defdelegate delete_subscription(adapter_meta, stream_uuid, subscription_name), to: EventStore
end

name_changed.ex

defmodule TeamWebsite.Users.Events.NameChanged do
  @moduledoc """
  The first and/or last name of a User has been changed.
  """

  @schema_version 1

  use TypedStruct

  @derive Jason.Encoder
  typedstruct do
    field :schema_version, non_neg_integer(), default: @schema_version

    field :encrypted, boolean(), default: false
    field :encryption_key, binary()

    field :id, binary(), enforce: true
    field :team_id, binary(), enforce: true
    field :first_name, binary(), enforce: true
    field :last_name, binary(), enforce: true
  end

  use TeamWebsite.PIIEncryption.Encryptable

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encrypted?(data), do: data.encrypted

  @impl TeamWebsite.PIIEncryption.Encryptable
  def data_subject_id(data), do: data.id

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encryption_key(data), do: data.encryption_key

  @impl TeamWebsite.PIIEncryption.Encryptable
  def get_encrypted_field_map(data) do
    data
    |> Map.from_struct()
    |> Map.take([:first_name, :last_name])
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_encrypted_fields(data, encrypted_data, encrypted_key) do
    data
    |> Map.from_struct()
    |> Map.merge(encrypted_data)
    |> Map.put(:encrypted, true)
    |> Map.put(:encryption_key, encrypted_key)
    |> (&struct!(__MODULE__, &1)).()
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_decrypted_fields(data, decrypted_data) do
    data
    |> Map.from_struct()
    |> Map.merge(decrypted_data)
    |> Map.put(:encrypted, false)
    |> Map.put(:encryption_key, nil)
    |> (&struct!(__MODULE__, &1)).()
  end

  defimpl Commanded.Event.Upcaster, for: __MODULE__ do
    def upcast(event, _metadata) do
      {:ok, event} = TeamWebsite.PIIEncryption.decrypt(event)
      event
    end
  end
end

pii_encryption.ex

# TODO: Rework this to not rely on Hashicorp Vault. We can likely just store the data subject keys
# in our own DB and encrypt them with our own master encryption key that will *not* be stored in
# our own DB (possibly in AWS KMS?). At this point, we don't really need the other features of
# Vault, so it's not really worth the additional work to get a Vault cluster set up.
defprotocol TeamWebsite.PIIEncryption do
  @moduledoc """
  Provides encryption for PII using Hashicorp's Vault

  [Vault](https://www.vaultproject.io) must be running and configured. The following system
  environment variables must be set in production:

      VAULT_ADDR="https://your.vault.host.example:8200"
      VAULT_TOKEN="token-for-app-role"

  In development and test environments, these default to `"http://vault:8200"` and `"root"`,
  respectively. Those values will work if you are using the included `docker-compose.yml` to
  launch the services for development and testing.
  """

  @fallback_to_any true

  @type keypair() :: {data_subject_id :: binary(), encrypted_key :: binary()}
  @type key() ::
          {data_subject_id :: binary()}
          | keypair()
          | (unencrypted_key :: binary())

  @spec encrypt(data :: any(), key :: key() | nil) ::
          {:ok, encrypted_data :: any()}
          | {:ok, encrypted_data :: any(), keypair() | nil}
          | {:error, reason :: atom()}
  @doc ~S"""
  Encrypts PII data for the given data subject

  If no `encryption_key` is provided, we use `ExCrypto` to generate a 256-bit AES encryption key,
  and that key is then used to encrypt the `pii_data`. The key itself is then encrypted using
  [Vault's transit engine](https://www.vaultproject.io/api/secret/transit/index.html).

  If an `encryption_key` *is* provided, then the key (which we assume to be the encrypted version
  of the key) will be decrypted via Vault, and the decrypted encryption key will be used to
  encrypt the `pii_data`.

  In either case, both the encrypted key and the encrypted PII are returned upon success.

  ### Examples: ###

  Encrypting a single value, no key provided:
      iex> alias TeamWebsite.PIIEncryption, as: PII
      iex> data_subject_id = "68c57252-961a-43ef-85f4-e78fac327058"
      iex> pii_data = "John Doe"
      iex> {:ok, encrypted_data, key} = PII.encrypt(pii_data, {data_subject_id})
      iex> encrypted_data == pii_data
      false
      iex> PII.decrypt(encrypted_data, key)
      {:ok, "John Doe"}

  Encrypting a map of data, no key provided:
      iex> alias TeamWebsite.PIIEncryption, as: PII
      iex> data_subject_id = "b65b2f14-22ab-4df1-91ea-0a05bcb88e44"
      iex> data = %{email: "[email protected]", name: "John Doe"}
      iex> {:ok, encrypted_data, key} = PII.encrypt(data, {data_subject_id})
      iex> data.email == encrypted_data.email
      false
      iex> data.name == encrypted_data.name
      false
      iex> PII.decrypt(encrypted_data, key)
      {:ok, %{email: "[email protected]", name: "John Doe"}}

  Encrypting a single value, key provided:

      iex> alias TeamWebsite.PIIEncryption, as: PII
      iex> data_subject_id = "68c57252-961a-43ef-85f4-e78fac327058"
      iex> pii_data = "John Doe"
      iex> {:ok, _encrypted_data, key} = PII.encrypt("foo", {data_subject_id})
      iex> {:ok, encrypted_data, _key} = PII.encrypt(pii_data, key)
      iex> encrypted_data == pii_data
      false
      iex> PII.decrypt(encrypted_data, key)
      {:ok, "John Doe"}

  Encrypting a map of data, key provided:
      iex> alias TeamWebsite.PIIEncryption, as: PII
      iex> data_subject_id = "b65b2f14-22ab-4df1-91ea-0a05bcb88e44"
      iex> data = %{email: "[email protected]", name: "John Doe"}
      iex> {:ok, _encrypted_data, key} = PII.encrypt("foo", {data_subject_id})
      iex> {:ok, encrypted_data, _key} = PII.encrypt(data, key)
      iex> data.email == encrypted_data.email
      false
      iex> data.name == encrypted_data.name
      false
      iex> PII.decrypt(encrypted_data, key)
      {:ok, %{email: "[email protected]", name: "John Doe"}}
  """
  def encrypt(data, key \\ nil)

  @spec decrypt(encrypted_data :: any(), key :: key() | nil) ::
          {:ok, data :: any()} | {:error, reason :: atom()}
  @doc """
  Decrypts data using the provided encrypted key as returned by
  `TeamWebsite.PIIEncryption.encrypt/2`.
  """
  def decrypt(encrypted_data, key \\ nil)
end

defmodule TeamWebsite.PIIEncryption.Keys do
  @moduledoc false

  alias TeamWebsite.PIIEncryption.Vault

  defmacro __using__(_) do
    quote do
      alias TeamWebsite.PIIEncryption
      alias TeamWebsite.PIIEncryption.Keys

      def encrypt(data, key) when is_tuple(key) do
        Keys.encrypt_for(key, data, fn data, key -> encrypt(data, key) end)
      end

      def decrypt(encrypted_data, key) when is_tuple(key) do
        Keys.decrypt_for(key, encrypted_data, fn data, key -> decrypt(data, key) end)
      end
    end
  end

  def encrypt_key(data_subject_id, key) do
    Vault.write("transit/keys/#{data_subject_id}")
    key = Base.encode64(key)

    {:ok, %{"ciphertext" => encrypted_key}} =
      Vault.write("transit/encrypt/#{data_subject_id}", %{"plaintext" => key})

    {:ok, {data_subject_id, encrypted_key}}
  end

  def decrypt_key(data_subject_id, encrypted_key) do
    case Vault.write("transit/decrypt/#{data_subject_id}", %{"ciphertext" => encrypted_key}) do
      {:ok, %{"plaintext" => key}} -> {:ok, key |> Base.decode64!()}
      {:error, _} -> {:error, :decryption_failed}
    end
  end

  def encrypt_for({data_subject_id}, data, func) do
    {:ok, decrypted_key} = ExCrypto.generate_aes_key(:aes_256, :bytes)
    {:ok, encrypted_data} = func.(data, decrypted_key)
    {:ok, encrypted_key} = encrypt_key(data_subject_id, decrypted_key)
    {:ok, encrypted_data, encrypted_key}
  end

  def encrypt_for({data_subject_id, encrypted_key}, data, func) do
    {:ok, decrypted_key} = decrypt_key(data_subject_id, encrypted_key)
    {:ok, encrypted_data} = func.(data, decrypted_key)
    {:ok, encrypted_data, {data_subject_id, encrypted_key}}
  end

  def decrypt_for({data_subject_id, encrypted_key}, data, func) do
    {:ok, key} = decrypt_key(data_subject_id, encrypted_key)
    func.(data, key)
  end
end

defimpl TeamWebsite.PIIEncryption, for: BitString do
  use TeamWebsite.PIIEncryption.Keys

  def encrypt(data, key) when is_binary(key) do
    {:ok, {iv, encrypted_part}} = ExCrypto.encrypt(key, data)

    encrypted_data =
      [iv, encrypted_part]
      |> Enum.map(&Base.encode64/1)
      |> Enum.join(":")

    {:ok, encrypted_data}
  end

  def decrypt(encrypted_data, key) when is_binary(key) do
    [iv, encrypted_part] =
      encrypted_data
      |> String.split(":", parts: 2)
      |> Enum.map(&Base.decode64!/1)

    ExCrypto.decrypt(key, iv, encrypted_part)
  end
end

defimpl TeamWebsite.PIIEncryption, for: List do
  use TeamWebsite.PIIEncryption.Keys

  def encrypt(data, key) when is_binary(key) do
    encrypted_data =
      Enum.reduce(data, [], fn item, acc ->
        {:ok, encrypted_item} = PIIEncryption.encrypt(item, key)
        [encrypted_item | acc]
      end)
      |> Enum.reverse()

    {:ok, encrypted_data}
  end

  def decrypt(encrypted_data, key) when is_binary(key) do
    data =
      encrypted_data
      |> Enum.reverse()
      |> Enum.reduce([], fn item, acc ->
        {:ok, decrypted_item} = PIIEncryption.decrypt(item, key)
        [decrypted_item | acc]
      end)

    {:ok, data}
  end
end

defimpl TeamWebsite.PIIEncryption, for: Tuple do
  use TeamWebsite.PIIEncryption.Keys

  def encrypt({listkey, data}, key) when is_atom(listkey) and is_binary(key) do
    {:ok, encrypted_data} = PIIEncryption.encrypt(data, key)
    {:ok, {listkey, encrypted_data}}
  end

  def encrypt(data, key) when is_binary(key) do
    {:ok, encrypted_data} =
      data
      |> Tuple.to_list()
      |> PIIEncryption.encrypt(key)

    {:ok, List.to_tuple(encrypted_data)}
  end

  def decrypt({listkey, encrypted_data}, key) when is_atom(listkey) and is_binary(key) do
    {:ok, data} = PIIEncryption.decrypt(encrypted_data, key)
    {:ok, {listkey, data}}
  end

  def decrypt(encrypted_data, key) when is_binary(key) do
    {:ok, data} =
      encrypted_data
      |> Tuple.to_list()
      |> PIIEncryption.decrypt(key)

    {:ok, List.to_tuple(data)}
  end
end

defimpl TeamWebsite.PIIEncryption, for: Map do
  use TeamWebsite.PIIEncryption.Keys

  def encrypt(data, key) when is_binary(key) do
    {:ok, encrypted_data} =
      data
      |> Map.to_list()
      |> PIIEncryption.encrypt(key)

    {:ok, encrypted_data |> Enum.into(%{})}
  end

  def decrypt(encrypted_data, key) when is_binary(key) do
    {:ok, data} =
      encrypted_data
      |> Map.to_list()
      |> PIIEncryption.decrypt(key)

    {:ok, data |> Enum.into(%{})}
  end
end

defimpl TeamWebsite.PIIEncryption, for: Any do
  def encrypt(data, _key), do: {:ok, data, nil}
  def decrypt(data, _key), do: {:ok, data}
end

user.ex

# TODO: implement PIIEncryption for snapshots
defmodule TeamWebsite.Users.User do
  @moduledoc """
  Represents an individual person who is associated with a particular `TeamWebsite.Teams.Team`
  """

  use TypedStruct

  alias TeamWebsite.Users.Events.{EmailChangeConfirmed, EmailChanged, NameChanged, UserCreated}

  @type id() :: Ecto.UUID.t()

  @typedoc """
  The user's password id hashed using `TeamWebsite.Authentication.hash_password/1`.
  """
  @type password_hash() :: binary()

  @derive Jason.Encoder
  typedstruct do
    field :encrypted, boolean(), default: false
    field :encryption_key, binary()

    field :id, id()
    field :team_id, TeamWebsite.Teams.Team.id()
    field :email, binary() | String.t()
    field :pending_email, binary() | String.t()
    field :pending_email_confirmation_code, String.t()
    field :first_name, binary() | String.t()
    field :last_name, binary() | String.t()
    field :password_hash, password_hash()
  end

  use TeamWebsite.PIIEncryption.Encryptable

  def apply(%{id: nil}, %UserCreated{} = event) do
    %__MODULE__{
      id: event.id,
      team_id: event.team_id,
      password_hash: event.password_hash
    }
  end

  def apply(aggregate, %NameChanged{} = event) do
    %{aggregate | first_name: event.first_name, last_name: event.last_name}
  end

  def apply(aggregate, %EmailChanged{} = event) do
    %{
      aggregate
      | pending_email: event.email,
        pending_email_confirmation_code: event.confirmation_code
    }
  end

  def apply(aggregate, %EmailChangeConfirmed{} = event) do
    %{aggregate | email: event.email}
    |> maybe_clear_unconfirmed_email()
  end

  def generate_id, do: Ecto.UUID.generate()

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encrypted?(data), do: data.encrypted

  @impl TeamWebsite.PIIEncryption.Encryptable
  def data_subject_id(data), do: data.id

  @impl TeamWebsite.PIIEncryption.Encryptable
  def encryption_key(data), do: data.encryption_key

  @impl TeamWebsite.PIIEncryption.Encryptable
  def get_encrypted_field_map(data) do
    data
    |> Map.from_struct()
    |> Map.take([:email, :pending_email, :first_name, :last_name])
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_encrypted_fields(data, encrypted_data, encrypted_key) do
    data
    |> Map.from_struct()
    |> Map.merge(encrypted_data)
    |> Map.put(:encrypted, true)
    |> Map.put(:encryption_key, encrypted_key)
    |> (&struct!(__MODULE__, &1)).()
  end

  @impl TeamWebsite.PIIEncryption.Encryptable
  def apply_decrypted_fields(data, decrypted_data) do
    data
    |> Map.from_struct()
    |> Map.merge(decrypted_data)
    |> Map.put(:encrypted, false)
    |> Map.put(:encryption_key, nil)
    |> (&struct!(__MODULE__, &1)).()
  end

  defp maybe_clear_unconfirmed_email(aggregate) do
    if aggregate.email == aggregate.pending_email do
      %{aggregate | pending_email: nil, pending_email_confirmation_code: nil}
    else
      aggregate
    end
  end
end

user_created.ex

defmodule TeamWebsite.Users.Events.UserCreated do
  @moduledoc """
  Represents the initial creation of a new `TeamWebsite.Aggregates.User` aggregate
  """

  @schema_version 1

  use TypedStruct

  @derive Jason.Encoder
  typedstruct do
    field :schema_version, non_neg_integer(), default: @schema_version

    field :id, binary(), enforce: true
    field :team_id, binary(), enforce: true
    field :password_hash, binary(), enforce: true
  end
end

vault.ex

defmodule TeamWebsite.PIIEncryption.Vault do
  @moduledoc false

  alias Vaultex.Client

  def write(path, data \\ %{}) do
    Client.write(path, data, :token, {token()})
  end

  def read(path) do
    Client.read(path, :token, {token()})
  end

  def delete(path) do
    Client.delete(path, :token, {token()})
  end

  defp token, do: Application.get_env(:teamwebsite, :vault_token)
end