mz/
server.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::future::{Future, IntoFuture};
11use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
12
13use axum::{
14    Router,
15    extract::Query,
16    response::{Html, IntoResponse, Response},
17    routing::get,
18};
19use serde::Deserialize;
20use tokio::net::TcpListener;
21use tokio::sync::mpsc::UnboundedSender;
22
23use crate::error::Error;
24
25const LOGO_URL: &str = "https://materialize.com/svgs/brand-guide/materialize-purple-mark.svg";
26
27/// Produces an HTML string formatted
28/// with a message centered in the middle of the page
29/// and Materialize logo on top
30fn format_as_html_message(msg: &str) -> Html<String> {
31    Html(String::from(&format!(" \
32        <body style=\"margin: 0; display: flex; justify-content: center; align-items: center; min-height: 100vh; background-color: #f0f0f0;\">
33            <div style=\"text-align: center; padding: 100px; background-color: #ffffff; border-radius: 10px; box-shadow: 0 2px 6px rgba(0, 0, 0, 0.1);\"> \
34                <img src=\"{}\"> \
35                <h2 style=\"padding-top: 20px; font-family: Inter, Arial, sans-serif;\">{}</h2> \
36            </div>
37        </body>
38    ", LOGO_URL, msg)))
39}
40
41/// The query string of an OAuth authorization response (RFC 6749 §4.1.2).
42#[derive(Debug, Deserialize)]
43struct OAuthCallback {
44    code: Option<String>,
45    state: Option<String>,
46    error: Option<String>,
47    error_description: Option<String>,
48}
49
50#[allow(clippy::unused_async)]
51async fn oauth_callback(
52    Query(callback): Query<OAuthCallback>,
53    expected_state: String,
54    tx: UnboundedSender<Result<String, Error>>,
55) -> Response {
56    // Any local process or web page can reach this listener, so a callback is
57    // only trusted when it echoes the `state` that `mz` sent to the
58    // authorization endpoint. Requests that fail the check are rejected without
59    // ending the login, so a stray or forged request cannot cancel it.
60    if callback.state.as_deref() != Some(expected_state.as_str()) {
61        return format_as_html_message("Invalid login request.").into_response();
62    }
63
64    let (result, msg) = match (callback.code, callback.error) {
65        (Some(code), None) => (Ok(code), "You can now close the tab."),
66        (_, Some(error)) if error == "access_denied" => (
67            Err(Error::LoginOperationCanceled),
68            "Login canceled. You can now close the tab.",
69        ),
70        (_, error) => (
71            Err(Error::OAuthError(
72                callback
73                    .error_description
74                    .or(error)
75                    .unwrap_or_else(|| "missing authorization code".into()),
76            )),
77            "Login failed. Please, try again or communicate with support.",
78        ),
79    };
80    tx.send(result)
81        .unwrap_or_else(|_| panic!("Error handling login details."));
82    format_as_html_message(msg).into_response()
83}
84
85/// Binds a loopback listener for an OAuth authorization response that must
86/// carry `expected_state`, and sends the authorization code (or the error the
87/// authorization server returned) to `tx`.
88pub async fn oauth_callback_server(
89    expected_state: String,
90    tx: UnboundedSender<Result<String, Error>>,
91) -> (impl Future<Output = Result<(), std::io::Error>>, u16) {
92    let app = Router::new().route("/", get(|query| oauth_callback(query, expected_state, tx)));
93    serve(app).await
94}
95
96async fn serve(app: Router) -> (impl Future<Output = Result<(), std::io::Error>>, u16) {
97    let addr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0));
98    let listener = TcpListener::bind(addr).await.unwrap_or_else(|e| {
99        panic!("error binding to {}: {}", addr, e);
100    });
101    let port = listener.local_addr().unwrap().port();
102    let server = axum::serve(listener, app.into_make_service());
103
104    (server.into_future(), port)
105}
106
107#[cfg(test)]
108mod tests {
109    use tokio::sync::mpsc;
110
111    use super::*;
112
113    async fn get(port: u16, query: &str) {
114        reqwest::get(format!("http://127.0.0.1:{port}/?{query}"))
115            .await
116            .expect("callback server reachable");
117    }
118
119    #[mz_ore::test(tokio::test)]
120    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `socket`
121    async fn test_oauth_callback_ignores_wrong_state() {
122        let (tx, mut rx) = mpsc::unbounded_channel();
123        let (server, port) = oauth_callback_server("expected".into(), tx).await;
124        mz_ore::task::spawn(|| "oauth-callback-server", server);
125
126        get(port, "code=forged&state=other").await;
127        get(port, "code=forged").await;
128        assert!(
129            rx.try_recv().is_err(),
130            "mismatched state must not end the login"
131        );
132
133        get(port, "code=real&state=expected").await;
134        assert_eq!(rx.recv().await.unwrap().unwrap(), "real");
135    }
136
137    #[mz_ore::test(tokio::test)]
138    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `socket`
139    async fn test_oauth_callback_errors() {
140        let (tx, mut rx) = mpsc::unbounded_channel();
141        let (server, port) = oauth_callback_server("s".into(), tx).await;
142        mz_ore::task::spawn(|| "oauth-callback-server", server);
143
144        get(port, "error=access_denied&state=s").await;
145        assert!(matches!(
146            rx.recv().await.unwrap(),
147            Err(Error::LoginOperationCanceled)
148        ));
149
150        get(port, "error=server_error&error_description=boom&state=s").await;
151        assert!(matches!(
152            rx.recv().await.unwrap(),
153            Err(Error::OAuthError(msg)) if msg == "boom"
154        ));
155
156        get(port, "state=s").await;
157        assert!(matches!(
158            rx.recv().await.unwrap(),
159            Err(Error::OAuthError(_))
160        ));
161    }
162}
mz/server.rs

mz/
server.rs
