use super::*;
use serde::Serialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use sqlx::Row;
const EVIDENCE_MANIFEST_SCHEMA_VERSION: &str = "1";
const EVIDENCE_MANIFEST_KIND: &str = "udb.compliance.evidence.manifest";
const EVIDENCE_CHAIN_GENESIS: &str =
"0000000000000000000000000000000000000000000000000000000000000000";
const EVIDENCE_CHAIN_ALGORITHM: &str = "sha256";
const AUTH_AUDIT_LOG_RELATION: &str = "udb_system.auth_audit_log";
const DEFAULT_EVIDENCE_PREFIX: &str = "compliance-evidence";
const DEFAULT_EVIDENCE_BACKEND: &str = "s3";
const DEFAULT_EVIDENCE_BUCKET: &str = "udb-compliance-evidence";
const EVIDENCE_BUNDLE_CONTENT_TYPE: &str = "application/x-ndjson";
#[derive(Debug, Clone, Serialize, PartialEq)]
struct EvidenceRecord {
event_id: String,
event_type: String,
tenant_id: String,
actor: String,
target_resource: String,
operation: String,
outcome: String,
reason_code: String,
occurred_at: String,
envelope: Value,
}
fn chain_hash(prev_hash: &str, canonical_record: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(prev_hash.as_bytes());
hasher.update(b"\n");
hasher.update(canonical_record.as_bytes());
let digest = hasher.finalize();
let mut hex = String::with_capacity(digest.len() * 2);
for byte in digest {
hex.push_str(&format!("{byte:02x}"));
}
hex
}
fn build_evidence_bundle(records: &[EvidenceRecord]) -> Result<(String, String), String> {
let mut prev = EVIDENCE_CHAIN_GENESIS.to_string();
let mut jsonl = String::new();
for record in records {
let canonical = serde_json::to_string(record)
.map_err(|err| format!("serialize evidence record failed: {err}"))?;
let hash = chain_hash(&prev, &canonical);
let mut line = serde_json::to_value(record)
.map_err(|err| format!("encode evidence record failed: {err}"))?;
line["prev_hash"] = json!(prev);
line["chain_hash"] = json!(hash);
jsonl.push_str(
&serde_json::to_string(&line)
.map_err(|err| format!("encode evidence line failed: {err}"))?,
);
jsonl.push('\n');
prev = hash;
}
Ok((jsonl, prev))
}
#[allow(clippy::too_many_arguments)]
fn render_evidence_manifest(
since: &str,
until: &str,
tenant: &str,
record_count: usize,
chain_head: &str,
backend: &str,
bucket: &str,
evidence_key: &str,
evidence_bytes: usize,
manifest_key: &str,
written: bool,
generated_at: &str,
) -> Value {
let opt = |value: &str| {
if value.trim().is_empty() {
Value::Null
} else {
json!(value)
}
};
json!({
"schema_version": EVIDENCE_MANIFEST_SCHEMA_VERSION,
"kind": EVIDENCE_MANIFEST_KIND,
"generated_at": generated_at,
"source_relation": AUTH_AUDIT_LOG_RELATION,
"window": { "since": opt(since), "until": opt(until) },
"tenant_filter": opt(tenant),
"record_count": record_count,
"chain": {
"algorithm": EVIDENCE_CHAIN_ALGORITHM,
"genesis": EVIDENCE_CHAIN_GENESIS,
"head": chain_head,
},
"evidence_object": {
"backend": backend,
"bucket": bucket,
"object_key": evidence_key,
"content_type": EVIDENCE_BUNDLE_CONTENT_TYPE,
"bytes": evidence_bytes,
"written": written,
},
"manifest_object": {
"backend": backend,
"bucket": bucket,
"object_key": manifest_key,
"content_type": "application/json",
"written": written,
},
})
}
fn object_put_request_json(bucket: &str, object_key: &str, content_type: &str) -> String {
json!({
"op": "put",
"bucket": bucket,
"container": bucket,
"object_key": object_key,
"key": object_key,
"object": object_key,
"blob": object_key,
"content_type": content_type,
})
.to_string()
}
fn evidence_object_slug(since: &str, until: &str, generated_at: &str) -> String {
let sanitize = |value: &str| -> String {
value
.chars()
.map(|c| if c.is_ascii_alphanumeric() { c } else { '-' })
.collect()
};
let since = if since.trim().is_empty() {
"begin".to_string()
} else {
sanitize(since.trim())
};
let until = if until.trim().is_empty() {
"now".to_string()
} else {
sanitize(until.trim())
};
format!("{since}_{until}__{}", sanitize(generated_at))
}
fn resolve_backend(flag: &str) -> String {
let flag = flag.trim();
if !flag.is_empty() {
return flag.to_string();
}
env::var("UDB_OBJECT_BACKEND")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| DEFAULT_EVIDENCE_BACKEND.to_string())
}
fn resolve_bucket(flag: &str) -> String {
let flag = flag.trim();
if !flag.is_empty() {
return flag.to_string();
}
env::var("UDB_COMPLIANCE_EVIDENCE_BUCKET")
.or_else(|_| env::var("UDB_OBJECT_BUCKET"))
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| DEFAULT_EVIDENCE_BUCKET.to_string())
}
fn resolve_project(flag: &str) -> String {
let flag = flag.trim();
if !flag.is_empty() {
return flag.to_string();
}
env::var("UDB_PROJECT_ID")
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| udb::runtime::catalog::DEFAULT_PROJECT_ID.to_string())
}
async fn collect_evidence_records(
pool: &sqlx::PgPool,
since: &str,
until: &str,
tenant: &str,
limit: i64,
) -> Result<Vec<EvidenceRecord>, String> {
let since_opt = (!since.trim().is_empty()).then(|| since.trim().to_string());
let until_opt = (!until.trim().is_empty()).then(|| until.trim().to_string());
let tenant = tenant.trim().to_string();
let row_limit = if limit <= 0 { i64::MAX } else { limit };
let sql = format!(
"SELECT event_id::text AS event_id, event_type, tenant_id, actor, \
target_resource, operation, outcome, reason_code, \
occurred_at::text AS occurred_at, envelope::text AS envelope \
FROM {rel} \
WHERE ($1::timestamptz IS NULL OR occurred_at >= $1::timestamptz) \
AND ($2::timestamptz IS NULL OR occurred_at <= $2::timestamptz) \
AND ($3 = '' OR tenant_id = $3) \
ORDER BY occurred_at ASC, event_id ASC \
LIMIT $4",
rel = AUTH_AUDIT_LOG_RELATION
);
let rows = sqlx::query(&sql)
.bind(since_opt)
.bind(until_opt)
.bind(&tenant)
.bind(row_limit)
.fetch_all(pool)
.await
.map_err(|err| format!("query audit window failed: {err}"))?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let get = |key: &str| -> String { row.try_get::<String, _>(key).unwrap_or_default() };
let envelope_text: String = row.try_get("envelope").unwrap_or_default();
let envelope: Value = serde_json::from_str(&envelope_text).unwrap_or(Value::Null);
out.push(EvidenceRecord {
event_id: get("event_id"),
event_type: get("event_type"),
tenant_id: get("tenant_id"),
actor: get("actor"),
target_resource: get("target_resource"),
operation: get("operation"),
outcome: get("outcome"),
reason_code: get("reason_code"),
occurred_at: get("occurred_at"),
envelope,
});
}
Ok(out)
}
pub(crate) fn run_compliance_command(command: ComplianceCommand) -> i32 {
let runtime = match tokio::runtime::Runtime::new() {
Ok(runtime) => runtime,
Err(err) => {
eprintln!("compliance: failed to create tokio runtime: {err}");
return 1;
}
};
match runtime.block_on(run_compliance_command_async(command)) {
Ok(value) => {
output_json(&value, "compliance evidence manifest");
0
}
Err(err) => {
eprintln!("compliance: {err}");
1
}
}
}
async fn run_compliance_command_async(command: ComplianceCommand) -> Result<Value, String> {
match command {
ComplianceCommand::Evidence {
since,
until,
tenant,
backend,
bucket,
prefix,
project,
limit,
out,
dry_run,
} => {
let rt = udb::DataBrokerRuntime::from_env().await;
let Some(pool) = rt.pg_pool_clone() else {
return Err(
"PostgreSQL is not configured (set UDB_PG_DSN or DATABASE_URL)".to_string(),
);
};
let records = collect_evidence_records(&pool, &since, &until, &tenant, limit).await?;
let (jsonl, chain_head) = build_evidence_bundle(&records)?;
let backend = resolve_backend(&backend);
let bucket = resolve_bucket(&bucket);
let project = resolve_project(&project);
let generated_at = chrono::Utc::now().to_rfc3339();
let prefix = {
let trimmed = prefix.trim();
if trimmed.is_empty() {
DEFAULT_EVIDENCE_PREFIX.to_string()
} else {
trimmed.trim_end_matches('/').to_string()
}
};
let slug = evidence_object_slug(&since, &until, &generated_at);
let evidence_key = format!("{prefix}/evidence-{slug}.jsonl");
let manifest_key = format!("{prefix}/evidence-{slug}.manifest.json");
let manifest = render_evidence_manifest(
&since,
&until,
&tenant,
records.len(),
&chain_head,
&backend,
&bucket,
&evidence_key,
jsonl.len(),
&manifest_key,
!dry_run,
&generated_at,
);
if !out.trim().is_empty() {
let pretty = serde_json::to_string_pretty(&manifest)
.map_err(|err| format!("render local manifest failed: {err}"))?;
std::fs::write(out.trim(), pretty).map_err(|err| {
format!("write local manifest '{}' failed: {err}", out.trim())
})?;
}
if dry_run {
return Ok(manifest);
}
let evidence_req =
object_put_request_json(&bucket, &evidence_key, EVIDENCE_BUNDLE_CONTENT_TYPE);
rt.put_object_backend_target_for_project(
&backend,
None,
&project,
&evidence_req,
jsonl.into_bytes(),
)
.await
.map_err(|err| format!("write evidence bundle failed: {err}"))?;
let manifest_bytes = serde_json::to_vec(&manifest)
.map_err(|err| format!("encode manifest bytes failed: {err}"))?;
let manifest_req = object_put_request_json(&bucket, &manifest_key, "application/json");
rt.put_object_backend_target_for_project(
&backend,
None,
&project,
&manifest_req,
manifest_bytes,
)
.await
.map_err(|err| format!("write manifest object failed: {err}"))?;
Ok(manifest)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn rec(id: &str, op: &str) -> EvidenceRecord {
EvidenceRecord {
event_id: id.to_string(),
event_type: "udb.authz.access.denied.v1".to_string(),
tenant_id: "t1".to_string(),
actor: "user-1".to_string(),
target_resource: "doc/1".to_string(),
operation: op.to_string(),
outcome: "deny".to_string(),
reason_code: "abac_default_deny".to_string(),
occurred_at: "2026-06-25T00:00:00Z".to_string(),
envelope: json!({ "event_id": id, "operation": op }),
}
}
#[test]
fn empty_window_chains_to_genesis() {
let (jsonl, head) = build_evidence_bundle(&[]).expect("bundle");
assert!(jsonl.is_empty());
assert_eq!(head, EVIDENCE_CHAIN_GENESIS);
}
#[test]
fn chain_is_deterministic_and_links_records() {
let records = vec![rec("e1", "create"), rec("e2", "delete")];
let (jsonl, head) = build_evidence_bundle(&records).expect("bundle");
let (jsonl2, head2) = build_evidence_bundle(&records).expect("bundle");
assert_eq!(jsonl, jsonl2, "bundle must be deterministic");
assert_eq!(head, head2);
let lines: Vec<Value> = jsonl
.lines()
.map(|line| serde_json::from_str(line).expect("valid json line"))
.collect();
assert_eq!(lines.len(), 2);
assert_eq!(lines[0]["prev_hash"], json!(EVIDENCE_CHAIN_GENESIS));
assert_eq!(lines[1]["prev_hash"], lines[0]["chain_hash"]);
assert_eq!(lines[1]["chain_hash"], json!(head));
assert_ne!(lines[0]["chain_hash"], lines[1]["chain_hash"]);
}
#[test]
fn tampering_an_early_record_changes_the_head() {
let base = vec![rec("e1", "create"), rec("e2", "delete")];
let mut tampered = base.clone();
tampered[0].outcome = "allow".to_string();
let (_, head_a) = build_evidence_bundle(&base).expect("bundle");
let (_, head_b) = build_evidence_bundle(&tampered).expect("bundle");
assert_ne!(
head_a, head_b,
"altering an early record must change the head"
);
}
#[test]
fn manifest_shape_is_machine_readable() {
let manifest = render_evidence_manifest(
"2026-06-01T00:00:00Z",
"",
"tenant-9",
3,
"deadbeef",
"s3",
"udb-compliance-evidence",
"compliance-evidence/evidence-x.jsonl",
1234,
"compliance-evidence/evidence-x.manifest.json",
true,
"2026-06-25T12:00:00Z",
);
assert_eq!(
manifest["schema_version"],
json!(EVIDENCE_MANIFEST_SCHEMA_VERSION)
);
assert_eq!(manifest["kind"], json!(EVIDENCE_MANIFEST_KIND));
assert_eq!(manifest["source_relation"], json!(AUTH_AUDIT_LOG_RELATION));
assert_eq!(manifest["record_count"], json!(3));
assert_eq!(manifest["window"]["since"], json!("2026-06-01T00:00:00Z"));
assert_eq!(manifest["window"]["until"], Value::Null);
assert_eq!(manifest["tenant_filter"], json!("tenant-9"));
assert_eq!(
manifest["chain"]["algorithm"],
json!(EVIDENCE_CHAIN_ALGORITHM)
);
assert_eq!(manifest["chain"]["head"], json!("deadbeef"));
assert_eq!(
manifest["evidence_object"]["bucket"],
json!("udb-compliance-evidence")
);
assert_eq!(manifest["evidence_object"]["bytes"], json!(1234));
assert_eq!(manifest["evidence_object"]["written"], json!(true));
}
#[test]
fn object_put_request_carries_executor_aliases() {
let req: Value =
serde_json::from_str(&object_put_request_json("b", "k", "application/json"))
.expect("valid json");
assert_eq!(req["op"], json!("put"));
assert_eq!(req["bucket"], json!("b"));
assert_eq!(req["container"], json!("b"));
assert_eq!(req["object_key"], json!("k"));
assert_eq!(req["key"], json!("k"));
assert_eq!(req["content_type"], json!("application/json"));
}
#[test]
fn arg_parse_evidence_reads_flags() {
let args: Vec<String> = [
"compliance",
"evidence",
"--since",
"2026-01-01T00:00:00Z",
"--tenant",
"t9",
"--bucket",
"ev",
"--prefix",
"ev-prefix",
"--limit",
"50",
"--dry-run",
]
.iter()
.map(|value| value.to_string())
.collect();
let (command, _, _, _) = parse_args(&args);
match command {
Command::Compliance(ComplianceCommand::Evidence {
since,
tenant,
bucket,
prefix,
limit,
dry_run,
..
}) => {
assert_eq!(since, "2026-01-01T00:00:00Z");
assert_eq!(tenant, "t9");
assert_eq!(bucket, "ev");
assert_eq!(prefix, "ev-prefix");
assert_eq!(limit, 50);
assert!(dry_run);
}
_ => panic!("expected compliance evidence command"),
}
}
}