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_http::websocket::WebSocket;
  21 use proxmox::{identity, sortable};
  22
  23 use pbs_tools::ticket::{self, Empty, Ticket};
  24
  25 use crate::api2::types::*;
  26 use crate::config::acl::PRIV_SYS_CONSOLE;
  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 tasks;
  37 pub mod subscription;
  38
  39 pub(crate) mod rrd;
  40
  41 mod journal;
  42 pub(crate) mod services;
  43 mod status;
  44 mod syslog;
  45 mod time;
  46 mod report;
  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(
  97     cmd: Option<String>,
  98     rpcenv: &mut dyn RpcEnvironment,
  99 ) -> 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)?
 123         .sign(
 124             crate::auth_helpers::private_auth_key(),
 125             Some(&tools::ticket::term_aad(&userid, &path, port)),
 126         )?;
 127
 128     let mut command = Vec::new();
 129     match cmd.as_deref() {
 130         Some("login") | None => {
 131             command.push("login");
 132             if userid == "root@pam" {
 133                 command.push("-f");
 134                 command.push("root");
 135             }
 136         }
 137         Some("upgrade") => {
 138             if userid != "root@pam" {
 139                 bail!("only root@pam can upgrade");
 140             }
 141             // TODO: add nicer/safer wrapper like in PVE instead
 142             command.push("sh");
 143             command.push("-c");
 144             command.push("apt full-upgrade; bash -l");
 145         }
 146         _ => bail!("invalid command"),
 147     };
 148
 149     let username = userid.name().to_owned();
 150     let upid = WorkerTask::spawn(
 151         "termproxy",
 152         None,
 153         auth_id,
 154         false,
 155         move |worker| async move {
 156             // move inside the worker so that it survives and does not close the port
 157             // remove CLOEXEC from listenere so that we can reuse it in termproxy
 158             tools::fd_change_cloexec(listener.as_raw_fd(), false)?;
 159
 160             let mut arguments: Vec<&str> = Vec::new();
 161             let fd_string = listener.as_raw_fd().to_string();
 162             arguments.push(&fd_string);
 163             arguments.extend_from_slice(&[
 164                 "--path",
 165                 &path,
 166                 "--perm",
 167                 "Sys.Console",
 168                 "--authport",
 169                 "82",
 170                 "--port-as-fd",
 171                 "--",
 172             ]);
 173             arguments.extend_from_slice(&command);
 174
 175             let mut cmd = tokio::process::Command::new("/usr/bin/termproxy");
 176
 177             cmd.args(&arguments)
 178                 .stdout(std::process::Stdio::piped())
 179                 .stderr(std::process::Stdio::piped());
 180
 181             let mut child = cmd.spawn().expect("error executing termproxy");
 182
 183             let stdout = child.stdout.take().expect("no child stdout handle");
 184             let stderr = child.stderr.take().expect("no child stderr handle");
 185
 186             let worker_stdout = worker.clone();
 187             let stdout_fut = async move {
 188                 let mut reader = BufReader::new(stdout).lines();
 189                 while let Some(line) = reader.next_line().await? {
 190                     worker_stdout.log(line);
 191                 }
 192                 Ok::<(), Error>(())
 193             };
 194
 195             let worker_stderr = worker.clone();
 196             let stderr_fut = async move {
 197                 let mut reader = BufReader::new(stderr).lines();
 198                 while let Some(line) = reader.next_line().await? {
 199                     worker_stderr.warn(line);
 200                 }
 201                 Ok::<(), Error>(())
 202             };
 203
 204             let mut needs_kill = false;
 205             let res = tokio::select!{
 206                 res = child.wait() => {
 207                     let exit_code = res?;
 208                     if !exit_code.success() {
 209                         match exit_code.code() {
 210                             Some(code) => bail!("termproxy exited with {}", code),
 211                             None => bail!("termproxy exited by signal"),
 212                         }
 213                     }
 214                     Ok(())
 215                 },
 216                 res = stdout_fut => res,
 217                 res = stderr_fut => res,
 218                 res = worker.abort_future() => {
 219                     needs_kill = true;
 220                     res.map_err(Error::from)
 221                 }
 222             };
 223
 224             if needs_kill {
 225                 if res.is_ok() {
 226                     child.kill().await?;
 227                     return Ok(());
 228                 }
 229
 230                 if let Err(err) = child.kill().await {
 231                     worker.warn(format!("error killing termproxy: {}", err));
 232                 } else if let Err(err) = child.wait().await {
 233                     worker.warn(format!("error awaiting termproxy: {}", err));
 234                 }
 235             }
 236
 237             res
 238         },
 239     )?;
 240
 241     // FIXME: We're returning the user NAME only?
 242     Ok(json!({
 243         "user": username,
 244         "ticket": ticket,
 245         "port": port,
 246         "upid": upid,
 247     }))
 248 }
 249
 250 #[sortable]
 251 pub const API_METHOD_WEBSOCKET: ApiMethod = ApiMethod::new(
 252     &ApiHandler::AsyncHttp(&upgrade_to_websocket),
 253     &ObjectSchema::new(
 254         "Upgraded to websocket",
 255         &sorted!([
 256             ("node", false, &NODE_SCHEMA),
 257             (
 258                 "vncticket",
 259                 false,
 260                 &StringSchema::new("Terminal ticket").schema()
 261             ),
 262             ("port", false, &IntegerSchema::new("Terminal port").schema()),
 263         ]),
 264     ),
 265 )
 266 .access(
 267     Some("The user needs Sys.Console on /system."),
 268     &Permission::Privilege(&["system"], PRIV_SYS_CONSOLE, false),
 269 );
 270
 271 fn upgrade_to_websocket(
 272     parts: Parts,
 273     req_body: Body,
 274     param: Value,
 275     _info: &ApiMethod,
 276     rpcenv: Box<dyn RpcEnvironment>,
 277 ) -> ApiResponseFuture {
 278     async move {
 279         // intentionally user only for now
 280         let auth_id: Authid = rpcenv
 281             .get_auth_id()
 282             .ok_or_else(|| format_err!("no authid available"))?
 283             .parse()?;
 284
 285         if auth_id.is_token() {
 286             bail!("API tokens cannot access this API endpoint");
 287         }
 288
 289         let userid = auth_id.user();
 290         let ticket = tools::required_string_param(&param, "vncticket")?;
 291         let port: u16 = tools::required_integer_param(&param, "port")? as u16;
 292
 293         // will be checked again by termproxy
 294         Ticket::<Empty>::parse(ticket)?
 295             .verify(
 296                 crate::auth_helpers::public_auth_key(),
 297                 ticket::TERM_PREFIX,
 298                 Some(&tools::ticket::term_aad(&userid, "/system", port)),
 299             )?;
 300
 301         let (ws, response) = WebSocket::new(parts.headers.clone())?;
 302
 303         crate::server::spawn_internal_task(async move {
 304             let conn: Upgraded = match hyper::upgrade::on(Request::from_parts(parts, req_body)).map_err(Error::from).await {
 305                 Ok(upgraded) => upgraded,
 306                 _ => bail!("error"),
 307             };
 308
 309             let local = tokio::net::TcpStream::connect(format!("localhost:{}", port)).await?;
 310             ws.serve_connection(conn, local).await
 311         });
 312
 313         Ok(response)
 314     }
 315     .boxed()
 316 }
 317
 318 pub const SUBDIRS: SubdirMap = &[
 319     ("apt", &apt::ROUTER),
 320     ("certificates", &certificates::ROUTER),
 321     ("config", &config::ROUTER),
 322     ("disks", &disks::ROUTER),
 323     ("dns", &dns::ROUTER),
 324     ("journal", &journal::ROUTER),
 325     ("network", &network::ROUTER),
 326     ("report", &report::ROUTER),
 327     ("rrd", &rrd::ROUTER),
 328     ("services", &services::ROUTER),
 329     ("status", &status::ROUTER),
 330     ("subscription", &subscription::ROUTER),
 331     ("syslog", &syslog::ROUTER),
 332     ("tasks", &tasks::ROUTER),
 333     ("termproxy", &Router::new().post(&API_METHOD_TERMPROXY)),
 334     ("time", &time::ROUTER),
 335     (
 336         "vncwebsocket",
 337         &Router::new().upgrade(&API_METHOD_WEBSOCKET),
 338     ),
 339 ];
 340
 341 pub const ROUTER: Router = Router::new()
 342     .get(&list_subdirs_api_method!(SUBDIRS))
 343     .subdirs(SUBDIRS);