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