diff --git a/apps/control_plane/lib/workharbor/config_trust.ex b/apps/control_plane/lib/workharbor/config_trust.ex index e44ac673..b1098a45 100644 --- a/apps/control_plane/lib/workharbor/config_trust.ex +++ b/apps/control_plane/lib/workharbor/config_trust.ex @@ -25,46 +25,49 @@ defmodule WorkHarbor.ConfigTrust do def approve(%Principal{} = principal, descriptor) do with :ok <- Accounts.authorize(principal, :config_approve), {:ok, normalized, bytes, digest} <- normalize(descriptor) do - transaction(fn -> - authorize!(principal, :config_approve) - - case Repo.get_by(Approval, [owner_id: principal.owner_id, descriptor_sha256: digest], - log: false - ) do - nil -> - approval = + result = + transaction(fn -> + authorize!(principal, :config_approve) + + case Repo.get_by(Approval, [owner_id: principal.owner_id, descriptor_sha256: digest], + log: false + ) do + nil -> + approval = + Repo.insert!( + %Approval{ + owner_id: principal.owner_id, + descriptor: normalized, + descriptor_bytes: bytes, + descriptor_sha256: digest, + approved_at: Accounts.now() + }, + log: false + ) + Repo.insert!( - %Approval{ + %AuditEvent{ owner_id: principal.owner_id, - descriptor: normalized, - descriptor_bytes: bytes, - descriptor_sha256: digest, - approved_at: Accounts.now() + event: "revision_approved", + action: "config_approve", + channel: channel(principal), + outcome: "recorded", + code: "ok", + metadata: %{"approval_id" => approval.id} }, log: false ) - Repo.insert!( - %AuditEvent{ - owner_id: principal.owner_id, - event: "revision_approved", - action: "config_approve", - channel: channel(principal), - outcome: "recorded", - code: "ok", - metadata: %{"approval_id" => approval.id} - }, - log: false - ) - - {:ok, approval} - - approval -> - if approval.descriptor_bytes == bytes and approval.descriptor == normalized, - do: {:ok, approval}, - else: Repo.rollback(:trust_conflict) - end - end) + {:ok, approval} + + approval -> + if approval.descriptor_bytes == bytes and approval.descriptor == normalized, + do: {:ok, approval}, + else: Repo.rollback(:trust_conflict) + end + end) + + with {:error, {:denied, action}} <- result, do: Accounts.deny(principal, action) end end @@ -109,7 +112,7 @@ defmodule WorkHarbor.ConfigTrust do end defp authorize!(principal, action) do - if Accounts.authorize(principal, action) != :ok, do: Repo.rollback(:forbidden) + if Accounts.authorize(principal, action) != :ok, do: Repo.rollback({:denied, action}) end defp channel(%Principal{kind: "owner_browser"}), do: "browser" diff --git a/apps/control_plane/lib/workharbor/host_connections.ex b/apps/control_plane/lib/workharbor/host_connections.ex index 71f33044..dda18f69 100644 --- a/apps/control_plane/lib/workharbor/host_connections.ex +++ b/apps/control_plane/lib/workharbor/host_connections.ex @@ -6,6 +6,11 @@ defmodule WorkHarbor.HostConnections do alias WorkHarbor.{Repo, Hosts, Accounts} alias WorkHarbor.HostConnections.Protocol alias WorkHarbor.Hosts.Peer + # Contract cadence is 10 s. Each heartbeat costs a transaction and an undeletable audit + # row, so arrivals are limited by a token bucket (capacity 3, one token per 5 s) measured + # on receive time, not on the previous commit; the first heartbeat after hello is free. + @heartbeat_refill_ms 5000 + @heartbeat_capacity 3 def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) def open(peer, socket), do: call(__MODULE__, {:open, peer, socket}) def receive_hello(ref, message), do: peer_call(ref, {:hello, message}) @@ -116,6 +121,8 @@ defmodule WorkHarbor.HostConnections do negotiated: false, opened: mono, seen: nil, + hb_arrived: nil, + hb_tokens: 0, lease: nil, expiry: expiry, expiry_timer: expiry_timer @@ -213,44 +220,9 @@ defmodule WorkHarbor.HostConnections do end defp peer_message(state, ref, entry, {:heartbeat, message}) do - with true <- entry.negotiated, - {:ok, %{"kind" => "heartbeat"}} <- - Protocol.validate(message, entry.peer.owner_id, entry.peer.host_id) do - case Hosts.transaction(fn -> - current_slot!(entry) - if not Hosts.current_peer?(entry.peer), do: Repo.rollback(:unauthorized_host) - now = DateTime.from_unix!(Accounts.now()) |> DateTime.to_iso8601() - lease = DateTime.from_unix!(Accounts.now() + 30) |> DateTime.to_iso8601() - - Hosts.sql("UPDATE hosts SET last_seen=?,lease_expires_at=? WHERE id=?", [ - now, - lease, - entry.peer.host_id - ]) - - Hosts.sql( - "UPDATE host_connection_slots SET state='connected' WHERE host_id=? AND connection_id=?", - [entry.peer.host_id, entry.id] - ) - - Hosts.audit( - entry.peer.owner_id, - entry.peer.host_id, - "host_heartbeat", - "host_heartbeat", - "authenticated" - ) - end) do - {:ok, _} -> - entry = %{entry | seen: monotonic(), lease: make_ref()} - Process.send_after(self(), {:stale, ref, entry.lease}, 30000) - {:reply, {:ok, ack(entry, "heartbeat_ack")}, put_entry(state, ref, entry)} - - _ -> - {:reply, {:error, :persistence_failed}, close(state, ref, :persistence_failed)} - end - else - _ -> {:reply, {:error, :invalid_connection_message}, close(state, ref, :invalid_message)} + case take_heartbeat_token(entry, monotonic()) do + {:ok, entry} -> heartbeat(state, ref, entry, message) + :limited -> {:reply, {:error, :heartbeat_too_fast}, close(state, ref, :heartbeat_too_fast)} end end @@ -429,6 +401,63 @@ defmodule WorkHarbor.HostConnections do defp fresh?(entry), do: entry.negotiated and not is_nil(entry.seen) and not stale?(entry) and valid_entry?(entry) + defp heartbeat(state, ref, entry, message) do + with true <- entry.negotiated, + {:ok, %{"kind" => "heartbeat"}} <- + Protocol.validate(message, entry.peer.owner_id, entry.peer.host_id) do + case Hosts.transaction(fn -> + current_slot!(entry) + if not Hosts.current_peer?(entry.peer), do: Repo.rollback(:unauthorized_host) + now = DateTime.from_unix!(Accounts.now()) |> DateTime.to_iso8601() + lease = DateTime.from_unix!(Accounts.now() + 30) |> DateTime.to_iso8601() + + Hosts.sql("UPDATE hosts SET last_seen=?,lease_expires_at=? WHERE id=?", [ + now, + lease, + entry.peer.host_id + ]) + + Hosts.sql( + "UPDATE host_connection_slots SET state='connected' WHERE host_id=? AND connection_id=?", + [entry.peer.host_id, entry.id] + ) + + Hosts.audit( + entry.peer.owner_id, + entry.peer.host_id, + "host_heartbeat", + "host_heartbeat", + "authenticated" + ) + end) do + {:ok, _} -> + entry = %{entry | seen: monotonic(), lease: make_ref()} + Process.send_after(self(), {:stale, ref, entry.lease}, 30000) + {:reply, {:ok, ack(entry, "heartbeat_ack")}, put_entry(state, ref, entry)} + + _ -> + {:reply, {:error, :persistence_failed}, close(state, ref, :persistence_failed)} + end + else + _ -> {:reply, {:error, :invalid_connection_message}, close(state, ref, :invalid_message)} + end + end + + defp take_heartbeat_token(%{hb_arrived: nil} = entry, now), + do: {:ok, %{entry | hb_arrived: now, hb_tokens: @heartbeat_capacity * @heartbeat_refill_ms}} + + defp take_heartbeat_token(entry, now) do + tokens = + min( + @heartbeat_capacity * @heartbeat_refill_ms, + entry.hb_tokens + max(0, now - entry.hb_arrived) + ) + + if tokens >= @heartbeat_refill_ms, + do: {:ok, %{entry | hb_arrived: now, hb_tokens: tokens - @heartbeat_refill_ms}}, + else: :limited + end + defp stale?(entry), do: monotonic() - (entry.seen || entry.opened) >= 30000 defp monotonic, diff --git a/apps/control_plane/lib/workharbor/hosts.ex b/apps/control_plane/lib/workharbor/hosts.ex index ffe5ada9..6a236213 100644 --- a/apps/control_plane/lib/workharbor/hosts.ex +++ b/apps/control_plane/lib/workharbor/hosts.ex @@ -13,7 +13,7 @@ defmodule WorkHarbor.Hosts do alias WorkHarbor.Hosts.{Issuer, Peer} def issue_enrollment(principal, host_id) do - with :ok <- Accounts.authorize(principal, :host_enrollment_issue) do + with_authorization(principal, :host_enrollment_issue, fn -> transaction(fn -> authorize!(principal, :host_enrollment_issue) @@ -39,6 +39,11 @@ defmodule WorkHarbor.Hosts do code = "whe1_" <> Base.url_encode64(:crypto.strong_rand_bytes(32), padding: false) now = Accounts.now() + sql( + "UPDATE host_enrollments SET revoked_at=? WHERE host_id=? AND consumed_at IS NULL AND revoked_at IS NULL", + [now, host.id] + ) + sql( "INSERT INTO host_enrollments(id,host_id,owner_id,code_hash,issued_at,expires_at) VALUES(?,?,?,?,?,?)", [Ecto.UUID.generate(), host.id, host.owner_id, digest(code), now, now + 300] @@ -50,7 +55,7 @@ defmodule WorkHarbor.Hosts do %{host_id: host.id, owner_id: host.owner_id, code: code, expires_at: now + 300} end) - end + end) end def redeem(code, csr) do @@ -236,7 +241,7 @@ defmodule WorkHarbor.Hosts do def activate(_), do: {:error, :unauthorized_host} def revoke(principal, host_id) do - with :ok <- Accounts.authorize(principal, :host_revoke) do + with_authorization(principal, :host_revoke, fn -> result = transaction(fn -> authorize!(principal, :host_revoke) @@ -275,7 +280,7 @@ defmodule WorkHarbor.Hosts do do: Phoenix.PubSub.broadcast(WorkHarbor.PubSub, "host: " <> host_id, :identity_changed) normalize_ok(result) - end + end) end def status(principal, host_id) do @@ -371,8 +376,19 @@ defmodule WorkHarbor.Hosts do Repo.get_by(Host, id: id, owner_id: owner) || Repo.rollback(:not_found) end + # The in-transaction check rolls back its own denial audit, so the denial is + # surfaced as a marker and audited here, after the rollback (authentication.md section 4). defp authorize!(p, action), - do: if(Accounts.authorize(p, action) != :ok, do: Repo.rollback(:forbidden)) + do: if(Accounts.authorize(p, action) != :ok, do: Repo.rollback({:denied, action})) + + defp with_authorization(principal, action, fun) do + with :ok <- Accounts.authorize(principal, action) do + case fun.() do + {:error, {:denied, ^action}} -> Accounts.deny(principal, action) + result -> result + end + end + end defp channel(%{kind: "cli_owner"}), do: "cli" defp channel(_), do: "browser" diff --git a/apps/control_plane/test/workharbor/config_trust_test.exs b/apps/control_plane/test/workharbor/config_trust_test.exs index 4cd3ecb0..4309c14a 100644 --- a/apps/control_plane/test/workharbor/config_trust_test.exs +++ b/apps/control_plane/test/workharbor/config_trust_test.exs @@ -21,6 +21,18 @@ defmodule WorkHarbor.ConfigTrustTest do %{owner: owner, secret: secret, principal: principal} end + test "denial inside the transaction keeps its audit when authorization changes after the pre-check", + ctx do + expire_after_precheck(ctx.principal, :config_approve) + assert {:error, :forbidden} = ConfigTrust.approve(ctx.principal, DB.config()) + + assert WorkHarbor.Repo.query!( + "SELECT count(*) FROM audit_events WHERE event='authorization_denied' AND action='config_approve'", + [], + log: false + ).rows == [[1]] + end + test "exact approval repeats without a second approval or audit", ctx do descriptor = DB.config() assert {:ok, first} = ConfigTrust.approve(ctx.principal, descriptor) @@ -186,4 +198,31 @@ defmodule WorkHarbor.ConfigTrustTest do assert results |> Enum.map(fn {:ok, row} -> row.id end) |> Enum.uniq() |> length() == 1 assert DB.count("revision_trust_approvals") == 1 end + + # Counts clock reads of one pre-check, then expires the credential right after it, so only + # the in-transaction check can deny. + defp expire_after_precheck(principal, action) do + {:ok, counter} = Agent.start_link(fn -> 0 end) + real = Application.get_env(:workharbor, :clock) + now = WorkHarbor.Accounts.now() + + Application.put_env(:workharbor, :clock, fn -> + Agent.update(counter, &(&1 + 1)) + now + end) + + :ok = WorkHarbor.Accounts.authorize(principal, action) + reads = Agent.get(counter, & &1) + Agent.update(counter, fn _ -> 0 end) + + Application.put_env(:workharbor, :clock, fn -> + if Agent.get_and_update(counter, &{&1 + 1, &1 + 1}) > reads, do: now + 10_000_000, else: now + end) + + on_exit(fn -> + if real, + do: Application.put_env(:workharbor, :clock, real), + else: Application.delete_env(:workharbor, :clock) + end) + end end diff --git a/apps/control_plane/test/workharbor/host_connections_test.exs b/apps/control_plane/test/workharbor/host_connections_test.exs index 9d78903f..b9e1b17f 100644 --- a/apps/control_plane/test/workharbor/host_connections_test.exs +++ b/apps/control_plane/test/workharbor/host_connections_test.exs @@ -95,6 +95,46 @@ defmodule WorkHarbor.HostConnectionsTest do {operation, binding} end + test "first heartbeat after hello passes and a catch-up burst within capacity stays connected", + c do + ref = connect(c) + Agent.update(c.mono, fn _ -> 10_000 end) + + for _ <- 1..3 do + assert {:ok, %{"kind" => "heartbeat_ack"}} = + HostConnections.heartbeat(ref, message(c, "heartbeat")) + end + + refute_received {:host_connection_close, _} + end + + test "sustained heartbeat burst beyond capacity closes without a heartbeat write", c do + ref = connect(c) + + count = fn -> + Repo.query!("SELECT count(*) FROM audit_events WHERE event='host_heartbeat'").rows + end + + for _ <- 1..3 do + assert {:ok, _} = HostConnections.heartbeat(ref, message(c, "heartbeat")) + end + + before = count.() + assert {:error, :heartbeat_too_fast} = HostConnections.heartbeat(ref, message(c, "heartbeat")) + assert_receive {:host_connection_close, :heartbeat_too_fast} + assert count.() == before + end + + test "tokens refill at one per five seconds measured from arrival", c do + ref = connect(c) + + for _ <- 1..3, do: assert({:ok, _} = HostConnections.heartbeat(ref, message(c, "heartbeat"))) + + Agent.update(c.mono, fn _ -> 5000 end) + assert {:ok, _} = HostConnections.heartbeat(ref, message(c, "heartbeat")) + assert {:error, :heartbeat_too_fast} = HostConnections.heartbeat(ref, message(c, "heartbeat")) + end + test "authenticated heartbeat becomes stale exactly30seconds without changing history", c do ref = connect(c) {operation, binding} = runtime(c) @@ -234,6 +274,7 @@ defmodule WorkHarbor.HostConnectionsTest do next = connect(c) send(c.server, {:stale, elem(ref, 1), old_lease}) assert {:ok, "connected"} = HostConnections.host_state(c.peer.host_id) + Agent.update(c.mono, fn _ -> 35000 end) assert {:ok, _} = HostConnections.heartbeat(next, message(c, "heartbeat")) end @@ -285,6 +326,7 @@ defmodule WorkHarbor.HostConnectionsTest do "CREATE TRIGGER fail_liveness BEFORE UPDATE ON host_connection_slots BEGIN SELECT RAISE(ABORT,'SECRET_SENTINEL'); END" ) + Agent.update(c.mono, fn _ -> 5000 end) assert {:error, :persistence_failed} = HostConnections.heartbeat(ref, message(c, "heartbeat")) assert {:ok, %{observed_state: "unknown"}} = diff --git a/apps/control_plane/test/workharbor/host_identity_test.exs b/apps/control_plane/test/workharbor/host_identity_test.exs index ba8beaee..549fc5e9 100644 --- a/apps/control_plane/test/workharbor/host_identity_test.exs +++ b/apps/control_plane/test/workharbor/host_identity_test.exs @@ -67,6 +67,16 @@ defmodule WorkHarbor.HostIdentityTest do refute inspect(Repo.all(WorkHarbor.Foundation.AuditEvent)) =~ issued.code end + test "issuing a code revokes the host's earlier unconsumed codes", c do + {:ok, first} = Hosts.issue_enrollment(c.principal, nil) + {:ok, second} = Hosts.issue_enrollment(c.principal, first.host_id) + {csr, _} = HostFixture.csr() + assert {:error, :invalid_enrollment} = Hosts.redeem(first.code, csr) + assert {:ok, enrolled} = Hosts.redeem(second.code, csr) + assert enrolled["host_id"] == first.host_id + assert {:ok, _} = Hosts.resolve_identity(Base.decode64!(enrolled["certificate"]), :active) + end + test "enrollment exact expiry issues no identities", c do now = Accounts.now() Application.put_env(:workharbor, :clock, fn -> now end) @@ -105,6 +115,38 @@ defmodule WorkHarbor.HostIdentityTest do assert WorkHarbor.FoundationDatabase.count("host_identities") == 1 end + test "denied host actions keep their authorization_denied audit", c do + {:ok, token, _} = Accounts.login(c.principal.owner_id, c.secret, "dashboard") + {:ok, dashboard} = Accounts.authenticate(token) + + denied = fn action -> + Repo.query!( + "SELECT count(*) FROM audit_events WHERE event='authorization_denied' AND action=?", + [action], + log: false + ).rows + end + + assert denied.("host_enrollment_issue") == [[0]] + assert {:error, :forbidden} = Hosts.issue_enrollment(dashboard, nil) + assert denied.("host_enrollment_issue") == [[1]] + assert denied.("host_revoke") == [[0]] + assert {:error, :forbidden} = Hosts.revoke(dashboard, Ecto.UUID.generate()) + assert denied.("host_revoke") == [[1]] + end + + test "denial inside the transaction keeps its audit when authorization changes after the pre-check", + c do + expire_after_precheck(c.principal, :host_revoke) + assert {:error, :forbidden} = Hosts.revoke(c.principal, Ecto.UUID.generate()) + + assert Repo.query!( + "SELECT count(*) FROM audit_events WHERE event='authorization_denied' AND action='host_revoke'", + [], + log: false + ).rows == [[1]] + end + test "explicit host actions reject dashboard and cross owner before mutation", c do {:ok, token, _} = Accounts.login(c.principal.owner_id, c.secret, "dashboard") {:ok, dashboard} = Accounts.authenticate(token) @@ -482,4 +524,31 @@ defmodule WorkHarbor.HostIdentityTest do assert File.ls!(dir) == ["issuer.json"] end end + + # Counts clock reads of one pre-check, then expires the credential right after it, so only + # the in-transaction check can deny. + defp expire_after_precheck(principal, action) do + {:ok, counter} = Agent.start_link(fn -> 0 end) + real = Application.get_env(:workharbor, :clock) + now = WorkHarbor.Accounts.now() + + Application.put_env(:workharbor, :clock, fn -> + Agent.update(counter, &(&1 + 1)) + now + end) + + :ok = WorkHarbor.Accounts.authorize(principal, action) + reads = Agent.get(counter, & &1) + Agent.update(counter, fn _ -> 0 end) + + Application.put_env(:workharbor, :clock, fn -> + if Agent.get_and_update(counter, &{&1 + 1, &1 + 1}) > reads, do: now + 10_000_000, else: now + end) + + on_exit(fn -> + if real, + do: Application.put_env(:workharbor, :clock, real), + else: Application.delete_env(:workharbor, :clock) + end) + end end diff --git a/docs/development/ci-acceptance.json b/docs/development/ci-acceptance.json index 7ea0d607..271a6fa2 100644 --- a/docs/development/ci-acceptance.json +++ b/docs/development/ci-acceptance.json @@ -1095,7 +1095,14 @@ "WorkHarbor.HostIdentityTest/test issuer refuses final and ancestor symlinks before initialization writes", "WorkHarbor.HostIdentityTest/test native leaf and issuer profiles reject extra purposes critical and duplicate extensions", "WorkHarbor.HostIdentityTest/test issuer startup rejects signed critical path length and duration profile variants", + "WorkHarbor.HostConnectionsTest/test first heartbeat after hello passes and a catch-up burst within capacity stays connected", + "WorkHarbor.HostConnectionsTest/test sustained heartbeat burst beyond capacity closes without a heartbeat write", + "WorkHarbor.HostConnectionsTest/test tokens refill at one per five seconds measured from arrival", "WorkHarbor.HostConnectionsTest/test authenticated heartbeat becomes stale exactly30seconds without changing history", + "WorkHarbor.HostIdentityTest/test issuing a code revokes the host's earlier unconsumed codes", + "WorkHarbor.HostIdentityTest/test denied host actions keep their authorization_denied audit", + "WorkHarbor.HostIdentityTest/test denial inside the transaction keeps its audit when authorization changes after the pre-check", + "WorkHarbor.ConfigTrustTest/test denial inside the transaction keeps its audit when authorization changes after the pre-check", "WorkHarbor.HostConnectionsTest/test reconnect and replaced socket cannot refresh old runtime annotation", "WorkHarbor.HostConnectionsTest/test revocation and open socket certificate expiry fence existing connections", "WorkHarbor.HostConnectionsTest/test closed connection messages reject malformed scope fields versions and bounds", diff --git a/specs/006-whr-cli/contracts/host-enrollment.md b/specs/006-whr-cli/contracts/host-enrollment.md index bd46edc2..aa48d3a6 100644 --- a/specs/006-whr-cli/contracts/host-enrollment.md +++ b/specs/006-whr-cli/contracts/host-enrollment.md @@ -181,6 +181,10 @@ clock jumps on idle sockets. No message at/after wall expiry is admitted before watchdog close. Agent mirrors expiry and stops expired active/pending use. Heartbeat immediately after hello then every10s; acknowledge each heartbeat. +Rate limit: token bucket, capacity3, one token per5s, measured on monotonic receive time +(not previous commit); first heartbeat after hello free. Heartbeat arriving with no token +is refused without a heartbeat write and closes the connection (slot/unknown audit as for +any close). Without authenticated heartbeat for>=30s mark unknown and close stale socket. Only transport reconnect uses bounded1/2/4/8/16/30s backoff then30s maximum; no blind provider resends or command-history resets. Health requires authenticated hello