inbuxa/inbuxa-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 23 Wiki Activity

Merge pull request 'Evaluate the personal-data catalog: the data inventory and its history' (#89) from feature/data-inventory into main
ci / fork-checks (push) Successful in 36s
Details
ci / build (push) Successful in 43m2s
Details

Browse Source
This commit was merged in pull request #89.
This commit is contained in:
jcoffey-dev
2026-09-28 17:15:10 +00:00
parent c61497e2b2 a8fb10458b
commit 7bbbff0648
29 changed files with 1810 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+31
View File
@@ -3956,6 +3956,7 @@ dependencies = [
"sha2 0.11.0",
"store",
"tokio",
"toml",
"trc",
"types",
"utils",
@@ -7642,6 +7643,15 @@ dependencies = [
"syn 3.0.6",
]
[[package]]
name = "serde_spanned"
version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26"
dependencies = [
"serde_core",
]
[[package]]
name = "serde_urlencoded"
version = "0.7.1"
@@ -8883,6 +8893,21 @@ dependencies = [
"webpki-roots 0.26.11",
]
[[package]]
name = "toml"
version = "1.1.6+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "920602543f0911ab71da12c50d59701da54c196d1a2bf5cb4b75667f137a406a"
dependencies = [
"indexmap 2.14.2",
"serde_core",
"serde_spanned",
"toml_datetime",
"toml_parser",
"toml_writer",
"winnow",
]
[[package]]
name = "toml_datetime"
version = "1.1.1+spec-1.1.0"
@@ -8913,6 +8938,12 @@ dependencies = [
"winnow",
]
[[package]]
name = "toml_writer"
version = "1.1.2+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
[[package]]
name = "tonic"
version = "0.14.6"
crates/common/src/lib.rs
+1
View File
@@ -69,6 +69,7 @@ pub mod auth;
pub mod cache;
pub mod audit; // inbuxa: the audit log (audit-hold-lock spec, AU)
pub mod hold; // inbuxa: legal holds (audit-hold-lock spec, LH)
pub mod privacy; // inbuxa: the personal-data catalog, evaluated
pub mod config;
pub mod expr;
pub mod i18n;
crates/common/src/privacy.rs
+406
View File
@@ -0,0 +1,406 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! The live facts the personal-data catalog is evaluated against
//! (personal-data catalog spec, §6): which sources are switched on, what
//! bounds each one's retention, which stores and endpoints are elsewhere.
//! Read from the registry on each request, so every node answers alike.
use crate::Server;
use inbuxa_features::privacy::{
self, Days, Inventory, LiveFacts, is_loopback,
snapshot::{self, Snapshot, Trigger},
};
use registry::schema::{
prelude::Object,
structs::{
AiModel, BlobStore, DataRetention, DataStore, InMemoryStore, Jmap, MtaHook, MtaMilter,
MtaRoute, Search, SearchStore, SpamClassifier, SpamClassifierModel, SpamDnsblServer,
SpamLlm, SpamPyzor, Tracer, TracingStore, WebHook,
},
};
use registry::types::duration::Duration;
use serde_json::Value;
use types::id::Id;
/// The objects [`Server::privacy_facts`] reads: a write to one may change
/// the inventory.
pub const INVENTORY_OBJECTS: &[&str] = &[
"x:DataRetention",
"x:SpamClassifier",
"x:Jmap",
"x:TracingStore",
"x:Search",
"x:Tracer",
"x:WebHook",
"x:AiModel",
"x:SpamLlm",
"x:SpamDnsblServer",
"x:SpamPyzor",
"x:MtaMilter",
"x:MtaHook",
"x:MtaRoute",
"x:DataStore",
"x:BlobStore",
"x:SearchStore",
"x:InMemoryStore",
"inbuxa:AuditSettings",
"inbuxa:LogSettings",
"inbuxa:AiLimits",
];
/// A store or endpoint object's type and host, from its JSON: local types
/// stay on the host.
fn remote_host(value: &Value) -> Option<String> {
let kind = value.get("@type").and_then(Value::as_str).unwrap_or_default();
if matches!(kind, "" | "RocksDb" | "Sqlite" | "FileSystem" | "Default" | "Disabled") {
return None;
}
for key in ["host", "url", "endpoint", "address", "hostname"] {
if let Some(host) = value.get(key).and_then(Value::as_str).filter(|h| !h.is_empty()) {
return Some(host.to_string());
}
}
// A list of URLs, as an array or as a map keyed by URL
match value.get("urls") {
Some(Value::Array(urls)) => {
if let Some(url) = urls.first().and_then(Value::as_str) {
return Some(url.to_string());
}
}
Some(Value::Object(urls)) => {
if let Some(url) = urls.keys().next() {
return Some(url.clone());
}
}
_ => {}
}
Some(kind.to_string())
}
fn days(duration: Option<&Duration>) -> Days {
match duration {
Some(d) => Days::Days(d.into_inner().as_secs().div_ceil(86_400)),
None => Days::Unbounded,
}
}
/// An expression's text, if it is a plain constant (a zone, a switch).
fn expression_text(value: &Value) -> Option<String> {
match value {
Value::String(s) => Some(s.clone()),
Value::Object(o) => o.get("else").and_then(|v| v.as_str()).map(str::to_string),
_ => None,
}
}
impl Server {
async fn singleton<T: registry::types::ObjectImpl + From<Object> + Default>(&self) -> trc::Result<T> {
Ok(self.registry().object::<T>(Id::singleton()).await?.unwrap_or_default())
}
/// The facts the catalog is evaluated against, from the live settings.
pub async fn privacy_facts(&self) -> trc::Result<LiveFacts> {
let mut facts = LiveFacts::default();
let data = &self.core.storage.data;
let endpoint = |facts: &mut LiveFacts, id: &str, url: String| {
if !url.is_empty() && !is_loopback(&url) {
facts.endpoints.entry(id.to_string()).or_default().push(url);
}
};
// Retention
let retention = self.singleton::<DataRetention>().await?;
for (name, value) in [
("x:DataRetention.holdTracesFor", &retention.hold_traces_for),
("x:DataRetention.holdMetricsFor", &retention.hold_metrics_for),
("x:DataRetention.holdMtaReportsFor", &retention.hold_mta_reports_for),
("x:DataRetention.archiveDeletedItemsFor", &retention.archive_deleted_items_for),
("x:DataRetention.archiveDeletedAccountsFor", &retention.archive_deleted_accounts_for),
("x:DataRetention.expungeTrashAfter", &retention.expunge_trash_after),
("x:DataRetention.expungeSubmissionsAfter", &retention.expunge_submissions_after),
] {
facts.durations.insert(name.into(), days(value.as_ref()));
}
let classifier = self.singleton::<SpamClassifier>().await?;
facts.durations.insert(
"x:SpamClassifier.holdSamplesFor".into(),
days(Some(&classifier.hold_samples_for)),
);
let jmap = self.singleton::<Jmap>().await?;
facts
.durations
.insert("x:Jmap.uploadTtl".into(), days(Some(&jmap.upload_ttl)));
let audit = inbuxa_features::audit::log::settings(data).await?;
facts.durations.insert(
"inbuxa:AuditSettings.keepForDays".into(),
Days::Days(audit.keep_for_secs.div_ceil(86_400)),
);
let logs = inbuxa_features::security::log_files::get(data).await?;
facts.durations.insert(
"inbuxa:LogSettings.keepForDays".into(),
logs.keep_for_days.map_or(Days::Unbounded, Days::Days),
);
// What's switched on
let tracing = self.singleton::<TracingStore>().await?;
let tracing_on = !matches!(tracing, TracingStore::Disabled);
let search = self.singleton::<Search>().await?;
for id in ["x:Trace", "x:TraceEvent", "x:TraceKeyValue", "x:TraceValueIpAddr", "x:TraceValueString"] {
facts.collected.insert(id.into(), tracing_on);
}
facts
.collected
.insert("trace-index".into(), tracing_on && search.index_telemetry);
facts.collected.insert(
"full-text-index".into(),
search.index_email || search.index_calendar || search.index_contacts,
);
let archive_on = retention.archive_deleted_items_for.is_some();
for id in [
"x:ArchivedEmail",
"x:ArchivedFileNode",
"x:ArchivedCalendarEvent",
"x:ArchivedContactCard",
"x:ArchivedSieveScript",
] {
facts.collected.insert(id.into(), archive_on);
}
facts.collected.insert(
"inbuxa:DeletedAccount".into(),
retention.archive_deleted_accounts_for.is_some(),
);
let reports_on = retention.hold_mta_reports_for.is_some();
for id in [
"x:ArfExternalReport",
"x:ArfFeedbackReport",
"x:DmarcExternalReport",
"x:DmarcReport",
"x:DmarcReportRecord",
"x:TlsExternalReport",
"x:TlsReport",
"x:TlsFailureDetails",
] {
facts.collected.insert(id.into(), reports_on);
}
let classifier_on = !matches!(classifier.model, SpamClassifierModel::Disabled);
facts
.collected
.insert("x:SpamTrainingSample".into(), classifier_on);
facts
.collected
.insert("spam-trainer-state".into(), classifier_on);
// Tracers
let (mut log_on, mut console_on, mut otel_on) = (false, false, false);
for tracer in self.registry().list::<Tracer>().await? {
match tracer.object {
Tracer::Log(t) => log_on |= t.enable,
Tracer::Stdout(t) => console_on |= t.enable,
Tracer::Journal(t) => console_on |= t.enable,
Tracer::OtelHttp(t) if t.enable => {
otel_on = true;
endpoint(&mut facts, "otel-tracer", t.endpoint);
}
Tracer::OtelGrpc(t) if t.enable => {
otel_on = true;
endpoint(&mut facts, "otel-tracer", t.endpoint.unwrap_or_default());
}
_ => {}
}
}
facts.collected.insert("log-file".into(), log_on);
facts.collected.insert("x:Log".into(), log_on);
facts.collected.insert("console-and-journal".into(), console_on);
facts.collected.insert("otel-tracer".into(), otel_on);
// Webhooks
let mut hooks_on = false;
for hook in self.registry().list::<WebHook>().await? {
if hook.object.enable {
hooks_on = true;
endpoint(&mut facts, "webhooks", hook.object.url);
}
}
facts.collected.insert("webhooks".into(), hooks_on);
// AI: the classifier's model, and Explain's
let models = self.registry().list::<AiModel>().await?;
let model_url = |id: Id| {
models
.iter()
.find(|m| Id::from(m.id.id()) == id)
.map(|m| m.object.url.clone())
};
let llm_on = match self.singleton::<SpamLlm>().await? {
SpamLlm::Enable(props) => {
if let Some(url) = model_url(props.model_id) {
endpoint(&mut facts, "spam-llm", url);
}
true
}
SpamLlm::Disable => false,
};
facts.collected.insert("spam-llm".into(), llm_on);
let limits = self.ai_limits().await;
let explain = self.ai_explain_model(&limits).await;
if let Some((_, model)) = &explain {
endpoint(&mut facts, "inbuxa:Explanation", model.url.clone());
}
facts
.collected
.insert("explain-cache".into(), explain.is_some());
facts
.collected
.insert("inbuxa:Explanation".into(), explain.is_some());
// Spam lookups off the host
let mut dnsbl_on = false;
for server in self.registry().list::<SpamDnsblServer>().await? {
let value = serde_json::to_value(&server.object).unwrap_or_default();
if value.get("enable").and_then(Value::as_bool).unwrap_or(false) {
dnsbl_on = true;
if let Some(zone) = value.get("zone").and_then(expression_text) {
endpoint(&mut facts, "spam-dnsbl", zone);
}
}
}
facts.collected.insert("spam-dnsbl".into(), dnsbl_on);
let pyzor = self.singleton::<SpamPyzor>().await?;
if pyzor.enable {
endpoint(&mut facts, "spam-pyzor", format!("{}:{}", pyzor.host, pyzor.port));
}
facts.collected.insert("spam-pyzor".into(), pyzor.enable);
// Mail handed to others
let mut hooks = false;
for milter in self.registry().list::<MtaMilter>().await? {
hooks = true;
endpoint(
&mut facts,
"mta-milter-and-hooks",
format!("{}:{}", milter.object.hostname, milter.object.port),
);
}
for hook in self.registry().list::<MtaHook>().await? {
hooks = true;
endpoint(&mut facts, "mta-milter-and-hooks", hook.object.url);
}
facts.collected.insert("mta-milter-and-hooks".into(), hooks);
let mut relays = false;
for route in self.registry().list::<MtaRoute>().await? {
if let MtaRoute::Relay(relay) = route.object {
relays = true;
endpoint(&mut facts, "relay", format!("{}:{}", relay.address, relay.port));
}
}
facts.collected.insert("relay".into(), relays);
// Stores elsewhere
let stores = [
("data-store", serde_json::to_value(self.singleton::<DataStore>().await.ok()).unwrap_or_default()),
("blob-store", serde_json::to_value(self.singleton::<BlobStore>().await?).unwrap_or_default()),
("search-store", serde_json::to_value(self.singleton::<SearchStore>().await?).unwrap_or_default()),
("in-memory-store", serde_json::to_value(self.singleton::<InMemoryStore>().await?).unwrap_or_default()),
];
for (place, value) in stores {
if let Some(host) = remote_host(&value) {
facts.remote_stores.insert(place.into(), host);
}
}
if let Some(host) = remote_host(&serde_json::to_value(&tracing).unwrap_or_default()) {
for id in ["x:Trace", "x:TraceEvent", "x:TraceKeyValue", "x:TraceValueIpAddr", "x:TraceValueString"] {
endpoint(&mut facts, id, host.clone());
}
}
Ok(facts)
}
/// The server's inventory, or a tenant's slice of it.
pub async fn data_inventory(&self, tenant_only: bool) -> trc::Result<Inventory> {
let facts = self.privacy_facts().await?;
Ok(privacy::evaluate(privacy::catalog(), &facts, tenant_only))
}
/// Records a snapshot of the server's inventory if it differs from the
/// newest one, or if there is none: the history shows when what the
/// server holds changed, not a copy a day. Returns whether it recorded.
pub async fn inventory_snapshot(&self, trigger: Trigger) -> trc::Result<bool> {
let data = &self.core.storage.data;
let inventory = self.data_inventory(false).await?;
if let Some(latest) = snapshot::latest(data).await?
&& let Some(previous) = snapshot::get(data, latest).await?
&& previous.inventory == inventory
{
return Ok(false);
}
snapshot::record(
data,
&Snapshot {
taken_at: store::write::now(),
trigger,
summary: inventory.summary(),
inventory,
},
)
.await?;
Ok(true)
}
/// A snapshot after a registry write, when the object is one the
/// inventory reads. Failures are logged: a snapshot is history, not
/// worth failing the write over.
pub async fn inventory_snapshot_after(&self, object: &str) {
if !INVENTORY_OBJECTS.contains(&object) {
return;
}
if let Err(err) = self
.inventory_snapshot(Trigger::SettingChanged {
setting: object.to_string(),
})
.await
{
trc::error!(err.details("Failed to record an inventory snapshot"));
}
}
/// Removes snapshots past the audit log's retention (settled
/// 2026-09-28: snapshots are kept as long as audit records).
pub async fn purge_inventory_snapshots(&self) -> trc::Result<usize> {
let data = &self.core.storage.data;
let keep = inbuxa_features::audit::log::settings(data).await?.keep_for_secs;
snapshot::purge(data, store::write::now().saturating_sub(keep)).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn local_stores_stay_and_others_name_their_host() {
assert_eq!(remote_host(&json!({"@type": "RocksDb", "path": "/var/lib"})), None);
assert_eq!(remote_host(&json!({"@type": "Default"})), None);
assert_eq!(
remote_host(&json!({"@type": "PostgreSql", "host": "db.example.net"})),
Some("db.example.net".into())
);
assert_eq!(
remote_host(&json!({"@type": "ElasticSearch", "url": "https://es.example.net:9200"})),
Some("https://es.example.net:9200".into())
);
assert_eq!(remote_host(&json!({"@type": "S3", "bucket": "mail"})), Some("S3".into()));
}
#[test]
fn days_round_up() {
assert_eq!(days(Some(&Duration::from_millis(86_400_000))), Days::Days(1));
assert_eq!(days(Some(&Duration::from_millis(3_600_000))), Days::Days(1));
assert_eq!(days(None), Days::Unbounded);
}
}
crates/features/Cargo.toml
+1
View File
@@ -15,6 +15,7 @@ utils = { path = "../utils" }
ahash = { version = "0.8.12", features = ["serde"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
toml = "1.1"
xxhash-rust = { version = "0.8.18", features = ["xxh3"] }
base64 = "0.23"
sha2 = "0.11"
crates/features/src/lib.rs
+1
View File
@@ -24,6 +24,7 @@ pub mod branding;
pub mod hold;
pub mod lock;
pub mod masked_email;
pub mod privacy;
pub mod security;
pub mod tenancy;
pub mod undelete;
crates/features/src/privacy/mod.rs
+486
View File
@@ -0,0 +1,486 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! The personal-data catalog, evaluated (personal-data catalog spec, §6).
//!
//! `resources/privacy/catalog.toml` says what the server *can* hold; this
//! module turns it into what *this* server holds, given the live facts the
//! caller gathers from its settings ([`LiveFacts`]). Facts in, facts out:
//! nothing here judges, and nothing here reads the store, so every
//! configuration can be tested with made-up facts.
pub mod snapshot;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::sync::OnceLock;
/// The catalog, as shipped with this build.
pub const CATALOG: &str = include_str!("../../../../resources/privacy/catalog.toml");
#[derive(Debug, Clone, Deserialize)]
#[serde(untagged)]
pub enum Retention {
Word(String),
Setting { setting: String },
}
#[derive(Debug, Clone, Default, Deserialize)]
pub struct ObjectEntry {
#[serde(default)]
pub whose: Vec<String>,
#[serde(default, rename = "where")]
pub location: Vec<String>,
pub scope: Option<String>,
pub retention: Option<Retention>,
#[serde(default)]
pub properties: BTreeMap<String, Vec<String>>,
}
#[derive(Debug, Clone, Default, Deserialize)]
pub struct SourceEntry {
#[serde(default)]
pub categories: Vec<String>,
#[serde(default)]
pub whose: Vec<String>,
#[serde(default, rename = "where")]
pub location: Vec<String>,
pub scope: Option<String>,
pub retention: Option<Retention>,
#[serde(default)]
pub enabled_by: Vec<String>,
#[serde(default)]
pub captures: Vec<String>,
#[serde(default)]
pub leaves_host: bool,
}
#[derive(Debug, Clone, Default, Deserialize)]
pub struct Catalog {
#[serde(default)]
pub object: BTreeMap<String, ObjectEntry>,
#[serde(default)]
pub source: BTreeMap<String, SourceEntry>,
}
/// The shipped catalog, parsed once. It is checked in CI
/// (`tools/fork/privacy-check.py`), so a parse failure is a build bug.
pub fn catalog() -> &'static Catalog {
static CATALOG_PARSED: OnceLock<Catalog> = OnceLock::new();
CATALOG_PARSED.get_or_init(|| toml::from_str(CATALOG).expect("resources/privacy/catalog.toml parses"))
}
/// A duration setting's live value.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Days {
/// Set, in whole days (rounded up).
Days(u64),
/// Unset: nothing bounds it.
Unbounded,
}
/// Everything the evaluation needs from the running server.
#[derive(Debug, Clone, Default)]
pub struct LiveFacts {
/// Duration settings by name (`x:DataRetention.holdTracesFor`,
/// `inbuxa:AuditSettings.keepForDays` ...). A setting not here is
/// reported by name, without a value.
pub durations: BTreeMap<String, Days>,
/// Whether each source or object is collected at all, by catalog id. An
/// id not here is taken as collected.
pub collected: BTreeMap<String, bool>,
/// The endpoints each source or object sends to, by catalog id: hosts or
/// URLs as configured. Loopback endpoints are left out: what goes there
/// stays on the host ([`is_loopback`]).
pub endpoints: BTreeMap<String, Vec<String>>,
/// Stores pointed at a remote backend, by location (`data-store`,
/// `blob-store`, `search-store`, `in-memory-store`), with the host.
pub remote_stores: BTreeMap<String, String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RetentionOut {
/// `unbounded`, `days`, `object-life`, `receiver`, or `setting` (named but
/// not evaluated).
pub kind: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub days: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub setting: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Item {
pub id: String,
/// `object` or `source`.
pub kind: String,
pub categories: Vec<String>,
pub whose: Vec<String>,
#[serde(rename = "where")]
pub location: Vec<String>,
pub scope: String,
pub collected: bool,
pub retention: RetentionOut,
pub leaves_host: bool,
pub controlled_by: Vec<String>,
pub endpoints: Vec<String>,
}
/// A host that receives personal data: a candidate processor, since whether
/// it is one in law is the operator's determination.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Processor {
pub host: String,
pub receives: Vec<String>,
pub sources: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Inventory {
pub items: Vec<Item>,
pub processors: Vec<Processor>,
}
/// Counts for a snapshot's summary and the Overview.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Summary {
pub collected: u64,
pub unbounded: u64,
pub leaving_host: u64,
pub processors: u64,
}
impl Inventory {
pub fn summary(&self) -> Summary {
let collected = self.items.iter().filter(|i| i.collected);
Summary {
collected: collected.clone().count() as u64,
unbounded: collected
.clone()
.filter(|i| i.retention.kind == "unbounded")
.count() as u64,
leaving_host: collected.filter(|i| i.leaves_host).count() as u64,
processors: self.processors.len() as u64,
}
}
}
/// The host part of an endpoint as configured: a URL's host, or the string
/// itself when it is a bare host or zone.
pub fn host_of(endpoint: &str) -> String {
let rest = endpoint.split_once("://").map_or(endpoint, |(_, rest)| rest);
let rest = rest.rsplit_once('@').map_or(rest, |(_, host)| host);
let host = rest.split(['/', '?', '#']).next().unwrap_or(rest);
let host = if host.starts_with('[') {
host.split_once(']').map_or(host, |(h, _)| h.trim_start_matches('['))
} else {
host.rsplit_once(':')
.filter(|(_, port)| port.chars().all(|c| c.is_ascii_digit()))
.map_or(host, |(h, _)| h)
};
host.trim().trim_end_matches('.').to_ascii_lowercase()
}
/// Whether an endpoint is this host: what is sent there stays here.
pub fn is_loopback(endpoint: &str) -> bool {
let host = host_of(endpoint);
host == "localhost"
|| host.ends_with(".localhost")
|| host == "::1"
|| host.parse::<std::net::IpAddr>().is_ok_and(|ip| ip.is_loopback())
}
fn retention_out(retention: Option<&Retention>, facts: &LiveFacts) -> RetentionOut {
match retention {
Some(Retention::Setting { setting }) => match facts.durations.get(setting) {
Some(Days::Days(days)) => RetentionOut {
kind: "days".into(),
days: Some(*days),
setting: Some(setting.clone()),
},
Some(Days::Unbounded) => RetentionOut {
kind: "unbounded".into(),
days: None,
setting: Some(setting.clone()),
},
None => RetentionOut {
kind: "setting".into(),
days: None,
setting: Some(setting.clone()),
},
},
Some(Retention::Word(word)) => RetentionOut {
kind: word.clone(),
days: None,
setting: None,
},
None => RetentionOut {
kind: "object-life".into(),
days: None,
setting: None,
},
}
}
/// What the catalog says `id` holds, evaluated against `facts`. Objects with
/// nothing personal are left out. `tenant_only` keeps the entries a tenant
/// can be told about: tenant-scoped, and none of the server's processors.
pub fn evaluate(catalog: &Catalog, facts: &LiveFacts, tenant_only: bool) -> Inventory {
let mut items = Vec::new();
let mut add = |id: &str,
kind: &str,
categories: Vec<String>,
whose: &[String],
location: &[String],
scope: Option<&String>,
retention: Option<&Retention>,
controlled_by: Vec<String>,
leaves: bool| {
let scope = scope.cloned().unwrap_or_else(|| "server".into());
if tenant_only && scope != "tenant" {
return;
}
let mut endpoints: Vec<String> = facts.endpoints.get(id).cloned().unwrap_or_default();
// Anything sent to an endpoint off this host leaves it
let mut leaves_host = leaves || !endpoints.is_empty();
for place in location {
if let Some(host) = facts.remote_stores.get(place) {
leaves_host = true;
endpoints.push(host.clone());
}
}
endpoints.sort();
endpoints.dedup();
items.push(Item {
id: id.to_string(),
kind: kind.to_string(),
categories,
whose: whose.to_vec(),
location: location.to_vec(),
scope,
collected: facts.collected.get(id).copied().unwrap_or(true),
retention: retention_out(retention, facts),
leaves_host,
controlled_by,
endpoints,
});
};
for (id, entry) in &catalog.source {
let controlled_by = entry
.enabled_by
.iter()
.chain(&entry.captures)
.cloned()
.collect();
add(
id,
"source",
entry.categories.clone(),
&entry.whose,
&entry.location,
entry.scope.as_ref(),
entry.retention.as_ref(),
controlled_by,
entry.leaves_host,
);
}
for (id, entry) in &catalog.object {
if entry.properties.is_empty() || entry.whose.is_empty() {
// Nothing personal, or a credential field of a configuration
// object with no place of its own in the inventory
continue;
}
let categories: BTreeSet<String> = entry.properties.values().flatten().cloned().collect();
let leaves = entry.location.iter().any(|place| place == "external");
add(
id,
"object",
categories.into_iter().collect(),
&entry.whose,
&entry.location,
entry.scope.as_ref(),
entry.retention.as_ref(),
Vec::new(),
leaves,
);
}
// Candidate processors: each host that receives something, once
let mut processors: BTreeMap<String, (BTreeSet<String>, BTreeSet<String>)> = BTreeMap::new();
if !tenant_only {
for item in items.iter().filter(|i| i.collected && i.leaves_host) {
for endpoint in &item.endpoints {
let entry = processors.entry(host_of(endpoint)).or_default();
entry.0.extend(item.categories.iter().cloned());
entry.1.insert(item.id.clone());
}
}
}
Inventory {
items,
processors: processors
.into_iter()
.filter(|(host, _)| !host.is_empty())
.map(|(host, (receives, sources))| Processor {
host,
receives: receives.into_iter().collect(),
sources: sources.into_iter().collect(),
})
.collect(),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn facts() -> LiveFacts {
LiveFacts::default()
}
fn item<'a>(inventory: &'a Inventory, id: &str) -> &'a Item {
inventory
.items
.iter()
.find(|i| i.id == id)
.unwrap_or_else(|| panic!("{id} not in the inventory"))
}
#[test]
fn the_shipped_catalog_parses() {
let catalog = catalog();
assert!(catalog.source.contains_key("log-file"));
assert!(catalog.object.contains_key("x:UserAccount"));
}
#[test]
fn defaults_a_new_install_would_report() {
let mut facts = facts();
facts
.durations
.insert("x:DataRetention.holdTracesFor".into(), Days::Days(14));
facts
.durations
.insert("inbuxa:LogSettings.keepForDays".into(), Days::Days(30));
facts.endpoints.insert("spam-pyzor".into(), vec!["public.pyzor.org:24441".into()]);
facts.collected.insert("spam-pyzor".into(), false);
facts.endpoints.insert(
"spam-dnsbl".into(),
vec!["zen.spamhaus.org".into(), "bl.spamcop.net".into()],
);
let inventory = evaluate(catalog(), &facts, false);
let trace = item(&inventory, "x:Trace");
assert_eq!(trace.retention.kind, "days");
assert_eq!(trace.retention.days, Some(14));
assert!(!trace.leaves_host);
assert_eq!(item(&inventory, "log-file").retention.days, Some(30));
// Pyzor off: listed, not collected, not a processor
assert!(!item(&inventory, "spam-pyzor").collected);
let hosts: Vec<_> = inventory.processors.iter().map(|p| p.host.as_str()).collect();
assert_eq!(hosts, vec!["bl.spamcop.net", "zen.spamhaus.org"]);
// Nothing personal isn't listed
assert!(inventory.items.iter().all(|i| i.id != "x:Http"));
}
#[test]
fn an_external_store_makes_what_lives_there_leave_the_host() {
let mut facts = facts();
facts
.remote_stores
.insert("blob-store".into(), "https://s3.example.net/mail".into());
let inventory = evaluate(catalog(), &facts, false);
let archived = item(&inventory, "x:ArchivedEmail");
assert!(archived.leaves_host);
assert_eq!(archived.endpoints, vec!["https://s3.example.net/mail"]);
assert!(inventory.processors.iter().any(|p| p.host == "s3.example.net"
&& p.sources.contains(&"x:ArchivedEmail".to_string())));
// What lives only in the data store stays
assert!(!item(&inventory, "x:UserAccount").leaves_host);
}
#[test]
fn a_hosted_ai_endpoint_is_a_processor_of_content() {
let mut facts = facts();
facts.collected.insert("spam-llm".into(), true);
facts
.endpoints
.insert("spam-llm".into(), vec!["https://api.example-ai.com/v1".into()]);
let inventory = evaluate(catalog(), &facts, false);
let ai = inventory
.processors
.iter()
.find(|p| p.host == "api.example-ai.com")
.expect("the AI endpoint is listed");
assert_eq!(ai.receives, vec!["content"]);
}
#[test]
fn telemetry_off_is_reported_as_not_collected() {
let mut facts = facts();
for id in ["x:Trace", "trace-index", "log-file"] {
facts.collected.insert(id.into(), false);
}
let inventory = evaluate(catalog(), &facts, false);
for id in ["x:Trace", "trace-index", "log-file"] {
assert!(!item(&inventory, id).collected, "{id}");
}
assert_eq!(item(&inventory, "log-file").retention.kind, "setting");
}
#[test]
fn a_tenant_sees_its_slice_and_no_processors() {
let mut facts = facts();
facts.endpoints.insert("spam-dnsbl".into(), vec!["zen.spamhaus.org".into()]);
let inventory = evaluate(catalog(), &facts, true);
assert!(inventory.items.iter().all(|i| i.scope == "tenant"));
assert!(inventory.items.iter().any(|i| i.id == "x:UserAccount"));
assert!(inventory.items.iter().all(|i| i.id != "log-file"));
assert!(inventory.processors.is_empty());
}
#[test]
fn hosts_are_read_from_urls_and_bare_names() {
assert_eq!(host_of("https://user:[email protected]:8443/path?x"), "hooks.example.com");
assert_eq!(host_of("public.pyzor.org:24441"), "public.pyzor.org");
assert_eq!(host_of("zen.spamhaus.org."), "zen.spamhaus.org");
assert_eq!(host_of("http://[::1]:11434/v1"), "::1");
assert_eq!(host_of("postgres://db.internal:5432/mail"), "db.internal");
}
#[test]
fn loopback_stays_on_the_host() {
assert!(is_loopback("http://127.0.0.1:11434/v1"));
assert!(is_loopback("http://localhost:8080"));
assert!(is_loopback("http://[::1]:11434"));
assert!(!is_loopback("http://10.77.0.2:11434"), "another node leaves the host");
assert!(!is_loopback("https://api.example-ai.com"));
}
#[test]
fn a_configured_endpoint_means_it_leaves() {
let mut facts = facts();
facts.endpoints.insert("x:Trace".into(), vec!["postgres://traces.example.net".into()]);
let inventory = evaluate(catalog(), &facts, false);
assert!(item(&inventory, "x:Trace").leaves_host);
}
#[test]
fn a_summary_counts_what_is_collected() {
let mut facts = facts();
facts.collected.insert("log-file".into(), true);
let inventory = evaluate(catalog(), &facts, false);
let summary = inventory.summary();
assert!(summary.collected > 10);
assert!(summary.unbounded >= 1, "sources the catalog marks unbounded");
}
}
crates/features/src/privacy/snapshot.rs
+155
View File
@@ -0,0 +1,155 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! Dated copies of the evaluated inventory (personal-data catalog spec, §6,
//! `inbuxa:InventorySnapshot`), so the Overview can show when and why what
//! the server holds changed. Stored as JSON under `C` `i` and the time taken
//! (seconds, big-endian) in the fork's subspace; kept as long as the audit
//! log keeps its records (settled 2026-09-28).
use super::{Inventory, Summary};
use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize};
use store::{
Deserialize, IterateParams, SUBSPACE_INBUXA, Store, U64_LEN, ValueKey,
write::{AnyClass, BatchBuilder, ValueClass, key::DeserializeBigEndian},
};
use trc::AddContext;
const PREFIX: &[u8] = b"Ci";
/// Why a snapshot was taken.
#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
#[serde(rename_all = "camelCase", tag = "kind")]
pub enum Trigger {
/// A setting the catalog names changed: the object type that changed.
SettingChanged { setting: String },
/// The daily snapshot.
Daily,
}
#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
#[serde(rename_all = "camelCase")]
pub struct Snapshot {
/// Seconds since the epoch; also the snapshot's id.
pub taken_at: u64,
pub trigger: Trigger,
pub summary: Summary,
pub inventory: Inventory,
}
fn key(taken_at: u64) -> Vec<u8> {
let mut key = PREFIX.to_vec();
key.extend_from_slice(&taken_at.to_be_bytes());
key
}
fn class(taken_at: u64) -> ValueClass {
ValueClass::Any(AnyClass {
subspace: SUBSPACE_INBUXA,
key: key(taken_at),
})
}
struct Json(Snapshot);
impl Deserialize for Json {
fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
serde_json::from_slice(bytes).map(Json).map_err(|err| {
trc::StoreEvent::DataCorruption
.caused_by(trc::location!())
.reason(err)
})
}
}
/// Stores a snapshot. Two in the same second: the later one wins.
pub async fn record(data: &Store, snapshot: &Snapshot) -> trc::Result<()> {
let bytes = serde_json::to_vec(snapshot).map_err(|err| {
trc::StoreEvent::UnexpectedError
.caused_by(trc::location!())
.reason(err)
})?;
let mut batch = BatchBuilder::new();
batch.set(class(snapshot.taken_at), bytes);
data.write(batch.build_all())
.await
.caused_by(trc::location!())
.map(|_| ())
}
/// One snapshot, by the time it was taken.
pub async fn get(data: &Store, taken_at: u64) -> trc::Result<Option<Snapshot>> {
Ok(data
.get_value::<Json>(ValueKey::from(class(taken_at)))
.await
.caused_by(trc::location!())?
.map(|Json(snapshot)| snapshot))
}
/// The times snapshots were taken between `after` and `before` (inclusive,
/// seconds), newest first.
pub async fn list(data: &Store, after: u64, before: u64) -> trc::Result<Vec<u64>> {
let mut times = Vec::new();
data.iterate(
IterateParams::new(
ValueKey::from(class(after)),
ValueKey::from(class(before)),
)
.no_values(),
|key, _| {
times.push(key.deserialize_be_u64(key.len() - U64_LEN)?);
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
times.reverse();
Ok(times)
}
/// The newest snapshot's time, if any.
pub async fn latest(data: &Store) -> trc::Result<Option<u64>> {
Ok(list(data, 0, u64::MAX).await?.first().copied())
}
/// Removes snapshots taken before `before` (seconds). Returns how many went.
pub async fn purge(data: &Store, before: u64) -> trc::Result<usize> {
let old = list(data, 0, before.saturating_sub(1)).await?;
if old.is_empty() {
return Ok(0);
}
let mut batch = BatchBuilder::new();
for taken_at in &old {
batch.clear(class(*taken_at));
}
data.write(batch.build_all())
.await
.caused_by(trc::location!())?;
Ok(old.len())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn keys_sort_by_time() {
assert!(key(1) < key(2));
assert!(key(255) < key(256));
assert_eq!(&key(7)[..2], PREFIX);
}
#[test]
fn a_trigger_reads_as_json_names_it() {
let changed = serde_json::to_value(Trigger::SettingChanged {
setting: "x:DataRetention".into(),
})
.unwrap();
assert_eq!(changed["kind"], "settingChanged");
assert_eq!(changed["setting"], "x:DataRetention");
assert_eq!(serde_json::to_value(Trigger::Daily).unwrap()["kind"], "daily");
}
}
crates/jmap-proto/src/object/inbuxa_data_inventory.rs
+165
View File
@@ -0,0 +1,165 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! `inbuxa:DataInventory/get` under `urn:inbuxa:jmap`: the personal-data
//! catalog evaluated against this server's live settings (personal-data
//! catalog spec, §6). A singleton, id `singleton`; read-only.
use crate::object::{AnyId, JmapObject, JmapObjectId};
use jmap_tools::{Element, Key, Property};
use std::{borrow::Cow, str::FromStr};
use types::id::Id;
#[derive(Debug, Clone, Default)]
pub struct DataInventory;
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum DataInventoryProperty {
Id,
EvaluatedAt,
CatalogVersion,
Summary,
Items,
Processors,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum DataInventoryValue {
Id(Id),
}
impl Property for DataInventoryProperty {
fn try_parse(_: Option<&Key<'_, Self>>, value: &str) -> Option<Self> {
DataInventoryProperty::parse(value)
}
fn to_cow(&self) -> Cow<'static, str> {
match self {
DataInventoryProperty::Id => "id",
DataInventoryProperty::EvaluatedAt => "evaluatedAt",
DataInventoryProperty::CatalogVersion => "catalogVersion",
DataInventoryProperty::Summary => "summary",
DataInventoryProperty::Items => "items",
DataInventoryProperty::Processors => "processors",
}
.into()
}
}
impl DataInventoryProperty {
fn parse(value: &str) -> Option<Self> {
hashify::tiny_map!(value.as_bytes(),
b"id" => DataInventoryProperty::Id,
b"evaluatedAt" => DataInventoryProperty::EvaluatedAt,
b"catalogVersion" => DataInventoryProperty::CatalogVersion,
b"summary" => DataInventoryProperty::Summary,
b"items" => DataInventoryProperty::Items,
b"processors" => DataInventoryProperty::Processors,
)
}
}
impl FromStr for DataInventoryProperty {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
DataInventoryProperty::parse(s).ok_or(())
}
}
impl Element for DataInventoryValue {
type Property = DataInventoryProperty;
fn try_parse<P>(key: &Key<'_, Self::Property>, value: &str) -> Option<Self> {
match key {
Key::Property(DataInventoryProperty::Id) => {
Id::from_str(value).ok().map(DataInventoryValue::Id)
}
_ => None,
}
}
fn to_cow(&self) -> Cow<'static, str> {
match self {
DataInventoryValue::Id(id) => id.to_string().into(),
}
}
}
impl JmapObject for DataInventory {
type Property = DataInventoryProperty;
type Element = DataInventoryValue;
type Id = Id;
type Filter = ();
type Comparator = ();
type GetArguments = ();
type SetArguments<'de> = ();
type QueryArguments = ();
type CopyArguments = ();
type ParseArguments = ();
const ID_PROPERTY: Self::Property = DataInventoryProperty::Id;
}
impl From<Id> for DataInventoryValue {
fn from(id: Id) -> Self {
DataInventoryValue::Id(id)
}
}
impl JmapObjectId for DataInventoryValue {
fn as_id(&self) -> Option<Id> {
match self {
DataInventoryValue::Id(id) => Some(*id),
}
}
fn as_any_id(&self) -> Option<AnyId> {
match self {
DataInventoryValue::Id(id) => Some(AnyId::Id(*id)),
}
}
fn as_id_ref(&self) -> Option<&str> {
None
}
fn try_set_id(&mut self, new_id: AnyId) -> bool {
if let AnyId::Id(id) = new_id {
*self = DataInventoryValue::Id(id);
true
} else {
false
}
}
}
impl JmapObjectId for DataInventoryProperty {
fn as_id(&self) -> Option<Id> {
None
}
fn as_any_id(&self) -> Option<AnyId> {
None
}
fn as_id_ref(&self) -> Option<&str> {
None
}
fn try_set_id(&mut self, _: AnyId) -> bool {
false
}
}
crates/jmap-proto/src/object/inbuxa_inventory_snapshot.rs
+162
View File
@@ -0,0 +1,162 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! `inbuxa:InventorySnapshot/get` under `urn:inbuxa:jmap`: dated copies of
//! the evaluated inventory (personal-data catalog spec, §6). The id is the
//! time taken; `ids: null` lists every snapshot kept, newest first.
use crate::object::{AnyId, JmapObject, JmapObjectId};
use jmap_tools::{Element, Key, Property};
use std::{borrow::Cow, str::FromStr};
use types::id::Id;
#[derive(Debug, Clone, Default)]
pub struct InventorySnapshot;
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum InventorySnapshotProperty {
Id,
TakenAt,
Trigger,
Summary,
Inventory,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum InventorySnapshotValue {
Id(Id),
}
impl Property for InventorySnapshotProperty {
fn try_parse(_: Option<&Key<'_, Self>>, value: &str) -> Option<Self> {
InventorySnapshotProperty::parse(value)
}
fn to_cow(&self) -> Cow<'static, str> {
match self {
InventorySnapshotProperty::Id => "id",
InventorySnapshotProperty::TakenAt => "takenAt",
InventorySnapshotProperty::Trigger => "trigger",
InventorySnapshotProperty::Summary => "summary",
InventorySnapshotProperty::Inventory => "inventory",
}
.into()
}
}
impl InventorySnapshotProperty {
fn parse(value: &str) -> Option<Self> {
hashify::tiny_map!(value.as_bytes(),
b"id" => InventorySnapshotProperty::Id,
b"takenAt" => InventorySnapshotProperty::TakenAt,
b"trigger" => InventorySnapshotProperty::Trigger,
b"summary" => InventorySnapshotProperty::Summary,
b"inventory" => InventorySnapshotProperty::Inventory,
)
}
}
impl FromStr for InventorySnapshotProperty {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
InventorySnapshotProperty::parse(s).ok_or(())
}
}
impl Element for InventorySnapshotValue {
type Property = InventorySnapshotProperty;
fn try_parse<P>(key: &Key<'_, Self::Property>, value: &str) -> Option<Self> {
match key {
Key::Property(InventorySnapshotProperty::Id) => {
Id::from_str(value).ok().map(InventorySnapshotValue::Id)
}
_ => None,
}
}
fn to_cow(&self) -> Cow<'static, str> {
match self {
InventorySnapshotValue::Id(id) => id.to_string().into(),
}
}
}
impl JmapObject for InventorySnapshot {
type Property = InventorySnapshotProperty;
type Element = InventorySnapshotValue;
type Id = Id;
type Filter = ();
type Comparator = ();
type GetArguments = ();
type SetArguments<'de> = ();
type QueryArguments = ();
type CopyArguments = ();
type ParseArguments = ();
const ID_PROPERTY: Self::Property = InventorySnapshotProperty::Id;
}
impl From<Id> for InventorySnapshotValue {
fn from(id: Id) -> Self {
InventorySnapshotValue::Id(id)
}
}
impl JmapObjectId for InventorySnapshotValue {
fn as_id(&self) -> Option<Id> {
match self {
InventorySnapshotValue::Id(id) => Some(*id),
}
}
fn as_any_id(&self) -> Option<AnyId> {
match self {
InventorySnapshotValue::Id(id) => Some(AnyId::Id(*id)),
}
}
fn as_id_ref(&self) -> Option<&str> {
None
}
fn try_set_id(&mut self, new_id: AnyId) -> bool {
if let AnyId::Id(id) = new_id {
*self = InventorySnapshotValue::Id(id);
true
} else {
false
}
}
}
impl JmapObjectId for InventorySnapshotProperty {
fn as_id(&self) -> Option<Id> {
None
}
fn as_any_id(&self) -> Option<AnyId> {
None
}
fn as_id_ref(&self) -> Option<&str> {
None
}
fn try_set_id(&mut self, _: AnyId) -> bool {
false
}
}
crates/jmap-proto/src/object/mod.rs
+2
View File
@@ -24,6 +24,8 @@ pub mod fastmail_masked_email; // inbuxa: masked email
pub mod inbuxa_account_lock; // inbuxa: account lock with delegation
pub mod inbuxa_ai_limits; // inbuxa: AI spam classification
pub mod inbuxa_log_settings; // inbuxa: personal-data catalog, D1
pub mod inbuxa_data_inventory; // inbuxa: personal-data catalog
pub mod inbuxa_inventory_snapshot; // inbuxa: personal-data catalog
pub mod inbuxa_audit; // inbuxa: the audit log
pub mod inbuxa_legal_hold; // inbuxa: legal hold
pub mod inbuxa_hold_export; // inbuxa: legal hold exports
crates/jmap-proto/src/references/eval.rs
+6
View File
@@ -64,6 +64,12 @@ impl Response<'_> {
GetResponseMethod::LogSettings(response) => {
response.eval_jptr(path, &mut results)
}
GetResponseMethod::DataInventory(response) => {
response.eval_jptr(path, &mut results)
}
GetResponseMethod::InventorySnapshot(response) => {
response.eval_jptr(path, &mut results)
}
GetResponseMethod::AuditEvent(response) => {
response.eval_jptr(path, &mut results)
}
crates/jmap-proto/src/references/resolve.rs
+2
View File
@@ -47,6 +47,8 @@ impl Response<'_> {
GetRequestMethod::DeletedAccount(request) => request.resolve_references(self)?,
GetRequestMethod::AiLimits(request) => request.resolve_references(self)?,
GetRequestMethod::LogSettings(request) => request.resolve_references(self)?,
GetRequestMethod::DataInventory(request) => request.resolve_references(self)?,
GetRequestMethod::InventorySnapshot(request) => request.resolve_references(self)?,
GetRequestMethod::AuditEvent(request) => request.resolve_references(self)?,
GetRequestMethod::AuditSettings(request) => request.resolve_references(self)?,
GetRequestMethod::AccountLock(request) => request.resolve_references(self)?,
crates/jmap-proto/src/request/method.rs
+10
View File
@@ -50,6 +50,8 @@ pub enum MethodObject {
// inbuxa: AI call limits
AiLimits,
LogSettings,
DataInventory,
InventorySnapshot,
// inbuxa: "Explain this" with the local model
Explanation,
// inbuxa: the audit log
@@ -91,6 +93,8 @@ impl MethodObject {
MethodObject::DeletedAccount => Capability::Inbuxa,
MethodObject::AiLimits => Capability::Inbuxa,
MethodObject::LogSettings => Capability::Inbuxa,
MethodObject::DataInventory => Capability::Inbuxa,
MethodObject::InventorySnapshot => Capability::Inbuxa,
MethodObject::Explanation => Capability::Inbuxa,
MethodObject::AuditEvent
| MethodObject::AuditSettings
@@ -279,6 +283,8 @@ impl MethodName {
(MethodFunction::Get, MethodObject::AiLimits) => "inbuxa:AiLimits/get",
(MethodFunction::Set, MethodObject::AiLimits) => "inbuxa:AiLimits/set",
(MethodFunction::Get, MethodObject::LogSettings) => "inbuxa:LogSettings/get",
(MethodFunction::Get, MethodObject::DataInventory) => "inbuxa:DataInventory/get",
(MethodFunction::Get, MethodObject::InventorySnapshot) => "inbuxa:InventorySnapshot/get",
(MethodFunction::Set, MethodObject::LogSettings) => "inbuxa:LogSettings/set",
(MethodFunction::Set, MethodObject::Explanation) => "inbuxa:Explanation/set",
(MethodFunction::Get, MethodObject::AuditEvent) => "inbuxa:AuditEvent/get",
@@ -429,6 +435,8 @@ impl MethodName {
"inbuxa:AiLimits/get" => (MethodObject::AiLimits, MethodFunction::Get),
"inbuxa:AiLimits/set" => (MethodObject::AiLimits, MethodFunction::Set),
"inbuxa:LogSettings/get" => (MethodObject::LogSettings, MethodFunction::Get),
"inbuxa:DataInventory/get" => (MethodObject::DataInventory, MethodFunction::Get),
"inbuxa:InventorySnapshot/get" => (MethodObject::InventorySnapshot, MethodFunction::Get),
"inbuxa:LogSettings/set" => (MethodObject::LogSettings, MethodFunction::Set),
"inbuxa:Explanation/set" => (MethodObject::Explanation, MethodFunction::Set),
"inbuxa:AuditEvent/get" => (MethodObject::AuditEvent, MethodFunction::Get),
@@ -501,6 +509,8 @@ impl Display for MethodObject {
MethodObject::DeletedAccount => "inbuxa:DeletedAccount",
MethodObject::AiLimits => "inbuxa:AiLimits",
MethodObject::LogSettings => "inbuxa:LogSettings",
MethodObject::DataInventory => "inbuxa:DataInventory",
MethodObject::InventorySnapshot => "inbuxa:InventorySnapshot",
MethodObject::Explanation => "inbuxa:Explanation",
MethodObject::AuditEvent => "inbuxa:AuditEvent",
MethodObject::AuditSettings => "inbuxa:AuditSettings",
crates/jmap-proto/src/request/mod.rs
+2
View File
@@ -117,6 +117,8 @@ pub enum GetRequestMethod {
DeletedAccount(Box<GetRequest<crate::object::inbuxa_deleted_account::DeletedAccount>>),
AiLimits(Box<GetRequest<crate::object::inbuxa_ai_limits::AiLimits>>),
LogSettings(Box<GetRequest<crate::object::inbuxa_log_settings::LogSettings>>),
DataInventory(Box<GetRequest<crate::object::inbuxa_data_inventory::DataInventory>>),
InventorySnapshot(Box<GetRequest<crate::object::inbuxa_inventory_snapshot::InventorySnapshot>>),
AuditEvent(Box<GetRequest<crate::object::inbuxa_audit::AuditEvent>>),
AuditSettings(Box<GetRequest<crate::object::inbuxa_audit::AuditSettings>>),
AccountLock(Box<GetRequest<crate::object::inbuxa_account_lock::AccountLock>>),
crates/jmap-proto/src/request/parser.rs
+14
View File
@@ -176,6 +176,20 @@ impl<'de> Visitor<'de> for CallVisitor {
return Err(de::Error::invalid_length(1, &self));
}
},
(MethodFunction::Get, MethodObject::DataInventory) => match seq.next_element() {
Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::DataInventory(value)),
Err(err) => RequestMethod::invalid(err),
Ok(None) => {
return Err(de::Error::invalid_length(1, &self));
}
},
(MethodFunction::Get, MethodObject::InventorySnapshot) => match seq.next_element() {
Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::InventorySnapshot(value)),
Err(err) => RequestMethod::invalid(err),
Ok(None) => {
return Err(de::Error::invalid_length(1, &self));
}
},
(MethodFunction::Get, MethodObject::ProtocolPolicy) => match seq.next_element() {
Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::ProtocolPolicy(value)),
Err(err) => RequestMethod::invalid(err),
crates/jmap-proto/src/response/mod.rs
+14
View File
@@ -104,6 +104,8 @@ pub enum GetResponseMethod {
DeletedAccount(GetResponse<crate::object::inbuxa_deleted_account::DeletedAccount>),
AiLimits(GetResponse<crate::object::inbuxa_ai_limits::AiLimits>),
LogSettings(GetResponse<crate::object::inbuxa_log_settings::LogSettings>),
DataInventory(GetResponse<crate::object::inbuxa_data_inventory::DataInventory>),
InventorySnapshot(GetResponse<crate::object::inbuxa_inventory_snapshot::InventorySnapshot>),
AuditEvent(GetResponse<crate::object::inbuxa_audit::AuditEvent>),
AuditSettings(GetResponse<crate::object::inbuxa_audit::AuditSettings>),
AccountLock(GetResponse<crate::object::inbuxa_account_lock::AccountLock>),
@@ -357,6 +359,18 @@ impl<'x> From<GetResponse<crate::object::inbuxa_log_settings::LogSettings>> for
}
}
impl<'x> From<GetResponse<crate::object::inbuxa_data_inventory::DataInventory>> for ResponseMethod<'x> {
fn from(value: GetResponse<crate::object::inbuxa_data_inventory::DataInventory>) -> Self {
ResponseMethod::Get(GetResponseMethod::DataInventory(value))
}
}
impl<'x> From<GetResponse<crate::object::inbuxa_inventory_snapshot::InventorySnapshot>> for ResponseMethod<'x> {
fn from(value: GetResponse<crate::object::inbuxa_inventory_snapshot::InventorySnapshot>) -> Self {
ResponseMethod::Get(GetResponseMethod::InventorySnapshot(value))
}
}
impl<'x> From<SetResponse<crate::object::inbuxa_ai_limits::AiLimits>> for ResponseMethod<'x> {
fn from(value: SetResponse<crate::object::inbuxa_ai_limits::AiLimits>) -> Self {
ResponseMethod::Set(SetResponseMethod::AiLimits(Box::new(value)))
crates/jmap/src/api/auth.rs
+6
View File
@@ -92,6 +92,10 @@ impl JmapAuthorization for AccessToken {
GetRequestMethod::AiLimits(_) => Permission::SysSpamLlmGet,
// inbuxa: log file retention, with the tracers' permissions
GetRequestMethod::LogSettings(_) => Permission::SysTracerGet,
// inbuxa: personal-data catalog, the inventory and its history
GetRequestMethod::DataInventory(_) | GetRequestMethod::InventorySnapshot(_) => {
Permission::SysComplianceGet
}
// inbuxa: the audit log (AU-9)
GetRequestMethod::AuditEvent(_) | GetRequestMethod::AuditSettings(_) => {
Permission::SysAuditGet
@@ -393,6 +397,8 @@ impl JmapAuthorization for AccessToken {
| MethodObject::DeletedAccount
| MethodObject::AiLimits
| MethodObject::LogSettings
| MethodObject::DataInventory
| MethodObject::InventorySnapshot
| MethodObject::Explanation
| MethodObject::AuditEvent
| MethodObject::AuditSettings
crates/jmap/src/api/request.rs
+14
View File
@@ -455,6 +455,20 @@ impl RequestHandler for Server {
.await?
.into()
}
// inbuxa: inbuxa:DataInventory/get
GetRequestMethod::DataInventory(mut req) => {
resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
crate::inbuxa::data_inventory::inventory_get(self, access_token, *req)
.await?
.into()
}
// inbuxa: inbuxa:InventorySnapshot/get
GetRequestMethod::InventorySnapshot(mut req) => {
resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
crate::inbuxa::data_inventory::snapshot_get(self, access_token, *req)
.await?
.into()
}
// inbuxa: account lock with delegation (AL-1)
GetRequestMethod::AccountLock(mut req) => {
resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
crates/jmap/src/changes/get.rs
+2
View File
@@ -419,6 +419,8 @@ impl IntermediateChangesResponse {
| MethodObject::DeletedAccount
| MethodObject::AiLimits
| MethodObject::LogSettings
| MethodObject::DataInventory
| MethodObject::InventorySnapshot
| MethodObject::Explanation
| MethodObject::AuditEvent
| MethodObject::AuditSettings
crates/jmap/src/inbuxa/ai_limits.rs
+2
View File
@@ -206,6 +206,8 @@ pub async fn set(
Some(error) => response.not_updated.append(id, error),
None => {
limits::set(data, &limits).await?;
// inbuxa: personal-data catalog: the inventory's history
server.inventory_snapshot_after("inbuxa:AiLimits").await;
response.updated.append(id, None);
}
}
crates/jmap/src/inbuxa/audit_log.rs
+2
View File
@@ -392,6 +392,8 @@ pub async fn settings_set(
Some(error) => response.not_updated.append(id, error),
None => {
log::set_settings(server.store(), &settings).await?;
// Audit retention is also the inventory's (personal-data catalog)
server.inventory_snapshot_after("inbuxa:AuditSettings").await;
response.updated.append(id, None);
}
}
crates/jmap/src/inbuxa/data_inventory.rs
+179
View File
@@ -0,0 +1,179 @@
/*
* SPDX-FileCopyrightText: 2026 Coffey Labs
*
* SPDX-License-Identifier: AGPL-3.0-only
*/
//! `inbuxa:DataInventory/get` and `inbuxa:InventorySnapshot/get`: the
//! personal-data catalog evaluated against this server, and its history
//! (personal-data catalog spec, §6). Read-only, with `sysComplianceGet`.
//!
//! Inside a tenant both answer with the tenant's slice: tenant-scoped
//! sources only, and none of the server's processors, which describe the
//! whole server. The same holds for snapshots, which are taken of the whole
//! server and cut to the slice when read.
use common::{Server, auth::AccessToken};
use inbuxa_features::privacy::{Inventory, snapshot};
use jmap_proto::{
method::get::{GetRequest, GetResponse},
object::{
inbuxa_data_inventory::{DataInventory, DataInventoryProperty as P, DataInventoryValue},
inbuxa_inventory_snapshot::{
InventorySnapshot, InventorySnapshotProperty as S, InventorySnapshotValue,
},
},
};
use jmap_tools::{Element, Key, Map, Property, Value};
use std::borrow::Cow;
use types::{brand_version, id::Id};
fn json_to_value<Pr: Property, E: Element<Property = Pr>>(
json: serde_json::Value,
) -> Value<'static, Pr, E> {
match json {
serde_json::Value::Null => Value::Null,
serde_json::Value::Bool(b) => Value::Bool(b),
serde_json::Value::Number(n) => {
if let Some(n) = n.as_u64() {
Value::Number(n.into())
} else if let Some(n) = n.as_i64() {
Value::Number(n.into())
} else {
Value::Number(n.as_f64().unwrap_or_default().into())
}
}
serde_json::Value::String(s) => Value::Str(Cow::Owned(s)),
serde_json::Value::Array(items) => {
Value::Array(items.into_iter().map(json_to_value).collect())
}
serde_json::Value::Object(map) => {
let mut out = Map::with_capacity(map.len());
for (key, value) in map {
out.insert_unchecked(Key::Owned(key), json_to_value(value));
}
Value::Object(out)
}
}
}
fn to_json<T: serde::Serialize>(value: &T) -> serde_json::Value {
serde_json::to_value(value).unwrap_or_default()
}
fn utc(seconds: u64) -> String {
jmap_proto::types::date::UTCDate::from_timestamp(seconds as i64).to_string()
}
/// A snapshot's inventory, cut to a tenant's slice when asked from one.
fn slice(mut inventory: Inventory, tenant_only: bool) -> Inventory {
if tenant_only {
inventory.items.retain(|item| item.scope == "tenant");
inventory.processors.clear();
}
inventory
}
/// `inbuxa:DataInventory/get`.
pub async fn inventory_get(
server: &Server,
access_token: &AccessToken,
mut request: GetRequest<DataInventory>,
) -> trc::Result<GetResponse<DataInventory>> {
let properties = request.unwrap_properties(&[
P::Id,
P::EvaluatedAt,
P::CatalogVersion,
P::Summary,
P::Items,
P::Processors,
]);
let (ids, not_found) = request.unwrap_ids(1)?;
let mut response = GetResponse {
account_id: request.account_id.into(),
state: None,
list: Vec::new(),
not_found,
};
let wanted = match ids {
None => true,
Some(ids) => {
let mut wanted = false;
for id in ids {
if id.is_singleton() {
wanted = true;
} else {
response.push_not_found(id);
}
}
wanted
}
};
if wanted {
let inventory = server
.data_inventory(access_token.tenant_id().is_some())
.await?;
let mut out = Map::with_capacity(properties.len());
for property in &properties {
let value = match property {
P::Id => Value::Element(DataInventoryValue::Id(Id::singleton())),
P::EvaluatedAt => Value::Str(utc(store::write::now()).into()),
P::CatalogVersion => Value::Str(brand_version!().into()),
P::Summary => json_to_value(to_json(&inventory.summary())),
P::Items => json_to_value(to_json(&inventory.items)),
P::Processors => json_to_value(to_json(&inventory.processors)),
};
out.insert_unchecked(Key::Property(property.clone()), value);
}
response.list.push(Value::Object(out));
}
Ok(response)
}
/// `inbuxa:InventorySnapshot/get`: by id (the time taken, as an id), or
/// `ids: null` for every snapshot kept, newest first. `inventory` is the
/// whole evaluated inventory; leave it out of `properties` for the list.
pub async fn snapshot_get(
server: &Server,
access_token: &AccessToken,
mut request: GetRequest<InventorySnapshot>,
) -> trc::Result<GetResponse<InventorySnapshot>> {
let tenant_only = access_token.tenant_id().is_some();
let data = &server.core.storage.data;
let properties =
request.unwrap_properties(&[S::Id, S::TakenAt, S::Trigger, S::Summary, S::Inventory]);
let times: Vec<u64> = match request.ids.take() {
None => snapshot::list(data, 0, u64::MAX).await?,
Some(_) => {
let (ids, _) = request.unwrap_ids(server.core.jmap.get_max_objects)?;
ids.unwrap_or_default().into_iter().map(|id| id.id()).collect()
}
};
let mut response = GetResponse {
account_id: request.account_id.into(),
state: None,
list: Vec::new(),
not_found: vec![],
};
for taken_at in times {
let Some(found) = snapshot::get(data, taken_at).await? else {
response.push_not_found(Id::from(taken_at));
continue;
};
let inventory = slice(found.inventory, tenant_only);
let summary = if tenant_only { inventory.summary() } else { found.summary };
let mut out = Map::with_capacity(properties.len());
for property in &properties {
let value = match property {
S::Id => Value::Element(InventorySnapshotValue::Id(Id::from(taken_at))),
S::TakenAt => Value::Str(utc(found.taken_at).into()),
S::Trigger => json_to_value(to_json(&found.trigger)),
S::Summary => json_to_value(to_json(&summary)),
S::Inventory => json_to_value(to_json(&inventory)),
};
out.insert_unchecked(Key::Property(property.clone()), value);
}
response.list.push(Value::Object(out));
}
Ok(response)
}
crates/jmap/src/inbuxa/log_settings.rs
+1
View File
@@ -154,6 +154,7 @@ pub async fn set(
log_files::set(data, &settings).await?;
// This node purges now; the others within the hour
log_files::CHANGED.notify_one();
server.inventory_snapshot_after("inbuxa:LogSettings").await;
response.updated.append(id, None);
}
}
crates/jmap/src/inbuxa/mod.rs
+1
View File
@@ -16,6 +16,7 @@ pub mod audit;
pub mod audit_log;
pub mod ai_limits;
pub mod log_settings;
pub mod data_inventory;
pub mod explanation;
pub mod protocol_policy;
pub mod tenant_protocol_policy;
crates/jmap/src/registry/set.rs
+14 -1
View File
@@ -902,7 +902,20 @@ impl RegistrySet for Server {
// Finalize cache invalidation
self.invalidate_caches(cache_invalidator).await?;
Ok(set.into_response())
// inbuxa: personal-data catalog: what the server holds may
// have changed, so the inventory's history is brought up to date
let response = set.into_response();
if !response.created.is_empty()
|| !response.updated.is_empty()
|| !response.destroyed.is_empty()
{
self.inventory_snapshot_after(&format!(
"x:{}",
registry::types::EnumImpl::as_str(&object_type)
))
.await;
}
Ok(response)
}
ObjectType::ArfExternalReport
| ObjectType::DmarcExternalReport
crates/services/src/task_manager/maintenance.rs
+12
View File
@@ -269,6 +269,18 @@ async fn store_maintenance(
trc::error!(err.details("Failed to re-apply account locks"));
}
// inbuxa: personal-data catalog: the inventory's daily look for
// a change, and snapshots past the audit log's retention go
if let Err(err) = server
.inventory_snapshot(inbuxa_features::privacy::snapshot::Trigger::Daily)
.await
{
trc::error!(err.details("Failed to record an inventory snapshot"));
}
if let Err(err) = server.purge_inventory_snapshots().await {
trc::error!(err.details("Failed to purge inventory snapshots"));
}
// inbuxa: personal-data catalog, D2: bans past their period go
if let Err(err) = server.purge_expired_blocked_ips().await {
trc::error!(err.details("Failed to purge expired IP bans"));
docs/spec/features/personal-data-catalog.md
+20
View File
@@ -497,6 +497,26 @@ leaving the host, processors), and on `get` the full inventory as above.
Kept for `inbuxa:AuditSettings.keepForDays`, so history is as long as the
audit log's. Same permission and tenant scoping as the inventory.
### As built (2026-09-28)
- The catalog is embedded and parsed at start
(`crates/features/src/privacy/`, `toml` crate); the evaluation is a pure
function of the catalog and the live facts, which
`crates/common/src/privacy.rs` gathers: retention settings, what is
switched on (tracers, webhooks, the classifier and its AI model, Explain,
DNSBL, Pyzor, milters, hooks, relays, archiving, report keeping), and
stores or endpoints off the host. Loopback endpoints stay on the host;
any other configured endpoint counts as leaving it.
- Objects with nothing personal aren't listed. An object's categories are
the union of its properties'.
- `inbuxa:InventorySnapshot` has `get` only: `ids: null` lists every
snapshot kept, newest first, and `inventory` is sent only when asked for
in `properties` (the `query` above folds into this).
- A snapshot is recorded only when the evaluated inventory differs from
the newest one (or there is none): after a registry write to an object
the inventory reads, after inbuxa's log, audit or AI settings change, and
on the daily clean-up. Snapshots past the audit log's retention go then.
## 7. The compliance role
### What it holds
resources/privacy/catalog.toml
+10
View File
@@ -165,6 +165,16 @@ default = "none"
file = "inbuxa_log_settings.rs"
default = "none"
# The inventory itself: which kinds of data the server holds, where, and the
# hosts that receive them. Facts about the server, not about people.
[object."inbuxa:DataInventory"]
file = "inbuxa_data_inventory.rs"
default = "none"
[object."inbuxa:InventorySnapshot"]
file = "inbuxa_inventory_snapshot.rs"
default = "none"
# ---------------------------------------------------------------------------
# Sources that are no object. Settings are `<object>.<property>`: a schema
# field object, or one of inbuxa's own.
tests/src/system/compliance.rs
+89
View File
@@ -185,6 +185,95 @@ pub async fn test(test: &mut TestServer) {
let (name, response) = call(&t_officer, "inbuxa:LegalHold/get", json!({"ids": null})).await;
assert_eq!(name, "error", "LH-13: the tenant officer read holds: {response}");
// The data inventory (§6): the officer reads it, facts not verdicts
let (name, response) = call(&officer, "inbuxa:DataInventory/get", json!({"ids": null})).await;
assert_eq!(name, "inbuxa:DataInventory/get", "{response}");
let inventory = response["list"][0].clone();
let ids: Vec<&str> = inventory["items"]
.as_array()
.unwrap()
.iter()
.filter_map(|i| i["id"].as_str())
.collect();
assert!(ids.contains(&"x:UserAccount") && ids.contains(&"log-file"), "{ids:?}");
assert!(inventory["summary"]["collected"].as_u64().unwrap_or(0) > 0);
assert!(!inventory.to_string().to_lowercase().contains("complian"), "facts only");
// Somebody without the permission is refused
let plain = admin
.create_user_account("[email protected]", "plain-secret-1182", "Plain", &[], vec![])
.await;
let (name, _) = call(&plain, "inbuxa:DataInventory/get", json!({"ids": null})).await;
assert_eq!(name, "error", "a user without sysComplianceGet read the inventory");
// A tenant's officer sees the tenant's slice, and no processors
let (_, response) = call(&t_officer, "inbuxa:DataInventory/get", json!({"ids": null})).await;
let slice = &response["list"][0];
assert!(
slice["items"].as_array().is_some_and(|items| !items.is_empty()
&& items.iter().all(|i| i["scope"] == "tenant")),
"{slice}"
);
assert_eq!(slice["processors"], json!([]));
// A webhook to another host makes it a candidate processor, and the
// change is in the inventory's history
let (_, response) = call(
&admin,
"x:WebHook/set",
json!({"create": {"w": {"url": "https://hooks.example.net/in", "enable": true}}}),
)
.await;
assert!(response["created"].get("w").is_some(), "{response}");
let (_, response) = call(&officer, "inbuxa:DataInventory/get", json!({"ids": null})).await;
let inventory = &response["list"][0];
assert!(
inventory["processors"]
.as_array()
.is_some_and(|p| p.iter().any(|p| p["host"] == "hooks.example.net")),
"{inventory}"
);
let webhooks = inventory["items"]
.as_array()
.unwrap()
.iter()
.find(|i| i["id"] == "webhooks")
.unwrap();
assert_eq!(webhooks["collected"], json!(true));
assert_eq!(webhooks["leavesHost"], json!(true));
let (_, response) = call(
&officer,
"inbuxa:InventorySnapshot/get",
json!({"ids": null, "properties": ["id", "takenAt", "trigger", "summary"]}),
)
.await;
let snapshots = response["list"].as_array().cloned().unwrap_or_default();
assert!(
snapshots
.iter()
.any(|s| s["trigger"]["kind"] == "settingChanged" && s["trigger"]["setting"] == "x:WebHook"),
"the webhook's snapshot: {snapshots:?}"
);
assert!(snapshots.iter().all(|s| s.get("inventory").is_none()), "left out when not asked");
// A retention change reads through
let (_, response) = call(
&admin,
"x:DataRetention/set",
json!({"update": {"singleton": {"holdTracesFor": 604800000}}}),
)
.await;
assert!(response["updated"].get("singleton").is_some(), "{response}");
let (_, response) = call(&officer, "inbuxa:DataInventory/get", json!({"ids": null})).await;
let trace = response["list"][0]["items"]
.as_array()
.unwrap()
.iter()
.find(|i| i["id"] == "x:Trace")
.cloned()
.unwrap();
assert_eq!(trace["retention"]["days"], json!(7), "{trace}");
// A tenant can still be deleted: its unused role goes with it
let spare = admin
.registry_create_object(Tenant {