ota_lib/
setup.rs

1// Copyright 2019 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::{Context as _, Error};
6use fuchsia_async as fasync;
7use http_body_util::{BodyExt, Full};
8use hyper::{Request, Response};
9use serde::{Deserialize, Serialize};
10use std::convert::Infallible;
11use std::future::Future;
12use std::net::{IpAddr, SocketAddr};
13
14const SERVER_PORT: u16 = 8880;
15
16pub type Body = Full<hyper::body::Bytes>;
17
18pub enum SetupEvent {
19    Root,
20    DevhostOta { cfg: DevhostConfig },
21}
22
23/// Devhost configuration, passed to the actual OTA process.
24pub struct DevhostConfig {
25    pub url: String,
26}
27
28#[derive(Deserialize, Serialize)]
29/// Configuration provided by the host for the devhost OTA. Only used for de/serialization.
30struct DevhostRequestInfo {
31    /// We assume that the OTA server is running on the requester's address
32    /// at the given port.
33    pub port: u16,
34}
35
36async fn parse_ota_json(
37    request: Request<hyper::body::Incoming>,
38    remote_addr: IpAddr,
39) -> Result<DevhostConfig, Error> {
40    let body = request.into_body().collect().await.context("read request")?.to_bytes();
41    let DevhostRequestInfo { port } =
42        serde_json::from_slice(&body).context("Failed to parse JSON")?;
43
44    let url = format!("http://{}/config.json", SocketAddr::new(remote_addr, port));
45    Ok(DevhostConfig { url })
46}
47
48async fn serve<Fut, F>(
49    request: Request<hyper::body::Incoming>,
50    remote_addr: SocketAddr,
51    handler: F,
52) -> Response<Body>
53where
54    Fut: Future<Output = ()>,
55    F: FnOnce(SetupEvent) -> Fut,
56{
57    use hyper::{Method, StatusCode};
58
59    match (request.method(), request.uri().path()) {
60        (&Method::GET, "/") => {
61            let () = handler(SetupEvent::Root).await;
62            Response::new(Full::new("Root document".into()))
63        }
64        (&Method::POST, "/ota/devhost") => {
65            // get devhost info out of POST request.
66            match parse_ota_json(request, remote_addr.ip()).await {
67                Err(e) => {
68                    let mut response =
69                        Response::new(Full::new(format!("Bad request: {:?}", e).into()));
70                    *response.status_mut() = StatusCode::BAD_REQUEST;
71                    response
72                }
73                Ok(cfg) => {
74                    let () = handler(SetupEvent::DevhostOta { cfg }).await;
75                    Response::new(Full::new("Started OTA".into()))
76                }
77            }
78        }
79        _ => {
80            let mut response = Response::new(Full::new("Unknown command".into()));
81            *response.status_mut() = StatusCode::NOT_FOUND;
82            response
83        }
84    }
85}
86
87pub fn start_server<Fut, F>(handler: F) -> impl Future<Output = Result<(), hyper::Error>>
88where
89    Fut: Future<Output = ()>,
90    F: FnOnce(SetupEvent) -> Fut,
91    Fut: Send + 'static,
92    F: Clone + Send + 'static,
93{
94    use futures::{FutureExt as _, StreamExt as _};
95    use hyper::service::service_fn;
96    use hyper_util::rt::TokioIo;
97
98    println!("recovery: start_server");
99
100    let addr = SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), SERVER_PORT);
101    let listener = fasync::net::TcpListener::bind(&addr).expect("bind");
102    let listener = listener.accept_stream();
103
104    let builder = hyper_util::server::conn::auto::Builder::new(fuchsia_hyper::Executor);
105
106    async move {
107        let mut listener = listener.fuse();
108        let mut tasks = futures::stream::FuturesUnordered::new();
109        loop {
110            futures::select! {
111                res = listener.next() => {
112                    match res {
113                        Some(Ok((stream, remote_addr))) => {
114                            let handler = handler.clone();
115                            let service = service_fn(move |request| {
116                                let handler = handler.clone();
117                                serve(request, remote_addr, handler).map(Ok::<_, Infallible>)
118                            });
119                            let builder = builder.clone();
120                            let stream = fuchsia_hyper::TcpStream { stream };
121                            tasks.push(fasync::Task::spawn(async move {
122                                if let Err(e) = builder.serve_connection(TokioIo::new(stream), service).await {
123                                    println!("recovery server connection error: {e}");
124                                }
125                            }));
126                        }
127                        Some(Err(e)) => {
128                            println!("recovery accept error: {e}");
129                        }
130                        None => break,
131                    }
132                }
133                _ = tasks.next() => {}
134            }
135        }
136        while let Some(_) = tasks.next().await {}
137        Ok(())
138    }
139}
ota_lib/setup.rs

ota_lib/
setup.rs
