src/api2/node/mod.rs

   1 //! Server/Node Configuration and Administration
   2
   3 use std::net::TcpListener;
   4 use std::os::unix::io::AsRawFd;
   5
   6 use anyhow::{bail, format_err, Error};
   7 use futures::future::{FutureExt, TryFutureExt};
   8 use hyper::body::Body;
   9 use hyper::http::request::Parts;
  10 use hyper::upgrade::Upgraded;
  11 use hyper::Request;
  12 use serde_json::{json, Value};
  13 use tokio::io::{AsyncBufReadExt, BufReader};
  14
  15 use proxmox::api::router::{Router, SubdirMap};
  16 use proxmox::api::{
  17     api, schema::*, ApiHandler, ApiMethod, ApiResponseFuture, Permission, RpcEnvironment,
  18 };
  19 use proxmox::list_subdirs_api_method;
  20 use proxmox::{identity, sortable};
  21 use proxmox_http::websocket::WebSocket;
  22
  23 use pbs_api_types::{Authid, NODE_SCHEMA, PRIV_SYS_CONSOLE};
  24 use pbs_tools::auth::private_auth_key;
  25 use pbs_tools::ticket::{self, Empty, Ticket};
  26
  27 use crate::server::WorkerTask;
  28 use crate::tools;
  29
  30 pub mod apt;
  31 pub mod certificates;
  32 pub mod config;
  33 pub mod disks;
  34 pub mod dns;
  35 pub mod network;
  36 pub mod subscription;
  37 pub mod tasks;
  38
  39 pub(crate) mod rrd;
  40
  41 mod journal;
  42 mod report;
  43 pub(crate) mod services;
  44 mod status;
  45 mod syslog;
  46 mod time;
  47
  48 pub const SHELL_CMD_SCHEMA: Schema = StringSchema::new("The command to run.")
  49     .format(&ApiStringFormat::Enum(&[
  50         EnumEntry::new("login", "Login"),
  51         EnumEntry::new("upgrade", "Upgrade"),
  52     ]))
  53     .schema();
  54
  55 #[api(
  56     protected: true,
  57     input: {
  58         properties: {
  59             node: {
  60                 schema: NODE_SCHEMA,
  61             },
  62             cmd: {
  63                 schema: SHELL_CMD_SCHEMA,
  64                 optional: true,
  65             },
  66         },
  67     },
  68     returns: {
  69         type: Object,
  70         description: "Object with the user, ticket, port and upid",
  71         properties: {
  72             user: {
  73                 description: "",
  74                 type: String,
  75             },
  76             ticket: {
  77                 description: "",
  78                 type: String,
  79             },
  80             port: {
  81                 description: "",
  82                 type: String,
  83             },
  84             upid: {
  85                 description: "",
  86                 type: String,
  87             },
  88         }
  89     },
  90     access: {
  91         description: "Restricted to users on realm 'pam'",
  92         permission: &Permission::Privilege(&["system"], PRIV_SYS_CONSOLE, false),
  93     }
  94 )]
  95 /// Call termproxy and return shell ticket
  96 async fn termproxy(cmd: Option<String>, rpcenv: &mut dyn RpcEnvironment) -> Result<Value, Error> {
  97     // intentionally user only for now
  98     let auth_id: Authid = rpcenv
  99         .get_auth_id()
 100         .ok_or_else(|| format_err!("no authid available"))?
 101         .parse()?;
 102
 103     if auth_id.is_token() {
 104         bail!("API tokens cannot access this API endpoint");
 105     }
 106
 107     let userid = auth_id.user();
 108
 109     if userid.realm() != "pam" {
 110         bail!("only pam users can use the console");
 111     }
 112
 113     let path = "/system";
 114
 115     // use port 0 and let the kernel decide which port is free
 116     let listener = TcpListener::bind("localhost:0")?;
 117     let port = listener.local_addr()?.port();
 118
 119     let ticket = Ticket::new(ticket::TERM_PREFIX, &Empty)?.sign(
 120         private_auth_key(),
 121         Some(&tools::ticket::term_aad(&userid, &path, port)),
 122     )?;
 123
 124     let mut command = Vec::new();
 125     match cmd.as_deref() {
 126         Some("login") | None => {
 127             command.push("login");
 128             if userid == "root@pam" {
 129                 command.push("-f");
 130                 command.push("root");
 131             }
 132         }
 133         Some("upgrade") => {
 134             if userid != "root@pam" {
 135                 bail!("only root@pam can upgrade");
 136             }
 137             // TODO: add nicer/safer wrapper like in PVE instead
 138             command.push("sh");
 139             command.push("-c");
 140             command.push("apt full-upgrade; bash -l");
 141         }
 142         _ => bail!("invalid command"),
 143     };
 144
 145     let username = userid.name().to_owned();
 146     let upid = WorkerTask::spawn(
 147         "termproxy",
 148         None,
 149         auth_id,
 150         false,
 151         move |worker| async move {
 152             // move inside the worker so that it survives and does not close the port
 153             // remove CLOEXEC from listenere so that we can reuse it in termproxy
 154             tools::fd_change_cloexec(listener.as_raw_fd(), false)?;
 155
 156             let mut arguments: Vec<&str> = Vec::new();
 157             let fd_string = listener.as_raw_fd().to_string();
 158             arguments.push(&fd_string);
 159             arguments.extend_from_slice(&[
 160                 "--path",
 161                 &path,
 162                 "--perm",
 163                 "Sys.Console",
 164                 "--authport",
 165                 "82",
 166                 "--port-as-fd",
 167                 "--",
 168             ]);
 169             arguments.extend_from_slice(&command);
 170
 171             let mut cmd = tokio::process::Command::new("/usr/bin/termproxy");
 172
 173             cmd.args(&arguments)
 174                 .stdout(std::process::Stdio::piped())
 175                 .stderr(std::process::Stdio::piped());
 176
 177             let mut child = cmd.spawn().expect("error executing termproxy");
 178
 179             let stdout = child.stdout.take().expect("no child stdout handle");
 180             let stderr = child.stderr.take().expect("no child stderr handle");
 181
 182             let worker_stdout = worker.clone();
 183             let stdout_fut = async move {
 184                 let mut reader = BufReader::new(stdout).lines();
 185                 while let Some(line) = reader.next_line().await? {
 186                     worker_stdout.log(line);
 187                 }
 188                 Ok::<(), Error>(())
 189             };
 190
 191             let worker_stderr = worker.clone();
 192             let stderr_fut = async move {
 193                 let mut reader = BufReader::new(stderr).lines();
 194                 while let Some(line) = reader.next_line().await? {
 195                     worker_stderr.warn(line);
 196                 }
 197                 Ok::<(), Error>(())
 198             };
 199
 200             let mut needs_kill = false;
 201             let res = tokio::select! {
 202                 res = child.wait() => {
 203                     let exit_code = res?;
 204                     if !exit_code.success() {
 205                         match exit_code.code() {
 206                             Some(code) => bail!("termproxy exited with {}", code),
 207                             None => bail!("termproxy exited by signal"),
 208                         }
 209                     }
 210                     Ok(())
 211                 },
 212                 res = stdout_fut => res,
 213                 res = stderr_fut => res,
 214                 res = worker.abort_future() => {
 215                     needs_kill = true;
 216                     res.map_err(Error::from)
 217                 }
 218             };
 219
 220             if needs_kill {
 221                 if res.is_ok() {
 222                     child.kill().await?;
 223                     return Ok(());
 224                 }
 225
 226                 if let Err(err) = child.kill().await {
 227                     worker.warn(format!("error killing termproxy: {}", err));
 228                 } else if let Err(err) = child.wait().await {
 229                     worker.warn(format!("error awaiting termproxy: {}", err));
 230                 }
 231             }
 232
 233             res
 234         },
 235     )?;
 236
 237     // FIXME: We're returning the user NAME only?
 238     Ok(json!({
 239         "user": username,
 240         "ticket": ticket,
 241         "port": port,
 242         "upid": upid,
 243     }))
 244 }
 245
 246 #[sortable]
 247 pub const API_METHOD_WEBSOCKET: ApiMethod = ApiMethod::new(
 248     &ApiHandler::AsyncHttp(&upgrade_to_websocket),
 249     &ObjectSchema::new(
 250         "Upgraded to websocket",
 251         &sorted!([
 252             ("node", false, &NODE_SCHEMA),
 253             (
 254                 "vncticket",
 255                 false,
 256                 &StringSchema::new("Terminal ticket").schema()
 257             ),
 258             ("port", false, &IntegerSchema::new("Terminal port").schema()),
 259         ]),
 260     ),
 261 )
 262 .access(
 263     Some("The user needs Sys.Console on /system."),
 264     &Permission::Privilege(&["system"], PRIV_SYS_CONSOLE, false),
 265 );
 266
 267 fn upgrade_to_websocket(
 268     parts: Parts,
 269     req_body: Body,
 270     param: Value,
 271     _info: &ApiMethod,
 272     rpcenv: Box<dyn RpcEnvironment>,
 273 ) -> ApiResponseFuture {
 274     async move {
 275         // intentionally user only for now
 276         let auth_id: Authid = rpcenv
 277             .get_auth_id()
 278             .ok_or_else(|| format_err!("no authid available"))?
 279             .parse()?;
 280
 281         if auth_id.is_token() {
 282             bail!("API tokens cannot access this API endpoint");
 283         }
 284
 285         let userid = auth_id.user();
 286         let ticket = pbs_tools::json::required_string_param(&param, "vncticket")?;
 287         let port: u16 = pbs_tools::json::required_integer_param(&param, "port")? as u16;
 288
 289         // will be checked again by termproxy
 290         Ticket::<Empty>::parse(ticket)?.verify(
 291             crate::auth_helpers::public_auth_key(),
 292             ticket::TERM_PREFIX,
 293             Some(&tools::ticket::term_aad(&userid, "/system", port)),
 294         )?;
 295
 296         let (ws, response) = WebSocket::new(parts.headers.clone())?;
 297
 298         crate::server::spawn_internal_task(async move {
 299             let conn: Upgraded = match hyper::upgrade::on(Request::from_parts(parts, req_body))
 300                 .map_err(Error::from)
 301                 .await
 302             {
 303                 Ok(upgraded) => upgraded,
 304                 _ => bail!("error"),
 305             };
 306
 307             let local = tokio::net::TcpStream::connect(format!("localhost:{}", port)).await?;
 308             ws.serve_connection(conn, local).await
 309         });
 310
 311         Ok(response)
 312     }
 313     .boxed()
 314 }
 315
 316 #[api]
 317 /// List Nodes (only for compatiblity)
 318 fn list_nodes() -> Result<Value, Error> {
 319     Ok(json!([ { "node": proxmox::tools::nodename().to_string() } ]))
 320 }
 321
 322 pub const SUBDIRS: SubdirMap = &[
 323     ("apt", &apt::ROUTER),
 324     ("certificates", &certificates::ROUTER),
 325     ("config", &config::ROUTER),
 326     ("disks", &disks::ROUTER),
 327     ("dns", &dns::ROUTER),
 328     ("journal", &journal::ROUTER),
 329     ("network", &network::ROUTER),
 330     ("report", &report::ROUTER),
 331     ("rrd", &rrd::ROUTER),
 332     ("services", &services::ROUTER),
 333     ("status", &status::ROUTER),
 334     ("subscription", &subscription::ROUTER),
 335     ("syslog", &syslog::ROUTER),
 336     ("tasks", &tasks::ROUTER),
 337     ("termproxy", &Router::new().post(&API_METHOD_TERMPROXY)),
 338     ("time", &time::ROUTER),
 339     (
 340         "vncwebsocket",
 341         &Router::new().upgrade(&API_METHOD_WEBSOCKET),
 342     ),
 343 ];
 344
 345 pub const ITEM_ROUTER: Router = Router::new()
 346     .get(&list_subdirs_api_method!(SUBDIRS))
 347     .subdirs(SUBDIRS);
 348
 349 pub const ROUTER: Router = Router::new()
 350     .get(&API_METHOD_LIST_NODES)
 351     .match_all("node", &ITEM_ROUTER);