[proxmox-backup.git] / proxmox-restore-daemon / src / main.rs

///! Daemon binary to run inside a micro-VM for secure single file restore of disk images
use std::fs::File;
use std::io::prelude::*;
use std::os::unix::{
    io::{FromRawFd, RawFd},
    net,
};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::future::Future;
use std::pin::Pin;

use anyhow::{bail, format_err, Error};
use lazy_static::lazy_static;
use log::{error, info};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use http::request::Parts;
use http::Response;
use hyper::{Body, StatusCode};
use hyper::header;

use proxmox::api::RpcEnvironmentType;

use pbs_client::DEFAULT_VSOCK_PORT;
use proxmox_rest_server::{ApiConfig, RestServer, RestEnvironment};

mod proxmox_restore_daemon;
use proxmox_restore_daemon::*;

/// Maximum amount of pending requests. If saturated, virtio-vsock returns ETIMEDOUT immediately.
/// We should never have more than a few requests in queue, so use a low number.
pub const MAX_PENDING: usize = 32;

/// Will be present in base initramfs
pub const VM_DETECT_FILE: &str = "/restore-vm-marker";

lazy_static! {
    /// The current disks state. Use for accessing data on the attached snapshots.
    pub static ref DISK_STATE: Arc<Mutex<DiskState>> = {
        Arc::new(Mutex::new(DiskState::scan().unwrap()))
    };
}

/// This is expected to be run by 'proxmox-file-restore' within a mini-VM
fn main() -> Result<(), Error> {
    if !Path::new(VM_DETECT_FILE).exists() {
        bail!(
            "This binary is not supposed to be run manually, use 'proxmox-file-restore' instead."
        );
    }

    // don't have a real syslog (and no persistance), so use env_logger to print to a log file (via
    // stdout to a serial terminal attached by QEMU)
    env_logger::from_env(env_logger::Env::default().default_filter_or("info"))
        .write_style(env_logger::WriteStyle::Never)
        .format_timestamp_millis()
        .init();

    info!("setup basic system environment...");
    setup_system_env().map_err(|err| format_err!("system environment setup failed: {}", err))?;

    // scan all attached disks now, before starting the API
    // this will panic and stop the VM if anything goes wrong
    info!("scanning all disks...");
    {
        let _disk_state = DISK_STATE.lock().unwrap();
    }

    info!("disk scan complete, starting main runtime...");

    pbs_runtime::main(run())
}

/// ensure we have our /run dirs, system users and stuff like that setup
fn setup_system_env() -> Result<(), Error> {
    // the API may save some stuff there, e.g., the memcon tracking file
    // we do not care much, but it's way less headache to just create it
    std::fs::create_dir_all("/run/proxmox-backup")?;

    // we now ensure that all lock files are owned by the backup user, and as we reuse the
    // specialized REST module from pbs api/daemon we have some checks there for user/acl stuff
    // that gets locked, and thus needs the backup system user to work.
    std::fs::create_dir_all("/etc")?;
    let mut passwd = File::create("/etc/passwd")?;
    writeln!(passwd, "root:x:0:0:root:/root:/bin/sh")?;
    writeln!(passwd, "backup:x:34:34:backup:/var/backups:/usr/sbin/nologin")?;

    let mut group = File::create("/etc/group")?;
    writeln!(group, "root:x:0:")?;
    writeln!(group, "backup:x:34:")?;

    Ok(())
}

fn get_index<'a>(
    _env: RestEnvironment,
    _parts: Parts,
) -> Pin<Box<dyn Future<Output = http::Response<Body>> + Send + 'a>> {
    Box::pin(async move {

        let index = "<center><h1>Proxmox Backup Restore Daemon/h1></center>";

        Response::builder()
            .status(StatusCode::OK)
            .header(header::CONTENT_TYPE, "text/html")
            .body(index.into())
            .unwrap()
    })
}

async fn run() -> Result<(), Error> {
    watchdog_init();

    let auth_config = Arc::new(
        auth::ticket_auth().map_err(|err| format_err!("reading ticket file failed: {}", err))?,
    );
    let config = ApiConfig::new("", &ROUTER, RpcEnvironmentType::PUBLIC, auth_config, &get_index)?;
    let rest_server = RestServer::new(config);

    let vsock_fd = get_vsock_fd()?;
    let connections = accept_vsock_connections(vsock_fd);
    let receiver_stream = ReceiverStream::new(connections);
    let acceptor = hyper::server::accept::from_stream(receiver_stream);

    hyper::Server::builder(acceptor).serve(rest_server).await?;

    bail!("hyper server exited");
}

