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    pub fn serve(self, db: &Db) {
118        sofie_db::provide::<Self>(db, Arc::new(self));
119    }
120
121    async fn answer(
122        &self,
123        asked: Asked<'_>,
124        db: &Db,
125        keys: &SigningKeys,
126        body: &mut Incoming,
127    ) -> ControlFlow<Response> {
128        let request = asked.request;
129        let path = request.uri.path();
130        let call = if let Some(rest) = path.strip_prefix(CONTRACTS_PATH) {
131            rest.rsplit_once('/')
132                .and_then(|(contract, method)| Some((self.contracts.get(contract)?, method)))
133        } else if let Some(name) = path.strip_prefix(LOOKUPS_PATH) {
134            self.lookups.get(name).map(|dispatch| (dispatch, ""))
135        } else if path == FEED_PATH && self.feed && request.method == Method::GET {
136            let feed = verified(asked, db, keys).map(|_| feed(db, request.uri.query()));
137            return ControlFlow::Break(match feed {
138                Ok(feed) => verified_answer(json(feed.await)),
139                Err(refused) => *refused,
140            });
141        } else {
142            return ControlFlow::Continue(());
143        };
144        let Some((dispatch, method)) = call.filter(|_| request.method == Method::POST) else {
145            return ControlFlow::Break(answer::failure(Error::NotFound, Caller::Json));
146        };
147        let acting = match verified(asked, db, keys) {
148            Ok(acting) => acting,
149            Err(refused) => return ControlFlow::Break(*refused),
150        };
151        let body = match read_body(body, BODY_LIMIT, Caller::Json).await {
152            Ok(body) => body,
153            Err(refused) => return ControlFlow::Break(verified_answer(*refused)),
154        };
155        let called = async {
156            let arguments = serde_json::from_slice(&body).map_err(|error| {
157                Error::BadRequest(format!("a contract call's body is not JSON: {error}"))
158            })?;
159            sofie_db::remote::acting_as(acting, dispatch(db, method, arguments)).await
160        };
161        ControlFlow::Break(verified_answer(json(called.await)))
162    }
163}
164
165fn verified_answer(response: Response) -> Response {
166    #[cfg(feature = "shield")]
167    return crate::shield::unobserved(response);
168    #[cfg(not(feature = "shield"))]
169    response
170}
171
172fn verified(
173    asked: Asked<'_>,
174    db: &Db,
175    keys: &SigningKeys,
176) -> std::result::Result<Option<Acting>, Box<Response>> {
177    let request = asked.request;
178    let token = request
179        .headers
180        .get(PRINCIPAL_HEADER)
181        .and_then(|token| token.to_str().ok())
182        .unwrap_or_default();
183    let now = db.clock().now().timestamp();
184    keys.verify_principal_token(token, request.uri.path(), now)
185        .map_err(|refused| {
186            tracing::warn!(
187                http.route = request.uri.path(),
188                error.kind = "principal_token",
189                error = %refused,
190                "refused a call between apps"
191            );
192            Box::new(answer::refusal(
193                StatusCode::UNAUTHORIZED,
194                UNVERIFIED,
195                Caller::Json,
196            ))
197        })
198}
199
200async fn feed(db: &Db, query: Option<&str>) -> Result<Value> {
201    let number = |name: &str| {
202        query
203            .unwrap_or_default()
204            .split('&')
205            .filter_map(|pair| pair.split_once('='))
206            .find(|(key, _)| *key == name)
207            .map(|(_, value)| {
208                value
209                    .parse::<i64>()
210                    .map_err(|_| Error::BadRequest(format!("`{name}` is not a whole number")))
211            })
212            .transpose()
213    };
214    let after = number("after")?.unwrap_or(0);
215    let limit = number("limit")?.unwrap_or(sofie_db::FEED_PAGE);
216    answered(&sofie_db::read_feed(db, after, limit).await?)
217}
218
219fn json(result: Result<Value>) -> Response {
220    match result {
221        Ok(value) => answer::response(StatusCode::OK, Some(JSON), value.to_string()),
222        Err(error) => answer::failure(error, Caller::Json),
223    }
224}
225
226fn unprovided<T: ?Sized>() -> Error {
227    Error::internal_message(format!(
228        "nothing provides {}; run `sofie gen` so the wiring does",
229        std::any::type_name::<T>()
230    ))
231}
sofie_http/remote.rs

sofie_http/
remote.rs
