use super::*;
use crate::auth_membership::{AuthorityState, MembershipAuthority};
use lys_identity::grants::change_stream::{ChangeFrame, ChangesPage, LogName};
use std::sync::Arc;
pub fn configuration() -> Result<GrantStreamConfig, StreamError> {
toml::from_str(
r#"
log_identity = "fixture-log"
log_epoch = 1
credential_path = "/fixture/app-credential"
service_key_path = "/fixture/service-public-key"
page_positions = 16
max_frames = 18
max_response_bytes = 8388608
max_event_bytes = 65536
max_revoked_grants = 65536
max_credential_bytes = 65536
max_keyset_bytes = 65536
max_transport_reopens = 2
max_rights = 64
max_chain_grants = 16
max_chains = 8
enforce_revocation = true
[admission]
kind = "liminal.network"
id = "fixture-network"
action = "call"
[[subjects]]
kind = "liminal.subject"
resource = "fixture-channel"
prefix = "fixture.channel"
"#,
)
.map_err(|_| StreamError::named("fixture_config_invalid"))
}
fn log() -> lys_pass::membership::GrantLog {
lys_pass::membership::GrantLog {
identity: "fixture-log".into(),
epoch: 1,
}
}
fn baseline(revision: u64) -> ChangesPage {
ChangesPage {
log: LogName {
identity: "fixture-log".into(),
epoch: 1,
},
frames: vec![
ChangeFrame::Baseline {
revision,
revoked: Vec::new(),
},
ChangeFrame::PlacementMark {
revision,
placement: 0,
},
ChangeFrame::Ready { revision },
],
}
}
#[test]
fn an_installed_authority_is_closed_until_the_complete_ready_barrier()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
assert_eq!(state.state(), AuthorityState::NotReady);
let snapshot = serde_json::to_value(state.snapshot()?)?;
assert_eq!(snapshot["readiness"], "not_ready");
assert!(snapshot["cursor"].is_null());
assert!(snapshot["last_error"].is_null());
Ok(())
}
#[test]
fn complete_baseline_and_tail_have_one_shared_applied_cursor()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(3))?;
assert_eq!(state.snapshot()?.readiness, "ready");
assert_eq!(state.snapshot()?.cursor, Some(3));
engine.page(ChangesPage {
log: baseline(3).log,
frames: vec![
ChangeFrame::Watermark { revision: 5 },
ChangeFrame::PlacementMark {
revision: 5,
placement: 1,
},
ChangeFrame::Ready { revision: 5 },
],
})?;
assert_eq!(state.snapshot()?.cursor, Some(5));
assert_eq!(state.snapshot()?.placed, Some(1));
assert_eq!(
engine
.authority
.diagnostics()?
.applied
.map(crate::auth_revocation::Revision::get),
Some(5)
);
Ok(())
}
#[test]
fn ready_ahead_of_cursor_is_a_named_gap_and_never_admits() -> Result<(), Box<dyn std::error::Error>>
{
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(1))?;
let error = engine
.page(ChangesPage {
log: baseline(1).log,
frames: vec![ChangeFrame::Ready { revision: 4 }],
})
.err()
.ok_or("ahead Ready was accepted")?;
assert_eq!(error.name(), "revocation_stream_gap");
assert_eq!(state.snapshot()?.readiness, "gap");
assert_eq!(state.snapshot()?.cursor, Some(1));
Ok(())
}
#[test]
fn reset_retains_the_old_cursor_and_closes_the_old_log_domain()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(2))?;
let mut reset = baseline(0);
reset.log.epoch = 2;
reset.frames = vec![ChangeFrame::Reset {
reason: lys_identity::grants::change_stream::ResetReason::OtherLog,
from: Some(baseline(2).log),
}];
assert_eq!(
engine.page(reset).err().ok_or("reset accepted")?.name(),
"revocation_log_reset"
);
assert_eq!(state.snapshot()?.readiness, "reset");
assert_eq!(state.snapshot()?.cursor, Some(2));
Ok(())
}
#[test]
fn unknown_frame_and_remote_words_are_never_debug_or_diagnostic_data()
-> Result<(), Box<dyn std::error::Error>> {
let payload = br#"{"log":{"identity":"fixture-log","epoch":1},"frames":[{"frame":"draft_decision","words":"secret payload"}]}"#;
assert!(serde_json::from_slice::<ChangesPage>(payload).is_err());
let state = StreamState::for_log(log());
state.unavailable("grant_stream_frame_invalid")?;
assert!(!format!("{state:?}").contains("secret payload"));
assert!(crate::auth_lys::live::drafts::stream_signal("revoke").is_none());
assert!(crate::auth_lys::live::drafts::stream_signal("draft_decision").is_some());
Ok(())
}
#[test]
fn cold_decisions_use_the_complete_ready_baseline_and_never_a_stale_allow()
-> Result<(), Box<dyn std::error::Error>> {
use crate::auth_revocation::{GrantDecisionSource, GrantId, GrantLogId, Revision};
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
let grant = GrantId::new("grant-00000000000000000000000000000001")?;
let asked = GrantLogId::new("fixture-log", 1)?;
assert!(
state
.decide(&asked, std::slice::from_ref(&grant), Revision::new(1))
.is_err()
);
let mut page = baseline(1);
page.frames[0] = ChangeFrame::Baseline {
revision: 1,
revoked: vec![grant.as_str().into()],
};
engine.page(page)?;
assert!(state.decide(&asked, &[grant], Revision::new(1))?.decisions[0].1);
state.unavailable("revocation_stream_unavailable")?;
assert!(state.decide(&asked, &[], Revision::new(1)).is_err());
Ok(())
}
#[path = "root_question_tests.rs"]
mod root_question;
#[path = "socket_tests.rs"]
mod sockets;
#[test]
fn transport_loss_preserves_both_cursors_and_requires_a_new_ready_fence()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(2))?;
engine.lost()?;
assert_eq!(state.snapshot()?.cursor, Some(2));
assert_eq!(state.snapshot()?.placed, Some(0));
assert_eq!(
state.snapshot()?.last_error,
Some("revocation_stream_unavailable")
);
assert_eq!(
engine.authority.diagnostics()?.readiness,
crate::auth_revocation::Readiness::Unavailable
);
engine.page(ChangesPage {
log: baseline(2).log,
frames: vec![
ChangeFrame::Watermark { revision: 3 },
ChangeFrame::PlacementMark {
revision: 3,
placement: 0,
},
ChangeFrame::Ready { revision: 3 },
],
})?;
assert_eq!(state.snapshot()?.readiness, "ready");
assert_eq!(state.snapshot()?.generation, 2);
Ok(())
}
fn signed_revoke(
seed: u8,
grant: u8,
) -> Result<(lys_core::Ed25519Identity, ChangeFrame), Box<dyn std::error::Error>> {
use lys_identity::grants::{GrantChange, GrantEvent, GrantId, sign_grant_event};
let identity = lys_core::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([seed; 32]));
let event = GrantEvent::new(
lys_identity::OperationId::from_bytes([3; 16]),
lys_identity::IdentityId::Person(lys_identity::PersonId::from_bytes([2; 16])),
1000,
GrantChange::Revoke {
grant: GrantId::from_bytes([grant; 16]),
reason: "right withdrawn".into(),
},
)?;
let event = sign_grant_event(event, &identity)?;
let frame = ChangeFrame::Change {
revision: 1,
grant: event.event().grant().to_string(),
change: lys_identity::grants::change_stream::ChangeKind::Revoke,
event: lys_identity::grants::change_stream::event_hex(event.bytes()),
};
Ok((identity, frame))
}
#[test]
fn signed_revocation_duplicates_are_idempotent_and_conflicting_evidence_is_terminal()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
let (key, frame) = signed_revoke(7, 1)?;
engine.service_key = key.public_key_bytes();
engine.page(baseline(0))?;
let page = |frame| ChangesPage {
log: baseline(0).log,
frames: vec![
frame,
ChangeFrame::PlacementMark {
revision: 1,
placement: 0,
},
ChangeFrame::Ready { revision: 1 },
],
};
engine.page(page(frame.clone()))?;
engine.page(page(frame))?;
let diagnostics = engine.authority.diagnostics()?;
assert_eq!(diagnostics.counts.observed_revocations, 1);
assert_eq!(diagnostics.counts.duplicate_revisions, 1);
let (_, conflict) = signed_revoke(7, 2)?;
assert_eq!(
engine
.page(page(conflict))
.err()
.ok_or("conflicting evidence was accepted")?
.name(),
"grant_stream_evidence_invalid"
);
assert_eq!(
engine.authority.diagnostics()?.readiness,
crate::auth_revocation::Readiness::Terminal
);
Ok(())
}
#[test]
fn an_invalid_signature_and_a_substituted_grant_never_advance_the_cursor()
-> Result<(), Box<dyn std::error::Error>> {
for substitution in [false, true] {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
let (key, mut frame) = signed_revoke(7, 1)?;
engine.service_key = if substitution {
key.public_key_bytes()
} else {
signed_revoke(8, 1)?.0.public_key_bytes()
};
if substitution && let ChangeFrame::Change { grant, .. } = &mut frame {
*grant = lys_identity::grants::GrantId::from_bytes([9; 16]).to_string();
}
engine.page(baseline(0))?;
assert_eq!(
engine
.page(ChangesPage {
log: baseline(0).log,
frames: vec![frame]
})
.err()
.ok_or("invalid evidence was accepted")?
.name(),
"grant_stream_evidence_invalid"
);
assert_eq!(state.snapshot()?.cursor, Some(0));
assert_ne!(state.snapshot()?.readiness, "ready");
}
Ok(())
}
#[test]
fn ready_without_the_placement_barrier_and_an_over_bound_baseline_are_refused()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
let mut incomplete = baseline(0);
incomplete.frames.remove(1);
assert_eq!(
engine
.page(incomplete)
.err()
.ok_or("missing placement barrier accepted")?
.name(),
"grant_stream_placement_barrier_missing"
);
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.config.max_revoked_grants = 1;
let mut page = baseline(2);
page.frames[0] = ChangeFrame::Baseline {
revision: 2,
revoked: vec![
lys_identity::grants::GrantId::from_bytes([1; 16]).to_string(),
lys_identity::grants::GrantId::from_bytes([2; 16]).to_string(),
],
};
assert_eq!(
engine
.page(page)
.err()
.ok_or("over-bound baseline accepted")?
.name(),
"grant_stream_revoked_over_bound"
);
assert_eq!(state.snapshot()?.cursor, None);
Ok(())
}
#[tokio::test]
async fn membership_completion_refuses_a_lost_or_replaced_stream_generation()
-> Result<(), Box<dyn std::error::Error>> {
use crate::auth_membership::{HttpMembershipProvider, MembershipGate, MembershipRefusal};
let state = StreamState::for_log(log());
let mapping = crate::auth_membership::fixture::config(
&log(),
("liminal.workspace", "liminal.channel"),
&[("fixture.channel", "fixture-workspace", "fixture-channel")],
&[(42, "fixture.channel")],
(16, 65536, 64),
);
let provider = HttpMembershipProvider::new(
"http://127.0.0.1/api/",
zeroize::Zeroizing::new("fixture-app-credential".into()),
tokio::runtime::Handle::current(),
65536,
state.cancellation(),
)?;
let gate = MembershipGate::new(&mapping, Arc::new(provider))?;
gate.install_authority(state.clone())
.map_err(|_| "authority already installed")?;
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(0))?;
assert_eq!(gate.stream_fence(1)?, 0);
engine.lost()?;
assert!(matches!(
gate.stream_fence(1),
Err(MembershipRefusal::Unavailable { .. })
));
engine.page(ChangesPage {
log: baseline(0).log,
frames: vec![
ChangeFrame::Watermark { revision: 1 },
ChangeFrame::PlacementMark {
revision: 1,
placement: 0,
},
ChangeFrame::Ready { revision: 1 },
],
})?;
assert_eq!(gate.stream_fence(1), Err(MembershipRefusal::ResumePending));
assert_eq!(gate.stream_fence(2)?, 1);
Ok(())
}
#[test]
fn a_terminal_reader_closes_the_revocation_fence_after_a_previous_ready()
-> Result<(), Box<dyn std::error::Error>> {
let state = StreamState::for_log(log());
let mut engine = engine::Engine::for_tests(Arc::clone(&state))?;
engine.page(baseline(0))?;
engine.terminal("grant_stream_no_progress")?;
assert_eq!(
engine.authority.diagnostics()?.readiness,
crate::auth_revocation::Readiness::Terminal
);
assert_eq!(state.snapshot()?.readiness, "terminal");
assert_eq!(
state.snapshot()?.last_error,
Some("grant_stream_no_progress")
);
Ok(())
}