fn accept_vsock_connections(
    vsock_fd: RawFd,
) -> mpsc::Receiver<Result<tokio::net::UnixStream, Error>> {
    use nix::sys::socket::*;
    let (sender, receiver) = mpsc::channel(MAX_PENDING);

    tokio::spawn(async move {
        loop {
            let stream: Result<tokio::net::UnixStream, Error> = tokio::task::block_in_place(|| {
                // we need to accept manually, as UnixListener aborts if socket type != AF_UNIX ...
                let client_fd = accept(vsock_fd)?;
                let stream = unsafe { net::UnixStream::from_raw_fd(client_fd) };
                stream.set_nonblocking(true)?;
                tokio::net::UnixStream::from_std(stream).map_err(|err| err.into())
            });

            match stream {
                Ok(stream) => {
                    if sender.send(Ok(stream)).await.is_err() {
                        error!("connection accept channel was closed");
                    }
                }
                Err(err) => {
                    error!("error accepting vsock connetion: {}", err);
                }
            }
        }
    });

    receiver
}

fn get_vsock_fd() -> Result<RawFd, Error> {
    use nix::sys::socket::*;
    let sock_fd = socket(
        AddressFamily::Vsock,
        SockType::Stream,
        SockFlag::empty(),
        None,
    )?;
    let sock_addr = VsockAddr::new(libc::VMADDR_CID_ANY, DEFAULT_VSOCK_PORT as u32);
    bind(sock_fd, &SockAddr::Vsock(sock_addr))?;
    listen(sock_fd, MAX_PENDING)?;
    Ok(sock_fd)
}
Commit	Line	Data
dd9cef56	1	///! Daemon binary to run inside a micro-VM for secure single file restore of disk images
4c1b7761 WB	2	use std::fs::File;
4c1b7761 WB	3	use std::io::prelude::*;
dd9cef56 SR	4	use std::os::unix::{
	5	io::{FromRawFd, RawFd},
	6	net,
	7	};
	8	use std::path::Path;
d32a8652	9	use std::sync::{Arc, Mutex};
6680878b DM	10	use std::future::Future;
6680878b DM	11	use std::pin::Pin;
dd9cef56	12
4c1b7761 WB	13	use anyhow::{bail, format_err, Error};
	14	use lazy_static::lazy_static;
	15	use log::{error, info};
dd9cef56 SR	16	use tokio::sync::mpsc;
dd9cef56 SR	17	use tokio_stream::wrappers::ReceiverStream;
7fa9a37c DM	18	use http::request::Parts;
	19	use http::Response;
	20	use hyper::{Body, StatusCode};
	21	use hyper::header;
dd9cef56 SR	22
dd9cef56 SR	23	use proxmox::api::RpcEnvironmentType;
dd9cef56	24
4c1b7761	25	use pbs_client::DEFAULT_VSOCK_PORT;
48176b0a	26	use proxmox_rest_server::{ApiConfig, RestServer, RestEnvironment};
9edf96e6	27
dd9cef56 SR	28	mod proxmox_restore_daemon;
	29	use proxmox_restore_daemon::*;
	30
	31	/// Maximum amount of pending requests. If saturated, virtio-vsock returns ETIMEDOUT immediately.
	32	/// We should never have more than a few requests in queue, so use a low number.
	33	pub const MAX_PENDING: usize = 32;
	34
	35	/// Will be present in base initramfs
	36	pub const VM_DETECT_FILE: &str = "/restore-vm-marker";
	37
d32a8652 SR	38	lazy_static! {
	39	/// The current disks state. Use for accessing data on the attached snapshots.
	40	pub static ref DISK_STATE: Arc<Mutex<DiskState>> = {
	41	Arc::new(Mutex::new(DiskState::scan().unwrap()))
	42	};
	43	}
	44
dd9cef56 SR	45	/// This is expected to be run by 'proxmox-file-restore' within a mini-VM
	46	fn main() -> Result<(), Error> {
	47	if !Path::new(VM_DETECT_FILE).exists() {
309e14eb TL	48	bail!(
	49	"This binary is not supposed to be run manually, use 'proxmox-file-restore' instead."
	50	);
dd9cef56 SR	51	}
	52
	53	// don't have a real syslog (and no persistance), so use env_logger to print to a log file (via
	54	// stdout to a serial terminal attached by QEMU)
	55	env_logger::from_env(env_logger::Env::default().default_filter_or("info"))
	56	.write_style(env_logger::WriteStyle::Never)
ecd66eca	57	.format_timestamp_millis()
dd9cef56 SR	58	.init();
dd9cef56 SR	59
3f780ddf TL	60	info!("setup basic system environment...");
3f780ddf TL	61	setup_system_env().map_err(\|err\| format_err!("system environment setup failed: {}", err))?;
33d7292f	62
d32a8652 SR	63	// scan all attached disks now, before starting the API
d32a8652 SR	64	// this will panic and stop the VM if anything goes wrong
9a06eb16	65	info!("scanning all disks...");
d32a8652 SR	66	{
	67	let _disk_state = DISK_STATE.lock().unwrap();
	68	}
	69
9a06eb16 TL	70	info!("disk scan complete, starting main runtime...");
9a06eb16 TL	71
d420962f	72	pbs_runtime::main(run())
dd9cef56 SR	73	}
dd9cef56 SR	74
73e1ba65 TL	75	/// ensure we have our /run dirs, system users and stuff like that setup
	76	fn setup_system_env() -> Result<(), Error> {
	77	// the API may save some stuff there, e.g., the memcon tracking file
	78	// we do not care much, but it's way less headache to just create it
	79	std::fs::create_dir_all("/run/proxmox-backup")?;
	80
9edf96e6 TL	81	// we now ensure that all lock files are owned by the backup user, and as we reuse the
	82	// specialized REST module from pbs api/daemon we have some checks there for user/acl stuff
	83	// that gets locked, and thus needs the backup system user to work.
	84	std::fs::create_dir_all("/etc")?;
	85	let mut passwd = File::create("/etc/passwd")?;
	86	writeln!(passwd, "root:x:0:0:root:/root:/bin/sh")?;
	87	writeln!(passwd, "backup:x:34:34:backup:/var/backups:/usr/sbin/nologin")?;
	88
	89	let mut group = File::create("/etc/group")?;
	90	writeln!(group, "root:x:0:")?;
	91	writeln!(group, "backup:x:34:")?;
	92
73e1ba65 TL	93	Ok(())
	94	}
	95
