aster_forge_webdav/
multistatus.rs

1//! Bounded incremental RFC 4918 Multi-Status response writing.
2
3use std::collections::{BTreeMap, VecDeque};
4use std::io::{self, Write};
5use std::pin::Pin;
6use std::task::{Context, Poll};
7
8use aster_forge_xml::{Error as ForgeXmlError, XmlSafetyError, XmlStreamWriter, XmlWriteOptions};
9use bytes::{Bytes, BytesMut};
10use futures::Stream;
11use http::header::CONTENT_TYPE;
12use http::{HeaderValue, StatusCode};
13
14use crate::xml::write_element;
15use crate::xml_response::error_condition_parts;
16use crate::{
17    DavBackendError, DavCancellationToken, DavErrorCondition, DavMultiStatusItem, DavPropStat,
18    DavResponse, DavResponseBody,
19};
20
21const DAV_NAMESPACE: &str = "DAV:";
22const DEFAULT_MAXIMUM_OUTPUT_BYTES: usize = 64 * 1024 * 1024;
23const DEFAULT_MAXIMUM_ITEMS: usize = 100_000;
24const DEFAULT_MAXIMUM_PROPERTIES_PER_ITEM: usize = 4_096;
25const DEFAULT_CHUNK_BYTES: usize = 16 * 1024;
26
27/// Product-configured resource limits for a Multi-Status response.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub struct DavMultiStatusLimits {
30    pub maximum_output_bytes: usize,
31    pub maximum_items: usize,
32    pub maximum_properties_per_item: usize,
33    pub chunk_bytes: usize,
34}
35
36impl DavMultiStatusLimits {
37    #[must_use]
38    pub const fn new(
39        maximum_output_bytes: usize,
40        maximum_items: usize,
41        maximum_properties_per_item: usize,
42        chunk_bytes: usize,
43    ) -> Self {
44        Self {
45            maximum_output_bytes,
46            maximum_items,
47            maximum_properties_per_item,
48            chunk_bytes,
49        }
50    }
51
52    fn validate(self) -> Result<(), DavMultiStatusError> {
53        if self.maximum_output_bytes == 0
54            || self.maximum_items == 0
55            || self.maximum_properties_per_item == 0
56            || self.chunk_bytes == 0
57        {
58            Err(DavMultiStatusError::new(
59                DavMultiStatusErrorKind::InvalidLimits,
60                DavMultiStatusProgress::default(),
61            ))
62        } else {
63            Ok(())
64        }
65    }
66}
67
68impl Default for DavMultiStatusLimits {
69    fn default() -> Self {
70        Self::new(
71            DEFAULT_MAXIMUM_OUTPUT_BYTES,
72            DEFAULT_MAXIMUM_ITEMS,
73            DEFAULT_MAXIMUM_PROPERTIES_PER_ITEM,
74            DEFAULT_CHUNK_BYTES,
75        )
76    }
77}
78
79/// Progress retained when bounded response generation stops.
80#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
81pub struct DavMultiStatusProgress {
82    pub response_started: bool,
83    pub emitted_items: usize,
84    pub emitted_bytes: usize,
85}
86
87/// Stable Multi-Status failure classification.
88#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
89pub enum DavMultiStatusErrorKind {
90    #[error("invalid Multi-Status resource limits")]
91    InvalidLimits,
92    #[error("Multi-Status item limit exceeded")]
93    ItemLimitExceeded,
94    #[error("Multi-Status property limit exceeded")]
95    PropertyLimitExceeded,
96    #[error("invalid Multi-Status response item")]
97    InvalidItem,
98    #[error("Multi-Status output byte limit exceeded")]
99    OutputLimitExceeded,
100    #[error("Multi-Status source was cancelled")]
101    Cancelled,
102    #[error(transparent)]
103    Backend(#[from] DavBackendError),
104    #[error("Multi-Status XML is malformed")]
105    Xml,
106    #[error("Multi-Status output sink failed")]
107    Write,
108}
109
110/// Multi-Status failure plus the exact response progress at the failure boundary.
111#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
112#[error("{kind}")]
113pub struct DavMultiStatusError {
114    pub kind: DavMultiStatusErrorKind,
115    pub progress: DavMultiStatusProgress,
116}
117
118impl DavMultiStatusError {
119    const fn new(kind: DavMultiStatusErrorKind, progress: DavMultiStatusProgress) -> Self {
120        Self { kind, progress }
121    }
122}
123
124/// Failure produced by the product-owned item source before XML composition.
125#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
126pub enum DavMultiStatusSourceError {
127    #[error(transparent)]
128    Backend(#[from] DavBackendError),
129    #[error("Multi-Status source was cancelled")]
130    Cancelled,
131}
132
133/// Type-erased incremental Multi-Status response stream.
134pub type DavMultiStatusStream =
135    Pin<Box<dyn Stream<Item = Result<Bytes, DavMultiStatusError>> + Send + 'static>>;
136
137/// Stateful writer for one RFC 4918 `<D:multistatus>` document.
138pub struct DavMultiStatusWriter<W: Write> {
139    writer: XmlStreamWriter<TrackingWriter<W>>,
140    inherited_namespaces: BTreeMap<String, String>,
141    limits: DavMultiStatusLimits,
142    emitted_items: usize,
143}
144
145impl<W: Write> DavMultiStatusWriter<W> {
146    ///
147    /// # Errors
148    ///
149    /// Returns [`DavMultiStatusError`] when limits are zero or the XML prologue exceeds them.
150    pub fn new(inner: W, limits: DavMultiStatusLimits) -> Result<Self, DavMultiStatusError> {
151        limits.validate()?;
152        let options = XmlWriteOptions::new().max_output_bytes(limits.maximum_output_bytes);
153        let tracking = TrackingWriter::new(inner);
154        let mut writer = XmlStreamWriter::with_options(tracking, options)
155            .map_err(|error| map_writer_error(&error, DavMultiStatusProgress::default()))?;
156        if let Err(error) = writer.start_element("D:multistatus", [("xmlns:D", DAV_NAMESPACE)]) {
157            return Err(map_writer_error(&error, writer_progress(&writer, 0)));
158        }
159        let mut inherited_namespaces = BTreeMap::new();
160        inherited_namespaces.insert("D".to_owned(), DAV_NAMESPACE.to_owned());
161        Ok(Self {
162            writer,
163            inherited_namespaces,
164            limits,
165            emitted_items: 0,
166        })
167    }
168
169    ///
170    /// # Errors
171    ///
172    /// Returns [`DavMultiStatusError`] when an item or cumulative output exceeds a limit.
173    pub fn append(&mut self, item: DavMultiStatusItem) -> Result<(), DavMultiStatusError> {
174        let next_items = self
175            .emitted_items
176            .checked_add(1)
177            .ok_or_else(|| self.error(DavMultiStatusErrorKind::ItemLimitExceeded))?;
178        if next_items > self.limits.maximum_items {
179            return Err(self.error(DavMultiStatusErrorKind::ItemLimitExceeded));
180        }
181        validate_item(&item, self.limits.maximum_properties_per_item)
182            .map_err(|kind| self.error(kind))?;
183
184        if let Err(error) = write_response_item(&mut self.writer, &self.inherited_namespaces, item)
185        {
186            return Err(map_writer_error(&error, self.progress()));
187        }
188        self.emitted_items = next_items;
189        Ok(())
190    }
191
192    /// Returns the number of bytes successfully written to the underlying sink.
193    #[must_use]
194    pub fn written_bytes(&self) -> usize {
195        self.writer.get_ref().written
196    }
197
198    /// Returns a mutable reference to the underlying sink without finishing the document.
199    ///
200    /// Writing to the sink directly bypasses XML state and byte accounting and can corrupt the
201    /// document. This access is intended only for draining sink-managed completed chunks.
202    pub fn get_mut(&mut self) -> &mut W {
203        &mut self.writer.get_mut().inner
204    }
205
206    ///
207    /// # Errors
208    ///
209    /// Returns [`DavMultiStatusError`] when no item was written or the closing XML exceeds limits.
210    pub fn finish(mut self) -> Result<W, DavMultiStatusError> {
211        if let Err(error) = self.writer.end_element() {
212            return Err(map_writer_error(&error, self.progress()));
213        }
214        let progress = self.progress();
215        self.writer
216            .finish()
217            .map(|tracking| tracking.inner)
218            .map_err(|error| map_writer_error(&error, progress))
219    }
220
221    fn progress(&self) -> DavMultiStatusProgress {
222        writer_progress(&self.writer, self.emitted_items)
223    }
224
225    fn error(&self, kind: DavMultiStatusErrorKind) -> DavMultiStatusError {
226        DavMultiStatusError::new(kind, self.progress())
227    }
228}
229
230/// Serializes a complete bounded Multi-Status document through the incremental writer contract.
231///
232/// # Errors
233///
234/// Returns [`DavMultiStatusError`] when an item is invalid or any output limit is exceeded.
235pub fn dav_multistatus_bytes(
236    items: impl IntoIterator<Item = DavMultiStatusItem>,
237    limits: DavMultiStatusLimits,
238) -> Result<Vec<u8>, DavMultiStatusError> {
239    let mut writer = DavMultiStatusWriter::new(Vec::new(), limits)?;
240    for item in items {
241        writer.append(item)?;
242    }
243    writer.finish()
244}
245
246/// Creates a transport-neutral streaming 207 response from a product-owned item stream.
247///
248/// # Errors
249///
250/// Returns [`DavMultiStatusError`] when limits are invalid or response headers cannot be encoded.
251pub fn multistatus_stream_response<S>(
252    source: S,
253    limits: DavMultiStatusLimits,
254) -> Result<DavResponse, DavMultiStatusError>
255where
256    S: Stream<Item = Result<DavMultiStatusItem, DavMultiStatusSourceError>> + Send + 'static,
257{
258    multistatus_stream_response_inner(source, limits, None)
259}
260
261/// Creates a streaming 207 response that cancels shared product work when its body is dropped.
262///
263/// Products should pass clones of `cancellation` to directory enumeration and property/lock
264/// preload operations. A transport-side client disconnect drops the response body and flips the
265/// same token, allowing those operations to stop at their next cancellation boundary.
266///
267/// # Errors
268///
269/// Returns [`DavMultiStatusError`] when limits are invalid or response headers cannot be encoded.
270pub fn multistatus_stream_response_with_cancellation<S>(
271    source: S,
272    limits: DavMultiStatusLimits,
273    cancellation: DavCancellationToken,
274) -> Result<DavResponse, DavMultiStatusError>
275where
276    S: Stream<Item = Result<DavMultiStatusItem, DavMultiStatusSourceError>> + Send + 'static,
277{
278    multistatus_stream_response_inner(source, limits, Some(cancellation))
279}
280
281fn multistatus_stream_response_inner<S>(
282    source: S,
283    limits: DavMultiStatusLimits,
284    cancellation: Option<DavCancellationToken>,
285) -> Result<DavResponse, DavMultiStatusError>
286where
287    S: Stream<Item = Result<DavMultiStatusItem, DavMultiStatusSourceError>> + Send + 'static,
288{
289    limits.validate()?;
290    let stream = StreamingMultiStatus::new(Box::pin(source), limits, cancellation);
291    let mut response = DavResponse {
292        status: StatusCode::MULTI_STATUS,
293        headers: http::HeaderMap::new(),
294        body: DavResponseBody::MultiStatus(Box::pin(stream)),
295    };
296    response.headers.insert(
297        CONTENT_TYPE,
298        HeaderValue::from_static("application/xml; charset=utf-8"),
299    );
300    Ok(response)
301}
302
303struct StreamingMultiStatus {
304    source: Pin<
305        Box<
306            dyn Stream<Item = Result<DavMultiStatusItem, DavMultiStatusSourceError>>
307                + Send
308                + 'static,
309        >,
310    >,
311    writer: Option<DavMultiStatusWriter<ChunkBuffer>>,
312    pending: VecDeque<Bytes>,
313    limits: DavMultiStatusLimits,
314    progress: DavMultiStatusProgress,
315    done: bool,
316    cancellation: Option<DavCancellationToken>,
317}
318
319impl StreamingMultiStatus {
320    fn new(
321        source: Pin<
322            Box<
323                dyn Stream<Item = Result<DavMultiStatusItem, DavMultiStatusSourceError>>
324                    + Send
325                    + 'static,
326            >,
327        >,
328        limits: DavMultiStatusLimits,
329        cancellation: Option<DavCancellationToken>,
330    ) -> Self {
331        Self {
332            source,
333            writer: None,
334            pending: VecDeque::new(),
335            limits,
336            progress: DavMultiStatusProgress::default(),
337            done: false,
338            cancellation,
339        }
340    }
341
342    fn new_writer(&self) -> Result<DavMultiStatusWriter<ChunkBuffer>, DavMultiStatusError> {
343        let buffer = ChunkBuffer::new(
344            self.limits
345                .chunk_bytes
346                .min(self.limits.maximum_output_bytes),
347        );
348        DavMultiStatusWriter::new(buffer, self.limits)
349    }
350
351    fn fail(
352        &mut self,
353        kind: DavMultiStatusErrorKind,
354    ) -> Poll<Option<Result<Bytes, DavMultiStatusError>>> {
355        self.done = true;
356        self.pending.clear();
357        self.writer = None;
358        Poll::Ready(Some(Err(DavMultiStatusError::new(kind, self.progress))))
359    }
360}
361
362impl Drop for StreamingMultiStatus {
363    fn drop(&mut self) {
364        if let Some(cancellation) = &self.cancellation {
365            cancellation.cancel();
366        }
367    }
368}
369
370impl Stream for StreamingMultiStatus {
371    type Item = Result<Bytes, DavMultiStatusError>;
372
373    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
374        loop {
375            if let Some(chunk) = self.pending.pop_front() {
376                self.progress.response_started = true;
377                self.progress.emitted_bytes =
378                    self.progress.emitted_bytes.saturating_add(chunk.len());
379                return Poll::Ready(Some(Ok(chunk)));
380            }
381            if self.done {
382                return Poll::Ready(None);
383            }
384
385            match self.source.as_mut().poll_next(context) {
386                Poll::Pending => return Poll::Pending,
387                Poll::Ready(Some(Err(DavMultiStatusSourceError::Backend(error)))) => {
388                    return self.fail(DavMultiStatusErrorKind::Backend(error));
389                }
390                Poll::Ready(Some(Err(DavMultiStatusSourceError::Cancelled))) => {
391                    return self.fail(DavMultiStatusErrorKind::Cancelled);
392                }
393                Poll::Ready(Some(Ok(item))) => {
394                    let mut writer = match self.writer.take() {
395                        Some(writer) => writer,
396                        None => match self.new_writer() {
397                            Ok(writer) => writer,
398                            Err(mut error) => {
399                                error.progress = self.progress;
400                                self.done = true;
401                                return Poll::Ready(Some(Err(error)));
402                            }
403                        },
404                    };
405                    let result = writer.append(item);
406                    if let Err(mut error) = result {
407                        error.progress.response_started = self.progress.response_started;
408                        error.progress.emitted_bytes = self.progress.emitted_bytes;
409                        self.done = true;
410                        self.writer = None;
411                        return Poll::Ready(Some(Err(error)));
412                    }
413                    self.progress.emitted_items = self.progress.emitted_items.saturating_add(1);
414                    self.pending.append(&mut writer.get_mut().take_chunks());
415                    self.writer = Some(writer);
416                }
417                Poll::Ready(None) => {
418                    let writer = match self.writer.take() {
419                        Some(writer) => writer,
420                        None => match self.new_writer() {
421                            Ok(writer) => writer,
422                            Err(mut error) => {
423                                error.progress = self.progress;
424                                self.done = true;
425                                return Poll::Ready(Some(Err(error)));
426                            }
427                        },
428                    };
429                    match writer.finish() {
430                        Ok(mut buffer) => {
431                            self.pending.append(&mut buffer.take_chunks());
432                            self.done = true;
433                        }
434                        Err(mut error) => {
435                            error.progress.response_started = self.progress.response_started;
436                            error.progress.emitted_bytes = self.progress.emitted_bytes;
437                            self.done = true;
438                            return Poll::Ready(Some(Err(error)));
439                        }
440                    }
441                }
442            }
443        }
444    }
445}
446
447fn validate_item(
448    item: &DavMultiStatusItem,
449    maximum_properties: usize,
450) -> Result<(), DavMultiStatusErrorKind> {
451    let property_count = item.propstats.iter().try_fold(0usize, |count, propstat| {
452        count.checked_add(propstat.properties.len())
453    });
454    if property_count.is_none_or(|count| count > maximum_properties) {
455        return Err(DavMultiStatusErrorKind::PropertyLimitExceeded);
456    }
457    if item
458        .status
459        .is_some_and(|status| StatusCode::from_u16(status).is_err())
460        || item
461            .propstats
462            .iter()
463            .any(|propstat| StatusCode::from_u16(propstat.status).is_err())
464    {
465        return Err(DavMultiStatusErrorKind::InvalidItem);
466    }
467    let property_form = item.status.is_none() && !item.propstats.is_empty();
468    let status_form = item.status.is_some() && item.propstats.is_empty();
469    if item.href.is_empty() || !(property_form || status_form) {
470        return Err(DavMultiStatusErrorKind::InvalidItem);
471    }
472    Ok(())
473}
474
475fn write_response_item<W: Write>(
476    writer: &mut XmlStreamWriter<TrackingWriter<W>>,
477    inherited_namespaces: &BTreeMap<String, String>,
478    item: DavMultiStatusItem,
479) -> Result<(), ForgeXmlError> {
480    writer.start("D:response")?;
481    write_text_element(writer, "D:href", &item.href)?;
482    for propstat in &item.propstats {
483        write_propstat(writer, inherited_namespaces, propstat)?;
484    }
485    if let Some(status) = item.status {
486        write_status(writer, status)?;
487    }
488    if let Some(error) = item.error {
489        write_error(writer, &error)?;
490    }
491    writer.end_element()
492}
493
494fn write_propstat<W: Write>(
495    writer: &mut XmlStreamWriter<TrackingWriter<W>>,
496    inherited_namespaces: &BTreeMap<String, String>,
497    propstat: &DavPropStat,
498) -> Result<(), ForgeXmlError> {
499    writer.start("D:propstat")?;
500    writer.start("D:prop")?;
501    for property in &propstat.properties {
502        write_element(writer, property, inherited_namespaces)?;
503    }
504    writer.end_element()?;
505    write_status(writer, propstat.status)?;
506    writer.end_element()
507}
508
509fn write_status<W: Write>(
510    writer: &mut XmlStreamWriter<TrackingWriter<W>>,
511    status: u16,
512) -> Result<(), ForgeXmlError> {
513    let status = StatusCode::from_u16(status)
514        .map_err(|_| ForgeXmlError::InvalidData("invalid HTTP status code".to_owned()))?;
515    let line = format!(
516        "HTTP/1.1 {} {}",
517        status.as_u16(),
518        status.canonical_reason().unwrap_or("Unknown"),
519    );
520    write_text_element(writer, "D:status", &line)
521}
522
523fn write_error<W: Write>(
524    writer: &mut XmlStreamWriter<TrackingWriter<W>>,
525    error: &DavErrorCondition,
526) -> Result<(), ForgeXmlError> {
527    let (name, href) = error_condition_parts(error);
528    writer.start("D:error")?;
529    if let Some(href) = href {
530        writer.start(&format!("D:{name}"))?;
531        write_text_element(writer, "D:href", href)?;
532        writer.end_element()?;
533    } else {
534        writer.empty(&format!("D:{name}"))?;
535    }
536    writer.end_element()
537}
538
539fn write_text_element<W: Write>(
540    writer: &mut XmlStreamWriter<TrackingWriter<W>>,
541    name: &str,
542    text: &str,
543) -> Result<(), ForgeXmlError> {
544    writer.start(name)?;
545    writer.text(text)?;
546    writer.end_element()
547}
548
549fn writer_progress<W: Write>(
550    writer: &XmlStreamWriter<TrackingWriter<W>>,
551    emitted_items: usize,
552) -> DavMultiStatusProgress {
553    let emitted_bytes = writer.get_ref().written;
554    DavMultiStatusProgress {
555        response_started: emitted_bytes != 0,
556        emitted_items,
557        emitted_bytes,
558    }
559}
560
561fn map_writer_error(
562    error: &ForgeXmlError,
563    progress: DavMultiStatusProgress,
564) -> DavMultiStatusError {
565    let kind = match error {
566        ForgeXmlError::Safety(XmlSafetyError::OutputTooLarge) => {
567            DavMultiStatusErrorKind::OutputLimitExceeded
568        }
569        ForgeXmlError::Safety(_) => DavMultiStatusErrorKind::Xml,
570        ForgeXmlError::InvalidXml(_) | ForgeXmlError::InvalidData(_) => {
571            DavMultiStatusErrorKind::Xml
572        }
573        ForgeXmlError::Io(_) => DavMultiStatusErrorKind::Write,
574    };
575    DavMultiStatusError::new(kind, progress)
576}
577
578struct TrackingWriter<W> {
579    inner: W,
580    written: usize,
581}
582
583impl<W> TrackingWriter<W> {
584    const fn new(inner: W) -> Self {
585        Self { inner, written: 0 }
586    }
587}
588
589impl<W: Write> Write for TrackingWriter<W> {
590    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
591        let written = self.inner.write(buffer)?;
592        self.written = self
593            .written
594            .checked_add(written)
595            .ok_or_else(|| io::Error::other("Multi-Status byte count overflow"))?;
596        Ok(written)
597    }
598
599    fn flush(&mut self) -> io::Result<()> {
600        self.inner.flush()
601    }
602}
603
604struct ChunkBuffer {
605    chunk_bytes: usize,
606    current: BytesMut,
607    ready: VecDeque<Bytes>,
608}
609
610impl ChunkBuffer {
611    fn new(chunk_bytes: usize) -> Self {
612        Self {
613            chunk_bytes,
614            current: BytesMut::with_capacity(chunk_bytes.min(DEFAULT_CHUNK_BYTES)),
615            ready: VecDeque::new(),
616        }
617    }
618
619    fn take_chunks(&mut self) -> VecDeque<Bytes> {
620        if !self.current.is_empty() {
621            self.ready.push_back(self.current.split().freeze());
622        }
623        std::mem::take(&mut self.ready)
624    }
625}
626
627impl Write for ChunkBuffer {
628    fn write(&mut self, mut buffer: &[u8]) -> io::Result<usize> {
629        let input_len = buffer.len();
630        while !buffer.is_empty() {
631            let remaining = self.chunk_bytes - self.current.len();
632            let take = remaining.min(buffer.len());
633            self.current.extend_from_slice(&buffer[..take]);
634            buffer = &buffer[take..];
635            if self.current.len() == self.chunk_bytes {
636                self.ready.push_back(self.current.split().freeze());
637            }
638        }
639        Ok(input_len)
640    }
641
642    fn flush(&mut self) -> io::Result<()> {
643        Ok(())
644    }
645}
646
647#[cfg(test)]
648mod tests {
649    use super::*;
650
651    #[test]
652    fn status_writer_rejects_an_unvalidated_invalid_code() {
653        let tracking = TrackingWriter::new(Vec::new());
654        let mut writer = XmlStreamWriter::new(tracking).expect("writer");
655        writer.start("root").expect("root");
656        assert!(matches!(
657            write_status(&mut writer, 99),
658            Err(ForgeXmlError::InvalidData(_))
659        ));
660    }
661}
aster_forge_webdav/multistatus.rs

aster_forge_webdav/
multistatus.rs
