use super::membership::{COVERAGE_LIMIT, EvidenceRef, FLOOR_LIMIT};
use super::*;
use crate::db::installation::InstallationGuard;
use crate::sync::{
SeedPublicationIntent,
seed_claim::{
Secret, SeedAuthority,
membership::{
self, Declaration, Device, Evidence, Invitation, Joiner, Membership,
VerifiedEnrollment, VerifiedKeys,
},
},
};
use anyhow::{Context, Result, ensure};
use serde::{Deserialize, Serialize};
#[cfg(any(test, feature = "test-support"))]
use std::sync::atomic::Ordering;
type Hash = [u8; 32];
const IDENTITY_LIMIT: usize = 8192;
const JOURNAL_LIMIT: usize = 4096;
const CANDIDATE_LIMIT: usize = 4 * membership::MAX_RECORD_BYTES + 1024;
const RESPONSE_LIMIT: usize = CANDIDATE_LIMIT + 4096;
const ATTEMPT_LIMIT: usize = 512;
pub const MAX_JOIN_ATTEMPTS: usize = 16;
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Identity {
role: String,
client: String,
incarnation: Hash,
locator: String,
authority: Vec<u8>,
}
impl Drop for Identity {
fn drop(&mut self) {
self.authority.zeroize();
}
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Outbound {
pub handle: Hash,
invitation: Vec<u8>,
pub declaration: Vec<u8>,
}
impl Drop for Outbound {
fn drop(&mut self) {
self.invitation.zeroize();
}
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Candidate {
predecessor: EvidenceRef,
record: Vec<u8>,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Verified {
evidence: EvidenceRef,
outcome: Hash,
key: Vec<u8>,
}
impl Drop for Verified {
fn drop(&mut self) {
self.key.zeroize();
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum EnrollmentReadiness {
NotSelected,
Pending,
Enrolled { head: Hash },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OpenInvitation {
pub handle: Hash,
pub expires_at: u64,
pub keys_may_have_been_sent: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum InvitationProgress {
Open,
Admitted,
Closed,
}
#[cfg(any(test, feature = "test-support"))]
#[derive(Debug, PartialEq, Eq)]
pub enum OutboundInvitation {
Pending,
Disclosed,
}
#[derive(Debug)]
pub struct PublishingBlocked;
impl std::fmt::Display for PublishingBlocked {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("error withdrawal-rotation-required")
}
}
impl std::error::Error for PublishingBlocked {}
enum Keys {
Seed(Box<SeedAuthority>),
Peer(Box<Joiner>),
}
impl Keys {
fn authority(&self) -> Device<'_> {
match self {
Self::Seed(s) => Device::seed(s),
Self::Peer(p) => p.authority(),
}
}
fn device(&self) -> Hash {
match self {
Self::Seed(s) => s.genesis().device_id(),
Self::Peer(p) => p.device(),
}
}
fn bearer(&self) -> &Secret {
match self {
Self::Seed(s) => s.bearer(),
Self::Peer(p) => p.bearer(),
}
}
}
pub struct ActiveInputs {
id: Identity,
keys: Keys,
pub membership: Membership,
pub evidence: Evidence,
coverage: VerifiedKeys,
_installation: InstallationGuard,
_lock: StoreLock,
}
impl ActiveInputs {
pub(super) fn authority(&self) -> Device<'_> {
self.keys.authority()
}
pub fn generation_keys(&self) -> &VerifiedKeys {
&self.coverage
}
pub fn device(&self) -> Hash {
self.keys.device()
}
pub fn bearer(&self) -> &Secret {
self.keys.bearer()
}
}
impl ProtectedLocalKeyStore {
pub(super) fn peer_authority_exists(&self) -> StoreResult<bool> {
Ok(self.read_record("peer-identity-authority", 32)?.is_some()
|| self.load_secret("peer-identity", IDENTITY_LIMIT)?.is_some())
}
pub(super) async fn phase(
&self,
db: &Database,
name: &str,
size: usize,
) -> Result<Option<Zeroizing<Vec<u8>>>> {
let pin = self.enrollment_pin(db, name).await?;
let bytes = self.read_owned(name, size, pin.is_some())?;
if let (Some(pin), Some(bytes)) = (pin, &bytes) {
ensure!(
pin == Sha256::digest(bytes.as_slice()).as_slice(),
"error enrollment-phase-mismatch"
);
}
Ok(bytes)
}
async fn save_phase(
&self,
db: &Database,
id: &Identity,
name: &str,
size: usize,
bytes: &[u8],
) -> Result<()> {
self.phase(db, name, size).await?;
self.write_owned(name, size, bytes)?;
let digest: Hash = Sha256::digest(bytes).into();
db.pin_enrollment_artifact(id.incarnation, name, digest)
.await?;
self.with_index(|index| {
if let Some(pins) = &mut index.pins {
pins.insert(name.to_string(), digest.to_vec());
}
});
Ok(())
}
async fn enrollment_pin(&self, db: &Database, name: &str) -> Result<Option<Vec<u8>>> {
match self.with_index(|index| index.pins.as_ref().map(|pins| pins.get(name).cloned())) {
None => db.enrollment_artifact(name).await,
Some(Some(pin)) => Ok(pin),
Some(None) => {
let pins = db.enrollment_artifacts().await?;
let pin = pins.get(name).cloned();
self.with_index(|index| index.pins = Some(pins));
Ok(pin)
}
}
}
pub(super) async fn save_management_phase(
&self,
db: &Database,
inputs: &ActiveInputs,
name: &str,
size: usize,
bytes: &[u8],
) -> Result<()> {
self.save_phase(db, &inputs.id, name, size, bytes).await
}
async fn identity(&self, db: &Database, guard: &InstallationGuard) -> Result<Option<Identity>> {
let pin = db.enrollment_pin().await?;
let Some(bytes) = self.read_owned("peer-identity", IDENTITY_LIMIT, pin.is_some())? else {
ensure!(
db.membership_checkpoint_mirror().await?.is_none()
&& self.journals(db).await?.is_empty(),
"error enrollment-protected-missing"
);
for (kind, size) in [
("peer-sent", 128),
("peer-response", RESPONSE_LIMIT),
("peer-verified", 2048),
("peer-ready", 128),
("peer-installed", 128),
] {
ensure!(
self.phase(db, kind, size).await?.is_none(),
"error enrollment-protected-missing"
);
}
for index in 1..MAX_JOIN_ATTEMPTS {
ensure!(
self.phase(db, &format!("peer-attempt-{index}"), ATTEMPT_LIMIT)
.await?
.is_none(),
"error enrollment-protected-missing"
);
}
for sequence in 1..=membership::MAX_TRANSITIONS + 1 {
ensure!(
self.read_owned(&format!("membership-floor-{sequence}"), FLOOR_LIMIT, false)?
.is_none(),
"error enrollment-protected-missing"
);
ensure!(
self.read_owned(
&format!("membership-coverage-{sequence}"),
COVERAGE_LIMIT,
false
)?
.is_none(),
"error enrollment-protected-missing"
);
}
return Ok(None);
};
let id: Identity = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
ensure!(
id.locator.len() <= 2048 && matches!(id.role.as_str(), "inviter" | "peer"),
"error enrollment-protected-framing"
);
if pin.is_none() {
ensure!(
self.read_owned("peer-ready", 128, false)?.is_none()
&& self.read_owned("peer-sent", 128, false)?.is_none()
&& self
.read_owned("outbound-0", JOURNAL_LIMIT, false)?
.is_none(),
"error enrollment-database-lost"
);
}
db.pin_enrollment(id.incarnation, &id.client, &id.role, guard)
.await?;
Ok(Some(id))
}
fn adopted_inputs(&self, seed: &SeedAuthority) -> Result<SeedPublicationIntent> {
let source = self.decode_source(
&self
.load_adoption_record("source", 104, true)?
.context("error enrollment-source-missing")?,
seed,
)?;
SeedPublicationIntent::from_protected_storage(
&self
.load_adoption_record("intent", 65536, true)?
.context("error enrollment-intent-missing")?,
&source,
seed.genesis(),
)
}
async fn seed_inputs(
&self,
db: &Database,
) -> Result<(SeedAuthority, Evidence, LocalSharedStatePackageKey)> {
let package = self.load_required_locked()?;
let seed = self.required_seed(&package)?;
let intent = self.adopted_inputs(&seed)?;
let (saved, state) = db
.seed_publication_intent_bytes()
.await?
.context("error seed-intent-missing")?;
ensure!(
state == "adopted" && saved == intent.protected_storage_bytes(),
"error seed-intent-mismatch"
);
let source = self.decode_source(
&self
.load_adoption_record("source", 104, true)?
.context("error seed-source-missing")?,
&seed,
)?;
ensure!(
db.seed_source_pin().await?.as_deref() == Some(source.protected_storage_bytes()),
"error seed-source-mismatch"
);
let publication = intent.publication(seed.genesis())?;
let evidence = Evidence {
genesis: seed.genesis().record().to_vec(),
publication: publication.record().to_vec(),
descriptor: intent.descriptor().to_vec(),
transitions: vec![],
};
Ok((
seed,
evidence,
LocalSharedStatePackageKey::new(*package.package_key().protected_storage_bytes()),
))
}
pub async fn active_inputs(&self, db: &Database, locator: &str) -> Result<ActiveInputs> {
Box::pin(self.load_active_inputs(db, locator)).await
}
async fn load_active_inputs(&self, db: &Database, locator: &str) -> Result<ActiveInputs> {
let installation = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
ensure!(locator.len() <= 2048, "error enrollment-locator-limit");
self.prepare()?;
let lock = self.lock()?;
let id = match self.identity(db, &installation).await? {
Some(id) => id,
None => {
let client = db.adopted_enrollment_client().await?;
self.seed_inputs(db).await?;
let id = Identity {
role: "inviter".into(),
client,
incarnation: *Secret::generate()?.expose(),
locator: locator.into(),
authority: vec![],
};
self.write_owned(
"peer-identity",
IDENTITY_LIMIT,
&Zeroizing::new(serde_json::to_vec(&id)?),
)?;
db.pin_enrollment(id.incarnation, &id.client, &id.role, &installation)
.await?;
id
}
};
ensure!(id.locator == locator, "error enrollment-context");
let (keys, original, original_membership, original_keys) = if id.role == "peer" {
ensure!(
self.phase(db, "peer-ready", 128).await?.is_some(),
"error enrollment-unresolved"
);
let (peer, verified, _, (evidence, current)) = self.verified_peer(db, &id).await?;
let installed = self
.phase(db, "peer-installed", 128)
.await?
.context("error snapshot-not-installed")?;
ensure!(
installed.as_slice() == verified.checkpoint()
&& db
.peer_snapshot_receipt(&verified, id.incarnation, &id.client, &installation)
.await?
.is_some(),
"error snapshot-receipt-mismatch"
);
(
Keys::Peer(Box::new(peer)),
evidence,
current,
verified.keys().clone(),
)
} else {
ensure!(
db.adopted_enrollment_client().await? == id.client,
"error enrollment-client-mismatch"
);
let (seed, evidence, key) = self.seed_inputs(db).await?;
let m = evidence.verify()?;
let coverage = m.verify_initial_key(&key)?;
(Keys::Seed(Box::new(seed)), evidence, m, coverage)
};
let (membership, evidence, coverage) =
if let Some(floor) = self.membership_floor(db, id.incarnation).await? {
floor
} else {
ensure!(
id.role == "inviter" && self.journals(db).await?.is_empty(),
"error membership-floor-missing"
);
let m = self
.adopt_membership(
db,
id.incarnation,
&original,
original_membership.clone(),
&original_keys,
)
.await?;
(m, original, original_keys)
};
ensure!(
membership.extends(&original_membership),
"error membership-original-mismatch"
);
keys.authority().validate(&membership)?;
coverage.validate(&membership)?;
let now = self.enrollment_now()?;
for journal in self.journals(db).await? {
if let Some(ready) = self.phase(db, &journal.name("ready"), 128).await? {
ensure!(
membership.contains_head(&ready.as_slice().try_into()?),
"error enrollment-checkpoint-mismatch"
);
} else if !self.retired(db, &journal).await?
&& !self.may_have_sent(db, &journal).await?
&& now >= Declaration::from_record(&membership, &journal.declaration)?.expiry()
{
self.save_phase(db, &id, &journal.name("retired"), 128, &journal.handle)
.await?;
}
}
Ok(ActiveInputs {
id,
keys,
membership,
evidence,
coverage,
_installation: installation,
_lock: lock,
})
}
pub async fn adopt_refresh(
&self,
db: &Database,
inputs: &mut ActiveInputs,
evidence: Evidence,
) -> Result<()> {
let target = evidence.verify()?;
self.adopt_verified_refresh(db, inputs, evidence, target)
.await
}
pub async fn adopt_verified_refresh(
&self,
db: &Database,
inputs: &mut ActiveInputs,
evidence: Evidence,
target: Membership,
) -> Result<()> {
let (membership, coverage) = self
.refresh_membership(
db,
inputs.id.incarnation,
inputs.keys.authority(),
&evidence,
target,
super::membership::Verified {
membership: &inputs.membership,
keys: &inputs.coverage,
},
)
.await?;
inputs.membership = membership;
inputs.coverage = coverage;
inputs.evidence = evidence;
Ok(())
}
async fn journals(&self, db: &Database) -> Result<Vec<Outbound>> {
let mut journals = Vec::new();
let mut gap = false;
for index in 0..membership::MAX_INVITATIONS {
let Some(bytes) = self
.phase(db, &format!("outbound-{index}"), JOURNAL_LIMIT)
.await?
else {
gap = true;
continue;
};
ensure!(!gap, "error enrollment-journal-missing");
let journal: Outbound = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-journal-corrupt"))?;
let invitation = Invitation::from_protected_storage(&journal.invitation)?;
ensure!(
journal.handle == invitation.handle(),
"error enrollment-journal-corrupt"
);
self.phase(db, &journal.name("registered"), 128).await?;
let bound = self.phase(db, &journal.name("bound"), 2048).await?;
let ready = self.phase(db, &journal.name("ready"), 128).await?;
let mut gap = false;
let mut resolved = ready.is_none();
for candidate in 0..membership::MAX_CANDIDATES {
let bytes = self
.phase(
db,
&journal.name(&format!("candidate-{candidate}")),
CANDIDATE_LIMIT,
)
.await?;
let sent = self
.phase(db, &journal.name(&format!("sent-{candidate}")), 128)
.await?;
if let Some(bytes) = bytes {
ensure!(!gap, "error enrollment-candidate-missing");
let saved: Candidate = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
let (_, before) = self.load_evidence(&saved.predecessor)?;
let after = before.append(
&journal.declaration,
bound.as_ref().context("error enrollment-binding-missing")?,
&saved.record,
)?;
if let Some(sent) = &sent {
ensure!(
sent.as_slice() == after.head(),
"error enrollment-sent-mismatch"
);
}
if ready.as_ref().is_some_and(|r| r.as_slice() == after.head()) {
ensure!(sent.is_some(), "error enrollment-sent-missing");
resolved = true;
}
} else {
gap = true;
ensure!(sent.is_none(), "error enrollment-candidate-missing");
}
}
ensure!(resolved, "error enrollment-candidate-missing");
journals.push(journal);
}
Ok(journals)
}
async fn retired(&self, db: &Database, journal: &Outbound) -> Result<bool> {
Ok(self
.phase(db, &journal.name("retired"), 128)
.await?
.is_some())
}
async fn closed(&self, db: &Database, journal: &Outbound) -> Result<bool> {
Ok(self.retired(db, journal).await?
|| self
.phase(db, &journal.name("withdrawn"), 128)
.await?
.is_some())
}
async fn may_have_sent(&self, db: &Database, journal: &Outbound) -> Result<bool> {
for attempt in 0..membership::MAX_CANDIDATES {
if self
.phase(db, &journal.name(&format!("sent-{attempt}")), 128)
.await?
.is_some()
{
return Ok(true);
}
}
Ok(false)
}
async fn unresolved_disclosures(
&self,
db: &Database,
inputs: &ActiveInputs,
) -> Result<Vec<(Outbound, u64)>> {
let mut unresolved = Vec::new();
for journal in self.journals(db).await? {
if self.phase(db, &journal.name("ready"), 128).await?.is_none()
&& !self.closed(db, &journal).await?
&& self.may_have_sent(db, &journal).await?
{
let expiry =
Declaration::from_record(&inputs.membership, &journal.declaration)?.expiry();
unresolved.push((journal, expiry));
}
}
Ok(unresolved)
}
pub async fn open_invitation(
&self,
db: &Database,
inputs: &ActiveInputs,
) -> Result<Option<OpenInvitation>> {
for journal in self.journals(db).await? {
if self.phase(db, &journal.name("ready"), 128).await?.is_none()
&& !self.closed(db, &journal).await?
{
let declaration =
Declaration::from_record(&inputs.membership, &journal.declaration)?;
return Ok(Some(OpenInvitation {
handle: journal.handle,
expires_at: declaration.expiry(),
keys_may_have_been_sent: self.may_have_sent(db, &journal).await?,
}));
}
}
Ok(None)
}
pub async fn invitation_progress(
&self,
db: &Database,
handle: &Hash,
) -> Result<InvitationProgress> {
let recorded = async |name| {
Ok::<_, anyhow::Error>(
self.phase(db, &invitation_phase_name(handle, name), 128)
.await?
.is_some(),
)
};
if recorded("ready").await? {
Ok(InvitationProgress::Admitted)
} else if recorded("retired").await? || recorded("withdrawn").await? {
Ok(InvitationProgress::Closed)
} else {
Ok(InvitationProgress::Open)
}
}
pub async fn retire_open_invitation(
&self,
db: &Database,
inputs: &ActiveInputs,
) -> Result<(Option<Outbound>, Option<OpenInvitation>)> {
let Some(state) = self.open_invitation(db, inputs).await? else {
return Ok((None, None));
};
if state.keys_may_have_been_sent {
return Ok((None, Some(state)));
}
let journal = self
.journals(db)
.await?
.into_iter()
.find(|journal| journal.handle == state.handle)
.context("error enrollment-invitation-missing")?;
self.save_phase(
db,
&inputs.id,
&journal.name("retired"),
128,
&journal.handle,
)
.await?;
Ok((Some(journal), Some(state)))
}
pub async fn prepare_invitation(
&self,
db: &Database,
inputs: &ActiveInputs,
expires: Option<u64>,
handle: Option<Hash>,
) -> Result<Outbound> {
let mut journals = self.journals(db).await?;
if let Some(handle) = handle {
return journals
.into_iter()
.find(|j| j.handle == handle)
.context("error enrollment-invitation-missing");
}
if expires.is_none() {
return journals
.pop()
.context("error enrollment-invitation-missing");
}
for journal in &journals {
if self.phase(db, &journal.name("ready"), 128).await?.is_none()
&& !self.closed(db, journal).await?
{
ensure!(
self.phase(db, &journal.name("sent-0"), 128)
.await?
.is_none(),
"error withdrawal-required-unsupported"
);
return journals
.pop()
.context("error enrollment-invitation-missing");
}
}
ensure!(
journals.len() < membership::MAX_INVITATIONS,
"error membership-limit"
);
let (invitation, d) = inputs.keys.authority().prepare_invitation(
&inputs.membership,
expires.context("error enrollment-expiry")?,
)?;
let journal = Outbound {
handle: invitation.handle(),
invitation: invitation.protected_storage_bytes().to_vec(),
declaration: d.record().to_vec(),
};
self.save_phase(
db,
&inputs.id,
&format!("outbound-{}", journals.len()),
JOURNAL_LIMIT,
&Zeroizing::new(serde_json::to_vec(&journal)?),
)
.await?;
Ok(journal)
}
pub async fn registered_invitation(
&self,
db: &Database,
inputs: &ActiveInputs,
journal: &Outbound,
) -> Result<Invitation> {
self.save_phase(
db,
&inputs.id,
&journal.name("registered"),
128,
&Sha256::digest(&journal.declaration),
)
.await?;
Invitation::from_protected_storage(&journal.invitation)
}
pub async fn prepare_admission(
&self,
db: &Database,
inputs: &ActiveInputs,
journal: &Outbound,
request: &[u8],
) -> Result<Vec<u8>> {
ensure!(
self.phase(db, &journal.name("registered"), 128)
.await?
.is_some(),
"error enrollment-not-registered"
);
ensure!(
!self.retired(db, journal).await?,
"error enrollment-invitation-retired"
);
ensure!(
self.phase(db, &journal.name("withdrawing"), 128)
.await?
.is_none()
&& !self.closed(db, journal).await?,
"error enrollment-invitation-withdrawing"
);
let d = Declaration::from_record(&inputs.membership, &journal.declaration)?;
let expired = self.enrollment_now()? >= d.expiry();
let mut index = 0;
for attempt in 0..membership::MAX_CANDIDATES {
let Some(bytes) = self
.phase(
db,
&journal.name(&format!("candidate-{attempt}")),
CANDIDATE_LIMIT,
)
.await?
else {
break;
};
self.save_phase(db, &inputs.id, &journal.name("bound"), 2048, request)
.await?;
let candidate: Candidate = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
let (_, predecessor) = self.load_evidence(&candidate.predecessor)?;
let result = predecessor.append(&journal.declaration, request, &candidate.record)?;
ensure!(
inputs.membership.extends(&predecessor),
"error enrollment-candidate-fork"
);
if inputs.membership.contains_head(&result.head())
|| inputs.membership.head() == predecessor.head()
{
let sent = journal.name(&format!("sent-{attempt}"));
ensure!(
!expired || self.phase(db, &sent, 128).await?.is_some(),
"error enrollment-expired"
);
self.save_phase(
db,
&inputs.id,
&journal.name(&format!("sent-{attempt}")),
128,
&result.head(),
)
.await?;
return Ok(candidate.record);
}
ensure!(
inputs
.membership
.head_at(result.sequence())
.is_some_and(|head| head != result.head()),
"error enrollment-candidate-unresolved"
);
index = attempt + 1;
}
ensure!(
index < membership::MAX_CANDIDATES,
"error membership-candidate-limit"
);
ensure!(!expired, "error enrollment-expired");
let inv = Invitation::from_protected_storage(&journal.invitation)?;
let record = inputs.keys.authority().prepare_admission(
&inputs.membership,
&d,
&inv,
request,
inputs.generation_keys(),
)?;
self.save_phase(db, &inputs.id, &journal.name("bound"), 2048, request)
.await?;
let candidate = Candidate {
predecessor: self.save_evidence(&inputs.evidence)?,
record: record.clone(),
};
self.save_phase(
db,
&inputs.id,
&journal.name(&format!("candidate-{index}")),
CANDIDATE_LIMIT,
&serde_json::to_vec(&candidate)?,
)
.await?;
self.save_phase(
db,
&inputs.id,
&journal.name(&format!("sent-{index}")),
128,
&Sha256::digest(&record),
)
.await?;
Ok(record)
}
pub async fn expired_disclosure(
&self,
db: &Database,
inputs: &ActiveInputs,
) -> Result<Option<Outbound>> {
let now = self.enrollment_now()?;
Ok(self
.unresolved_disclosures(db, inputs)
.await?
.into_iter()
.find(|(_, expiry)| now >= *expiry)
.map(|(journal, _)| journal))
}
pub async fn mark_withdrawing(
&self,
db: &Database,
inputs: &ActiveInputs,
journal: &Outbound,
) -> Result<()> {
self.save_phase(
db,
&inputs.id,
&journal.name("withdrawing"),
128,
&journal.handle,
)
.await
}
pub async fn reconcile_disclosure(
&self,
db: &Database,
inputs: &ActiveInputs,
journal: &Outbound,
) -> Result<Disclosure> {
let bound = self
.phase(db, &journal.name("bound"), 2048)
.await?
.context("error enrollment-binding-missing")?;
let mut candidates = Vec::new();
for attempt in 0..membership::MAX_CANDIDATES {
let Some(bytes) = self
.phase(
db,
&journal.name(&format!("candidate-{attempt}")),
CANDIDATE_LIMIT,
)
.await?
else {
break;
};
let saved: Candidate = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
let (_, before) = self.load_evidence(&saved.predecessor)?;
ensure!(
inputs.membership.extends(&before),
"error enrollment-candidate-fork"
);
let after = before.append(&journal.declaration, &bound, &saved.record)?;
match inputs.membership.head_at(after.sequence()) {
Some(head) if head == after.head() => {
self.finish_inviter(db, inputs, journal, &saved.record)
.await?;
return Ok(Disclosure::Admitted);
}
Some(_) => {}
None => return Ok(Disclosure::Unresolved),
}
let recipient = after
.devices()
.find(|device| !before.has_device(*device))
.context("error enrollment-candidate-recipient")?;
candidates.push((before, after.sequence(), recipient));
}
let disclosed = candidates
.iter()
.flat_map(|(before, _, _)| before.generations().iter().map(|g| g.id))
.collect::<std::collections::HashSet<_>>();
let last_slot = candidates.iter().map(|(_, slot, _)| *slot).max();
let mut state = candidates
.iter()
.map(|(before, _, _)| before)
.min_by_key(|before| before.sequence())
.context("error enrollment-candidate-missing")?
.clone();
let mut qualifying = None;
for t in inputs
.evidence
.transitions
.iter()
.skip(state.sequence() as usize - 1)
{
state = state.append(&t.declaration, &t.request, &t.record)?;
ensure!(
candidates
.iter()
.all(|(_, _, recipient)| !state.has_device(*recipient)),
"error withdrawal-recipient-member"
);
if qualifying.is_none()
&& Some(state.sequence()) > last_slot
&& !state.rotation_pending()
&& !disclosed.contains(&state.current_generation().id)
{
qualifying = Some(state.head());
}
}
ensure!(
state.head() == inputs.membership.head(),
"error enrollment-checkpoint-mismatch"
);
let Some(head) = qualifying else {
return Ok(Disclosure::Unresolved);
};
self.save_phase(db, &inputs.id, &journal.name("withdrawn"), 128, &head)
.await?;
Ok(Disclosure::Withdrawn)
}
pub async fn finish_inviter(
&self,
db: &Database,
inputs: &ActiveInputs,
journal: &Outbound,
record: &[u8],
) -> Result<()> {
let head: Hash = Sha256::digest(record).into();
ensure!(
inputs.membership.contains_head(&head),
"error enrollment-outcome-missing"
);
let mut matched = false;
for attempt in 0..membership::MAX_CANDIDATES {
let Some(bytes) = self
.phase(
db,
&journal.name(&format!("candidate-{attempt}")),
CANDIDATE_LIMIT,
)
.await?
else {
break;
};
let candidate: Candidate = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
if candidate.record == record {
let sent = self
.phase(db, &journal.name(&format!("sent-{attempt}")), 128)
.await?
.context("error enrollment-sent-missing")?;
ensure!(sent.as_slice() == head, "error enrollment-sent-mismatch");
matched = true;
}
}
ensure!(matched, "error enrollment-outcome-mismatch");
self.save_phase(db, &inputs.id, &journal.name("ready"), 128, &head)
.await
}
pub async fn prepare_peer(
&self,
db: &Database,
locator: &str,
invitation: Option<Invitation>,
) -> Result<Joiner> {
self.select_peer(db, locator, invitation, false).await
}
pub async fn replace_peer(
&self,
db: &Database,
locator: &str,
invitation: Invitation,
) -> Result<Joiner> {
self.select_peer(db, locator, Some(invitation), true).await
}
async fn select_peer(
&self,
db: &Database,
locator: &str,
invitation: Option<Invitation>,
replace: bool,
) -> Result<Joiner> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
ensure!(locator.len() <= 2048, "error enrollment-locator-limit");
self.prepare()?;
let _lock = self.lock()?;
let id = match self.identity(db, &guard).await? {
Some(id) => id,
None => {
let client = db.peer_target_preflight().await?;
guard.ensure_unbound()?;
ensure!(
self.load_secret(KEYRING_ITEM, KEYRING_BYTES)?.is_none()
&& self.read_marker()?.is_none()
&& !self.seed_authority_exists()?,
"error enrollment-existing-authority"
);
let peer = Joiner::generate(
invitation
.as_ref()
.map(|i| Invitation::from_protected_storage(&i.protected_storage_bytes()))
.transpose()?
.context("error enrollment-invitation-required")?,
)?;
guard.fence()?;
let id = Identity {
role: "peer".into(),
client,
incarnation: *Secret::generate()?.expose(),
locator: locator.into(),
authority: peer.protected_storage_bytes().to_vec(),
};
self.write_owned(
"peer-identity",
IDENTITY_LIMIT,
&Zeroizing::new(serde_json::to_vec(&id)?),
)?;
db.pin_enrollment(id.incarnation, &id.client, &id.role, &guard)
.await?;
id
}
};
ensure!(
id.role == "peer" && id.locator == locator,
"error enrollment-context"
);
let mut attempts = self.attempts(db, &id).await?;
let known = invitation
.as_ref()
.map(|i| attempts.iter().position(|peer| peer.matches_invitation(i)));
Ok(match (known, self.response(db).await?) {
(Some(None), _) => {
ensure!(replace, "error enrollment-invitation-conflict");
self.ensure_retry_allowed(db, &id, &guard, attempts.len())
.await?;
let peer = attempts[0].retry(invitation.expect("unknown invitation"))?;
self.save_phase(
db,
&id,
&format!("peer-attempt-{}", attempts.len()),
ATTEMPT_LIMIT,
&peer.attempt_bytes(),
)
.await?;
peer
}
(_, Some(mail)) => responding(attempts, &mail)?,
(Some(Some(index)), None) => attempts.swap_remove(index),
(None, None) => attempts.pop().expect("original attempt"),
})
}
pub async fn peer_attempts(&self, db: &Database, locator: &str) -> Result<Vec<Joiner>> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
let _lock = self.lock()?;
let id = self
.identity(db, &guard)
.await?
.context("error enrollment-missing")?;
ensure!(
id.role == "peer" && id.locator == locator,
"error enrollment-context"
);
let attempts = self.attempts(db, &id).await?;
Ok(match self.response(db).await? {
Some(mail) => vec![responding(attempts, &mail)?],
None => attempts,
})
}
async fn attempts(&self, db: &Database, id: &Identity) -> Result<Vec<Joiner>> {
let original = Joiner::from_protected_storage(&id.authority)?;
let sent = Sha256::digest(original.request());
if let Some(saved) = self.phase(db, "peer-sent", 128).await? {
ensure!(
saved.as_slice() == sent.as_slice(),
"error enrollment-sent-mismatch"
);
}
self.save_phase(db, id, "peer-sent", 128, &sent).await?;
let mut attempts = Vec::new();
let mut gap = false;
for index in 1..MAX_JOIN_ATTEMPTS {
let name = format!("peer-attempt-{index}");
let Some(bytes) = self.phase(db, &name, ATTEMPT_LIMIT).await? else {
gap = true;
continue;
};
ensure!(!gap, "error enrollment-attempt-missing");
if db.enrollment_artifact(&name).await?.is_none() {
self.save_phase(db, id, &name, ATTEMPT_LIMIT, &bytes)
.await?;
}
attempts.push(original.attempt(&bytes)?);
}
attempts.insert(0, original);
Ok(attempts)
}
async fn ensure_retry_allowed(
&self,
db: &Database,
id: &Identity,
guard: &InstallationGuard,
attempts: usize,
) -> Result<()> {
for (kind, size) in [
("peer-response", RESPONSE_LIMIT),
("peer-verified", 2048),
("peer-ready", 128),
("peer-installed", 128),
] {
ensure!(
self.phase(db, kind, size).await?.is_none(),
"error enrollment-retry-unavailable"
);
}
ensure!(
self.membership_floor(db, id.incarnation).await?.is_none()
&& db.membership_checkpoint_mirror().await?.is_none()
&& self.journals(db).await?.is_empty(),
"error enrollment-retry-unavailable"
);
db.peer_retry_preflight(id.incarnation, &id.client, guard)
.await?;
ensure!(attempts < MAX_JOIN_ATTEMPTS, "error enrollment-retry-limit");
Ok(())
}
async fn response(&self, db: &Database) -> Result<Option<membership::Mailbox>> {
self.phase(db, "peer-response", RESPONSE_LIMIT)
.await?
.map(|bytes| {
serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))
})
.transpose()
}
pub async fn open_peer_response(
&self,
db: &Database,
mail: &membership::Mailbox,
) -> Result<membership::ProvisionalGrant> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
let _lock = self.lock()?;
let id = self
.identity(db, &guard)
.await?
.context("error enrollment-missing")?;
ensure!(id.role == "peer", "error enrollment-role");
responding(self.attempts(db, &id).await?, mail)?.open_provisional(
&mail.declaration,
mail.admission
.as_deref()
.context("error enrollment-outcome-missing")?,
)
}
pub async fn finish_peer(
&self,
db: &Database,
mail: &membership::Mailbox,
evidence: &Evidence,
) -> Result<()> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
let _lock = self.lock()?;
let id = self
.identity(db, &guard)
.await?
.context("error enrollment-missing")?;
ensure!(id.role == "peer", "error enrollment-role");
let peer = responding(self.attempts(db, &id).await?, mail)?;
let grant = peer.open_provisional(
&mail.declaration,
mail.admission
.as_deref()
.context("error enrollment-outcome-missing")?,
)?;
let (verified, current) = evidence.enrollment(&peer, grant.outcome)?;
self.save_phase(
db,
&id,
"peer-response",
RESPONSE_LIMIT,
&serde_json::to_vec(mail)?,
)
.await?;
let mut original = evidence.clone();
original
.transitions
.truncate(verified.membership().sequence() as usize - 1);
let record = Verified {
evidence: self.save_evidence(&original)?,
outcome: grant.outcome,
key: verified.key().protected_storage_bytes().to_vec(),
};
self.save_phase(
db,
&id,
"peer-verified",
2048,
&Zeroizing::new(serde_json::to_vec(&record)?),
)
.await?;
if let Some((floor, _, _)) = self.membership_floor(db, id.incarnation).await?
&& floor.extends(¤t)
{
peer.authority().validate(&floor)?;
self.save_phase(db, &id, "peer-ready", 128, &record.outcome)
.await?;
return Ok(());
}
self.refresh_membership(
db,
id.incarnation,
peer.authority(),
evidence,
current,
super::membership::Verified {
membership: verified.membership(),
keys: verified.keys(),
},
)
.await?;
self.save_phase(db, &id, "peer-ready", 128, &record.outcome)
.await
}
async fn verified_peer(
&self,
db: &Database,
id: &Identity,
) -> Result<(Joiner, VerifiedEnrollment, Verified, (Evidence, Membership))> {
let ready = self
.phase(db, "peer-ready", 128)
.await?
.context("error enrollment-not-ready")?;
let bytes = self
.phase(db, "peer-verified", 2048)
.await?
.context("error enrollment-key-coverage-missing")?;
let record: Verified = serde_json::from_slice(&bytes)
.map_err(|_| anyhow::anyhow!("error enrollment-protected-framing"))?;
let mail = self
.response(db)
.await?
.context("error enrollment-response-missing")?;
let peer = responding(self.attempts(db, id).await?, &mail)?;
let evidence = Evidence::decode_unverified(&self.load_evidence_bytes(&record.evidence)?)?;
let (verified, current) = evidence.enrollment(&peer, record.outcome)?;
ensure!(
ready.as_slice() == record.outcome
&& verified.key().protected_storage_bytes().as_slice() == record.key,
"error enrollment-verified-corrupt"
);
Ok((peer, verified, record, (evidence, current)))
}
pub async fn install_peer_snapshot(
&self,
db: &Database,
transport: &crate::sync::client::enrollment::Client,
locator: &str,
downloaded: &(dyn Fn(u64, Option<u64>) + Sync),
) -> Result<crate::sync::SharedStateInstallReport> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
let _lock = self.lock()?;
let id = self
.identity(db, &guard)
.await?
.context("error enrollment-missing")?;
ensure!(
id.role == "peer" && id.locator == locator,
"error enrollment-context"
);
let (peer, verified, _, (evidence, _)) = self.verified_peer(db, &id).await?;
let completed = self.phase(db, "peer-installed", 128).await?;
if let Some(report) = db
.peer_snapshot_receipt(&verified, id.incarnation, &id.client, &guard)
.await?
{
let (floor, _, _) = self
.membership_floor(db, id.incarnation)
.await?
.context("error membership-floor-missing")?;
peer.authority().validate(&floor)?;
self.save_phase(db, &id, "peer-installed", 128, &verified.checkpoint())
.await?;
return Ok(report);
}
ensure!(
completed.is_none(),
"error snapshot-installed-database-lost"
);
let (floor, _, _) = self
.membership_floor(db, id.incarnation)
.await?
.context("error membership-floor-missing")?;
peer.authority().validate(&floor)?;
let package = transport
.download(
self,
db,
id.incarnation,
&peer,
&verified,
&evidence.descriptor,
downloaded,
)
.await?;
let report = db
.install_peer_snapshot(&verified, id.incarnation, &id.client, &guard, &package)
.await?;
self.save_phase(db, &id, "peer-installed", 128, &verified.checkpoint())
.await?;
Ok(report)
}
pub async fn adopt_download_refresh(
&self,
db: &Database,
identity: Hash,
peer: &Joiner,
verified: &VerifiedEnrollment,
evidence: &Evidence,
target: Membership,
) -> Result<Membership> {
Ok(self
.refresh_membership(
db,
identity,
peer.authority(),
evidence,
target,
super::membership::Verified {
membership: verified.membership(),
keys: verified.keys(),
},
)
.await?
.0)
}
pub async fn tail_inputs(&self, db: &Database, locator: &str) -> Result<TailSnapshot> {
let inputs = self.active_inputs(db, locator).await?;
let withdrawal_deadline = self
.unresolved_disclosures(db, &inputs)
.await?
.into_iter()
.map(|(_, expiry)| expiry)
.min();
let b = inputs.membership.publication().binding();
let association = format!(
"{}:{}:{}",
hex::encode(b.vault_id),
hex::encode(b.stream_id),
hex::encode(b.bootstrap_id)
);
ensure!(
db.meta("e2ee_association").await?.as_deref() == Some(association.as_str()),
"error encrypted-tail-association"
);
let authority = crate::sync::encrypted_tail::Authority {
context: crate::sync::encrypted_tail::Context {
vault: b.vault_id,
genesis: inputs.membership.genesis().commitment(),
device: inputs.device(),
head: inputs.membership.head(),
stream: b.stream_id,
descriptor: b.descriptor_commitment,
},
membership: inputs.membership.clone(),
keys: inputs.coverage.clone(),
prefix: i64::try_from(b.prefix_count)?,
association,
sync_generation: db
.meta("sync_generation")
.await?
.context("error encrypted-tail-generation")?
.parse()?,
};
let snapshot = TailSnapshot {
authority,
bearer: Secret::new(*inputs.bearer().expose()),
withdrawal_deadline,
enrollment_marker: db.enrollment_artifact_marker().await?,
#[cfg(any(test, feature = "test-support"))]
enrollment_clock: self.enrollment_clock.clone(),
};
drop(inputs);
Ok(snapshot)
}
pub async fn association(&self, db: &Database) -> Result<Option<(bool, String)>> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
self.prepare()?;
let _lock = self.lock()?;
Ok(self
.identity(db, &guard)
.await?
.map(|id| (id.role == "peer", id.locator.clone())))
}
#[cfg(any(test, feature = "test-support"))]
pub async fn outbound_invitation(&self, db: &Database) -> Result<Option<OutboundInvitation>> {
let _guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
self.prepare()?;
let _lock = self.lock()?;
for journal in self.journals(db).await? {
if self.phase(db, &journal.name("ready"), 128).await?.is_none()
&& !self.closed(db, &journal).await?
{
return Ok(Some(if self.may_have_sent(db, &journal).await? {
OutboundInvitation::Disclosed
} else {
OutboundInvitation::Pending
}));
}
}
Ok(None)
}
pub async fn enrollment_readiness(&self, db: &Database) -> Result<EnrollmentReadiness> {
let guard = InstallationGuard::acquire(db.path())?;
self.validate_database(db).await?;
self.prepare()?;
let _lock = self.lock()?;
let Some(id) = self.identity(db, &guard).await? else {
return Ok(EnrollmentReadiness::NotSelected);
};
if id.role == "peer" && self.phase(db, "peer-ready", 128).await?.is_none() {
return Ok(EnrollmentReadiness::Pending);
}
let Some((m, _, _)) = self.membership_floor(db, id.incarnation).await? else {
return Ok(EnrollmentReadiness::Pending);
};
if id.role == "peer" {
self.verified_peer(db, &id)
.await?
.0
.authority()
.validate(&m)?;
} else {
Device::seed(&self.seed_inputs(db).await?.0).validate(&m)?;
}
Ok(EnrollmentReadiness::Enrolled { head: m.head() })
}
}
fn responding(attempts: Vec<Joiner>, mail: &membership::Mailbox) -> Result<Joiner> {
attempts
.into_iter()
.find(|peer| mail.request.as_deref() == Some(peer.request()))
.context("error enrollment-request-mismatch")
}
impl Outbound {
fn name(&self, phase: &str) -> String {
invitation_phase_name(&self.handle, phase)
}
}
fn invitation_phase_name(handle: &Hash, phase: &str) -> String {
format!("invite-{}-{phase}", hex::encode(handle))
}
#[derive(Debug, PartialEq, Eq)]
pub enum Disclosure {
Admitted,
Withdrawn,
Unresolved,
}
pub struct TailSnapshot {
pub authority: crate::sync::encrypted_tail::Authority,
pub bearer: Secret,
withdrawal_deadline: Option<u64>,
enrollment_marker: i64,
#[cfg(any(test, feature = "test-support"))]
enrollment_clock: Option<std::sync::Arc<std::sync::atomic::AtomicU64>>,
}
impl TailSnapshot {
pub fn require_publishing_ready(&self) -> Result<()> {
if self.publishing_blocked()? {
return Err(PublishingBlocked.into());
}
Ok(())
}
pub fn publishing_blocked(&self) -> Result<bool> {
#[cfg(any(test, feature = "test-support"))]
let now = enrollment_now(self.enrollment_clock.as_ref())?;
#[cfg(not(any(test, feature = "test-support")))]
let now = enrollment_now()?;
Ok(self
.withdrawal_deadline
.is_some_and(|deadline| now >= deadline))
}
pub async fn is_current(&self, db: &Database) -> Result<bool> {
Ok(self.enrollment_marker == db.enrollment_artifact_marker().await?)
}
}
impl ProtectedLocalKeyStore {
fn enrollment_now(&self) -> Result<u64> {
#[cfg(any(test, feature = "test-support"))]
return enrollment_now(self.enrollment_clock.as_ref());
#[cfg(not(any(test, feature = "test-support")))]
enrollment_now()
}
}
#[cfg(any(test, feature = "test-support"))]
fn enrollment_now(clock: Option<&std::sync::Arc<std::sync::atomic::AtomicU64>>) -> Result<u64> {
match clock {
Some(clock) => Ok(clock.load(Ordering::SeqCst)),
None => crate::sync::client::engine::unix_now(),
}
}
#[cfg(not(any(test, feature = "test-support")))]
fn enrollment_now() -> Result<u64> {
crate::sync::client::engine::unix_now()
}