mkit_server/connect/
service.rs

1//! `mkit.transport.v1.TransportService` over the pipeline: decode, one
2//! pipeline call inside [`send_wrap`], encode.
3
4use core::fmt;
5use std::sync::Arc;
6
7use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
8use buffa::Message as _;
9use bytes::Bytes;
10use connectrpc::{
11    ConnectError, InboundStream, RequestContext, Response, ServiceRequest, ServiceResult,
12    ServiceStream,
13};
14use futures::{StreamExt, stream};
15use mkit_core::hash::Hash;
16use mkit_core::protocol::{AdvanceOutcome, PackKey};
17
18use super::error::recorded;
19use super::proto::mkit::transport::v1::__buffa::oneof::begin_upload_response::Result as BeginResult;
20use super::proto::mkit::transport::v1::__buffa::oneof::download_pack_response::Body as DownloadBody;
21use super::proto::mkit::transport::v1::__buffa::oneof::issue_object_url_request::Target;
22use super::proto::mkit::transport::v1::__buffa::oneof::set_repo_visibility_request::Mode as VisibilityMode;
23use super::proto::mkit::transport::v1::__buffa::oneof::upload_pack_request::Body as UploadBody;
24use super::proto::mkit::transport::v1::__buffa::oneof::upload_part_request::Msg as PartMsg;
25use super::proto::mkit::transport::v1::{
26    AdvanceOutcome as WireOutcome, AdvanceRefsRequest, AdvanceRefsResponse, BeginUploadRequest,
27    BeginUploadResponse, CompleteUploadRequest, CompleteUploadResponse, DownloadPackHeader,
28    DownloadPackRequest, DownloadPackResponse, GetReceiptRequest, GetReceiptResponse,
29    GetServerInfoRequest, GetServerInfoResponse, ListRefsRequest, ListRefsResponse, PackChunk,
30    PackExistsRequest, PackExistsResponse, ReadRefRequest, ReadRefResponse, RefEntry,
31    RefExpectation, TransportService, UpdateRefRequest, UpdateRefResponse, UploadPackRequest,
32    UploadPackResponse, UploadPartRequest, UploadPartResponse, UploadTicket,
33};
34use super::proto::mkit::transport::v1::{
35    GetAuthorityGenerationRequest, GetAuthorityGenerationResponse, GetGrantEpochRequest,
36    GetGrantEpochResponse, IssueObjectUrlRequest, IssueObjectUrlResponse, RepoVisibility,
37    SetAuthorityGenerationRequest, SetAuthorityGenerationResponse, SetGrantEpochRequest,
38    SetGrantEpochResponse, SetRepoVisibilityRequest, SetRepoVisibilityResponse,
39};
40use super::{Shared, authenticated};
41use crate::error::ServerError;
42use crate::op::RefUpdate;
43use crate::pipeline::{
44    Authenticated, DownloadChunk, HookSet, Pipeline, ServerInfo, VisibilityRequest,
45};
46use crate::refs::{DigestField, UnusedExpectedId, condition_from_wire, hash_from_slice};
47use crate::replay::{BeginUploadResult, UpdateRefResult};
48use crate::rt::{send_wrap, send_wrap_stream};
49use crate::store::{MultipartBlobStore, NamespaceStore};
50use crate::upload::UploadError;
51use crate::url_token::UrlTarget;
52
53/// `TransportService` over a [`Pipeline`]. Each handler takes the
54/// [`Authenticated`] that [`super::AuthInterceptor`] stored for existing RPCs;
55/// `GetServerInfo` is deliberately unauthenticated. The M2 RPCs remain
56/// explicit stubs until their implementing WPs land.
57pub struct ConnectTransport<B, N, H> {
58    pipe: Shared<Pipeline<B, N, H>>,
59}
60
61impl<B, N, H> fmt::Debug for ConnectTransport<B, N, H> {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        f.debug_struct("ConnectTransport").finish_non_exhaustive()
64    }
65}
66
67impl<B, N, H> ConnectTransport<B, N, H> {
68    /// The service over `pipeline`.
69    #[must_use]
70    pub fn new(pipeline: Arc<Pipeline<B, N, H>>) -> Self {
71        Self {
72            pipe: Shared::new(pipeline),
73        }
74    }
75}
76
77/// A wire `(name, expectation, expected_id, new_id)` as a [`RefUpdate`].
78/// The name is checked by the pipeline.
79fn ref_update(
80    name: Option<String>,
81    expectation: Option<buffa::EnumValue<RefExpectation>>,
82    expected_id: Option<&[u8]>,
83    new_id: Option<&[u8]>,
84    delete: bool,
85) -> Result<RefUpdate, ServerError> {
86    let expectation = expectation.map_or(0, |e| e.to_i32());
87    let expected_id = expected_id.unwrap_or_default();
88    let condition = condition_from_wire(expectation, expected_id, UnusedExpectedId::Reject)?;
89    let new = if delete {
90        if !matches!(condition, mkit_core::refs::RefWriteCondition::Match(_))
91            || !new_id.unwrap_or_default().is_empty()
92        {
93            return Err(ServerError::invalid_argument(
94                "delete requires MATCH and an empty new_id",
95            ));
96        }
97        None
98    } else {
99        Some(hash_from_slice(DigestField::NewId, new_id)?)
100    };
101    Ok(RefUpdate {
102        name: name.unwrap_or_default(),
103        condition,
104        new,
105    })
106}
107
108fn not_yet() -> ServerError {
109    ServerError::unimplemented("not implemented yet")
110}
111
112fn pack_key(pack_id: Option<&[u8]>) -> Result<PackKey, ServerError> {
113    Ok(PackKey::new(hash_from_slice(DigestField::PackId, pack_id)?))
114}
115
116fn download_message(body: DownloadBody) -> DownloadPackResponse {
117    DownloadPackResponse {
118        body: Some(body),
119        ..Default::default()
120    }
121}
122
123fn chunk_message(pack_id: &[u8], chunk: &DownloadChunk) -> DownloadPackResponse {
124    download_message(DownloadBody::Chunk(Box::new(PackChunk {
125        pack_id: Some(pack_id.to_vec()),
126        offset: Some(chunk.offset),
127        data: Some(chunk.data.to_vec()),
128        last: Some(chunk.last),
129        ..Default::default()
130    })))
131}
132
133/// Stream `requests` into an upload session: the header, then each chunk
134/// up to the `last` one, then `finish`. The caller keeps the pipeline
135/// alive across the stream. Reading stops at `last`; connectrpc drains
136/// what follows within its bounds.
137async fn upload<B: MultipartBlobStore, N: NamespaceStore, H: HookSet>(
138    pipe: &Pipeline<B, N, H>,
139    a: &Authenticated,
140    mut requests: InboundStream<UploadPackRequest>,
141) -> Result<(), ConnectError> {
142    let header = match requests.next().await.transpose()? {
143        None => Err(UploadError::HeaderMissing { stream_empty: true }),
144        Some(first) => match first.to_owned_message().body {
145            Some(UploadBody::Header(header)) => Ok(header),
146            _ => Err(UploadError::HeaderMissing {
147                stream_empty: false,
148            }),
149        },
150    };
151    let header = header.map_err(ServerError::from)?;
152    let token = header.ticket_token.as_deref().unwrap_or_default();
153    let mut session = if token.is_empty() {
154        pipe.open_upload(a, header.pack_id.as_deref(), header.total_bytes)
155            .await?
156    } else {
157        pipe.open_ticketed_upload(a, header.pack_id.as_deref(), header.total_bytes, token)
158            .await?
159    };
160    while let Some(item) = requests.next().await {
161        let chunk = match item.map(|m| m.to_owned_message().body) {
162            Ok(Some(UploadBody::Chunk(chunk))) => *chunk,
163            // The request is recorded with the code the client receives.
164            Ok(body) => {
165                let err = ServerError::from(UploadError::UnexpectedMessage {
166                    header: body.is_some(),
167                });
168                session.abort_with(&err).await;
169                return Err(err.into());
170            }
171            Err(e) => {
172                session.abort_with(&recorded(&e)).await;
173                return Err(e);
174            }
175        };
176        let data = Bytes::from(chunk.data.unwrap_or_default());
177        let last = chunk.last.unwrap_or(false);
178        match session
179            .push(chunk.pack_id.as_deref(), chunk.offset, data, last)
180            .await
181        {
182            Ok(false) => {}
183            Ok(true) => break,
184            // `push` already recorded `e`; this only discards the sink.
185            Err(e) => {
186                session.abort_with(&e).await;
187                return Err(e.into());
188            }
189        }
190    }
191    session.finish().await?;
192    Ok(())
193}
194
195/// Read one part's header before opening its sink, then forward nonempty
196/// chunks without buffering the part in the Connect layer.
197async fn upload_part<B: MultipartBlobStore, N: NamespaceStore, H: HookSet>(
198    pipe: &Pipeline<B, N, H>,
199    a: &Authenticated,
200    mut requests: InboundStream<UploadPartRequest>,
201) -> Result<Vec<u8>, ConnectError> {
202    let first = match requests.next().await.transpose() {
203        Ok(first) => first,
204        Err(err) => {
205            pipe.record_part_error(a, &recorded(&err));
206            return Err(err);
207        }
208    };
209    let header = match first {
210        None => Err(UploadError::HeaderMissing { stream_empty: true }),
211        Some(first) => match first.to_owned_message().msg {
212            Some(PartMsg::Header(header)) => Ok(*header),
213            _ => Err(UploadError::HeaderMissing {
214                stream_empty: false,
215            }),
216        },
217    }
218    .map_err(|error| {
219        ServerError::invalid_argument(error.connect_message().replace("UploadPack", "UploadPart"))
220    });
221    let header = match header {
222        Ok(header) => header,
223        Err(err) => {
224            pipe.record_part_error(a, &err);
225            return Err(err.into());
226        }
227    };
228    let mut session = pipe
229        .open_part(
230            a,
231            header.ticket_token.as_deref().unwrap_or_default(),
232            header.index.unwrap_or_default(),
233        )
234        .await?;
235    while let Some(item) = requests.next().await {
236        let chunk = match item {
237            Ok(message) => match message.to_owned_message().msg {
238                Some(PartMsg::Chunk(chunk)) => chunk,
239                Some(PartMsg::Header(_)) => {
240                    let err =
241                        ServerError::invalid_argument("UploadPart: saw a second `header` message");
242                    session.abort_with(&err).await;
243                    return Err(err.into());
244                }
245                None => {
246                    let err = ServerError::invalid_argument(
247                        "UploadPart: message with neither `header` nor `chunk` set",
248                    );
249                    session.abort_with(&err).await;
250                    return Err(err.into());
251                }
252            },
253            Err(err) => {
254                session.abort_with(&recorded(&err)).await;
255                return Err(err);
256            }
257        };
258        if let Err(err) = session.push(Bytes::from(chunk)).await {
259            session.abort_with(&err).await;
260            return Err(err.into());
261        }
262    }
263    session.finish().await.map_err(Into::into)
264}
265
266#[allow(refining_impl_trait)]
267impl<B, N, H> TransportService for ConnectTransport<B, N, H>
268where
269    B: MultipartBlobStore + 'static,
270    N: NamespaceStore + 'static,
271    H: HookSet + 'static,
272{
273    async fn list_repos(
274        &self,
275        ctx: RequestContext,
276        request: ServiceRequest<'_, super::proto::mkit::transport::v1::ListReposRequest>,
277    ) -> ServiceResult<super::proto::mkit::transport::v1::ListReposResponse> {
278        let a = authenticated(&ctx)?;
279        let m = request.to_owned_message();
280        let pipe = self.pipe.arc();
281        send_wrap(async move {
282            let token = m
283                .page_token
284                .as_deref()
285                .filter(|s| !s.is_empty())
286                .map(|s| {
287                    if s.len() > 512 {
288                        return Err(ServerError::invalid_argument(
289                            "invalid repository page token",
290                        ));
291                    }
292                    URL_SAFE_NO_PAD
293                        .decode(s)
294                        .map_err(|_| ServerError::invalid_argument("invalid repository page token"))
295                })
296                .transpose()?;
297            let page = pipe
298                .list_repos_page(
299                    &a,
300                    m.namespace.as_deref().unwrap_or_default(),
301                    m.name_prefix.as_deref().unwrap_or_default(),
302                    m.page_size,
303                    token.as_deref(),
304                )
305                .await?;
306            let response = super::proto::mkit::transport::v1::ListReposResponse {
307                repos: page
308                    .repos
309                    .into_iter()
310                    .map(|entry| super::proto::mkit::transport::v1::RepoEntry {
311                        name: Some(entry.name),
312                        visibility: Some(
313                            if entry.visibility == crate::pipeline::RepoVisibility::Public {
314                                RepoVisibility::REPO_VISIBILITY_PUBLIC
315                            } else {
316                                RepoVisibility::REPO_VISIBILITY_PRIVATE
317                            }
318                            .into(),
319                        ),
320                        ..Default::default()
321                    })
322                    .collect(),
323                next_page_token: page.next.map(|bytes| URL_SAFE_NO_PAD.encode(bytes)),
324                ..Default::default()
325            };
326            if response.encoded_len() > 64 * 1024 {
327                return Err(ServerError::unavailable("repository listing unavailable").into());
328            }
329            Response::ok(response)
330        })
331        .await
332    }
333
334    async fn list_refs(
335        &self,
336        ctx: RequestContext,
337        request: ServiceRequest<'_, ListRefsRequest>,
338    ) -> ServiceResult<ListRefsResponse> {
339        let a = authenticated(&ctx)?;
340        let m = request.to_owned_message();
341        let prefix = m.prefix.unwrap_or_default();
342        let pipe = self.pipe.arc();
343        send_wrap(async move {
344            let token = m
345                .page_token
346                .as_deref()
347                .filter(|t| !t.is_empty())
348                .map(|t| {
349                    if t.len() > 730 {
350                        return Err(ServerError::invalid_argument("invalid page token"));
351                    }
352                    URL_SAFE_NO_PAD
353                        .decode(t)
354                        .map_err(|_| ServerError::invalid_argument("invalid page token"))
355                })
356                .transpose()
357                .map_err(ConnectError::from)?;
358            let page = pipe
359                .list_refs_page(&a, &prefix, m.page_size, token.as_deref())
360                .await?;
361            let refs = page.refs.into_iter().map(|entry| RefEntry {
362                name: Some(entry.name),
363                object_id: Some(entry.id.to_vec()),
364                ..Default::default()
365            });
366            let response = ListRefsResponse {
367                refs: refs.collect(),
368                next_page_token: page.next.map(|bytes| URL_SAFE_NO_PAD.encode(bytes)),
369                ..Default::default()
370            };
371            if response.encoded_len() > 2 * 1024 * 1024 {
372                return Err(ServerError::unavailable("ref listing unavailable").into());
373            }
374            Response::ok(response)
375        })
376        .await
377    }
378
379    async fn read_ref(
380        &self,
381        ctx: RequestContext,
382        request: ServiceRequest<'_, ReadRefRequest>,
383    ) -> ServiceResult<ReadRefResponse> {
384        let a = authenticated(&ctx)?;
385        let name = request.to_owned_message().name.unwrap_or_default();
386        let pipe = self.pipe.arc();
387        send_wrap(async move {
388            let id = pipe.read_ref(&a, &name).await?;
389            Response::ok(ReadRefResponse {
390                exists: Some(id.is_some()),
391                object_id: Some(id.map(|id| id.to_vec()).unwrap_or_default()),
392                ..Default::default()
393            })
394        })
395        .await
396    }
397
398    async fn update_ref(
399        &self,
400        ctx: RequestContext,
401        request: ServiceRequest<'_, UpdateRefRequest>,
402    ) -> ServiceResult<UpdateRefResponse> {
403        let a = authenticated(&ctx)?;
404        let m = request.to_owned_message();
405        let upd = ref_update(
406            m.name,
407            m.expectation,
408            m.expected_id.as_deref(),
409            m.new_id.as_deref(),
410            m.delete.unwrap_or(false),
411        )?;
412        let pipe = self.pipe.arc();
413        send_wrap(async move {
414            match pipe.update_ref_with_meta(&a, upd).await? {
415                (UpdateRefResult::Committed, meta) => {
416                    let mut response = Response::new(UpdateRefResponse::default());
417                    for (name, value) in meta.headers() {
418                        response = response.with_header(name, value);
419                    }
420                    Ok(response)
421                }
422                // SPEC-TRANSPORT-CONNECT §3: the response never carries the
423                // current value.
424                (UpdateRefResult::Conflict { .. }, _) => Err(ServerError::failed_precondition(
425                    "ref CAS precondition failed — read_ref to disambiguate",
426                )
427                .into()),
428            }
429        })
430        .await
431    }
432
433    async fn advance_refs(
434        &self,
435        ctx: RequestContext,
436        request: ServiceRequest<'_, AdvanceRefsRequest>,
437    ) -> ServiceResult<AdvanceRefsResponse> {
438        let a = authenticated(&ctx)?;
439        let m = request.to_owned_message();
440        let delete = m.delete.unwrap_or(false);
441        if delete && !m.ticket_ids.is_empty() {
442            return Err(ServerError::invalid_argument("delete consumes no tickets").into());
443        }
444        if m.ticket_ids.len() > crate::store::outbox::MAX_TICKETS_PER_ADVANCE {
445            return Err(ServerError::invalid_argument("too many tickets in one advance").into());
446        }
447        let mut tickets = Vec::with_capacity(m.ticket_ids.len());
448        for raw in &m.ticket_ids {
449            let id: [u8; 32] = raw
450                .as_slice()
451                .try_into()
452                .map_err(|_| ServerError::invalid_argument("ticket id must be 32 bytes"))?;
453            if tickets.contains(&id) {
454                return Err(ServerError::invalid_argument("duplicate ticket id").into());
455            }
456            tickets.push(id);
457        }
458        let head = ref_update(
459            m.head_ref,
460            m.head_expectation,
461            m.head_expected_id.as_deref(),
462            m.head_new_id.as_deref(),
463            delete,
464        )?;
465        let packmap = ref_update(
466            m.packmap_ref,
467            m.packmap_expectation,
468            m.packmap_expected_id.as_deref(),
469            m.packmap_new_id.as_deref(),
470            delete,
471        )?;
472        let pipe = self.pipe.arc();
473        send_wrap(async move {
474            // A conflict is a typed outcome, never an error (§4).
475            let (outcome, meta) = pipe
476                .advance_refs_with_tickets_with_meta(&a, head, packmap, tickets)
477                .await?;
478            let outcome = match outcome {
479                AdvanceOutcome::Committed => WireOutcome::ADVANCE_OUTCOME_COMMITTED,
480                AdvanceOutcome::HeadConflict => WireOutcome::ADVANCE_OUTCOME_HEAD_CONFLICT,
481                AdvanceOutcome::PackmapConflict => WireOutcome::ADVANCE_OUTCOME_PACKMAP_CONFLICT,
482            };
483            let mut response = Response::new(AdvanceRefsResponse {
484                outcome: Some(outcome.into()),
485                ..Default::default()
486            });
487            for (name, value) in meta.headers() {
488                response = response.with_header(name, value);
489            }
490            Ok(response)
491        })
492        .await
493    }
494
495    async fn pack_exists(
496        &self,
497        ctx: RequestContext,
498        request: ServiceRequest<'_, PackExistsRequest>,
499    ) -> ServiceResult<PackExistsResponse> {
500        let a = authenticated(&ctx)?;
501        let key = pack_key(request.to_owned_message().pack_id.as_deref())?;
502        let pipe = self.pipe.arc();
503        send_wrap(async move {
504            let exists = pipe.pack_exists(&a, key).await?;
505            Response::ok(PackExistsResponse {
506                exists: Some(exists),
507                ..Default::default()
508            })
509        })
510        .await
511    }
512
513    async fn upload_pack(
514        &self,
515        ctx: RequestContext,
516        requests: InboundStream<UploadPackRequest>,
517    ) -> ServiceResult<UploadPackResponse> {
518        let a = authenticated(&ctx)?;
519        // The session borrows the pipeline: this owned handle keeps it
520        // alive for the whole stream.
521        let pipe = self.pipe.arc();
522        send_wrap(async move {
523            upload(&pipe, &a, requests).await?;
524            Response::ok(UploadPackResponse::default())
525        })
526        .await
527    }
528
529    async fn download_pack(
530        &self,
531        ctx: RequestContext,
532        request: ServiceRequest<'_, DownloadPackRequest>,
533    ) -> ServiceResult<ServiceStream<DownloadPackResponse>> {
534        let a = authenticated(&ctx)?;
535        let key = pack_key(request.to_owned_message().pack_id.as_deref())?;
536        let pipe = self.pipe.arc();
537        // `not_found` and every other failure to open come back here,
538        // before any message (§6.2).
539        let download = send_wrap(async move { pipe.download(&a, key).await }).await?;
540        let header = download_message(DownloadBody::Header(Box::new(DownloadPackHeader {
541            total_bytes: Some(download.total_bytes),
542            ..Default::default()
543        })));
544        let chunks = download.chunks.map(move |chunk| {
545            chunk
546                .map(|c| chunk_message(&key.0, &c))
547                .map_err(ConnectError::from)
548        });
549        Response::stream_ok(stream::iter([Ok(header)]).chain(send_wrap_stream(chunks)))
550    }
551
552    async fn get_server_info(
553        &self,
554        _ctx: RequestContext,
555        _request: ServiceRequest<'_, GetServerInfoRequest>,
556    ) -> ServiceResult<GetServerInfoResponse> {
557        // SECURITY: deliberately unauthenticated and outside Procedure; it
558        // never resolves a repository or reads its state (STC §2.1).
559        let info = self.pipe.get().server_info();
560        Ok(Response::new(info.into()).with_header("cache-control", "private, max-age=60"))
561    }
562
563    async fn get_receipt(
564        &self,
565        _ctx: RequestContext,
566        _request: ServiceRequest<'_, GetReceiptRequest>,
567    ) -> ServiceResult<GetReceiptResponse> {
568        // WP-5.8 implements receipt retention and writer-view retrieval.
569        Err(not_yet().into())
570    }
571
572    async fn begin_upload(
573        &self,
574        ctx: RequestContext,
575        request: ServiceRequest<'_, BeginUploadRequest>,
576    ) -> ServiceResult<BeginUploadResponse> {
577        let a = authenticated(&ctx)?;
578        let message = request.to_owned_message();
579        let pipe = self.pipe.arc();
580        send_wrap(async move {
581            let (result, meta) = pipe
582                .begin_upload_with_meta(
583                    &a,
584                    &message.r#ref.unwrap_or_default(),
585                    message.pack_id.as_deref().unwrap_or_default(),
586                    message.bytes.unwrap_or_default(),
587                )
588                .await?;
589            let result = match result {
590                BeginUploadResult::AlreadyPresent => BeginResult::AlreadyPresent(Box::default()),
591                BeginUploadResult::Ticket {
592                    id,
593                    part_size,
594                    expires_at_ms,
595                    token,
596                } => {
597                    let expires_unix_ms = i64::try_from(expires_at_ms).map_err(|_| {
598                        ServerError::internal(
599                            "upload ticket expiry exceeds wire clock",
600                            expires_at_ms,
601                        )
602                    })?;
603                    BeginResult::Ticket(Box::new(UploadTicket {
604                        id: Some(id.to_vec()),
605                        part_size: Some(part_size),
606                        expires_unix_ms: Some(expires_unix_ms),
607                        token: Some(token),
608                        ..Default::default()
609                    }))
610                }
611            };
612            let mut response = Response::new(BeginUploadResponse {
613                result: Some(result),
614                ..Default::default()
615            });
616            for (name, value) in meta.headers() {
617                response = response.with_header(name, value);
618            }
619            Ok(response)
620        })
621        .await
622    }
623
624    async fn upload_part(
625        &self,
626        ctx: RequestContext,
627        requests: InboundStream<UploadPartRequest>,
628    ) -> ServiceResult<UploadPartResponse> {
629        let a = authenticated(&ctx)?;
630        let pipe = self.pipe.arc();
631        send_wrap(async move {
632            let receipt = upload_part(&pipe, &a, requests).await?;
633            Response::ok(UploadPartResponse {
634                receipt: Some(receipt),
635                ..Default::default()
636            })
637        })
638        .await
639    }
640
641    async fn complete_upload(
642        &self,
643        ctx: RequestContext,
644        request: ServiceRequest<'_, CompleteUploadRequest>,
645    ) -> ServiceResult<CompleteUploadResponse> {
646        let a = authenticated(&ctx)?;
647        let message = request.to_owned_message();
648        let pipe = self.pipe.arc();
649        send_wrap(async move {
650            pipe.complete_upload(
651                &a,
652                message.ticket_token.as_deref().unwrap_or_default(),
653                &message.receipts,
654            )
655            .await?;
656            Response::ok(CompleteUploadResponse::default())
657        })
658        .await
659    }
660
661    async fn get_grant_epoch(
662        &self,
663        _ctx: RequestContext,
664        request: ServiceRequest<'_, GetGrantEpochRequest>,
665    ) -> ServiceResult<GetGrantEpochResponse> {
666        // Unsigned by design: no repository selector or auth headers are read.
667        let message = request.to_owned_message();
668        let pipe = self.pipe.arc();
669        send_wrap(async move {
670            let epoch = pipe
671                .get_grant_epoch(message.namespace.as_deref().unwrap_or_default())
672                .await?;
673            Response::ok(GetGrantEpochResponse {
674                epoch: Some(epoch),
675                ..Default::default()
676            })
677        })
678        .await
679    }
680
681    async fn set_grant_epoch(
682        &self,
683        _ctx: RequestContext,
684        request: ServiceRequest<'_, SetGrantEpochRequest>,
685    ) -> ServiceResult<SetGrantEpochResponse> {
686        // Unsigned forever: the owner statement is its sole authorization.
687        let message = request.to_owned_message();
688        let pipe = self.pipe.arc();
689        send_wrap(async move {
690            let epoch = pipe
691                .set_grant_epoch(message.signed_statement.as_deref().unwrap_or_default())
692                .await?;
693            Response::ok(SetGrantEpochResponse {
694                epoch: Some(epoch),
695                ..Default::default()
696            })
697        })
698        .await
699    }
700
701    async fn get_authority_generation(
702        &self,
703        _ctx: RequestContext,
704        request: ServiceRequest<'_, GetAuthorityGenerationRequest>,
705    ) -> ServiceResult<GetAuthorityGenerationResponse> {
706        // Unsigned by design: no repository selector or auth headers are read.
707        let message = request.to_owned_message();
708        let pipe = self.pipe.arc();
709        send_wrap(async move {
710            let generation = pipe
711                .get_authority_generation(message.namespace.as_deref().unwrap_or_default())
712                .await?;
713            Response::ok(GetAuthorityGenerationResponse {
714                generation: Some(generation),
715                ..Default::default()
716            })
717        })
718        .await
719    }
720
721    async fn set_authority_generation(
722        &self,
723        _ctx: RequestContext,
724        request: ServiceRequest<'_, SetAuthorityGenerationRequest>,
725    ) -> ServiceResult<SetAuthorityGenerationResponse> {
726        // Unsigned forever: the deployment-authority statement is its sole authorization.
727        let message = request.to_owned_message();
728        let pipe = self.pipe.arc();
729        send_wrap(async move {
730            let generation = pipe
731                .set_authority_generation(message.signed_statement.as_deref().unwrap_or_default())
732                .await?;
733            Response::ok(SetAuthorityGenerationResponse {
734                generation: Some(generation),
735                ..Default::default()
736            })
737        })
738        .await
739    }
740
741    async fn set_repo_visibility(
742        &self,
743        ctx: RequestContext,
744        request: ServiceRequest<'_, SetRepoVisibilityRequest>,
745    ) -> ServiceResult<SetRepoVisibilityResponse> {
746        let a = authenticated(&ctx)?;
747        let req = match request.to_owned_message().mode {
748            Some(VisibilityMode::Visibility(visibility)) => match visibility.as_known() {
749                Some(RepoVisibility::REPO_VISIBILITY_PUBLIC) => {
750                    VisibilityRequest::Envelope(mkit_attest::grant::Visibility::Public)
751                }
752                Some(RepoVisibility::REPO_VISIBILITY_PRIVATE) => {
753                    VisibilityRequest::Envelope(mkit_attest::grant::Visibility::Private)
754                }
755                _ => return Err(ServerError::invalid_argument("invalid visibility").into()),
756            },
757            Some(VisibilityMode::SignedStatement(statement)) => {
758                VisibilityRequest::Statement(statement)
759            }
760            None => {
761                return Err(
762                    ServerError::invalid_argument("SetRepoVisibility requires a mode").into(),
763                );
764            }
765        };
766        let pipe = self.pipe.arc();
767        send_wrap(async move {
768            pipe.set_repo_visibility(&a, req).await?;
769            Response::ok(SetRepoVisibilityResponse::default())
770        })
771        .await
772    }
773
774    async fn issue_object_url(
775        &self,
776        ctx: RequestContext,
777        request: ServiceRequest<'_, IssueObjectUrlRequest>,
778    ) -> ServiceResult<IssueObjectUrlResponse> {
779        let a = authenticated(&ctx)?;
780        let message = request.to_owned_message();
781        let invalid = || ServerError::invalid_argument("invalid object URL target");
782        let target = match message.target {
783            Some(Target::ObjectId(id)) => {
784                UrlTarget::Object(Hash::try_from(id.as_slice()).map_err(|_| invalid())?)
785            }
786            Some(Target::RefPath(path)) => UrlTarget::path(
787                path.r#ref.unwrap_or_default(),
788                path.path.unwrap_or_default(),
789            )
790            .map_err(|_| invalid())?,
791            None => {
792                return Err(
793                    ServerError::invalid_argument("IssueObjectUrl requires a target").into(),
794                );
795            }
796        };
797        let pipe = self.pipe.arc();
798        send_wrap(async move {
799            let minted = pipe
800                .issue_object_url(&a, target, message.ttl_seconds.unwrap_or_default())
801                .await?;
802            Response::ok(IssueObjectUrlResponse {
803                token: Some(minted.expose().to_owned()),
804                expires_unix_ms: Some(minted.expires_at_ms),
805                ..Default::default()
806            })
807        })
808        .await
809    }
810}
811
812impl From<ServerInfo> for GetServerInfoResponse {
813    fn from(info: ServerInfo) -> Self {
814        Self {
815            protocol: Some(info.protocol.into()),
816            spec_version: Some(info.spec_version),
817            max_pack_bytes: Some(info.max_pack_bytes),
818            part_size: Some(info.part_size),
819            max_parts: Some(info.max_parts),
820            max_list_refs_page_size: Some(info.max_list_refs_page_size),
821            begin_upload_threshold_bytes: Some(info.begin_upload_threshold_bytes),
822            atomic_advance: Some(info.atomic_advance),
823            indexed_mode: Some(info.indexed_mode),
824            admission: Some(info.admission),
825            receipt_public_key: Some(info.receipt_public_key),
826            receipt_key_id: Some(info.receipt_key_id),
827            grant_schemes: info.grant_schemes,
828            namespace_policy: Some(info.namespace_policy.into()),
829            index_fanout: Some(info.index_fanout),
830            max_delta_chain_depth: Some(info.max_delta_chain_depth),
831            leases: Some(false), // Launch retains content permanently; no storage leases.
832            // Launch inspection is synchronous only (SPEC-SERVER §18).
833            async_inspection: Some(false),
834            inspection_max_objects: info.inspection_max_objects,
835            __buffa_unknown_fields: buffa::UnknownFields::default(),
836        }
837    }
838}
839
840#[cfg(test)]
841mod proto_roundtrip {
842    use super::super::proto::mkit::transport::v1::__buffa::oneof::{
843        begin_upload_response::Result as BeginResult, upload_part_request::Msg as PartMsg,
844    };
845    use super::super::proto::mkit::transport::v1::*;
846    use super::ServerInfo;
847    use buffa::Message;
848
849    fn roundtrip<M: Message + PartialEq + core::fmt::Debug>(message: &M) {
850        let encoded = message.encode_to_vec();
851        assert_eq!(
852            &M::decode_from_slice(&encoded).expect("decode generated message"),
853            message
854        );
855    }
856
857    fn ticket() -> UploadTicket {
858        UploadTicket {
859            id: Some(vec![0x12; 32]),
860            part_size: Some(8 << 20),
861            expires_unix_ms: Some(1_700_000_000_000),
862            token: Some(vec![0xa1, 0xb2]),
863            ..Default::default()
864        }
865    }
866
867    #[test]
868    fn admission_challenge_roundtrips_with_zero_one_and_eight_entries() {
869        for count in [0, 1, 8] {
870            let message = AdmissionChallenge {
871                challenges: (0..count)
872                    .map(|index| Challenge {
873                        scheme: Some(format!("scheme{index}")),
874                        value: Some(format!("opaque-{index}")),
875                        ..Default::default()
876                    })
877                    .collect(),
878                description: (count != 0).then(|| "Admission required".into()),
879                ..Default::default()
880            };
881            roundtrip(&message);
882            let json = serde_json::to_vec(&message).expect("serialize challenge");
883            assert_eq!(
884                serde_json::from_slice::<AdmissionChallenge>(&json).expect("parse challenge"),
885                message
886            );
887        }
888    }
889
890    #[test]
891    fn discovery_messages_roundtrip() {
892        // Empty messages have no declared fields to populate.
893        roundtrip(&GetServerInfoRequest::default());
894        roundtrip(&GetServerInfoResponse {
895            protocol: Some("mkit.transport.v1".into()),
896            spec_version: Some(2),
897            max_pack_bytes: Some(1 << 34),
898            part_size: Some(8 << 20),
899            max_parts: Some(1024),
900            max_list_refs_page_size: Some(512),
901            begin_upload_threshold_bytes: Some(8 << 20),
902            atomic_advance: Some(true),
903            indexed_mode: Some(true),
904            admission: Some(true),
905            receipt_public_key: Some(vec![0x34; 32]),
906            receipt_key_id: Some("receipt-key".into()),
907            grant_schemes: vec!["ed25519".into(), "secp256k1-eip191".into()],
908            namespace_policy: Some("allowlist".into()),
909            index_fanout: Some(4096),
910            inspection_max_objects: Some(10_000),
911            ..Default::default()
912        });
913    }
914
915    #[test]
916    fn discovery_explicitly_denies_unimplemented_lease_and_async_capabilities() {
917        for indexed_mode in [false, true] {
918            let response = GetServerInfoResponse::from(ServerInfo {
919                protocol: "mkit.transport.v1",
920                spec_version: 2,
921                max_pack_bytes: 1 << 30,
922                part_size: 8 << 20,
923                max_parts: 128,
924                max_list_refs_page_size: 512,
925                begin_upload_threshold_bytes: 0,
926                atomic_advance: false,
927                indexed_mode,
928                admission: false,
929                receipt_public_key: Vec::new(),
930                receipt_key_id: String::new(),
931                grant_schemes: Vec::new(),
932                namespace_policy: "allowlist",
933                index_fanout: 4096,
934                max_delta_chain_depth: if indexed_mode { 50 } else { 0 },
935                inspection_max_objects: None,
936            });
937            roundtrip(&response);
938            let json = serde_json::to_value(&response).unwrap();
939            assert_eq!(json["leases"], false);
940            assert_eq!(json["asyncInspection"], false);
941            assert_eq!(json["indexedMode"], indexed_mode);
942            assert!(response.receipt_public_key.unwrap().is_empty());
943            assert!(response.receipt_key_id.unwrap().is_empty());
944        }
945    }
946
947    #[test]
948    fn inspection_limit_roundtrips_and_is_absent_without_inspection() {
949        for limit in [None, Some(10_000)] {
950            let response = GetServerInfoResponse {
951                inspection_max_objects: limit,
952                ..Default::default()
953            };
954            roundtrip(&response);
955            let json = serde_json::to_value(&response).unwrap();
956            assert_eq!(
957                json.get("inspectionMaxObjects"),
958                limit.map(serde_json::Value::from).as_ref()
959            );
960            assert_eq!(
961                serde_json::from_value::<GetServerInfoResponse>(json).unwrap(),
962                response
963            );
964        }
965    }
966
967    #[test]
968    fn ticket_messages_and_both_results_roundtrip() {
969        roundtrip(&BeginUploadRequest {
970            r#ref: Some("refs/heads/main".into()),
971            pack_id: Some(vec![0x56; 32]),
972            bytes: Some(1 << 33),
973            ..Default::default()
974        });
975        roundtrip(&AlreadyPresent::default());
976        roundtrip(&ticket());
977        roundtrip(&BeginUploadResponse {
978            result: Some(BeginResult::AlreadyPresent(Box::default())),
979            ..Default::default()
980        });
981        roundtrip(&BeginUploadResponse {
982            result: Some(BeginResult::Ticket(Box::new(ticket()))),
983            ..Default::default()
984        });
985    }
986
987    #[test]
988    fn part_messages_and_both_stream_alternatives_roundtrip() {
989        let header = UploadPartHeader {
990            ticket_token: Some(vec![0x78, 0x9a]),
991            index: Some(3),
992            ..Default::default()
993        };
994        roundtrip(&header);
995        roundtrip(&UploadPartRequest {
996            msg: Some(PartMsg::Header(Box::new(header))),
997            ..Default::default()
998        });
999        roundtrip(&UploadPartRequest {
1000            msg: Some(PartMsg::Chunk(vec![0x01, 0x23, 0x45])),
1001            ..Default::default()
1002        });
1003        roundtrip(&UploadPartResponse {
1004            receipt: Some(vec![0xab, 0xcd]),
1005            ..Default::default()
1006        });
1007        roundtrip(&CompleteUploadRequest {
1008            ticket_token: Some(vec![0x78, 0x9a]),
1009            receipts: vec![vec![0xab, 0xcd], vec![0xef, 0x01]],
1010            ..Default::default()
1011        });
1012        roundtrip(&CompleteUploadResponse::default());
1013    }
1014
1015    #[test]
1016    fn m2_messages_roundtrip() {
1017        use super::super::proto::mkit::transport::v1::__buffa::oneof::issue_object_url_request::Target;
1018        use super::super::proto::mkit::transport::v1::__buffa::oneof::set_repo_visibility_request::Mode;
1019        use super::super::proto::mkit::transport::v1::{RefPath, RepoVisibility};
1020
1021        roundtrip(&GetGrantEpochRequest {
1022            namespace: Some("namespace".into()),
1023            ..Default::default()
1024        });
1025        roundtrip(&GetGrantEpochResponse {
1026            epoch: Some(42),
1027            ..Default::default()
1028        });
1029        roundtrip(&SetGrantEpochRequest {
1030            signed_statement: Some("statement.scheme.blob".into()),
1031            ..Default::default()
1032        });
1033        roundtrip(&SetGrantEpochResponse {
1034            epoch: Some(43),
1035            ..Default::default()
1036        });
1037        for visibility in [
1038            RepoVisibility::Unspecified,
1039            RepoVisibility::Public,
1040            RepoVisibility::Private,
1041        ] {
1042            roundtrip(&SetRepoVisibilityRequest {
1043                mode: Some(Mode::Visibility(visibility.into())),
1044                ..Default::default()
1045            });
1046        }
1047        roundtrip(&SetRepoVisibilityRequest {
1048            mode: Some(Mode::SignedStatement("statement.scheme.blob".into())),
1049            ..Default::default()
1050        });
1051        roundtrip(&SetRepoVisibilityResponse::default());
1052        roundtrip(&IssueObjectUrlRequest {
1053            target: Some(Target::ObjectId(vec![0x42; 32])),
1054            ttl_seconds: Some(15),
1055            ..Default::default()
1056        });
1057        let ref_path = RefPath {
1058            r#ref: Some("refs/heads/main".into()),
1059            path: Some("dir/file".into()),
1060            ..Default::default()
1061        };
1062        roundtrip(&ref_path);
1063        roundtrip(&IssueObjectUrlRequest {
1064            target: Some(Target::RefPath(Box::new(ref_path))),
1065            ttl_seconds: Some(20),
1066            ..Default::default()
1067        });
1068        roundtrip(&IssueObjectUrlResponse {
1069            token: Some("signed.token".into()),
1070            expires_unix_ms: Some(1_700_000_000_000),
1071            ..Default::default()
1072        });
1073    }
1074
1075    #[test]
1076    fn additive_fields_roundtrip() {
1077        roundtrip(&ListRefsRequest {
1078            prefix: Some("refs/heads/".into()),
1079            page_size: Some(1),
1080            page_token: Some("continuation".into()),
1081            ..Default::default()
1082        });
1083        roundtrip(&ListRefsResponse {
1084            refs: vec![RefEntry {
1085                name: Some("main".into()),
1086                object_id: Some(vec![0x23; 32]),
1087                ..Default::default()
1088            }],
1089            next_page_token: Some("next".into()),
1090            ..Default::default()
1091        });
1092        roundtrip(&UpdateRefRequest {
1093            delete: Some(true),
1094            ..Default::default()
1095        });
1096        roundtrip(&AdvanceRefsRequest {
1097            ticket_ids: vec![vec![0x45; 32], vec![0x67; 32]],
1098            delete: Some(true),
1099            ..Default::default()
1100        });
1101        roundtrip(&UploadPackHeader {
1102            ticket_token: Some(vec![0x89, 0xab]),
1103            ..Default::default()
1104        });
1105    }
1106}
mkit_server/connect/service.rs

mkit_server/connect/
service.rs
