Files
buzz/crates/buzz-db/src/store/api_token.rs
Luke Tornquist bd73490418 ci: run PostgreSQL tests in isolated lane (#6730)
## Why

PR #6660 introduced four PostgreSQL-backed persistence tests and a
focused CI selector, but an exact test-name list does not automatically
cover future database tests. Broad ignored-test execution also exposed
shared-schema races and ambiguity between desired-state and
migration-applied schema expectations.

## What

- Establish a discoverable convention: PostgreSQL unit modules use
`postgres_tests`, PostgreSQL integration binaries use a `postgres_`
prefix, and unrelated external-infrastructure tests use an
`external_infra_` prefix.
- Add a dedicated nextest PostgreSQL profile and archive-backed CI job
covering the relevant crates without enumerating test names.
- Create a run-scoped desired-state source database and a unique
PostgreSQL database per test process for parallel-safe isolation, with
cleanup on success, failure, or interruption.
- Route destructive migration and migration-parity tests to clean
`template0` databases while desired-state tests clone the desired-state
database.
- Document discovery, schema modes, required role privileges, and a
bounded portable local runner.
- Keep the existing infrastructure-free unit-test jobs unchanged.

This is intentionally limited to test and CI harness behavior. It does
not move production database code, change persistence semantics, or
implement issue #20's broader shared test-utility refactor.

## Risk

Low production risk because all changes are confined to tests, CI
configuration, documentation, and test harness scripts.

Remaining operational risks:

- Cleanup retries `dropdb --force` five times across roughly four
seconds. Exhaustion warns with the database name but deliberately does
not mask the test result; individual PostgreSQL diagnostics are
suppressed.
- The seven-package boundary is duplicated between the runner and CI
archive and must remain synchronized if PostgreSQL tests move to a new
crate.
- Portable SHA fallback branches were exercised on Linux; no macOS Blox
workstation was needed for this Linux CI artifact.
- GitHub's unchanged generic Unit Tests job currently fails while
cold-linking `buzz-voice` because `sherpa-onnx-c-api` is absent. The
branch changes no voice/build/toolchain inputs; the exact parent
previously passed that job, and Blox passes the same infrastructure-free
suite (1,450/1,450). A prior-head one-job retry and the final-head run
both reproduced the hosted-runner failure.

## Verification

Author Blox workstation `2020088`:

- `cargo fmt --all -- --check`
- `just clippy`
- `just test-unit`: 1,450 passed; PostgreSQL tests remain skipped in
fast jobs
- PR #6660 focused tests: 4/4 passed through the final runner
- Full lane: 282/282 passed in 10.087s with a non-superuser role limited
to `CREATEDB`, `CREATEROLE`, and `pg_read_all_stats`
  - 272 desired-state tests
  - 10 migration-applied tests
  - 6 explicitly filtered external-infrastructure tests
- Cleanup fault injection: a deliberately underprivileged 282-test run
produced the expected 277 passes and 5 failures, and the post-run
catalog audit found zero lane databases after retry cleanup. The
restored successful run also left zero lane databases.

GitHub exact final head `92c231e48f299ea2af23817763f5ae80ac013d68`:

- [PostgreSQL
Tests](https://github.com/block/buzz/actions/runs/32789030203/job/97630827823):
282/282 passed across 10 binaries in 98.886s; 1,240 skipped, including 6
via the profile filter
- The shared relay/PostgreSQL archive predecessor completed successfully
- [Unit
Tests](https://github.com/block/buzz/actions/runs/32789030203/job/97626930420):
unrelated `sherpa-onnx-c-api` native-link failure described above
- Independent exact-head review on separate Blox workstation `2022762`:
no substantive findings after all initial findings were addressed. The
reviewer independently verified cleanup retry behavior, exit-status
preservation, shell portability, syntax, and a clean exact-head
worktree.

## References

- Stacked on #6660 at exact head
`561de54be4d9c2b622b7c2aa5b61bc3068f47e2a`
- https://github.com/TheSentinel454/buzz/issues/20
-
https://github.com/TheSentinel454/buzz/issues/20#issuecomment-5398526792

Generated with Codex.

## Update — August 24, 2026 review follow-up

- Centralized the repeated PostgreSQL test URL resolution in
crate-local, test-only helpers without introducing the broader shared
utility refactor from issue #20.
- Restored descriptive hybrid/Redis test function names. Structural
`external_infra_*_tests` modules now own exclusion, and the nextest
filter only recognizes module path segments.
- Added a three-second source guard that scans every Rust file and fails
CI when an ignored PostgreSQL test would be omitted or an
external-infrastructure test would be included. Fixture tests cover
accepted modules/binaries and both failure modes.
- Removed the two implementation-plan documents.

Performance profiling on Blox workstation `2027352` (same prebuilt
archive and cargo-nextest 0.9.143 for every comparison):

- Current per-test database model, 8 workers: 10.15–10.40s across three
runs. The 283 `createdb` calls and 284 `dropdb` calls consumed
28.85–29.46 aggregate seconds, about 39–40% of aggregate test-process
duration.
- Reusing one database per worker without cleanup: 6.55–6.84s, about 35%
faster, but all three runs failed because global matcher-queue tests
inherited a quiesced community from an earlier test; the failing test
varied with scheduling.
- Reusing workers with `TRUNCATE … RESTART IDENTITY CASCADE`: all 282
tests passed, but total time regressed to 10.47–10.67s. The 265 truncate
calls consumed 24.34–25.45 aggregate seconds, so truncation merely
replaced most clone/drop cost.
- Raising concurrency to 16 workers reduced one isolated-database run to
8.79s, but 24 workers exposed a cluster-global `pg_stat_activity` race.
The profile remains at the proven-safe 8 workers.

Decision: retain unique per-test databases. Harness cleanup uses `dropdb
--if-exists --force`; it drops the database rather than truncating
tables or deleting rows.

Verification at local branch head
`60086c1f6ee34a193c342255a5e8e6293b60e988`:

- Discovery inventory: exactly 282 intended tests; all 6 hybrid/Redis
tests present and structurally excluded.
- PR #6660 focused persistence tests: 4/4 passed.
- Full PostgreSQL lane: 282/282 passed three times; 1,240 skipped each
run; wall time 9.72–9.96s without profiling shims.
- Non-ignored tests with a desired-state database: 1,234/1,234 passed;
one mesh-demo timeout from the first run passed immediately in
isolation.
- `cargo fmt --all -- --check` and all-target/all-feature clippy for
`buzz-db`, `buzz-deletion`, and `buzz-relay` passed.

Final independent re-review at exact head
`42097c0136ab3fa8f0efe5720ae829c04575b9b6` on separate Blox workstation
`2028455`: no substantive residual findings. The reviewer independently
exercised ordinary strings, zero/one/three-hash raw strings, line/block
comments, and string-contained lookalikes, then reran the full 353-file
scan, shell syntax, Python AST parsing, and diff check.

Independent review found that the source guard's initial regular
expression could miss raw-string ignore reasons and treat a commented
attribute as real. Final head `42097c0136ab3fa8f0efe5720ae829c04575b9b6`
parses valid ordinary/raw Rust string literals only at attributes found
in comment-sanitized source; regression fixtures cover both cases. The
fixture suite, Python compilation, shell syntax, diff check, and full
353-file repository scan pass on Blox.

## Update — August 25, 2026 restack

PR #6660 merged, so this follow-up was rebased from its former exact
parent onto current `origin/main`
(`8d2d0ff5ad42733e9949442c4b6358d0ba87f9a8`). The final candidate head
is `05dcc2ab28948e3ab79bb44839a69f2ba44648a2`; the PR no longer carries
#6660's pre-squash history.

Fresh exact-head verification on Blox workstation `2028572`:

- Discovery guard passed across 353 Rust files; inventory remained
exactly 282 intended PostgreSQL tests.
- PR #6660 focused persistence tests: 4/4 passed.
- Full PostgreSQL lane: 282/282 passed three times; 1,243 skipped each
run; wall time 9.70–9.87s.
- `cargo fmt --all -- --check` and all-target/all-feature clippy for
`buzz-db`, `buzz-deletion`, and `buzz-relay` passed.

## Update — August 25, 2026 final hosted verification

- [GitHub PostgreSQL
Tests](https://github.com/block/buzz/actions/runs/32801679226/job/97667835080):
282/282 passed in 53.787s; 1,243 skipped.
- The final guard scanned all 353 Rust files successfully before the
hosted lane ran.
- The shared relay/archive prerequisite and infrastructure-free Unit
Tests job both passed at exact final head
`42097c0136ab3fa8f0efe5720ae829c04575b9b6`.

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Signed-off-by: tornquist <tornquist@squareup.com>
2026-09-01 12:07:28 -04:00

802 lines
27 KiB
Rust

//! API token CRUD operations.
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use uuid::Uuid;
use crate::error::{DbError, Result};
use crate::Db;
use buzz_core::CommunityId;
use buzz_datastore_tracing::datastore_span;
/// Create a new API token record. The caller is responsible for generating
/// the raw token and computing its SHA-256 hash.
///
/// `community_id` is row zero: every token is scoped to a community, derived
/// from the request's resolved tenant — never client-supplied here.
#[allow(clippy::too_many_arguments)]
pub async fn create_api_token(
pool: &PgPool,
community_id: Uuid,
token_hash: &[u8],
owner_pubkey: &[u8],
name: &str,
scopes: &[String],
channel_ids: Option<&[Uuid]>,
expires_at: Option<DateTime<Utc>>,
) -> Result<Uuid> {
let id = Uuid::new_v4();
let scopes_json =
serde_json::to_value(scopes).map_err(|e| DbError::InvalidData(e.to_string()))?;
// Serialize channel_ids; propagate errors rather than silently dropping to NULL.
let channel_ids_json: Option<serde_json::Value> = channel_ids
.map(|ids| {
serde_json::to_value(ids.iter().map(|id| id.to_string()).collect::<Vec<_>>())
.map_err(|e| DbError::InvalidData(format!("channel_ids serialization: {e}")))
})
.transpose()?;
sqlx::query(
r#"
INSERT INTO api_tokens
(community_id, id, token_hash, owner_pubkey, name, scopes, channel_ids, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
"#,
)
.bind(community_id)
.bind(id)
.bind(token_hash)
.bind(owner_pubkey)
.bind(name)
.bind(&scopes_json)
.bind(&channel_ids_json)
.bind(expires_at)
.execute(pool)
.await?;
Ok(id)
}
/// Atomic conditional INSERT: create a token only if the owner has fewer than 10 active tokens.
///
/// Uses a subquery so the check and insert are atomic --
/// no TOCTOU race between a separate count query and the insert.
///
/// The 10-token limit is per (community, owner) — a user's quota is scoped to
/// their community, never global.
///
/// Returns `Ok(Some(uuid))` on success, `Ok(None)` if the 10-token limit is exceeded.
#[allow(clippy::too_many_arguments)]
pub async fn create_api_token_if_under_limit(
pool: &PgPool,
community_id: Uuid,
token_hash: &[u8],
owner_pubkey: &[u8],
name: &str,
scopes: &[String],
channel_ids: Option<&[Uuid]>,
expires_at: Option<DateTime<Utc>>,
) -> Result<Option<Uuid>> {
let id = Uuid::new_v4();
let scopes_json =
serde_json::to_value(scopes).map_err(|e| DbError::InvalidData(e.to_string()))?;
let channel_ids_json: Option<serde_json::Value> = channel_ids
.map(|ids| {
serde_json::to_value(ids.iter().map(|id| id.to_string()).collect::<Vec<_>>())
.map_err(|e| DbError::InvalidData(format!("channel_ids serialization: {e}")))
})
.transpose()?;
// Conditional INSERT: only inserts if active (non-revoked, non-expired) token count < 10
// **for this (community, owner) pair**. The subquery and insert execute atomically --
// no separate count + insert race.
let result = sqlx::query(
r#"
INSERT INTO api_tokens
(community_id, id, token_hash, owner_pubkey, name, scopes, channel_ids, expires_at, created_by_self_mint)
SELECT $1, $2, $3, $4, $5, $6, $7, $8, TRUE
WHERE (
SELECT COUNT(*)
FROM api_tokens
WHERE community_id = $1
AND owner_pubkey = $9
AND revoked_at IS NULL
AND (expires_at IS NULL OR expires_at > NOW())
) < 10
"#,
)
.bind(community_id)
.bind(id)
.bind(token_hash)
.bind(owner_pubkey)
.bind(name)
.bind(&scopes_json)
.bind(&channel_ids_json)
.bind(expires_at)
.bind(owner_pubkey)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
// Limit exceeded -- the WHERE clause prevented the INSERT.
return Ok(None);
}
Ok(Some(id))
}
/// Look up an API token by its SHA-256 hash, **including revoked tokens**,
/// scoped to the request's community.
///
/// The lookup is keyed on `(community_id, token_hash)` — the same key the
/// storage UNIQUE index uses. This closes the row-44 conformance obligation:
/// a token minted in community A must never authorize in community B, even
/// if (by birthday-style collision or adversarial mint) the same hash exists
/// in both. The UNIQUE index is a *storage* guarantee; this `AND community_id`
/// clause is the *query* guarantee — both must hold for the property to be
/// load-bearing under all schemas.
///
/// Unlike [`crate::Db::get_api_token_by_hash`] (which filters `revoked_at IS NULL`),
/// this function returns the full record regardless of revocation status.
/// The relay layer uses this to return distinct `token_revoked` vs `invalid_token`
/// error responses rather than treating both as "not found".
pub async fn get_api_token_by_hash_including_revoked(
pool: &PgPool,
community_id: Uuid,
hash: &[u8],
) -> Result<Option<crate::ApiTokenRecord>> {
let row = sqlx::query(
r#"
SELECT id, token_hash, owner_pubkey, name, scopes, channel_ids,
created_at, expires_at, last_used_at, revoked_at
FROM api_tokens
WHERE community_id = $1 AND token_hash = $2
"#,
)
.bind(community_id)
.bind(hash)
.fetch_optional(pool)
.await?;
let row = match row {
None => return Ok(None),
Some(r) => r,
};
let id: Uuid = row.try_get("id")?;
let scopes_json: serde_json::Value = row.try_get("scopes")?;
let scopes: Vec<String> = serde_json::from_value(scopes_json)
.map_err(|e| DbError::InvalidData(format!("scopes JSON: {e}")))?;
let channel_ids: Option<Vec<Uuid>> = {
let raw: Option<serde_json::Value> = row.try_get("channel_ids")?;
match raw {
None => None,
Some(v) => {
let strings: Vec<String> = serde_json::from_value(v)
.map_err(|e| DbError::InvalidData(format!("channel_ids JSON: {e}")))?;
let uuids: std::result::Result<Vec<Uuid>, _> =
strings.iter().map(|s| s.parse::<Uuid>()).collect();
Some(uuids.map_err(|e| DbError::InvalidData(format!("channel_ids UUID: {e}")))?)
}
}
};
Ok(Some(crate::ApiTokenRecord {
id,
token_hash: row.try_get("token_hash")?,
owner_pubkey: row.try_get("owner_pubkey")?,
name: row.try_get("name")?,
scopes,
channel_ids,
created_at: row.try_get("created_at")?,
expires_at: row.try_get("expires_at")?,
last_used_at: row.try_get("last_used_at")?,
revoked_at: row.try_get("revoked_at")?,
}))
}
/// List all tokens (including revoked) for a (community, owner) pair,
/// ordered by creation time descending.
///
/// Returns the full [`crate::ApiTokenRecord`] including `token_hash`. Callers are
/// responsible for stripping `token_hash` before returning data to clients -- the
/// raw token value is never exposed after the initial mint response.
/// Used by `GET /api/tokens` to show a user their full token history.
pub async fn list_tokens_by_owner(
pool: &PgPool,
community_id: Uuid,
pubkey: &[u8],
) -> Result<Vec<crate::ApiTokenRecord>> {
let rows = sqlx::query(
r#"
SELECT id, token_hash, owner_pubkey, name, scopes, channel_ids,
created_at, expires_at, last_used_at, revoked_at
FROM api_tokens
WHERE community_id = $1 AND owner_pubkey = $2
ORDER BY created_at DESC
"#,
)
.bind(community_id)
.bind(pubkey)
.fetch_all(pool)
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let id: Uuid = row.try_get("id")?;
let scopes_json: serde_json::Value = row.try_get("scopes")?;
let scopes: Vec<String> = serde_json::from_value(scopes_json)
.map_err(|e| DbError::InvalidData(format!("scopes JSON: {e}")))?;
let channel_ids: Option<Vec<Uuid>> = {
let raw: Option<serde_json::Value> = row.try_get("channel_ids")?;
match raw {
None => None,
Some(v) => {
let strings: Vec<String> = serde_json::from_value(v)
.map_err(|e| DbError::InvalidData(format!("channel_ids JSON: {e}")))?;
let uuids: std::result::Result<Vec<Uuid>, _> =
strings.iter().map(|s| s.parse::<Uuid>()).collect();
Some(
uuids
.map_err(|e| DbError::InvalidData(format!("channel_ids UUID: {e}")))?,
)
}
}
};
out.push(crate::ApiTokenRecord {
id,
token_hash: row.try_get("token_hash")?,
owner_pubkey: row.try_get("owner_pubkey")?,
name: row.try_get("name")?,
scopes,
channel_ids,
created_at: row.try_get("created_at")?,
expires_at: row.try_get("expires_at")?,
last_used_at: row.try_get("last_used_at")?,
revoked_at: row.try_get("revoked_at")?,
});
}
Ok(out)
}
/// Revoke a single token by ID, scoped to (community, owner).
///
/// Only revokes if the token is in `community_id`, owned by `owner_pubkey`, and not already revoked.
/// Returns `true` if the token was revoked, `false` if not found, not owned, or already revoked.
pub async fn revoke_token(
pool: &PgPool,
community_id: Uuid,
id: Uuid,
owner_pubkey: &[u8],
revoked_by: &[u8],
) -> Result<bool> {
let result = sqlx::query(
r#"
UPDATE api_tokens
SET revoked_at = NOW(), revoked_by = $1
WHERE community_id = $2
AND id = $3
AND owner_pubkey = $4
AND revoked_at IS NULL
"#,
)
.bind(revoked_by)
.bind(community_id)
.bind(id)
.bind(owner_pubkey)
.execute(pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Revoke all active tokens for a (community, owner) pair.
///
/// Skips already-revoked tokens (idempotent). Returns the count of newly revoked tokens.
/// If all tokens are already revoked, returns 0 with no error.
pub async fn revoke_all_tokens(
pool: &PgPool,
community_id: Uuid,
owner_pubkey: &[u8],
revoked_by: &[u8],
) -> Result<u64> {
let result = sqlx::query(
r#"
UPDATE api_tokens
SET revoked_at = NOW(), revoked_by = $1
WHERE community_id = $2
AND owner_pubkey = $3
AND revoked_at IS NULL
"#,
)
.bind(revoked_by)
.bind(community_id)
.bind(owner_pubkey)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
/// Token summary returned by [`Db::list_active_tokens`].
#[derive(Debug, Clone)]
pub struct TokenSummary {
/// Unique token identifier.
pub id: Uuid,
/// Human-readable token name.
pub name: String,
/// Compressed public key bytes of the token owner.
pub owner_pubkey: Vec<u8>,
/// Permission scopes granted to this token.
pub scopes: Vec<String>,
/// When the token was created.
pub created_at: DateTime<Utc>,
/// Optional expiry timestamp; `None` means no expiry.
pub expires_at: Option<DateTime<Utc>>,
}
/// A full API token record.
#[derive(Debug, Clone)]
pub struct ApiTokenRecord {
/// Unique token identifier.
pub id: Uuid,
/// SHA-256 hash of the raw token value.
pub token_hash: Vec<u8>,
/// Compressed public key bytes of the token owner.
pub owner_pubkey: Vec<u8>,
/// Human-readable token name.
pub name: String,
/// Permission scopes granted to this token.
pub scopes: Vec<String>,
/// Optional channel ID restrictions.
pub channel_ids: Option<Vec<Uuid>>,
/// When the token was created.
pub created_at: DateTime<Utc>,
/// Optional expiry timestamp.
pub expires_at: Option<DateTime<Utc>>,
/// When the token was last used.
pub last_used_at: Option<DateTime<Utc>>,
/// When the token was revoked.
pub revoked_at: Option<DateTime<Utc>>,
}
fn parse_api_token_row(row: sqlx::postgres::PgRow) -> Result<ApiTokenRecord> {
let id: Uuid = row.try_get("id")?;
let scopes_json: serde_json::Value = row.try_get("scopes")?;
let scopes: Vec<String> = serde_json::from_value(scopes_json)
.map_err(|e| DbError::InvalidData(format!("scopes JSON: {e}")))?;
let channel_ids: Option<Vec<Uuid>> = {
let raw: Option<serde_json::Value> = row.try_get("channel_ids")?;
match raw {
None => None,
Some(v) => {
let strings: Vec<String> = serde_json::from_value(v)
.map_err(|e| DbError::InvalidData(format!("channel_ids JSON: {e}")))?;
let uuids: std::result::Result<Vec<Uuid>, _> =
strings.iter().map(|s| s.parse::<Uuid>()).collect();
Some(uuids.map_err(|e| DbError::InvalidData(format!("channel_ids UUID: {e}")))?)
}
}
};
Ok(ApiTokenRecord {
id,
token_hash: row.try_get("token_hash")?,
owner_pubkey: row.try_get("owner_pubkey")?,
name: row.try_get("name")?,
scopes,
channel_ids,
created_at: row.try_get("created_at")?,
expires_at: row.try_get("expires_at")?,
last_used_at: row.try_get("last_used_at")?,
revoked_at: row.try_get("revoked_at")?,
})
}
impl Db {
/// Create a new API token record.
#[allow(clippy::too_many_arguments)]
#[datastore_span(name = "create_api_token", system = "postgresql")]
pub async fn create_api_token(
&self,
community_id: CommunityId,
token_hash: &[u8],
owner_pubkey: &[u8],
name: &str,
scopes: &[String],
channel_ids: Option<&[Uuid]>,
expires_at: Option<DateTime<Utc>>,
) -> Result<Uuid> {
create_api_token(
&self.pool,
*community_id.as_uuid(),
token_hash,
owner_pubkey,
name,
scopes,
channel_ids,
expires_at,
)
.await
}
/// Atomic conditional INSERT with 10-token limit (per (community, owner)).
#[allow(clippy::too_many_arguments)]
#[datastore_span(name = "create_api_token_if_under_limit", system = "postgresql")]
pub async fn create_api_token_if_under_limit(
&self,
community_id: CommunityId,
token_hash: &[u8],
owner_pubkey: &[u8],
name: &str,
scopes: &[String],
channel_ids: Option<&[Uuid]>,
expires_at: Option<DateTime<Utc>>,
) -> Result<Option<Uuid>> {
create_api_token_if_under_limit(
&self.pool,
*community_id.as_uuid(),
token_hash,
owner_pubkey,
name,
scopes,
channel_ids,
expires_at,
)
.await
}
/// Look up an active (non-revoked) API token by its SHA-256 hash,
/// scoped to the request's community.
///
/// See [`get_api_token_by_hash_including_revoked`] for the
/// row-44 conformance rationale — the `(community_id, token_hash)` key
/// is enforced both by the storage UNIQUE index and by this WHERE clause.
#[datastore_span(name = "get_api_token_by_hash", system = "postgresql")]
pub async fn get_api_token_by_hash(
&self,
community_id: CommunityId,
hash: &[u8],
) -> Result<Option<ApiTokenRecord>> {
let row = sqlx::query(
r#"
SELECT id, token_hash, owner_pubkey, name, scopes, channel_ids,
created_at, expires_at, last_used_at, revoked_at
FROM api_tokens
WHERE community_id = $1 AND token_hash = $2 AND revoked_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(hash)
.fetch_optional(&self.pool)
.await?;
match row {
None => Ok(None),
Some(r) => parse_api_token_row(r).map(Some),
}
}
/// Look up an API token by hash, including revoked, scoped to community.
#[datastore_span(
name = "get_api_token_by_hash_including_revoked",
system = "postgresql"
)]
pub async fn get_api_token_by_hash_including_revoked(
&self,
community_id: CommunityId,
hash: &[u8],
) -> Result<Option<ApiTokenRecord>> {
get_api_token_by_hash_including_revoked(&self.pool, *community_id.as_uuid(), hash).await
}
/// Record a token usage (update `last_used_at`), scoped to community.
#[datastore_span(name = "touch_api_token", system = "postgresql")]
pub async fn touch_api_token(&self, community_id: CommunityId, hash: &[u8]) -> Result<()> {
sqlx::query(
"UPDATE api_tokens SET last_used_at = NOW() WHERE community_id = $1 AND token_hash = $2",
)
.bind(community_id.as_uuid())
.bind(hash)
.execute(&self.pool)
.await?;
Ok(())
}
/// Alias for [`Self::touch_api_token`].
pub async fn update_token_last_used(
&self,
community_id: CommunityId,
hash: &[u8],
) -> Result<()> {
self.touch_api_token(community_id, hash).await
}
/// List all active (non-revoked) tokens in a community, newest first.
#[datastore_span(name = "list_active_tokens", system = "postgresql")]
pub async fn list_active_tokens(&self, community_id: CommunityId) -> Result<Vec<TokenSummary>> {
let rows = sqlx::query(
r#"
SELECT id, name, owner_pubkey, scopes, created_at, expires_at
FROM api_tokens
WHERE community_id = $1 AND revoked_at IS NULL
ORDER BY created_at DESC
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.fetch_all(&self.pool)
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let id: Uuid = row.try_get("id")?;
let scopes_json: serde_json::Value = row.try_get("scopes")?;
let scopes: Vec<String> = serde_json::from_value(scopes_json)
.map_err(|e| DbError::InvalidData(format!("scopes JSON: {e}")))?;
out.push(TokenSummary {
id,
name: row.try_get("name")?,
owner_pubkey: row.try_get("owner_pubkey")?,
scopes,
created_at: row.try_get("created_at")?,
expires_at: row.try_get("expires_at")?,
});
}
Ok(out)
}
/// List all tokens for a (community, owner) pair (including revoked).
#[datastore_span(name = "list_tokens_by_owner", system = "postgresql")]
pub async fn list_tokens_by_owner(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<ApiTokenRecord>> {
list_tokens_by_owner(&self.pool, *community_id.as_uuid(), pubkey).await
}
/// Revoke a single token by ID, scoped to (community, owner).
#[datastore_span(name = "revoke_token", system = "postgresql")]
pub async fn revoke_token(
&self,
community_id: CommunityId,
id: Uuid,
owner_pubkey: &[u8],
revoked_by: &[u8],
) -> Result<bool> {
revoke_token(
&self.pool,
*community_id.as_uuid(),
id,
owner_pubkey,
revoked_by,
)
.await
}
/// Revoke all active tokens for a (community, owner) pair.
#[datastore_span(name = "revoke_all_tokens", system = "postgresql")]
pub async fn revoke_all_tokens(
&self,
community_id: CommunityId,
owner_pubkey: &[u8],
revoked_by: &[u8],
) -> Result<u64> {
revoke_all_tokens(
&self.pool,
*community_id.as_uuid(),
owner_pubkey,
revoked_by,
)
.await
}
}
#[cfg(test)]
mod postgres_tests {
//! Row-44 conformance: API token lookups MUST be keyed on
//! `(community_id, token_hash)`, not on `token_hash` alone. The storage
//! UNIQUE index is a *storage* guarantee; the WHERE clause here is the
//! *query* guarantee. Both must hold — a query that filters on hash
//! alone could return a foreign-community row, defeating the row-zero
//! tenancy fence. This test directly inserts two same-hash rows in two
//! communities (only possible by bypassing the unique index, which we
//! achieve via distinct hashes that the test then queries-by-hash for
//! both — see below for the actual property under test).
//!
//! The load-bearing property: even if storage uniqueness is ever relaxed
//! or a hash collision occurs, the query-side `AND community_id = $N`
//! clause guarantees the lookup returns the row for the *requested*
//! tenant. Mutate-bite proof: drop the clause, the test fails.
use super::*;
use crate::{ApiTokenRecord, Db};
use sqlx::PgPool;
async fn setup_db() -> Db {
let pool = PgPool::connect(&crate::test_support::database_url())
.await
.expect("connect to test DB");
Db::from_pool(pool)
}
async fn make_community(pool: &PgPool) -> Uuid {
let id = Uuid::new_v4();
let host = format!("api-token-tenancy-{}.example", id.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(host)
.execute(pool)
.await
.expect("insert community");
id
}
async fn insert_user(pool: &PgPool, community_id: Uuid, pubkey: &[u8]) {
sqlx::query(
r#"
INSERT INTO users (community_id, pubkey)
VALUES ($1, $2)
"#,
)
.bind(community_id)
.bind(pubkey)
.execute(pool)
.await
.expect("insert user");
}
/// Direct INSERT bypassing `create_api_token` so the test pins the
/// **lookup**'s scoping, not the insert path's.
async fn raw_insert_token(
pool: &PgPool,
community_id: Uuid,
token_hash: &[u8],
owner_pubkey: &[u8],
name: &str,
) -> Uuid {
let id = Uuid::new_v4();
let scopes = serde_json::json!(["files:read", "files:write"]);
sqlx::query(
r#"
INSERT INTO api_tokens
(community_id, id, token_hash, owner_pubkey, name, scopes)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
)
.bind(community_id)
.bind(id)
.bind(token_hash)
.bind(owner_pubkey)
.bind(name)
.bind(&scopes)
.execute(pool)
.await
.expect("insert api_token");
id
}
/// Row-44 sharp test: two communities, **same** 32-byte token hash in each,
/// lookup scoped to community A returns A's row only (and B-scoped lookup
/// returns B's row only). The storage UNIQUE index is `(community_id,
/// token_hash)` so this is a legal state. The lookup must not return the
/// foreign row.
///
/// Mutate-bite handle: the WHERE clause in
/// `get_api_token_by_hash_including_revoked` is the only thing keeping
/// this test green. Strip `AND community_id = $1` and the lookup becomes
/// hash-only — Postgres returns whichever row it picks (insert-order
/// dependent), and the cross-tenancy assertion fails.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn lookup_by_hash_is_scoped_to_community() {
let db = setup_db().await;
let community_a = make_community(&db.pool).await;
let community_b = make_community(&db.pool).await;
// Distinct pubkeys per community — FK is (community_id, owner_pubkey).
let owner_a = vec![0xAAu8; 32];
let owner_b = vec![0xBBu8; 32];
insert_user(&db.pool, community_a, &owner_a).await;
insert_user(&db.pool, community_b, &owner_b).await;
// SAME hash in both communities — legal under UNIQUE(community_id, token_hash).
let shared_hash = vec![0xCCu8; 32];
let id_a = raw_insert_token(&db.pool, community_a, &shared_hash, &owner_a, "token-A").await;
let id_b = raw_insert_token(&db.pool, community_b, &shared_hash, &owner_b, "token-B").await;
assert_ne!(id_a, id_b, "ids must differ");
let cid_a = buzz_core::CommunityId::from_uuid(community_a);
let cid_b = buzz_core::CommunityId::from_uuid(community_b);
// Lookup scoped to A returns A's row, never B's.
let from_a: ApiTokenRecord = db
.get_api_token_by_hash_including_revoked(cid_a, &shared_hash)
.await
.expect("lookup A")
.expect("row in A");
assert_eq!(from_a.id, id_a, "community-A lookup must return A's row");
assert_eq!(
from_a.owner_pubkey, owner_a,
"community-A lookup must return A's owner",
);
// Lookup scoped to B returns B's row, never A's.
let from_b: ApiTokenRecord = db
.get_api_token_by_hash_including_revoked(cid_b, &shared_hash)
.await
.expect("lookup B")
.expect("row in B");
assert_eq!(from_b.id, id_b, "community-B lookup must return B's row");
assert_eq!(
from_b.owner_pubkey, owner_b,
"community-B lookup must return B's owner",
);
// Lookup with the hash but a third (unrelated) community returns None.
let community_c = make_community(&db.pool).await;
let cid_c = buzz_core::CommunityId::from_uuid(community_c);
let from_c = db
.get_api_token_by_hash_including_revoked(cid_c, &shared_hash)
.await
.expect("lookup C");
assert!(
from_c.is_none(),
"community-C has no token with this hash; lookup must return None, got {from_c:?}",
);
}
/// Active (non-revoked) lookup also enforces community scope.
/// Mirrors the obligation for the `revoked_at IS NULL` variant at
/// `Db::get_api_token_by_hash`.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn active_lookup_by_hash_is_scoped_to_community() {
let db = setup_db().await;
let community_a = make_community(&db.pool).await;
let community_b = make_community(&db.pool).await;
let owner_a = vec![0x11u8; 32];
let owner_b = vec![0x22u8; 32];
insert_user(&db.pool, community_a, &owner_a).await;
insert_user(&db.pool, community_b, &owner_b).await;
let shared_hash = vec![0x33u8; 32];
let id_a =
raw_insert_token(&db.pool, community_a, &shared_hash, &owner_a, "active-A").await;
let id_b =
raw_insert_token(&db.pool, community_b, &shared_hash, &owner_b, "active-B").await;
let cid_a = buzz_core::CommunityId::from_uuid(community_a);
let cid_b = buzz_core::CommunityId::from_uuid(community_b);
let from_a = db
.get_api_token_by_hash(cid_a, &shared_hash)
.await
.expect("active lookup A")
.expect("row in A");
assert_eq!(from_a.id, id_a);
let from_b = db
.get_api_token_by_hash(cid_b, &shared_hash)
.await
.expect("active lookup B")
.expect("row in B");
assert_eq!(from_b.id, id_b);
}
}