Start9/start-os
2
0
Fork 0
mirror of https://github.com/Start9Labs/start-os.git synced 2026-03-26 02:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

v2 migration

Browse Source
This commit is contained in:
Aiden McClelland
2021-10-07 16:33:51 -06:00
committed by Aiden McClelland
parent 7fc4cb175c
commit e58df7ec4a
22 changed files with 466 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
appmgr/build-dev.sh
View File

@@ -10,7 +10,7 @@ fi
alias 'rust-arm64-builder'='docker run --rm -it -v "$HOME/.cargo/registry":/root/.cargo/registry -v "$(pwd)":/home/rust/src start9/rust-arm-cross:aarch64'
cd ../..
rust-arm64-builder sh -c "(cd embassy-os/appmgr && cargo build)"
cd embassy-os/appmgr
cd ..
rust-arm64-builder sh -c "(cd appmgr && cargo build)"
cd appmgr
#rust-arm64-builder aarch64-linux-gnu-strip target/aarch64-unknown-linux-gnu/release/embassyd

6
appmgr/build-portable-dev.sh
View File

@@ -10,6 +10,6 @@ fi
alias 'rust-musl-builder'='docker run --rm -it -v "$HOME"/.cargo/registry:/root/.cargo/registry -v "$(pwd)":/home/rust/src start9/rust-musl-cross:x86_64-musl'
cd ../..
rust-musl-builder sh -c "(cd embassy-os/appmgr && cargo +beta build --target=x86_64-unknown-linux-musl --no-default-features)"
cd embassy-os/appmgr
cd ..
rust-musl-builder sh -c "(cd appmgr && cargo +beta build --target=x86_64-unknown-linux-musl --no-default-features)"
cd appmgr

6
appmgr/build-portable.sh
View File

@@ -10,6 +10,6 @@ fi
alias 'rust-musl-builder'='docker run --rm -it -v "$HOME"/.cargo/registry:/root/.cargo/registry -v "$(pwd)":/home/rust/src start9/rust-musl-cross:x86_64-musl'
cd ../..
rust-musl-builder sh -c "(cd embassy-os/appmgr && cargo +beta build --release --target=x86_64-unknown-linux-musl --no-default-features)"
cd embassy-os/appmgr
cd ..
rust-musl-builder sh -c "(cd appmgr && cargo +beta build --release --target=x86_64-unknown-linux-musl --no-default-features)"
cd appmgr

6
appmgr/build-prod.sh
View File

@@ -10,7 +10,7 @@ fi
alias 'rust-arm64-builder'='docker run --rm -it -v "$HOME/.cargo/registry":/root/.cargo/registry -v "$(pwd)":/home/rust/src start9/rust-arm-cross:aarch64'
cd ../..
rust-arm64-builder sh -c "(cd embassy-os/appmgr && cargo build --release)"
cd embassy-os/appmgr
cd ..
rust-arm64-builder sh -c "(cd appmgr && cargo build --release)"
cd appmgr
#rust-arm64-builder aarch64-linux-gnu-strip target/aarch64-unknown-linux-gnu/release/embassyd

8
appmgr/src/action/docker.rs
View File

