sofie_http/
remote.rs

1//! What a split app serves the apps it left: contracts, batch lookups and its event feed, each
2//! call proven by a principal token signed with the keys every app of the deployment shares.
3
4use std::collections::HashMap;
5use std::ops::ControlFlow;
6use std::sync::Arc;
7
8use http::{Method, StatusCode};
9use hyper::body::Incoming;
10use serde::Serialize;
11use serde::de::DeserializeOwned;
12use serde_json::Value;
13use sofie_auth::{Acting, SigningKeys};
14use sofie_core::{Error, Result, Uuid};
15use sofie_db::remote::{CONTRACTS_PATH, FEED_PATH, LOOKUPS_PATH, PRINCIPAL_HEADER, lookup_name};
16use sofie_db::{BoxFuture, Db, Lookup};
17
18use crate::answer::{self, JSON, Response};
19use crate::caller::Caller;
20use crate::pipeline::{Asked, read_body};
21use crate::routes::BODY_LIMIT;
22
23const UNVERIFIED: &str = "The call between apps could not be verified.";
24
25/// Written by `#[contract]` for every contract trait, so a split app can serve it.
26pub trait Serve {
27    /// `<contract crate>/<trait>`, the same in the app that calls and the one that serves.
28    const CONTRACT: &'static str;
29
30    fn serve<'a>(&'a self, method: &'a str, arguments: Value) -> BoxFuture<'a, Result<Value>>;
31}
32
33pub fn arguments<T: DeserializeOwned>(arguments: Value) -> Result<T> {
34    serde_json::from_value(arguments).map_err(|error| {
35        Error::BadRequest(format!(
36            "the arguments of a contract call do not fit: {error}"
37        ))
38    })
39}
40
41pub fn encoded<T: Serialize>(arguments: &T) -> Result<Value> {
42    serde_json::to_value(arguments)
43        .map_err(|error| Error::internal("serializing a contract call's arguments", error))
44}
45
46pub fn answered<T: Serialize>(answer: &T) -> Result<Value> {
47    serde_json::to_value(answer)
48        .map_err(|error| Error::internal("serializing a contract's answer", error))
49}
50
51#[must_use]
52pub fn unknown_method(contract: &str, method: &str) -> Error {
53    Error::BadRequest(format!("`{contract}` has no method `{method}`"))
54}
55
56type Dispatch = Arc<dyn Fn(&Db, &str, Value) -> BoxFuture<'static, Result<Value>> + Send + Sync>;
57
58#[derive(Default)]
59pub struct Served {
60    contracts: HashMap<&'static str, Dispatch>,
61    lookups: HashMap<&'static str, Dispatch>,
62    feed: bool,
63}
64
65/// Looked up only for a path under `/_sofie/`, so no other request pays for it.
66pub(crate) async fn answer(
67    asked: Asked<'_>,
68    db: &Db,
69    keys: &SigningKeys,
70    body: &mut Incoming,
71) -> ControlFlow<Response> {
72    if !asked.request.uri.path().starts_with("/_sofie/") {
73        return ControlFlow::Continue(());
74    }
75    match sofie_db::provided::<Served>(db) {
76        Some(served) => served.answer(asked, db, keys, body).await,
77        None => ControlFlow::Continue(()),
78    }
79}
80
81impl Served {
82    #[must_use]
83    pub fn contract<T: ?Sized + Serve + Send + Sync + 'static>(mut self) -> Self {
84        let dispatch: Dispatch = Arc::new(|db, method, arguments| {
85            let implementation = sofie_db::provided::<T>(db);
86            let method = method.to_owned();
87            Box::pin(async move {
88                let implementation = implementation.ok_or_else(|| unprovided::<T>())?;
89                implementation.serve(&method, arguments).await
90            })
91        });
92        self.contracts.insert(T::CONTRACT, dispatch);
93        self
94    }
95
96    #[must_use]
97    pub fn lookup<D: Serialize + Send + 'static>(mut self) -> Self {
98        let dispatch: Dispatch = Arc::new(|db, _, arguments| {
99            let lookup = sofie_db::provided::<dyn Lookup<D>>(db);
100            Box::pin(async move {
101                let lookup = lookup.ok_or_else(unprovided::<dyn Lookup<D>>)?;
102                let ids: Vec<Uuid> = self::arguments(arguments)?;
103                answered(&lookup.get_many(&ids).await?)
104            })
105        });
106        self.lookups.insert(lookup_name::<D>(), dispatch);
107        self
108    }
109
110    #[must_use]
111    pub fn feed(mut self) -> Self {
112        self.feed = true;
113        self
114    }
115
116    /// Answered under `/_sofie/`, never through an area; the reverse proxy keeps those paths inside.
117    /// A feed is refused here on a database that keeps none, not at the first app pulling it.
118    pub fn serve(self, db: &Db) -> Result<()> {
119        if self.feed {
120            sofie_db::keeps_feed(db)?;
121        }
122        sofie_db::provide::<Self>(db, Arc::new(self));
123        Ok(())
124    }
125
126    async fn answer(
127        &self,
128        asked: Asked<'_>,
129        db: &Db,
130        keys: &SigningKeys,
131        body: &mut Incoming,
132    ) -> ControlFlow<Response> {
133        let request = asked.request;
134        let path = request.uri.path();
135        let call = if let Some(rest) = path.strip_prefix(CONTRACTS_PATH) {
136            rest.rsplit_once('/')
137                .and_then(|(contract, method)| Some((self.contracts.get(contract)?, method)))
138        } else if let Some(name) = path.strip_prefix(LOOKUPS_PATH) {
139            self.lookups.get(name).map(|dispatch| (dispatch, ""))
140        } else if path == FEED_PATH && self.feed && request.method == Method::GET {
141            let feed = verified(asked, db, keys).map(|_| feed(db, request.uri.query()));
142            return ControlFlow::Break(match feed {
143                Ok(feed) => verified_answer(json(feed.await)),
144                Err(refused) => *refused,
145            });
146        } else {
147            return ControlFlow::Continue(());
148        };
149        let Some((dispatch, method)) = call.filter(|_| request.method == Method::POST) else {
150            return ControlFlow::Break(answer::failure(Error::NotFound, Caller::Json));
151        };
152        let acting = match verified(asked, db, keys) {
153            Ok(acting) => acting,
154            Err(refused) => return ControlFlow::Break(*refused),
155        };
156        let body = match read_body(body, BODY_LIMIT, Caller::Json).await {
157            Ok(body) => body,
158            Err(refused) => return ControlFlow::Break(verified_answer(*refused)),
159        };
160        let called = async {
161            let arguments = serde_json::from_slice(&body).map_err(|error| {
162                Error::BadRequest(format!("a contract call's body is not JSON: {error}"))
163            })?;
164            sofie_db::remote::acting_as(acting, dispatch(db, method, arguments)).await
165        };
166        ControlFlow::Break(verified_answer(json(called.await)))
167    }
168}
169
170fn verified_answer(response: Response) -> Response {
171    #[cfg(feature = "shield")]
172    return crate::shield::unobserved(response);
173    #[cfg(not(feature = "shield"))]
174    response
175}
176
177fn verified(
178    asked: Asked<'_>,
179    db: &Db,
180    keys: &SigningKeys,
181) -> std::result::Result<Option<Acting>, Box<Response>> {
182    let request = asked.request;
183    let token = request
184        .headers
185        .get(PRINCIPAL_HEADER)
186        .and_then(|token| token.to_str().ok())
187        .unwrap_or_default();
188    let now = db.clock().now().timestamp();
189    keys.verify_principal_token(token, request.uri.path(), now)
190        .map_err(|refused| {
191            tracing::warn!(
192                http.route = request.uri.path(),
193                error.kind = "principal_token",
194                error = %refused,
195                "refused a call between apps"
196            );
197            Box::new(answer::refusal(
198                StatusCode::UNAUTHORIZED,
199                UNVERIFIED,
200                Caller::Json,
201            ))
202        })
203}
204
205async fn feed(db: &Db, query: Option<&str>) -> Result<Value> {
206    let number = |name: &str| {
207        query
208            .unwrap_or_default()
209            .split('&')
210            .filter_map(|pair| pair.split_once('='))
211            .find(|(key, _)| *key == name)
212            .map(|(_, value)| {
213                value
214                    .parse::<i64>()
215                    .map_err(|_| Error::BadRequest(format!("`{name}` is not a whole number")))
216            })
217            .transpose()
218    };
219    let after = number("after")?.unwrap_or(0);
220    let limit = number("limit")?.unwrap_or(sofie_db::FEED_PAGE);
221    answered(&sofie_db::read_feed(db, after, limit).await?)
222}
223
224fn json(result: Result<Value>) -> Response {
225    match result {
226        Ok(value) => answer::response(StatusCode::OK, Some(JSON), value.to_string()),
227        Err(error) => answer::failure(error, Caller::Json),
228    }
229}
230
231fn unprovided<T: ?Sized>() -> Error {
232    Error::internal_message(format!(
233        "nothing provides {}; run `sofie gen` so the wiring does",
234        std::any::type_name::<T>()
235    ))
236}
sofie_http/remote.rs

sofie_http/
remote.rs
