coven_protocol/
provider.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::fmt;
3
4use async_trait::async_trait;
5use serde::{Deserialize, Deserializer, Serialize, Serializer};
6use sha2::{Digest, Sha256};
7
8use crate::membership::{MembershipCoord, MembershipEntry, MembershipGrantId, OwnerStreamBarrier};
9use crate::objects::ObjectSlot;
10use crate::objects::{ExactObjectRef, ProviderDeviceBinding, StorageError, StoreProviderBinding};
11use crate::store_commit::{
12    DeviceJoinAttemptId, ObjectHash, StoreBatchCommitRef, StoreDeviceRegistration,
13    StoreDeviceRegistrationRef, StoreRootRef,
14};
15
16const EXACT_TRANSCRIPT_DOMAIN: &[u8] = b"coven.provider-exact-slot-probe.v1\0";
17const CROSS_TRANSCRIPT_DOMAIN: &[u8] = b"coven.provider-cross-principal-probe.v1\0";
18const CROSS_CHALLENGE_DOMAIN: &[u8] = b"coven.provider-cross-principal-challenge.v1\0";
19const CROSS_RESPONSE_DOMAIN: &[u8] = b"coven.provider-cross-principal-response.v1\0";
20const PAYLOAD_DOMAIN: &[u8] = b"coven.provider-probe-payload.v1\0";
21const MEMBER_ACCESS_GRANT_DOMAIN: &[u8] = b"coven.provider-member-access-grant.v1\0";
22pub(crate) const PROBE_PAYLOAD_LEN: usize = 256;
23pub const PROBE_RANGE_START: u64 = 31;
24pub const PROBE_RANGE_END: u64 = 173;
25
26mod access;
27mod admin;
28mod cross_principal;
29mod probe;
30
31pub use access::*;
32pub use admin::*;
33pub use cross_principal::*;
34pub use probe::*;
35
36pub fn canonical_custom_s3_origin(input: &str) -> Result<String, StorageError> {
37    if input.ends_with('/') {
38        return Err(StorageError::Configuration(
39            "custom S3 endpoint must not have a trailing slash".to_string(),
40        ));
41    }
42    let parsed = url::Url::parse(input)?;
43    if !matches!(parsed.scheme(), "http" | "https")
44        || !parsed.username().is_empty()
45        || parsed.password().is_some()
46        || parsed.query().is_some()
47        || parsed.fragment().is_some()
48        || parsed.path() != "/"
49    {
50        return Err(StorageError::Configuration(
51            "custom S3 endpoint must be an HTTP origin without user info, path, query, or fragment"
52                .to_string(),
53        ));
54    }
55    let host = parsed
56        .host_str()
57        .ok_or_else(|| StorageError::Configuration("custom S3 endpoint has no host".to_string()))?;
58    let port = parsed.port();
59    let default_port = matches!(
60        (parsed.scheme(), port),
61        ("http", Some(80)) | ("https", Some(443))
62    );
63    Ok(if let Some(port) = port.filter(|_| !default_port) {
64        format!("{}://{}:{port}", parsed.scheme(), host.to_ascii_lowercase())
65    } else {
66        format!("{}://{}", parsed.scheme(), host.to_ascii_lowercase())
67    })
68}
69
70pub async fn advance_cross_completion(
71    journal: &dyn ProviderProbeJournal,
72    durable: &mut ProviderProbeJournalRecord,
73    record: &mut CrossPrincipalCompletionJournal,
74    progress: CrossPrincipalCompletionProgress,
75) -> Result<(), ProviderProbeError> {
76    record.progress = progress;
77    let next = ProviderProbeJournalRecord::CrossPrincipal(record.clone());
78    journal.advance(durable, next.clone()).await?;
79    *durable = next;
80    Ok(())
81}
82
83pub async fn advance_exact(
84    journal: &dyn ProviderProbeJournal,
85    durable: &mut ProviderProbeJournalRecord,
86    record: &mut ExactProbeJournal,
87    progress: ExactProbeProgress,
88) -> Result<(), ProviderProbeError> {
89    record.progress = progress;
90    let next = ProviderProbeJournalRecord::Exact(record.clone());
91    journal.advance(durable, next.clone()).await?;
92    *durable = next;
93    Ok(())
94}
95
96mod ordered_owner_barriers {
97    use super::*;
98
99    pub(super) fn serialize<S>(
100        map: &BTreeMap<MembershipGrantId, OwnerStreamBarrier>,
101        serializer: S,
102    ) -> Result<S::Ok, S::Error>
103    where
104        S: Serializer,
105    {
106        map.iter().collect::<Vec<_>>().serialize(serializer)
107    }
108
109    pub(super) fn deserialize<'de, D>(
110        deserializer: D,
111    ) -> Result<BTreeMap<MembershipGrantId, OwnerStreamBarrier>, D::Error>
112    where
113        D: Deserializer<'de>,
114    {
115        let entries = Vec::<(MembershipGrantId, OwnerStreamBarrier)>::deserialize(deserializer)?;
116        let count = entries.len();
117        let map = entries.into_iter().collect::<BTreeMap<_, _>>();
118        if map.len() != count {
119            return Err(serde::de::Error::custom(
120                "provider administrator owner barriers contain a duplicate grant",
121            ));
122        }
123        Ok(map)
124    }
125}
126
127#[cfg(any(test, feature = "test-utils"))]
128pub mod test_fixtures;
129#[cfg(test)]
130mod tests;
coven_protocol/provider.rs

coven_protocol/
provider.rs
