use std::sync::{Arc, Mutex, PoisonError};
use std::time::Instant;
use lean_ctx_protocol::context_gateway::{
ContextDecisionReceiptV1, ContextDecisionV1, ContextDispositionV1, DeliveryOutcomeV1,
DestinationLocalityV1, DestinationV1, GatewayModeV1, MAX_DECISIONS, PolicyRefV1,
PrincipalKindV1, PrincipalV1, RECEIPT_V2_SCHEMA_VERSION, ReasonCodeV1, SecurityCountsV1,
SourceCountsV1, TokenAccountV1,
};
use lean_ctx_protocol::{PolicyId, ProtocolReference, Sha256Digest, TaskId};
use super::AdmissionCounts;
tokio::task_local! {
pub static ADMISSIONS: Option<Arc<AdmissionCapture>>;
}
pub fn current() -> Option<Arc<AdmissionCapture>> {
ADMISSIONS.try_with(Clone::clone).ok().flatten()
}
#[derive(Debug, Clone)]
struct Recorded {
decision: ContextDecisionV1,
counts: AdmissionCounts,
external: bool,
}
pub const UNTRUSTED_ORIGIN: &str = "origin.untrusted";
pub const MAX_LEDGER_REFS: usize = 256;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OriginLedger {
pub count: u64,
pub refs: std::collections::BTreeSet<super::origin::InputRef>,
pub overflow: bool,
}
impl OriginLedger {
fn note(&mut self, input: Option<super::origin::InputRef>) {
self.count = self.count.saturating_add(1);
match input {
Some(input) if self.refs.contains(&input) => {}
Some(input) if self.refs.len() < MAX_LEDGER_REFS => {
self.refs.insert(input);
}
_ => self.overflow = true,
}
}
fn merge(&mut self, other: &Self) {
self.count = self.count.saturating_add(other.count);
for input in &other.refs {
if !self.refs.contains(input) {
if self.refs.len() < MAX_LEDGER_REFS {
self.refs.insert(input.clone());
} else {
self.overflow = true;
}
}
}
self.overflow |= other.overflow;
}
pub fn is_empty(&self) -> bool {
self.count == 0 && !self.overflow
}
}
pub static ORPHAN_EXTERNAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
thread_local! {
static OUTSIDE_CALL: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
static IN_SPAN: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
static EGRESS: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}
pub fn outside_call_active() -> bool {
OUTSIDE_CALL.with(std::cell::Cell::get) > 0
}
pub fn egress_check_active() -> bool {
EGRESS.with(std::cell::Cell::get) > 0
}
pub fn egress_check<T>(check: impl FnOnce() -> T) -> T {
struct Leave;
impl Drop for Leave {
fn drop(&mut self) {
EGRESS.with(|depth| depth.set(depth.get().saturating_sub(1)));
}
}
EGRESS.with(|depth| depth.set(depth.get().saturating_add(1)));
let _leave = Leave;
outside_call(check)
}
pub fn outside_call<T>(operation: impl FnOnce() -> T) -> T {
struct Leave;
impl Drop for Leave {
fn drop(&mut self) {
OUTSIDE_CALL.with(|depth| depth.set(depth.get().saturating_sub(1)));
}
}
OUTSIDE_CALL.with(|depth| depth.set(depth.get().saturating_add(1)));
let _leave = Leave;
operation()
}
pub fn note_origin(class: &super::origin::SourceClass) {
if !class.is_external_or_unknown() || outside_call_active() {
return;
}
let input = class.input_ref();
match current() {
Some(capture) => capture.note_origin_ref(input),
None => {
ORPHAN_EXTERNAL.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
}
}
#[derive(Clone)]
pub struct Carried {
capture: Option<Arc<AdmissionCapture>>,
outside: bool,
project: Option<std::path::PathBuf>,
}
pub fn carry() -> Carried {
Carried {
capture: current(),
outside: outside_call_active(),
project: crate::core::policy::runtime::REQUEST_PROJECT
.try_with(|slot| slot.borrow().clone())
.ok()
.flatten(),
}
}
impl Carried {
pub fn enter<T>(&self, operation: impl FnOnce() -> T) -> T {
let inner = || {
if self.outside {
outside_call(operation)
} else {
operation()
}
};
let with_project = || match &self.project {
Some(project) if crate::core::policy::diagnostics::request_project().is_none() => {
crate::core::policy::runtime::in_project(project.clone(), inner)
}
_ => inner(),
};
if self.capture.is_some() && current().is_none() {
ADMISSIONS.sync_scope(self.capture.clone(), with_project)
} else {
with_project()
}
}
}
pub fn external_span(
class: &super::origin::SourceClass,
admit: impl FnOnce() -> Option<String>,
) -> Option<String> {
let Some(capture) =
current().filter(|_| class.is_external_or_unknown() && !outside_call_active())
else {
return admit();
};
struct Leave;
impl Drop for Leave {
fn drop(&mut self) {
IN_SPAN.with(|depth| depth.set(depth.get().saturating_sub(1)));
}
}
IN_SPAN.with(|depth| depth.set(depth.get().saturating_add(1)));
let admitted = {
let _leave = Leave;
admit()
};
if let Some(text) = &admitted
&& !text.is_empty()
{
capture
.spans
.lock()
.unwrap_or_else(PoisonError::into_inner)
.push((class.input_ref(), text.clone()));
}
admitted
}
fn in_span() -> bool {
IN_SPAN.with(std::cell::Cell::get) > 0
}
#[derive(Debug, Clone)]
pub struct CallSnapshot {
admission: super::stores::StoreAdmission,
protected: bool,
pack: Option<(String, String)>,
org_policy: Option<OrgObservation>,
}
pub(crate) type OrgObservation = crate::core::policy::org::Observation;
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum OrgEvidence {
#[default]
None,
One(OrgObservation),
Diverged,
Unavailable,
}
impl OrgEvidence {
fn of(observation: Option<OrgObservation>) -> Self {
observation.map_or(Self::None, Self::One)
}
}
impl CallSnapshot {
pub fn take() -> Self {
let admission = super::carrier::gateway_snapshot();
let protected =
super::stored::source_authority_required_under(admission.admission_policy());
let (active, org_policy) = crate::core::policy::runtime::active_observed();
let pack = active.map(|active| {
(
active.resolved.name.clone(),
active.resolved.version.clone(),
)
});
Self {
admission,
protected,
pack,
org_policy,
}
}
pub fn admission(&self) -> &super::stores::StoreAdmission {
&self.admission
}
pub(crate) fn protected(&self) -> bool {
self.protected
}
pub(crate) fn org_policy(&self) -> OrgEvidence {
OrgEvidence::of(self.org_policy.clone())
}
pub(crate) fn pack(&self) -> Option<&(String, String)> {
self.pack.as_ref()
}
}
pub(crate) const MAX_SOURCE_REFERENCES: usize = 64;
pub(crate) const MAX_SOURCE_REFERENCE_BYTES: usize = 256;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SourceName {
Path(String),
Id(String),
}
#[derive(Debug, Default)]
struct SourceNotes {
named: Vec<(Sha256Digest, SourceName)>,
omitted: u32,
}
thread_local! {
static UNNAMED: std::cell::RefCell<Option<(usize, Sha256Digest)>> =
const { std::cell::RefCell::new(None) };
}
pub(crate) fn note_source(project: &std::path::Path, canonical_path: &str) {
let Some((capture, object)) = unnamed_decision() else {
return;
};
let path = std::path::Path::new(canonical_path);
let relative = path
.strip_prefix(project)
.ok()
.map(std::path::Path::to_path_buf)
.or_else(|| {
let project = crate::core::pathutil::canonicalize_secure(project).ok()?;
path.strip_prefix(&project)
.ok()
.map(std::path::Path::to_path_buf)
});
let reference = relative
.as_deref()
.and_then(std::path::Path::to_str)
.filter(|reference| !reference.is_empty() && reference.len() <= MAX_SOURCE_REFERENCE_BYTES)
.map(|reference| SourceName::Path(reference.replace('\\', "/")));
capture.name_source(object, reference);
}
pub fn note_source_id(id: &str) {
let Some((capture, object)) = unnamed_decision() else {
return;
};
let reference = (!id.is_empty() && id.len() <= MAX_SOURCE_REFERENCE_BYTES)
.then(|| SourceName::Id(id.to_owned()));
capture.name_source(object, reference);
}
fn unnamed_decision() -> Option<(Arc<AdmissionCapture>, Sha256Digest)> {
let capture = current()?;
let owner = std::ptr::from_ref::<AdmissionCapture>(&capture).addr();
let (recorded_by, object) = UNNAMED.with(|slot| slot.borrow_mut().take())?;
(recorded_by == owner).then_some((capture, object))
}
#[derive(Debug)]
pub struct AdmissionCapture {
started: Instant,
records: Mutex<Vec<Recorded>>,
overflow: Mutex<Overflow>,
original_tokens: Mutex<Option<u64>>,
policy: Mutex<Option<(GatewayModeV1, Sha256Digest)>>,
task: Mutex<Option<(String, String)>>,
snapshot: Option<CallSnapshot>,
diverged: std::sync::atomic::AtomicBool,
sources: Mutex<SourceNotes>,
pack: Mutex<Option<(String, String)>>,
org_policy: Mutex<Option<OrgEvidence>>,
origin: Mutex<OriginLedger>,
spans: Mutex<Vec<(Option<super::origin::InputRef>, String)>>,
orphan_mark: u64,
output_class: Mutex<Option<super::origin::SourceClass>>,
released: Mutex<std::collections::BTreeSet<[u8; 32]>>,
}
const MAX_RELEASED_BLOCKS: usize = 4096;
impl Default for AdmissionCapture {
fn default() -> Self {
Self {
released: Mutex::new(std::collections::BTreeSet::new()),
origin: Mutex::new(OriginLedger::default()),
spans: Mutex::new(Vec::new()),
orphan_mark: ORPHAN_EXTERNAL.load(std::sync::atomic::Ordering::SeqCst),
output_class: Mutex::new(None),
started: Instant::now(),
records: Mutex::new(Vec::new()),
overflow: Mutex::new(Overflow::default()),
original_tokens: Mutex::new(None),
policy: Mutex::new(None),
task: Mutex::new(None),
snapshot: None,
diverged: std::sync::atomic::AtomicBool::new(false),
sources: Mutex::new(SourceNotes::default()),
pack: Mutex::new(None),
org_policy: Mutex::new(None),
}
}
}
impl AdmissionCapture {
pub fn new() -> Arc<Self> {
Arc::new(Self::default())
}
pub fn for_call(snapshot: CallSnapshot) -> Arc<Self> {
Arc::new(Self {
snapshot: Some(snapshot),
..Self::default()
})
}
pub fn snapshot(&self) -> Option<&CallSnapshot> {
self.snapshot.as_ref()
}
pub fn set_output_class(&self, class: super::origin::SourceClass) {
*self
.output_class
.lock()
.unwrap_or_else(PoisonError::into_inner) = Some(class);
}
pub fn output_class(&self) -> Option<super::origin::SourceClass> {
self.output_class
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
fn note_origin_ref(&self, input: Option<super::origin::InputRef>) {
if in_span() {
return;
}
self.origin
.lock()
.unwrap_or_else(PoisonError::into_inner)
.note(input);
}
pub fn origin_ledger(&self) -> OriginLedger {
let mut ledger = self
.origin
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
if self.orphan_external() {
ledger.overflow = true;
}
ledger
}
pub fn orphan_external(&self) -> bool {
ORPHAN_EXTERNAL.load(std::sync::atomic::Ordering::SeqCst) != self.orphan_mark
}
pub fn external_spans(&self) -> Vec<(Option<super::origin::InputRef>, String)> {
self.spans
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
pub fn mark_released(&self, text: &str) {
let mut released = self.released.lock().unwrap_or_else(PoisonError::into_inner);
if released.len() < MAX_RELEASED_BLOCKS {
released.insert(*blake3::hash(text.as_bytes()).as_bytes());
}
}
pub fn was_released(&self, text: &str) -> bool {
self.released
.lock()
.unwrap_or_else(PoisonError::into_inner)
.contains(blake3::hash(text.as_bytes()).as_bytes())
}
pub fn output_origins(&self) -> OriginLedger {
let mut ledger = self.origin_ledger();
for (input, _) in self.external_spans() {
ledger.note(input);
}
ledger
}
pub fn renote_origins(&self, origins: &OriginLedger) {
self.origin
.lock()
.unwrap_or_else(PoisonError::into_inner)
.merge(origins);
}
pub fn spans_into_ledger(&self) {
let spans = std::mem::take(&mut *self.spans.lock().unwrap_or_else(PoisonError::into_inner));
let mut ledger = self.origin.lock().unwrap_or_else(PoisonError::into_inner);
for (input, _) in spans {
ledger.note(input);
}
}
pub(crate) fn policy_diverged(&self) -> bool {
self.diverged.load(std::sync::atomic::Ordering::SeqCst)
}
pub fn source_references(&self) -> (Vec<(Sha256Digest, SourceName)>, u32) {
let sources = self.sources.lock().unwrap_or_else(PoisonError::into_inner);
(sources.named.clone(), sources.omitted)
}
fn name_source(&self, object: Sha256Digest, name: Option<SourceName>) {
let mut sources = self.sources.lock().unwrap_or_else(PoisonError::into_inner);
match name {
Some(name) if sources.named.len() < MAX_SOURCE_REFERENCES => {
sources.named.push((object, name));
}
_ => sources.omitted = sources.omitted.saturating_add(1),
}
}
pub(super) fn note_pack(&self, name: &str, version: &str) {
*self.pack.lock().unwrap_or_else(PoisonError::into_inner) =
Some((name.to_owned(), version.to_owned()));
}
pub(super) fn note_org_policy(&self, observation: Option<OrgObservation>) {
self.merge_org_policy(OrgEvidence::of(observation));
}
pub(super) fn note_org_policy_unavailable(&self) {
self.merge_org_policy(OrgEvidence::Unavailable);
}
fn merge_org_policy(&self, incoming: OrgEvidence) {
let mut noted = self
.org_policy
.lock()
.unwrap_or_else(PoisonError::into_inner);
*noted = Some(match noted.take() {
None => incoming,
Some(present) if present == incoming => present,
Some(_) => OrgEvidence::Diverged,
});
}
pub(crate) fn org_policy(&self) -> OrgEvidence {
match &self.snapshot {
Some(snapshot) => OrgEvidence::of(snapshot.org_policy.clone()),
None => self
.org_policy
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
.unwrap_or_default(),
}
}
pub(crate) fn pack(&self) -> Option<(String, String)> {
self.snapshot
.as_ref()
.and_then(|snapshot| snapshot.pack.clone())
.or_else(|| {
self.pack
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
})
}
pub(super) fn record(&self, decision: &ContextDecisionV1, counts: AdmissionCounts) {
self.record_as(decision, counts, false);
}
pub(super) fn record_as(
&self,
decision: &ContextDecisionV1,
counts: AdmissionCounts,
external: bool,
) {
if let Some(envelope) = crate::core::task_spine::TaskSpine::current() {
let mut task = self.task.lock().unwrap_or_else(PoisonError::into_inner);
if task.is_none() {
*task = Some((
envelope.task_id.as_str().to_owned(),
crate::core::context_store::task_scope(
envelope.tenant_id.as_ref(),
&envelope.project_id,
),
));
}
}
let nameable = decision.object != note_object() && decision.disposition.delivers_content();
UNNAMED.with(|slot| {
*slot.borrow_mut() =
nameable.then(|| (std::ptr::from_ref(self).addr(), decision.object.clone()));
});
let mut decision = decision.clone();
if external {
let code =
ReasonCodeV1::new(UNTRUSTED_ORIGIN).expect("built-in reason codes are valid");
if !decision.reason_codes.contains(&code) {
decision.reason_codes.push(code);
}
}
let incoming = Recorded {
decision,
counts,
external,
};
let mut records = self.records.lock().unwrap_or_else(PoisonError::into_inner);
if records.len() < MAX_DECISIONS {
records.push(incoming);
return;
}
let displaced = if incoming.is_clean() {
incoming
} else if let Some(position) = records.iter().position(Recorded::is_clean) {
std::mem::replace(&mut records[position], incoming)
} else if incoming.is_note() {
let Some(displaced) = mark_notes_not_itemized(&mut records) else {
return;
};
displaced
} else {
incoming
};
self.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner)
.add(&displaced);
}
pub(super) fn bind_policy(&self, mode: GatewayModeV1, digest: Sha256Digest) {
if let Some(snapshot) = &self.snapshot
&& snapshot.admission.digest() != digest.hex()
{
self.diverged
.store(true, std::sync::atomic::Ordering::SeqCst);
}
*self.policy.lock().unwrap_or_else(PoisonError::into_inner) = Some((mode, digest));
}
pub(crate) fn task_scope(&self) -> Option<String> {
self.task
.lock()
.unwrap_or_else(PoisonError::into_inner)
.as_ref()
.map(|(_, scope)| scope.clone())
}
pub fn set_original_tokens(&self, tokens: u64) {
*self
.original_tokens
.lock()
.unwrap_or_else(PoisonError::into_inner) = Some(tokens);
}
pub fn is_empty(&self) -> bool {
self.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.is_empty()
}
pub(crate) fn describes_sources(&self) -> bool {
let note = note_object();
let itemized = self
.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
.any(|record| record.decision.object != note);
itemized
|| self
.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner)
.objects
> 0
}
pub(super) fn record_unsourced(
&self,
output: Sha256Digest,
mode: GatewayModeV1,
policy: Sha256Digest,
) {
if self.bound_policy().is_none() {
self.bind_policy(mode, policy);
}
let decision = ContextDecisionV1 {
object: output,
disposition: ContextDispositionV1::Allow,
reason_codes: vec![
ReasonCodeV1::new(NO_SOURCE_DECISION).expect("built-in reason codes are valid"),
],
signals: Vec::new(),
required_transformations: Vec::new(),
};
self.record(&decision, AdmissionCounts::default());
}
pub fn itemized(&self) -> Vec<(ContextDecisionV1, AdmissionCounts)> {
self.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
.map(|record| (record.decision.clone(), record.counts))
.collect()
}
pub(crate) fn counted_only(&self) -> u32 {
self.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner)
.objects
}
pub(super) fn bound_policy(&self) -> Option<(GatewayModeV1, Sha256Digest)> {
self.policy
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
pub(super) fn absorb(&self, earlier: &Self) {
if let Some((name, version)) = earlier.pack()
&& self.pack().is_none()
{
self.note_pack(&name, &version);
}
let theirs = earlier
.org_policy
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
if let Some(theirs) = theirs {
self.merge_org_policy(theirs);
}
let records = earlier
.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
for record in &records {
self.record_as(&record.decision, record.counts, record.external);
}
let ledger = earlier
.origin
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
self.origin
.lock()
.unwrap_or_else(PoisonError::into_inner)
.merge(&ledger);
let spans = earlier.external_spans();
self.spans
.lock()
.unwrap_or_else(PoisonError::into_inner)
.extend(spans);
let released = earlier
.released
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
{
let mut ours = self.released.lock().unwrap_or_else(PoisonError::into_inner);
for digest in released {
if ours.len() >= MAX_RELEASED_BLOCKS {
break;
}
ours.insert(digest);
}
}
let theirs = earlier
.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner);
let mut ours = self.overflow.lock().unwrap_or_else(PoisonError::into_inner);
ours.objects = ours.objects.saturating_add(theirs.objects);
ours.permitted = ours.permitted.saturating_add(theirs.permitted);
ours.blocked = ours.blocked.saturating_add(theirs.blocked);
ours.redactions = ours.redactions.saturating_add(theirs.redactions);
ours.injection = ours.injection.saturating_add(theirs.injection);
ours.incomplete = ours.incomplete.saturating_add(theirs.incomplete);
ours.security.merge(&theirs.security);
ours.injection_flags
.extend(theirs.injection_flags.iter().cloned());
}
pub(super) fn deny_reason_codes(
&self,
) -> Vec<lean_ctx_protocol::context_gateway::ReasonCodeV1> {
let mut codes = Vec::new();
for record in self
.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
.filter(|record| record.decision.disposition == ContextDispositionV1::Deny)
{
for code in &record.decision.reason_codes {
if !codes.contains(code) {
codes.push(code.clone());
}
}
}
codes
}
pub fn flagged_injection(&self) -> bool {
self.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
.any(|record| record.counts.injection > 0)
}
pub(crate) fn security_tally(&self) -> crate::core::security_events::SecurityCounts {
let mut tally = self
.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner)
.security;
for record in self
.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
{
tally.merge(&record.security());
}
tally
}
fn injection_flags(&self) -> Vec<crate::core::security_events::InjectionFlag> {
let mut flags = self
.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner)
.injection_flags
.clone();
flags.extend(
self.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.iter()
.filter_map(Recorded::injection_flag),
);
flags
}
pub(crate) fn record_security(
&self,
tool: &str,
agent_id: &str,
) -> crate::core::security_events::SecurityCounts {
use crate::core::security_events::{self, SecurityKind};
let mut tally = self.security_tally();
tally.injection_flagged = 0;
security_events::record(tool, agent_id, &tally);
let flagged = security_events::record_injections(tool, agent_id, &self.injection_flags());
tally.add(SecurityKind::InjectionFlagged, flagged);
tally
}
}
pub(super) const NOTES_NOT_ITEMIZED: &str = "recovery.notes_not_itemized";
pub(crate) const NO_SOURCE_DECISION: &str = "delivery.no_source_decision";
fn mark_notes_not_itemized(records: &mut [Recorded]) -> Option<Recorded> {
if records.iter().any(Recorded::states_notes_not_itemized) {
return None;
}
let marker = Recorded {
decision: ContextDecisionV1 {
object: note_object(),
disposition: ContextDispositionV1::Allow,
reason_codes: vec![
ReasonCodeV1::new(NOTES_NOT_ITEMIZED).expect("built-in reason codes are valid"),
],
signals: Vec::new(),
required_transformations: Vec::new(),
},
counts: AdmissionCounts::default(),
external: false,
};
let slot = records
.iter()
.rposition(Recorded::is_note)
.or_else(|| {
records
.iter()
.rposition(|record| record.decision.disposition.delivers_content())
})
.unwrap_or(records.len().saturating_sub(1));
let displaced = std::mem::replace(records.get_mut(slot)?, marker);
(!displaced.is_note()).then_some(displaced)
}
impl Recorded {
fn injection_flag(&self) -> Option<crate::core::security_events::InjectionFlag> {
(self.counts.injection > 0).then(|| crate::core::security_events::InjectionFlag {
object_sha256: self.decision.object.hex().to_owned(),
patterns: self.counts.injection_patterns,
})
}
fn is_clean(&self) -> bool {
self.decision.disposition == ContextDispositionV1::Allow
&& self.decision.reason_codes.is_empty()
&& !self.external
}
fn is_note(&self) -> bool {
self.decision.object == note_object()
}
fn states_notes_not_itemized(&self) -> bool {
self.is_note()
&& self
.decision
.reason_codes
.iter()
.any(|reason| reason.as_str() == NOTES_NOT_ITEMIZED)
}
fn security(&self) -> crate::core::security_events::SecurityCounts {
use crate::core::security_events::{SecurityCounts, SecurityKind};
let mut tally = SecurityCounts::default();
let has = |code: &str| {
self.decision
.reason_codes
.iter()
.any(|reason| reason.as_str() == code)
};
if has("secret.redacted") {
tally.add(SecurityKind::SecretRedacted, u64::from(self.counts.secrets));
}
if has("pii.redacted") {
tally.add(SecurityKind::PiiRedacted, u64::from(self.counts.pii));
}
if self.counts.injection > 0 {
tally.add(
SecurityKind::InjectionFlagged,
u64::from(self.counts.injection),
);
}
if self.decision.disposition == ContextDispositionV1::Deny {
tally.add(SecurityKind::ContentWithheld, 1);
} else if self.counts.incomplete > 0 {
tally.add(SecurityKind::CoverageIncomplete, 1);
}
tally
}
fn redactions(&self) -> u32 {
if self.decision.disposition == ContextDispositionV1::AllowRedacted {
self.counts.secrets.saturating_add(self.counts.pii)
} else {
0
}
}
}
#[derive(Debug, Default)]
struct Overflow {
objects: u32,
permitted: u32,
blocked: u32,
redactions: u32,
injection: u32,
incomplete: u32,
security: crate::core::security_events::SecurityCounts,
injection_flags: Vec<crate::core::security_events::InjectionFlag>,
}
impl Overflow {
fn add(&mut self, record: &Recorded) {
if record.decision.object == note_object() {
return;
}
let disposition = record.decision.disposition;
self.objects = self.objects.saturating_add(1);
if disposition.delivers_content() {
self.permitted = self.permitted.saturating_add(1);
}
if disposition == ContextDispositionV1::Deny {
self.blocked = self.blocked.saturating_add(1);
}
self.redactions = self.redactions.saturating_add(record.redactions());
self.injection = self.injection.saturating_add(record.counts.injection);
if record.counts.incomplete > 0 {
self.incomplete = self.incomplete.saturating_add(1);
}
self.security.merge(&record.security());
self.injection_flags.extend(record.injection_flag());
}
}
pub struct CallIdentity<'a> {
pub agent_id: Option<&'a str>,
pub destination: Option<DestinationV1>,
}
pub struct Delivered<'a> {
pub text: &'a str,
pub is_error: bool,
pub tokens: u64,
}
fn reference(prefix: &str, raw: &str) -> Option<ProtocolReference> {
ProtocolReference::new(format!("{prefix}{raw}"))
.ok()
.or_else(|| {
let digest = blake3::hash(raw.as_bytes()).to_hex();
ProtocolReference::new(format!("{prefix}{}", &digest[..32])).ok()
})
}
fn sha256(bytes: &[u8]) -> Sha256Digest {
use sha2::{Digest, Sha256};
let hex = crate::core::agent_identity::hex_encode(&Sha256::digest(bytes));
Sha256Digest::new(format!("sha256:{hex}")).expect("SHA-256 digest is canonical")
}
fn saturating_u32(n: usize) -> u32 {
u32::try_from(n).unwrap_or(u32::MAX)
}
pub(super) fn note_object() -> Sha256Digest {
sha256(b"proxy-block-kept-whole")
}
pub fn is_note(decision: &ContextDecisionV1) -> bool {
let note = note_object();
decision.object == note
|| super::receipt_store::keyed::object_key()
.is_ok_and(|key| decision.object == key.object(¬e))
}
#[cfg(any(test, feature = "test-support"))]
pub fn finish(
capture: &AdmissionCapture,
identity: &CallIdentity<'_>,
delivered: &Delivered<'_>,
) -> Option<ContextDecisionReceiptV1> {
finish_for_presenter(capture, identity, None, delivered)
}
pub(crate) fn finish_for_presenter(
capture: &AdmissionCapture,
identity: &CallIdentity<'_>,
presenter_ref: Option<&str>,
delivered: &Delivered<'_>,
) -> Option<ContextDecisionReceiptV1> {
let records = capture
.records
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone();
if records.is_empty() {
return None;
}
let overflow = capture
.overflow
.lock()
.unwrap_or_else(PoisonError::into_inner);
let (mode, policy_digest) = capture
.policy
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()?;
let decisions: Vec<ContextDecisionV1> = records
.iter()
.map(|record| record.decision.clone())
.collect();
let note = note_object();
let notes = decisions
.iter()
.filter(|decision| decision.object == note)
.count();
if notes == decisions.len() && overflow.objects == 0 {
return None;
}
let count = |disposition: ContextDispositionV1| {
decisions
.iter()
.filter(|decision| decision.disposition == disposition)
.count()
};
let blocked = count(ContextDispositionV1::Deny);
let quarantined = count(ContextDispositionV1::Quarantine);
let permitted = saturating_u32(
decisions
.iter()
.filter(|decision| decision.object != note && decision.disposition.delivers_content())
.count(),
)
.saturating_add(overflow.permitted);
let redactions: u32 = records
.iter()
.map(Recorded::redactions)
.fold(overflow.redactions, u32::saturating_add);
let injection: u32 = records
.iter()
.map(|record| record.counts.injection)
.fold(overflow.injection, u32::saturating_add);
let incomplete = saturating_u32(
records
.iter()
.filter(|record| record.counts.incomplete > 0)
.count(),
)
.saturating_add(overflow.incomplete);
let object_key = super::receipt_store::keyed::object_key().ok()?;
let decisions: Vec<ContextDecisionV1> = decisions
.into_iter()
.map(|mut decision| {
decision.object = object_key.object(&decision.object);
decision
})
.collect();
let outcome = if permitted == 0 {
DeliveryOutcomeV1::Withheld
} else if delivered.is_error {
DeliveryOutcomeV1::Failed
} else {
DeliveryOutcomeV1::Delivered
};
let final_context =
(outcome == DeliveryOutcomeV1::Delivered).then(|| sha256(delivered.text.as_bytes()));
let principal = presenter_ref
.and_then(|presenter| ProtocolReference::new(presenter).ok())
.and_then(|id| PrincipalV1::new(PrincipalKindV1::Session, id).ok())
.or_else(|| {
identity
.agent_id
.and_then(|agent| reference("agent:", agent))
.and_then(|id| PrincipalV1::new(PrincipalKindV1::Agent, id).ok())
})
.unwrap_or_else(PrincipalV1::unknown);
let receipt_seed = format!(
"{}|{}|{}|{:?}",
policy_digest.as_str(),
decisions
.iter()
.map(|decision| decision.object.as_str())
.collect::<Vec<_>>()
.join(","),
final_context.as_ref().map_or("", Sha256Digest::as_str),
capture.started
);
let receipt = ContextDecisionReceiptV1 {
schema_version: RECEIPT_V2_SCHEMA_VERSION,
object_digest: Some(object_key.descriptor()),
receipt_id: reference(
"rcpt-",
&blake3::hash(receipt_seed.as_bytes()).to_hex()[..32],
)?,
mode,
principal,
task: capture
.task
.lock()
.unwrap_or_else(PoisonError::into_inner)
.as_ref()
.and_then(|(task, _)| TaskId::new(task.as_str()).ok()),
destination: identity
.destination
.clone()
.unwrap_or_else(|| DestinationV1 {
provider: ProtocolReference::new("mcp-host").expect("valid provider ref"),
model: None,
locality: DestinationLocalityV1::Unknown,
organization_managed: false,
account_ref: None,
region: None,
}),
policy: Some(PolicyRefV1 {
id: PolicyId::new("context-gateway").expect("valid policy id"),
version: None,
digest: policy_digest,
}),
sources: SourceCountsV1 {
inspected: saturating_u32(decisions.len() - notes).saturating_add(overflow.objects),
permitted,
selected: permitted,
blocked: saturating_u32(blocked).saturating_add(overflow.blocked),
},
decisions,
security: SecurityCountsV1 {
redactions,
blocked_objects: saturating_u32(blocked),
quarantined_objects: saturating_u32(quarantined),
injection_signals: injection,
incomplete_coverage: incomplete,
},
tokens: TokenAccountV1 {
original: capture
.original_tokens
.lock()
.unwrap_or_else(PoisonError::into_inner)
.filter(|tokens| *tokens > 0)
.unwrap_or(delivered.tokens),
delivered: delivered.tokens,
},
final_context,
outcome,
duration_us: u64::try_from(capture.started.elapsed().as_micros()).unwrap_or(u64::MAX),
quality: None,
};
receipt.validate().ok()?;
Some(receipt)
}