mkit_server/connect/interceptor.rs
1//! Pipeline stage 0 as a Connect interceptor.
2
3use core::fmt;
4use std::sync::Arc;
5
6use connectrpc::interceptor::{
7 NextStream, PayloadStream, StreamRequest, StreamResponse, UnaryRequest, UnaryResponse,
8};
9use connectrpc::{ConnectError, Interceptor, Next, RequestContext, async_trait};
10use futures::StreamExt as _;
11
12use super::Shared;
13use crate::auth_v2;
14use crate::error::ServerError;
15use crate::op::Procedure;
16use crate::pipeline::{HookSet, Pipeline, RequestMeta};
17use crate::principal::Principal;
18use crate::store::{MultipartBlobStore, NamespaceStore};
19
20/// Runs [`Pipeline::authenticate`] once per call, before any message
21/// reaches a handler, and stores the resulting
22/// [`crate::pipeline::Authenticated`] (with its test directives under
23/// `test-faults`) in the request extensions. A unary call is checked
24/// against its exact request bytes; a client stream against its headers
25/// only; `DownloadPack`'s single request envelope is buffered, verified
26/// over its reconstructed framed body (`0x00‖be32(len)‖message`, R-129)
27/// and re-injected.
28///
29/// Calls outside `mkit.transport.v1.TransportService` (health) pass
30/// through unauthenticated. For [`crate::pipeline::AuthMode::TransportIdentity`]
31/// an adapter inserts the peer's [`Principal`] into the HTTP request
32/// extensions; nothing a client sends can set it.
33///
34/// It must be the first (outermost) interceptor. An adapter that builds its
35/// own chain from [`super::router`] registers it before any other, so no
36/// interceptor can rewrite the message before the signature is checked. A
37/// rewritten unary payload is verified over its re-encoded bytes, which
38/// fails closed.
39///
40/// A unary body is verified as connectrpc hands it over, after any
41/// `Content-Encoding` is undone. The client signs the bytes it sends
42/// (SPEC-WRITE-GRANTS §9.2, SPEC-TRANSPORT-CONNECT §7.1), so a compressed
43/// signed request does not verify and is rejected `unauthenticated`, as in
44/// `vcs-worker`.
45pub struct AuthInterceptor<B, N, H> {
46 pipe: Shared<Pipeline<B, N, H>>,
47}
48
49impl<B, N, H> fmt::Debug for AuthInterceptor<B, N, H> {
50 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51 f.debug_struct("AuthInterceptor").finish_non_exhaustive()
52 }
53}
54
55impl<B: MultipartBlobStore, N: NamespaceStore, H: HookSet> AuthInterceptor<B, N, H> {
56 /// An interceptor authenticating against `pipeline`'s auth mode.
57 #[must_use]
58 pub fn new(pipeline: Arc<Pipeline<B, N, H>>) -> Self {
59 Self {
60 pipe: Shared::new(pipeline),
61 }
62 }
63
64 /// A request header as the pipeline sees it. A present-but-undecodable
65 /// auth v2 header maps to `"~"`, never `None`: its presence decides
66 /// signed-ness, and a value that fails to parse fails verification
67 /// instead of silently turning a signed read anonymous.
68 fn header(ctx: &RequestContext) -> impl Fn(&str) -> Option<String> + '_ {
69 move |name: &str| {
70 if name == "x-write-grant" {
71 let values: Vec<_> = ctx.headers().get_all(name).iter().collect();
72 if values.is_empty() {
73 return None;
74 }
75 return Some(
76 values
77 .iter()
78 .map(|value| value.to_str().unwrap_or("~"))
79 .collect::<Vec<_>>()
80 .join(", "),
81 );
82 }
83 ctx.header(name).and_then(|v| {
84 if name == "x-mkit-ref" {
85 core::str::from_utf8(v.as_bytes()).ok().map(str::to_owned)
86 } else if auth_v2::HEADER_NAMES.contains(&name) {
87 Some(v.to_str().unwrap_or("~").to_owned())
88 } else {
89 v.to_str().ok().map(str::to_owned)
90 }
91 })
92 }
93 }
94
95 /// Authenticate the call in `ctx` and record the result in its
96 /// extensions; a call outside the transport service is left alone.
97 fn authenticate(
98 &self,
99 ctx: &mut RequestContext,
100 unary_body: Option<&[u8]>,
101 ) -> Result<(), ServerError> {
102 let Some(procedure) = ctx.path().and_then(Procedure::from_connect_path) else {
103 return Ok(());
104 };
105 let transport_principal = ctx.extensions().get::<Principal>().cloned();
106 let authenticated = {
107 let header = Self::header(ctx);
108 let meta = RequestMeta {
109 procedure,
110 header: &header,
111 header_values: Some(&|name| {
112 ctx.headers()
113 .get_all(name)
114 .iter()
115 .map(|value| value.to_str().unwrap_or("\n").to_owned())
116 .collect()
117 }),
118 unary_body,
119 transport_principal,
120 };
121 self.pipe.get().authenticate(&meta)?
122 };
123 ctx.extensions_mut().insert(authenticated);
124 Ok(())
125 }
126}
127
128#[async_trait]
129impl<B, N, H> Interceptor for AuthInterceptor<B, N, H>
130where
131 B: MultipartBlobStore + 'static,
132 N: NamespaceStore + 'static,
133 H: HookSet + 'static,
134{
135 async fn intercept_unary(
136 &self,
137 mut req: UnaryRequest,
138 next: Next<'_>,
139 ) -> Result<UnaryResponse, ConnectError> {
140 // The bytes the handler will decode: the received ones unless an
141 // earlier interceptor replaced the message (then re-encoded, which
142 // fails the body commitment). See the type docs on compression.
143 let body = req.payload.encoded()?;
144 self.authenticate(&mut req.ctx, Some(&body))?;
145 next.run(req).await
146 }
147
148 async fn intercept_streaming(
149 &self,
150 mut req: StreamRequest,
151 mut inbound: PayloadStream,
152 next: NextStream<'_>,
153 ) -> Result<StreamResponse, ConnectError> {
154 if req.ctx.path().and_then(Procedure::from_connect_path) != Some(Procedure::DownloadPack) {
155 self.authenticate(&mut req.ctx, None)?;
156 return next.run(req, inbound).await;
157 }
158 // `DownloadPack` is server-streaming, but its request is one
159 // envelope and the `body:` commitment covers its exact wire bytes
160 // `0x00‖be32(len)‖message` (R-129). The dispatcher already decoded
161 // that envelope, so the frame is rebuilt from the single payload
162 // (buffered, bounded) and handed on afterwards.
163 let Some(item) = inbound.next().await else {
164 // No request envelope: a signed call fails its body check.
165 self.authenticate(&mut req.ctx, None)?;
166 return next.run(req, inbound).await;
167 };
168 let payload = item?;
169 let bytes = payload.encoded()?;
170 // The 1 KiB bound applies to the decoded message: connectrpc has
171 // already read the frame (capped by the transport's message limit)
172 // and undone any per-message compression before this point, so a
173 // pre-decode check is not available here.
174 if bytes.len() > 1024 {
175 return Err(ServerError::invalid_argument("DownloadPack request too large").into());
176 }
177 // The envelope's compression flag is already consumed; a
178 // `connect-content-encoding` (or `grpc-encoding`) header other
179 // than `identity` is its proxy. An uncompressed reconstruction
180 // can also never match bytes the client compressed before
181 // signing, so this fails closed either way.
182 let compressed_signed = {
183 let header = Self::header(&req.ctx);
184 auth_v2::carries_auth_headers(&header)
185 && ["connect-content-encoding", "grpc-encoding"]
186 .iter()
187 .any(|name| {
188 req.ctx
189 .header(*name)
190 .is_some_and(|v| !matches!(v.to_str(), Ok("identity")))
191 })
192 };
193 if compressed_signed {
194 return Err(ServerError::unauthenticated("compressed signed request").into());
195 }
196 let mut frame = Vec::with_capacity(5 + bytes.len());
197 frame.push(0);
198 frame.extend_from_slice(
199 &u32::try_from(bytes.len())
200 .map_err(|_| ServerError::invalid_argument("DownloadPack request too large"))?
201 .to_be_bytes(),
202 );
203 frame.extend_from_slice(&bytes);
204 self.authenticate(&mut req.ctx, Some(&frame))?;
205 // connectrpc collects a server-streaming call's request before
206 // the interceptors run and refuses a second message itself, so
207 // `inbound` holds at most this one payload; the chain below only
208 // hands it on.
209 let inbound: PayloadStream =
210 Box::pin(futures::stream::once(async move { Ok(payload) }).chain(inbound));
211 next.run(req, inbound).await
212 }
213}