use super::*;
pub struct SeqAllocator {
current: u64,
}
impl SeqAllocator {
pub async fn load(store: &Store) -> Self {
let current = match store.get_raw_bytes(SEQ_KEY).await {
Ok(Some(bytes)) if bytes.len() == 8 => {
u64::from_be_bytes(bytes[..8].try_into().unwrap_or([0; 8]))
}
_ => 0,
};
Self { current }
}
pub async fn next(&mut self, store: &Store) -> Result<u64> {
self.current += 1;
store.put_raw(SEQ_KEY, &self.current.to_be_bytes()).await?;
Ok(self.current)
}
pub fn current(&self) -> u64 {
self.current
}
}
pub async fn get_or_create_installation_id(store: &Store) -> Result<String> {
if let Ok(Some(bytes)) = store.get_raw_bytes(INSTALLATION_ID_KEY).await {
if let Ok(id) = std::str::from_utf8(&bytes) {
if !id.is_empty() {
return Ok(id.to_string());
}
}
}
let id = uuid::Uuid::new_v4().to_string();
store.put_raw(INSTALLATION_ID_KEY, id.as_bytes()).await?;
Ok(id)
}
pub fn get_local_actor() -> Option<ActorLocal> {
let username = std::env::var("USER")
.or_else(|_| std::env::var("USERNAME"))
.ok()?;
#[cfg(unix)]
let uid = Some(unsafe { libc::getuid() } as u32);
#[cfg(not(unix))]
let uid = None;
Some(ActorLocal {
username,
uid,
verified: false,
})
}
pub fn canonicalize_file_key(path: &str, repo_root: &Path) -> String {
let abs_path = if Path::new(path).is_relative() {
repo_root.join(path)
} else {
PathBuf::from(path)
};
let normalized = normalize_components(&abs_path);
let resolved = std::fs::canonicalize(&normalized).unwrap_or(normalized);
let repo_root_canonical =
std::fs::canonicalize(repo_root).unwrap_or_else(|_| repo_root.to_path_buf());
let relative = resolved
.strip_prefix(&repo_root_canonical)
.unwrap_or(&resolved);
let mut key = relative
.components()
.map(|c| c.as_os_str().to_string_lossy().to_string())
.collect::<Vec<_>>()
.join("/");
if is_case_insensitive() {
key = key.to_lowercase();
}
key
}
pub(crate) fn normalize_components(path: &Path) -> PathBuf {
let mut components = Vec::new();
for component in path.components() {
match component {
Component::CurDir => {} Component::ParentDir => {
if matches!(components.last(), Some(Component::Normal(_))) {
components.pop();
} else {
components.push(component);
}
}
_ => components.push(component),
}
}
components.iter().collect()
}
pub(crate) fn is_case_insensitive() -> bool {
cfg!(target_os = "macos") || cfg!(target_os = "windows")
}
pub fn canonical_subject_hash(canonical_key: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(canonical_key.as_bytes());
format!("{:x}", hasher.finalize())
}
fn uuid7_string() -> String {
uuid::Uuid::now_v7().to_string()
}
pub(crate) fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
pub struct EnforcementEventWriter {
seq: SeqAllocator,
installation_id: String,
prev_hash: String,
pub(super) agent_session: Option<String>,
pub(super) agent_id: Option<String>,
pub(super) parent_agent_id: Option<String>,
}
impl EnforcementEventWriter {
pub async fn new(store: &Store) -> Result<Self> {
let seq = SeqAllocator::load(store).await;
let installation_id = get_or_create_installation_id(store).await?;
let prev_hash = Self::load_last_hash(store).await;
Ok(Self {
seq,
installation_id,
prev_hash,
agent_session: None,
agent_id: None,
parent_agent_id: None,
})
}
async fn load_last_hash(store: &Store) -> String {
let keys = match store.scan_keys(EVENT_PREFIX).await {
Ok(k) => k,
Err(_) => return String::new(),
};
if keys.is_empty() {
return String::new();
}
let last_key = keys
.iter()
.max_by_key(|k| {
k.strip_prefix(EVENT_PREFIX)
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0)
})
.cloned();
if let Some(key) = last_key {
if let Ok(Some(bytes)) = store.get_raw_bytes(&key).await {
if let Ok(event) = serde_json::from_slice::<EnforcementEvent>(&bytes) {
return event.event_hash;
}
}
}
String::new()
}
#[allow(clippy::too_many_arguments)]
pub async fn write(
&mut self,
store: &Store,
event_type: EnforcementEventType,
subject_kind: SubjectKind,
subject_key: String,
agent_type: String,
receipt_id: Option<String>,
decision_reason_code: String,
decision_basis_hash: Option<String>,
) -> Result<EnforcementEvent> {
let seq_no = self.seq.next(store).await?;
let canonical_subject_hash_value = if subject_kind == SubjectKind::File {
Some(canonical_subject_hash(&subject_key))
} else {
None
};
let mut event = EnforcementEvent {
event_id: uuid7_string(),
schema_version: SCHEMA_VERSION,
seq_no,
recorded_at_ms: now_ms(),
event_type,
event_hash: String::new(), prev_hash: self.prev_hash.clone(),
installation_id: self.installation_id.clone(),
actor_local: get_local_actor(),
agent_type,
subject_kind,
subject_key,
canonical_subject_hash: canonical_subject_hash_value,
receipt_id,
decision_reason_code,
decision_basis_hash,
agent_session: self.agent_session.clone(),
agent_id: self.agent_id.clone(),
parent_agent_id: self.parent_agent_id.clone(),
};
event.event_hash = event.compute_hash();
let key = format!("{EVENT_PREFIX}{:020}", seq_no);
let json = serde_json::to_vec(&event)?;
store.put_raw(&key, &json).await?;
self.prev_hash = event.event_hash.clone();
self.agent_session = None;
self.agent_id = None;
self.parent_agent_id = None;
Ok(event)
}
pub fn installation_id(&self) -> &str {
&self.installation_id
}
pub fn current_seq(&self) -> u64 {
self.seq.current()
}
pub fn prev_hash(&self) -> &str {
&self.prev_hash
}
pub async fn detect_and_record_gap(
&mut self,
store: &Store,
gap_start_ms: u64,
gap_end_ms: u64,
cause: GapCause,
) -> Result<EnforcementEvent> {
let mode = get_enforcement_mode(store).await;
self.write(
store,
EnforcementEventType::RecordingGap {
gap_start_ms,
gap_end_ms,
cause,
enforcement_mode_during_gap: mode,
missed_event_count: MissedEventCount::Unknown,
certainty: GapCertainty::Inferred,
},
SubjectKind::System,
"enforcement:stream".to_string(),
"system".to_string(),
None,
"recording_gap_detected".to_string(),
None,
)
.await
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct EnforcementEventsWithSkips {
pub events: Vec<EnforcementEvent>,
pub skipped_seqs: Vec<u64>,
}
pub async fn scan_enforcement_events(
store: &Store,
since_seq: u64,
until_seq: u64,
) -> Result<Vec<EnforcementEvent>> {
Ok(
scan_enforcement_events_with_skips(store, since_seq, until_seq)
.await?
.events,
)
}
pub async fn scan_enforcement_events_with_skips(
store: &Store,
since_seq: u64,
until_seq: u64,
) -> Result<EnforcementEventsWithSkips> {
let keys = store.scan_keys(EVENT_PREFIX).await?;
let mut events = Vec::new();
let mut skipped_seqs = Vec::new();
let start = keys.partition_point(|key| {
key.strip_prefix(EVENT_PREFIX)
.and_then(|s| s.parse::<u64>().ok())
.map(|seq| seq < since_seq)
.unwrap_or(true)
});
for key in keys.iter().skip(start) {
let seq = match key
.strip_prefix(EVENT_PREFIX)
.and_then(|s| s.parse::<u64>().ok())
{
Some(s) => s,
None => continue,
};
if seq > until_seq {
break;
}
if seq < since_seq {
continue;
}
match store.get_raw_bytes(key).await {
Ok(Some(bytes)) => match serde_json::from_slice::<EnforcementEvent>(&bytes) {
Ok(event) => events.push(event),
Err(e) => {
tracing::warn!(key, "skipping corrupt enforcement event: {e}");
skipped_seqs.push(seq);
}
},
Ok(None) => {
tracing::warn!(key, "skipping enforcement event key with no value");
skipped_seqs.push(seq);
}
Err(e) => {
tracing::warn!(key, "skipping unreadable enforcement event: {e}");
skipped_seqs.push(seq);
}
}
}
events.sort_by_key(|e| e.seq_no);
skipped_seqs.sort_unstable();
Ok(EnforcementEventsWithSkips {
events,
skipped_seqs,
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EnforcementEventScan {
pub events: Vec<EnforcementEvent>,
pub oldest_recorded_at_ms: Option<u64>,
pub scanned_keys: usize,
}
pub async fn scan_enforcement_events_since_ms(
store: &Store,
since_ms: u64,
until_ms: u64,
) -> Result<EnforcementEventScan> {
let keys = store.scan_keys(EVENT_PREFIX).await?;
let valid_keys: Vec<&String> = keys
.iter()
.filter(|key| {
key.strip_prefix(EVENT_PREFIX)
.and_then(|s| s.parse::<u64>().ok())
.is_some()
})
.collect();
async fn read_event(store: &Store, key: &str) -> Option<EnforcementEvent> {
let bytes = store.get_raw_bytes(key).await.ok()??;
serde_json::from_slice(&bytes).ok()
}
let oldest_recorded_at_ms = match valid_keys.first() {
Some(key) => read_event(store, key)
.await
.map(|event| event.recorded_at_ms),
None => None,
};
let mut low = 0;
let mut high = valid_keys.len();
while low < high {
let mid = low + (high - low) / 2;
match read_event(store, valid_keys[mid]).await {
Some(event) if event.recorded_at_ms < since_ms => low = mid + 1,
Some(_) => high = mid,
None => low = mid + 1,
}
}
let mut events = Vec::new();
let mut scanned_keys = 0;
for key in valid_keys.into_iter().skip(low) {
scanned_keys += 1;
let Some(event) = read_event(store, key).await else {
continue;
};
if event.recorded_at_ms > until_ms {
break;
}
if event.recorded_at_ms >= since_ms {
events.push(event);
}
}
events.sort_by_key(|event| event.seq_no);
Ok(EnforcementEventScan {
events,
oldest_recorded_at_ms,
scanned_keys,
})
}