Add a basic web server (#1041)

Trello:
https://trello.com/c/msRHsnpV/3570-3-build-a-basic-agama-web-server

This PR introduces the web server for [Agama's 2024
architecture](https://github.com/openSUSE/agama/blob/master/doc/new_architecture.md).
At this point, it implements:

* A `/ping` endpoint.
* A simple WebSocket to receive progress events. It is just a
proof-of-concept and even the format should be adapted.
* Tracing/logging.

Additionally, it is able to export the description of the API using
[OpenAPI](https://www.openapis.org/).

## Usage

The new binary implements two subcommands: `serve` and `openapi`.

### Running the server

```
$ agama-web-server serve --help
Start the API server

Usage: agama-web-server serve [OPTIONS]

Options:
      --address <ADDRESS>  Address to listen on (default: "0.0.0.0:3000") [default: 0.0.0.0:3000]
  -h, --help               Print help
```

### Generating the OpenAPI documentation

```
$ agama-web-sever openapi
{
  "openapi": "3.0.3",
  "info": {
    "title": "agama-dbus-server",
    "description": "Agama web API description",
    "license": {
      "name": ""
    },
    "version": "0.1.0"
  },
...
```


Additionally, it adds a new `agama-web-server` package to the RPM spec
file.

## To do

- [x] Improve logging/tracing.
- [x] Expose the API documentation (openAPI?).

## Out of scope

* Better error handling.
* Read the D-Bus address from `/run/agama/bus`. We should do pretty much
the same for `agama-cli`.

## Testing

- Added a new Rus integration test
- Tested manually

rust/agama-cli/Cargo.toml

@@ -24,6 +24,7 @@ fs_extra = "1.3.0"
 nix = { version = "0.27.1", features = ["user"] }
 zbus = { version = "3", default-features = false, features = ["tokio"] }
 tokio = { version = "1.33.0", features = ["macros", "rt-multi-thread"] }
+async-trait = "0.1.77"
 
 [[bin]]
 name = "agama"

rust/agama-cli/src/progress.rs

@@ -1,4 +1,5 @@
 use agama_lib::progress::{Progress, ProgressPresenter};
+use async_trait::async_trait;
 use console::style;
 use indicatif::{ProgressBar, ProgressStyle};
 use std::time::Duration;
@@ -26,14 +27,15 @@ impl InstallerProgress {
     }
 }
 
+#[async_trait]
 impl ProgressPresenter for InstallerProgress {
-    fn start(&mut self, progress: &Progress) {
+    async fn start(&mut self, progress: &Progress) {
         if !progress.finished {
-            self.update_main(progress);
+            self.update_main(progress).await;
         }
     }
 
-    fn update_main(&mut self, progress: &Progress) {
+    async fn update_main(&mut self, progress: &Progress) {
         let counter = format!("[{}/{}]", &progress.current_step, &progress.max_steps);
 
         println!(
@@ -43,7 +45,7 @@ impl ProgressPresenter for InstallerProgress {
         );
     }
 
-    fn update_detail(&mut self, progress: &Progress) {
+    async fn update_detail(&mut self, progress: &Progress) {
         if progress.finished {
             if let Some(bar) = self.bar.take() {
                 bar.finish_and_clear();
@@ -53,7 +55,7 @@ impl ProgressPresenter for InstallerProgress {
         }
     }
 
-    fn finish(&mut self) {
+    async fn finish(&mut self) {
         if let Some(bar) = self.bar.take() {
             bar.finish_and_clear();
         }

rust/agama-dbus-server/Cargo.toml

@@ -27,3 +27,23 @@ regex = "1.10.2"
 once_cell = "1.18.0"
 macaddr = "1.0"
 async-trait = "0.1.75"
+axum = { version = "0.7.4", features = ["ws"] }
+serde_json = "1.0.113"
+tower-http = { version = "0.5.1", features = ["trace"] }
+tracing-subscriber = "0.3.18"
+tracing-journald = "0.3.0"
+tracing = "0.1.40"
+clap = { version = "4.5.0", features = ["derive", "wrap_help"] }
+tower = "0.4.13"
+utoipa = { version = "4.2.0", features = ["axum_extras"] }
+
+[[bin]]
+name = "agama-dbus-server"
+path = "src/agama-dbus-server.rs"
+
+[[bin]]
+name = "agama-web-server"
+path = "src/agama-web-server.rs"
+
+[dev-dependencies]
+http-body-util = "0.1.0"

rust/agama-dbus-server/src/agama-web-server.rs

@@ -0,0 +1,57 @@
+use agama_dbus_server::web;
+use agama_lib::connection;
+use clap::{Parser, Subcommand};
+use tracing_subscriber::prelude::*;
+use utoipa::OpenApi;
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    /// Start the API server.
+    Serve {
+        /// Address to listen on (default: "0.0.0.0:3000")
+        #[arg(long, default_value = "0.0.0.0:3000")]
+        address: String,
+    },
+    /// Display the API documentation in OpenAPI format.
+    Openapi,
+}
+
+#[derive(Parser, Debug)]
+#[command(
+    version,
+    about = "Starts the Agama web-based API.",
+    long_about = None)]
+struct Cli {
+    #[command(subcommand)]
+    pub command: Commands,
+}
+
+/// Start serving the API.
+async fn serve_command(address: &str) {
+    let journald = tracing_journald::layer().expect("could not connect to journald");
+    tracing_subscriber::registry().with(journald).init();
+
+    let listener = tokio::net::TcpListener::bind(address)
+        .await
+        .unwrap_or_else(|_| panic!("could not listen on {}", address));
+
+    let dbus_connection = connection().await.unwrap();
+    axum::serve(listener, web::service(dbus_connection))
+        .await
+        .expect("could not mount app on listener");
+}
+
+/// Display the API documentation in OpenAPI format.
+fn openapi_command() {
+    println!("{}", web::ApiDoc::openapi().to_pretty_json().unwrap());
+}
+
+#[tokio::main]
+async fn main() {
+    let cli = Cli::parse();
+
+    match cli.command {
+        Commands::Serve { address } => serve_command(&address).await,
+        Commands::Openapi => openapi_command(),
+    }
+}

rust/agama-dbus-server/src/lib.rs

@@ -2,3 +2,5 @@ pub mod error;
 pub mod l10n;
 pub mod network;
 pub mod questions;
+pub mod web;
+pub use web::service;

rust/agama-dbus-server/src/web.rs

@@ -0,0 +1,13 @@
+//! This module implements a web-based API for Agama. It is responsible for:
+//!
+//! * Exposing an HTTP API to interact with Agama.
+//! * Emit relevant events via websocket.
+//! * Serve the code for the web user interface (not implemented yet).
+
+mod docs;
+mod http;
+mod service;
+mod ws;
+
+pub use docs::ApiDoc;
+pub use service::service;

rust/agama-dbus-server/src/web/docs.rs

@@ -0,0 +1,9 @@
+use utoipa::OpenApi;
+
+#[derive(OpenApi)]
+#[openapi(
+    info(description = "Agama web API description"),
+    paths(super::http::ping),
+    components(schemas(super::http::PingResponse))
+)]
+pub struct ApiDoc;

rust/agama-dbus-server/src/web/http.rs

@@ -0,0 +1,20 @@
+//! Implements the handlers for the HTTP-based API.
+
+use axum::Json;
+use serde::Serialize;
+use utoipa::ToSchema;
+
+#[derive(Serialize, ToSchema)]
+pub struct PingResponse {
+    /// API status
+    status: String,
+}
+
+#[utoipa::path(get, path = "/ping", responses(
+    (status = 200, description = "The API is working", body = PingResponse)
+))]
+pub async fn ping() -> Json<PingResponse> {
+    Json(PingResponse {
+        status: "success".to_string(),
+    })
+}

rust/agama-dbus-server/src/web/service.rs

@@ -0,0 +1,17 @@
+use axum::{routing::get, Router};
+use tower_http::trace::TraceLayer;
+
+/// Returns a service that implements the web-based Agama API.
+pub fn service(dbus_connection: zbus::Connection) -> Router {
+    let state = ServiceState { dbus_connection };
+    Router::new()
+        .route("/ping", get(super::http::ping))
+        .route("/ws", get(super::ws::ws_handler))
+        .layer(TraceLayer::new_for_http())
+        .with_state(state)
+}
+
+#[derive(Clone)]
+pub struct ServiceState {
+    pub dbus_connection: zbus::Connection,
+}

rust/agama-dbus-server/src/web/ws.rs

@@ -0,0 +1,56 @@
+//! Implements the websocket handling.
+
+use super::service::ServiceState;
+use agama_lib::progress::{Progress, ProgressMonitor, ProgressPresenter};
+use async_trait::async_trait;
+use axum::{
+    extract::{
+        ws::{Message, WebSocket},
+        State, WebSocketUpgrade,
+    },
+    response::IntoResponse,
+};
+
+pub async fn ws_handler(
+    State(state): State<ServiceState>,
+    ws: WebSocketUpgrade,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| handle_socket(socket, state.dbus_connection))
+}
+
+async fn handle_socket(socket: WebSocket, connection: zbus::Connection) {
+    let presenter = WebSocketProgressPresenter::new(socket);
+    let mut monitor = ProgressMonitor::new(connection).await.unwrap();
+    _ = monitor.run(presenter).await;
+}
+
+/// Experimental ProgressPresenter to emit progress events over a WebSocket.
+struct WebSocketProgressPresenter(WebSocket);
+
+impl WebSocketProgressPresenter {
+    pub fn new(socket: WebSocket) -> Self {
+        Self(socket)
+    }
+
+    pub async fn report_progress(&mut self, progress: &Progress) {
+        let payload = serde_json::to_string(&progress).unwrap();
+        _ = self.0.send(Message::Text(payload)).await;
+    }
+}
+
+#[async_trait]
+impl ProgressPresenter for WebSocketProgressPresenter {
+    async fn start(&mut self, progress: &Progress) {
+        self.report_progress(progress).await;
+    }
+
+    async fn update_main(&mut self, progress: &Progress) {
+        self.report_progress(progress).await;
+    }
+
+    async fn update_detail(&mut self, progress: &Progress) {
+        self.report_progress(progress).await;
+    }
+
+    async fn finish(&mut self) {}
+}

rust/agama-dbus-server/tests/service.rs

@@ -0,0 +1,31 @@
+mod common;
+
+use self::common::DBusServer;
+use agama_dbus_server::service;
+use axum::{
+    body::Body,
+    http::{Request, StatusCode},
+};
+use http_body_util::BodyExt;
+use std::error::Error;
+use tokio::test;
+use tower::ServiceExt;
+
+async fn body_to_string(body: Body) -> String {
+    let bytes = body.collect().await.unwrap().to_bytes();
+    String::from_utf8(bytes.to_vec()).unwrap()
+}
+
+#[test]
+async fn test_ping() -> Result<(), Box<dyn Error>> {
+    let dbus_server = DBusServer::new().start().await?;
+    let web_server = service(dbus_server.connection());
+    let request = Request::builder().uri("/ping").body(Body::empty()).unwrap();
+
+    let response = web_server.oneshot(request).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+
+    let body = body_to_string(response.into_body()).await;
+    assert_eq!(&body, "{\"status\":\"success\"}");
+    Ok(())
+}

rust/agama-lib/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 [dependencies]
 agama-settings = { path="../agama-settings" }
 anyhow = "1.0"
+async-trait = "0.1.77"
 cidr = { version = "0.2.2", features = ["serde"] }
 curl = { version = "0.4.44", features = ["protocol-ftp"] }
 futures-util = "0.3.29"