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