use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use crate::budget_store::{
budget_snapshot_anchor_chain_digest, BudgetSnapshotAnchorCommitment,
SignedBudgetSnapshotAnchorCommitment,
};
use chio_core::capability::{
scope::ChioScope,
token::{CapabilityToken, CapabilityTokenBody},
};
use chio_core::crypto::{Keypair, PublicKey, Signature};
use chio_kernel::{
ensure_capability_issuance_supported, AuthoritySnapshot, AuthorityStatus, AuthorityStoreError,
AuthorityTrustedKeySnapshot, CapabilityAuthority, KernelError,
};
use rusqlite::{params, Connection, OptionalExtension};
use uuid::Uuid;
pub struct SqliteCapabilityAuthority {
path: PathBuf,
cached_public_key: Mutex<PublicKey>,
cached_trusted_public_keys: Mutex<Vec<PublicKey>>,
}
const AUTHORITY_STORE_SUPPORTED_SCHEMA_VERSION: i32 = 1;
const AUTHORITY_STORE_SCHEMA_KEY: &str = "authority";
const AUTHORITY_STORE_LEGACY_ANCHOR_TABLES: &[&str] = &["authority_state"];
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthorityClusterFence {
pub leader_url: Option<String>,
pub election_term: u64,
pub updated_at: u64,
pub authority_generation: u64,
pub authority_rotated_at: u64,
}
impl SqliteCapabilityAuthority {
pub fn open(path: impl AsRef<Path>) -> Result<Self, AuthorityStoreError> {
let path = path.as_ref().to_path_buf();
if let Some(parent) = crate::sqlite_parent_dir_to_create(&path) {
fs::create_dir_all(&parent)?;
}
let bootstrap = Keypair::generate();
let connection = Self::open_connection(&path)?;
connection.execute(
r#"
INSERT INTO authority_state (singleton_id, seed_hex, public_key_hex, generation, rotated_at)
VALUES (1, ?1, ?2, 1, ?3)
ON CONFLICT(singleton_id) DO NOTHING
"#,
params![
bootstrap.seed_hex(),
bootstrap.public_key().to_hex(),
unix_now() as i64
],
)?;
let current_public_key = connection
.query_row(
r#"
SELECT seed_hex
FROM authority_state
WHERE singleton_id = 1
"#,
[],
|row| row.get::<_, String>(0),
)
.map(|seed_hex| Keypair::from_seed_hex(seed_hex.trim()))
.map_err(AuthorityStoreError::from)??;
connection.execute(
r#"
UPDATE authority_state
SET public_key_hex = COALESCE(NULLIF(public_key_hex, ''), ?1)
WHERE singleton_id = 1
"#,
params![current_public_key.public_key().to_hex()],
)?;
connection.execute(
r#"
INSERT INTO authority_trusted_keys (public_key_hex, generation, activated_at)
VALUES (?1, 1, ?2)
ON CONFLICT(public_key_hex) DO NOTHING
"#,
params![current_public_key.public_key().to_hex(), unix_now() as i64],
)?;
let status = Self::read_status_from_connection(&connection)?;
Ok(Self {
path,
cached_public_key: Mutex::new(status.public_key),
cached_trusted_public_keys: Mutex::new(status.trusted_public_keys),
})
}
pub fn status(&self) -> Result<AuthorityStatus, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let status = Self::read_status_from_connection(&connection)?;
self.update_cached_public_key(status.public_key.clone());
self.update_cached_trusted_public_keys(status.trusted_public_keys.clone());
Ok(status)
}
pub fn rotate(&self) -> Result<AuthorityStatus, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let status_before = Self::read_status_from_connection(&connection)?;
let keypair = Keypair::generate();
let rotated_at = unix_now();
let next_generation = status_before.generation.saturating_add(1);
connection.execute(
r#"
UPDATE authority_state
SET seed_hex = ?1, public_key_hex = ?2, generation = ?3, rotated_at = ?4
WHERE singleton_id = 1
"#,
params![
keypair.seed_hex(),
keypair.public_key().to_hex(),
next_generation as i64,
rotated_at as i64,
],
)?;
connection.execute(
r#"
INSERT INTO authority_trusted_keys (public_key_hex, generation, activated_at)
VALUES (?1, ?2, ?3)
ON CONFLICT(public_key_hex) DO NOTHING
"#,
params![
keypair.public_key().to_hex(),
next_generation as i64,
rotated_at as i64
],
)?;
let status = Self::read_status_from_connection(&connection)?;
self.update_cached_public_key(status.public_key.clone());
self.update_cached_trusted_public_keys(status.trusted_public_keys.clone());
Ok(status)
}
pub fn snapshot(&self) -> Result<AuthoritySnapshot, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let status = Self::read_status_from_connection(&connection)?;
Ok(AuthoritySnapshot {
public_key_hex: status.public_key.to_hex(),
generation: status.generation,
rotated_at: status.rotated_at,
trusted_keys: Self::read_trusted_key_snapshots(&connection)?,
})
}
pub fn apply_snapshot(
&self,
snapshot: &AuthoritySnapshot,
) -> Result<bool, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let local_snapshot = self.snapshot()?;
let remote_public_key = PublicKey::from_hex(snapshot.public_key_hex.trim())?;
connection.execute(
r#"
INSERT INTO authority_trusted_keys (public_key_hex, generation, activated_at)
VALUES (?1, ?2, ?3)
ON CONFLICT(public_key_hex) DO UPDATE SET
generation = MAX(generation, excluded.generation),
activated_at = MIN(activated_at, excluded.activated_at)
"#,
params![
remote_public_key.to_hex(),
snapshot.generation as i64,
snapshot.rotated_at as i64
],
)?;
for trusted_key in &snapshot.trusted_keys {
connection.execute(
r#"
INSERT INTO authority_trusted_keys (public_key_hex, generation, activated_at)
VALUES (?1, ?2, ?3)
ON CONFLICT(public_key_hex) DO UPDATE SET
generation = MAX(generation, excluded.generation),
activated_at = MIN(activated_at, excluded.activated_at)
"#,
params![
trusted_key.public_key_hex,
trusted_key.generation as i64,
trusted_key.activated_at as i64
],
)?;
}
let should_replace = snapshot.generation > local_snapshot.generation
|| (snapshot.generation == local_snapshot.generation
&& (snapshot.rotated_at, snapshot.public_key_hex.as_str())
> (
local_snapshot.rotated_at,
local_snapshot.public_key_hex.as_str(),
));
if should_replace {
connection.execute(
r#"
UPDATE authority_state
SET public_key_hex = ?1, generation = ?2, rotated_at = ?3
WHERE singleton_id = 1
"#,
params![
remote_public_key.to_hex(),
snapshot.generation as i64,
snapshot.rotated_at as i64,
],
)?;
}
let status = Self::read_status_from_connection(&connection)?;
self.update_cached_public_key(status.public_key);
self.update_cached_trusted_public_keys(status.trusted_public_keys);
Ok(should_replace)
}
pub fn current_keypair(&self) -> Result<Keypair, AuthorityStoreError> {
self.read_current_keypair()
}
pub fn local_keypair(&self) -> Result<Keypair, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
Self::read_keypair_from_connection(&connection)
}
pub fn commit_budget_snapshot_anchor_set(
&self,
anchor_set_digest: &str,
leader_url: &str,
election_term: u64,
committed_at: u64,
) -> Result<Vec<SignedBudgetSnapshotAnchorCommitment>, AuthorityStoreError> {
if anchor_set_digest.len() != 64
|| !anchor_set_digest
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
|| leader_url.is_empty()
|| election_term == 0
{
return Err(AuthorityStoreError::Fence(
"budget snapshot anchor commitment identity is invalid".to_string(),
));
}
let election_term_sqlite = i64::try_from(election_term).map_err(|_| {
AuthorityStoreError::Fence(
"budget snapshot anchor election term exceeds SQLite range".to_string(),
)
})?;
let committed_at_sqlite = i64::try_from(committed_at).map_err(|_| {
AuthorityStoreError::Fence(
"budget snapshot anchor commit time exceeds SQLite range".to_string(),
)
})?;
let mut connection = Self::open_connection(&self.path)?;
let transaction = connection.transaction()?;
let keypair = Self::read_keypair_from_connection(&transaction)?;
let signer_public_key = keypair.public_key().to_hex();
let latest = transaction
.query_row(
r#"
SELECT commit_sequence, anchor_set_digest, leader_url, election_term,
signer_public_key
FROM authority_budget_anchor_commits
ORDER BY commit_sequence DESC LIMIT 1
"#,
[],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, String>(4)?,
))
},
)
.optional()?;
let already_committed = latest
.as_ref()
.is_some_and(|(_, digest, leader, term, signer)| {
digest == anchor_set_digest
&& leader == leader_url
&& *term == election_term_sqlite
&& signer == &signer_public_key
});
if !already_committed {
let (previous_sequence, previous_chain_digest) = transaction
.query_row(
r#"
SELECT commit_sequence, chain_digest
FROM authority_budget_anchor_commits
ORDER BY commit_sequence DESC LIMIT 1
"#,
[],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)
.optional()?
.unwrap_or((
0,
"0000000000000000000000000000000000000000000000000000000000000000".to_string(),
));
let commit_sequence = u64::try_from(previous_sequence)
.ok()
.and_then(|value| value.checked_add(1))
.ok_or_else(|| {
AuthorityStoreError::Fence(
"budget snapshot anchor commitment sequence overflowed".to_string(),
)
})?;
let mut body = BudgetSnapshotAnchorCommitment {
schema: "chio.budget-snapshot-anchor-commitment.v1".to_string(),
commit_sequence,
previous_chain_digest,
chain_digest: String::new(),
anchor_set_digest: anchor_set_digest.to_string(),
leader_url: leader_url.to_string(),
election_term,
committed_at,
signer_public_key,
};
body.chain_digest = budget_snapshot_anchor_chain_digest(&body)
.map_err(|error| AuthorityStoreError::Fence(error.to_string()))?;
let signature = keypair
.sign_canonical(&body)
.map_err(|error| AuthorityStoreError::Fence(error.to_string()))?
.0;
let commit_sequence_sqlite = i64::try_from(commit_sequence).map_err(|_| {
AuthorityStoreError::Fence(
"budget snapshot anchor commitment sequence exceeds SQLite range".to_string(),
)
})?;
transaction.execute(
r#"
INSERT INTO authority_budget_anchor_commits (
commit_sequence, previous_chain_digest, chain_digest,
anchor_set_digest, leader_url, election_term, committed_at,
signer_public_key, signature
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
"#,
params![
commit_sequence_sqlite,
&body.previous_chain_digest,
&body.chain_digest,
&body.anchor_set_digest,
&body.leader_url,
election_term_sqlite,
committed_at_sqlite,
&body.signer_public_key,
signature.to_hex(),
],
)?;
}
let chain = Self::read_budget_anchor_commit_chain(&transaction)?;
transaction.commit()?;
Ok(chain)
}
pub fn cluster_fence(&self) -> Result<AuthorityClusterFence, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
Self::read_cluster_fence_from_connection(&connection)
}
pub fn seed_cluster_fence(
&self,
leader_url: Option<&str>,
election_term: u64,
) -> Result<bool, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let current = Self::read_cluster_fence_from_connection(&connection)?;
let (_, authority_generation, authority_rotated_at) =
Self::read_public_state_from_connection(&connection)?;
let next_leader = leader_url.map(ToOwned::to_owned);
let same_term_same_leader = election_term == current.election_term
&& current.leader_url.as_deref() == next_leader.as_deref();
let fence_authority_state_is_stale = (current.election_term > 0
|| current.leader_url.is_some())
&& (current.authority_generation != authority_generation
|| current.authority_rotated_at != authority_rotated_at);
let should_update = election_term > current.election_term
|| (election_term == current.election_term
&& current.leader_url.is_none()
&& next_leader.is_some())
|| (same_term_same_leader && fence_authority_state_is_stale);
if should_update {
Self::write_cluster_fence_to_connection(&connection, next_leader, election_term)?;
}
Ok(should_update)
}
pub fn enforce_cluster_fence(
&self,
leader_url: &str,
election_term: u64,
) -> Result<(), AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let current = Self::read_cluster_fence_from_connection(&connection)?;
let (_, authority_generation, authority_rotated_at) =
Self::read_public_state_from_connection(&connection)?;
if (current.election_term > 0 || current.leader_url.is_some())
&& (current.authority_generation != authority_generation
|| current.authority_rotated_at != authority_rotated_at)
{
return Err(AuthorityStoreError::Fence(format!(
"persisted authority fence generation `{}` rotated_at `{}` does not match current authority generation `{authority_generation}` rotated_at `{authority_rotated_at}`",
current.authority_generation, current.authority_rotated_at
)));
}
if election_term < current.election_term {
return Err(AuthorityStoreError::Fence(format!(
"stale authority term `{election_term}` is below persisted term `{}`",
current.election_term
)));
}
if election_term == current.election_term
&& current
.leader_url
.as_deref()
.is_some_and(|current_leader| current_leader != leader_url)
{
return Err(AuthorityStoreError::Fence(format!(
"authority term `{election_term}` is already fenced to leader `{}`",
current.leader_url.unwrap_or_default()
)));
}
Self::write_cluster_fence_to_connection(
&connection,
Some(leader_url.to_string()),
election_term,
)
}
fn open_connection(path: &Path) -> Result<Connection, AuthorityStoreError> {
let connection = Connection::open(path)?;
crate::check_schema_version(
&connection,
AUTHORITY_STORE_SCHEMA_KEY,
AUTHORITY_STORE_SUPPORTED_SCHEMA_VERSION,
AUTHORITY_STORE_LEGACY_ANCHOR_TABLES,
)
.map_err(|error| AuthorityStoreError::Schema(error.to_string()))?;
connection.execute_batch(
r#"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = FULL;
PRAGMA busy_timeout = 5000;
CREATE TABLE IF NOT EXISTS authority_state (
singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1),
seed_hex TEXT NOT NULL,
public_key_hex TEXT,
generation INTEGER NOT NULL,
rotated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS authority_trusted_keys (
public_key_hex TEXT PRIMARY KEY,
generation INTEGER NOT NULL,
activated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS authority_cluster_fence (
singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1),
leader_url TEXT,
election_term INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
authority_generation INTEGER NOT NULL DEFAULT 0,
authority_rotated_at INTEGER NOT NULL DEFAULT 0
);
INSERT INTO authority_cluster_fence (singleton_id, leader_url, election_term, updated_at)
VALUES (1, NULL, 0, 0)
ON CONFLICT(singleton_id) DO NOTHING;
CREATE TABLE IF NOT EXISTS authority_budget_anchor_commits (
commit_sequence INTEGER PRIMARY KEY CHECK (commit_sequence > 0),
previous_chain_digest TEXT NOT NULL,
chain_digest TEXT NOT NULL UNIQUE,
anchor_set_digest TEXT NOT NULL,
leader_url TEXT NOT NULL CHECK (leader_url <> ''),
election_term INTEGER NOT NULL CHECK (election_term > 0),
committed_at INTEGER NOT NULL CHECK (committed_at >= 0),
signer_public_key TEXT NOT NULL CHECK (signer_public_key <> ''),
signature TEXT NOT NULL CHECK (signature <> '')
);
CREATE TRIGGER IF NOT EXISTS authority_budget_anchor_commits_immutable
BEFORE UPDATE ON authority_budget_anchor_commits
BEGIN
SELECT RAISE(ABORT, 'budget anchor commitment is immutable');
END;
CREATE TRIGGER IF NOT EXISTS authority_budget_anchor_commits_no_delete
BEFORE DELETE ON authority_budget_anchor_commits
BEGIN
SELECT RAISE(ABORT, 'budget anchor commitment is immutable');
END;
"#,
)?;
if !Self::table_has_column(&connection, "authority_state", "public_key_hex")? {
connection.execute(
"ALTER TABLE authority_state ADD COLUMN public_key_hex TEXT",
[],
)?;
}
if !Self::table_has_column(
&connection,
"authority_cluster_fence",
"authority_generation",
)? {
connection.execute(
"ALTER TABLE authority_cluster_fence ADD COLUMN authority_generation INTEGER NOT NULL DEFAULT 0",
[],
)?;
}
if !Self::table_has_column(
&connection,
"authority_cluster_fence",
"authority_rotated_at",
)? {
connection.execute(
"ALTER TABLE authority_cluster_fence ADD COLUMN authority_rotated_at INTEGER NOT NULL DEFAULT 0",
[],
)?;
}
crate::stamp_schema_version(
&connection,
AUTHORITY_STORE_SCHEMA_KEY,
AUTHORITY_STORE_SUPPORTED_SCHEMA_VERSION,
)
.map_err(|error| AuthorityStoreError::Schema(error.to_string()))?;
Ok(connection)
}
fn read_status_from_connection(
connection: &Connection,
) -> Result<AuthorityStatus, AuthorityStoreError> {
let (public_key, generation, rotated_at) =
Self::read_public_state_from_connection(connection)?;
Ok(AuthorityStatus {
public_key,
generation,
rotated_at,
trusted_public_keys: Self::read_trusted_public_keys(connection)?,
})
}
fn read_keypair_from_connection(
connection: &Connection,
) -> Result<Keypair, AuthorityStoreError> {
let seed_hex = connection.query_row(
r#"
SELECT seed_hex
FROM authority_state
WHERE singleton_id = 1
"#,
[],
|row| row.get::<_, String>(0),
)?;
Keypair::from_seed_hex(seed_hex.trim()).map_err(Into::into)
}
fn read_budget_anchor_commit_chain(
connection: &Connection,
) -> Result<Vec<SignedBudgetSnapshotAnchorCommitment>, AuthorityStoreError> {
let mut statement = connection.prepare(
r#"
SELECT commit_sequence, previous_chain_digest, chain_digest,
anchor_set_digest, leader_url, election_term, committed_at,
signer_public_key, signature
FROM authority_budget_anchor_commits
ORDER BY commit_sequence
"#,
)?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, i64>(5)?,
row.get::<_, i64>(6)?,
row.get::<_, String>(7)?,
row.get::<_, String>(8)?,
))
})?;
rows.map(|row| {
let (sequence, previous, chain, anchors, leader, term, at, signer, signature) = row?;
Ok(SignedBudgetSnapshotAnchorCommitment {
body: BudgetSnapshotAnchorCommitment {
schema: "chio.budget-snapshot-anchor-commitment.v1".to_string(),
commit_sequence: sequence.max(0) as u64,
previous_chain_digest: previous,
chain_digest: chain,
anchor_set_digest: anchors,
leader_url: leader,
election_term: term.max(0) as u64,
committed_at: at.max(0) as u64,
signer_public_key: signer,
},
signature: Signature::from_hex(&signature)?,
})
})
.collect()
}
fn read_public_state_from_connection(
connection: &Connection,
) -> Result<(PublicKey, u64, u64), AuthorityStoreError> {
let (seed_hex, public_key_hex, generation, rotated_at) = connection.query_row(
r#"
SELECT seed_hex, public_key_hex, generation, rotated_at
FROM authority_state
WHERE singleton_id = 1
"#,
[],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, i64>(3)?,
))
},
)?;
let public_key = match public_key_hex
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
Some(public_key_hex) => PublicKey::from_hex(public_key_hex)?,
None => Keypair::from_seed_hex(seed_hex.trim())?.public_key(),
};
Ok((
public_key,
generation.max(0) as u64,
rotated_at.max(0) as u64,
))
}
fn read_current_keypair(&self) -> Result<Keypair, AuthorityStoreError> {
let connection = Self::open_connection(&self.path)?;
let keypair = Self::read_keypair_from_connection(&connection)?;
let status = Self::read_status_from_connection(&connection)?;
self.update_cached_public_key(status.public_key.clone());
self.update_cached_trusted_public_keys(status.trusted_public_keys.clone());
if keypair.public_key() != status.public_key {
return Err(AuthorityStoreError::Fence(format!(
"local signing seed public key {} does not match replicated authority public key {}",
keypair.public_key().to_hex(),
status.public_key.to_hex(),
)));
}
Ok(keypair)
}
fn table_has_column(
connection: &Connection,
table: &str,
column: &str,
) -> Result<bool, AuthorityStoreError> {
let pragma = format!("PRAGMA table_info({table})");
let mut statement = connection.prepare(&pragma)?;
let mut rows = statement.query([])?;
while let Some(row) = rows.next()? {
if row.get::<_, String>(1)? == column {
return Ok(true);
}
}
Ok(false)
}
fn update_cached_public_key(&self, public_key: PublicKey) {
match self.cached_public_key.lock() {
Ok(mut guard) => *guard = public_key,
Err(poisoned) => {
tracing::error!(
"cached_public_key mutex is poisoned - recovering possibly-stale key material"
);
*poisoned.into_inner() = public_key;
}
}
}
fn update_cached_trusted_public_keys(&self, public_keys: Vec<PublicKey>) {
match self.cached_trusted_public_keys.lock() {
Ok(mut guard) => *guard = public_keys,
Err(poisoned) => {
tracing::error!(
"cached_trusted_public_keys mutex is poisoned - recovering possibly-stale key material"
);
*poisoned.into_inner() = public_keys;
}
}
}
fn cached_public_key(&self) -> PublicKey {
match self.cached_public_key.lock() {
Ok(guard) => guard.clone(),
Err(poisoned) => {
tracing::error!(
"cached_public_key mutex is poisoned - reading possibly-stale key material"
);
poisoned.into_inner().clone()
}
}
}
fn cached_trusted_public_keys(&self) -> Vec<PublicKey> {
match self.cached_trusted_public_keys.lock() {
Ok(guard) => guard.clone(),
Err(poisoned) => {
tracing::error!(
"cached_trusted_public_keys mutex is poisoned - reading possibly-stale key material"
);
poisoned.into_inner().clone()
}
}
}
fn read_trusted_public_keys(
connection: &Connection,
) -> Result<Vec<PublicKey>, AuthorityStoreError> {
let mut statement = connection.prepare(
r#"
SELECT public_key_hex
FROM authority_trusted_keys
ORDER BY generation ASC, activated_at ASC
"#,
)?;
let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
rows.map(|row| {
let public_key_hex = row?;
PublicKey::from_hex(public_key_hex.trim()).map_err(AuthorityStoreError::from)
})
.collect()
}
fn read_trusted_key_snapshots(
connection: &Connection,
) -> Result<Vec<AuthorityTrustedKeySnapshot>, AuthorityStoreError> {
let mut statement = connection.prepare(
r#"
SELECT public_key_hex, generation, activated_at
FROM authority_trusted_keys
ORDER BY generation ASC, activated_at ASC, public_key_hex ASC
"#,
)?;
let rows = statement.query_map([], |row| {
Ok(AuthorityTrustedKeySnapshot {
public_key_hex: row.get(0)?,
generation: row.get::<_, i64>(1)?.max(0) as u64,
activated_at: row.get::<_, i64>(2)?.max(0) as u64,
})
})?;
rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}
fn read_cluster_fence_from_connection(
connection: &Connection,
) -> Result<AuthorityClusterFence, AuthorityStoreError> {
let (leader_url, election_term, updated_at, authority_generation, authority_rotated_at) =
connection.query_row(
r#"
SELECT leader_url, election_term, updated_at, authority_generation, authority_rotated_at
FROM authority_cluster_fence
WHERE singleton_id = 1
"#,
[],
|row| {
Ok((
row.get::<_, Option<String>>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, i64>(4)?,
))
},
)?;
Ok(AuthorityClusterFence {
leader_url,
election_term: election_term.max(0) as u64,
updated_at: updated_at.max(0) as u64,
authority_generation: authority_generation.max(0) as u64,
authority_rotated_at: authority_rotated_at.max(0) as u64,
})
}
fn write_cluster_fence_to_connection(
connection: &Connection,
leader_url: Option<String>,
election_term: u64,
) -> Result<(), AuthorityStoreError> {
let (_, authority_generation, authority_rotated_at) =
Self::read_public_state_from_connection(connection)?;
connection.execute(
r#"
UPDATE authority_cluster_fence
SET
leader_url = ?1,
election_term = ?2,
updated_at = ?3,
authority_generation = ?4,
authority_rotated_at = ?5
WHERE singleton_id = 1
"#,
params![
leader_url,
election_term as i64,
unix_now() as i64,
authority_generation as i64,
authority_rotated_at as i64,
],
)?;
Ok(())
}
}
impl CapabilityAuthority for SqliteCapabilityAuthority {
fn authority_public_key(&self) -> PublicKey {
self.status()
.map(|status| status.public_key)
.unwrap_or_else(|_| self.cached_public_key())
}
fn trusted_public_keys(&self) -> Vec<PublicKey> {
self.status()
.map(|status| status.trusted_public_keys)
.unwrap_or_else(|_| self.cached_trusted_public_keys())
}
fn issue_capability(
&self,
subject: &PublicKey,
scope: ChioScope,
ttl_seconds: u64,
) -> Result<CapabilityToken, KernelError> {
ensure_capability_issuance_supported(&scope)?;
let keypair = self
.read_current_keypair()
.map_err(|error| KernelError::CapabilityIssuanceFailed(error.to_string()))?;
let now = unix_now();
let body = CapabilityTokenBody {
id: format!("cap-{}", Uuid::now_v7()),
issuer: keypair.public_key(),
subject: subject.clone(),
scope,
issued_at: now,
expires_at: now.saturating_add(ttl_seconds),
delegation_chain: vec![],
aggregate_invocation_budget: None,
};
CapabilityToken::sign(body, &keypair)
.map_err(|error| KernelError::CapabilityIssuanceFailed(error.to_string()))
}
}
fn unix_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or(0)
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
use std::fs;
use std::time::{SystemTime, UNIX_EPOCH};
use super::*;
use chio_core::capability::scope::{Operation, ToolGrant};
use chio_kernel::LocalCapabilityAuthority;
fn unique_db_path(prefix: &str) -> std::path::PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time before epoch")
.as_nanos();
std::env::temp_dir().join(format!("{prefix}-{nonce}.sqlite3"))
}
#[test]
fn local_capability_authority_signs_capabilities() {
let authority = LocalCapabilityAuthority::new(Keypair::generate());
let subject = Keypair::generate().public_key();
let capability = authority
.issue_capability(
&subject,
ChioScope {
grants: vec![ToolGrant {
server_id: "srv-a".to_string(),
tool_name: "read_file".to_string(),
operations: vec![Operation::Invoke],
constraints: vec![],
max_invocations: None,
max_cost_per_invocation: None,
max_total_cost: None,
dpop_required: None,
}],
resource_grants: vec![],
prompt_grants: vec![],
},
300,
)
.unwrap();
assert_eq!(capability.subject, subject);
assert_eq!(capability.issuer, authority.authority_public_key());
assert!(capability.id.starts_with("cap-"));
assert!(capability.verify_signature().unwrap());
}
#[test]
fn sqlite_capability_authority_persists_and_rotates_across_handles() {
let path = unique_db_path("chio-authority");
let authority_a = SqliteCapabilityAuthority::open(&path).unwrap();
let authority_b = SqliteCapabilityAuthority::open(&path).unwrap();
let before = authority_a.status().unwrap();
assert_eq!(before.generation, 1);
assert_eq!(authority_b.status().unwrap().public_key, before.public_key);
let rotated = authority_a.rotate().unwrap();
assert_eq!(rotated.generation, 2);
assert_ne!(rotated.public_key, before.public_key);
let observed = authority_b.status().unwrap();
assert_eq!(observed.public_key, rotated.public_key);
assert_eq!(observed.generation, rotated.generation);
let _ = fs::remove_file(path);
}
#[test]
fn sqlite_capability_authority_issues_with_current_rotated_key() {
let path = unique_db_path("chio-authority-issue");
let authority = SqliteCapabilityAuthority::open(&path).unwrap();
let subject = Keypair::generate().public_key();
let first = authority
.issue_capability(
&subject,
ChioScope {
grants: vec![ToolGrant {
server_id: "srv-a".to_string(),
tool_name: "read_file".to_string(),
operations: vec![Operation::Invoke],
constraints: vec![],
max_invocations: None,
max_cost_per_invocation: None,
max_total_cost: None,
dpop_required: None,
}],
resource_grants: vec![],
prompt_grants: vec![],
},
300,
)
.unwrap();
let rotated = authority.rotate().unwrap();
let second = authority
.issue_capability(
&subject,
ChioScope {
grants: vec![ToolGrant {
server_id: "srv-a".to_string(),
tool_name: "read_file".to_string(),
operations: vec![Operation::Invoke],
constraints: vec![],
max_invocations: None,
max_cost_per_invocation: None,
max_total_cost: None,
dpop_required: None,
}],
resource_grants: vec![],
prompt_grants: vec![],
},
300,
)
.unwrap();
assert_ne!(first.issuer, second.issuer);
assert_eq!(second.issuer, rotated.public_key);
assert!(second.verify_signature().unwrap());
let _ = fs::remove_file(path);
}
#[test]
fn sqlite_capability_authority_snapshot_updates_public_view_without_copying_seed() {
let source_path = unique_db_path("chio-authority-source");
let follower_path = unique_db_path("chio-authority-follower");
let source = SqliteCapabilityAuthority::open(&source_path).unwrap();
let follower = SqliteCapabilityAuthority::open(&follower_path).unwrap();
let follower_local_key = follower.current_keypair().unwrap().public_key();
let rotated = source.rotate().unwrap();
let snapshot = source.snapshot().unwrap();
assert!(follower.apply_snapshot(&snapshot).unwrap());
let follower_status = follower.status().unwrap();
assert_eq!(follower_status.public_key, rotated.public_key);
assert_eq!(follower_status.generation, rotated.generation);
let error = match follower.current_keypair() {
Ok(_) => panic!("mismatched follower keypair should be rejected"),
Err(error) => error.to_string(),
};
assert!(error.contains("does not match replicated authority public key"));
assert!(error.contains(&follower_local_key.to_hex()));
let _ = fs::remove_file(source_path);
let _ = fs::remove_file(follower_path);
}
#[test]
fn sqlite_capability_authority_same_term_snapshot_does_not_clear_fenced_leader() {
let path = unique_db_path("chio-authority-fence");
let authority = SqliteCapabilityAuthority::open(&path).unwrap();
assert!(authority
.seed_cluster_fence(Some("http://leader-a"), 7)
.unwrap());
assert!(!authority.seed_cluster_fence(None, 7).unwrap());
let fence = authority.cluster_fence().unwrap();
let status = authority.status().unwrap();
assert_eq!(fence.election_term, 7);
assert_eq!(fence.leader_url.as_deref(), Some("http://leader-a"));
assert_eq!(fence.authority_generation, status.generation);
assert_eq!(fence.authority_rotated_at, status.rotated_at);
let _ = fs::remove_file(path);
}
#[test]
fn sqlite_capability_authority_enforce_cluster_fence_rejects_stale_rotation() {
let path = unique_db_path("chio-authority-fence-stale-rotation");
let authority = SqliteCapabilityAuthority::open(&path).unwrap();
assert!(authority
.seed_cluster_fence(Some("http://leader-a"), 7)
.unwrap());
let fence = authority.cluster_fence().unwrap();
authority.rotate().unwrap();
let error = authority
.enforce_cluster_fence("http://leader-a", 7)
.expect_err("stale persisted fence should fail closed")
.to_string();
assert!(error.contains("persisted authority fence generation"));
assert!(error.contains(&fence.authority_generation.to_string()));
let _ = fs::remove_file(path);
}
#[test]
fn sqlite_capability_authority_seed_cluster_fence_refreshes_same_term_authority_state() {
let path = unique_db_path("chio-authority-fence-same-term-refresh");
let authority = SqliteCapabilityAuthority::open(&path).unwrap();
assert!(authority
.seed_cluster_fence(Some("http://leader-a"), 7)
.unwrap());
authority.rotate().unwrap();
assert!(authority
.seed_cluster_fence(Some("http://leader-a"), 7)
.unwrap());
authority
.enforce_cluster_fence("http://leader-a", 7)
.expect("same-term reseed should refresh authority generation");
let fence = authority.cluster_fence().unwrap();
let status = authority.status().unwrap();
assert_eq!(fence.election_term, 7);
assert_eq!(fence.leader_url.as_deref(), Some("http://leader-a"));
assert_eq!(fence.authority_generation, status.generation);
assert_eq!(fence.authority_rotated_at, status.rotated_at);
let _ = fs::remove_file(path);
}
}