From 7a1285fbb2e2fac2fe542635c40499af71059093 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:11:54 +0000
Subject: [PATCH 01/25] feat(integrations): add device authorization claims
---
lib/kodo/integrations.ex | 112 +++++++++++++++++++++++++++++
test/kodo/integrations_test.exs | 124 ++++++++++++++++++++++++++++++++
2 files changed, 236 insertions(+)
diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex
index d52b81e..1754b93 100644
--- a/lib/kodo/integrations.ex
+++ b/lib/kodo/integrations.ex
@@ -25,6 +25,7 @@ defmodule Kodo.Integrations do
@device_authorization_lifetime_seconds 15 * 60
@device_authorization_secret_max_bytes 1_024
@device_authorization_cleanup_age_seconds 24 * 60 * 60
+ @device_authorization_claim_lease_ms 30_000
@secret_repo_options [log: false, telemetry_event: nil]
def list_integrations(%Scope{user: user}) do
@@ -130,6 +131,48 @@ defmodule Kodo.Integrations do
end
end
+ def claim_device_authorization(%Scope{user: user}, integration_id, claim_owner_id) do
+ with {:ok, integration_id} <- cast_uuid(integration_id),
+ {:ok, claim_owner_id} <- cast_uuid(claim_owner_id) do
+ result =
+ Repo.transaction(fn ->
+ expire_device_authorization(user.id, integration_id)
+ claim_device_authorization_locked(user.id, integration_id, claim_owner_id)
+ end)
+
+ case result do
+ {:ok, {:error, reason}} -> {:error, reason}
+ other -> other
+ end
+ end
+ end
+
+ def renew_device_authorization_claim(
+ %Scope{user: user},
+ attempt_id,
+ attempt_generation,
+ claim_owner_id,
+ claim_epoch
+ )
+ when is_integer(attempt_generation) and attempt_generation > 0 and
+ is_integer(claim_epoch) and claim_epoch > 0 do
+ with {:ok, attempt_id} <- cast_uuid(attempt_id),
+ {:ok, claim_owner_id} <- cast_uuid(claim_owner_id) do
+ Repo.transaction(fn ->
+ renew_device_authorization_claim_locked(
+ user.id,
+ attempt_id,
+ attempt_generation,
+ claim_owner_id,
+ claim_epoch
+ )
+ end)
+ end
+ end
+
+ def renew_device_authorization_claim(%Scope{}, _attempt_id, _generation, _owner, _epoch),
+ do: {:error, :stale_device_authorization_claim}
+
def cancel_device_authorization(%Scope{user: user}, attempt_id, attempt_generation)
when is_integer(attempt_generation) and attempt_generation > 0 do
with {:ok, attempt_id} <- cast_uuid(attempt_id) do
@@ -417,6 +460,75 @@ defmodule Kodo.Integrations do
end
end
+ defp claim_device_authorization_locked(user_id, integration_id, claim_owner_id) do
+ timestamp = database_now()
+ lease_expires_at = DateTime.add(timestamp, @device_authorization_claim_lease_ms, :millisecond)
+
+ query =
+ claimable_device_authorization_query(user_id, integration_id, claim_owner_id, timestamp)
+ |> select([attempt, _integration], attempt)
+
+ case Repo.update_all(
+ query,
+ [
+ set: [
+ claim_owner_id: claim_owner_id,
+ claim_lease_expires_at: lease_expires_at,
+ updated_at: timestamp
+ ],
+ inc: [claim_epoch: 1]
+ ],
+ @secret_repo_options
+ ) do
+ {1, [attempt]} -> decrypt_device_authorization(attempt)
+ {0, []} -> {:error, :device_authorization_not_claimable}
+ end
+ end
+
+ defp claimable_device_authorization_query(user_id, integration_id, claim_owner_id, timestamp) do
+ from attempt in DeviceAuthorizationAttempt,
+ join: integration in Integration,
+ on: integration.id == attempt.integration_id and integration.user_id == attempt.user_id,
+ where:
+ attempt.user_id == ^user_id and attempt.integration_id == ^integration_id and
+ attempt.state == "active" and attempt.provider_deadline > ^timestamp and
+ integration.credential_generation == attempt.expected_integration_generation and
+ (is_nil(attempt.claim_owner_id) or attempt.claim_lease_expires_at <= ^timestamp or
+ attempt.claim_owner_id == ^claim_owner_id)
+ end
+
+ defp renew_device_authorization_claim_locked(
+ user_id,
+ attempt_id,
+ attempt_generation,
+ claim_owner_id,
+ claim_epoch
+ ) do
+ timestamp = database_now()
+ lease_expires_at = DateTime.add(timestamp, @device_authorization_claim_lease_ms, :millisecond)
+
+ query =
+ from attempt in DeviceAuthorizationAttempt,
+ join: integration in Integration,
+ on: integration.id == attempt.integration_id and integration.user_id == attempt.user_id,
+ where:
+ attempt.id == ^attempt_id and attempt.user_id == ^user_id and
+ attempt.attempt_generation == ^attempt_generation and attempt.state == "active" and
+ attempt.claim_owner_id == ^claim_owner_id and attempt.claim_epoch == ^claim_epoch and
+ attempt.claim_lease_expires_at > ^timestamp and attempt.provider_deadline > ^timestamp and
+ integration.credential_generation == attempt.expected_integration_generation,
+ select: attempt
+
+ case Repo.update_all(
+ query,
+ [set: [claim_lease_expires_at: lease_expires_at, updated_at: timestamp]],
+ @secret_repo_options
+ ) do
+ {1, [attempt]} -> decrypt_device_authorization(attempt)
+ {0, []} -> Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
defp cancel_device_authorization_locked(user_id, attempt_id, attempt_generation) do
integration_id =
DeviceAuthorizationAttempt
diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs
index 70c6821..0f45936 100644
--- a/test/kodo/integrations_test.exs
+++ b/test/kodo/integrations_test.exs
@@ -826,6 +826,130 @@ defmodule Kodo.IntegrationsTest do
)
end
+ test "claims one active attempt and fences renewal by owner and epoch", %{scope: scope} do
+ integration = oauth_integration(scope)
+ first_owner = Ecto.UUID.generate()
+ second_owner = Ecto.UUID.generate()
+ payload = %{"device_auth_id" => "device", "user_code" => "CODE"}
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ payload,
+ 1_000
+ )
+
+ assert {:ok, {first_claim, ^payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, first_owner)
+
+ assert first_claim.claim_owner_id == first_owner
+ assert first_claim.claim_epoch == 1
+ assert DateTime.after?(first_claim.claim_lease_expires_at, DateTime.utc_now())
+
+ assert {:error, :device_authorization_not_claimable} =
+ Integrations.claim_device_authorization(scope, integration.id, second_owner)
+
+ assert {:ok, {replacement_claim, ^payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, first_owner)
+
+ assert replacement_claim.claim_epoch == 2
+
+ assert {:error, :stale_device_authorization_claim} =
+ Integrations.renew_device_authorization_claim(
+ scope,
+ attempt.id,
+ attempt.attempt_generation,
+ first_owner,
+ first_claim.claim_epoch
+ )
+
+ assert {:ok, {renewed, ^payload}} =
+ Integrations.renew_device_authorization_claim(
+ scope,
+ attempt.id,
+ attempt.attempt_generation,
+ first_owner,
+ replacement_claim.claim_epoch
+ )
+
+ assert renewed.claim_epoch == replacement_claim.claim_epoch
+
+ assert DateTime.compare(
+ renewed.claim_lease_expires_at,
+ replacement_claim.claim_lease_expires_at
+ ) in [
+ :eq,
+ :gt
+ ]
+ end
+
+ test "permits takeover only after lease expiry and fences the prior claimant", %{scope: scope} do
+ integration = oauth_integration(scope)
+ first_owner = Ecto.UUID.generate()
+ second_owner = Ecto.UUID.generate()
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 1_000
+ )
+
+ assert {:ok, {first_claim, _payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, first_owner)
+
+ Repo.update!(
+ change(first_claim,
+ claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)
+ )
+ )
+
+ assert {:ok, {takeover, _payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, second_owner)
+
+ assert takeover.claim_owner_id == second_owner
+ assert takeover.claim_epoch == first_claim.claim_epoch + 1
+
+ assert {:error, :stale_device_authorization_claim} =
+ Integrations.renew_device_authorization_claim(
+ scope,
+ attempt.id,
+ attempt.attempt_generation,
+ first_owner,
+ first_claim.claim_epoch
+ )
+ end
+
+ test "claiming lazily expires an overdue attempt", %{scope: scope} do
+ integration = oauth_integration(scope)
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 1_000
+ )
+
+ Repo.update!(
+ change(attempt, provider_deadline: DateTime.add(DateTime.utc_now(), -1, :second))
+ )
+
+ assert {:error, :device_authorization_not_claimable} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ assert %{state: "expired", encrypted_payload: nil} = Repo.reload!(attempt)
+ end
+
test "cancellation is owned and generation fenced and removes the one-time code", %{
scope: scope
} do
From 5a922dca10d01d4d26f4f907dfb19f104c9bcb56 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:20:24 +0000
Subject: [PATCH 02/25] feat(integrations): normalize Codex authorization
tokens
---
.../device_authorization_tokens.ex | 70 +++++++++++++++++++
.../device_authorization_tokens_test.exs | 61 ++++++++++++++++
2 files changed, 131 insertions(+)
create mode 100644 lib/kodo/integrations/device_authorization_tokens.ex
create mode 100644 test/kodo/integrations/device_authorization_tokens_test.exs
diff --git a/lib/kodo/integrations/device_authorization_tokens.ex b/lib/kodo/integrations/device_authorization_tokens.ex
new file mode 100644
index 0000000..1064c6c
--- /dev/null
+++ b/lib/kodo/integrations/device_authorization_tokens.ex
@@ -0,0 +1,70 @@
+defmodule Kodo.Integrations.DeviceAuthorizationTokens do
+ @moduledoc false
+
+ @openai_auth_claim "https://api.openai.com/auth"
+
+ def normalize(tokens, now \\ DateTime.utc_now())
+
+ def normalize(
+ %{
+ "access_token" => access_token,
+ "refresh_token" => refresh_token,
+ "id_token" => id_token
+ },
+ now
+ ) do
+ with {:ok, access_claims} <- decode_claims(access_token),
+ {:ok, identity_claims} <- decode_claims(id_token),
+ {:ok, expires_at} <- expiration(access_claims, now),
+ {:ok, account_id} <- account_id(identity_claims) do
+ {:ok,
+ %{
+ credentials: %{
+ "access_token" => access_token,
+ "refresh_token" => refresh_token,
+ "id_token" => id_token,
+ "account_id" => account_id
+ },
+ expires_at: expires_at
+ }}
+ else
+ _invalid -> {:error, :device_authorization_response_invalid}
+ end
+ end
+
+ def normalize(_tokens, _now), do: {:error, :device_authorization_response_invalid}
+
+ defp decode_claims(token) when is_binary(token) do
+ case String.split(token, ".") do
+ [_header, payload, _signature] ->
+ with {:ok, decoded} <- Base.url_decode64(payload, padding: false),
+ {:ok, claims} when is_map(claims) <- Jason.decode(decoded) do
+ {:ok, claims}
+ else
+ _invalid -> :error
+ end
+
+ _invalid ->
+ :error
+ end
+ end
+
+ defp decode_claims(_token), do: :error
+
+ defp expiration(%{"exp" => unix}, now) when is_integer(unix) do
+ with {:ok, expires_at} <- DateTime.from_unix(unix),
+ true <- DateTime.after?(expires_at, now) do
+ {:ok, expires_at}
+ else
+ _invalid -> :error
+ end
+ end
+
+ defp expiration(_claims, _now), do: :error
+
+ defp account_id(%{@openai_auth_claim => %{"chatgpt_account_id" => account_id}})
+ when is_binary(account_id) and account_id != "",
+ do: {:ok, account_id}
+
+ defp account_id(_claims), do: :error
+end
diff --git a/test/kodo/integrations/device_authorization_tokens_test.exs b/test/kodo/integrations/device_authorization_tokens_test.exs
new file mode 100644
index 0000000..92332a8
--- /dev/null
+++ b/test/kodo/integrations/device_authorization_tokens_test.exs
@@ -0,0 +1,61 @@
+defmodule Kodo.Integrations.DeviceAuthorizationTokensTest do
+ use ExUnit.Case, async: true
+
+ alias Kodo.Integrations.DeviceAuthorizationTokens
+
+ @now ~U[2026-09-10 12:00:00Z]
+
+ test "normalizes required tokens, account identity, and access-token expiry" do
+ tokens = %{
+ "access_token" => jwt(%{"exp" => DateTime.to_unix(@now) + 3_600}),
+ "refresh_token" => "refresh-secret",
+ "id_token" =>
+ jwt(%{
+ "https://api.openai.com/auth" => %{"chatgpt_account_id" => "account-secret"}
+ })
+ }
+
+ assert {:ok, normalized} = DeviceAuthorizationTokens.normalize(tokens, @now)
+ assert normalized.expires_at == DateTime.add(@now, 3_600, :second)
+
+ assert normalized.credentials ==
+ Map.put(tokens, "account_id", "account-secret")
+ end
+
+ test "rejects missing identity, malformed tokens, and non-future expiration" do
+ valid_identity =
+ jwt(%{"https://api.openai.com/auth" => %{"chatgpt_account_id" => "account"}})
+
+ invalid_tokens = [
+ %{
+ "access_token" => jwt(%{"exp" => DateTime.to_unix(@now)}),
+ "refresh_token" => "refresh",
+ "id_token" => valid_identity
+ },
+ %{
+ "access_token" => "not-a-jwt",
+ "refresh_token" => "refresh",
+ "id_token" => valid_identity
+ },
+ %{
+ "access_token" => jwt(%{"exp" => DateTime.to_unix(@now) + 60}),
+ "refresh_token" => "refresh",
+ "id_token" => jwt(%{"https://api.openai.com/auth" => %{}})
+ },
+ %{
+ "access_token" => jwt(%{"exp" => DateTime.to_unix(@now) + 60}),
+ "refresh_token" => "refresh"
+ }
+ ]
+
+ for tokens <- invalid_tokens do
+ assert {:error, :device_authorization_response_invalid} =
+ DeviceAuthorizationTokens.normalize(tokens, @now)
+ end
+ end
+
+ defp jwt(claims) do
+ encoded = claims |> Jason.encode!() |> Base.url_encode64(padding: false)
+ "header.#{encoded}.signature"
+ end
+end
From cd8f29e9e9431edb8e95efed85912c503367924b Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:20:35 +0000
Subject: [PATCH 03/25] feat(integrations): fence device authorization
completion
---
lib/kodo/integrations.ex | 315 ++++++++++++++++++++++++++-
lib/kodo/integrations/audit_event.ex | 2 +
test/kodo/integrations_test.exs | 228 +++++++++++++++++++
3 files changed, 543 insertions(+), 2 deletions(-)
diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex
index 1754b93..2a8484b 100644
--- a/lib/kodo/integrations.ex
+++ b/lib/kodo/integrations.ex
@@ -24,8 +24,15 @@ defmodule Kodo.Integrations do
)
@device_authorization_lifetime_seconds 15 * 60
@device_authorization_secret_max_bytes 1_024
+ @device_authorization_token_max_bytes 8_192
@device_authorization_cleanup_age_seconds 24 * 60 * 60
@device_authorization_claim_lease_ms 30_000
+ @device_authorization_terminal_errors ~w(
+ provider_unavailable
+ provider_rejected
+ response_invalid
+ redirect
+ )
@secret_repo_options [log: false, telemetry_event: nil]
def list_integrations(%Scope{user: user}) do
@@ -173,6 +180,69 @@ defmodule Kodo.Integrations do
def renew_device_authorization_claim(%Scope{}, _attempt_id, _generation, _owner, _epoch),
do: {:error, :stale_device_authorization_claim}
+ def admit_device_authorization_poll(%Scope{user: user}, %DeviceAuthorizationAttempt{} = claim) do
+ Repo.transaction(fn -> admit_device_authorization_poll_locked(user.id, claim) end)
+ end
+
+ def store_device_authorization_exchange(
+ %Scope{user: user},
+ %DeviceAuthorizationAttempt{} = claim,
+ payload
+ ) do
+ with :ok <- validate_device_authorization_exchange_payload(payload),
+ {:ok, encrypted} <- DeviceAuthorizationEncryption.encrypt(claim, payload) do
+ Repo.transaction(fn ->
+ update_claimed_device_authorization_payload(user.id, claim, encrypted)
+ end)
+ end
+ end
+
+ def admit_device_authorization_exchange(
+ %Scope{user: user},
+ %DeviceAuthorizationAttempt{} = claim
+ ) do
+ Repo.transaction(fn -> admit_device_authorization_exchange_locked(user.id, claim) end)
+ end
+
+ def complete_device_authorization(
+ %Scope{user: user},
+ %DeviceAuthorizationAttempt{} = claim,
+ credentials,
+ %DateTime{} = expires_at
+ ) do
+ with :ok <- validate_device_authorization_credentials(credentials) do
+ Repo.transaction(fn ->
+ complete_device_authorization_locked(user.id, claim, credentials, expires_at)
+ end)
+ |> notify_integration_change()
+ end
+ end
+
+ def complete_device_authorization(
+ %Scope{},
+ %DeviceAuthorizationAttempt{},
+ _credentials,
+ _expiry
+ ),
+ do: {:error, :device_authorization_response_invalid}
+
+ def fail_device_authorization(
+ %Scope{user: user},
+ %DeviceAuthorizationAttempt{} = claim,
+ error_code
+ )
+ when error_code in @device_authorization_terminal_errors do
+ Repo.transaction(fn ->
+ attempt = terminalize_claimed_device_authorization(user.id, claim, "failed", error_code)
+ integration = Repo.get!(Integration, attempt.integration_id)
+ audit!(user.id, integration, "device_authorization_failed")
+ attempt
+ end)
+ end
+
+ def fail_device_authorization(%Scope{}, %DeviceAuthorizationAttempt{}, _error_code),
+ do: {:error, :unsafe_device_authorization_error}
+
def cancel_device_authorization(%Scope{user: user}, attempt_id, attempt_generation)
when is_integer(attempt_generation) and attempt_generation > 0 do
with {:ok, attempt_id} <- cast_uuid(attempt_id) do
@@ -529,6 +599,205 @@ defmodule Kodo.Integrations do
end
end
+ defp admit_device_authorization_poll_locked(user_id, claim) do
+ timestamp = database_now()
+ lease_expires_at = DateTime.add(timestamp, @device_authorization_claim_lease_ms, :millisecond)
+
+ query =
+ claimed_device_authorization_query(user_id, claim, timestamp)
+ |> where([attempt, _integration], attempt.next_poll_at <= ^timestamp)
+ |> update(
+ [attempt, _integration],
+ set: [
+ next_poll_at:
+ fragment(
+ "CAST(? AS timestamp) + (? * interval '1 millisecond')",
+ ^timestamp,
+ attempt.polling_interval_ms
+ ),
+ claim_lease_expires_at: ^lease_expires_at,
+ updated_at: ^timestamp
+ ]
+ )
+ |> select([attempt, _integration], attempt)
+
+ case Repo.update_all(query, [], @secret_repo_options) do
+ {1, [attempt]} ->
+ decrypt_device_authorization(attempt)
+
+ {0, []} ->
+ if claimed_device_authorization?(user_id, claim, timestamp),
+ do: Repo.rollback(:device_authorization_poll_not_due),
+ else: Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
+ defp claimed_device_authorization?(user_id, claim, timestamp) do
+ claim
+ |> then(&claimed_device_authorization_query(user_id, &1, timestamp))
+ |> Repo.exists?(@secret_repo_options)
+ end
+
+ defp update_claimed_device_authorization_payload(user_id, claim, encrypted) do
+ timestamp = database_now()
+
+ query =
+ claimed_device_authorization_query(user_id, claim, timestamp)
+ |> select([attempt, _integration], attempt)
+
+ changes =
+ encrypted
+ |> Map.put(:updated_at, timestamp)
+ |> Map.to_list()
+
+ case Repo.update_all(query, [set: changes], @secret_repo_options) do
+ {1, [attempt]} -> attempt
+ {0, []} -> Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
+ defp admit_device_authorization_exchange_locked(user_id, claim) do
+ timestamp = database_now()
+ lease_expires_at = DateTime.add(timestamp, @device_authorization_claim_lease_ms, :millisecond)
+
+ query =
+ claimed_device_authorization_query(user_id, claim, timestamp)
+ |> select([attempt, _integration], attempt)
+
+ case Repo.update_all(
+ query,
+ [set: [claim_lease_expires_at: lease_expires_at, updated_at: timestamp]],
+ @secret_repo_options
+ ) do
+ {1, [attempt]} ->
+ case decrypt_device_authorization(attempt) do
+ {attempt, payload} ->
+ case validate_device_authorization_exchange_payload(payload) do
+ :ok -> {attempt, payload}
+ {:error, reason} -> Repo.rollback(reason)
+ end
+ end
+
+ {0, []} ->
+ Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
+ defp complete_device_authorization_locked(user_id, claim, credentials, expires_at) do
+ timestamp = database_now()
+ lock_user!(user_id)
+ provider = owned_integration_provider!(user_id, claim.integration_id)
+ lock_provider_identity(user_id, provider)
+ integration = lock_owned_integration!(user_id, claim.integration_id)
+
+ if !DateTime.after?(expires_at, timestamp),
+ do: Repo.rollback(:device_authorization_response_invalid)
+
+ attempt = lock_claimed_device_authorization!(user_id, integration, claim, timestamp)
+
+ case CredentialEncryption.encrypt(integration, credentials) do
+ {:ok, encrypted} ->
+ integration =
+ integration
+ |> change(
+ Map.merge(encrypted, %{
+ connection_status: "connected",
+ validation_status: "unverified",
+ credential_generation: integration.credential_generation + 1,
+ expires_at: expires_at,
+ validated_at: nil,
+ refreshed_at: timestamp,
+ validation_error_code: nil,
+ updated_at: timestamp
+ })
+ )
+ |> Repo.update!()
+ |> maybe_activate_completed_oauth(user_id)
+
+ attempt
+ |> change(terminal_attempt_changes("completed", nil, timestamp))
+ |> Repo.update!(@secret_repo_options)
+
+ audit!(user_id, integration, "device_authorization_completed")
+ integration
+
+ {:error, reason} ->
+ Repo.rollback(reason)
+ end
+ end
+
+ defp owned_integration_provider!(user_id, integration_id) do
+ case Integration
+ |> where(
+ [integration],
+ integration.id == ^integration_id and integration.user_id == ^user_id
+ )
+ |> select([integration], integration.provider)
+ |> Repo.one() do
+ nil -> Repo.rollback(:integration_not_found)
+ provider -> provider
+ end
+ end
+
+ defp lock_claimed_device_authorization!(user_id, integration, claim, timestamp) do
+ query =
+ DeviceAuthorizationAttempt
+ |> where(
+ [attempt],
+ attempt.id == ^claim.id and attempt.user_id == ^user_id and
+ attempt.integration_id == ^integration.id and
+ attempt.attempt_generation == ^claim.attempt_generation and attempt.state == "active" and
+ attempt.claim_owner_id == ^claim.claim_owner_id and
+ attempt.claim_epoch == ^claim.claim_epoch and
+ attempt.claim_lease_expires_at > ^timestamp and
+ attempt.provider_deadline > ^timestamp and
+ attempt.expected_integration_generation == ^integration.credential_generation
+ )
+ |> lock("FOR UPDATE")
+
+ case Repo.one(query, @secret_repo_options) do
+ %DeviceAuthorizationAttempt{} = attempt -> attempt
+ nil -> Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
+ defp maybe_activate_completed_oauth(integration, user_id) do
+ if integration.active or active_provider_account_exists?(user_id, integration.provider) do
+ integration
+ else
+ integration |> change(active: true, updated_at: now()) |> Repo.update!()
+ end
+ end
+
+ defp terminalize_claimed_device_authorization(user_id, claim, state, error_code) do
+ timestamp = database_now()
+
+ query = claimed_device_authorization_query(user_id, claim, timestamp)
+
+ case Repo.update_all(
+ query,
+ [set: terminal_attempt_changes(state, error_code, timestamp)],
+ @secret_repo_options
+ ) do
+ {1, nil} -> Repo.get!(DeviceAuthorizationAttempt, claim.id, @secret_repo_options)
+ {0, nil} -> Repo.rollback(:stale_device_authorization_claim)
+ end
+ end
+
+ defp claimed_device_authorization_query(user_id, claim, timestamp) do
+ from attempt in DeviceAuthorizationAttempt,
+ join: integration in Integration,
+ on: integration.id == attempt.integration_id and integration.user_id == attempt.user_id,
+ where:
+ attempt.id == ^claim.id and attempt.user_id == ^user_id and
+ attempt.attempt_generation == ^claim.attempt_generation and attempt.state == "active" and
+ attempt.claim_owner_id == ^claim.claim_owner_id and
+ attempt.claim_epoch == ^claim.claim_epoch and
+ attempt.claim_lease_expires_at > ^timestamp and
+ attempt.provider_deadline > ^timestamp and
+ integration.credential_generation == attempt.expected_integration_generation
+ end
+
defp cancel_device_authorization_locked(user_id, attempt_id, attempt_generation) do
integration_id =
DeviceAuthorizationAttempt
@@ -672,11 +941,53 @@ defmodule Kodo.Integrations do
defp validate_device_authorization_payload(_payload),
do: {:error, :device_authorization_invalid}
+ defp validate_device_authorization_exchange_payload(
+ %{
+ "authorization_code" => authorization_code,
+ "code_challenge" => code_challenge,
+ "code_verifier" => code_verifier
+ } = payload
+ )
+ when map_size(payload) == 3 do
+ if Enum.all?([authorization_code, code_challenge, code_verifier], fn value ->
+ bounded_secret?(value, @device_authorization_token_max_bytes)
+ end),
+ do: :ok,
+ else: {:error, :device_authorization_response_invalid}
+ end
+
+ defp validate_device_authorization_exchange_payload(_payload),
+ do: {:error, :device_authorization_response_invalid}
+
+ defp validate_device_authorization_credentials(
+ %{
+ "access_token" => access_token,
+ "refresh_token" => refresh_token,
+ "id_token" => id_token,
+ "account_id" => account_id
+ } = credentials
+ )
+ when map_size(credentials) == 4 do
+ valid_tokens? =
+ Enum.all?([access_token, refresh_token, id_token], fn value ->
+ bounded_secret?(value, @device_authorization_token_max_bytes)
+ end)
+
+ if valid_tokens? and bounded_secret?(account_id),
+ do: :ok,
+ else: {:error, :device_authorization_response_invalid}
+ end
+
+ defp validate_device_authorization_credentials(_credentials),
+ do: {:error, :device_authorization_response_invalid}
+
defp bounded_secret?(value) do
- is_binary(value) and byte_size(value) > 0 and
- byte_size(value) <= @device_authorization_secret_max_bytes
+ bounded_secret?(value, @device_authorization_secret_max_bytes)
end
+ defp bounded_secret?(value, max_bytes),
+ do: is_binary(value) and byte_size(value) > 0 and byte_size(value) <= max_bytes
+
defp require_device_authorization_integration(%Integration{
provider: "openai_codex",
authentication_type: "oauth"
diff --git a/lib/kodo/integrations/audit_event.ex b/lib/kodo/integrations/audit_event.ex
index 665ef20..e78ba87 100644
--- a/lib/kodo/integrations/audit_event.ex
+++ b/lib/kodo/integrations/audit_event.ex
@@ -14,6 +14,8 @@ defmodule Kodo.Integrations.AuditEvent do
integration_disconnected
device_authorization_started
device_authorization_cancelled
+ device_authorization_completed
+ device_authorization_failed
oauth_succeeded
refresh_succeeded
refresh_invalid_grant
diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs
index 0f45936..d95564e 100644
--- a/test/kodo/integrations_test.exs
+++ b/test/kodo/integrations_test.exs
@@ -950,6 +950,234 @@ defmodule Kodo.IntegrationsTest do
assert %{state: "expired", encrypted_payload: nil} = Repo.reload!(attempt)
end
+ test "admits each claimed network phase and persists exchange state encrypted", %{
+ scope: scope
+ } do
+ integration = oauth_integration(scope)
+ owner = Ecto.UUID.generate()
+ polling_payload = %{"device_auth_id" => "device", "user_code" => "CODE"}
+
+ exchange_payload = %{
+ "authorization_code" => "authorization-secret",
+ "code_challenge" => "challenge-secret",
+ "code_verifier" => "verifier-secret"
+ }
+
+ assert {:ok, _attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ polling_payload,
+ 0
+ )
+
+ assert {:ok, {claim, ^polling_payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, owner)
+
+ assert {:ok, {poll_admission, ^polling_payload}} =
+ Integrations.admit_device_authorization_poll(scope, claim)
+
+ assert {:ok, persisted} =
+ Integrations.store_device_authorization_exchange(scope, claim, exchange_payload)
+
+ refute persisted.encrypted_payload =~ "authorization-secret"
+
+ assert {:ok, {exchange_admission, ^exchange_payload}} =
+ Integrations.admit_device_authorization_exchange(scope, claim)
+
+ assert exchange_admission.claim_epoch == claim.claim_epoch
+ assert poll_admission.claim_epoch == claim.claim_epoch
+
+ assert {:ok, failed} =
+ Integrations.fail_device_authorization(scope, claim, "provider_rejected")
+
+ assert failed.state == "failed"
+ assert failed.terminal_error_code == "provider_rejected"
+ assert is_nil(failed.encrypted_payload)
+ assert is_nil(failed.claim_owner_id)
+ end
+
+ test "rejects early polling and stale phase results without changing payload", %{scope: scope} do
+ integration = oauth_integration(scope)
+ first_owner = Ecto.UUID.generate()
+ second_owner = Ecto.UUID.generate()
+ polling_payload = %{"device_auth_id" => "device", "user_code" => "CODE"}
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ polling_payload,
+ 60_000
+ )
+
+ assert {:ok, {first_claim, ^polling_payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, first_owner)
+
+ assert {:error, :device_authorization_poll_not_due} =
+ Integrations.admit_device_authorization_poll(scope, first_claim)
+
+ Repo.update!(
+ change(first_claim,
+ claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)
+ )
+ )
+
+ assert {:ok, {_takeover, ^polling_payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, second_owner)
+
+ exchange_payload = %{
+ "authorization_code" => "stale-secret",
+ "code_challenge" => "challenge",
+ "code_verifier" => "verifier"
+ }
+
+ assert {:error, :stale_device_authorization_claim} =
+ Integrations.store_device_authorization_exchange(
+ scope,
+ first_claim,
+ exchange_payload
+ )
+
+ assert {:ok, {_active, ^polling_payload}} =
+ Integrations.get_active_device_authorization(scope, integration.id)
+
+ assert Repo.reload!(attempt).state == "active"
+ end
+
+ test "rejects unsafe terminal errors and malformed exchange state", %{scope: scope} do
+ integration = oauth_integration(scope)
+
+ assert {:ok, _attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 0
+ )
+
+ assert {:ok, {claim, _payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ assert {:error, :device_authorization_response_invalid} =
+ Integrations.store_device_authorization_exchange(scope, claim, %{
+ "authorization_code" => "code",
+ "code_verifier" => "verifier"
+ })
+
+ assert {:error, :unsafe_device_authorization_error} =
+ Integrations.fail_device_authorization(scope, claim, "private provider detail")
+ end
+
+ test "atomically completes a claimed attempt and installs OAuth credentials", %{scope: scope} do
+ integration = oauth_integration(scope)
+
+ credentials = %{
+ "access_token" => "access-secret",
+ "refresh_token" => "refresh-secret",
+ "id_token" => "identity-secret",
+ "account_id" => "account-secret"
+ }
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 0
+ )
+
+ assert {:ok, {claim, _payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ expires_at = DateTime.add(DateTime.utc_now(), 3_600, :second)
+
+ assert {:ok, completed} =
+ Integrations.complete_device_authorization(scope, claim, credentials, expires_at)
+
+ assert completed.connection_status == "connected"
+ assert completed.validation_status == "unverified"
+ assert completed.credential_generation == claim.expected_integration_generation + 1
+ assert completed.active
+ assert completed.expires_at == expires_at
+ assert {:ok, ^credentials} = CredentialEncryption.decrypt(completed)
+
+ assert %{state: "completed", encrypted_payload: nil, claim_owner_id: nil} =
+ Repo.reload!(attempt)
+
+ assert Enum.map(Integrations.list_audit_events(scope), & &1.event_type) == [
+ "device_authorization_started",
+ "device_authorization_completed"
+ ]
+ end
+
+ test "completion is fenced after claim takeover and rejects expired credentials", %{
+ scope: scope
+ } do
+ integration = oauth_integration(scope)
+ first_owner = Ecto.UUID.generate()
+
+ assert {:ok, _attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 0
+ )
+
+ assert {:ok, {claim, _payload}} =
+ Integrations.claim_device_authorization(scope, integration.id, first_owner)
+
+ credentials = %{
+ "access_token" => "access",
+ "refresh_token" => "refresh",
+ "id_token" => "identity",
+ "account_id" => "account"
+ }
+
+ assert {:error, :device_authorization_response_invalid} =
+ Integrations.complete_device_authorization(
+ scope,
+ claim,
+ credentials,
+ DateTime.add(DateTime.utc_now(), -1, :second)
+ )
+
+ Repo.update!(
+ change(claim, claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second))
+ )
+
+ assert {:ok, {_takeover, _payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ assert {:error, :stale_device_authorization_claim} =
+ Integrations.complete_device_authorization(
+ scope,
+ claim,
+ credentials,
+ DateTime.add(DateTime.utc_now(), 3_600, :second)
+ )
+
+ assert Repo.reload!(integration).connection_status == "disconnected"
+ end
+
test "cancellation is owned and generation fenced and removes the one-time code", %{
scope: scope
} do
From bf3a9527368faa200fb4b30738465889124c179d Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:27:05 +0000
Subject: [PATCH 04/25] fix(integrations): preserve OAuth expiry precision
---
lib/kodo/integrations/device_authorization_tokens.ex | 2 +-
test/kodo/integrations/device_authorization_tokens_test.exs | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/lib/kodo/integrations/device_authorization_tokens.ex b/lib/kodo/integrations/device_authorization_tokens.ex
index 1064c6c..295cd45 100644
--- a/lib/kodo/integrations/device_authorization_tokens.ex
+++ b/lib/kodo/integrations/device_authorization_tokens.ex
@@ -54,7 +54,7 @@ defmodule Kodo.Integrations.DeviceAuthorizationTokens do
defp expiration(%{"exp" => unix}, now) when is_integer(unix) do
with {:ok, expires_at} <- DateTime.from_unix(unix),
true <- DateTime.after?(expires_at, now) do
- {:ok, expires_at}
+ {:ok, %{expires_at | microsecond: {0, 6}}}
else
_invalid -> :error
end
diff --git a/test/kodo/integrations/device_authorization_tokens_test.exs b/test/kodo/integrations/device_authorization_tokens_test.exs
index 92332a8..569f2c2 100644
--- a/test/kodo/integrations/device_authorization_tokens_test.exs
+++ b/test/kodo/integrations/device_authorization_tokens_test.exs
@@ -16,7 +16,7 @@ defmodule Kodo.Integrations.DeviceAuthorizationTokensTest do
}
assert {:ok, normalized} = DeviceAuthorizationTokens.normalize(tokens, @now)
- assert normalized.expires_at == DateTime.add(@now, 3_600, :second)
+ assert DateTime.to_unix(normalized.expires_at) == DateTime.to_unix(@now) + 3_600
assert normalized.credentials ==
Map.put(tokens, "account_id", "account-secret")
From ede247ed267f394823981a1972c53eb0127cebca Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:27:12 +0000
Subject: [PATCH 05/25] feat(integrations): run durable device authorization
---
lib/kodo/integrations/device_authorization.ex | 217 ++++++++++++++++
.../device_authorization_test.exs | 245 ++++++++++++++++++
.../fake_device_authorization_client.ex | 23 ++
3 files changed, 485 insertions(+)
create mode 100644 lib/kodo/integrations/device_authorization.ex
create mode 100644 test/kodo/integrations/device_authorization_test.exs
create mode 100644 test/support/fake_device_authorization_client.ex
diff --git a/lib/kodo/integrations/device_authorization.ex b/lib/kodo/integrations/device_authorization.ex
new file mode 100644
index 0000000..ab3722b
--- /dev/null
+++ b/lib/kodo/integrations/device_authorization.ex
@@ -0,0 +1,217 @@
+defmodule Kodo.Integrations.DeviceAuthorization do
+ @moduledoc "Runs finite, durable ChatGPT device authorization tasks."
+
+ alias Kodo.Accounts.Scope
+ alias Kodo.Cluster.InstanceManager
+ alias Kodo.Integrations
+ alias Kodo.Integrations.DeviceAuthorizationAttempt
+ alias Kodo.Integrations.DeviceAuthorizationTokens
+
+ @lease_renewal_interval_ms 15_000
+ @verification_url "https://auth.openai.com/codex/device"
+
+ def begin(%Scope{} = scope, integration_id, expected_generation, opts \\ []) do
+ client = Keyword.get(opts, :client, configured_client())
+ client_options = Keyword.get(opts, :client_options, [])
+
+ with {:ok, created} <- safe_create(client, client_options),
+ {:ok, attempt} <-
+ Integrations.begin_device_authorization(
+ scope,
+ integration_id,
+ expected_generation,
+ created.payload,
+ created.polling_interval_ms
+ ),
+ {:ok, task} <- resume(scope, integration_id, opts) do
+ {:ok,
+ %{
+ attempt: attempt,
+ task: task,
+ verification_url: created.verification_url
+ }}
+ end
+ end
+
+ def resume(%Scope{} = scope, integration_id, opts \\ []) do
+ claim_owner_id = Keyword.get_lazy(opts, :claim_owner_id, &InstanceManager.current_boot_id/0)
+ supervisor = Keyword.get(opts, :supervisor, Kodo.ControlPlaneTaskSupervisor)
+
+ with owner when is_binary(owner) <- claim_owner_id,
+ {:ok, {claim, payload}} <-
+ Integrations.claim_device_authorization(scope, integration_id, owner) do
+ task =
+ Task.Supervisor.async_nolink(supervisor, fn ->
+ run(scope, claim, payload, opts)
+ end)
+
+ {:ok, task}
+ else
+ nil -> {:error, :device_authorization_owner_unavailable}
+ {:error, _reason} = error -> error
+ end
+ end
+
+ @doc false
+ def run(
+ %Scope{} = scope,
+ %DeviceAuthorizationAttempt{} = claim,
+ payload,
+ opts \\ []
+ ) do
+ client = Keyword.get(opts, :client, configured_client())
+ client_options = Keyword.get(opts, :client_options, [])
+ continue(scope, claim, payload, client, client_options)
+ end
+
+ defp continue(scope, claim, %{"device_auth_id" => _, "user_code" => _}, client, options) do
+ with {:ok, {admission, payload}} <- await_poll_admission(scope, claim),
+ result <- safe_poll(client, payload, options) do
+ case result do
+ :pending ->
+ continue(scope, admission, payload, client, options)
+
+ {:ok, exchange_payload} ->
+ persist_and_exchange(scope, admission, exchange_payload, client, options)
+
+ {:error, reason} ->
+ fail(scope, admission, reason)
+ end
+ else
+ {:error, reason} -> stop(reason)
+ end
+ end
+
+ defp continue(scope, claim, %{"authorization_code" => _} = _payload, client, options) do
+ exchange(scope, claim, client, options)
+ end
+
+ defp continue(scope, claim, _payload, _client, _options) do
+ fail(scope, claim, :device_authorization_response_invalid)
+ end
+
+ defp await_poll_admission(scope, claim) do
+ delay = max(DateTime.diff(claim.next_poll_at, DateTime.utc_now(), :millisecond), 0)
+
+ if delay == 0 do
+ case Integrations.admit_device_authorization_poll(scope, claim) do
+ {:error, :device_authorization_poll_not_due} -> wait_and_renew(scope, claim, 1)
+ result -> result
+ end
+ else
+ wait_and_renew(scope, claim, min(delay, @lease_renewal_interval_ms))
+ end
+ end
+
+ defp wait_and_renew(scope, claim, delay) do
+ receive do
+ after
+ delay ->
+ case Integrations.renew_device_authorization_claim(
+ scope,
+ claim.id,
+ claim.attempt_generation,
+ claim.claim_owner_id,
+ claim.claim_epoch
+ ) do
+ {:ok, {renewed, _payload}} -> await_poll_admission(scope, renewed)
+ {:error, reason} -> {:error, reason}
+ end
+ end
+ end
+
+ defp persist_and_exchange(scope, claim, payload, client, options) do
+ case Integrations.store_device_authorization_exchange(scope, claim, payload) do
+ {:ok, persisted} -> exchange(scope, persisted, client, options)
+ {:error, reason} -> stop(reason)
+ end
+ end
+
+ defp exchange(scope, claim, client, options) do
+ with {:ok, {admission, payload}} <-
+ Integrations.admit_device_authorization_exchange(scope, claim),
+ {:ok, tokens} <- safe_exchange(client, payload, options),
+ {:ok, normalized} <- DeviceAuthorizationTokens.normalize(tokens),
+ {:ok, _integration} <-
+ Integrations.complete_device_authorization(
+ scope,
+ admission,
+ normalized.credentials,
+ normalized.expires_at
+ ) do
+ :ok
+ else
+ {:error, :stale_device_authorization_claim} = error -> stop(error)
+ {:error, reason} -> fail(scope, claim, reason)
+ end
+ end
+
+ defp fail(scope, claim, reason) do
+ case Integrations.fail_device_authorization(scope, claim, terminal_error(reason)) do
+ {:ok, _attempt} -> {:error, reason}
+ {:error, stale_reason} -> stop(stale_reason)
+ end
+ end
+
+ defp stop({:error, reason}), do: {:error, reason}
+ defp stop(reason), do: {:error, reason}
+
+ defp terminal_error(:redirect), do: "redirect"
+ defp terminal_error(:device_authorization_rejected), do: "provider_rejected"
+ defp terminal_error(:device_authorization_response_invalid), do: "response_invalid"
+ defp terminal_error(_reason), do: "provider_unavailable"
+
+ defp safe_create(client, options) do
+ case safe_client_call(fn -> client.create(options) end) do
+ {:ok,
+ %{
+ payload: payload,
+ polling_interval_ms: interval,
+ verification_url: verification_url
+ } = created}
+ when is_map(payload) and is_integer(interval) and verification_url == @verification_url ->
+ {:ok, created}
+
+ {:error, reason} when is_atom(reason) ->
+ {:error, reason}
+
+ _invalid ->
+ {:error, :device_authorization_response_invalid}
+ end
+ end
+
+ defp safe_poll(client, payload, options) do
+ case safe_client_call(fn -> client.poll(payload, options) end) do
+ :pending -> :pending
+ {:ok, exchange_payload} when is_map(exchange_payload) -> {:ok, exchange_payload}
+ {:error, reason} when is_atom(reason) -> {:error, reason}
+ _invalid -> {:error, :device_authorization_response_invalid}
+ end
+ end
+
+ defp safe_exchange(client, payload, options) do
+ case safe_client_call(fn -> client.exchange(payload, options) end) do
+ {:ok, tokens} when is_map(tokens) -> {:ok, tokens}
+ {:error, reason} when is_atom(reason) -> {:error, reason}
+ _invalid -> {:error, :device_authorization_response_invalid}
+ end
+ end
+
+ defp safe_client_call(operation) do
+ try do
+ operation.()
+ rescue
+ _exception -> {:error, :provider_unavailable}
+ catch
+ _kind, _reason -> {:error, :provider_unavailable}
+ end
+ end
+
+ defp configured_client do
+ Application.get_env(
+ :kodo,
+ :device_authorization_client,
+ Kodo.Integrations.ReqDeviceAuthorizationClient
+ )
+ end
+end
diff --git a/test/kodo/integrations/device_authorization_test.exs b/test/kodo/integrations/device_authorization_test.exs
new file mode 100644
index 0000000..f6b6ccb
--- /dev/null
+++ b/test/kodo/integrations/device_authorization_test.exs
@@ -0,0 +1,245 @@
+defmodule Kodo.Integrations.DeviceAuthorizationTest do
+ use Kodo.DataCase, async: false
+
+ alias Kodo.AccountsFixtures
+ alias Kodo.Integrations
+ alias Kodo.Integrations.CredentialEncryption
+ alias Kodo.Integrations.DeviceAuthorization
+ alias Kodo.Integrations.DeviceAuthorizationAttempt
+ alias Kodo.Integrations.Integration
+ alias Kodo.Test.FakeDeviceAuthorizationClient
+
+ setup do
+ scope = AccountsFixtures.user_scope_fixture()
+
+ integration =
+ %Integration{user_id: scope.user.id}
+ |> Integration.create_changeset(%{
+ provider: "openai_codex",
+ authentication_type: "oauth"
+ })
+ |> Repo.insert!()
+
+ %{scope: scope, integration: integration}
+ end
+
+ test "polls until authorized, exchanges once, and installs the tokens", %{
+ scope: scope,
+ integration: integration
+ } do
+ exchange_payload = exchange_payload()
+ tokens = tokens()
+
+ client =
+ fake_client(
+ poll: [:pending, {:ok, exchange_payload}],
+ exchange: [{:ok, tokens}]
+ )
+
+ {claim, payload} = claimed_attempt(scope, integration)
+
+ assert :ok = DeviceAuthorization.run(scope, claim, payload, client_opts(client))
+
+ completed = Repo.reload!(integration)
+ assert completed.connection_status == "connected"
+ assert completed.active
+ assert completed.credential_generation == 2
+ assert {:ok, credentials} = CredentialEncryption.decrypt(completed)
+ assert credentials["access_token"] == tokens["access_token"]
+ assert credentials["refresh_token"] == "refresh-secret"
+ assert credentials["account_id"] == "account-secret"
+
+ assert %{state: "completed", encrypted_payload: nil} = Repo.reload!(claim)
+
+ assert calls(client) == [
+ {:poll, payload},
+ {:poll, payload},
+ {:exchange, exchange_payload}
+ ]
+ end
+
+ test "resumes a persisted authorization code directly at exchange", %{
+ scope: scope,
+ integration: integration
+ } do
+ client = fake_client(exchange: [{:ok, tokens()}])
+ {claim, _payload} = claimed_attempt(scope, integration)
+
+ assert {:ok, persisted} =
+ Integrations.store_device_authorization_exchange(scope, claim, exchange_payload())
+
+ assert {:ok, {persisted, payload}} =
+ Integrations.admit_device_authorization_exchange(scope, persisted)
+
+ assert :ok = DeviceAuthorization.run(scope, persisted, payload, client_opts(client))
+ assert calls(client) == [{:exchange, exchange_payload()}]
+ end
+
+ test "terminal provider errors fail the attempt without changing credentials", %{
+ scope: scope,
+ integration: integration
+ } do
+ client = fake_client(poll: [{:error, :device_authorization_rejected}])
+ {claim, payload} = claimed_attempt(scope, integration)
+
+ assert {:error, :device_authorization_rejected} =
+ DeviceAuthorization.run(scope, claim, payload, client_opts(client))
+
+ assert %{state: "failed", terminal_error_code: "provider_rejected"} = Repo.reload!(claim)
+ assert Repo.reload!(integration).connection_status == "disconnected"
+
+ assert Enum.map(Integrations.list_audit_events(scope), & &1.event_type) == [
+ "device_authorization_started",
+ "device_authorization_failed"
+ ]
+ end
+
+ test "a stale claimant performs no provider operation", %{
+ scope: scope,
+ integration: integration
+ } do
+ client = fake_client(poll: [{:ok, exchange_payload()}])
+ {claim, payload} = claimed_attempt(scope, integration)
+
+ Repo.update!(
+ change(claim, claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second))
+ )
+
+ assert {:ok, {_takeover, _payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ assert {:error, :stale_device_authorization_claim} =
+ DeviceAuthorization.run(scope, claim, payload, client_opts(client))
+
+ assert calls(client) == []
+ end
+
+ test "begin persists the provider response and launches a supervised finite task", %{
+ scope: scope,
+ integration: integration
+ } do
+ supervisor = start_supervised!(Task.Supervisor)
+
+ client =
+ fake_client(
+ create: [
+ {:ok,
+ %{
+ payload: %{"device_auth_id" => "device", "user_code" => "CODE"},
+ polling_interval_ms: 0,
+ verification_url: "https://auth.openai.com/codex/device"
+ }}
+ ],
+ poll: [{:error, :device_authorization_rejected}]
+ )
+
+ assert {:ok, started} =
+ DeviceAuthorization.begin(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ client_opts(client, supervisor: supervisor, claim_owner_id: Ecto.UUID.generate())
+ )
+
+ assert started.verification_url == "https://auth.openai.com/codex/device"
+ assert %Task{} = started.task
+ assert {:error, :device_authorization_rejected} = Task.await(started.task)
+ assert Repo.reload!(started.attempt).state == "failed"
+ end
+
+ test "begin rejects a client-selected verification destination before persistence", %{
+ scope: scope,
+ integration: integration
+ } do
+ client =
+ fake_client(
+ create: [
+ {:ok,
+ %{
+ payload: %{"device_auth_id" => "device", "user_code" => "CODE"},
+ polling_interval_ms: 0,
+ verification_url: "https://attacker.example/device"
+ }}
+ ]
+ )
+
+ assert {:error, :device_authorization_response_invalid} =
+ DeviceAuthorization.begin(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ client_opts(client)
+ )
+
+ refute Repo.exists?(DeviceAuthorizationAttempt)
+ end
+
+ defp claimed_attempt(scope, integration) do
+ assert {:ok, _attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ integration.credential_generation,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 0
+ )
+
+ assert {:ok, claimed} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ claimed
+ end
+
+ defp fake_client(responses) do
+ defaults = %{create: [], poll: [], exchange: []}
+
+ start_supervised!(
+ {Agent, fn -> %{responses: Map.merge(defaults, Map.new(responses)), calls: []} end},
+ id: {:fake_device_client, System.unique_integer([:positive])}
+ )
+ end
+
+ defp client_opts(agent, opts \\ []) do
+ [client: FakeDeviceAuthorizationClient, client_options: [agent: agent]] ++ opts
+ end
+
+ defp calls(agent) do
+ agent
+ |> Agent.get(& &1.calls)
+ |> Enum.reverse()
+ end
+
+ defp exchange_payload do
+ %{
+ "authorization_code" => "authorization-secret",
+ "code_challenge" => "challenge-secret",
+ "code_verifier" => "verifier-secret"
+ }
+ end
+
+ defp tokens do
+ now = DateTime.utc_now()
+
+ %{
+ "access_token" => jwt(%{"exp" => DateTime.to_unix(now) + 3_600}),
+ "refresh_token" => "refresh-secret",
+ "id_token" =>
+ jwt(%{
+ "https://api.openai.com/auth" => %{"chatgpt_account_id" => "account-secret"}
+ })
+ }
+ end
+
+ defp jwt(claims) do
+ encoded = claims |> Jason.encode!() |> Base.url_encode64(padding: false)
+ "header.#{encoded}.signature"
+ end
+end
diff --git a/test/support/fake_device_authorization_client.ex b/test/support/fake_device_authorization_client.ex
new file mode 100644
index 0000000..5feffcd
--- /dev/null
+++ b/test/support/fake_device_authorization_client.ex
@@ -0,0 +1,23 @@
+defmodule Kodo.Test.FakeDeviceAuthorizationClient do
+ @moduledoc false
+
+ def create(opts), do: respond(opts, :create, nil)
+ def poll(payload, opts), do: respond(opts, :poll, payload)
+ def exchange(payload, opts), do: respond(opts, :exchange, payload)
+
+ defp respond(opts, operation, payload) do
+ agent = Keyword.fetch!(opts, :agent)
+
+ Agent.get_and_update(agent, fn state ->
+ [response | remaining] = Map.fetch!(state.responses, operation)
+
+ new_state = %{
+ state
+ | calls: [{operation, payload} | state.calls],
+ responses: Map.put(state.responses, operation, remaining)
+ }
+
+ {response, new_state}
+ end)
+ end
+end
From 824431a386e0a282a008906c8bc04c423cd7b745 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:28:43 +0000
Subject: [PATCH 06/25] docs(rfd-5): track device authorization worker
---
rfd/0005/IMPLEMENTATION.org | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/rfd/0005/IMPLEMENTATION.org b/rfd/0005/IMPLEMENTATION.org
index 145affb..8afd3f8 100644
--- a/rfd/0005/IMPLEMENTATION.org
+++ b/rfd/0005/IMPLEMENTATION.org
@@ -321,7 +321,7 @@ manufactured after that no-go decision.
binding attempt ID, integration ID, user ID, provider, immutable attempt
generation, and payload-format version; fail closed on swaps, tampering,
unknown formats, and unavailable keys.
-- [ ] Implement device-code creation, interval-aware asynchronous polling,
+- [X] Implement device-code creation, interval-aware asynchronous polling,
cancellation, expiry, terminal errors, and authorization-code exchange using
Req rather than embedded browser scripts.
- [ ] Render the fixed allowlisted HTTPS verification destination and one-time
@@ -332,13 +332,13 @@ manufactured after that no-go decision.
returns its claim epoch, and run a successful claim as a finite task under
~Kodo.ControlPlaneTaskSupervisor~ that survives the originating LiveView
closing.
-- [ ] Conditionally renew the claim lease before expiry and verify active state,
+- [X] Conditionally renew the claim lease before expiry and verify active state,
current claim epoch, and lease ownership before admitting every poll or token
exchange. Make its duration exceed the hard HTTP timeout plus a safety margin.
Stop future work immediately on ownership loss; fence completion by claim
epoch plus attempt and expected integration generations, while allowing an
already admitted network operation to finish without persisting stale results.
-- [ ] Do not run a device-polling sweeper on every replica; expire attempts
+- [X] Do not run a device-polling sweeper on every replica; expire attempts
lazily and delete terminal rows in bounded batches, using one database-elected
cleanup owner only if periodic cleanup becomes necessary.
- [ ] Make a new attempt atomically supersede the prior attempt and fence
@@ -346,7 +346,7 @@ manufactured after that no-go decision.
generation. Starting reauthorization advances the integration generation and
the new attempt captures that post-advance value; install OAuth credentials
only when both attempt and expected integration generations still match.
-- [ ] Store access tokens, refresh tokens, account identifiers, expiries, and
+- [X] Store access tokens, refresh tokens, account identifiers, expiries, and
credential generations in the encrypted integration payload without logging
device or token material.
- [ ] Test user deletion with OAuth credentials, an active authorization
From 970eef3c646ea9a36ee5e9036c3b3beed6ce7f02 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:30:07 +0000
Subject: [PATCH 07/25] test(integrations): fence completed device polls
---
.../device_authorization_test.exs | 44 +++++++++++++++++++
.../fake_device_authorization_client.ex | 21 +++++----
2 files changed, 56 insertions(+), 9 deletions(-)
diff --git a/test/kodo/integrations/device_authorization_test.exs b/test/kodo/integrations/device_authorization_test.exs
index f6b6ccb..7003fd9 100644
--- a/test/kodo/integrations/device_authorization_test.exs
+++ b/test/kodo/integrations/device_authorization_test.exs
@@ -118,6 +118,50 @@ defmodule Kodo.Integrations.DeviceAuthorizationTest do
assert calls(client) == []
end
+ test "a poll admitted before takeover may finish but cannot persist its result", %{
+ scope: scope,
+ integration: integration
+ } do
+ owner = self()
+
+ blocking_poll = fn :poll, _payload ->
+ send(owner, {:poll_admitted, self()})
+
+ receive do
+ :finish_poll -> {:ok, exchange_payload()}
+ end
+ end
+
+ client = fake_client(poll: [blocking_poll])
+ {claim, payload} = claimed_attempt(scope, integration)
+ supervisor = start_supervised!(Task.Supervisor)
+
+ task =
+ Task.Supervisor.async_nolink(supervisor, fn ->
+ DeviceAuthorization.run(scope, claim, payload, client_opts(client))
+ end)
+
+ assert_receive {:poll_admitted, poller}
+
+ Repo.update!(
+ change(claim, claim_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second))
+ )
+
+ assert {:ok, {takeover, ^payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ send(poller, :finish_poll)
+
+ assert {:error, :stale_device_authorization_claim} = Task.await(task)
+ assert Repo.reload!(takeover).state == "active"
+ assert Repo.reload!(integration).connection_status == "disconnected"
+ assert calls(client) == [{:poll, payload}]
+ end
+
test "begin persists the provider response and launches a supervised finite task", %{
scope: scope,
integration: integration
diff --git a/test/support/fake_device_authorization_client.ex b/test/support/fake_device_authorization_client.ex
index 5feffcd..d38e5be 100644
--- a/test/support/fake_device_authorization_client.ex
+++ b/test/support/fake_device_authorization_client.ex
@@ -8,16 +8,19 @@ defmodule Kodo.Test.FakeDeviceAuthorizationClient do
defp respond(opts, operation, payload) do
agent = Keyword.fetch!(opts, :agent)
- Agent.get_and_update(agent, fn state ->
- [response | remaining] = Map.fetch!(state.responses, operation)
+ response =
+ Agent.get_and_update(agent, fn state ->
+ [response | remaining] = Map.fetch!(state.responses, operation)
- new_state = %{
- state
- | calls: [{operation, payload} | state.calls],
- responses: Map.put(state.responses, operation, remaining)
- }
+ new_state = %{
+ state
+ | calls: [{operation, payload} | state.calls],
+ responses: Map.put(state.responses, operation, remaining)
+ }
- {response, new_state}
- end)
+ {response, new_state}
+ end)
+
+ if is_function(response, 2), do: response.(operation, payload), else: response
end
end
From a9299ae551e998fb92c829001bb36ddadbf1dc91 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:31:25 +0000
Subject: [PATCH 08/25] fix(integrations): serialize authorization terminal
states
---
lib/kodo/integrations.ex | 3 +-
test/kodo/integrations_concurrency_test.exs | 57 +++++++++++++++++++++
2 files changed, 59 insertions(+), 1 deletion(-)
diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex
index 2a8484b..3fd7f31 100644
--- a/lib/kodo/integrations.ex
+++ b/lib/kodo/integrations.ex
@@ -233,8 +233,9 @@ defmodule Kodo.Integrations do
)
when error_code in @device_authorization_terminal_errors do
Repo.transaction(fn ->
+ lock_user!(user.id)
+ integration = lock_owned_integration!(user.id, claim.integration_id)
attempt = terminalize_claimed_device_authorization(user.id, claim, "failed", error_code)
- integration = Repo.get!(Integration, attempt.integration_id)
audit!(user.id, integration, "device_authorization_failed")
attempt
end)
diff --git a/test/kodo/integrations_concurrency_test.exs b/test/kodo/integrations_concurrency_test.exs
index 5571403..bf4e3a7 100644
--- a/test/kodo/integrations_concurrency_test.exs
+++ b/test/kodo/integrations_concurrency_test.exs
@@ -193,6 +193,63 @@ defmodule Kodo.IntegrationsConcurrencyTest do
end
end
+ test "device authorization failure and cancellation use one lock order without deadlocks", %{
+ scope: scope,
+ supervisor: supervisor
+ } do
+ for iteration <- 1..10 do
+ integration =
+ %Integration{user_id: scope.user.id}
+ |> Integration.create_changeset(%{
+ provider: "openai_codex",
+ authentication_type: "oauth",
+ display_name: "Failure race #{iteration}"
+ })
+ |> Repo.insert!()
+
+ assert {:ok, attempt} =
+ Integrations.begin_device_authorization(
+ scope,
+ integration.id,
+ 0,
+ %{"device_auth_id" => "device", "user_code" => "CODE"},
+ 0
+ )
+
+ assert {:ok, {claim, _payload}} =
+ Integrations.claim_device_authorization(
+ scope,
+ integration.id,
+ Ecto.UUID.generate()
+ )
+
+ cancel =
+ contended_task(supervisor, fn ->
+ Integrations.cancel_device_authorization(
+ scope,
+ attempt.id,
+ attempt.attempt_generation
+ )
+ end)
+
+ fail =
+ contended_task(supervisor, fn ->
+ Integrations.fail_device_authorization(scope, claim, "provider_rejected")
+ end)
+
+ results = [cancel, fail] |> release_contenders() |> Task.await_many()
+
+ assert Enum.count(results, &match?({:ok, %DeviceAuthorizationAttempt{}}, &1)) == 1
+
+ assert Enum.any?(results, fn result ->
+ result in [
+ {:error, :stale_device_authorization},
+ {:error, :stale_device_authorization_claim}
+ ]
+ end)
+ end
+ end
+
defp connect(scope, name) do
Integrations.connect(scope, "openai", "api_key", %{"api_key" => "#{name}-secret"},
display_name: name
From be3b41a44e6f9c6c45d24e913a3c8a8223e2c1a4 Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:32:45 +0000
Subject: [PATCH 09/25] feat(integrations): add ChatGPT subscription accounts
---
lib/kodo/integrations.ex | 28 ++++++++++++++++++++++++++++
test/kodo/integrations_test.exs | 22 ++++++++++++++++++++++
2 files changed, 50 insertions(+)
diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex
index 3fd7f31..f52226c 100644
--- a/lib/kodo/integrations.ex
+++ b/lib/kodo/integrations.ex
@@ -279,6 +279,34 @@ defmodule Kodo.Integrations do
def cleanup_device_authorizations(_limit), do: {:error, :cleanup_limit_invalid}
+ def create_oauth_integration(scope, provider, opts \\ [])
+
+ def create_oauth_integration(%Scope{user: user}, "openai_codex", opts) do
+ changeset =
+ %Integration{user_id: user.id}
+ |> Integration.create_changeset(%{
+ provider: "openai_codex",
+ authentication_type: "oauth",
+ display_name: opts[:display_name]
+ })
+
+ if changeset.valid? do
+ Repo.transaction(fn ->
+ lock_provider_identity(user.id, "openai_codex")
+
+ case Repo.insert(changeset) do
+ {:ok, integration} -> integration
+ {:error, reason} -> Repo.rollback(reason)
+ end
+ end)
+ else
+ {:error, changeset}
+ end
+ end
+
+ def create_oauth_integration(%Scope{}, _provider, _opts),
+ do: {:error, :authentication_type_mismatch}
+
def connect(scope, provider, authentication_type, credentials, opts \\ [])
def connect(%Scope{user: user}, provider, "api_key", credentials, opts) do
diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs
index d95564e..9780eed 100644
--- a/test/kodo/integrations_test.exs
+++ b/test/kodo/integrations_test.exs
@@ -600,6 +600,28 @@ defmodule Kodo.IntegrationsTest do
%{scope: AccountsFixtures.user_scope_fixture()}
end
+ test "creates several named disconnected ChatGPT subscription accounts", %{scope: scope} do
+ assert {:ok, first} =
+ Integrations.create_oauth_integration(scope, "openai_codex",
+ display_name: "Personal subscription"
+ )
+
+ assert {:ok, second} =
+ Integrations.create_oauth_integration(scope, "openai_codex",
+ display_name: "Work subscription"
+ )
+
+ assert first.authentication_type == "oauth"
+ assert first.connection_status == "disconnected"
+ refute first.active
+ refute second.active
+
+ assert Enum.map(Integrations.list_integrations(scope), & &1.display_name) == [
+ "Personal subscription",
+ "Work subscription"
+ ]
+ end
+
test "starts and reads an encrypted, generation-fenced attempt", %{scope: scope} do
integration = oauth_integration(scope)
payload = %{"device_auth_id" => "device-secret", "user_code" => "ABCD-EFGH"}
From a37bb9671940972b001c41996949ff8b29f24dfc Mon Sep 17 00:00:00 2001
From: Darwin Wu
Date: Thu, 10 Sep 2026 04:43:07 +0000
Subject: [PATCH 10/25] feat(integrations): add ChatGPT device authorization UI
---
lib/kodo_web/live/integrations_live.ex | 379 ++++++++++++++++--
test/kodo_web/live/integrations_live_test.exs | 234 ++++++++++-
.../fake_device_authorization_client.ex | 4 +-
3 files changed, 577 insertions(+), 40 deletions(-)
diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex
index 3c7186e..1487bb0 100644
--- a/lib/kodo_web/live/integrations_live.ex
+++ b/lib/kodo_web/live/integrations_live.ex
@@ -3,6 +3,7 @@ defmodule KodoWeb.IntegrationsLive do
alias Kodo.Integrations
alias Kodo.Integrations.APIKeyValidation
+ alias Kodo.Integrations.DeviceAuthorization
@provider_configs [
%{
@@ -28,10 +29,19 @@ defmodule KodoWeb.IntegrationsLive do
description: "Use an OpenRouter API key for models billed through OpenRouter.",
key_label: "OpenRouter API key",
revoke_url: "https://openrouter.ai/settings/keys"
+ },
+ %{
+ id: "openai_codex",
+ name: "ChatGPT Subscription",
+ badge: "Subscription billing · Beta",
+ description:
+ "Use an eligible ChatGPT subscription through OpenAI's device authorization flow.",
+ key_label: nil,
+ revoke_url: "https://chatgpt.com/#settings/Security"
}
]
@providers Enum.map(@provider_configs, & &1.id)
- @actions ~w(connect replace disconnect)
+ @actions ~w(connect replace reauthorize disconnect)
@max_api_key_bytes 4_096
@impl true
@@ -40,18 +50,22 @@ defmodule KodoWeb.IntegrationsLive do
Phoenix.PubSub.subscribe(Kodo.PubSub, "integration:#{socket.assigns.current_scope.user.id}")
end
- {:ok,
- socket
- |> assign(:action, nil)
- |> assign(:action_provider, nil)
- |> assign(:action_target, nil)
- |> assign(:modal_token, nil)
- |> assign(:modal_error, nil)
- |> assign(:provider_configs, @provider_configs)
- |> assign(:max_api_key_bytes, @max_api_key_bytes)
- |> assign(:api_key_form, empty_form())
- |> assign(:validation_tasks, %{})
- |> load_integrations()}
+ socket =
+ socket
+ |> assign(:action, nil)
+ |> assign(:action_provider, nil)
+ |> assign(:action_target, nil)
+ |> assign(:modal_token, nil)
+ |> assign(:modal_error, nil)
+ |> assign(:provider_configs, @provider_configs)
+ |> assign(:max_api_key_bytes, @max_api_key_bytes)
+ |> assign(:api_key_form, empty_form())
+ |> assign(:validation_tasks, %{})
+ |> assign(:device_authorization_tasks, %{})
+ |> load_integrations()
+
+ socket = if connected?(socket), do: resume_device_authorizations(socket), else: socket
+ {:ok, socket}
end
@impl true
@@ -197,6 +211,70 @@ defmodule KodoWeb.IntegrationsLive do
def handle_event("check_access", _params, socket), do: stale_action(socket)
+ def handle_event("start_device_authorization", %{"integration" => params}, socket) do
+ with true <- socket.assigns.action in ~w(connect reauthorize),
+ true <- socket.assigns.action_provider == "openai_codex",
+ true <- modal_token_valid?(socket, params),
+ {:ok, integration} <- device_authorization_target(socket, params),
+ {:ok, started} <-
+ DeviceAuthorization.begin(
+ socket.assigns.current_scope,
+ integration.id,
+ integration.credential_generation
+ ) do
+ {:noreply,
+ socket
+ |> track_device_authorization_task(started.task, integration.id)
+ |> put_flash(:info, "Authorization started. Enter the one-time code with OpenAI.")
+ |> push_patch(to: ~p"/integrations")}
+ else
+ false ->
+ stale_action(socket)
+
+ {:error, %Ecto.Changeset{}} ->
+ {:noreply,
+ assign(socket, :modal_error, "Enter an account name between 1 and 80 characters.")}
+
+ {:error, :stale_credential_generation} ->
+ stale_action(socket)
+
+ {:error, _reason} ->
+ {:noreply,
+ socket
+ |> put_flash(:error, "Authorization could not be started. Try again.")
+ |> push_patch(to: ~p"/integrations")}
+ end
+ end
+
+ def handle_event("start_device_authorization", _params, socket), do: stale_action(socket)
+
+ def handle_event(
+ "cancel_device_authorization",
+ %{"attempt" => attempt_id, "generation" => generation},
+ socket
+ ) do
+ with {attempt_generation, ""} <- Integer.parse(generation),
+ %{attempt_id: ^attempt_id, attempt_generation: ^attempt_generation} <-
+ Enum.find_value(socket.assigns.device_authorizations, fn {_integration_id, attempt} ->
+ if attempt.attempt_id == attempt_id, do: attempt
+ end),
+ {:ok, _attempt} <-
+ Integrations.cancel_device_authorization(
+ socket.assigns.current_scope,
+ attempt_id,
+ attempt_generation
+ ) do
+ {:noreply,
+ socket
+ |> load_integrations()
+ |> put_flash(:info, "Authorization cancelled.")}
+ else
+ _reason -> stale_action(socket)
+ end
+ end
+
+ def handle_event("cancel_device_authorization", _params, socket), do: stale_action(socket)
+
def handle_event("activate", %{"integration" => id, "generation" => generation}, socket) do
socket = load_integrations(socket)
@@ -240,19 +318,30 @@ defmodule KodoWeb.IntegrationsLive do
@impl true
def handle_info({reference, _result}, socket) when is_reference(reference) do
- if Map.has_key?(socket.assigns.validation_tasks, reference) do
- Process.demonitor(reference, [:flush])
- {:noreply, socket |> drop_validation_task(reference) |> load_integrations()}
- else
- {:noreply, socket}
+ cond do
+ Map.has_key?(socket.assigns.validation_tasks, reference) ->
+ Process.demonitor(reference, [:flush])
+ {:noreply, socket |> drop_validation_task(reference) |> load_integrations()}
+
+ Map.has_key?(socket.assigns.device_authorization_tasks, reference) ->
+ Process.demonitor(reference, [:flush])
+ {:noreply, socket |> drop_device_authorization_task(reference) |> load_integrations()}
+
+ true ->
+ {:noreply, socket}
end
end
def handle_info({:DOWN, reference, :process, _pid, _reason}, socket) do
- if Map.has_key?(socket.assigns.validation_tasks, reference) do
- {:noreply, socket |> drop_validation_task(reference) |> load_integrations()}
- else
- {:noreply, socket}
+ cond do
+ Map.has_key?(socket.assigns.validation_tasks, reference) ->
+ {:noreply, socket |> drop_validation_task(reference) |> load_integrations()}
+
+ Map.has_key?(socket.assigns.device_authorization_tasks, reference) ->
+ {:noreply, socket |> drop_device_authorization_task(reference) |> load_integrations()}
+
+ true ->
+ {:noreply, socket}
end
end
@@ -339,7 +428,11 @@ defmodule KodoWeb.IntegrationsLive do
|> Enum.filter(&(&1.provider in @providers))
|> Enum.map(&integration_metadata/1)
- assign(socket, :integrations, integrations)
+ device_authorizations = load_device_authorizations(socket.assigns.current_scope, integrations)
+
+ socket
+ |> assign(:integrations, integrations)
+ |> assign(:device_authorizations, device_authorizations)
end
# Browser-facing state needs lifecycle metadata only. In particular, keeping
@@ -348,6 +441,7 @@ defmodule KodoWeb.IntegrationsLive do
Map.take(integration, [
:id,
:provider,
+ :authentication_type,
:display_name,
:active,
:connection_status,
@@ -371,6 +465,63 @@ defmodule KodoWeb.IntegrationsLive do
update(socket, :validation_tasks, &Map.delete(&1, reference))
end
+ defp track_device_authorization_task(socket, task, integration_id) do
+ update(socket, :device_authorization_tasks, fn tasks ->
+ Map.put(tasks, task.ref, integration_id)
+ end)
+ end
+
+ defp drop_device_authorization_task(socket, reference) do
+ update(socket, :device_authorization_tasks, &Map.delete(&1, reference))
+ end
+
+ defp resume_device_authorizations(socket) do
+ Enum.reduce(socket.assigns.device_authorizations, socket, fn {integration_id, _attempt},
+ acc ->
+ case DeviceAuthorization.resume(acc.assigns.current_scope, integration_id) do
+ {:ok, task} -> track_device_authorization_task(acc, task, integration_id)
+ {:error, _reason} -> acc
+ end
+ end)
+ end
+
+ defp load_device_authorizations(scope, integrations) do
+ integrations
+ |> Enum.filter(&(&1.provider == "openai_codex"))
+ |> Enum.reduce(%{}, fn integration, attempts ->
+ case Integrations.get_active_device_authorization(scope, integration.id) do
+ {:ok, {attempt, %{"user_code" => user_code}}} ->
+ Map.put(attempts, integration.id, %{
+ attempt_id: attempt.id,
+ attempt_generation: attempt.attempt_generation,
+ user_code: user_code,
+ provider_deadline: attempt.provider_deadline
+ })
+
+ _missing_or_exchanging ->
+ attempts
+ end
+ end)
+ end
+
+ defp device_authorization_target(
+ %{assigns: %{action_target: :new, current_scope: scope}},
+ params
+ ) do
+ Integrations.create_oauth_integration(scope, "openai_codex",
+ display_name: Map.get(params, "display_name", "")
+ )
+ end
+
+ defp device_authorization_target(
+ %{assigns: %{action_target: %{provider: "openai_codex"} = integration}},
+ _params
+ ),
+ do: {:ok, integration}
+
+ defp device_authorization_target(_socket, _params),
+ do: {:error, :stale_credential_generation}
+
defp validate_api_key(api_key)
when is_binary(api_key) and byte_size(api_key) > 0 and
byte_size(api_key) <= @max_api_key_bytes,
@@ -395,15 +546,28 @@ defmodule KodoWeb.IntegrationsLive do
defp action_target("connect", %{connection_status: "disconnected"} = integration),
do: {:ok, target_metadata(integration)}
- defp action_target(action, %{connection_status: "connected"} = integration)
- when action in ~w(replace disconnect),
+ defp action_target(
+ "reauthorize",
+ %{provider: "openai_codex", connection_status: "connected"} = integration
+ ),
do: {:ok, target_metadata(integration)}
+ defp action_target(
+ "replace",
+ %{authentication_type: "api_key", connection_status: "connected"} = integration
+ ),
+ do: {:ok, target_metadata(integration)}
+
+ defp action_target("disconnect", %{connection_status: "connected"} = integration),
+ do: {:ok, target_metadata(integration)}
+
defp action_target(_action, _integration), do: :error
defp target_metadata(integration) do
Map.take(integration, [
:id,
+ :provider,
+ :authentication_type,
:display_name,
:credential_generation,
:connection_status,
@@ -640,6 +804,73 @@ defmodule KodoWeb.IntegrationsLive do
>
{validation_detail(integration)}
+
+
+ Waiting for OpenAI authorization
+
+
+
+
Open this verification page
+ <.link
+ id={dom_id(integration, "device-verification-link")}
+ href="https://auth.openai.com/codex/device"
+ target="_blank"
+ rel="noopener noreferrer"
+ class="mt-1 block break-all text-sm font-semibold text-sky-800 underline decoration-sky-300 underline-offset-2 hover:text-sky-950 dark:text-sky-300 dark:hover:text-sky-200"
+ >
+ https://auth.openai.com/codex/device
+
(opens in a new tab)
+
+
+
+
One-time code
+
+
+ {authorization.user_code}
+
+
+
+
+
+
+
+ Keep this page open. This attempt expires at .
+
+
+
+
@@ -651,7 +882,7 @@ defmodule KodoWeb.IntegrationsLive do
phx-click={JS.push_focus()}
class="rounded-xl bg-zinc-950 px-3.5 py-2 text-sm font-semibold text-white transition hover:bg-zinc-800 dark:bg-white dark:text-zinc-950 dark:hover:bg-zinc-200"
>
- Reconnect
+ {if(integration.provider == "openai_codex", do: "Authorize", else: "Reconnect")}