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")} + + + +
<% provider = provider_config(@action_provider) %>

@@ -823,7 +1127,7 @@ defmodule KodoWeb.IntegrationsLive do rel="noopener noreferrer" class="font-semibold text-red-800 underline decoration-red-300 underline-offset-2 hover:text-red-950 dark:text-red-300 dark:hover:text-red-200" > - Revoke the key in {provider.name}(opens in a new tab) + {revoke_action_label(@action_target)} in {provider.name}(opens in a new tab) if it must stop outside Kodo.

@@ -852,6 +1156,17 @@ defmodule KodoWeb.IntegrationsLive do
+ """ end @@ -859,5 +1174,9 @@ defmodule KodoWeb.IntegrationsLive do defp modal_title("connect", :new), do: "Add provider account" defp modal_title("connect", target), do: "Reconnect #{target.display_name}" defp modal_title("replace", target), do: "Replace key for #{target.display_name}" + defp modal_title("reauthorize", target), do: "Reauthorize #{target.display_name}" defp modal_title("disconnect", target), do: "Disconnect #{target.display_name}?" + + defp revoke_action_label(%{authentication_type: "oauth"}), do: "Revoke account access" + defp revoke_action_label(_target), do: "Revoke the key" end diff --git a/test/kodo_web/live/integrations_live_test.exs b/test/kodo_web/live/integrations_live_test.exs index 8ceaebe..c61266d 100644 --- a/test/kodo_web/live/integrations_live_test.exs +++ b/test/kodo_web/live/integrations_live_test.exs @@ -4,9 +4,12 @@ defmodule KodoWeb.IntegrationsLiveTest do import Kodo.AccountsFixtures import Phoenix.LiveViewTest + alias Kodo.Cluster.InstanceManager alias Kodo.Integrations alias Kodo.Integrations.CredentialEncryption + alias Kodo.Integrations.DeviceAuthorizationAttempt alias Kodo.Repo + alias Kodo.Test.FakeDeviceAuthorizationClient setup %{conn: conn} do user = user_fixture() @@ -24,7 +27,12 @@ defmodule KodoWeb.IntegrationsLiveTest do assert has_element?(view, "#add-integration-menu") for {provider, name} <- - [{"openai", "OpenAI API"}, {"anthropic", "Anthropic"}, {"openrouter", "OpenRouter"}] do + [ + {"openai", "OpenAI API"}, + {"anthropic", "Anthropic"}, + {"openrouter", "OpenRouter"}, + {"openai_codex", "ChatGPT Subscription"} + ] do assert has_element?(view, "#add-#{provider}", name) end end @@ -87,20 +95,181 @@ defmodule KodoWeb.IntegrationsLiveTest do assert error =~ "changed in another session" end - test "stored providers without settings support do not crash the page", %{ + test "renders stored ChatGPT subscription accounts", %{ conn: conn, scope: scope } do - %Kodo.Integrations.Integration{user_id: scope.user.id} - |> Kodo.Integrations.Integration.create_changeset(%{ - provider: "openai_codex", - authentication_type: "oauth" - }) - |> Kodo.Repo.insert!() + {:ok, integration} = + Integrations.create_oauth_integration(scope, "openai_codex", + display_name: "Personal subscription" + ) assert {:ok, view, _html} = live(conn, ~p"/integrations") assert has_element?(view, "#settings-shell") - assert has_element?(view, "#integrations-empty") + assert has_element?(view, card(integration), "Personal subscription") + assert has_element?(view, "#integration-#{integration.id}-reconnect", "Authorize") + end + + test "starts, renders, copies, and cancels ChatGPT device authorization", %{ + conn: conn, + scope: scope + } do + owner = self() + + blocking_poll = fn :poll, _payload -> + send(owner, {:live_device_poll_started, self()}) + + receive do + :finish_poll -> :pending + end + end + + _client = + configure_device_client( + create: [ + {:ok, + %{ + payload: %{"device_auth_id" => "private-device", "user_code" => "ABCD-EFGH"}, + polling_interval_ms: 0, + verification_url: "https://auth.openai.com/codex/device" + }} + ], + poll: [blocking_poll] + ) + + {:ok, view, _html} = live(conn, new_path("openai_codex")) + assert has_element?(view, "#device-authorization-form") + refute has_element?(view, "#device-authorization-form input[type='password']") + + view + |> form("#device-authorization-form", %{ + "integration" => %{"display_name" => "Personal subscription"} + }) + |> render_submit() + + assert_receive {:live_device_poll_started, poller} + [integration] = Integrations.list_integrations(scope) + panel = "#integration-#{integration.id}-device-authorization" + + assert has_element?(view, panel, "https://auth.openai.com/codex/device") + assert has_element?(view, panel, "ABCD-EFGH") + + assert has_element?( + view, + "#integration-#{integration.id}-copy-device-code[data-copy-text='ABCD-EFGH'][aria-label='Copy one-time code']" + ) + + refute inspect(:sys.get_state(view.pid)) =~ "private-device" + + render_click(view, "cancel_device_authorization", %{ + "attempt" => Ecto.UUID.generate(), + "generation" => "1" + }) + + assert Repo.one!(DeviceAuthorizationAttempt).state == "active" + assert render(view) =~ "changed in another session" + + view + |> element("#integration-#{integration.id}-cancel-device-authorization") + |> render_click() + + refute has_element?(view, panel) + assert Repo.one!(DeviceAuthorizationAttempt).state == "cancelled" + + monitor = Process.monitor(poller) + send(poller, :finish_poll) + assert_receive {:DOWN, ^monitor, :process, ^poller, :normal} + end + + test "device authorization uses the normal aged authenticated boundary", %{ + scope: scope, + user: user + } do + blocking_poll = fn :poll, _payload -> + receive do + :finish_poll -> :pending + end + end + + _client = + configure_device_client( + create: [ + {:ok, + %{ + payload: %{"device_auth_id" => "device", "user_code" => "CODE"}, + polling_interval_ms: 0, + verification_url: "https://auth.openai.com/codex/device" + }} + ], + poll: [blocking_poll] + ) + + conn = + build_conn() + |> log_in_user(user, + token_authenticated_at: DateTime.add(DateTime.utc_now(:second), -11, :minute) + ) + + {:ok, view, _html} = live(conn, new_path("openai_codex")) + + view + |> form("#device-authorization-form", %{ + "integration" => %{"display_name" => "Aged session"} + }) + |> render_submit() + + assert [%{provider: "openai_codex"}] = Integrations.list_integrations(scope) + end + + test "shows completion and reauthorization after the device flow succeeds", %{ + conn: conn, + scope: scope + } do + Phoenix.PubSub.subscribe(Kodo.PubSub, "integration:#{scope.user.id}") + + _client = + configure_device_client( + create: [ + {:ok, + %{ + payload: %{"device_auth_id" => "device", "user_code" => "CODE"}, + polling_interval_ms: 0, + verification_url: "https://auth.openai.com/codex/device" + }} + ], + poll: [ + {:ok, + %{ + "authorization_code" => "authorization", + "code_challenge" => "challenge", + "code_verifier" => "verifier" + }} + ], + exchange: [{:ok, device_tokens()}] + ) + + {:ok, view, _html} = live(conn, new_path("openai_codex")) + + view + |> form("#device-authorization-form", %{ + "integration" => %{"display_name" => "Completed subscription"} + }) + |> render_submit() + + assert_receive {:integration_changed, integration_id, _generation} + _ = :sys.get_state(view.pid) + assert {:ok, integration} = Integrations.get_integration(scope, integration_id) + assert integration.connection_status == "connected" + assert has_element?(view, card(integration), "Completed subscription") + assert has_element?(view, "#integration-#{integration.id}-reauthorize", "Reauthorize") + refute has_element?(view, "#integration-#{integration.id}-device-authorization") + + view + |> element("#integration-#{integration.id}-reauthorize") + |> render_click() + + assert has_element?(view, "#device-authorization-form") + assert has_element?(view, "#integration-modal-title", "Reauthorize Completed subscription") end test "adds multiple accounts for one provider without exposing secrets", %{ @@ -487,6 +656,53 @@ defmodule KodoWeb.IntegrationsLiveTest do defp activate_button(integration), do: "#integration-#{integration.id}-activate" defp check_button(integration), do: "#integration-#{integration.id}-check-access" + defp configure_device_client(responses) do + start_supervised!( + {InstanceManager, + enabled: true, boot_id: Ecto.UUID.generate(), heartbeat_interval: :infinity} + ) + + agent = + start_supervised!( + {Agent, + fn -> + %{ + responses: + %{create: [], poll: [], exchange: []} + |> Map.merge(Map.new(responses)), + calls: [] + } + end} + ) + + Application.put_env(:kodo, :device_authorization_client, FakeDeviceAuthorizationClient) + Application.put_env(:kodo, :fake_device_authorization_client_agent, agent) + + on_exit(fn -> + Application.delete_env(:kodo, :device_authorization_client) + Application.delete_env(:kodo, :fake_device_authorization_client_agent) + end) + + agent + end + + defp device_tokens do + %{ + "access_token" => + jwt(%{"exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.+(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 + defp live_assign(view, name) do :sys.get_state(view.pid).socket.assigns[name] end diff --git a/test/support/fake_device_authorization_client.ex b/test/support/fake_device_authorization_client.ex index d38e5be..1a26c9f 100644 --- a/test/support/fake_device_authorization_client.ex +++ b/test/support/fake_device_authorization_client.ex @@ -6,7 +6,9 @@ defmodule Kodo.Test.FakeDeviceAuthorizationClient do def exchange(payload, opts), do: respond(opts, :exchange, payload) defp respond(opts, operation, payload) do - agent = Keyword.fetch!(opts, :agent) + agent = + Keyword.get(opts, :agent) || + Application.fetch_env!(:kodo, :fake_device_authorization_client_agent) response = Agent.get_and_update(agent, fn state -> From 3129d353aac98a5755d6add831a505ab48569603 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:46:07 +0000 Subject: [PATCH 11/25] docs(rfd-5): track device authorization UI --- 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 8afd3f8..6311020 100644 --- a/rfd/0005/IMPLEMENTATION.org +++ b/rfd/0005/IMPLEMENTATION.org @@ -324,10 +324,10 @@ manufactured after that no-go decision. - [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 +- [X] Render the fixed allowlisted HTTPS verification destination and one-time code separately, reject a provider-returned verification URL that does not match the exact allowlist, and never automatically navigate to that URL. -- [ ] Trigger claims only from attempt creation or LiveView mount/remount, use +- [X] Trigger claims only from attempt creation or LiveView mount/remount, use one conditional update against the exact attempt row that increments and returns its claim epoch, and run a successful claim as a finite task under ~Kodo.ControlPlaneTaskSupervisor~ that survives the originating LiveView @@ -341,7 +341,7 @@ manufactured after that no-go decision. - [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 +- [X] Make a new attempt atomically supersede the prior attempt and fence cancellation, expiry, completion, replacement, and disconnection by attempt generation. Starting reauthorization advances the integration generation and the new attempt captures that post-advance value; install OAuth credentials @@ -376,7 +376,7 @@ manufactured after that no-go decision. explanation, progress state, cancellation, disconnect, and reauthorization to the authenticated integrations page; show subscription quota errors only on the affected session or turn with retry and billing guidance. -- [ ] Use the normal authenticated-user boundary to start authorization or +- [X] Use the normal authenticated-user boundary to start authorization or reauthorization and to cancel an owned attempt; do not require renewed sudo mode. Permit the same active generation-fenced attempt to complete while its ownership and generation checks continue to succeed. From 69de0c655143c92c59913e164c100b4aff7ea16d Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:49:34 +0000 Subject: [PATCH 12/25] feat(integrations): add OAuth refresh leases --- lib/kodo/integrations/integration.ex | 10 ++++++ ...refresh_claim_to_provider_integrations.exs | 36 +++++++++++++++++++ priv/repo/structure.sql | 18 ++++++++-- 3 files changed, 62 insertions(+), 2 deletions(-) create mode 100644 priv/repo/migrations/20260910044734_add_refresh_claim_to_provider_integrations.exs diff --git a/lib/kodo/integrations/integration.ex b/lib/kodo/integrations/integration.ex index 2b0f0bc..2e80063 100644 --- a/lib/kodo/integrations/integration.ex +++ b/lib/kodo/integrations/integration.ex @@ -34,6 +34,10 @@ defmodule Kodo.Integrations.Integration do field :validated_at, :utc_datetime_usec field :refreshed_at, :utc_datetime_usec field :validation_error_code, :string + field :refresh_claim_owner_id, Ecto.UUID + field :refresh_claim_epoch, :integer, default: 0 + field :refresh_claim_generation, :integer + field :refresh_lease_expires_at, :utc_datetime_usec belongs_to :user, Kodo.Accounts.User, type: :id @@ -72,6 +76,12 @@ defmodule Kodo.Integrations.Integration do |> check_constraint(:credential_generation, name: :provider_integrations_credential_generation_valid ) + |> check_constraint(:refresh_claim_epoch, + name: :provider_integrations_refresh_claim_epoch_valid + ) + |> check_constraint(:refresh_claim_owner_id, + name: :provider_integrations_refresh_claim_valid + ) |> check_constraint(:connection_status, name: :provider_integrations_state_valid) |> check_constraint(:active, name: :provider_integrations_active_connected) |> unique_constraint(:active, name: :provider_integrations_one_active_index) diff --git a/priv/repo/migrations/20260910044734_add_refresh_claim_to_provider_integrations.exs b/priv/repo/migrations/20260910044734_add_refresh_claim_to_provider_integrations.exs new file mode 100644 index 0000000..09656a0 --- /dev/null +++ b/priv/repo/migrations/20260910044734_add_refresh_claim_to_provider_integrations.exs @@ -0,0 +1,36 @@ +defmodule Kodo.Repo.Migrations.AddRefreshClaimToProviderIntegrations do + use Ecto.Migration + + def change do + alter table(:provider_integrations) do + add :refresh_claim_owner_id, :binary_id + add :refresh_claim_epoch, :bigint, null: false, default: 0 + add :refresh_claim_generation, :bigint + add :refresh_lease_expires_at, :utc_datetime_usec + end + + create index(:provider_integrations, [:refresh_lease_expires_at], + where: "refresh_claim_owner_id IS NOT NULL", + name: :provider_integrations_refresh_lease_index + ) + + create constraint(:provider_integrations, :provider_integrations_refresh_claim_epoch_valid, + check: "refresh_claim_epoch >= 0" + ) + + create constraint(:provider_integrations, :provider_integrations_refresh_claim_valid, + check: """ + ( + refresh_claim_owner_id IS NULL AND + refresh_claim_generation IS NULL AND + refresh_lease_expires_at IS NULL + ) OR ( + refresh_claim_owner_id IS NOT NULL AND + refresh_claim_generation IS NOT NULL AND + refresh_claim_generation >= 0 AND + refresh_lease_expires_at IS NOT NULL + ) + """ + ) + end +end diff --git a/priv/repo/structure.sql b/priv/repo/structure.sql index ba404cc..bef1f4c 100644 --- a/priv/repo/structure.sql +++ b/priv/repo/structure.sql @@ -2,7 +2,7 @@ -- PostgreSQL database dump -- -\restrict XOpxqSYwtvGfTiqRIBStCvyZpzZaCE6LIgMjOIQHCBPW9ZYUkI1mr4M5q3RWTPU +\restrict VoWETYGZi4vEwyliUNWQG3BVneag5OunvKzcS2NOBFzVtuTzyBWpDfhYIneu2ek -- Dumped from database version 15.19 (Debian 15.19-0+deb12u1) -- Dumped by pg_dump version 15.19 (Debian 15.19-0+deb12u1) @@ -183,11 +183,17 @@ CREATE TABLE public.provider_integrations ( updated_at timestamp without time zone NOT NULL, display_name character varying(80) NOT NULL, active boolean DEFAULT false NOT NULL, + refresh_claim_owner_id uuid, + refresh_claim_epoch bigint DEFAULT 0 NOT NULL, + refresh_claim_generation bigint, + refresh_lease_expires_at timestamp without time zone, CONSTRAINT provider_integrations_active_connected CHECK (((NOT active) OR ((connection_status)::text = 'connected'::text))), CONSTRAINT provider_integrations_authentication_type_valid CHECK (((authentication_type)::text = ANY ((ARRAY['api_key'::character varying, 'oauth'::character varying])::text[]))), CONSTRAINT provider_integrations_credential_generation_valid CHECK ((credential_generation >= 0)), CONSTRAINT provider_integrations_provider_authentication_valid CHECK (((((provider)::text = 'openai_codex'::text) AND ((authentication_type)::text = 'oauth'::text)) OR (((provider)::text = ANY ((ARRAY['openai'::character varying, 'anthropic'::character varying, 'openrouter'::character varying])::text[])) AND ((authentication_type)::text = 'api_key'::text)))), CONSTRAINT provider_integrations_provider_valid CHECK (((provider)::text = ANY ((ARRAY['openai'::character varying, 'openai_codex'::character varying, 'anthropic'::character varying, 'openrouter'::character varying])::text[]))), + CONSTRAINT provider_integrations_refresh_claim_epoch_valid CHECK ((refresh_claim_epoch >= 0)), + CONSTRAINT provider_integrations_refresh_claim_valid CHECK ((((refresh_claim_owner_id IS NULL) AND (refresh_claim_generation IS NULL) AND (refresh_lease_expires_at IS NULL)) OR ((refresh_claim_owner_id IS NOT NULL) AND (refresh_claim_generation IS NOT NULL) AND (refresh_claim_generation >= 0) AND (refresh_lease_expires_at IS NOT NULL)))), CONSTRAINT provider_integrations_state_valid CHECK (((((connection_status)::text = 'disconnected'::text) AND ((validation_status)::text = 'unverified'::text) AND (encrypted_credentials IS NULL) AND (encryption_key_version IS NULL) AND (credential_format_version IS NULL) AND (expires_at IS NULL) AND (validated_at IS NULL) AND (refreshed_at IS NULL) AND (validation_error_code IS NULL)) OR (((connection_status)::text = 'connected'::text) AND ((validation_status)::text = ANY ((ARRAY['unverified'::character varying, 'valid'::character varying, 'invalid'::character varying, 'unavailable'::character varying])::text[])) AND (encrypted_credentials IS NOT NULL) AND (encryption_key_version IS NOT NULL) AND (credential_format_version IS NOT NULL)) OR (((connection_status)::text = 'reauthorization_required'::text) AND ((validation_status)::text = 'unverified'::text) AND (encrypted_credentials IS NOT NULL) AND (encryption_key_version IS NOT NULL) AND (credential_format_version IS NOT NULL)))) ); @@ -553,6 +559,13 @@ CREATE UNIQUE INDEX provider_integrations_attempt_owner_identity_index ON public CREATE UNIQUE INDEX provider_integrations_one_active_index ON public.provider_integrations USING btree (user_id, provider) WHERE active; +-- +-- Name: provider_integrations_refresh_lease_index; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX provider_integrations_refresh_lease_index ON public.provider_integrations USING btree (refresh_lease_expires_at) WHERE (refresh_claim_owner_id IS NOT NULL); + + -- -- Name: provider_integrations_user_id_provider_index; Type: INDEX; Schema: public; Owner: - -- @@ -767,7 +780,7 @@ ALTER TABLE ONLY public.users_tokens -- PostgreSQL database dump complete -- -\unrestrict XOpxqSYwtvGfTiqRIBStCvyZpzZaCE6LIgMjOIQHCBPW9ZYUkI1mr4M5q3RWTPU +\unrestrict VoWETYGZi4vEwyliUNWQG3BVneag5OunvKzcS2NOBFzVtuTzyBWpDfhYIneu2ek INSERT INTO public."schema_migrations" (version) VALUES (20260807073017); INSERT INTO public."schema_migrations" (version) VALUES (20260808062115); @@ -786,3 +799,4 @@ INSERT INTO public."schema_migrations" (version) VALUES (20260903201206); INSERT INTO public."schema_migrations" (version) VALUES (20260904051620); INSERT INTO public."schema_migrations" (version) VALUES (20260907083117); INSERT INTO public."schema_migrations" (version) VALUES (20260907161929); +INSERT INTO public."schema_migrations" (version) VALUES (20260910044734); From 9b30beb723ae8206fefe1fd9cb0c83e59902a084 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:52:02 +0000 Subject: [PATCH 13/25] feat(integrations): coordinate OAuth refresh claims --- lib/kodo/integrations.ex | 98 ++++++++++++++++++++++++++++- test/kodo/integrations_test.exs | 108 ++++++++++++++++++++++++++++++++ 2 files changed, 203 insertions(+), 3 deletions(-) diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index f52226c..f746477 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -27,6 +27,7 @@ defmodule Kodo.Integrations do @device_authorization_token_max_bytes 8_192 @device_authorization_cleanup_age_seconds 24 * 60 * 60 @device_authorization_claim_lease_ms 30_000 + @refresh_claim_lease_ms 30_000 @device_authorization_terminal_errors ~w( provider_unavailable provider_rejected @@ -393,6 +394,62 @@ defmodule Kodo.Integrations do ) end + @doc "Claims one generation of an OAuth integration for a bounded refresh operation." + def claim_refresh(%Scope{user: user}, id, generation, claim_owner_id) + when is_integer(generation) and generation >= 0 do + with {:ok, id} <- cast_uuid(id), + {:ok, claim_owner_id} <- cast_uuid(claim_owner_id) do + lease_expires_at = + DateTime.add(database_now(), @refresh_claim_lease_ms, :millisecond) + + query = + from integration in Integration, + where: + integration.id == ^id and integration.user_id == ^user.id and + integration.provider == "openai_codex" and + integration.authentication_type == "oauth" and + integration.connection_status in ["connected", "reauthorization_required"] and + integration.credential_generation == ^generation and + (is_nil(integration.refresh_claim_owner_id) or + integration.refresh_lease_expires_at <= fragment("clock_timestamp()")) + + updates = [ + set: [ + refresh_claim_owner_id: claim_owner_id, + refresh_claim_generation: generation, + refresh_lease_expires_at: lease_expires_at, + updated_at: now() + ], + inc: [refresh_claim_epoch: 1] + ] + + case Repo.update_all(query, updates) do + {1, nil} -> {:ok, Repo.get_by!(Integration, id: id, user_id: user.id)} + {0, nil} -> refresh_claim_conflict(user.id, id, generation) + end + end + end + + def claim_refresh(%Scope{}, _id, _generation, _claim_owner_id), + do: {:error, :stale_credential_generation} + + @doc "Releases only the exact refresh claim represented by the supplied snapshot." + def release_refresh_claim(%Scope{user: user}, %Integration{} = claim) do + query = exact_refresh_claim_query(user.id, claim) + + case Repo.update_all(query, + set: [ + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil, + updated_at: now() + ] + ) do + {1, nil} -> :ok + {0, nil} -> {:error, :stale_refresh_claim} + end + end + def validation_succeeded(%Scope{} = scope, id, generation) do update_fenced(scope, id, generation, ["connected"], "validation_succeeded", %{ validation_status: "valid", @@ -455,7 +512,10 @@ defmodule Kodo.Integrations do expires_at: nil, validated_at: nil, refreshed_at: nil, - validation_error_code: nil + validation_error_code: nil, + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil } ) |> notify_integration_change() @@ -914,7 +974,13 @@ defmodule Kodo.Integrations do end defp device_authorization_integration_changes(%Integration{}, generation, timestamp) do - [credential_generation: generation, updated_at: timestamp] + [ + credential_generation: generation, + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil, + updated_at: timestamp + ] end defp supersede_active_device_authorization(integration_id, timestamp) do @@ -1086,7 +1152,10 @@ defmodule Kodo.Integrations do expires_at: opts[:expires_at], validated_at: nil, refreshed_at: opts[:refreshed_at], - validation_error_code: nil + validation_error_code: nil, + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil }) update_fenced(scope, id, generation, allowed_connections, audit_event_type, changes) @@ -1106,6 +1175,29 @@ defmodule Kodo.Integrations do defp require_connected(%Integration{connection_status: "connected"}), do: :ok defp require_connected(%Integration{}), do: {:error, :integration_not_connected} + defp refresh_claim_conflict(user_id, id, generation) do + case Repo.get_by(Integration, id: id, user_id: user_id) do + %Integration{credential_generation: ^generation, refresh_claim_owner_id: owner} + when not is_nil(owner) -> + {:error, :refresh_in_progress} + + %Integration{} -> + {:error, :stale_credential_generation} + + nil -> + {:error, :integration_not_found} + end + end + + defp exact_refresh_claim_query(user_id, %Integration{} = claim) do + from integration in Integration, + where: + integration.id == ^claim.id and integration.user_id == ^user_id and + integration.credential_generation == ^claim.refresh_claim_generation and + integration.refresh_claim_owner_id == ^claim.refresh_claim_owner_id and + integration.refresh_claim_epoch == ^claim.refresh_claim_epoch + end + defp lock_provider_integrations(user_id, integration_id) do provider = Integration diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs index 9780eed..bb9022b 100644 --- a/test/kodo/integrations_test.exs +++ b/test/kodo/integrations_test.exs @@ -258,6 +258,114 @@ defmodule Kodo.IntegrationsTest do |> then(fn {:ok, payload} -> {:ok, Map.take(payload, ["access_token"])} end) end + test "claims one OAuth refresh generation and fences release across takeover", %{scope: scope} do + integration = oauth_integration(scope) + + assert {:ok, connected} = + Integrations.oauth_succeeded(scope, integration.id, 0, %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "account_id" => "account" + }) + + first_owner = Ecto.UUID.generate() + second_owner = Ecto.UUID.generate() + + assert {:ok, first_claim} = + Integrations.claim_refresh( + scope, + connected.id, + connected.credential_generation, + first_owner + ) + + assert first_claim.refresh_claim_owner_id == first_owner + assert first_claim.refresh_claim_generation == connected.credential_generation + assert first_claim.refresh_claim_epoch == 1 + + assert {:error, :refresh_in_progress} = + Integrations.claim_refresh( + scope, + connected.id, + connected.credential_generation, + second_owner + ) + + Repo.update!( + change(first_claim, + refresh_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second) + ) + ) + + assert {:ok, second_claim} = + Integrations.claim_refresh( + scope, + connected.id, + connected.credential_generation, + second_owner + ) + + assert second_claim.refresh_claim_epoch == 2 + + assert {:error, :stale_refresh_claim} = + Integrations.release_refresh_claim(scope, first_claim) + + assert :ok = Integrations.release_refresh_claim(scope, second_claim) + + released = Repo.reload!(connected) + assert is_nil(released.refresh_claim_owner_id) + assert is_nil(released.refresh_claim_generation) + assert is_nil(released.refresh_lease_expires_at) + assert released.refresh_claim_epoch == 2 + end + + test "a credential-generation transition clears and permanently fences a refresh claim", %{ + scope: scope + } do + integration = oauth_integration(scope) + + assert {:ok, connected} = + Integrations.oauth_succeeded(scope, integration.id, 0, %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "account_id" => "account" + }) + + assert {:ok, claim} = + Integrations.claim_refresh( + scope, + connected.id, + connected.credential_generation, + Ecto.UUID.generate() + ) + + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + scope, + connected.id, + connected.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + fenced = Repo.reload!(connected) + assert fenced.credential_generation == connected.credential_generation + 1 + assert is_nil(fenced.refresh_claim_owner_id) + assert {:error, :stale_refresh_claim} = Integrations.release_refresh_claim(scope, claim) + + assert {:error, :stale_credential_generation} = + Integrations.refresh_succeeded( + scope, + connected.id, + claim.refresh_claim_generation, + %{ + "access_token" => "late-access", + "refresh_token" => "late-refresh", + "account_id" => "account" + } + ) + end + test "activates only the first OAuth account after authorization", %{scope: scope} do first = oauth_integration(scope) From d142e742f7abc9e1411fec5b040838ad27cfbb15 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:53:14 +0000 Subject: [PATCH 14/25] refactor(integrations): share OAuth credential transport --- ...ion_req_adapter.ex => credential_req_adapter.ex} | 8 ++++---- .../integrations/req_device_authorization_client.ex | 12 ++++++------ ...ter_test.exs => credential_req_adapter_test.exs} | 13 ++++++------- .../req_device_authorization_client_test.exs | 8 ++++---- 4 files changed, 20 insertions(+), 21 deletions(-) rename lib/kodo/integrations/{device_authorization_req_adapter.ex => credential_req_adapter.ex} (93%) rename test/kodo/integrations/{device_authorization_req_adapter_test.exs => credential_req_adapter_test.exs} (51%) diff --git a/lib/kodo/integrations/device_authorization_req_adapter.ex b/lib/kodo/integrations/credential_req_adapter.ex similarity index 93% rename from lib/kodo/integrations/device_authorization_req_adapter.ex rename to lib/kodo/integrations/credential_req_adapter.ex index 5756a83..406416a 100644 --- a/lib/kodo/integrations/device_authorization_req_adapter.ex +++ b/lib/kodo/integrations/credential_req_adapter.ex @@ -1,6 +1,6 @@ -defmodule Kodo.Integrations.DeviceAuthorizationReqAdapter do +defmodule Kodo.Integrations.CredentialReqAdapter do @moduledoc """ - Req adapter for credential-bearing device authorization requests. + Req adapter for credential-bearing OAuth requests. It deliberately uses Mint directly: Finch telemetry includes request and response values, which are secrets throughout this protocol. @@ -10,7 +10,7 @@ defmodule Kodo.Integrations.DeviceAuthorizationReqAdapter do @max_response_bytes 64 * 1024 def run(%Req.Request{} = request) do - timeout = request.options[:device_authorization_timeout] || @default_timeout + timeout = request.options[:credential_request_timeout] || @default_timeout started_at = System.monotonic_time(:millisecond) task = Task.async(fn -> safe_perform(request, started_at, timeout) end) @@ -30,7 +30,7 @@ defmodule Kodo.Integrations.DeviceAuthorizationReqAdapter do end defp perform( - %Req.Request{options: %{device_authorization_transport: transport}} = request, + %Req.Request{options: %{credential_request_transport: transport}} = request, started_at, timeout ) diff --git a/lib/kodo/integrations/req_device_authorization_client.ex b/lib/kodo/integrations/req_device_authorization_client.ex index 5b3ecd2..bdf8b61 100644 --- a/lib/kodo/integrations/req_device_authorization_client.ex +++ b/lib/kodo/integrations/req_device_authorization_client.ex @@ -134,25 +134,25 @@ defmodule Kodo.Integrations.ReqDeviceAuthorizationClient do req_options = Keyword.take(req_options, [ :plug, - :device_authorization_transport, - :device_authorization_timeout + :credential_request_transport, + :credential_request_timeout ]) options = [ url: url, - adapter: Kodo.Integrations.DeviceAuthorizationReqAdapter, + adapter: Kodo.Integrations.CredentialReqAdapter, redirect: false, retry: false, decode_body: false, - device_authorization_timeout: @timeout + credential_request_timeout: @timeout ] ++ request_options ++ req_options request = Req.new() |> Req.Request.register_options([ - :device_authorization_timeout, - :device_authorization_transport + :credential_request_timeout, + :credential_request_transport ]) case Req.post(request, options) do diff --git a/test/kodo/integrations/device_authorization_req_adapter_test.exs b/test/kodo/integrations/credential_req_adapter_test.exs similarity index 51% rename from test/kodo/integrations/device_authorization_req_adapter_test.exs rename to test/kodo/integrations/credential_req_adapter_test.exs index d623971..d495dce 100644 --- a/test/kodo/integrations/device_authorization_req_adapter_test.exs +++ b/test/kodo/integrations/credential_req_adapter_test.exs @@ -1,19 +1,18 @@ -defmodule Kodo.Integrations.DeviceAuthorizationReqAdapterTest do +defmodule Kodo.Integrations.CredentialReqAdapterTest do use ExUnit.Case, async: true - alias Kodo.Integrations.DeviceAuthorizationReqAdapter + alias Kodo.Integrations.CredentialReqAdapter test "builds origin-form request targets for every device authorization endpoint" do - assert DeviceAuthorizationReqAdapter.request_target( + assert CredentialReqAdapter.request_target( URI.parse("https://auth.openai.com/api/accounts/deviceauth/usercode") ) == "/api/accounts/deviceauth/usercode" - assert DeviceAuthorizationReqAdapter.request_target( + assert CredentialReqAdapter.request_target( URI.parse("https://auth.openai.com/api/accounts/deviceauth/token?attempt=1") ) == "/api/accounts/deviceauth/token?attempt=1" - assert DeviceAuthorizationReqAdapter.request_target( - URI.parse("https://auth.openai.com/oauth/token") - ) == "/oauth/token" + assert CredentialReqAdapter.request_target(URI.parse("https://auth.openai.com/oauth/token")) == + "/oauth/token" end end diff --git a/test/kodo/integrations/req_device_authorization_client_test.exs b/test/kodo/integrations/req_device_authorization_client_test.exs index e281d43..c8fca92 100644 --- a/test/kodo/integrations/req_device_authorization_client_test.exs +++ b/test/kodo/integrations/req_device_authorization_client_test.exs @@ -199,7 +199,7 @@ defmodule Kodo.Integrations.ReqDeviceAuthorizationClientTest do end assert {:error, :provider_unavailable} = - ReqDeviceAuthorizationClient.create(device_authorization_transport: transport) + ReqDeviceAuthorizationClient.create(credential_request_transport: transport) end test "accepts device fields at 1024 bytes and rejects 1025 bytes" do @@ -274,7 +274,7 @@ defmodule Kodo.Integrations.ReqDeviceAuthorizationClientTest do assert {:ok, %{"authorization_code" => ^sentinel}} = ReqDeviceAuthorizationClient.poll( %{"device_auth_id" => sentinel, "user_code" => sentinel}, - device_authorization_transport: transport + credential_request_transport: transport ) refute_receive {:finch_telemetry, _event, _measurements, _metadata} @@ -299,8 +299,8 @@ defmodule Kodo.Integrations.ReqDeviceAuthorizationClientTest do assert {:error, :provider_unavailable} = ReqDeviceAuthorizationClient.create( - device_authorization_transport: transport, - device_authorization_timeout: 10 + credential_request_transport: transport, + credential_request_timeout: 10 ) assert_received :transport_started From 64539d09a93f2bab7059e45c2fb7aa9cc2510100 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:55:01 +0000 Subject: [PATCH 15/25] feat(integrations): add Codex refresh client --- lib/kodo/integrations/refresh_client.ex | 5 + lib/kodo/integrations/req_refresh_client.ex | 85 +++++++++++++++++ .../integrations/req_refresh_client_test.exs | 94 +++++++++++++++++++ 3 files changed, 184 insertions(+) create mode 100644 lib/kodo/integrations/refresh_client.ex create mode 100644 lib/kodo/integrations/req_refresh_client.ex create mode 100644 test/kodo/integrations/req_refresh_client_test.exs diff --git a/lib/kodo/integrations/refresh_client.ex b/lib/kodo/integrations/refresh_client.ex new file mode 100644 index 0000000..2507453 --- /dev/null +++ b/lib/kodo/integrations/refresh_client.ex @@ -0,0 +1,5 @@ +defmodule Kodo.Integrations.RefreshClient do + @moduledoc false + + @callback refresh(String.t(), keyword()) :: {:ok, map()} | {:error, atom()} +end diff --git a/lib/kodo/integrations/req_refresh_client.ex b/lib/kodo/integrations/req_refresh_client.ex new file mode 100644 index 0000000..5e01825 --- /dev/null +++ b/lib/kodo/integrations/req_refresh_client.ex @@ -0,0 +1,85 @@ +defmodule Kodo.Integrations.ReqRefreshClient do + @moduledoc "Performs the pinned OpenAI Codex refresh-token contract." + + @behaviour Kodo.Integrations.RefreshClient + + @client_id "app_EMoamEEZ73f0CkXaXp7hrann" + @token_url "https://auth.openai.com/oauth/token" + @timeout 10_000 + @max_response_bytes 8_192 + + @impl true + def refresh(refresh_token, req_options \\ []) + + def refresh(refresh_token, req_options) + when is_binary(refresh_token) and byte_size(refresh_token) > 0 and + byte_size(refresh_token) <= @max_response_bytes and is_list(req_options) do + options = + [ + url: @token_url, + json: %{ + client_id: @client_id, + grant_type: "refresh_token", + refresh_token: refresh_token + }, + adapter: Kodo.Integrations.CredentialReqAdapter, + redirect: false, + retry: false, + decode_body: false, + credential_request_timeout: @timeout + ] ++ + Keyword.take(req_options, [ + :plug, + :credential_request_transport, + :credential_request_timeout + ]) + + request = + Req.new() + |> Req.Request.register_options([ + :credential_request_timeout, + :credential_request_transport + ]) + + case Req.post(request, options) do + {:ok, %Req.Response{status: status, body: body}} when status in 200..299 -> + decode_success(body) + + {:ok, %Req.Response{status: 400, body: body}} -> + if invalid_grant?(body), + do: {:error, :invalid_grant}, + else: {:error, :provider_unavailable} + + {:ok, %Req.Response{status: status}} when status in 300..399 -> + {:error, :redirect} + + {:ok, %Req.Response{}} -> + {:error, :provider_unavailable} + + {:error, _reason} -> + {:error, :provider_unavailable} + end + end + + def refresh(_refresh_token, _req_options), do: {:error, :oauth_refresh_response_invalid} + + defp decode_success(body) do + with {:ok, decoded} when is_map(decoded) <- decode_json(body), + access_token when is_binary(access_token) and byte_size(access_token) > 0 <- + decoded["access_token"] do + {:ok, decoded} + else + _invalid -> {:error, :oauth_refresh_response_invalid} + end + end + + defp invalid_grant?(body) do + case decode_json(body) do + {:ok, %{"error" => "invalid_grant"}} -> true + _other -> false + end + end + + defp decode_json(body) when is_binary(body), do: Jason.decode(body) + defp decode_json(_body), do: {:error, :invalid_body} +end diff --git a/test/kodo/integrations/req_refresh_client_test.exs b/test/kodo/integrations/req_refresh_client_test.exs new file mode 100644 index 0000000..53a7b58 --- /dev/null +++ b/test/kodo/integrations/req_refresh_client_test.exs @@ -0,0 +1,94 @@ +defmodule Kodo.Integrations.ReqRefreshClientTest do + use ExUnit.Case, async: true + + alias Kodo.Integrations.ReqRefreshClient + + @client_id "app_EMoamEEZ73f0CkXaXp7hrann" + + test "refreshes at the pinned endpoint with the published contract" do + plug = fn conn -> + assert conn.method == "POST" + assert conn.scheme == :https + assert conn.host == "auth.openai.com" + assert conn.request_path == "/oauth/token" + + assert json_body(conn) == %{ + "client_id" => @client_id, + "grant_type" => "refresh_token", + "refresh_token" => "refresh-secret" + } + + Req.Test.json(conn, %{ + "access_token" => "access-secret", + "refresh_token" => "rotated-secret", + "id_token" => "identity-secret" + }) + end + + assert {:ok, %{"access_token" => "access-secret"}} = + ReqRefreshClient.refresh("refresh-secret", plug: plug) + end + + test "classifies only an exact invalid_grant response as invalid credentials" do + invalid_grant = fn conn -> + conn |> Plug.Conn.put_status(400) |> Req.Test.json(%{"error" => "invalid_grant"}) + end + + assert {:error, :invalid_grant} = + ReqRefreshClient.refresh("refresh-secret", plug: invalid_grant) + + for body <- [%{"error" => "temporarily_unavailable"}, %{"message" => "invalid_grant"}] do + plug = fn conn -> conn |> Plug.Conn.put_status(400) |> Req.Test.json(body) end + + assert {:error, :provider_unavailable} = + ReqRefreshClient.refresh("refresh-secret", plug: plug) + end + end + + test "rejects redirects, malformed success, and unsafe caller input with bounded errors" do + redirect = fn conn -> Plug.Conn.send_resp(conn, 302, "private") end + malformed = fn conn -> Req.Test.json(conn, %{"refresh_token" => "private"}) end + + assert {:error, :redirect} = ReqRefreshClient.refresh("refresh-secret", plug: redirect) + + assert {:error, :oauth_refresh_response_invalid} = + ReqRefreshClient.refresh("refresh-secret", plug: malformed) + + assert {:error, :oauth_refresh_response_invalid} = ReqRefreshClient.refresh("") + assert {:error, :oauth_refresh_response_invalid} = ReqRefreshClient.refresh(:secret) + end + + test "uses the credential transport without emitting Finch telemetry" do + test_pid = self() + handler_id = "refresh-finch-canary-#{System.unique_integer()}" + + :ok = + :telemetry.attach_many( + handler_id, + [[:finch, :request, :start], [:finch, :request, :stop], [:finch, :request, :exception]], + fn event, measurements, metadata, pid -> + send(pid, {:finch_telemetry, event, measurements, metadata}) + end, + test_pid + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + + transport = fn request, _started_at, _timeout -> + assert IO.iodata_to_binary(request.body) =~ "refresh-secret" + + {request, + Req.Response.new(status: 200, body: Jason.encode!(%{"access_token" => "access-secret"}))} + end + + assert {:ok, _tokens} = + ReqRefreshClient.refresh("refresh-secret", credential_request_transport: transport) + + refute_receive {:finch_telemetry, _event, _measurements, _metadata} + end + + defp json_body(conn) do + {:ok, body, _conn} = Plug.Conn.read_body(conn) + Jason.decode!(body) + end +end From a0c32e2d9506b8be99e81e971b1f3e167cade593 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 04:56:59 +0000 Subject: [PATCH 16/25] feat(integrations): normalize rotated OAuth tokens --- lib/kodo/integrations/refresh_tokens.ex | 91 +++++++++++++++++++ .../kodo/integrations/refresh_tokens_test.exs | 82 +++++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 lib/kodo/integrations/refresh_tokens.ex create mode 100644 test/kodo/integrations/refresh_tokens_test.exs diff --git a/lib/kodo/integrations/refresh_tokens.ex b/lib/kodo/integrations/refresh_tokens.ex new file mode 100644 index 0000000..74df26e --- /dev/null +++ b/lib/kodo/integrations/refresh_tokens.ex @@ -0,0 +1,91 @@ +defmodule Kodo.Integrations.RefreshTokens do + @moduledoc false + + @account_claim "https://api.openai.com/auth" + @max_token_bytes 8_192 + + def normalize(response, current, now \\ DateTime.utc_now()) + + def normalize(response, current, %DateTime{} = now) + when is_map(response) and is_map(current) do + with {:ok, access_token} <- token(response, "access_token"), + {:ok, expires_at} <- access_expiry(access_token, now), + {:ok, refresh_token} <- optional_token(response, current, "refresh_token"), + {:ok, id_token} <- optional_token(response, current, "id_token"), + {:ok, current_account_id} <- token(current, "account_id"), + {:ok, account_id} <- account_id(response, id_token, current_account_id), + true <- account_id == current_account_id do + {:ok, + %{ + credentials: %{ + "access_token" => access_token, + "refresh_token" => refresh_token, + "id_token" => id_token, + "account_id" => account_id + }, + expires_at: expires_at + }} + else + false -> {:error, :refresh_account_identity_mismatch} + {:error, _reason} = error -> error + end + end + + def normalize(_response, _current, _now), do: {:error, :oauth_refresh_response_invalid} + + defp account_id(response, id_token, current_account_id) do + if Map.has_key?(response, "id_token") do + with {:ok, claims} <- jwt_claims(id_token), + %{"chatgpt_account_id" => account_id} <- claims[@account_claim], + :ok <- valid_token(account_id) do + {:ok, account_id} + else + _invalid -> {:error, :oauth_refresh_response_invalid} + end + else + {:ok, current_account_id} + end + end + + defp access_expiry(access_token, now) do + with {:ok, %{"exp" => unix}} when is_integer(unix) <- jwt_claims(access_token), + {:ok, expires_at} <- DateTime.from_unix(unix), + true <- DateTime.after?(expires_at, now) do + {:ok, %{expires_at | microsecond: {0, 6}}} + else + _invalid -> {:error, :oauth_refresh_response_invalid} + end + end + + defp optional_token(response, current, field) do + case Map.fetch(response, field) do + {:ok, value} -> checked_token(value) + :error -> token(current, field) + end + end + + defp token(payload, field), do: payload |> Map.get(field) |> checked_token() + + defp checked_token(value) do + case valid_token(value) do + :ok -> {:ok, value} + :error -> {:error, :oauth_refresh_response_invalid} + end + end + + defp valid_token(value) + when is_binary(value) and byte_size(value) > 0 and byte_size(value) <= @max_token_bytes, + do: :ok + + defp valid_token(_value), do: :error + + defp jwt_claims(token) do + with [_header, payload, _signature] <- String.split(token, "."), + {:ok, json} <- Base.url_decode64(payload, padding: false), + {:ok, claims} when is_map(claims) <- Jason.decode(json) do + {:ok, claims} + else + _invalid -> {:error, :oauth_refresh_response_invalid} + end + end +end diff --git a/test/kodo/integrations/refresh_tokens_test.exs b/test/kodo/integrations/refresh_tokens_test.exs new file mode 100644 index 0000000..ef8df2a --- /dev/null +++ b/test/kodo/integrations/refresh_tokens_test.exs @@ -0,0 +1,82 @@ +defmodule Kodo.Integrations.RefreshTokensTest do + use ExUnit.Case, async: true + + alias Kodo.Integrations.RefreshTokens + + test "requires a future access expiry and retains omitted rotated fields" do + now = ~U[2026-09-10 05:00:00.000000Z] + current = current_tokens() + + assert {:ok, normalized} = + RefreshTokens.normalize( + %{"access_token" => jwt(%{"exp" => DateTime.to_unix(now) + 60})}, + current, + now + ) + + assert normalized.credentials["refresh_token"] == current["refresh_token"] + assert normalized.credentials["id_token"] == current["id_token"] + assert normalized.credentials["account_id"] == "account" + assert normalized.expires_at == ~U[2026-09-10 05:01:00.000000Z] + + assert {:error, :oauth_refresh_response_invalid} = + RefreshTokens.normalize( + %{"access_token" => jwt(%{"exp" => DateTime.to_unix(now)})}, + current, + now + ) + end + + test "persists rotated refresh and identity tokens without changing billing identity" do + now = ~U[2026-09-10 05:00:00.000000Z] + rotated_id = id_token("account") + + assert {:ok, normalized} = + RefreshTokens.normalize( + %{ + "access_token" => jwt(%{"exp" => DateTime.to_unix(now) + 60}), + "refresh_token" => "rotated-refresh", + "id_token" => rotated_id + }, + current_tokens(), + now + ) + + assert normalized.credentials["refresh_token"] == "rotated-refresh" + assert normalized.credentials["id_token"] == rotated_id + end + + test "rejects an account identity change and malformed returned identity token" do + response = %{ + "access_token" => jwt(%{"exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.+(60)}), + "id_token" => id_token("other-account") + } + + assert {:error, :refresh_account_identity_mismatch} = + RefreshTokens.normalize(response, current_tokens()) + + assert {:error, :oauth_refresh_response_invalid} = + RefreshTokens.normalize( + %{response | "id_token" => "malformed"}, + current_tokens() + ) + end + + defp current_tokens do + %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "id_token" => id_token("account"), + "account_id" => "account" + } + end + + defp id_token(account_id) do + jwt(%{"https://api.openai.com/auth" => %{"chatgpt_account_id" => account_id}}) + end + + defp jwt(claims) do + encoded = claims |> Jason.encode!() |> Base.url_encode64(padding: false) + "header.#{encoded}.signature" + end +end From 82d7d2a75836586b9ee4c9053b9b1fcf68903d80 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 05:04:18 +0000 Subject: [PATCH 17/25] feat(integrations): refresh OAuth credentials just in time --- lib/kodo/integrations.ex | 33 ++- lib/kodo/integrations/audit_event.ex | 1 + lib/kodo/integrations/oauth_refresh.ex | 183 ++++++++++++++ lib/kodo/llm.ex | 3 +- lib/kodo/llm/credential_resolver.ex | 2 + lib/kodo/llm/provider_error.ex | 4 +- test/kodo/integrations/oauth_refresh_test.exs | 232 ++++++++++++++++++ test/kodo/llm/credential_resolver_test.exs | 16 +- test/support/fake_refresh_client.ex | 15 ++ 9 files changed, 481 insertions(+), 8 deletions(-) create mode 100644 lib/kodo/integrations/oauth_refresh.ex create mode 100644 test/kodo/integrations/oauth_refresh_test.exs create mode 100644 test/support/fake_refresh_client.ex diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index f746477..765fca9 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -408,7 +408,7 @@ defmodule Kodo.Integrations do integration.id == ^id and integration.user_id == ^user.id and integration.provider == "openai_codex" and integration.authentication_type == "oauth" and - integration.connection_status in ["connected", "reauthorization_required"] and + integration.connection_status == "connected" and integration.credential_generation == ^generation and (is_nil(integration.refresh_claim_owner_id) or integration.refresh_lease_expires_at <= fragment("clock_timestamp()")) @@ -450,6 +450,37 @@ defmodule Kodo.Integrations do end end + def require_refresh_reauthorization(%Scope{user: user}, %Integration{} = claim, error_code) + when error_code in ~w(refresh_invalid_grant refresh_account_identity_mismatch) do + Repo.transaction(fn -> + changes = [ + connection_status: "reauthorization_required", + active: false, + validation_status: "unverified", + validated_at: nil, + validation_error_code: error_code, + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil, + updated_at: now() + ] + + case Repo.update_all(exact_refresh_claim_query(user.id, claim), set: changes) do + {1, nil} -> + integration = Repo.get_by!(Integration, id: claim.id, user_id: user.id) + audit!(user.id, integration, "refresh_reauthorization_required") + integration + + {0, nil} -> + Repo.rollback(:stale_refresh_claim) + end + end) + |> notify_integration_change() + end + + def require_refresh_reauthorization(%Scope{}, %Integration{}, _error_code), + do: {:error, :unsafe_refresh_error} + def validation_succeeded(%Scope{} = scope, id, generation) do update_fenced(scope, id, generation, ["connected"], "validation_succeeded", %{ validation_status: "valid", diff --git a/lib/kodo/integrations/audit_event.ex b/lib/kodo/integrations/audit_event.ex index e78ba87..9ecfa2c 100644 --- a/lib/kodo/integrations/audit_event.ex +++ b/lib/kodo/integrations/audit_event.ex @@ -19,6 +19,7 @@ defmodule Kodo.Integrations.AuditEvent do oauth_succeeded refresh_succeeded refresh_invalid_grant + refresh_reauthorization_required ) @primary_key {:id, :binary_id, autogenerate: true} diff --git a/lib/kodo/integrations/oauth_refresh.ex b/lib/kodo/integrations/oauth_refresh.ex new file mode 100644 index 0000000..0c35ec6 --- /dev/null +++ b/lib/kodo/integrations/oauth_refresh.ex @@ -0,0 +1,183 @@ +defmodule Kodo.Integrations.OAuthRefresh do + @moduledoc "Runs generation-fenced, just-in-time OAuth refreshes as finite tasks." + + alias Kodo.Accounts.Scope + alias Kodo.Cluster.InstanceManager + alias Kodo.Integrations + alias Kodo.Integrations.CredentialEncryption + alias Kodo.Integrations.Integration + alias Kodo.Integrations.RefreshTokens + + @refresh_buffer_seconds 5 * 60 + @wait_timeout_ms 12_000 + @wait_interval_ms 250 + + def ensure_fresh(%Scope{} = scope, %Integration{} = integration, opts \\ []) do + now = Keyword.get(opts, :now, DateTime.utc_now()) + buffer = Keyword.get(opts, :refresh_buffer_seconds, @refresh_buffer_seconds) + + if refresh_due?(integration, now, buffer) do + refresh(scope, integration, opts) + else + {:ok, integration} + end + end + + defp refresh(scope, integration, opts) do + owner_id = Keyword.get_lazy(opts, :claim_owner_id, &InstanceManager.current_boot_id/0) + + deadline = + System.monotonic_time(:millisecond) + Keyword.get(opts, :wait_timeout, @wait_timeout_ms) + + if is_binary(owner_id) do + claim_or_wait(scope, integration, owner_id, deadline, opts) + else + {:error, :provider_unavailable} + end + end + + defp claim_or_wait(scope, integration, owner_id, deadline, opts) do + case Integrations.claim_refresh( + scope, + integration.id, + integration.credential_generation, + owner_id + ) do + {:ok, claim} -> + run_claim(scope, claim, opts) + + {:error, :refresh_in_progress} -> + wait_for_claim(scope, integration, owner_id, deadline, opts) + + {:error, :stale_credential_generation} -> + current_result(scope, integration) + + {:error, _reason} -> + {:error, :provider_unavailable} + end + end + + defp wait_for_claim(scope, integration, owner_id, deadline, opts) do + if System.monotonic_time(:millisecond) < deadline do + receive do + after + @wait_interval_ms -> claim_or_wait(scope, integration, owner_id, deadline, opts) + end + else + current_result(scope, integration) + end + end + + defp run_claim(scope, claim, opts) do + supervisor = Keyword.get(opts, :supervisor, Kodo.ControlPlaneTaskSupervisor) + + task = + Task.Supervisor.async_nolink(supervisor, fn -> + perform(scope, claim, opts) + end) + + case Task.yield(task, Keyword.get(opts, :wait_timeout, @wait_timeout_ms)) do + {:ok, result} -> + result + + {:exit, _reason} -> + {:error, :provider_unavailable} + + nil -> + _result = Task.shutdown(task, :brutal_kill) + {:error, :provider_unavailable} + end + end + + defp perform(scope, claim, opts) do + client = Keyword.get(opts, :client, configured_client()) + client_options = Keyword.get(opts, :client_options, []) + + with {:ok, current} <- CredentialEncryption.decrypt(claim), + {:ok, refresh_token} <- fetch_refresh_token(current), + {:ok, response} <- safe_refresh(client, refresh_token, client_options), + {:ok, normalized} <- RefreshTokens.normalize(response, current), + {:ok, integration} <- + Integrations.refresh_succeeded( + scope, + claim.id, + claim.refresh_claim_generation, + normalized.credentials, + expires_at: normalized.expires_at, + refreshed_at: DateTime.utc_now() + ) do + {:ok, integration} + else + {:error, :invalid_grant} -> + require_reauthorization(scope, claim, "refresh_invalid_grant") + + {:error, :refresh_account_identity_mismatch} -> + require_reauthorization(scope, claim, "refresh_account_identity_mismatch") + + {:error, :stale_credential_generation} -> + current_result(scope, claim) + + {:error, _reason} -> + _result = Integrations.release_refresh_claim(scope, claim) + {:error, :provider_unavailable} + end + end + + defp require_reauthorization(scope, claim, error_code) do + case Integrations.require_refresh_reauthorization(scope, claim, error_code) do + {:ok, _integration} -> {:error, :integration_reauthorization_required} + {:error, :stale_refresh_claim} -> current_result(scope, claim) + {:error, _reason} -> {:error, :provider_unavailable} + end + end + + defp current_result(scope, integration) do + case Integrations.get_integration(scope, integration.id) do + {:ok, %{connection_status: "connected"} = current} + when current.credential_generation > integration.credential_generation -> + {:ok, current} + + {:ok, %{connection_status: "reauthorization_required"}} -> + {:error, :integration_reauthorization_required} + + _other -> + {:error, :provider_unavailable} + end + end + + defp refresh_due?( + %Integration{ + provider: "openai_codex", + authentication_type: "oauth", + expires_at: expires_at + }, + %DateTime{} = now, + buffer + ) + when is_integer(buffer) and buffer >= 0 do + not match?(%DateTime{}, expires_at) or + DateTime.compare(expires_at, DateTime.add(now, buffer, :second)) != :gt + end + + defp refresh_due?(_integration, _now, _buffer), do: false + + defp fetch_refresh_token(%{"refresh_token" => token}) + when is_binary(token) and byte_size(token) > 0, + do: {:ok, token} + + defp fetch_refresh_token(_current), do: {:error, :oauth_refresh_response_invalid} + + defp safe_refresh(client, refresh_token, options) do + try do + client.refresh(refresh_token, options) + rescue + _exception -> {:error, :provider_unavailable} + catch + _kind, _reason -> {:error, :provider_unavailable} + end + end + + defp configured_client do + Application.get_env(:kodo, :oauth_refresh_client, Kodo.Integrations.ReqRefreshClient) + end +end diff --git a/lib/kodo/llm.ex b/lib/kodo/llm.ex index c54f1de..2029a02 100644 --- a/lib/kodo/llm.ex +++ b/lib/kodo/llm.ex @@ -124,7 +124,8 @@ defmodule Kodo.LLM do :integration_disconnected, :integration_reauthorization_required, :integration_invalid, - :stale_credential_generation + :stale_credential_generation, + :provider_unavailable ] -> {:error, ProviderError.from_integration(reason, model)} diff --git a/lib/kodo/llm/credential_resolver.ex b/lib/kodo/llm/credential_resolver.ex index 0659332..c5e400b 100644 --- a/lib/kodo/llm/credential_resolver.ex +++ b/lib/kodo/llm/credential_resolver.ex @@ -12,6 +12,7 @@ defmodule Kodo.LLM.CredentialResolver do alias Kodo.Integrations alias Kodo.Integrations.CredentialEncryption alias Kodo.Integrations.Integration + alias Kodo.Integrations.OAuthRefresh alias Kodo.LLM.Credential alias Kodo.LLM.IntegrationRef @@ -47,6 +48,7 @@ defmodule Kodo.LLM.CredentialResolver do :ok <- require_authentication_type(integration, reference), :ok <- require_billing_path(integration, reference), :ok <- require_usable(integration), + {:ok, integration} <- OAuthRefresh.ensure_fresh(scope, integration), {:ok, payload} <- CredentialEncryption.decrypt(integration) do build_credential(integration, reference, payload) end diff --git a/lib/kodo/llm/provider_error.ex b/lib/kodo/llm/provider_error.ex index 249f1bc..02ac85f 100644 --- a/lib/kodo/llm/provider_error.ex +++ b/lib/kodo/llm/provider_error.ex @@ -28,7 +28,8 @@ defmodule Kodo.LLM.ProviderError do :integration_disconnected, :integration_reauthorization_required, :integration_invalid, - :stale_credential_generation + :stale_credential_generation, + :provider_unavailable ] do provider = Atom.to_string(model.provider) @@ -96,6 +97,7 @@ defmodule Kodo.LLM.ProviderError do defp integration_kind(:integration_invalid), do: :authentication_rejected defp integration_kind(:integration_reauthorization_required), do: :authentication_rejected defp integration_kind(:stale_credential_generation), do: :integration_changed + defp integration_kind(:provider_unavailable), do: :provider_unavailable defp integration_kind(_reason), do: :integration_required defp billing_path("openai_codex"), do: :subscription diff --git a/test/kodo/integrations/oauth_refresh_test.exs b/test/kodo/integrations/oauth_refresh_test.exs new file mode 100644 index 0000000..dfc0fcd --- /dev/null +++ b/test/kodo/integrations/oauth_refresh_test.exs @@ -0,0 +1,232 @@ +defmodule Kodo.Integrations.OAuthRefreshTest do + use Kodo.DataCase, async: false + + alias Kodo.AccountsFixtures + alias Kodo.Integrations + alias Kodo.Integrations.CredentialEncryption + alias Kodo.Integrations.DeviceAuthorizationAttempt + alias Kodo.Integrations.Integration + alias Kodo.Integrations.OAuthRefresh + alias Kodo.Test.FakeRefreshClient + + setup do + supervisor = start_supervised!({Task.Supervisor, name: nil}) + scope = AccountsFixtures.user_scope_fixture() + + integration = + %Integration{user_id: scope.user.id} + |> Integration.create_changeset(%{ + provider: "openai_codex", + authentication_type: "oauth", + display_name: "Subscription" + }) + |> Repo.insert!() + + current = current_tokens() + + assert {:ok, connected} = + Integrations.oauth_succeeded(scope, integration.id, 0, current, + expires_at: DateTime.add(DateTime.utc_now(), 30, :second) + ) + + %{scope: scope, integration: connected, supervisor: supervisor} + end + + test "refreshes shortly before expiry and persists rotated token state", context do + client = fake_client([{:ok, rotated_tokens("account")}]) + + assert {:ok, refreshed} = refresh(context, client) + assert refreshed.credential_generation == context.integration.credential_generation + 1 + assert refreshed.connection_status == "connected" + assert %DateTime{} = refreshed.refreshed_at + assert %DateTime{} = refreshed.expires_at + assert is_nil(refreshed.refresh_claim_owner_id) + + assert {:ok, credentials} = CredentialEncryption.decrypt(refreshed) + assert credentials["access_token"] == rotated_tokens("account")["access_token"] + assert credentials["refresh_token"] == "rotated-refresh" + assert credentials["account_id"] == "account" + assert Agent.get(client, &Enum.reverse(&1.refresh_tokens)) == ["old-refresh"] + end + + test "does not refresh a token outside the refresh buffer", context do + integration = + context.integration + |> change(expires_at: DateTime.add(DateTime.utc_now(), 3_600, :second)) + |> Repo.update!() + + client = fake_client([]) + + assert {:ok, unchanged} = + OAuthRefresh.ensure_fresh(context.scope, integration, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + + assert unchanged.credential_generation == integration.credential_generation + assert Agent.get(client, & &1.refresh_tokens) == [] + end + + test "invalid grant and account changes require reauthorization without erasing credentials", + context do + for {response, error_code} <- [ + {{:error, :invalid_grant}, "refresh_invalid_grant"}, + {{:ok, rotated_tokens("other-account")}, "refresh_account_identity_mismatch"} + ] do + integration = Repo.reload!(context.integration) + client = fake_client([response]) + + assert {:error, :integration_reauthorization_required} = + refresh(%{context | integration: integration}, client) + + required = Repo.reload!(integration) + assert required.connection_status == "reauthorization_required" + assert required.validation_error_code == error_code + assert required.credential_generation == integration.credential_generation + assert required.encrypted_credentials == integration.encrypted_credentials + + Repo.update!( + change(required, + connection_status: "connected", + active: true, + validation_error_code: nil + ) + ) + end + end + + test "transient provider failures preserve credentials and release the exact claim", context do + client = fake_client([{:error, :provider_unavailable}]) + + assert {:error, :provider_unavailable} = refresh(context, client) + + unchanged = Repo.reload!(context.integration) + assert unchanged.connection_status == "connected" + assert unchanged.encrypted_credentials == context.integration.encrypted_credentials + assert is_nil(unchanged.refresh_claim_owner_id) + assert unchanged.refresh_claim_epoch == 1 + end + + test "the first generation-matching success wins after lease takeover", context do + owner = self() + first_tokens = rotated_tokens("account", "first") + second_tokens = rotated_tokens("account", "second") + + first_response = fn _refresh_token -> + send(owner, {:first_refresh_started, self()}) + + receive do + :finish_first_refresh -> {:ok, first_tokens} + end + end + + client = fake_client([first_response, {:ok, second_tokens}]) + + first = Task.async(fn -> refresh(context, client) end) + assert_receive {:first_refresh_started, first_worker} + + context.integration + |> Repo.reload!() + |> change(refresh_lease_expires_at: DateTime.add(DateTime.utc_now(), -1, :second)) + |> Repo.update!() + + assert {:ok, winner} = refresh(context, client) + assert winner.credential_generation == context.integration.credential_generation + 1 + + send(first_worker, :finish_first_refresh) + assert {:ok, same_winner} = Task.await(first) + assert same_winner.credential_generation == winner.credential_generation + + assert {:ok, persisted} = CredentialEncryption.decrypt(Repo.reload!(winner)) + assert persisted["refresh_token"] == "second-refresh" + assert persisted["access_token"] == second_tokens["access_token"] + end + + test "user deletion cascades OAuth state and fences an in-flight refresh", context do + assert {:ok, _attempt} = + Integrations.begin_device_authorization( + context.scope, + context.integration.id, + context.integration.credential_generation, + %{"device_auth_id" => "device", "user_code" => "CODE"}, + 0 + ) + + integration = Repo.reload!(context.integration) + owner = self() + + blocked = fn _refresh_token -> + send(owner, {:deletion_refresh_started, self()}) + + receive do + :finish_deleted_refresh -> {:ok, rotated_tokens("account")} + end + end + + client = fake_client([blocked]) + refresh = Task.async(fn -> refresh(%{context | integration: integration}, client) end) + assert_receive {:deletion_refresh_started, worker} + + assert Repo.reload!(integration).refresh_claim_owner_id + Repo.delete!(context.scope.user) + refute Repo.exists?(Integration) + refute Repo.exists?(DeviceAuthorizationAttempt) + + send(worker, :finish_deleted_refresh) + assert {:error, :provider_unavailable} = Task.await(refresh) + refute Repo.exists?(Integration) + refute Repo.exists?(DeviceAuthorizationAttempt) + end + + defp refresh(context, client) do + OAuthRefresh.ensure_fresh(context.scope, context.integration, + client: FakeRefreshClient, + client_options: [agent: client], + supervisor: context.supervisor, + claim_owner_id: Ecto.UUID.generate() + ) + end + + defp fake_client(responses) do + start_supervised!( + {Agent, fn -> %{responses: responses, refresh_tokens: []} end}, + id: {:fake_refresh_client, System.unique_integer([:positive])} + ) + end + + defp current_tokens do + %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "id_token" => id_token("account"), + "account_id" => "account" + } + end + + defp rotated_tokens(account_id) do + rotated_tokens(account_id, "rotated") + end + + defp rotated_tokens(account_id, label) do + %{ + "access_token" => + jwt(%{ + "exp" => DateTime.utc_now() |> DateTime.to_unix() |> Kernel.+(3_600), + "label" => label + }), + "refresh_token" => "#{label}-refresh", + "id_token" => id_token(account_id) + } + end + + defp id_token(account_id) do + jwt(%{"https://api.openai.com/auth" => %{"chatgpt_account_id" => account_id}}) + end + + defp jwt(claims) do + encoded = claims |> Jason.encode!() |> Base.url_encode64(padding: false) + "header.#{encoded}.signature" + end +end diff --git a/test/kodo/llm/credential_resolver_test.exs b/test/kodo/llm/credential_resolver_test.exs index ef34360..0a96f79 100644 --- a/test/kodo/llm/credential_resolver_test.exs +++ b/test/kodo/llm/credential_resolver_test.exs @@ -288,11 +288,17 @@ defmodule Kodo.LLM.CredentialResolverTest do integration = Kodo.Repo.insert!(integration) assert {:ok, connected} = - Integrations.oauth_succeeded(scope, integration.id, 0, %{ - "access_token" => "access-secret", - "refresh_token" => "refresh-secret", - "account_id" => "account-secret" - }) + Integrations.oauth_succeeded( + scope, + integration.id, + 0, + %{ + "access_token" => "access-secret", + "refresh_token" => "refresh-secret", + "account_id" => "account-secret" + }, + expires_at: DateTime.add(DateTime.utc_now(), 3_600, :second) + ) codex_model = LLMDB.Model.new!(%{id: "codex", provider: :openai_codex}) diff --git a/test/support/fake_refresh_client.ex b/test/support/fake_refresh_client.ex new file mode 100644 index 0000000..7c04bdf --- /dev/null +++ b/test/support/fake_refresh_client.ex @@ -0,0 +1,15 @@ +defmodule Kodo.Test.FakeRefreshClient do + @moduledoc false + + def refresh(refresh_token, opts) do + agent = Keyword.fetch!(opts, :agent) + + response = + Agent.get_and_update(agent, fn %{responses: [response | remaining]} = state -> + {response, + %{state | responses: remaining, refresh_tokens: [refresh_token | state.refresh_tokens]}} + end) + + if is_function(response, 1), do: response.(refresh_token), else: response + end +end From b5dabde26ddd0ee5602d10f768bf224a5657a823 Mon Sep 17 00:00:00 2001 From: Darwin Wu Date: Thu, 10 Sep 2026 05:07:53 +0000 Subject: [PATCH 18/25] feat(integrations): verify ChatGPT subscription access --- lib/kodo/integrations.ex | 21 ++++++ lib/kodo/integrations/audit_event.ex | 1 + .../integrations/oauth_access_validation.ex | 67 +++++++++++++++++++ lib/kodo/integrations/oauth_refresh.ex | 4 +- lib/kodo_web/live/integrations_live.ex | 12 ++-- test/kodo/integrations/oauth_refresh_test.exs | 20 ++++++ test/kodo_web/live/integrations_live_test.exs | 27 ++++++++ test/support/fake_refresh_client.ex | 3 +- 8 files changed, 147 insertions(+), 8 deletions(-) create mode 100644 lib/kodo/integrations/oauth_access_validation.ex diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex index 765fca9..8c6ece6 100644 --- a/lib/kodo/integrations.ex +++ b/lib/kodo/integrations.ex @@ -481,6 +481,27 @@ defmodule Kodo.Integrations do def require_refresh_reauthorization(%Scope{}, %Integration{}, _error_code), do: {:error, :unsafe_refresh_error} + def fail_refresh(%Scope{user: user}, %Integration{} = claim) do + Repo.transaction(fn -> + case Repo.update_all(exact_refresh_claim_query(user.id, claim), + set: [ + refresh_claim_owner_id: nil, + refresh_claim_generation: nil, + refresh_lease_expires_at: nil, + updated_at: now() + ] + ) do + {1, nil} -> + integration = Repo.get_by!(Integration, id: claim.id, user_id: user.id) + audit!(user.id, integration, "refresh_failed") + integration + + {0, nil} -> + Repo.rollback(:stale_refresh_claim) + end + end) + end + def validation_succeeded(%Scope{} = scope, id, generation) do update_fenced(scope, id, generation, ["connected"], "validation_succeeded", %{ validation_status: "valid", diff --git a/lib/kodo/integrations/audit_event.ex b/lib/kodo/integrations/audit_event.ex index 9ecfa2c..b1e3fdd 100644 --- a/lib/kodo/integrations/audit_event.ex +++ b/lib/kodo/integrations/audit_event.ex @@ -20,6 +20,7 @@ defmodule Kodo.Integrations.AuditEvent do refresh_succeeded refresh_invalid_grant refresh_reauthorization_required + refresh_failed ) @primary_key {:id, :binary_id, autogenerate: true} diff --git a/lib/kodo/integrations/oauth_access_validation.ex b/lib/kodo/integrations/oauth_access_validation.ex new file mode 100644 index 0000000..3091d52 --- /dev/null +++ b/lib/kodo/integrations/oauth_access_validation.ex @@ -0,0 +1,67 @@ +defmodule Kodo.Integrations.OAuthAccessValidation do + @moduledoc "Validates ChatGPT subscription access through the bounded refresh contract." + + alias Kodo.Accounts.Scope + alias Kodo.Integrations + alias Kodo.Integrations.Integration + alias Kodo.Integrations.OAuthRefresh + + def start(%Scope{} = scope, %{id: id, credential_generation: generation}) do + Task.Supervisor.async_nolink(Kodo.ControlPlaneTaskSupervisor, fn -> + validate(scope, id, generation) + end) + end + + @doc false + def validate(%Scope{} = scope, id, generation, opts \\ []) do + with {:ok, integration} <- Integrations.get_integration(scope, id), + :ok <- admit(integration, generation) do + case OAuthRefresh.ensure_fresh(scope, integration, Keyword.put(opts, :force, true)) do + {:ok, refreshed} -> + scope + |> Integrations.validation_succeeded(refreshed.id, refreshed.credential_generation) + |> broadcast_result(refreshed) + + {:error, :provider_unavailable} = error -> + result = + Integrations.validation_unavailable( + scope, + integration.id, + integration.credential_generation, + "provider_unavailable" + ) + + _result = broadcast_result(result, integration) + error + + {:error, _reason} = error -> + error + end + end + end + + defp admit( + %Integration{ + provider: "openai_codex", + authentication_type: "oauth", + connection_status: "connected", + credential_generation: generation + }, + generation + ), + do: :ok + + defp admit(%Integration{}, _generation), do: {:error, :stale_credential_generation} + + defp broadcast_result({:ok, validated} = result, integration) do + Phoenix.PubSub.broadcast( + Kodo.PubSub, + "integration:#{integration.user_id}", + {:integration_validation_finished, validated.id, validated.credential_generation} + ) + + result + end + + defp broadcast_result(error, _integration), do: error +end diff --git a/lib/kodo/integrations/oauth_refresh.ex b/lib/kodo/integrations/oauth_refresh.ex index 0c35ec6..76705d4 100644 --- a/lib/kodo/integrations/oauth_refresh.ex +++ b/lib/kodo/integrations/oauth_refresh.ex @@ -16,7 +16,7 @@ defmodule Kodo.Integrations.OAuthRefresh do now = Keyword.get(opts, :now, DateTime.utc_now()) buffer = Keyword.get(opts, :refresh_buffer_seconds, @refresh_buffer_seconds) - if refresh_due?(integration, now, buffer) do + if Keyword.get(opts, :force, false) or refresh_due?(integration, now, buffer) do refresh(scope, integration, opts) else {:ok, integration} @@ -118,7 +118,7 @@ defmodule Kodo.Integrations.OAuthRefresh do current_result(scope, claim) {:error, _reason} -> - _result = Integrations.release_refresh_claim(scope, claim) + _result = Integrations.fail_refresh(scope, claim) {:error, :provider_unavailable} end end diff --git a/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex index 1487bb0..add2f5c 100644 --- a/lib/kodo_web/live/integrations_live.ex +++ b/lib/kodo_web/live/integrations_live.ex @@ -4,6 +4,7 @@ defmodule KodoWeb.IntegrationsLive do alias Kodo.Integrations alias Kodo.Integrations.APIKeyValidation alias Kodo.Integrations.DeviceAuthorization + alias Kodo.Integrations.OAuthAccessValidation @provider_configs [ %{ @@ -454,7 +455,11 @@ defmodule KodoWeb.IntegrationsLive do end defp start_validation(socket, integration) do - task = APIKeyValidation.start(socket.assigns.current_scope, integration) + task = + case integration.authentication_type do + "oauth" -> OAuthAccessValidation.start(socket.assigns.current_scope, integration) + "api_key" -> APIKeyValidation.start(socket.assigns.current_scope, integration) + end update(socket, :validation_tasks, fn tasks -> Map.put(tasks, task.ref, {integration.id, integration.credential_generation}) @@ -897,10 +902,7 @@ defmodule KodoWeb.IntegrationsLive do Activate