不使用guardian_db，如何在登录端点生成Access/Refresh Token？

太棒了，不用guardian_db完全能实现短时效Access Token + 可刷新Refresh Token的方案！核心思路是自己搭建Refresh Token的存储与校验机制，同时借助Guardian的JWT生成能力。下面我一步步给你拆解实现步骤，结合你的代码基础来调整：

1. 完善Guardian配置 #

首先补全config/config.exs里的Guardian配置，明确两种Token的过期时间：

# config/config.exs
config :my_proj, MyProj.Guardian,
  issuer: "my_proj",
  verify_module: Guardian.JWT,
  secret_key: System.get_env("GUARDIAN_SECRET_KEY") || "your-development-secret-key", # 生产环境务必用环境变量
  token_ttl: %{
    access: {15, :minute},  # Access Token设15分钟，降低泄露风险
    refresh: {7, :day}      # Refresh Token设7天，平衡安全性和用户体验
  }

2. 定义Guardian核心模块 #

创建lib/my_proj/guardian.ex，实现Guardian的核心回调，用来关联用户和Token：

defmodule MyProj.Guardian do
  use Guardian, otp_app: :my_proj

  alias MyProj.{Repo, User}

  # 把用户ID作为JWT的sub字段（唯一标识）
  def subject_for_token(user, _claims) do
    {:ok, to_string(user.id)}
  end

  # 从JWT的sub字段反查用户
  def resource_from_claims(claims) do
    case Repo.get(User, claims["sub"]) do
      nil -> {:error, :resource_not_found}
      user -> {:ok, user}
    end
  end
end

3. 创建Refresh Token存储模型 #

因为不用guardian_db，我们需要自己存储Refresh Token来做校验和失效控制。先定义Ecto Schema：

# lib/my_proj/refresh_token.ex
defmodule MyProj.RefreshToken do
  use Ecto.Schema
  import Ecto.Changeset

  schema "refresh_tokens" do
    field :token, :string # 存储明文Refresh Token（需和前端传来的对比）
    belongs_to :user, MyProj.User

    timestamps()
  end

  @doc false
  def changeset(refresh_token, attrs) do
    refresh_token
    |> cast(attrs, [:token, :user_id])
    |> validate_required([:token, :user_id])
    |> unique_constraint(:token) # 确保Token唯一
  end

  # 创建新的Refresh Token记录
  def create(attrs) do
    %__MODULE__{}
    |> changeset(attrs)
    |> MyProj.Repo.insert()
  end

  # 通过Token字符串查询记录
  def get_by_token(token) do
    MyProj.Repo.get_by(__MODULE__, token: token)
  end

  # 删除指定Token的记录（用于登出/失效）
  def delete_by_token(token) do
    case get_by_token(token) do
      nil -> {:error, :not_found}
      record -> MyProj.Repo.delete(record)
    end
  end
end

然后生成对应的数据库迁移文件：

mix ecto.gen.migration create_refresh_tokens

修改迁移文件内容：

# priv/repo/migrations/xxxxxx_create_refresh_tokens.exs
defmodule MyProj.Repo.Migrations.CreateRefreshTokens do
  use Ecto.Migration

  def change do
    create table(:refresh_tokens) do
      add :token, :string, null: false
      add :user_id, references(:users, on_delete: :delete_all), null: false

      timestamps()
    end

    create unique_index(:refresh_tokens, [:token]) # 防止重复Token
    create index(:refresh_tokens, [:user_id]) # 优化用户关联查询
  end
end

执行迁移：

mix ecto.migrate

4. 登录端点生成双Token #

在你的登录控制器（比如lib/my_proj_web/controllers/session_controller.ex）里，实现生成Access和Refresh Token的逻辑：

defmodule MyProjWeb.SessionController do
  use MyProjWeb, :controller

  alias MyProj.{Repo, User, Guardian, RefreshToken}
  alias Comeonin.Bcrypt # 假设你用Bcrypt做密码校验，可替换成你用的加密库

  def create(conn, %{"email" => email, "password" => password}) do
    with %User{} = user <- Repo.get_by(User, email: email),
         true <- Bcrypt.checkpw(password, user.password_hash),
         # 生成Access Token
         {:ok, access_token, _access_claims} <- Guardian.encode_and_sign(user, %{}, token_type: "access"),
         # 生成Refresh Token
         {:ok, refresh_token, _refresh_claims} <- Guardian.encode_and_sign(user, %{}, token_type: "refresh"),
         # 把Refresh Token存入数据库
         {:ok, _refresh_record} <- RefreshToken.create(%{user_id: user.id, token: refresh_token}) do

      conn
      |> put_status(:ok)
      |> json(%{
        access_token: access_token,
        refresh_token: refresh_token,
        token_type: "Bearer",
        expires_in: 15 * 60 # 和Access Token的过期时间一致（15分钟转秒）
      })
    else
      nil ->
        conn |> put_status(:unauthorized) |> json(%{error: "Invalid email or password"})
      false ->
        conn |> put_status(:unauthorized) |> json(%{error: "Invalid email or password"})
      _error ->
        conn |> put_status(:internal_server_error) |> json(%{error: "Failed to generate tokens"})
    end
  end
end

5. 实现Token刷新逻辑 #

添加一个刷新端点，用Refresh Token生成新的Access Token：

def refresh(conn, %{"refresh_token" => refresh_token}) do
  with # 先校验Refresh Token的签名和过期时间
       {:ok, claims} <- Guardian.decode_and_verify(refresh_token, token_type: "refresh"),
       # 从Token反查用户
       {:ok, user} <- Guardian.resource_from_claims(claims),
       # 检查数据库中是否存在该Refresh Token（确保未被失效）
       %RefreshToken{} = token_record <- RefreshToken.get_by_token(refresh_token),
       # 确认Token属于当前用户
       true <- user.id == token_record.user_id,
       # 生成新的Access Token
       {:ok, new_access_token, _access_claims} <- Guardian.encode_and_sign(user, %{}, token_type: "access") do

    conn
    |> put_status(:ok)
    |> json(%{
      access_token: new_access_token,
      token_type: "Bearer",
      expires_in: 15 * 60
    })
  else
    _ ->
      conn |> put_status(:unauthorized) |> json(%{error: "Invalid or expired refresh token"})
  end
end

6. 处理Refresh Token失效（登出/密码修改） #

当用户登出或者修改密码时，需要删除对应的Refresh Token，使其无法再刷新Access Token：

def delete(conn, %{"refresh_token" => refresh_token}) do
  case RefreshToken.delete_by_token(refresh_token) do
    {:ok, _} ->
      conn |> put_status(:ok) |> json(%{message: "Successfully logged out"})
    {:error, _} ->
      conn |> put_status(:bad_request) |> json(%{error: "Failed to log out"})
  end
end

关键注意点 #

• Refresh Token的安全性：前端要把Refresh Token存在HttpOnly、Secure的Cookie里，避免XSS攻击；传输时用HTTPS，防止中间人窃取。
• Token类型校验：在保护API的Plug里，要校验Token的token_type为access，避免用Refresh Token直接访问业务接口。
• 定期清理过期Token：可以写一个定时任务，删除数据库中已过期的Refresh Token（根据inserted_at和配置的过期时间计算），减少数据库冗余。

内容的提问来源于stack exchange，提问作者Nuno_Coletiv