AshAuthentication.Phoenix.Oauth2Server.BearerPlug (ash_authentication_oauth2_server v0.1.1)

Resource-server side bearer token validation.

Validates an Authorization: Bearer <jwt> header against the configured authorization server. On success, loads the user via Ash.get/3 on the configured user_resource and sets it as the conn's actor.

Usage

pipeline :mcp_protected do
  plug AshAuthentication.Phoenix.Oauth2Server.BearerPlug,
    oauth2_server: MyApp.Oauth2Server
end

Options

:oauth2_server (required) — your Oauth2Server config module
:required? (default true) — when false, missing/invalid tokens pass through unchanged instead of returning 401. Useful for routes that should serve unauthenticated users with a different (e.g. session-based) signal.

Failure behavior

Per RFC 6750 §3, a missing or invalid token results in 401 with a WWW-Authenticate: Bearer resource_metadata="..." header pointing at the protected-resource metadata endpoint, so MCP-style clients can auto-discover the authorization server.

What ends up on the conn

On success two things are set, and downstream code reads from each for different purposes:

Ash.PlugHelpers.get_actor(conn) — the user record loaded via Ash.get/3 on the configured user_resource using the token's sub claim. Use this for "who is this" (Ash policies, tenant resolution, ownership checks).
conn.assigns.oauth_claims — the verified JWT claims map. Use this for "what is this bearer allowed to do" — most importantly the scope claim:
```
scopes =
  conn.assigns.oauth_claims["scope"]
  |> String.split(" ", trim: true)
```
Other useful claims: client_id (which OAuth client minted this), aud (which resource), jti (unique token id).

Note that scopes are conn-scoped, not actor-scoped. The same user with two access tokens minted for two different clients ends up with the same actor but different oauth_claims["scope"]. This is the right OAuth semantic — the access token is a delegated grant from user → client, distinct from the user's own permissions.

Gating an action on a scope

The minimum useful pattern is a plug that 403s when a required scope isn't present. Drop one of these into your pipeline after this plug, or write it inline in your controller:

defmodule MyAppWeb.RequireScope do
  @behaviour Plug
  import Plug.Conn

  @impl true
  def init(scope) when is_binary(scope), do: scope

  @impl true
  def call(conn, scope) do
    scopes =
      conn.assigns
      |> Map.get(:oauth_claims, %{})
      |> Map.get("scope", "")
      |> String.split(" ", trim: true)

    if scope in scopes do
      conn
    else
      conn |> send_resp(403, "") |> halt()
    end
  end
end

pipeline :mcp_read do
  plug AshAuthentication.Phoenix.Oauth2Server.BearerPlug,
    oauth2_server: MyApp.Oauth2Server
  plug MyAppWeb.RequireScope, "mcp.read"
end

Reading scopes inside an Ash policy

If you'd rather gate at the resource layer, copy oauth_claims into the actor's metadata or the action context before invoking the action. For example, a tiny plug between BearerPlug and your controller:

plug fn conn, _ ->
  Ash.PlugHelpers.update_context(conn, fn ctx ->
    Map.put(ctx || %{}, :oauth_scopes,
      conn.assigns.oauth_claims["scope"]
      |> String.split(" ", trim: true))
  end)
end

then in your resource:

policies do
  policy action(:read) do
    authorize_if expr(^context(:oauth_scopes) |> contains("mcp.read"))
  end
end