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