JWS X.509 validation

jws_x509.ex

defmodule JwsX509 do

  require Record

  # Some code taken from https://github.com/hexpm/hex/blob/8410f8d48607a67bec713a6cbb760bbba8447aab/lib/hex/http/ssl.ex
  # Licensed under Apache 2.0

  Record.defrecordp(
    :certificate,
    :OTPCertificate,
    Record.extract(:OTPCertificate, from_lib: "public_key/include/OTP-PUB-KEY.hrl")
  )

  Record.defrecordp(
    :tbs_certificate,
    :OTPTBSCertificate,
    Record.extract(:OTPTBSCertificate, from_lib: "public_key/include/OTP-PUB-KEY.hrl")
  )

  Record.defrecordp(
    :subject_public_key_info,
    :OTPSubjectPublicKeyInfo,
    Record.extract(:OTPSubjectPublicKeyInfo, from_lib: "public_key/include/OTP-PUB-KEY.hrl")
  )

  def validate(jws, verify_fun) do
    cacerts = {:der, :certifi.cacerts()}
    Process.put(:ssl_manager, :ssl_manager.name(:normal))
    {:ok, db_state} = :ssl_manager.connection_init(cacerts, :client, {:ssl_crl_cache, {:internal, []}})

    protected = JOSE.JWS.peek_protected(jws) |> Poison.decode!
    certs = Map.get(protected, "x5c")
    certs = Enum.map(certs, fn x -> Base.decode64!(x) end)

    # https://github.com/erlang/otp/blob/f529c3a4d5add57c2d65114781d1f3c4f96d2a7c/lib/ssl/src/ssl_handshake.erl#L338
    {trusted_cert, cert_path} = :ssl_certificate.trusted_cert_and_path(certs, db_state.cert_db_handle,
      db_state.cert_db_ref, &validate_partial_chain(cacerts, &1))
    {:ok, {{_, pubkey, _}, _}} = :public_key.pkix_path_validation(trusted_cert, cert_path, max_path_length: 20, verify_fun: verify_fun)

    jwk = JOSE.JWK.from_key(pubkey)
    JOSE.JWS.verify(jwk, jws)
  end

  defp validate_partial_chain(cacerts, certs) do
    certs = Enum.map(certs, &{&1, :public_key.pkix_decode_cert(&1, :otp)})
    cacerts = Enum.map(cacerts, &:public_key.pkix_decode_cert(&1, :otp))

    trusted =
      Enum.find_value(certs, fn {der, cert} ->
        trusted? =
          Enum.find(cacerts, fn cacert ->
            extract_public_key_info(cacert) == extract_public_key_info(cert)
          end)

        if trusted?, do: der
      end)

    if trusted do
      {:trusted_ca, trusted}
    else
      :unknown_ca
    end
  end

  defp extract_public_key_info(cert) do
    cert
    |> certificate(:tbsCertificate)
    |> tbs_certificate(:subjectPublicKeyInfo)
  end

  # A simple cert validation function

  def validate_cert(_, {:extension, _}, user_state), do: {:unknown, user_state}
  def validate_cert(_, {:bad_cert, _} = reason, _), do: {:fail, reason}
  def validate_cert(_, :valid, user_state), do: {:valid, user_state}
  def validate_cert(cert, :valid_peer, user_state) do
    case :public_key.pkix_verify_hostname(cert, [uri_id: Keyword.get(user_state, :hostname)]) do
      true -> {:valid, user_state}
      false -> {:fail, {:bad_cert, :hostname_check_failed}}
    end
  end

end

Usage example:

    JwsX509.validate(
      result,
      {
        &JwsX509.validate_cert/3,
        [hostname: "https://attest.android.com/"]
      }
    )