Git of dcc - anni/blob - lib/pleroma/web/o_auth/token/utils.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.OAuth.Token.Utils do
   6   @moduledoc """
   7   Auxiliary functions for dealing with tokens.
   8   """
   9
  10   alias Pleroma.Repo
  11   alias Pleroma.Web.OAuth.App
  12
  13   @doc "Fetch app by client credentials from request"
  14   @spec fetch_app(Plug.Conn.t()) :: {:ok, App.t()} | {:error, :not_found}
  15   def fetch_app(conn) do
  16     res =
  17       conn
  18       |> fetch_client_credentials()
  19       |> fetch_client
  20
  21     case res do
  22       %App{} = app -> {:ok, app}
  23       _ -> {:error, :not_found}
  24     end
  25   end
  26
  27   defp fetch_client({id, secret}) when is_binary(id) and is_binary(secret) do
  28     Repo.get_by(App, client_id: id, client_secret: secret)
  29   end
  30
  31   defp fetch_client({_id, _secret}), do: nil
  32
  33   defp fetch_client_credentials(conn) do
  34     # Per RFC 6749, HTTP Basic is preferred to body params
  35     with ["Basic " <> encoded] <- Plug.Conn.get_req_header(conn, "authorization"),
  36          {:ok, decoded} <- Base.decode64(encoded),
  37          [id, secret] <-
  38            Enum.map(
  39              String.split(decoded, ":"),
  40              fn s -> URI.decode_www_form(s) end
  41            ) do
  42       {id, secret}
  43     else
  44       _ -> {conn.params["client_id"], conn.params["client_secret"]}
  45     end
  46   end
  47
  48   @doc "convert token inserted_at to unix timestamp"
  49   def format_created_at(%{inserted_at: inserted_at} = _token) do
  50     inserted_at
  51     |> DateTime.from_naive!("Etc/UTC")
  52     |> DateTime.to_unix()
  53   end
  54
  55   @doc false
  56   @spec generate_token(keyword()) :: binary()
  57   def generate_token(opts \\ []) do
  58     opts
  59     |> Keyword.get(:size, 32)
  60     |> :crypto.strong_rand_bytes()
  61     |> Base.url_encode64(padding: false)
  62   end
  63
  64   # XXX - for whatever reason our token arrives urlencoded, but Plug.Conn should be
  65   # decoding it.  Investigate sometime.
  66   def fix_padding(token) do
  67     token
  68     |> URI.decode()
  69     |> Base.url_decode64!(padding: false)
  70     |> Base.url_encode64(padding: false)
  71   end
  72 end