@@ -69,7 +69,13 @@ impl DockerAction {
cmd.stdout(std::process::Stdio::piped());
cmd.stderr(std::process::Stdio::piped());
if log::log_enabled!(log::Level::Trace) {
log::trace!("{}", format!("{:?}", cmd).split(r#"" ""#).collect::<Vec<&str>>().join(" "));
log::trace!(
"{}",
format!("{:?}", cmd)
.split(r#"" ""#)
.collect::<Vec<&str>>()
.join(" ")
);
}
let mut handle = cmd.spawn().with_kind(crate::ErrorKind::Docker)?;
if let (Some(input), Some(stdin)) = (&input_buf, &mut handle.stdin) {

9
appmgr/src/bin/embassy-init.rs
View File

@@ -1,12 +1,10 @@
use std::path::Path;
use std::sync::Arc;
use embassy::context::rpc::RpcContextConfig;
use embassy::context::{DiagnosticContext, SetupContext};
use embassy::db::model::ServerStatus;
use embassy::db::DatabaseModel;
use embassy::disk::main::DEFAULT_PASSWORD;
use embassy::hostname::get_product_key;
use embassy::middleware::cors::cors;
use embassy::middleware::diagnostic::diagnostic;
use embassy::middleware::encrypt::encrypt;
@@ -46,7 +44,12 @@ async fn init(cfg_path: Option<&str>) -> Result<(), Error> {
.invoke(embassy::ErrorKind::Nginx)
.await?;
let ctx = SetupContext::init(cfg_path).await?;
let encrypt = encrypt(Arc::new(get_product_key().await?));
let keysource_ctx = ctx.clone();
let keysource = move || {
let ctx = keysource_ctx.clone();
async move { ctx.product_key().await }
};
let encrypt = encrypt(keysource);
MARIO_COIN.play().await?;
rpc_server!({
command: embassy::setup_api,

2
appmgr/src/bin/embassyd.rs
View File

@@ -18,7 +18,7 @@ use futures::{FutureExt, TryFutureExt};
use log::LevelFilter;
use reqwest::{Client, Proxy};
use rpc_toolkit::hyper::{Body, Response, Server, StatusCode};
use rpc_toolkit::{rpc_server, Context};
use rpc_toolkit::rpc_server;
use tokio::process::Command;
use tokio::signal::unix::signal;

10
appmgr/src/config/mod.rs
View File

@@ -6,7 +6,7 @@ use bollard::container::KillContainerOptions;
use futures::future::{BoxFuture, FutureExt};
use indexmap::IndexSet;
use itertools::Itertools;
use patch_db::{DbHandle, ModelData, OptionModel};
use patch_db::DbHandle;
use rand::SeedableRng;
use regex::Regex;
use rpc_toolkit::command;
@@ -14,16 +14,14 @@ use serde_json::Value;
use crate::action::docker::DockerAction;
use crate::context::RpcContext;
use crate::db::model::{
CurrentDependencyInfo, InstalledPackageDataEntry, InstalledPackageDataEntryModel,
};
use crate::db::model::CurrentDependencyInfo;
use crate::db::util::WithRevision;
use crate::dependencies::{
break_transitive, update_current_dependents, BreakageRes, DependencyError, DependencyErrors,
TaggedDependencyError,
};
use crate::install::cleanup::{remove_current_dependents, update_dependents};
use crate::s9pk::manifest::{Manifest, ManifestModel, PackageId};
use crate::install::cleanup::remove_current_dependents;
use crate::s9pk::manifest::{Manifest, PackageId};
use crate::util::{
display_none, display_serializable, parse_duration, parse_stdin_deserializable, IoFormat,
};

4
appmgr/src/context/cli.rs
View File

@@ -1,5 +1,5 @@
use std::fs::File;
use std::io::{BufReader, Read};
use std::io::BufReader;
use std::net::{Ipv4Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -14,7 +14,7 @@ use rpc_toolkit::url::Host;
use rpc_toolkit::Context;
use serde::Deserialize;
use crate::{Error, ResultExt};
use crate::ResultExt;
#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "kebab-case")]

3
appmgr/src/context/rpc.rs
View File

@@ -15,9 +15,8 @@ use reqwest::Url;
use rpc_toolkit::url::Host;
use rpc_toolkit::Context;
use serde::Deserialize;
use sqlx::migrate::MigrateDatabase;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{ConnectOptions, Sqlite, SqlitePool};
use sqlx::SqlitePool;
use tokio::fs::File;
use tokio::sync::broadcast::Sender;
use tokio::sync::RwLock;

25
appmgr/src/context/setup.rs
View File

@@ -7,18 +7,20 @@ use std::time::Duration;
use patch_db::json_ptr::JsonPointer;
use patch_db::PatchDb;
use rpc_toolkit::yajrc::RpcError;
use rpc_toolkit::Context;
use serde::Deserialize;
use sqlx::migrate::MigrateDatabase;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{Sqlite, SqlitePool};
use sqlx::SqlitePool;
use tokio::fs::File;
use tokio::sync::broadcast::Sender;
use tokio::sync::RwLock;
use url::Host;
use crate::db::model::Database;
use crate::hostname::{get_hostname, get_id};
use crate::hostname::{get_hostname, get_id, get_product_key};
use crate::net::tor::os_key;
use crate::setup::RecoveryStatus;
use crate::util::io::from_toml_async_reader;
use crate::util::AsyncFileExt;
use crate::{Error, ResultExt};
@@ -64,6 +66,9 @@ pub struct SetupContextSeed {
pub shutdown: Sender<()>,
pub datadir: PathBuf,
pub zfs_pool_name: Arc<String>,
pub selected_v2_drive: RwLock<Option<PathBuf>>,
pub cached_product_key: RwLock<Option<Arc<String>>>,
pub recovery_status: RwLock<Option<Result<RecoveryStatus, RpcError>>>,
}
#[derive(Clone)]
@@ -79,6 +84,9 @@ impl SetupContext {
shutdown,
datadir,
zfs_pool_name,
selected_v2_drive: RwLock::new(None),
cached_product_key: RwLock::new(None),
recovery_status: RwLock::new(None),
})))
}
pub async fn db(&self, secret_store: &SqlitePool) -> Result<PatchDb, Error> {
@@ -114,6 +122,17 @@ impl SetupContext {
.with_kind(crate::ErrorKind::Database)?;
Ok(secret_store)
}
pub async fn product_key(&self) -> Result<Arc<String>, Error> {
Ok(
if let Some(k) = { self.cached_product_key.read().await.clone() } {
k
} else {
let k = Arc::new(get_product_key().await?);
*self.cached_product_key.write().await = Some(k.clone());
k
},
)
}
}
impl Context for SetupContext {

11
appmgr/src/db/model.rs
View File

@@ -23,6 +23,8 @@ pub struct Database {
pub server_info: ServerInfo,
#[model]
pub package_data: AllPackageData,
#[model]
pub recovered_packages: BTreeMap<PackageId, RecoveredPackageInfo>,
pub broken_packages: Vec<PackageId>,
pub ui: Value,
}
@@ -54,6 +56,7 @@ impl Database {
update_progress: None,
},
package_data: AllPackageData::default(),
recovered_packages: BTreeMap::new(),
broken_packages: Vec::new(),
ui: Value::Object(Default::default()),
}
@@ -253,3 +256,11 @@ pub struct InterfaceAddresses {
#[model]
pub lan_address: Option<String>,
}
#[derive(Debug, Deserialize, Serialize, HasModel)]
#[serde(rename_all = "kebab-case")]
pub struct RecoveredPackageInfo {
pub title: String,
pub icon: String,
pub version: Version,
}

4
appmgr/src/disk/util.rs
View File

@@ -93,6 +93,7 @@ pub async fn get_capacity<P: AsRef<Path>>(path: P) -> Result<usize, Error> {
.invoke(crate::ErrorKind::BlockDevice)
.await?,
)?
.trim()
.parse()?)
}
@@ -203,7 +204,8 @@ pub async fn list() -> Result<Vec<DiskInfo>, Error> {
.unwrap_or_default();
let mut used = None;
let tmp_mountpoint = Path::new(TMP_MOUNTPOINT).join(&part);
let tmp_mountpoint =
Path::new(TMP_MOUNTPOINT).join(&part.strip_prefix("/").unwrap_or(&part));
if let Err(e) = mount(&part, &tmp_mountpoint).await {
log::warn!("Could not collect usage information: {}", e.source)
} else {

7
appmgr/src/install/cleanup.rs
View File

@@ -6,7 +6,6 @@ use patch_db::{DbHandle, PatchDbHandle};
use super::PKG_DOCKER_DIR;
use crate::context::RpcContext;
use crate::db::model::{CurrentDependencyInfo, InstalledPackageDataEntry, PackageDataEntry};
use crate::dependencies::update_current_dependents;
use crate::s9pk::manifest::PackageId;
use crate::util::Version;
use crate::Error;
@@ -218,6 +217,12 @@ pub async fn uninstall(
entry.current_dependents.keys(),
)
.await?;
tokio::fs::remove_dir_all(
ctx.datadir
.join(crate::volume::PKG_VOLUME_DIR)
.join(&entry.manifest.id),
)
.await?;
tx.commit(None).await?;
Ok(())
}

52
appmgr/src/install/mod.rs
View File

@@ -507,7 +507,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
log::info!("Install {}@{}: Unpacking Assets", pkg_id, version);
progress
.track_read_during(progress_model.clone(), &ctx.db, || async {
let asset_dir = asset_dir(ctx, pkg_id, version);
let asset_dir = asset_dir(&ctx.datadir, pkg_id, version);
if tokio::fs::metadata(&asset_dir).await.is_err() {
tokio::fs::create_dir_all(&asset_dir).await?;
}
@@ -628,6 +628,13 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
*dep_errs = DependencyErrors::init(ctx, &mut tx, &manifest, &current_dependencies).await?;
dep_errs.save(&mut tx).await?;
let recovered = crate::db::DatabaseModel::new()
.recovered_packages()
.idx_model(pkg_id)
.get(&mut tx, true)
.await?
.into_owned();
if let PackageDataEntry::Updating {
installed: prev,
manifest: prev_manifest,
@@ -686,10 +693,51 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
&mut BTreeMap::new(),
)
.await?;
todo!("set as running if viable");
let mut main_status = crate::db::DatabaseModel::new()
.package_data()
.idx_model(pkg_id)
.expect(&mut tx)
.await?
.installed()
.expect(&mut tx)
.await?
.status()
.main()
.get_mut(&mut tx)
.await?;
*main_status = prev.status.main;
main_status.save(&mut tx).await?;
}
} else {
update_dependents(ctx, &mut tx, pkg_id, current_dependents.keys()).await?;
if let Some(recovered) = recovered {
let configured = if let Some(res) = manifest
.migrations
.from(ctx, &recovered.version, pkg_id, version, &manifest.volumes)
.await?
{
res.configured
} else {
false
};
if configured {
crate::config::configure(
ctx,
&mut tx,
pkg_id,
None,
&None,
false,
&mut BTreeMap::new(),
&mut BTreeMap::new(),
)
.await?;
}
crate::db::DatabaseModel::new()
.recovered_packages()
.remove(&mut tx, pkg_id)
.await?
}
}
sql_tx.commit().await?;

2
appmgr/src/install/progress.rs
View File

@@ -120,10 +120,10 @@ impl<RW> InstallProgressTracker<RW> {
}
}
pub fn validated(&mut self) {
self.validating = false;
self.progress
.validation_complete
.store(true, Ordering::SeqCst);
self.validating = false;
}
}
impl<W: AsyncWrite> AsyncWrite for InstallProgressTracker<W> {

32
appmgr/src/middleware/encrypt.rs
View File

@@ -1,3 +1,4 @@
use std::future::Future;
use std::sync::Arc;
use aes::cipher::{CipherKey, NewCipher, Nonce, StreamCipher};
@@ -147,17 +148,38 @@ fn encrypted(headers: &HeaderMap) -> bool {
.unwrap_or_default()
}
pub fn encrypt<M: Metadata>(key: Arc<String>) -> DynMiddleware<M> {
pub fn encrypt<
F: Fn() -> Fut + Send + Sync + Clone + 'static,
Fut: Future<Output = Result<Arc<String>, Error>> + Send + Sync + 'static,
M: Metadata,
>(
keysource: F,
) -> DynMiddleware<M> {
Box::new(
move |req: &mut Request<Body>,
metadata: M|
-> BoxFuture<Result<Result<DynMiddlewareStage2, Response<Body>>, HttpError>> {
let key = key.clone();
let keysource = keysource.clone();
async move {
let encrypted = encrypted(req.headers());
if encrypted {
let key = if encrypted {
let key = match keysource().await {
Ok(s) => s,
Err(e) => {
let (res_parts, _) = Response::new(()).into_parts();
return Ok(Err(to_response(
req.headers(),
res_parts,
Err(e.into()),
|_| StatusCode::OK,
)?));
}
};
let body = std::mem::take(req.body_mut());
*req.body_mut() = Body::wrap_stream(DecryptStream::new(key.clone(), body));
Some(key)
} else {
None
};
let res: DynMiddlewareStage2 = Box::new(move |req, rpc_req| {
async move {
@@ -182,7 +204,7 @@ pub fn encrypt<M: Metadata>(key: Arc<String>) -> DynMiddleware<M> {
async move {
let res: DynMiddlewareStage4 = Box::new(move |res| {
async move {
if encrypted {
if let Some(key) = key {
res.headers_mut().insert(
"Content-Encoding",
HeaderValue::from_static("aesctr256"),
@@ -200,7 +222,7 @@ pub fn encrypt<M: Metadata>(key: Arc<String>) -> DynMiddleware<M> {
}
let body = std::mem::take(res.body_mut());
*res.body_mut() = Body::wrap_stream(
EncryptStream::new(&*key, body),
EncryptStream::new(key.as_ref(), body),
);
}
Ok(())

4
appmgr/src/notifications.rs
View File

@@ -4,10 +4,10 @@ use std::str::FromStr;
use anyhow::anyhow;
use chrono::{DateTime, Utc};
use futures::lock::Mutex;
use patch_db::{PatchDb, Revision};
use patch_db::PatchDb;
use rpc_toolkit::command;
use sqlx::SqlitePool;
use tokio::sync::Mutex;
use crate::context::RpcContext;
use crate::db::util::WithRevision;

313
appmgr/src/setup.rs
View File

@@ -1,7 +1,13 @@
use std::path::PathBuf;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use anyhow::anyhow;
use futures::future::BoxFuture;
use futures::{FutureExt, TryStreamExt};
use rpc_toolkit::command;
use rpc_toolkit::yajrc::RpcError;
use serde::{Deserialize, Serialize};
use tokio::fs::File;
use tokio::io::AsyncWriteExt;
@@ -9,9 +15,16 @@ use tokio::process::Command;
use torut::onion::TorSecretKeyV3;
use crate::context::SetupContext;
use crate::disk::disk;
use crate::db::model::RecoveredPackageInfo;
use crate::disk::main::DEFAULT_PASSWORD;
use crate::util::Invoke;
use crate::disk::util::{mount, unmount, DiskInfo};
use crate::id::Id;
use crate::install::PKG_PUBLIC_DIR;
use crate::s9pk::manifest::PackageId;
use crate::sound::BEETHOVEN;
use crate::util::io::from_yaml_async_reader;
use crate::util::{GeneralGuard, Invoke, Version};
use crate::volume::{data_dir, Volume, VolumeId};
use crate::{Error, ResultExt};
#[command(subcommands(status, disk, execute))]
@@ -22,22 +35,47 @@ pub fn setup() -> Result<(), Error> {
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct StatusRes {
is_recovering: bool,
tor_address: Option<String>,
product_key: bool,
migrating: bool,
}
#[command(rpc_only)]
pub fn status() -> Result<StatusRes, Error> {
#[command(rpc_only, metadata(authenticated = false))]
pub async fn status(#[context] ctx: SetupContext) -> Result<StatusRes, Error> {
Ok(StatusRes {
is_recovering: false,
tor_address: None,
product_key: tokio::fs::metadata("/embassy-os/product_key.txt")
.await
.is_ok(),
migrating: ctx.recovery_status.read().await.is_some(), // TODO
})
}
#[derive(Debug, Deserialize, Serialize)]
#[command(subcommands(list_disks))]
pub fn disk() -> Result<(), Error> {
Ok(())
}
#[command(rename = "list", rpc_only, metadata(authenticated = false))]
pub async fn list_disks() -> Result<Vec<DiskInfo>, Error> {
crate::disk::list(None).await
}
#[command(subcommands(recovery_status))]
pub fn recovery() -> Result<(), Error> {
Ok(())
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case")]
pub struct SetupResult {
tor_address: String,
pub struct RecoveryStatus {
bytes_transferred: u64,
total_bytes: u64,
}
#[command(rename = "status", rpc_only, metadata(authenticated = false))]
pub async fn recovery_status(
#[context] ctx: SetupContext,
) -> Result<Option<RecoveryStatus>, RpcError> {
ctx.recovery_status.read().await.clone().transpose()
}
#[command(rpc_only)]
@@ -45,10 +83,20 @@ pub async fn execute(
#[context] ctx: SetupContext,
#[arg(rename = "embassy-logicalname")] embassy_logicalname: PathBuf,
#[arg(rename = "embassy-password")] embassy_password: String,
) -> Result<SetupResult, Error> {
match execute_inner(ctx, embassy_logicalname, embassy_password).await {
#[arg(rename = "recovery-diskinfo")] recovery_diskinfo: Option<DiskInfo>,
#[arg(rename = "recovery-password")] recovery_password: Option<String>,
) -> Result<String, Error> {
match execute_inner(
ctx,
embassy_logicalname,
embassy_password,
recovery_diskinfo,
recovery_password,
)
.await
{
Ok(a) => {
log::info!("Setup Successful! Tor Address: {}", a.tor_address);
log::info!("Setup Successful! Tor Address: {}", a);
Ok(a)
}
Err(e) => {
@@ -58,11 +106,27 @@ pub async fn execute(
}
}
pub async fn complete_setup(ctx: SetupContext, guid: String) -> Result<(), Error> {
let mut guid_file = File::create("/embassy-os/disk.guid").await?;
guid_file.write_all(guid.as_bytes()).await?;
guid_file.sync_all().await?;
ctx.shutdown.send(()).expect("failed to shutdown");
Ok(())
}
pub async fn execute_inner(
ctx: SetupContext,
embassy_logicalname: PathBuf,
embassy_password: String,
) -> Result<SetupResult, Error> {
recovery_diskinfo: Option<DiskInfo>,
recovery_password: Option<String>,
) -> Result<String, Error> {
if ctx.recovery_status.read().await.is_some() {
return Err(Error::new(
anyhow!("Cannot execute setup while in recovery!"),
crate::ErrorKind::InvalidRequest,
));
}
let guid =
crate::disk::main::create(&ctx.zfs_pool_name, [embassy_logicalname], DEFAULT_PASSWORD)
.await?;
@@ -100,14 +164,217 @@ pub async fn execute_inner(
)
.execute(&mut sqlite_pool.acquire().await?)
.await?;
let mut guid_file = File::create("/embassy-os/disk.guid").await?;
guid_file.write_all(guid.as_bytes()).await?;
guid_file.sync_all().await?;
sqlite_pool.close().await;
ctx.shutdown.send(()).expect("failed to shutdown");
if let Some(recovery_diskinfo) = recovery_diskinfo {
if recovery_diskinfo
.embassy_os
.as_ref()
.map(|v| &*v.version < &emver::Version::new(0, 2, 8, 0))
.unwrap_or(true)
{
return Err(Error::new(anyhow!("Unsupported version of EmbassyOS. Please update to at least 0.2.8 before recovering."), crate::ErrorKind::VersionIncompatible));
}
tokio::spawn(async move {
if let Err(e) = recover(ctx.clone(), guid, recovery_diskinfo, recovery_password).await {
BEETHOVEN.play().await.unwrap_or_default(); // ignore error in playing the song
log::error!("Error recovering drive!: {}", e);
*ctx.recovery_status.write().await = Some(Err(e.into()));
}
});
} else {
complete_setup(ctx, guid).await?;
}
Ok(SetupResult {
tor_address: tor_key.public().get_onion_address().to_string(),
})
Ok(tor_key.public().get_onion_address().to_string())
}
async fn recover(
ctx: SetupContext,
guid: String,
recovery_diskinfo: DiskInfo,
recovery_password: Option<String>,
) -> Result<(), Error> {
let recovery_version = recovery_diskinfo
.embassy_os
.as_ref()
.map(|i| i.version.clone())
.unwrap_or_default();
if recovery_version.major() == 0 && recovery_version.minor() == 2 {
recover_v2(&ctx, recovery_diskinfo).await?;
} else if recovery_version.major() == 0 && recovery_version.minor() == 3 {
recover_v3(&ctx, recovery_diskinfo, recovery_password).await?;
} else {
return Err(Error::new(
anyhow!("Unsupported version of EmbassyOS: {}", recovery_version),
crate::ErrorKind::VersionIncompatible,
));
}
complete_setup(ctx, guid).await
}
fn dir_size<'a, P: AsRef<Path> + 'a + Send + Sync>(
path: P,
res: &'a AtomicU64,
) -> BoxFuture<'a, Result<(), std::io::Error>> {
async move {
tokio_stream::wrappers::ReadDirStream::new(tokio::fs::read_dir(path.as_ref()).await?)
.try_for_each_concurrent(None, |e| async move {
let m = e.metadata().await?;
if m.is_file() {
res.fetch_add(m.len(), Ordering::Relaxed);
} else if m.is_dir() {
dir_size(e.path(), res).await?;
}
Ok(())
})
.await
}
.boxed()
}
fn dir_copy<'a, P0: AsRef<Path> + 'a + Send + Sync, P1: AsRef<Path> + 'a + Send + Sync>(
src: P0,
dst: P1,
ctr: &'a AtomicU64,
) -> BoxFuture<'a, Result<(), std::io::Error>> {
async move {
let dst_path = dst.as_ref();
tokio_stream::wrappers::ReadDirStream::new(tokio::fs::read_dir(src.as_ref()).await?)
.try_for_each_concurrent(None, |e| async move {
let m = e.metadata().await?;
let src_path = e.path();
let dst_path = dst_path.join(e.file_name());
if m.is_file() {
tokio::fs::copy(src_path, dst_path).await?;
ctr.fetch_add(m.len(), Ordering::Relaxed);
} else if m.is_dir() {
dir_copy(src_path, dst_path, ctr).await?;
} else {
tokio::fs::symlink(tokio::fs::read_link(src_path).await?, &dst_path).await?;
tokio::fs::set_permissions(dst_path, m.permissions()).await?;
}
Ok(())
})
.await
}
.boxed()
}
async fn recover_v2(ctx: &SetupContext, recovery_diskinfo: DiskInfo) -> Result<(), Error> {
let tmp_mountpoint = Path::new("/mnt/recovery");
mount(
&recovery_diskinfo
.partitions
.get(1)
.ok_or_else(|| {
Error::new(
anyhow!("missing rootfs partition"),
crate::ErrorKind::Filesystem,
)
})?
.logicalname,
tmp_mountpoint,
)
.await?;
let mount_guard = GeneralGuard::new(|| tokio::spawn(unmount(tmp_mountpoint)));
let secret_store = ctx.secret_store().await?;
let db = ctx.db(&secret_store).await?;
let mut handle = db.handle();
let apps_yaml_path = tmp_mountpoint.join("root").join("appmgr").join("apps.yaml");
#[derive(Deserialize)]
struct LegacyAppInfo {
title: String,
version: Version,
}
let packages: BTreeMap<PackageId, LegacyAppInfo> =
from_yaml_async_reader(File::open(&apps_yaml_path).await.with_ctx(|_| {
(
crate::ErrorKind::Filesystem,
apps_yaml_path.display().to_string(),
)
})?)
.await?;
let volume_path = tmp_mountpoint.join("root/appmgr/volumes");
let total_bytes = AtomicU64::new(0);
for (pkg_id, _) in &packages {
dir_size(volume_path.join(pkg_id), &total_bytes).await?;
}
let total_bytes = total_bytes.load(Ordering::SeqCst);
*ctx.recovery_status.write().await = Some(Ok(RecoveryStatus {
bytes_transferred: 0,
total_bytes,
}));
let bytes_transferred = AtomicU64::new(0);
let volume_id = VolumeId::Custom(Id::try_from("main".to_owned())?);
for (pkg_id, info) in packages {
let volume_src_path = volume_path.join(&pkg_id);
let volume_dst_path = data_dir(&ctx.datadir, &pkg_id, &info.version, &volume_id);
tokio::select!(
res = dir_copy(
&volume_src_path,
&volume_dst_path,
&bytes_transferred
) => res.with_ctx(|_| {
(
crate::ErrorKind::Filesystem,
format!("{} -> {}", volume_src_path.display(), volume_dst_path.display()),
)
})?,
_ = async {
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
*ctx.recovery_status.write().await = Some(Ok(RecoveryStatus {
bytes_transferred: bytes_transferred.load(Ordering::Relaxed),
total_bytes,
}));
}
} => (),
);
let icon_leaf = AsRef::<Path>::as_ref(&pkg_id)
.join(info.version.as_str())
.join("icon.png");
let icon_src_path = tmp_mountpoint
.join("root/agent/icons")
.join(format!("{}.png", pkg_id));
let icon_dst_path = ctx.datadir.join(PKG_PUBLIC_DIR).join(&icon_leaf);
tokio::fs::copy(&icon_src_path, &icon_dst_path)
.await
.with_ctx(|_| {
(
crate::ErrorKind::Filesystem,
format!("{} -> {}", icon_src_path.display(), icon_dst_path.display()),
)
})?;
let icon_url = Path::new("/public/package-data").join(&icon_leaf);
crate::db::DatabaseModel::new()
.recovered_packages()
.idx_model(&pkg_id)
.put(
&mut handle,
&RecoveredPackageInfo {
title: info.title,
icon: icon_url.display().to_string(),
version: info.version,
},
)
.await?;
}
mount_guard
.drop()
.await
.with_kind(crate::ErrorKind::Unknown)?
}
async fn recover_v3(
ctx: &SetupContext,
recovery_diskinfo: DiskInfo,
recovery_password: Option<String>,
) -> Result<(), Error> {
todo!()
}

7
appmgr/src/status/mod.rs
View File

@@ -3,17 +3,14 @@ use std::sync::Arc;
use anyhow::anyhow;
use chrono::{DateTime, Utc};
use futures::future::BoxFuture;
use futures::{FutureExt, StreamExt};
use patch_db::{DbHandle, HasModel, Map, MapModel, ModelData};
use patch_db::{DbHandle, HasModel, Map, ModelData};
use serde::{Deserialize, Serialize};
use self::health_check::HealthCheckId;
use crate::context::RpcContext;
use crate::db::model::{CurrentDependencyInfo, InstalledPackageDataEntryModel};
use crate::dependencies::{
break_transitive, DependencyError, DependencyErrors, TaggedDependencyError,
};
use crate::dependencies::{break_transitive, DependencyError, DependencyErrors};
use crate::manager::{Manager, Status as ManagerStatus};
use crate::notifications::{NotificationLevel, NotificationSubtype};
use crate::s9pk::manifest::{Manifest, PackageId};

2
appmgr/src/system.rs
View File

@@ -1,9 +1,7 @@
use std::fmt;
use futures::future::try_join_all;
use futures::FutureExt;
use rpc_toolkit::command;
use serde::ser::SerializeStruct;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use tokio::sync::broadcast::Receiver;
use tokio::sync::RwLock;

28
appmgr/src/volume.rs
View File

@@ -133,8 +133,23 @@ impl HasModel for Volumes {
type Model = MapModel<Self>;
}
pub fn asset_dir(ctx: &RpcContext, pkg_id: &PackageId, version: &Version) -> PathBuf {
ctx.datadir
pub fn data_dir<P: AsRef<Path>>(
datadir: P,
pkg_id: &PackageId,
version: &Version,
volume_id: &VolumeId,
) -> PathBuf {
datadir
.as_ref()
.join(PKG_VOLUME_DIR)
.join(pkg_id)
.join("data")
.join(volume_id)
}
pub fn asset_dir<P: AsRef<Path>>(datadir: P, pkg_id: &PackageId, version: &Version) -> PathBuf {
datadir
.as_ref()
.join(PKG_VOLUME_DIR)
.join(pkg_id)
.join("assets")
@@ -189,13 +204,8 @@ impl Volume {
volume_id: &VolumeId,
) -> PathBuf {
match self {
Volume::Data { .. } => ctx
.datadir
.join(PKG_VOLUME_DIR)
.join(pkg_id)
.join("data")
.join(volume_id),
Volume::Assets {} => asset_dir(ctx, pkg_id, version).join(volume_id),
Volume::Data { .. } => data_dir(&ctx.datadir, pkg_id, version, volume_id),
Volume::Assets {} => asset_dir(&ctx.datadir, pkg_id, version).join(volume_id),
Volume::Pointer {
package_id,
volume_id,