6680878b	96	fn get_index<'a>(
48176b0a	97	_env: RestEnvironment,
7fa9a37c	98	_parts: Parts,
6680878b DM	99	) -> Pin<Box<dyn Future<Output = http::Response<Body>> + Send + 'a>> {
6680878b DM	100	Box::pin(async move {
7fa9a37c	101
6680878b	102	let index = "<center><h1>Proxmox Backup Restore Daemon/h1></center>";
7fa9a37c	103
6680878b DM	104	Response::builder()
	105	.status(StatusCode::OK)
	106	.header(header::CONTENT_TYPE, "text/html")
	107	.body(index.into())
	108	.unwrap()
	109	})
7fa9a37c DM	110	}
7fa9a37c DM	111
dd9cef56	112	async fn run() -> Result<(), Error> {
a26ebad5 SR	113	watchdog_init();
a26ebad5 SR	114
dd9cef56 SR	115	let auth_config = Arc::new(
	116	auth::ticket_auth().map_err(\|err\| format_err!("reading ticket file failed: {}", err))?,
	117	);
6680878b	118	let config = ApiConfig::new("", &ROUTER, RpcEnvironmentType::PUBLIC, auth_config, &get_index)?;
dd9cef56 SR	119	let rest_server = RestServer::new(config);
	120
	121	let vsock_fd = get_vsock_fd()?;
	122	let connections = accept_vsock_connections(vsock_fd);
	123	let receiver_stream = ReceiverStream::new(connections);
	124	let acceptor = hyper::server::accept::from_stream(receiver_stream);
	125
	126	hyper::Server::builder(acceptor).serve(rest_server).await?;
	127
	128	bail!("hyper server exited");
	129	}
	130
	131	fn accept_vsock_connections(
	132	vsock_fd: RawFd,
	133	) -> mpsc::Receiver<Result<tokio::net::UnixStream, Error>> {
	134	use nix::sys::socket::*;
	135	let (sender, receiver) = mpsc::channel(MAX_PENDING);
	136
	137	tokio::spawn(async move {
	138	loop {
	139	let stream: Result<tokio::net::UnixStream, Error> = tokio::task::block_in_place(\|\| {
	140	// we need to accept manually, as UnixListener aborts if socket type != AF_UNIX ...
	141	let client_fd = accept(vsock_fd)?;
	142	let stream = unsafe { net::UnixStream::from_raw_fd(client_fd) };
	143	stream.set_nonblocking(true)?;
	144	tokio::net::UnixStream::from_std(stream).map_err(\|err\| err.into())
	145	});
	146
	147	match stream {
	148	Ok(stream) => {
	149	if sender.send(Ok(stream)).await.is_err() {
	150	error!("connection accept channel was closed");
	151	}
	152	}
	153	Err(err) => {
	154	error!("error accepting vsock connetion: {}", err);
	155	}
	156	}
	157	}
	158	});
	159
	160	receiver
	161	}
	162
	163	fn get_vsock_fd() -> Result<RawFd, Error> {
	164	use nix::sys::socket::*;
	165	let sock_fd = socket(
	166	AddressFamily::Vsock,
	167	SockType::Stream,
	168	SockFlag::empty(),
	169	None,
	170	)?;
	171	let sock_addr = VsockAddr::new(libc::VMADDR_CID_ANY, DEFAULT_VSOCK_PORT as u32);
	172	bind(sock_fd, &SockAddr::Vsock(sock_addr))?;
	173	listen(sock_fd, MAX_PENDING)?;
	174	Ok(sock_fd)
	175	}