diff --git a/lib/kodo/integrations.ex b/lib/kodo/integrations.ex new file mode 100644 index 0000000..defb12e --- /dev/null +++ b/lib/kodo/integrations.ex @@ -0,0 +1,232 @@ +defmodule Kodo.Integrations do + @moduledoc "Owns scoped provider-integration credential lifecycle transitions." + + import Ecto.Changeset + import Ecto.Query + + alias Kodo.Accounts.Scope + alias Kodo.Integrations.CredentialEncryption + alias Kodo.Integrations.Integration + alias Kodo.Repo + + @safe_validation_errors ~w(network_error timeout tls_error provider_unavailable rate_limited) + + def list_integrations(%Scope{user: user}) do + Integration + |> where([integration], integration.user_id == ^user.id) + |> order_by([integration], asc: integration.provider) + |> Repo.all() + end + + def get_integration(%Scope{user: user}, id) do + with {:ok, id} <- Ecto.UUID.cast(id), + %Integration{} = integration <- Repo.get_by(Integration, id: id, user_id: user.id) do + {:ok, integration} + else + _missing -> {:error, :integration_not_found} + end + end + + def get_integration_by_provider(%Scope{user: user}, provider) do + case Repo.get_by(Integration, user_id: user.id, provider: provider) do + %Integration{} = integration -> {:ok, integration} + nil -> {:error, :integration_not_found} + end + end + + def connect(%Scope{user: user}, provider, authentication_type, credentials, opts \\ []) do + integration = %Integration{id: Ecto.UUID.generate(), user_id: user.id} + + changeset = + Integration.create_changeset(integration, %{ + provider: provider, + authentication_type: authentication_type + }) + + with true <- changeset.valid?, + {:ok, encrypted} <- CredentialEncryption.encrypt(apply_changes(changeset), credentials) do + changeset + |> change( + Map.merge(encrypted, %{ + connection_status: "connected", + validation_status: "unverified", + credential_generation: 1, + expires_at: opts[:expires_at] + }) + ) + |> Integration.constraint_changeset() + |> Repo.insert() + else + false -> {:error, changeset} + {:error, _reason} = error -> error + end + end + + def replace_credentials(%Scope{} = scope, id, generation, credentials, opts \\ []) do + install_credentials( + scope, + id, + generation, + credentials, + opts, + Integration.connection_statuses(), + :any + ) + end + + def oauth_succeeded(%Scope{} = scope, id, generation, credentials, opts \\ []) do + install_credentials( + scope, + id, + generation, + credentials, + opts, + Integration.connection_statuses(), + "oauth" + ) + end + + def refresh_succeeded(%Scope{} = scope, id, generation, credentials, opts \\ []) do + install_credentials( + scope, + id, + generation, + credentials, + opts, + ~w(connected reauthorization_required), + "oauth" + ) + end + + def validation_succeeded(%Scope{} = scope, id, generation) do + update_fenced(scope, id, generation, ["connected"], %{ + validation_status: "valid", + validated_at: now(), + validation_error_code: nil + }) + end + + def validation_invalid(%Scope{} = scope, id, generation) do + update_fenced(scope, id, generation, ["connected"], %{ + validation_status: "invalid", + validated_at: now(), + validation_error_code: "invalid_credentials" + }) + end + + def validation_unavailable(%Scope{} = scope, id, generation, error_code) + when error_code in @safe_validation_errors do + update_fenced(scope, id, generation, ["connected"], %{ + validation_status: "unavailable", + validated_at: now(), + validation_error_code: error_code + }) + end + + def validation_unavailable(%Scope{}, _id, _generation, _error_code), + do: {:error, :unsafe_validation_error} + + def refresh_invalid_grant(%Scope{} = scope, id, generation) do + with {:ok, integration} <- get_integration(scope, id), + true <- integration.authentication_type == "oauth" do + update_fenced(scope, id, generation, ["connected"], %{ + connection_status: "reauthorization_required", + validation_status: "unverified", + validated_at: nil, + validation_error_code: nil + }) + else + false -> {:error, :authentication_type_mismatch} + {:error, _reason} = error -> error + end + end + + def disconnect(%Scope{} = scope, id, generation) do + update_fenced(scope, id, generation, Integration.connection_statuses(), %{ + connection_status: "disconnected", + validation_status: "unverified", + encrypted_credentials: nil, + encryption_key_version: nil, + credential_format_version: nil, + credential_generation: generation + 1, + expires_at: nil, + validated_at: nil, + refreshed_at: nil, + validation_error_code: nil + }) + end + + def safe_validation_errors, do: @safe_validation_errors + + defp install_credentials( + scope, + id, + generation, + credentials, + opts, + allowed_connections, + authentication_type + ) do + with {:ok, integration} <- get_integration(scope, id), + :ok <- require_generation(integration, generation), + :ok <- require_authentication_type(integration, authentication_type), + {:ok, encrypted} <- CredentialEncryption.encrypt(integration, credentials) do + changes = + Map.merge(encrypted, %{ + connection_status: "connected", + validation_status: "unverified", + credential_generation: generation + 1, + expires_at: opts[:expires_at], + validated_at: nil, + refreshed_at: opts[:refreshed_at], + validation_error_code: nil + }) + + update_fenced(scope, id, generation, allowed_connections, changes) + else + {:error, _reason} = error -> error + end + end + + defp require_generation(%Integration{credential_generation: generation}, generation), do: :ok + defp require_generation(%Integration{}, _generation), do: {:error, :stale_credential_generation} + + defp require_authentication_type(%Integration{}, :any), do: :ok + + defp require_authentication_type(%Integration{authentication_type: type}, type), do: :ok + + defp require_authentication_type(%Integration{}, _type), + do: {:error, :authentication_type_mismatch} + + defp update_fenced(%Scope{user: user}, id, generation, allowed_connections, changes) + when is_integer(generation) and generation >= 0 do + case Ecto.UUID.cast(id) do + {:ok, id} -> + Repo.transaction(fn -> + execute_fenced_update(user.id, id, generation, allowed_connections, changes) + end) + + :error -> + {:error, :stale_credential_generation} + end + end + + defp update_fenced(%Scope{}, _id, _generation, _allowed_connections, _changes), + do: {:error, :stale_credential_generation} + + defp execute_fenced_update(user_id, id, generation, allowed_connections, changes) do + query = + from integration in Integration, + where: + integration.id == ^id and integration.user_id == ^user_id and + integration.credential_generation == ^generation and + integration.connection_status in ^allowed_connections + + case Repo.update_all(query, set: Map.to_list(Map.put(changes, :updated_at, now()))) do + {1, nil} -> Repo.get_by!(Integration, id: id, user_id: user_id) + {0, nil} -> Repo.rollback(:stale_credential_generation) + end + end + + defp now, do: DateTime.utc_now() +end diff --git a/test/kodo/integrations_test.exs b/test/kodo/integrations_test.exs new file mode 100644 index 0000000..867ef68 --- /dev/null +++ b/test/kodo/integrations_test.exs @@ -0,0 +1,226 @@ +defmodule Kodo.IntegrationsTest do + use Kodo.DataCase, async: true + + alias Kodo.AccountsFixtures + alias Kodo.Integrations + alias Kodo.Integrations.CredentialEncryption + + describe "scoped credential lifecycle" do + setup do + %{scope: AccountsFixtures.user_scope_fixture()} + end + + test "connects, lists, and retrieves only owned integration metadata", %{scope: scope} do + other_scope = AccountsFixtures.user_scope_fixture() + + assert {:ok, integration} = + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "owner-secret"}) + + assert integration.user_id == scope.user.id + assert integration.connection_status == "connected" + assert integration.validation_status == "unverified" + assert integration.credential_generation == 1 + assert {:ok, %{"api_key" => "owner-secret"}} = CredentialEncryption.decrypt(integration) + + assert [listed] = Integrations.list_integrations(scope) + assert listed.id == integration.id + assert Integrations.list_integrations(other_scope) == [] + + assert {:error, :integration_not_found} = + Integrations.get_integration(other_scope, integration.id) + + assert {:error, :integration_not_found} = + Integrations.get_integration_by_provider(other_scope, "openai") + end + + test "enforces one integration per user and provider", %{scope: scope} do + assert {:ok, _integration} = connect(scope) + assert {:error, changeset} = connect(scope) + assert "has already been taken" in errors_on(changeset).user_id + end + + test "replaces credentials with a new nonce and advances the generation", %{scope: scope} do + assert {:ok, integration} = connect(scope) + original_ciphertext = integration.encrypted_credentials + + assert {:ok, replaced} = + Integrations.replace_credentials( + scope, + integration.id, + integration.credential_generation, + %{"api_key" => "replacement-secret"} + ) + + assert replaced.credential_generation == 2 + refute replaced.encrypted_credentials == original_ciphertext + + assert {:ok, %{"api_key" => "replacement-secret"}} = + CredentialEncryption.decrypt(replaced) + + assert {:error, :stale_credential_generation} = + Integrations.replace_credentials( + scope, + replaced.id, + 1, + %{"api_key" => "stale-secret"} + ) + end + + test "records fenced validation outcomes without changing credential generation", %{ + scope: scope + } do + assert {:ok, integration} = connect(scope) + generation = integration.credential_generation + + assert {:ok, invalid} = + Integrations.validation_invalid(scope, integration.id, generation) + + assert invalid.validation_status == "invalid" + assert invalid.validation_error_code == "invalid_credentials" + assert invalid.credential_generation == generation + + assert {:ok, valid} = Integrations.validation_succeeded(scope, integration.id, generation) + assert valid.validation_status == "valid" + assert is_nil(valid.validation_error_code) + + assert {:ok, unavailable} = + Integrations.validation_unavailable(scope, integration.id, generation, "timeout") + + assert unavailable.validation_status == "unavailable" + assert unavailable.validation_error_code == "timeout" + + assert {:error, :unsafe_validation_error} = + Integrations.validation_unavailable( + scope, + integration.id, + generation, + "provider body with secret" + ) + end + + test "disconnects without a provider call and rejects delayed updates", %{scope: scope} do + assert {:ok, integration} = connect(scope) + generation = integration.credential_generation + + assert {:ok, disconnected} = + Integrations.disconnect(scope, integration.id, generation) + + assert disconnected.connection_status == "disconnected" + assert disconnected.validation_status == "unverified" + assert disconnected.credential_generation == generation + 1 + assert is_nil(disconnected.encrypted_credentials) + assert is_nil(disconnected.encryption_key_version) + assert is_nil(disconnected.credential_format_version) + + assert {:error, :stale_credential_generation} = + Integrations.validation_succeeded(scope, integration.id, generation) + end + + test "retains provisional OAuth credentials until a fenced refresh succeeds", %{scope: scope} do + assert {:ok, integration} = + Integrations.connect(scope, "openai_codex", "oauth", %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh", + "account_id" => "account" + }) + + generation = integration.credential_generation + + assert {:ok, reauthorization} = + Integrations.refresh_invalid_grant(scope, integration.id, generation) + + assert reauthorization.connection_status == "reauthorization_required" + assert reauthorization.validation_status == "unverified" + assert reauthorization.encrypted_credentials == integration.encrypted_credentials + assert reauthorization.credential_generation == generation + + assert {:ok, refreshed} = + Integrations.refresh_succeeded( + scope, + integration.id, + generation, + %{ + "access_token" => "new-access", + "refresh_token" => "new-refresh", + "account_id" => "account" + }, + refreshed_at: DateTime.utc_now() + ) + + assert refreshed.connection_status == "connected" + assert refreshed.credential_generation == generation + 1 + + assert {:ok, %{"access_token" => "new-access"}} = + CredentialEncryption.decrypt(refreshed) + |> then(fn {:ok, payload} -> {:ok, Map.take(payload, ["access_token"])} end) + end + + test "rejects OAuth-only transitions for API-key integrations", %{scope: scope} do + assert {:ok, integration} = connect(scope) + + assert {:error, :authentication_type_mismatch} = + Integrations.refresh_invalid_grant( + scope, + integration.id, + integration.credential_generation + ) + + assert {:error, :authentication_type_mismatch} = + Integrations.oauth_succeeded( + scope, + integration.id, + integration.credential_generation, + %{"access_token" => "wrong-route"} + ) + end + + test "installs a generation-fenced OAuth authorization after disconnection", %{scope: scope} do + assert {:ok, integration} = + Integrations.connect(scope, "openai_codex", "oauth", %{ + "access_token" => "old-access", + "refresh_token" => "old-refresh" + }) + + assert {:ok, disconnected} = + Integrations.disconnect( + scope, + integration.id, + integration.credential_generation + ) + + assert {:ok, connected} = + Integrations.oauth_succeeded( + scope, + integration.id, + disconnected.credential_generation, + %{"access_token" => "authorized", "refresh_token" => "refresh"} + ) + + assert connected.connection_status == "connected" + assert connected.validation_status == "unverified" + assert connected.credential_generation == disconnected.credential_generation + 1 + end + + test "rejects forged and cross-user generation-fenced transitions", %{scope: scope} do + other_scope = AccountsFixtures.user_scope_fixture() + assert {:ok, integration} = connect(scope) + + assert {:error, :stale_credential_generation} = + Integrations.disconnect( + other_scope, + integration.id, + integration.credential_generation + ) + + assert {:error, :stale_credential_generation} = + Integrations.disconnect(scope, Ecto.UUID.generate(), 0) + + assert {:ok, current} = Integrations.get_integration(scope, integration.id) + assert current.connection_status == "connected" + end + end + + defp connect(scope) do + Integrations.connect(scope, "openai", "api_key", %{"api_key" => "provider-secret"}) + end +end diff --git a/test/kodo/sessions/active_session_test.exs b/test/kodo/sessions/active_session_test.exs index a4a58b5..6f47a84 100644 --- a/test/kodo/sessions/active_session_test.exs +++ b/test/kodo/sessions/active_session_test.exs @@ -449,8 +449,6 @@ defmodule Kodo.Sessions.ActiveSessionTest do %{type: "session_status_changed", payload: %{"status" => "completed"}}} assert_receive {:DOWN, ^ref, :process, ^pid, :normal} - _ = :sys.get_state(Kodo.SessionRegistry) - assert Registry.lookup(Kodo.SessionRegistry, session.id) == [] end events = Sessions.events_after(session.id)