diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex
index d52b81e..4070e2b 100644
--- a/lib/kodo/integrations.ex
+++ b/lib/kodo/integrations.ex
@@ -24,7 +24,16 @@ 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
+ @refresh_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
@@ -130,6 +139,112 @@ 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 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 ->
+ lock_user!(user.id)
+ integration = lock_owned_integration!(user.id, claim.integration_id)
+ attempt = terminalize_claimed_device_authorization(user.id, claim, "failed", error_code)
+ 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
@@ -165,6 +280,36 @@ 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 -> insert_oauth_integration(user.id, changeset) end)
+ else
+ {:error, changeset}
+ end
+ end
+
+ def create_oauth_integration(%Scope{}, _provider, _opts),
+ do: {:error, :authentication_type_mismatch}
+
+ defp insert_oauth_integration(user_id, changeset) do
+ lock_provider_identity(user_id, "openai_codex")
+
+ case Repo.insert(changeset) do
+ {:ok, integration} -> integration
+ {:error, reason} -> Repo.rollback(reason)
+ end
+ end
+
def connect(scope, provider, authentication_type, credentials, opts \\ [])
def connect(%Scope{user: user}, provider, "api_key", credentials, opts) do
@@ -251,6 +396,116 @@ defmodule Kodo.Integrations do
)
end
+ @doc "Claims one generation of an OAuth integration for a bounded refresh operation."
+ # The boolean terms below are one atomic database fence, not control-flow branches.
+ # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
+ 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 == "connected" 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 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 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",
@@ -313,7 +568,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()
@@ -417,6 +675,279 @@ 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
+
+ # The boolean terms below are one atomic database fence, not control-flow branches.
+ # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
+ 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 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]} ->
+ {attempt, payload} = decrypt_device_authorization(attempt)
+
+ case validate_device_authorization_exchange_payload(payload) do
+ :ok -> {attempt, payload}
+ {:error, reason} -> Repo.rollback(reason)
+ 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
+
+ # The boolean terms below are one atomic database fence, not control-flow branches.
+ # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
+ 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
+
+ # The boolean terms below are one atomic database fence, not control-flow branches.
+ # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
+ 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
@@ -504,7 +1035,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
@@ -560,11 +1097,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"
@@ -634,7 +1213,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)
@@ -654,6 +1236,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/lib/kodo/integrations/audit_event.ex b/lib/kodo/integrations/audit_event.ex
index 665ef20..b1e3fdd 100644
--- a/lib/kodo/integrations/audit_event.ex
+++ b/lib/kodo/integrations/audit_event.ex
@@ -14,9 +14,13 @@ 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
+ refresh_reauthorization_required
+ refresh_failed
)
@primary_key {:id, :binary_id, autogenerate: true}
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/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/lib/kodo/integrations/device_authorization_tokens.ex b/lib/kodo/integrations/device_authorization_tokens.ex
new file mode 100644
index 0000000..295cd45
--- /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 | microsecond: {0, 6}}}
+ 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/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/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
new file mode 100644
index 0000000..76705d4
--- /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 Keyword.get(opts, :force, false) or 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.fail_refresh(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/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/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/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/lib/kodo/integrations/req_refresh_client.ex b/lib/kodo/integrations/req_refresh_client.ex
new file mode 100644
index 0000000..8e78832
--- /dev/null
+++ b/lib/kodo/integrations/req_refresh_client.ex
@@ -0,0 +1,89 @@
+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
+ ])
+
+ request |> Req.post(options) |> classify_response()
+ end
+
+ def refresh(_refresh_token, _req_options), do: {:error, :oauth_refresh_response_invalid}
+
+ defp classify_response(result) do
+ case result 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
+
+ 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/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..5c24671 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)
@@ -42,8 +43,7 @@ defmodule Kodo.LLM.ProviderError do
end
def guidance(%__MODULE__{kind: :integration_required, provider: "openai_codex"}),
- do:
- "ChatGPT connections are not available yet. Choose models from connected API-key providers for the required roles, then start a new session."
+ do: "Connect or activate a ChatGPT Subscription account, then retry the turn."
def guidance(%__MODULE__{kind: :integration_required, provider: provider}),
do: "Connect or activate an account for #{provider_name(provider)}, then retry the turn."
@@ -91,11 +91,15 @@ defmodule Kodo.LLM.ProviderError do
def provider_help_url(%__MODULE__{provider: "openrouter"}),
do: "https://openrouter.ai/credits"
+ def provider_help_url(%__MODULE__{provider: "openai_codex"}),
+ do: "https://chatgpt.com/#settings/Subscription"
+
def provider_help_url(%__MODULE__{}), do: nil
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/lib/kodo_web/live/integrations_live.ex b/lib/kodo_web/live/integrations_live.ex
index 3c7186e..06de0cb 100644
--- a/lib/kodo_web/live/integrations_live.ex
+++ b/lib/kodo_web/live/integrations_live.ex
@@ -3,6 +3,8 @@ defmodule KodoWeb.IntegrationsLive do
alias Kodo.Integrations
alias Kodo.Integrations.APIKeyValidation
+ alias Kodo.Integrations.DeviceAuthorization
+ alias Kodo.Integrations.OAuthAccessValidation
@provider_configs [
%{
@@ -28,10 +30,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 +51,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 +212,68 @@ 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} <-
+ find_device_authorization(socket.assigns.device_authorizations, attempt_id),
+ {: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 +317,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 +427,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 +440,7 @@ defmodule KodoWeb.IntegrationsLive do
Map.take(integration, [
:id,
:provider,
+ :authentication_type,
:display_name,
:active,
:connection_status,
@@ -360,7 +453,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})
@@ -371,6 +468,69 @@ 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 find_device_authorization(authorizations, attempt_id) do
+ Enum.find_value(authorizations, fn {_integration_id, attempt} ->
+ if attempt.attempt_id == attempt_id, do: attempt
+ 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 +555,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 +813,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 +891,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")}