diff --git a/crates/didbot-pds/src/policy.rs b/crates/didbot-pds/src/policy.rs index c7395d03..eee82ed0 100644 --- a/crates/didbot-pds/src/policy.rs +++ b/crates/didbot-pds/src/policy.rs @@ -196,7 +196,8 @@ pub fn token_subject<'a>( /// Builds the [`didbot_policy::Subject::Blob`] a [`PolicyGate`] judges one /// upload against. `sniffed` and `size` are what is known at this point — -/// an upload is judged when it starts and again once its bytes are in. +/// an upload is judged when it starts, while its bytes arrive, and once +/// they are in. #[allow(clippy::too_many_arguments)] #[must_use] pub fn blob_subject<'a>( diff --git a/crates/didbot-pds/src/provision/mod.rs b/crates/didbot-pds/src/provision/mod.rs index 226fb330..7a0cf406 100644 --- a/crates/didbot-pds/src/provision/mod.rs +++ b/crates/didbot-pds/src/provision/mod.rs @@ -790,27 +790,66 @@ impl PolicyGate for FreezeSyncingGate<'_, S> { /// firehose frame needs as its `prev`. type AppliedBatch = (Vec, Committed, Vec>); +/// How much of an upload's body [`JudgedUpload`] waits for before it judges +/// the upload again. Each judgment after that waits for what has arrived to +/// double. +/// +/// Past [`crate::blobs::SNIFF_BYTES`], so each of these judgments has the +/// type read from the bytes. +const FIRST_BLOB_CHECKPOINT: u64 = 64 * 1024; +const _: () = assert!(FIRST_BLOB_CHECKPOINT >= crate::blobs::SNIFF_BYTES as u64); + /// An upload [`Registry::begin_blob`] hands out: judged under `blob.write` -/// once its bytes have arrived, filed under the type those bytes carry, and -/// stored only if policy allows it. +/// while its bytes arrive and once they have, filed under the type those +/// bytes carry, and stored only if policy allows it. /// /// The first judgment is [`Registry::begin_blob`]'s own, on what the request /// claims, so a body a policy will refuse for its declared type or length is -/// refused before it is read. The type read from the bytes, and the size -/// that actually arrived, are only known at [`BlobUpload::commit`], which -/// judges again with both. A refused upload is dropped, which discards it. +/// refused before it is read. [`BlobUpload::write`] judges again at +/// [`FIRST_BLOB_CHECKPOINT`] and each time what has arrived doubles, so a +/// body a policy refuses for its size stops soon after it passes the limit, +/// whatever length it declared. Those judgments can refuse the upload but +/// never admit it: the size that actually arrived is only known at +/// [`BlobUpload::commit`], which judges once more and decides. A refused +/// upload is dropped, which discards it. struct JudgedUpload<'a, S> { provisioner: &'a Provisioner, account: HostedAccount, /// The application whose OAuth token presented the upload. client_id: Option, mime_type: String, + /// The length the request declared, if it declared one. + declared: Option, size: u64, + /// How much must have arrived before the next judgment. + checkpoint: u64, /// The first bytes of the body, which the type is read from. head: Vec, inner: Box, } +impl JudgedUpload<'_, S> { + /// Judges the upload as `size` bytes whose first bytes carry + /// `signature`. See [`Provisioner::judge_blob`] for `arrived`. + fn judge(&self, signature: Option<&str>, size: u64, arrived: bool) -> Result<(), BlobError> { + self.provisioner + .judge_blob( + &self.account, + self.client_id.as_ref(), + &self.mime_type, + Some(signature.unwrap_or(crate::blobs::DEFAULT_MIME_TYPE)), + Some(size), + arrived, + ) + .map_err(|err| match err { + ProvisionError::PolicyRejected { reason } => BlobError::PolicyRejected { reason }, + ProvisionError::PolicyFrozen { reason } => BlobError::PolicyFrozen { reason }, + // `judge_blob` answers with those two and nothing else. + other => BlobError::Backend(other.to_string()), + }) + } +} + impl BlobUpload for JudgedUpload<'_, S> where S: AccountStore, @@ -820,6 +859,22 @@ where self.size += chunk.len() as u64; let room = crate::blobs::SNIFF_BYTES.saturating_sub(self.head.len()); self.head.extend_from_slice(&chunk[..room.min(chunk.len())]); + if self.size >= self.checkpoint { + self.checkpoint = self.size.saturating_mul(2); + // A declared length is what the first judgment was asked about, + // and what the body comes to if it is honest. With none, the + // bytes so far are the least the blob will be. + let size = self + .declared + .map_or(self.size, |declared| declared.max(self.size)); + tracing::debug!( + did = %self.account.did, + size, + received = self.size, + "judging a blob upload while it arrives" + ); + self.judge(crate::blobs::sniff(&self.head), size, false)?; + } Ok(()) } @@ -847,20 +902,7 @@ where /// was true all along. fn commit(mut self: Box) -> Result { let signature = crate::blobs::sniff(&self.head); - if let Err(err) = self.provisioner.judge_blob( - &self.account, - self.client_id.as_ref(), - &self.mime_type, - Some(signature.unwrap_or(crate::blobs::DEFAULT_MIME_TYPE)), - Some(self.size), - ) { - return Err(match err { - ProvisionError::PolicyRejected { reason } => BlobError::PolicyRejected { reason }, - ProvisionError::PolicyFrozen { reason } => BlobError::PolicyFrozen { reason }, - // `judge_blob` answers with those two and nothing else. - other => BlobError::Backend(other.to_string()), - }); - } + self.judge(signature, self.size, true)?; let filed = crate::blobs::filed_type(&self.mime_type, signature); if filed != self.mime_type { let corrected = filed.to_owned(); diff --git a/crates/didbot-pds/src/provision/registry.rs b/crates/didbot-pds/src/provision/registry.rs index 12273e46..eaa3ca8c 100644 --- a/crates/didbot-pds/src/provision/registry.rs +++ b/crates/didbot-pds/src/provision/registry.rs @@ -1439,7 +1439,7 @@ where self.require_writable(&account)?; // What the request claims, before a byte is read: a policy that // refuses this type or this length refuses it here. - self.judge_blob(&account, client_id, mime_type, None, declared)?; + self.judge_blob(&account, client_id, mime_type, None, declared, false)?; let inner = self .blobs .begin(account.did.as_str(), mime_type, declared)?; @@ -1448,7 +1448,9 @@ where account, client_id: client_id.cloned(), mime_type: mime_type.to_owned(), + declared, size: 0, + checkpoint: FIRST_BLOB_CHECKPOINT, head: Vec::new(), inner, })) diff --git a/crates/didbot-pds/src/provision/write.rs b/crates/didbot-pds/src/provision/write.rs index cf005f7b..0755581d 100644 --- a/crates/didbot-pds/src/provision/write.rs +++ b/crates/didbot-pds/src/provision/write.rs @@ -703,10 +703,11 @@ where /// Judges one upload under `blob.write`. `sniffed` and `size` are what /// is known at this point — see [`didbot_policy::Subject::Blob`], which - /// an upload reaches twice. + /// an upload reaches before its body, while the body arrives, and once + /// it is in. `arrived` is true for the last of these. /// /// The attempt is observed only once, at the judgment that decides the - /// upload: the first one when it refuses, and the second otherwise. One + /// upload: the first one that refuses, or the last one otherwise. One /// upload is one attempt to a stateful evaluator counting them. pub(super) fn judge_blob( &self, @@ -715,6 +716,7 @@ where mime_type: &str, sniffed: Option<&str>, size: Option, + arrived: bool, ) -> Result<(), ProvisionError> { let pds_did = self.zone.service_did(); let subject = crate::policy::blob_subject( @@ -730,7 +732,7 @@ where ); let gate = self.freeze_syncing_gate(account); let outcome = gate.judge(&subject); - let decided = sniffed.is_some() || !matches!(outcome, Outcome::Allow); + let decided = arrived || !matches!(outcome, Outcome::Allow); if decided { gate.observe(&subject, &outcome); } @@ -742,6 +744,7 @@ where mime_type, sniffed, size, + arrived, %reason, "policy refused a blob upload" ); diff --git a/crates/didbot-policy-cedar/src/lib.rs b/crates/didbot-policy-cedar/src/lib.rs index 613247a9..e9a15476 100644 --- a/crates/didbot-policy-cedar/src/lib.rs +++ b/crates/didbot-policy-cedar/src/lib.rs @@ -251,9 +251,10 @@ impl Evaluator for CedarEvaluator { /// request built for it, whichever evaluator holds them. fn evaluate(&self, subject: &Subject<'_>, compiled: CompiledId) -> Result { let document = self.document(compiled)?; - // A blob is judged once when its upload starts and again once its - // bytes are in. `Blob` takes every fact as a `String` or a `Long`, - // so this engine judges the second, which carries them all. + // A blob is judged when its upload starts, before its bytes are + // read, and again while they arrive and once they are in. `Blob` + // takes every fact as a `String` or a `Long`, so this engine judges + // only the later ones, which carry them all. if matches!(subject, Subject::Blob { sniffed: None, .. }) { return Ok(Outcome::Allow); } diff --git a/crates/didbot-policy-cedar/src/shared.rs b/crates/didbot-policy-cedar/src/shared.rs index 6f003181..ae30e185 100644 --- a/crates/didbot-policy-cedar/src/shared.rs +++ b/crates/didbot-policy-cedar/src/shared.rs @@ -233,9 +233,9 @@ pub(crate) fn with_request(subject: &Subject<'_>, f: impl FnOnce(&Request<'_> now: universal.now, action: Action::BlobWrite { mime_type, - // Both are carried by the judgment this engine answers — + // Both are carried by every judgment this engine answers — // see `CedarEvaluator::evaluate`, which skips the one that - // runs before a blob's bytes are in. + // runs before a blob's bytes are read. sniffed: sniffed.unwrap_or_default(), size: size.unwrap_or_default(), space: None, diff --git a/crates/didbot-policy-records/src/subject.rs b/crates/didbot-policy-records/src/subject.rs index a9998daf..53e81054 100644 --- a/crates/didbot-policy-records/src/subject.rs +++ b/crates/didbot-policy-records/src/subject.rs @@ -241,8 +241,8 @@ pub fn write_subject<'a>( /// /// `sniffed` and `size` are what is known when the judgment happens: an /// upload is judged on what the request declares, before a byte is read, -/// and again once the bytes are in. `now` is the instant the request is -/// judged at. +/// again while the bytes arrive, and once they are in. `now` is the instant +/// the request is judged at. #[allow(clippy::too_many_arguments)] #[must_use] pub fn blob_subject<'a>( diff --git a/crates/didbot-policy-regex/src/lib.rs b/crates/didbot-policy-regex/src/lib.rs index 5f741aad..cbe331c6 100644 --- a/crates/didbot-policy-regex/src/lib.rs +++ b/crates/didbot-policy-regex/src/lib.rs @@ -619,8 +619,9 @@ enum CompiledPolicy { /// for the whole type — see `Fit`. Only applies to [`Subject::Blob`]. /// /// A condition over a fact the upload does not carry yet holds: an - /// upload is judged before its bytes arrive and again once they are in - /// (see [`Subject::Blob`]), and the second judgment has everything. + /// upload is judged before its bytes arrive, while they arrive, and once + /// they are in (see [`Subject::Blob`]), and the last judgment has + /// everything. DenyBlobUnless { mime_type: Vec, sniffed: Vec, diff --git a/crates/didbot-policy/src/subject.rs b/crates/didbot-policy/src/subject.rs index e72bbf5a..deaf6bc5 100644 --- a/crates/didbot-policy/src/subject.rs +++ b/crates/didbot-policy/src/subject.rs @@ -361,10 +361,11 @@ pub enum Subject<'a> { /// policy declares one, which is the ordinary case. lookups: &'a [Lookup<'a>], }, - /// A blob uploaded to this account, judged twice: when the upload - /// starts, on what the request claims, and again once the bytes are in - /// and before they are stored. A fact the first judgment does not have - /// is `None` — what is not known yet cannot deny. + /// A blob uploaded to this account. It is judged when the upload + /// starts, on what the request claims; at checkpoints while the body + /// arrives, each of which can refuse the upload but not admit it; and + /// once the bytes are in and before they are stored. A fact the first + /// judgment does not have is `None` — what is not known yet cannot deny. Blob { /// The account the blob is uploaded to. account: &'a str, @@ -377,12 +378,13 @@ pub enum Subject<'a> { client_id: Option<&'a ClientId>, /// The MIME type the uploader declared. mime_type: &'a str, - /// The MIME type read from the blob's own bytes. `None` until they - /// have arrived. + /// The MIME type read from the blob's own bytes. `None` before the + /// body is read. sniffed: Option<&'a str>, - /// The blob's size in bytes: what the request declared before the - /// bytes arrive, and what arrived after. `None` when the request - /// declared no length. + /// The blob's size in bytes. Before the body, the length the request + /// declared, or `None` when it declared none. While the body + /// arrives, that length, or the bytes received so far when there is + /// none. After, what arrived. size: Option, }, /// A pushed authorization request, judged on the application and the