gonzalo_server/
grpc.rs

1//! gRPC transport: adapts the generated `Gonzalo` service to the shared
2//! `Service`, carrying `gonzalo-core` types as JSON payloads.
3
4use crate::Service;
5use crate::auth::{Access, Auth, Principal};
6use gonzalo_core::{
7    ContentHash, CoreError, DeleteResult, Identity, KeyPrefix, PutResult, Record, RecordKey,
8    Revision,
9};
10use gonzalo_proto::v1::{
11    DeleteBlobRequest, DeleteBlobResponse, DeleteRequest, DeleteResponse, GetBlobRequest,
12    GetBlobResponse, GetRequest, GetResponse, GraphLocatedResponse, GraphNamesResponse,
13    GraphQueryRequest, HasBlobRequest, HasBlobResponse, ListBlobsRequest, ListBlobsResponse,
14    ListRequest, ListResponse, PurgeRequest, PurgeResponse, PutBlobRequest, PutBlobResponse,
15    PutRequest, PutResponse, TicketSyncRequest, TicketSyncResponse,
16    gonzalo_server::{Gonzalo, GonzaloServer},
17};
18use serde::Serialize;
19use std::sync::Arc;
20use tonic::metadata::MetadataMap;
21use tonic::{Request, Response, Status};
22
23/// Reserved authz namespace for namespace-agnostic blob ops (ADR 0015), matching
24/// the HTTP transport.
25const BLOB_NS: &str = "_blobs";
26
27/// Adapts [`Service`] to the generated gRPC trait, enforcing namespace-scoped
28/// auth (ADR 0015) per call from the request's bearer metadata.
29pub struct GrpcAdapter {
30    service: Service,
31    auth: Arc<Auth>,
32}
33
34impl GrpcAdapter {
35    /// Adapter with auth disabled (open) — used by tests and open deployments.
36    pub fn new(service: Service) -> Self {
37        Self::with_auth(service, Arc::new(Auth::Disabled))
38    }
39
40    /// Adapter enforcing `auth`.
41    pub fn with_auth(service: Service, auth: Arc<Auth>) -> Self {
42        Self { service, auth }
43    }
44
45    /// Authenticate the call's bearer token and authorize `access` on
46    /// `namespace`. Returns the [`Principal`] (for author stamping on writes).
47    #[allow(clippy::result_large_err)]
48    fn authorize(
49        &self,
50        metadata: &MetadataMap,
51        access: Access,
52        namespace: &str,
53    ) -> Result<Principal, Status> {
54        let principal = self.authenticate(metadata)?;
55        self.check_access(&principal, access, namespace)?;
56        Ok(principal)
57    }
58
59    /// Authenticate the call's bearer token into a [`Principal`], independent of
60    /// any namespace. Split out from [`authorize`] so a handler can reject an
61    /// unauthenticated caller *before* deserializing attacker-controlled JSON
62    /// (#146) and only then authorize against a namespace parsed from the body.
63    #[allow(clippy::result_large_err)]
64    fn authenticate(&self, metadata: &MetadataMap) -> Result<Principal, Status> {
65        self.auth
66            .authenticate(bearer(metadata))
67            .ok_or_else(|| Status::unauthenticated("invalid or missing token"))
68    }
69
70    /// Authorize an already-authenticated `principal` for `access` on `namespace`.
71    #[allow(clippy::result_large_err)]
72    fn check_access(
73        &self,
74        principal: &Principal,
75        access: Access,
76        namespace: &str,
77    ) -> Result<(), Status> {
78        if principal.allows(access, namespace) {
79            Ok(())
80        } else {
81            Err(Status::permission_denied(format!(
82                "principal {:?} lacks {access:?} on namespace {namespace:?}",
83                principal.name()
84            )))
85        }
86    }
87
88    /// Authenticate the call and require an admin principal (`read` and
89    /// `write` on `"*"`). Used by `purge` (gonzalo#203): an early purge is data
90    /// loss that surfaces later on another machine, so no namespace scope is
91    /// enough. Takes no body, so it runs before any deserialization (#146).
92    #[allow(clippy::result_large_err)]
93    fn authorize_admin(
94        &self,
95        metadata: &MetadataMap,
96        operation: &str,
97    ) -> Result<Principal, Status> {
98        let principal = self.authenticate(metadata)?;
99        if principal.is_admin() {
100            Ok(principal)
101        } else {
102            Err(Status::permission_denied(format!(
103                "principal {:?} is not an admin; {operation} requires admin",
104                principal.name()
105            )))
106        }
107    }
108
109    /// Authenticate, parse a `PutRequest` (malformed → `InvalidArgument`) and
110    /// authorize `Write` on the record's namespace. Shared by `Put` and `PutRaw`,
111    /// in #146 order.
112    #[allow(clippy::result_large_err)]
113    fn authorize_put(
114        &self,
115        metadata: &MetadataMap,
116        r: &PutRequest,
117    ) -> Result<(Principal, Record, Option<Revision>), Status> {
118        let principal = self.authenticate(metadata)?;
119        let record: Record = serde_json::from_slice(&r.record_json)
120            .map_err(|e| Status::invalid_argument(e.to_string()))?;
121        let expected: Option<Revision> = serde_json::from_slice(&r.expected_json)
122            .map_err(|e| Status::invalid_argument(e.to_string()))?;
123        self.check_access(&principal, Access::Write, &record.key.namespace)?;
124        Ok((principal, record, expected))
125    }
126
127    /// Authenticate the call and authorize a key listing (`read` on the
128    /// namespace, or on `"*"` when unscoped), then build its prefix. Shared by
129    /// `List` and `ListRaw`.
130    #[allow(clippy::result_large_err)]
131    fn authorize_list(&self, metadata: &MetadataMap, r: ListRequest) -> Result<KeyPrefix, Status> {
132        self.authorize(
133            metadata,
134            Access::Read,
135            r.namespace.as_deref().unwrap_or("*"),
136        )?;
137        Ok(KeyPrefix {
138            namespace: r.namespace,
139            collection: r.collection,
140        })
141    }
142}
143
144/// The bearer token from gRPC `authorization: Bearer <token>` metadata.
145fn bearer(metadata: &MetadataMap) -> Option<&str> {
146    metadata
147        .get("authorization")?
148        .to_str()
149        .ok()?
150        .strip_prefix("Bearer ")
151}
152
153/// Map a backend failure to an opaque `Internal` status. The full error is
154/// logged server-side; the client sees only "internal error" so on-disk graph
155/// paths, SQLite text, and S3 endpoint/bucket detail never leak to the network
156/// (#148).
157fn internal<E: std::fmt::Display>(e: E) -> Status {
158    eprintln!("gonzalod: internal error: {e}");
159    Status::internal("internal error")
160}
161
162#[tonic::async_trait]
163impl Gonzalo for GrpcAdapter {
164    async fn get(&self, req: Request<GetRequest>) -> Result<Response<GetResponse>, Status> {
165        let (metadata, _ext, r) = req.into_parts();
166        self.authorize(&metadata, Access::Read, &r.namespace)?;
167        let key = RecordKey::new(r.namespace, r.collection, r.id);
168        let rec = self.service.get(&key).await.map_err(internal)?;
169        Ok(Response::new(get_response(rec)?))
170    }
171
172    async fn put(&self, req: Request<PutRequest>) -> Result<Response<PutResponse>, Status> {
173        let (metadata, _ext, r) = req.into_parts();
174        // Authenticate BEFORE deserializing attacker-controlled JSON (#146): an
175        // unauthenticated caller is rejected without ever feeding its body to
176        // serde. Only then parse the body (malformed input is the caller's
177        // error → invalid_argument, not internal) and authorize the write
178        // against the namespace named in the record's key.
179        let (principal, mut record, expected) = self.authorize_put(&metadata, &r)?;
180        // Stamp the author from the authenticated principal — unforgeable (ADR
181        // 0015). Open mode (no auth) leaves the record's author untouched.
182        if principal.is_authenticated() {
183            record.meta.author = Identity::new(principal.name());
184        }
185        let outcome = self
186            .service
187            .put(record, expected)
188            .await
189            .map_err(put_error)?;
190        Ok(Response::new(put_response(outcome)?))
191    }
192
193    async fn delete(
194        &self,
195        req: Request<DeleteRequest>,
196    ) -> Result<Response<DeleteResponse>, Status> {
197        let (metadata, _ext, r) = req.into_parts();
198        // Authenticate BEFORE deserializing attacker-controlled JSON (#146), then
199        // parse the precondition and the optional claimed-author (malformed
200        // input is the caller's error → invalid_argument), authorize the write
201        // against the path's namespace, and build the key from
202        // (namespace, collection, id). The tombstone's author follows
203        // `Principal::delete_author` (gonzalo#203, mirrors `put_raw`'s
204        // authorship rule): a non-admin is always stamped as itself and any
205        // claimed author is ignored; an admin or open mode may name the
206        // deleter via `author_json`; and otherwise an authenticated admin is
207        // stamped as itself while open mode keeps the prior author (no
208        // identity to stamp, ADR 0015).
209        let principal = self.authenticate(&metadata)?;
210        let expected: Option<Revision> = serde_json::from_slice(&r.expected_json)
211            .map_err(|e| Status::invalid_argument(e.to_string()))?;
212        let claimed: Option<Identity> = if r.author_json.is_empty() {
213            None
214        } else {
215            Some(
216                serde_json::from_slice(&r.author_json)
217                    .map_err(|e| Status::invalid_argument(e.to_string()))?,
218            )
219        };
220        self.check_access(&principal, Access::Write, &r.namespace)?;
221        let author = principal.delete_author(claimed);
222        let key = RecordKey::new(r.namespace, r.collection, r.id);
223        let outcome = self
224            .service
225            .delete_as(&key, expected, author)
226            .await
227            .map_err(internal)?;
228        let (outcome, payload_json) = delete_outcome_parts(outcome)?;
229        Ok(Response::new(DeleteResponse {
230            outcome,
231            payload_json,
232        }))
233    }
234
235    async fn list(&self, req: Request<ListRequest>) -> Result<Response<ListResponse>, Status> {
236        let (metadata, _ext, r) = req.into_parts();
237        let prefix = self.authorize_list(&metadata, r)?;
238        let keys = self.service.list(&prefix).await.map_err(internal)?;
239        Ok(Response::new(list_response(&keys)?))
240    }
241
242    async fn get_raw(&self, req: Request<GetRequest>) -> Result<Response<GetResponse>, Status> {
243        let (metadata, _ext, r) = req.into_parts();
244        self.authorize(&metadata, Access::Read, &r.namespace)?;
245        let key = RecordKey::new(r.namespace, r.collection, r.id);
246        let rec = self.service.get_raw(&key).await.map_err(internal)?;
247        Ok(Response::new(get_response(rec)?))
248    }
249
250    async fn list_raw(&self, req: Request<ListRequest>) -> Result<Response<ListResponse>, Status> {
251        let (metadata, _ext, r) = req.into_parts();
252        // Same authorization as `List`: unscoped requires `read` on `"*"`.
253        let prefix = self.authorize_list(&metadata, r)?;
254        let keys = self.service.list_raw(&prefix).await.map_err(internal)?;
255        Ok(Response::new(list_response(&keys)?))
256    }
257
258    async fn put_raw(&self, req: Request<PutRequest>) -> Result<Response<PutResponse>, Status> {
259        let (metadata, _ext, r) = req.into_parts();
260        // Replication write (gonzalo#203), admin only (gonzalo#342) to match
261        // `Purge` and the HTTP route: this path reaches `plan_put_raw`, so a
262        // plain `Write` principal could otherwise discard a deleted manifest's
263        // restore window (ADR 0029/0030). The admin check runs before
264        // deserializing the caller's JSON (#146). Authorship stays unforgeable
265        // (ADR 0015) by construction: the replication credential is an admin
266        // token, so the record keeps its original author, and no other
267        // principal reaches this RPC.
268        self.authorize_admin(&metadata, "a raw record write")?;
269        let record: Record = serde_json::from_slice(&r.record_json)
270            .map_err(|e| Status::invalid_argument(e.to_string()))?;
271        let expected: Option<Revision> = serde_json::from_slice(&r.expected_json)
272            .map_err(|e| Status::invalid_argument(e.to_string()))?;
273        let outcome = self
274            .service
275            .put_raw(record, expected)
276            .await
277            .map_err(put_error)?;
278        Ok(Response::new(put_response(outcome)?))
279    }
280
281    async fn purge(&self, req: Request<PurgeRequest>) -> Result<Response<PurgeResponse>, Status> {
282        let (metadata, _ext, r) = req.into_parts();
283        // Admin check first, then parse the attacker-controlled precondition
284        // (malformed → invalid_argument, the caller's error).
285        self.authorize_admin(&metadata, "purge")?;
286        let expected: Revision = serde_json::from_slice(&r.expected_json)
287            .map_err(|e| Status::invalid_argument(e.to_string()))?;
288        let key = RecordKey::new(r.namespace, r.collection, r.id);
289        let outcome = self
290            .service
291            .purge(&key, expected)
292            .await
293            .map_err(purge_error)?;
294        let (outcome, payload_json) = delete_outcome_parts(outcome)?;
295        Ok(Response::new(PurgeResponse {
296            outcome,
297            payload_json,
298        }))
299    }
300
301    async fn ticket_sync(
302        &self,
303        req: Request<TicketSyncRequest>,
304    ) -> Result<Response<TicketSyncResponse>, Status> {
305        let (metadata, _ext, r) = req.into_parts();
306        // Ticket sync writes records in the `tickets` namespace.
307        self.authorize(&metadata, Access::Write, "tickets")?;
308        let conn: gonzalo_ticket_config::Connection = serde_json::from_slice(&r.connection_json)
309            .map_err(|e| Status::invalid_argument(e.to_string()))?;
310        let summary = self
311            .service
312            .ticket_sync(&conn, "gonzalod")
313            .await
314            .map_err(|e| match e {
315                // A misconfigured request is the caller's own input → safe to
316                // echo. An internal failure goes through `internal` so its
317                // detail is logged, not leaked (#148).
318                crate::service::TicketSyncError::BadRequest(m) => Status::invalid_argument(m),
319                crate::service::TicketSyncError::Internal(m) => internal(m),
320            })?;
321        Ok(Response::new(TicketSyncResponse {
322            imported: summary.imported as u64,
323            updated: summary.updated as u64,
324            unchanged: summary.unchanged as u64,
325        }))
326    }
327
328    async fn graph_definitions(
329        &self,
330        req: Request<GraphQueryRequest>,
331    ) -> Result<Response<GraphLocatedResponse>, Status> {
332        let (metadata, _ext, r) = req.into_parts();
333        self.authorize(&metadata, Access::Read, &r.repo)?;
334        let items = self
335            .service
336            .graph_definitions(&r.repo, &r.view_id, &r.name)
337            .await
338            .map_err(internal)?;
339        Ok(Response::new(located_response(&items)?))
340    }
341
342    async fn graph_references_to(
343        &self,
344        req: Request<GraphQueryRequest>,
345    ) -> Result<Response<GraphLocatedResponse>, Status> {
346        let (metadata, _ext, r) = req.into_parts();
347        self.authorize(&metadata, Access::Read, &r.repo)?;
348        let items = self
349            .service
350            .graph_references_to(&r.repo, &r.view_id, &r.name)
351            .await
352            .map_err(internal)?;
353        Ok(Response::new(located_response(&items)?))
354    }
355
356    async fn graph_callers_of(
357        &self,
358        req: Request<GraphQueryRequest>,
359    ) -> Result<Response<GraphNamesResponse>, Status> {
360        let (metadata, _ext, r) = req.into_parts();
361        self.authorize(&metadata, Access::Read, &r.repo)?;
362        let names = self
363            .service
364            .graph_callers_of(&r.repo, &r.view_id, &r.name)
365            .await
366            .map_err(internal)?;
367        Ok(Response::new(GraphNamesResponse { names }))
368    }
369
370    async fn graph_callees(
371        &self,
372        req: Request<GraphQueryRequest>,
373    ) -> Result<Response<GraphNamesResponse>, Status> {
374        let (metadata, _ext, r) = req.into_parts();
375        self.authorize(&metadata, Access::Read, &r.repo)?;
376        let names = self
377            .service
378            .graph_callees(&r.repo, &r.view_id, &r.name)
379            .await
380            .map_err(internal)?;
381        Ok(Response::new(GraphNamesResponse { names }))
382    }
383
384    async fn graph_impact(
385        &self,
386        req: Request<GraphQueryRequest>,
387    ) -> Result<Response<GraphNamesResponse>, Status> {
388        let (metadata, _ext, r) = req.into_parts();
389        self.authorize(&metadata, Access::Read, &r.repo)?;
390        let names = self
391            .service
392            .graph_impact_names(&r.repo, &r.view_id, &r.name)
393            .await
394            .map_err(internal)?;
395        Ok(Response::new(GraphNamesResponse { names }))
396    }
397
398    async fn put_blob(
399        &self,
400        req: Request<PutBlobRequest>,
401    ) -> Result<Response<PutBlobResponse>, Status> {
402        let (metadata, _ext, r) = req.into_parts();
403        self.authorize(&metadata, Access::Write, BLOB_NS)?;
404        // Verify the content hashes to the advertised value before writing —
405        // same integrity check as the HTTP hash-addressed PUT.
406        let computed = ContentHash::of(&r.content);
407        if computed.0 != r.hash {
408            return Err(Status::invalid_argument(
409                "blob content does not match the advertised hash",
410            ));
411        }
412        let hash = self.service.put_blob(&r.content).await.map_err(internal)?;
413        Ok(Response::new(PutBlobResponse { hash: hash.0 }))
414    }
415
416    async fn get_blob(
417        &self,
418        req: Request<GetBlobRequest>,
419    ) -> Result<Response<GetBlobResponse>, Status> {
420        let (metadata, _ext, r) = req.into_parts();
421        self.authorize(&metadata, Access::Read, BLOB_NS)?;
422        let found = self
423            .service
424            .get_blob(&ContentHash(r.hash))
425            .await
426            .map_err(internal)?;
427        let resp = match found {
428            Some(content) => GetBlobResponse {
429                found: true,
430                content,
431            },
432            None => GetBlobResponse {
433                found: false,
434                content: Vec::new(),
435            },
436        };
437        Ok(Response::new(resp))
438    }
439
440    async fn has_blob(
441        &self,
442        req: Request<HasBlobRequest>,
443    ) -> Result<Response<HasBlobResponse>, Status> {
444        let (metadata, _ext, r) = req.into_parts();
445        self.authorize(&metadata, Access::Read, BLOB_NS)?;
446        let found = self
447            .service
448            .has_blob(&ContentHash(r.hash))
449            .await
450            .map_err(internal)?;
451        Ok(Response::new(HasBlobResponse { found }))
452    }
453
454    async fn list_blobs(
455        &self,
456        req: Request<ListBlobsRequest>,
457    ) -> Result<Response<ListBlobsResponse>, Status> {
458        let (metadata, _ext, _r) = req.into_parts();
459        self.authorize(&metadata, Access::Read, BLOB_NS)?;
460        let entries = self.service.list_blobs().await.map_err(internal)?;
461        Ok(Response::new(ListBlobsResponse {
462            entries: entries
463                .into_iter()
464                .map(|e| gonzalo_proto::v1::BlobEntry {
465                    hash: e.hash.0,
466                    modified_unix_ms: e.modified_unix_ms,
467                })
468                .collect(),
469        }))
470    }
471
472    async fn delete_blob(
473        &self,
474        req: Request<DeleteBlobRequest>,
475    ) -> Result<Response<DeleteBlobResponse>, Status> {
476        let (metadata, _ext, r) = req.into_parts();
477        self.authorize(&metadata, Access::Write, BLOB_NS)?;
478        self.service
479            .delete_blob(&ContentHash(r.hash))
480            .await
481            .map_err(internal)?;
482        Ok(Response::new(DeleteBlobResponse {}))
483    }
484}
485
486/// Map a put failure. `CoreError::NotFound` (`expected` names a revision the
487/// store does not hold) is the caller's precondition failing →
488/// `FailedPrecondition`, which the client maps back to `NotFound`. Everything
489/// else is opaque (#148).
490fn put_error(e: CoreError) -> Status {
491    match e {
492        CoreError::NotFound(key) => Status::failed_precondition(format!("record not found: {key}")),
493        // A call the store will never accept (a consumer put of a tombstone):
494        // the caller's fault, not an outage, so not `Internal` (gonzalo#299).
495        CoreError::Invalid(reason) => Status::invalid_argument(reason),
496        other => internal(other),
497    }
498}
499
500/// `Status` for a purge error. A call the store will never accept — a purge of a
501/// live record (gonzalo#341) — is the caller's fault, not an outage, so it is
502/// `invalid_argument` rather than `internal` (same rule as `put_error`,
503/// gonzalo#299).
504fn purge_error(e: CoreError) -> Status {
505    match e {
506        CoreError::Invalid(reason) => Status::invalid_argument(reason),
507        other => internal(other),
508    }
509}
510
511/// `PutResponse` for a `PutResult` (shared by `Put` and `PutRaw`).
512#[allow(clippy::result_large_err)]
513fn put_response(outcome: PutResult) -> Result<PutResponse, Status> {
514    Ok(match outcome {
515        PutResult::Committed(rev) => PutResponse {
516            outcome: "committed".into(),
517            payload_json: serde_json::to_vec(&rev).map_err(internal)?,
518        },
519        PutResult::Conflict(c) => PutResponse {
520            outcome: "conflict".into(),
521            payload_json: serde_json::to_vec(&*c).map_err(internal)?,
522        },
523    })
524}
525
526/// `GetResponse` for an optional record (shared by `Get` and `GetRaw`).
527#[allow(clippy::result_large_err)]
528fn get_response(rec: Option<Record>) -> Result<GetResponse, Status> {
529    Ok(match rec {
530        Some(rec) => GetResponse {
531            found: true,
532            record_json: serde_json::to_vec(&rec).map_err(internal)?,
533        },
534        None => GetResponse {
535            found: false,
536            record_json: Vec::new(),
537        },
538    })
539}
540
541/// `ListResponse` of JSON-encoded keys (shared by `List` and `ListRaw`).
542#[allow(clippy::result_large_err)]
543fn list_response(keys: &[RecordKey]) -> Result<ListResponse, Status> {
544    let keys_json = keys
545        .iter()
546        .map(serde_json::to_vec)
547        .collect::<std::result::Result<Vec<_>, _>>()
548        .map_err(internal)?;
549    Ok(ListResponse { keys_json })
550}
551
552/// `(outcome, payload_json)` for a `DeleteResult` (shared by `Delete` and
553/// `Purge`): `"deleted"` + empty, or `"conflict"` + JSON of `Conflict`.
554#[allow(clippy::result_large_err)]
555fn delete_outcome_parts(outcome: DeleteResult) -> Result<(String, Vec<u8>), Status> {
556    Ok(match outcome {
557        DeleteResult::Deleted => ("deleted".into(), Vec::new()),
558        DeleteResult::Conflict(c) => (
559            "conflict".into(),
560            serde_json::to_vec(&*c).map_err(internal)?,
561        ),
562    })
563}
564
565/// JSON-encode each located item into a `GraphLocatedResponse` (the shared
566/// JSON-in-bytes convention).
567// `Status` is large but fixed by tonic's API, so the large-err lint can't be
568// acted on (same as `serve_grpc`).
569#[allow(clippy::result_large_err)]
570fn located_response<T: Serialize>(items: &[T]) -> Result<GraphLocatedResponse, Status> {
571    let items_json = items
572        .iter()
573        .map(serde_json::to_vec)
574        .collect::<std::result::Result<Vec<_>, _>>()
575        .map_err(internal)?;
576    Ok(GraphLocatedResponse { items_json })
577}
578
579/// Serve gRPC on an already-bound listener until the process ends. `auth`
580/// governs per-call namespace authorization (ADR 0015); `Auth::Disabled` serves
581/// open.
582///
583/// # The decode ceiling applies to every RPC, not just blobs
584///
585/// `max_decoding_message_size` below is raised to `max_blob_size` so 64 MiB
586/// blobs can transit `PutBlob`. tonic scopes that setting **per server**, not
587/// per method — there is no per-RPC knob — so it raises the decode ceiling for
588/// *every* RPC, including record `Put`, from tonic's 4 MiB default to
589/// `GONZALO_MAX_BLOB_SIZE`.
590///
591/// Two consequences worth knowing before you change the env var:
592///
593/// - Raising `GONZALO_MAX_BLOB_SIZE` to allow larger blobs also allows larger
594///   records. That is a surprising reach for a knob named for blobs.
595/// - It is asymmetric with HTTP, where the blob body limit is scoped to the
596///   blob sub-router, so record `PUT`s keep axum's 2 MiB default (`http.rs`).
597///
598/// This is bounded and every call is authenticated, so it is documented rather
599/// than closed (#194). Closing it would mean a tower layer inspecting
600/// content-length by gRPC method path — real machinery for a benign gap.
601pub async fn serve_grpc(
602    listener: tokio::net::TcpListener,
603    service: Service,
604    auth: Arc<Auth>,
605) -> Result<(), tonic::transport::Error> {
606    let max_blob = service.max_blob_size();
607    let adapter = GrpcAdapter::with_auth(service, auth);
608    let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
609    tonic::transport::Server::builder()
610        // Per-server, so this governs record RPCs too — see the doc comment.
611        .add_service(GonzaloServer::new(adapter).max_decoding_message_size(max_blob))
612        .serve_with_incoming(incoming)
613        .await
614}
615
616#[cfg(test)]
617mod tests {
618    use super::*;
619    use gonzalo_core::{BlobStore, Identity, Manifest, Meta, RecordKind, Store};
620    use gonzalo_graph::{Located, Symbol, build_rust};
621    use gonzalo_store_fs::FsStore;
622    use std::collections::BTreeMap;
623    use std::sync::Arc;
624
625    /// Seed one view (`r`/`main`) with two slices and return a gRPC adapter over it.
626    async fn seeded_adapter() -> GrpcAdapter {
627        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
628        let mut manifest = Manifest::new();
629        for (path, src) in [
630            ("lib.rs", "fn helper() {}"),
631            ("main.rs", "fn main() { helper(); }"),
632        ] {
633            let hash = fs
634                .put_blob(&build_rust(src).to_slice_bytes())
635                .await
636                .unwrap();
637            manifest.insert(path, hash);
638        }
639        let body = manifest.to_body();
640        let record = Record {
641            revision: Revision::initial(body.bytes()),
642            parent: None,
643            body,
644            kind: RecordKind::GraphManifest,
645            meta: Meta {
646                author: Identity::new("tester"),
647                origin_system: "test".into(),
648                created: 0,
649                updated: 0,
650                labels: BTreeMap::new(),
651            },
652            links: Vec::new(),
653            key: Manifest::key("r", "main"),
654            ancestors: Vec::new(),
655            deleted_at: None,
656            deleted_blob: None,
657            deleted_kind: None,
658        };
659        let outcome = fs.put(record, None).await.unwrap();
660        assert!(matches!(outcome, PutResult::Committed(_)));
661        GrpcAdapter::new(Service::new(fs.clone(), fs))
662    }
663
664    fn query(name: &str) -> Request<GraphQueryRequest> {
665        Request::new(GraphQueryRequest {
666            repo: "r".into(),
667            view_id: "main".into(),
668            name: name.into(),
669        })
670    }
671
672    #[tokio::test]
673    async fn graph_definitions_returns_located_json() {
674        let adapter = seeded_adapter().await;
675        let resp = adapter
676            .graph_definitions(query("helper"))
677            .await
678            .unwrap()
679            .into_inner();
680        assert_eq!(resp.items_json.len(), 1);
681        let located: Located<Symbol> = serde_json::from_slice(&resp.items_json[0]).unwrap();
682        assert_eq!(located.path, "lib.rs");
683        assert_eq!(located.item.name, "helper");
684    }
685
686    // --- namespace-scoped auth (ADR 0015) ---
687
688    use std::collections::HashMap;
689
690    /// A `writer` principal scoped to the `memory` namespace, plus an `admin`.
691    fn scoped_auth() -> Arc<Auth> {
692        Arc::new(Auth::Enabled(HashMap::from([
693            (
694                "wtok".to_string(),
695                Principal::new("writer", vec!["memory".into()], vec!["memory".into()]),
696            ),
697            ("atok".to_string(), Principal::admin("admin")),
698        ])))
699    }
700
701    fn with_token<T>(msg: T, token: &str) -> Request<T> {
702        let mut req = Request::new(msg);
703        req.metadata_mut()
704            .insert("authorization", format!("Bearer {token}").parse().unwrap());
705        req
706    }
707
708    fn fs_adapter(auth: Arc<Auth>) -> GrpcAdapter {
709        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
710        GrpcAdapter::with_auth(Service::new(fs.clone(), fs), auth)
711    }
712
713    fn get_req(namespace: &str) -> GetRequest {
714        GetRequest {
715            namespace: namespace.into(),
716            collection: "col".into(),
717            id: "x".into(),
718        }
719    }
720
721    fn put_req(namespace: &str, author: &str) -> PutRequest {
722        let record = Record {
723            revision: Revision::initial(b"{}"),
724            parent: None,
725            body: gonzalo_core::Body::Inline(b"{}".to_vec()),
726            kind: RecordKind::MemoryTier,
727            meta: Meta {
728                author: Identity::new(author),
729                origin_system: "test".into(),
730                created: 0,
731                updated: 0,
732                labels: BTreeMap::new(),
733            },
734            links: Vec::new(),
735            key: RecordKey::new(namespace, "col", "x"),
736            ancestors: Vec::new(),
737            deleted_at: None,
738            deleted_blob: None,
739            deleted_kind: None,
740        };
741        PutRequest {
742            record_json: serde_json::to_vec(&record).unwrap(),
743            expected_json: serde_json::to_vec(&Option::<Revision>::None).unwrap(),
744        }
745    }
746
747    /// A well-formed `PutRequest` whose `record_json` is not valid JSON.
748    fn malformed_put_req() -> PutRequest {
749        PutRequest {
750            record_json: b"definitely not a record".to_vec(),
751            expected_json: serde_json::to_vec(&Option::<Revision>::None).unwrap(),
752        }
753    }
754
755    /// A store whose every op fails, to force the `internal` (server-error) path.
756    struct DownStore;
757
758    fn leaky() -> gonzalo_core::CoreError {
759        gonzalo_core::CoreError::Backend("s3://secret-bucket".into())
760    }
761
762    #[async_trait::async_trait]
763    impl Store for DownStore {
764        async fn get(&self, _key: &RecordKey) -> gonzalo_core::Result<Option<Record>> {
765            Err(gonzalo_core::CoreError::Backend(
766                "/var/lib/gonzalo/graphs/secret.sqlite unreachable".into(),
767            ))
768        }
769        async fn put(
770            &self,
771            _record: Record,
772            _expected: Option<Revision>,
773        ) -> gonzalo_core::Result<PutResult> {
774            Err(leaky())
775        }
776        async fn list(
777            &self,
778            _prefix: &gonzalo_core::KeyPrefix,
779        ) -> gonzalo_core::Result<Vec<RecordKey>> {
780            Err(leaky())
781        }
782        async fn delete_as(
783            &self,
784            _key: &RecordKey,
785            _expected: Option<Revision>,
786            _author: Option<Identity>,
787        ) -> gonzalo_core::Result<DeleteResult> {
788            Err(leaky())
789        }
790        async fn get_raw(&self, _key: &RecordKey) -> gonzalo_core::Result<Option<Record>> {
791            Err(leaky())
792        }
793        async fn list_raw(
794            &self,
795            _prefix: &gonzalo_core::KeyPrefix,
796        ) -> gonzalo_core::Result<Vec<RecordKey>> {
797            Err(leaky())
798        }
799        async fn put_raw(
800            &self,
801            _record: Record,
802            _expected: Option<Revision>,
803        ) -> gonzalo_core::Result<PutResult> {
804            Err(leaky())
805        }
806        async fn purge(
807            &self,
808            _key: &RecordKey,
809            _expected: Revision,
810        ) -> gonzalo_core::Result<DeleteResult> {
811            Err(leaky())
812        }
813    }
814
815    #[tokio::test]
816    async fn put_authenticates_before_deserializing() {
817        // A malformed body with NO token is rejected at authentication, before
818        // serde ever runs on the attacker-controlled JSON (#146).
819        let adapter = fs_adapter(scoped_auth());
820        let err = adapter
821            .put(Request::new(malformed_put_req()))
822            .await
823            .unwrap_err();
824        assert_eq!(err.code(), tonic::Code::Unauthenticated);
825    }
826
827    #[tokio::test]
828    async fn put_malformed_body_is_invalid_argument_not_internal() {
829        // An authorized caller sending a malformed body gets InvalidArgument —
830        // the caller's own bad input — not Internal (#146).
831        let adapter = fs_adapter(scoped_auth());
832        let err = adapter
833            .put(with_token(malformed_put_req(), "wtok"))
834            .await
835            .unwrap_err();
836        assert_eq!(err.code(), tonic::Code::InvalidArgument);
837    }
838
839    #[test]
840    fn an_invalid_call_is_invalid_argument_not_internal() {
841        // A consumer put of a tombstone can never succeed: the caller's fault,
842        // so `InvalidArgument`, not an outage-looking `Internal` (#299). The
843        // reason is safe to return — it names the API to use, not the store.
844        let status = put_error(CoreError::Invalid(
845            gonzalo_core::CONSUMER_TOMBSTONE_REJECTED.to_string(),
846        ));
847        assert_eq!(status.code(), tonic::Code::InvalidArgument);
848        assert!(status.message().contains("delete_as"), "{status:?}");
849    }
850
851    #[tokio::test]
852    async fn backend_error_is_opaque() {
853        // A forced backend failure yields an opaque Internal status; the leaky
854        // path/bucket detail never reaches the client (#148).
855        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
856        let adapter = GrpcAdapter::new(Service::new(Arc::new(DownStore), fs));
857        let err = adapter.get(Request::new(get_req("any"))).await.unwrap_err();
858        assert_eq!(err.code(), tonic::Code::Internal);
859        assert_eq!(err.message(), "internal error");
860        assert!(!err.message().contains("secret"));
861    }
862
863    #[tokio::test]
864    async fn missing_token_is_unauthenticated() {
865        let adapter = fs_adapter(scoped_auth());
866        let err = adapter
867            .get(Request::new(get_req("memory")))
868            .await
869            .unwrap_err();
870        assert_eq!(err.code(), tonic::Code::Unauthenticated);
871    }
872
873    #[tokio::test]
874    async fn read_is_allowed_in_scope_denied_out_of_scope() {
875        let adapter = fs_adapter(scoped_auth());
876        // In-scope read succeeds (record absent → found:false, but authorized).
877        assert!(
878            adapter
879                .get(with_token(get_req("memory"), "wtok"))
880                .await
881                .is_ok()
882        );
883        // Out-of-scope read is denied.
884        let err = adapter
885            .get(with_token(get_req("secrets"), "wtok"))
886            .await
887            .unwrap_err();
888        assert_eq!(err.code(), tonic::Code::PermissionDenied);
889    }
890
891    #[tokio::test]
892    async fn write_is_scoped_and_author_is_stamped() {
893        let adapter = fs_adapter(scoped_auth());
894        // Write outside scope is denied.
895        let err = adapter
896            .put(with_token(put_req("secrets", "writer"), "wtok"))
897            .await
898            .unwrap_err();
899        assert_eq!(err.code(), tonic::Code::PermissionDenied);
900
901        // In-scope write commits — even though the client claimed author
902        // "forged", the daemon stamps the authenticated principal.
903        adapter
904            .put(with_token(put_req("memory", "forged"), "wtok"))
905            .await
906            .unwrap();
907        let resp = adapter
908            .get(with_token(get_req("memory"), "wtok"))
909            .await
910            .unwrap()
911            .into_inner();
912        let record: Record = serde_json::from_slice(&resp.record_json).unwrap();
913        assert_eq!(record.meta.author, Identity::new("writer"));
914    }
915
916    #[tokio::test]
917    async fn list_without_namespace_requires_admin() {
918        let adapter = fs_adapter(scoped_auth());
919        let scoped = adapter
920            .list(with_token(ListRequest::default(), "wtok"))
921            .await
922            .unwrap_err();
923        assert_eq!(scoped.code(), tonic::Code::PermissionDenied);
924        // Admin (wildcard) may list across all namespaces.
925        assert!(
926            adapter
927                .list(with_token(ListRequest::default(), "atok"))
928                .await
929                .is_ok()
930        );
931    }
932
933    // --- blob RPCs (#184) ---
934
935    #[tokio::test]
936    async fn grpc_blob_roundtrip_open() {
937        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
938        let adapter = GrpcAdapter::new(Service::new(fs.clone(), fs));
939        let content = b"grpc blob body".to_vec();
940        let hash = gonzalo_core::ContentHash::of(&content).0;
941
942        let put = adapter
943            .put_blob(Request::new(PutBlobRequest {
944                hash: hash.clone(),
945                content: content.clone(),
946            }))
947            .await
948            .unwrap()
949            .into_inner();
950        assert_eq!(put.hash, hash);
951
952        let got = adapter
953            .get_blob(Request::new(GetBlobRequest { hash: hash.clone() }))
954            .await
955            .unwrap()
956            .into_inner();
957        assert!(got.found);
958        assert_eq!(got.content, content);
959
960        let listed = adapter
961            .list_blobs(Request::new(ListBlobsRequest {}))
962            .await
963            .unwrap()
964            .into_inner();
965        assert_eq!(listed.entries.len(), 1);
966        assert_eq!(listed.entries[0].hash, hash);
967        assert_ne!(
968            listed.entries[0].modified_unix_ms, 0,
969            "a zero timestamp means the substrate never reported one"
970        );
971
972        adapter
973            .delete_blob(Request::new(DeleteBlobRequest { hash: hash.clone() }))
974            .await
975            .unwrap();
976        let gone = adapter
977            .get_blob(Request::new(GetBlobRequest { hash }))
978            .await
979            .unwrap()
980            .into_inner();
981        assert!(!gone.found);
982    }
983
984    #[tokio::test]
985    async fn grpc_put_blob_hash_mismatch_is_invalid_argument() {
986        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
987        let adapter = GrpcAdapter::new(Service::new(fs.clone(), fs));
988        let err = adapter
989            .put_blob(Request::new(PutBlobRequest {
990                hash: gonzalo_core::ContentHash::of(b"not the body").0,
991                content: b"the body".to_vec(),
992            }))
993            .await
994            .unwrap_err();
995        assert_eq!(err.code(), tonic::Code::InvalidArgument);
996    }
997
998    /// `HasBlob` reports presence and nothing else — there is no content field
999    /// to fill, so a client confirming a blob survived a commit cannot be made
1000    /// to pay for its bytes (#329).
1001    #[tokio::test]
1002    async fn grpc_has_blob_reports_presence() {
1003        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
1004        let adapter = GrpcAdapter::new(Service::new(fs.clone(), fs));
1005        let content = b"grpc presence".to_vec();
1006        let hash = gonzalo_core::ContentHash::of(&content).0;
1007
1008        let before = adapter
1009            .has_blob(Request::new(HasBlobRequest { hash: hash.clone() }))
1010            .await
1011            .unwrap()
1012            .into_inner();
1013        assert!(!before.found, "nothing stored yet");
1014
1015        adapter
1016            .put_blob(Request::new(PutBlobRequest {
1017                hash: hash.clone(),
1018                content,
1019            }))
1020            .await
1021            .unwrap();
1022
1023        let after = adapter
1024            .has_blob(Request::new(HasBlobRequest { hash }))
1025            .await
1026            .unwrap()
1027            .into_inner();
1028        assert!(after.found, "the blob just written is present");
1029    }
1030
1031    /// Presence is a read of `_blobs`: without that scope a principal must not
1032    /// learn whether a blob exists.
1033    #[tokio::test]
1034    async fn grpc_has_blob_requires_blobs_scope() {
1035        // `scoped_auth()` grants `memory` only, not `_blobs`.
1036        let adapter = fs_adapter(scoped_auth());
1037        let content = b"scoped presence".to_vec();
1038        let hash = gonzalo_core::ContentHash::of(&content).0;
1039        adapter
1040            .put_blob(with_token(
1041                PutBlobRequest {
1042                    hash: hash.clone(),
1043                    content,
1044                },
1045                "atok",
1046            ))
1047            .await
1048            .unwrap();
1049
1050        let denied = adapter
1051            .has_blob(with_token(HasBlobRequest { hash: hash.clone() }, "wtok"))
1052            .await
1053            .unwrap_err();
1054        assert_eq!(denied.code(), tonic::Code::PermissionDenied);
1055
1056        let allowed = adapter
1057            .has_blob(with_token(HasBlobRequest { hash }, "atok"))
1058            .await
1059            .unwrap()
1060            .into_inner();
1061        assert!(allowed.found);
1062    }
1063
1064    #[tokio::test]
1065    async fn grpc_blob_ops_require_blobs_scope() {
1066        // `scoped_auth()` grants `memory` only, not `_blobs`.
1067        let adapter = fs_adapter(scoped_auth());
1068        let content = b"scoped".to_vec();
1069        let hash = gonzalo_core::ContentHash::of(&content).0;
1070
1071        let denied = adapter
1072            .put_blob(with_token(
1073                PutBlobRequest {
1074                    hash: hash.clone(),
1075                    content: content.clone(),
1076                },
1077                "wtok",
1078            ))
1079            .await
1080            .unwrap_err();
1081        assert_eq!(denied.code(), tonic::Code::PermissionDenied);
1082
1083        // Admin token succeeds.
1084        adapter
1085            .put_blob(with_token(PutBlobRequest { hash, content }, "atok"))
1086            .await
1087            .unwrap();
1088    }
1089
1090    #[tokio::test]
1091    async fn graph_name_queries_return_names() {
1092        let adapter = seeded_adapter().await;
1093        let impact = adapter
1094            .graph_impact(query("helper"))
1095            .await
1096            .unwrap()
1097            .into_inner();
1098        assert_eq!(impact.names, vec!["main".to_string()]);
1099        let callees = adapter
1100            .graph_callees(query("main"))
1101            .await
1102            .unwrap()
1103            .into_inner();
1104        assert_eq!(callees.names, vec!["helper".to_string()]);
1105        let callers = adapter
1106            .graph_callers_of(query("helper"))
1107            .await
1108            .unwrap()
1109            .into_inner();
1110        assert_eq!(callers.names, vec!["main".to_string()]);
1111    }
1112
1113    // --- replication RPCs: GetRaw / ListRaw / PutRaw / Purge (gonzalo#203) ---
1114
1115    /// `reader` reads `memory` only; `writer` reads and writes `memory`; `admin`.
1116    fn tomb_auth() -> Arc<Auth> {
1117        Arc::new(Auth::Enabled(HashMap::from([
1118            (
1119                "rtok".to_string(),
1120                Principal::new("reader", vec!["memory".into()], vec![]),
1121            ),
1122            (
1123                "wtok".to_string(),
1124                Principal::new("writer", vec!["memory".into()], vec!["memory".into()]),
1125            ),
1126            ("atok".to_string(), Principal::admin("admin")),
1127        ])))
1128    }
1129
1130    fn delete_req(namespace: &str) -> DeleteRequest {
1131        DeleteRequest {
1132            namespace: namespace.into(),
1133            collection: "col".into(),
1134            id: "x".into(),
1135            expected_json: serde_json::to_vec(&Option::<Revision>::None).unwrap(),
1136            author_json: Vec::new(),
1137        }
1138    }
1139
1140    fn purge_req(namespace: &str, expected: &Revision) -> PurgeRequest {
1141        PurgeRequest {
1142            namespace: namespace.into(),
1143            collection: "col".into(),
1144            id: "x".into(),
1145            expected_json: serde_json::to_vec(expected).unwrap(),
1146        }
1147    }
1148
1149    /// A `PutRequest` for memory/col/x with an explicit author and precondition.
1150    fn put_req_with(author: &str, payload: &[u8], expected: Option<Revision>) -> PutRequest {
1151        let body = gonzalo_core::Body::Inline(payload.to_vec());
1152        let record = Record {
1153            revision: Revision::initial(body.bytes()),
1154            parent: None,
1155            body,
1156            kind: RecordKind::MemoryTier,
1157            meta: Meta {
1158                author: Identity::new(author),
1159                origin_system: "test".into(),
1160                created: 0,
1161                updated: 0,
1162                labels: BTreeMap::new(),
1163            },
1164            links: Vec::new(),
1165            ancestors: Vec::new(),
1166            deleted_at: None,
1167            deleted_blob: None,
1168            key: RecordKey::new("memory", "col", "x"),
1169            deleted_kind: None,
1170        };
1171        PutRequest {
1172            record_json: serde_json::to_vec(&record).unwrap(),
1173            expected_json: serde_json::to_vec(&expected).unwrap(),
1174        }
1175    }
1176
1177    fn memory_list() -> ListRequest {
1178        ListRequest {
1179            namespace: Some("memory".into()),
1180            collection: None,
1181        }
1182    }
1183
1184    async fn raw_record(adapter: &GrpcAdapter) -> Option<Record> {
1185        let raw = adapter
1186            .get_raw(with_token(get_req("memory"), "atok"))
1187            .await
1188            .unwrap()
1189            .into_inner();
1190        raw.found
1191            .then(|| serde_json::from_slice(&raw.record_json).unwrap())
1192    }
1193
1194    /// Put memory/col/x as `writer`, delete it with `delete_token`; return the
1195    /// tombstone.
1196    async fn seed_tombstone(adapter: &GrpcAdapter, delete_token: &str) -> Record {
1197        adapter
1198            .put(with_token(put_req("memory", "writer"), "wtok"))
1199            .await
1200            .unwrap();
1201        let del = adapter
1202            .delete(with_token(delete_req("memory"), delete_token))
1203            .await
1204            .unwrap()
1205            .into_inner();
1206        assert_eq!(del.outcome, "deleted");
1207        raw_record(adapter).await.expect("tombstone")
1208    }
1209
1210    fn decode_keys(resp: ListResponse) -> Vec<RecordKey> {
1211        resp.keys_json
1212            .iter()
1213            .map(|b| serde_json::from_slice(b).unwrap())
1214            .collect()
1215    }
1216
1217    #[tokio::test]
1218    async fn grpc_delete_leaves_a_stamped_tombstone_for_raw_reads() {
1219        let adapter = fs_adapter(tomb_auth());
1220        // `writer` wrote it; `admin` deletes it; the tombstone names the deleter.
1221        let tomb = seed_tombstone(&adapter, "atok").await;
1222        assert_eq!(tomb.kind, RecordKind::Tombstone);
1223        assert_eq!(tomb.revision.hash, gonzalo_core::tombstone_hash());
1224        assert_eq!(tomb.meta.author, Identity::new("admin"));
1225        let key = RecordKey::new("memory", "col", "x");
1226
1227        let got = adapter
1228            .get(with_token(get_req("memory"), "rtok"))
1229            .await
1230            .unwrap()
1231            .into_inner();
1232        assert!(!got.found);
1233        let live = decode_keys(
1234            adapter
1235                .list(with_token(memory_list(), "rtok"))
1236                .await
1237                .unwrap()
1238                .into_inner(),
1239        );
1240        assert!(!live.contains(&key));
1241
1242        let raw = adapter
1243            .get_raw(with_token(get_req("memory"), "rtok"))
1244            .await
1245            .unwrap()
1246            .into_inner();
1247        assert!(raw.found);
1248        let raw_keys = decode_keys(
1249            adapter
1250                .list_raw(with_token(memory_list(), "rtok"))
1251                .await
1252                .unwrap()
1253                .into_inner(),
1254        );
1255        assert!(raw_keys.contains(&key));
1256    }
1257
1258    #[tokio::test]
1259    async fn grpc_raw_reads_are_read_scoped() {
1260        let adapter = fs_adapter(tomb_auth());
1261        let err = adapter
1262            .get_raw(with_token(get_req("secrets"), "rtok"))
1263            .await
1264            .unwrap_err();
1265        assert_eq!(err.code(), tonic::Code::PermissionDenied);
1266        let err = adapter
1267            .get_raw(Request::new(get_req("memory")))
1268            .await
1269            .unwrap_err();
1270        assert_eq!(err.code(), tonic::Code::Unauthenticated);
1271    }
1272
1273    #[tokio::test]
1274    async fn grpc_unscoped_list_raw_requires_admin() {
1275        let adapter = fs_adapter(tomb_auth());
1276        for token in ["rtok", "wtok"] {
1277            let err = adapter
1278                .list_raw(with_token(ListRequest::default(), token))
1279                .await
1280                .unwrap_err();
1281            assert_eq!(err.code(), tonic::Code::PermissionDenied, "{token}");
1282        }
1283        assert!(
1284            adapter
1285                .list_raw(with_token(ListRequest::default(), "atok"))
1286                .await
1287                .is_ok()
1288        );
1289    }
1290
1291    #[tokio::test]
1292    async fn grpc_put_raw_requires_admin() {
1293        // Replication is admin-only (gonzalo#342): PutRaw reaches
1294        // `plan_put_raw`, so a namespace writer could otherwise discard a
1295        // deleted manifest's restore window. Nothing is written.
1296        let adapter = fs_adapter(tomb_auth());
1297        for token in ["rtok", "wtok"] {
1298            let err = adapter
1299                .put_raw(with_token(put_req_with("origin", b"{}", None), token))
1300                .await
1301                .unwrap_err();
1302            assert_eq!(err.code(), tonic::Code::PermissionDenied, "{token}");
1303        }
1304        assert_eq!(raw_record(&adapter).await, None);
1305    }
1306
1307    /// The admin check runs before the caller's JSON is deserialized (#146), so
1308    /// a non-admin sending garbage gets `PermissionDenied`, not
1309    /// `InvalidArgument` (gonzalo#342).
1310    #[tokio::test]
1311    async fn grpc_put_raw_authorizes_before_deserializing() {
1312        let adapter = fs_adapter(tomb_auth());
1313        let err = adapter
1314            .put_raw(with_token(malformed_put_req(), "wtok"))
1315            .await
1316            .unwrap_err();
1317        assert_eq!(err.code(), tonic::Code::PermissionDenied);
1318    }
1319
1320    #[tokio::test]
1321    async fn grpc_put_raw_by_admin_keeps_the_incoming_author() {
1322        // An admin token is the replication credential.
1323        let adapter = fs_adapter(tomb_auth());
1324        let resp = adapter
1325            .put_raw(with_token(put_req_with("origin", b"{}", None), "atok"))
1326            .await
1327            .unwrap()
1328            .into_inner();
1329        assert_eq!(resp.outcome, "committed");
1330        assert_eq!(
1331            raw_record(&adapter).await.expect("stored").meta.author,
1332            Identity::new("origin")
1333        );
1334    }
1335
1336    #[tokio::test]
1337    async fn grpc_put_raw_in_open_mode_keeps_the_incoming_author() {
1338        // Open mode's implicit principal is an admin.
1339        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
1340        let adapter = GrpcAdapter::new(Service::new(fs.clone(), fs));
1341        let resp = adapter
1342            .put_raw(Request::new(put_req_with("origin", b"{}", None)))
1343            .await
1344            .unwrap()
1345            .into_inner();
1346        assert_eq!(resp.outcome, "committed");
1347        let raw = adapter
1348            .get_raw(Request::new(get_req("memory")))
1349            .await
1350            .unwrap()
1351            .into_inner();
1352        let stored: Record = serde_json::from_slice(&raw.record_json).unwrap();
1353        assert_eq!(stored.meta.author, Identity::new("origin"));
1354    }
1355
1356    #[tokio::test]
1357    async fn grpc_put_raw_over_tombstone_none_conflicts_some_overwrites() {
1358        let adapter = fs_adapter(tomb_auth());
1359        let tomb = seed_tombstone(&adapter, "wtok").await;
1360
1361        // Replication is admin-only (gonzalo#342).
1362        let conflict = adapter
1363            .put_raw(with_token(put_req_with("peer", b"peer", None), "atok"))
1364            .await
1365            .unwrap()
1366            .into_inner();
1367        assert_eq!(conflict.outcome, "conflict");
1368
1369        let ok = adapter
1370            .put_raw(with_token(
1371                put_req_with("peer", b"peer", Some(tomb.revision.clone())),
1372                "atok",
1373            ))
1374            .await
1375            .unwrap()
1376            .into_inner();
1377        assert_eq!(ok.outcome, "committed");
1378        let stored = raw_record(&adapter).await.expect("stored");
1379        assert_eq!(stored.revision, Revision::initial(b"peer"));
1380    }
1381
1382    #[tokio::test]
1383    async fn grpc_put_not_found_is_failed_precondition() {
1384        let adapter = fs_adapter(tomb_auth());
1385        let never = Some(Revision::initial(b"never current"));
1386        let err = adapter
1387            .put_raw(with_token(put_req_with("w", b"{}", never.clone()), "atok"))
1388            .await
1389            .unwrap_err();
1390        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
1391
1392        seed_tombstone(&adapter, "wtok").await;
1393        let err = adapter
1394            .put(with_token(put_req_with("w", b"{}", never), "wtok"))
1395            .await
1396            .unwrap_err();
1397        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
1398    }
1399
1400    #[tokio::test]
1401    async fn grpc_purge_requires_admin() {
1402        let adapter = fs_adapter(tomb_auth());
1403        let tomb = seed_tombstone(&adapter, "wtok").await;
1404
1405        for token in ["rtok", "wtok"] {
1406            let err = adapter
1407                .purge(with_token(purge_req("memory", &tomb.revision), token))
1408                .await
1409                .unwrap_err();
1410            assert_eq!(err.code(), tonic::Code::PermissionDenied, "{token}");
1411        }
1412        let err = adapter
1413            .purge(Request::new(purge_req("memory", &tomb.revision)))
1414            .await
1415            .unwrap_err();
1416        assert_eq!(err.code(), tonic::Code::Unauthenticated);
1417
1418        let ok = adapter
1419            .purge(with_token(purge_req("memory", &tomb.revision), "atok"))
1420            .await
1421            .unwrap()
1422            .into_inner();
1423        assert_eq!(ok.outcome, "deleted");
1424        assert_eq!(raw_record(&adapter).await, None);
1425    }
1426
1427    /// gonzalo#341: the daemon refuses to purge a live record over gRPC too,
1428    /// and classifies it as the caller's error rather than an outage.
1429    #[tokio::test]
1430    async fn grpc_purge_of_a_live_record_is_invalid_argument() {
1431        let adapter = fs_adapter(tomb_auth());
1432        adapter
1433            .put(with_token(put_req("memory", "client"), "wtok"))
1434            .await
1435            .unwrap();
1436        let live = raw_record(&adapter).await.expect("live record");
1437        assert!(!live.is_tombstone());
1438
1439        let err = adapter
1440            .purge(with_token(purge_req("memory", &live.revision), "atok"))
1441            .await
1442            .unwrap_err();
1443        assert_eq!(err.code(), tonic::Code::InvalidArgument);
1444        assert!(
1445            err.message().contains("gonzalo delete"),
1446            "{}",
1447            err.message()
1448        );
1449
1450        // Still there, unchanged.
1451        let after = raw_record(&adapter).await.expect("live record survives");
1452        assert_eq!(after.revision, live.revision);
1453    }
1454
1455    #[tokio::test]
1456    async fn grpc_purge_stale_expected_is_conflict() {
1457        let adapter = fs_adapter(tomb_auth());
1458        let tomb = seed_tombstone(&adapter, "wtok").await;
1459        let resp = adapter
1460            .purge(with_token(
1461                purge_req("memory", &Revision::initial(b"not the tombstone")),
1462                "atok",
1463            ))
1464            .await
1465            .unwrap()
1466            .into_inner();
1467        assert_eq!(resp.outcome, "conflict");
1468        let conflict: gonzalo_core::Conflict = serde_json::from_slice(&resp.payload_json).unwrap();
1469        assert_eq!(conflict.current.revision, tomb.revision);
1470    }
1471
1472    #[tokio::test]
1473    async fn grpc_purge_authorizes_before_parsing() {
1474        let adapter = fs_adapter(tomb_auth());
1475        let garbage = PurgeRequest {
1476            namespace: "memory".into(),
1477            collection: "col".into(),
1478            id: "x".into(),
1479            expected_json: b"not json".to_vec(),
1480        };
1481        let err = adapter
1482            .purge(with_token(garbage.clone(), "wtok"))
1483            .await
1484            .unwrap_err();
1485        assert_eq!(err.code(), tonic::Code::PermissionDenied);
1486        let err = adapter
1487            .purge(with_token(garbage, "atok"))
1488            .await
1489            .unwrap_err();
1490        assert_eq!(err.code(), tonic::Code::InvalidArgument);
1491    }
1492
1493    #[tokio::test]
1494    async fn grpc_raw_backend_error_is_opaque() {
1495        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
1496        let adapter = GrpcAdapter::new(Service::new(Arc::new(DownStore), fs));
1497        let err = adapter
1498            .get_raw(Request::new(get_req("any")))
1499            .await
1500            .unwrap_err();
1501        assert_eq!(err.code(), tonic::Code::Internal);
1502        assert_eq!(err.message(), "internal error");
1503
1504        // The same opaque mapping applies to every other replication RPC
1505        // that reaches the store: `put_raw` (via `put_error`'s fallback),
1506        // `list_raw` and `purge`. Open mode's implicit principal is an admin,
1507        // so each request clears authorization and reaches `DownStore`.
1508        let err = adapter
1509            .put_raw(Request::new(put_req_with("origin", b"{}", None)))
1510            .await
1511            .unwrap_err();
1512        assert_eq!(err.code(), tonic::Code::Internal);
1513        assert_eq!(err.message(), "internal error");
1514
1515        let err = adapter
1516            .list_raw(Request::new(memory_list()))
1517            .await
1518            .unwrap_err();
1519        assert_eq!(err.code(), tonic::Code::Internal);
1520        assert_eq!(err.message(), "internal error");
1521
1522        let err = adapter
1523            .purge(Request::new(purge_req("memory", &Revision::initial(b"x"))))
1524            .await
1525            .unwrap_err();
1526        assert_eq!(err.code(), tonic::Code::Internal);
1527        assert_eq!(err.message(), "internal error");
1528    }
1529
1530    // --- gRPC delete carries a claimed author (R1) ---
1531
1532    #[tokio::test]
1533    async fn grpc_delete_by_non_admin_ignores_a_claimed_author() {
1534        // The live record is seeded by admin, so the prior author is "admin" —
1535        // distinct from both "forged" (the claim) and "writer" (the deleter).
1536        let adapter = fs_adapter(tomb_auth());
1537        adapter
1538            .put(with_token(put_req("memory", "admin"), "atok"))
1539            .await
1540            .unwrap();
1541        let mut req = delete_req("memory");
1542        req.author_json = serde_json::to_vec(&Identity::new("forged")).unwrap();
1543        let del = adapter
1544            .delete(with_token(req, "wtok"))
1545            .await
1546            .unwrap()
1547            .into_inner();
1548        assert_eq!(del.outcome, "deleted");
1549        let tomb = raw_record(&adapter).await.expect("tombstone");
1550        assert_eq!(tomb.meta.author, Identity::new("writer"));
1551    }
1552
1553    #[tokio::test]
1554    async fn grpc_delete_by_admin_keeps_a_claimed_author() {
1555        let adapter = fs_adapter(tomb_auth());
1556        adapter
1557            .put(with_token(put_req("memory", "writer"), "wtok"))
1558            .await
1559            .unwrap();
1560        let mut req = delete_req("memory");
1561        req.author_json = serde_json::to_vec(&Identity::new("origin")).unwrap();
1562        let del = adapter
1563            .delete(with_token(req, "atok"))
1564            .await
1565            .unwrap()
1566            .into_inner();
1567        assert_eq!(del.outcome, "deleted");
1568        let tomb = raw_record(&adapter).await.expect("tombstone");
1569        assert_eq!(tomb.meta.author, Identity::new("origin"));
1570    }
1571
1572    #[tokio::test]
1573    async fn grpc_open_mode_delete_keeps_a_claimed_author() {
1574        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
1575        let adapter = GrpcAdapter::new(Service::new(fs.clone(), fs));
1576        adapter
1577            .put(Request::new(put_req("memory", "writer")))
1578            .await
1579            .unwrap();
1580        let mut req = delete_req("memory");
1581        req.author_json = serde_json::to_vec(&Identity::new("origin")).unwrap();
1582        let del = adapter
1583            .delete(Request::new(req))
1584            .await
1585            .unwrap()
1586            .into_inner();
1587        assert_eq!(del.outcome, "deleted");
1588        let raw = adapter
1589            .get_raw(Request::new(get_req("memory")))
1590            .await
1591            .unwrap()
1592            .into_inner();
1593        let tomb: Record = serde_json::from_slice(&raw.record_json).unwrap();
1594        assert_eq!(tomb.meta.author, Identity::new("origin"));
1595    }
1596
1597    #[tokio::test]
1598    async fn grpc_delete_rejects_a_malformed_author() {
1599        let adapter = fs_adapter(tomb_auth());
1600        adapter
1601            .put(with_token(put_req("memory", "admin"), "atok"))
1602            .await
1603            .unwrap();
1604        let mut req = delete_req("memory");
1605        req.author_json = b"not json".to_vec();
1606        let err = adapter.delete(with_token(req, "atok")).await.unwrap_err();
1607        assert_eq!(err.code(), tonic::Code::InvalidArgument);
1608
1609        // The malformed author aborted the delete before the store call: the
1610        // record is still live.
1611        let got = adapter
1612            .get(with_token(get_req("memory"), "atok"))
1613            .await
1614            .unwrap()
1615            .into_inner();
1616        assert!(got.found);
1617    }
1618}
gonzalo_server/grpc.rs

gonzalo_server/
grpc.rs
