diff --git a/Cargo.lock b/Cargo.lock index 90ea64e7c82..155ed5cdc1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3496,10 +3496,14 @@ dependencies = [ name = "tedge" version = "0.12.0" dependencies = [ + "anstyle", "anyhow", "assert_cmd", "assert_matches", "base64 0.13.1", + "c8y-configuration-plugin", + "c8y-firmware-plugin", + "c8y-remote-access-plugin", "camino", "certificate", "clap", @@ -3507,6 +3511,7 @@ dependencies = [ "hyper", "mockito", "mqtt_tests", + "nix", "pad", "pem", "predicates 2.1.5", @@ -3516,6 +3521,11 @@ dependencies = [ "serde", "serde_json", "strum_macros", + "tedge-agent", + "tedge-apt-plugin", + "tedge-log-plugin", + "tedge-mapper", + "tedge-watchdog", "tedge_config", "tedge_utils", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index dc62134ec06..cd6e9e85477 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,7 @@ homepage = "https://thin-edge.io" repository = "https://github.com/thin-edge/thin-edge.io" [workspace.dependencies] +anstyle = "1.0" anyhow = "1.0" assert-json-diff = "2.0" assert_cmd = "2.0" @@ -49,7 +50,10 @@ c8y_firmware_manager = { path = "crates/extensions/c8y_firmware_manager" } c8y_http_proxy = { path = "crates/extensions/c8y_http_proxy" } c8y_auth_proxy = { path = "crates/extensions/c8y_auth_proxy" } c8y_log_manager = { path = "crates/extensions/c8y_log_manager" } +c8y-configuration-plugin = { path = "plugins/c8y_configuration_plugin" } +c8y-firmware-plugin = { path = "plugins/c8y_firmware_plugin" } c8y_mapper_ext = { path = "crates/extensions/c8y_mapper_ext" } +c8y-remote-access-plugin = { path = "plugins/c8y_remote_access_plugin" } camino = "1.1" certificate = { path = "crates/common/certificate" } clap = { version = "4.4", features = ["cargo", "derive"] } @@ -127,6 +131,8 @@ strum_macros = "0.24" syn = { version = "2", features = ["full", "extra-traits"] } tedge_actors = { path = "crates/core/tedge_actors" } tedge_api = { path = "crates/core/tedge_api" } +tedge-apt-plugin = { path = "plugins/tedge_apt_plugin" } +tedge-agent = { path = "crates/core/tedge_agent" } tedge_config = { path = "crates/common/tedge_config" } tedge_config_macros = { path = "crates/common/tedge_config_macros" } tedge_config_macros-impl = { path = "crates/common/tedge_config_macros/impl" } @@ -136,11 +142,14 @@ tedge_file_system_ext = { path = "crates/extensions/tedge_file_system_ext" } tedge_health_ext = { path = "crates/extensions/tedge_health_ext" } tedge_http_ext = { path = "crates/extensions/tedge_http_ext" } tedge_log_manager = { path = "crates/extensions/tedge_log_manager" } +tedge-log-plugin = { path = "plugins/tedge_log_plugin" } +tedge-mapper = { path = "crates/core/tedge_mapper" } tedge_mqtt_ext = { path = "crates/extensions/tedge_mqtt_ext" } tedge_signal_ext = { path = "crates/extensions/tedge_signal_ext" } tedge_test_utils = { path = "crates/tests/tedge_test_utils" } tedge_timer_ext = { path = "crates/extensions/tedge_timer_ext" } tedge_utils = { path = "crates/common/tedge_utils" } +tedge-watchdog = { path = "crates/core/tedge_watchdog" } tempfile = "3.5" test-case = "2.2" thiserror = "1.0" diff --git a/ci/build_scripts/package.sh b/ci/build_scripts/package.sh index bd25a8ad3b1..22766b18e4c 100755 --- a/ci/build_scripts/package.sh +++ b/ci/build_scripts/package.sh @@ -227,15 +227,16 @@ build_tarball() { tar_type="bsdtar" fi + # Only include the tedge binary as it is a multi-call binary case "$tar_type" in bsdtar) # bsd tar requires different options to prevent adding extra "AppleDouble" files, e.g. `._` files, to the archive echo "Using bsdtar, but please consider using gnu-tar instead. Install via: brew install gnu-tar" - COPYFILE_DISABLE=1 tar cfz "$TAR_FILE" --no-xattrs --no-mac-metadata -C "$source_dir" --files-from <(printf "%s\n" "${PACKAGES[@]}") + COPYFILE_DISABLE=1 tar cfz "$TAR_FILE" --no-xattrs --no-mac-metadata -C "$source_dir" --files-from <(printf "%s\n" "tedge") ;; *) # Default to gnu tar (as this is generally the default) - "$tar_cmd" cfz "$TAR_FILE" --no-xattrs --owner=0 --group=0 --mode='0755' -C "$source_dir" --files-from <(printf "%s\n" "${PACKAGES[@]}") + "$tar_cmd" cfz "$TAR_FILE" --no-xattrs --owner=0 --group=0 --mode='0755' -C "$source_dir" --files-from <(printf "%s\n" "tedge") ;; esac diff --git a/configuration/package_manifests/nfpm.c8y-configuration-plugin.yaml b/configuration/package_manifests/nfpm.c8y-configuration-plugin.yaml index f0f549e7b7e..1af877db4a6 100644 --- a/configuration/package_manifests/nfpm.c8y-configuration-plugin.yaml +++ b/configuration/package_manifests/nfpm.c8y-configuration-plugin.yaml @@ -28,10 +28,6 @@ deb: - c8y_configuration_plugin (<= 0.8.1) contents: - # binary - - src: .build/c8y-configuration-plugin - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/c8y-configuration-plugin.service dst: /lib/systemd/system/ diff --git a/configuration/package_manifests/nfpm.c8y-firmware-plugin.yaml b/configuration/package_manifests/nfpm.c8y-firmware-plugin.yaml index e840b0e014f..24786f7669a 100644 --- a/configuration/package_manifests/nfpm.c8y-firmware-plugin.yaml +++ b/configuration/package_manifests/nfpm.c8y-firmware-plugin.yaml @@ -14,6 +14,9 @@ vendor: "thin-edge.io" homepage: "https://thin-edge.io" license: "Apache-2.0" +depends: + - tedge + deb: fields: Vcs-Browser: ${CI_PROJECT_URL} @@ -21,10 +24,6 @@ deb: compression: xz contents: - # binary - - src: .build/c8y-firmware-plugin - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/c8y-firmware-plugin.service dst: /lib/systemd/system/ diff --git a/configuration/package_manifests/nfpm.c8y-remote-access-plugin.yaml b/configuration/package_manifests/nfpm.c8y-remote-access-plugin.yaml index 310e024b716..c436d591500 100644 --- a/configuration/package_manifests/nfpm.c8y-remote-access-plugin.yaml +++ b/configuration/package_manifests/nfpm.c8y-remote-access-plugin.yaml @@ -21,18 +21,8 @@ deb: compression: xz depends: - # Requires tedge user to create the supported operation file. - # The tedge user is created by the tedge package - # so it must be run before this package's postinst. - # Debian seems to work without this depends, however the rpm package - # seems to use another sorting order if the dependency is not defined. - tedge -contents: - # binary - - src: .build/c8y-remote-access-plugin - dst: /usr/bin/ - overrides: apk: scripts: diff --git a/configuration/package_manifests/nfpm.tedge-agent.yaml b/configuration/package_manifests/nfpm.tedge-agent.yaml index dc02745a98a..feab780b9f2 100644 --- a/configuration/package_manifests/nfpm.tedge-agent.yaml +++ b/configuration/package_manifests/nfpm.tedge-agent.yaml @@ -14,6 +14,8 @@ vendor: "thin-edge.io" homepage: "https://thin-edge.io" license: "Apache-2.0" +depends: + - tedge replaces: - tedge_agent provides: @@ -30,10 +32,6 @@ deb: - tedge_agent (<= 0.8.1) contents: - # binary - - src: .build/tedge-agent - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/tedge-agent.service dst: /lib/systemd/system/tedge-agent.service diff --git a/configuration/package_manifests/nfpm.tedge-apt-plugin.yaml b/configuration/package_manifests/nfpm.tedge-apt-plugin.yaml index abad5243902..a37ac6011d3 100644 --- a/configuration/package_manifests/nfpm.tedge-apt-plugin.yaml +++ b/configuration/package_manifests/nfpm.tedge-apt-plugin.yaml @@ -14,6 +14,8 @@ vendor: "thin-edge.io" homepage: "https://thin-edge.io" license: "Apache-2.0" +depends: + - tedge replaces: - tedge_apt_plugin conflicts: @@ -28,10 +30,6 @@ deb: - tedge_apt_plugin (<= 0.8.1) contents: - # binary - - src: .build/tedge-apt-plugin - dst: /usr/bin/ - # Symlink to sm plugin dir - src: /usr/bin/tedge-apt-plugin dst: /etc/tedge/sm-plugins/apt diff --git a/configuration/package_manifests/nfpm.tedge-log-plugin.yaml b/configuration/package_manifests/nfpm.tedge-log-plugin.yaml index 10a6f9ba9de..665818d7218 100644 --- a/configuration/package_manifests/nfpm.tedge-log-plugin.yaml +++ b/configuration/package_manifests/nfpm.tedge-log-plugin.yaml @@ -14,6 +14,8 @@ vendor: "thin-edge.io" homepage: "https://thin-edge.io" license: "Apache-2.0" +depends: + - tedge replaces: - c8y_log_plugin - c8y-log-plugin @@ -31,10 +33,6 @@ deb: - c8y-log-plugin contents: - # binary - - src: .build/tedge-log-plugin - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/tedge-log-plugin.service dst: /lib/systemd/system/ diff --git a/configuration/package_manifests/nfpm.tedge-mapper.yaml b/configuration/package_manifests/nfpm.tedge-mapper.yaml index 5372f221316..a787f2e2510 100644 --- a/configuration/package_manifests/nfpm.tedge-mapper.yaml +++ b/configuration/package_manifests/nfpm.tedge-mapper.yaml @@ -32,10 +32,6 @@ deb: - tedge_mapper (<= 0.8.1) contents: - # binary - - src: .build/tedge-mapper - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/tedge-mapper-aws.service dst: /lib/systemd/system/tedge-mapper-aws.service diff --git a/configuration/package_manifests/nfpm.tedge-watchdog.yaml b/configuration/package_manifests/nfpm.tedge-watchdog.yaml index 7fede4be953..a51590019a4 100644 --- a/configuration/package_manifests/nfpm.tedge-watchdog.yaml +++ b/configuration/package_manifests/nfpm.tedge-watchdog.yaml @@ -14,6 +14,8 @@ vendor: "thin-edge.io" homepage: "https://thin-edge.io" license: "Apache-2.0" +depends: + - tedge replaces: - tedge_watchdog conflicts: @@ -28,10 +30,6 @@ deb: - tedge_watchdog (<= 0.8.1) contents: - # binary - - src: .build/tedge-watchdog - dst: /usr/bin/ - # service definitions - src: ./configuration/init/systemd/tedge-watchdog.service dst: /lib/systemd/system/ diff --git a/crates/core/tedge/Cargo.toml b/crates/core/tedge/Cargo.toml index d84c25155f5..e882802ef24 100644 --- a/crates/core/tedge/Cargo.toml +++ b/crates/core/tedge/Cargo.toml @@ -11,13 +11,18 @@ repository = { workspace = true } readme = "README.md" [dependencies] +anstyle = { workspace = true } anyhow = { workspace = true } base64 = { workspace = true } +c8y-configuration-plugin = { workspace = true } +c8y-firmware-plugin = { workspace = true } +c8y-remote-access-plugin = { workspace = true } camino = { workspace = true } certificate = { workspace = true } -clap = { workspace = true, features = ["cargo", "derive"] } +clap = { workspace = true, features = ["cargo", "derive", "string", "unstable-styles"] } doku = { workspace = true } hyper = { workspace = true, default-features = false } +nix = { workspace = true } pad = { workspace = true } reqwest = { workspace = true, features = [ "blocking", @@ -30,9 +35,15 @@ rumqttc = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } strum_macros = { workspace = true } +tedge-agent = { workspace = true } +tedge-apt-plugin = { workspace = true } +tedge-log-plugin = { workspace = true } +tedge-mapper = { workspace = true } +tedge-watchdog = { workspace = true } tedge_config = { workspace = true } tedge_utils = { workspace = true } thiserror = { workspace = true } +tokio = { workspace = true, features = ["macros"] } toml = { workspace = true } tracing = { workspace = true } url = { workspace = true } diff --git a/crates/core/tedge/src/cli/init.rs b/crates/core/tedge/src/cli/init.rs index d963f31d1df..1d4c8bc8986 100644 --- a/crates/core/tedge/src/cli/init.rs +++ b/crates/core/tedge/src/cli/init.rs @@ -1,6 +1,11 @@ use crate::command::BuildContext; use crate::command::Command; +use crate::Component; +use anyhow::bail; use anyhow::Context; +use clap::Subcommand; +use std::os::unix::fs::MetadataExt; +use std::path::Path; use tedge_utils::file::create_directory; use tedge_utils::file::PermissionEntry; @@ -8,19 +13,94 @@ use tedge_utils::file::PermissionEntry; pub struct TEdgeInitCmd { user: String, group: String, + relative_links: bool, context: BuildContext, } impl TEdgeInitCmd { - pub fn new(user: String, group: String, context: BuildContext) -> Self { + pub fn new(user: String, group: String, relative_links: bool, context: BuildContext) -> Self { Self { user, group, + relative_links, context, } } fn initialize_tedge(&self) -> anyhow::Result<()> { + let executable_name = + std::env::current_exe().context("retrieving the current executable name")?; + let stat = std::fs::metadata(&executable_name).with_context(|| { + format!( + "reading metadata for the current executable ({})", + executable_name.display() + ) + })?; + let Some(executable_dir) = executable_name.parent() else { + bail!( + "current executable ({}) does not have a parent directory", + executable_name.display() + ) + }; + let Some(executable_file_name) = executable_name.file_name() else { + bail!( + "current executable ({}) does not have a file name", + executable_name.display() + ) + }; + + let component_subcommands: Vec = + Component::augment_subcommands(clap::Command::new("tedge")) + .get_subcommands() + .map(|c| c.get_name().to_owned()) + .chain(["tedge-apt-plugin".to_owned()]) + .collect(); + + for component in &component_subcommands { + let link = executable_dir.join(component); + match std::fs::symlink_metadata(&link) { + Err(e) if e.kind() != std::io::ErrorKind::NotFound => bail!( + "couldn't read metadata for {}. do you need to run with sudo?", + link.display() + ), + meta => { + let file_exists = meta.is_ok(); + if file_exists { + nix::unistd::unlink(&link).with_context(|| { + format!("removing old version of {component} at {}", link.display()) + })?; + } + + let tedge = if self.relative_links { + Path::new(executable_file_name) + } else { + &*executable_name + }; + std::os::unix::fs::symlink(tedge, &link).with_context(|| { + format!("creating symlink for {component} to {}", tedge.display()) + })?; + + let res = std::process::Command::new("chown") + .arg("--no-dereference") + .arg(&format!("{}:{}", stat.uid(), stat.gid())) + .arg(&link) + .output() + .with_context(|| { + format!( + "executing chown to change ownership of symlink at {}", + link.display() + ) + })?; + anyhow::ensure!( + res.status.success(), + "failed to change ownership of symlink at {}\n\nSTDERR: {}", + link.display(), + String::from_utf8_lossy(&res.stderr), + ) + } + } + } + let config_dir = self.context.config_location.tedge_config_root_path.clone(); create_directory( &config_dir, diff --git a/crates/core/tedge/src/cli/mod.rs b/crates/core/tedge/src/cli/mod.rs index 33bb7dfc8e3..ea3d36e5851 100644 --- a/crates/core/tedge/src/cli/mod.rs +++ b/crates/core/tedge/src/cli/mod.rs @@ -4,11 +4,17 @@ pub use self::certificate::*; use crate::command::BuildCommand; use crate::command::BuildContext; use crate::command::Command; +use c8y_configuration_plugin::ConfigPluginOpt; +use c8y_firmware_plugin::FirmwarePluginOpt; +use c8y_remote_access_plugin::C8yRemoteAccessPluginOpt; pub use connect::*; +use tedge_agent::AgentOpt; use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tedge_log_plugin::LogfilePluginOpt; +use tedge_mapper::MapperOpt; +use tedge_watchdog::WatchdogOpt; use self::init::TEdgeInitCmd; - mod certificate; mod common; pub mod config; @@ -23,22 +29,47 @@ mod reconnect; name = clap::crate_name!(), version = clap::crate_version!(), about = clap::crate_description!(), - arg_required_else_help(true) + arg_required_else_help(true), + allow_external_subcommands(true), + styles(styles()), + multicall(true), )] +pub enum TEdgeOptMulticall { + Tedge { + #[clap(subcommand)] + cmd: TEdgeOpt, -pub struct Opt { - /// Initialize the tedge - #[clap(long)] - pub init: bool, + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH, global = true)] + config_dir: PathBuf, + }, - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, + #[clap(flatten)] + Component(Component), +} - #[clap(subcommand)] - pub tedge: Option, +#[derive(clap::Parser, Debug)] +pub enum Component { + TedgeMapper(MapperOpt), + + TedgeAgent(AgentOpt), + + TedgeLogPlugin(LogfilePluginOpt), + + C8yConfigurationPlugin(ConfigPluginOpt), + + C8yFirmwarePlugin(FirmwarePluginOpt), + + TedgeWatchdog(WatchdogOpt), + + C8yRemoteAccessPlugin(C8yRemoteAccessPluginOpt), } #[derive(clap::Subcommand, Debug)] +#[clap( + name = clap::crate_name!(), + version = clap::crate_version!(), + about = clap::crate_description!(), +)] pub enum TEdgeOpt { /// Initialize Thin Edge Init { @@ -49,6 +80,11 @@ pub enum TEdgeOpt { /// The group who will own the directories created #[clap(long, default_value = "tedge")] group: String, + + /// Create symlinks to the tedge binary using a relative path + /// (e.g. ./tedge) instead of an absolute path (e.g. /usr/bin/tedge) + #[clap(long)] + relative_links: bool, }, /// Create and manage device certificate @@ -76,10 +112,57 @@ pub enum TEdgeOpt { Mqtt(mqtt::TEdgeMqttCli), } +fn styles() -> clap::builder::Styles { + clap::builder::Styles::styled() + .usage( + anstyle::Style::new() + .bold() + .underline() + .fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Yellow))), + ) + .header( + anstyle::Style::new() + .bold() + .underline() + .fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Yellow))), + ) + .literal( + anstyle::Style::new().fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Green))), + ) + .invalid( + anstyle::Style::new() + .bold() + .fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Red))), + ) + .error( + anstyle::Style::new() + .bold() + .fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Red))), + ) + .valid( + anstyle::Style::new() + .bold() + .underline() + .fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::Green))), + ) + .placeholder( + anstyle::Style::new().fg_color(Some(anstyle::Color::Ansi(anstyle::AnsiColor::White))), + ) +} + impl BuildCommand for TEdgeOpt { fn build_command(self, context: BuildContext) -> Result, crate::ConfigError> { match self { - TEdgeOpt::Init { user, group } => Ok(Box::new(TEdgeInitCmd::new(user, group, context))), + TEdgeOpt::Init { + user, + group, + relative_links, + } => Ok(Box::new(TEdgeInitCmd::new( + user, + group, + relative_links, + context, + ))), TEdgeOpt::Cert(opt) => opt.build_command(context), TEdgeOpt::Config(opt) => opt.build_command(context), TEdgeOpt::Connect(opt) => opt.build_command(context), @@ -89,3 +172,31 @@ impl BuildCommand for TEdgeOpt { } } } + +#[cfg(test)] +mod tests { + use crate::Component; + use crate::TEdgeOptMulticall; + use clap::Parser; + + #[test] + fn tedge_mapper_rejects_with_missing_argument() { + assert!(TEdgeOptMulticall::try_parse_from(["tedge-mapper"]).is_err()); + } + + #[test] + fn tedge_mapper_accepts_with_argument() { + assert!(matches!( + TEdgeOptMulticall::parse_from(["tedge-mapper", "c8y"]), + TEdgeOptMulticall::Component(Component::TedgeMapper(_)) + )); + } + + #[test] + fn tedge_agent_runs_with_no_additional_arguments() { + assert!(matches!( + TEdgeOptMulticall::parse_from(["tedge-agent"]), + TEdgeOptMulticall::Component(Component::TedgeAgent(_)) + )); + } +} diff --git a/crates/core/tedge/src/main.rs b/crates/core/tedge/src/main.rs index e6dc5faafa8..5e97c7e0829 100644 --- a/crates/core/tedge/src/main.rs +++ b/crates/core/tedge/src/main.rs @@ -3,37 +3,86 @@ use anyhow::Context; use clap::Parser; +use std::future::Future; +use std::path::PathBuf; use tedge::command::BuildCommand; use tedge::command::BuildContext; +use tedge::Component; +use tedge::TEdgeOptMulticall; +use tedge_apt_plugin::AptCli; use tedge_config::system_services::set_log_level; -use tracing::log::warn; fn main() -> anyhow::Result<()> { - set_log_level(tracing::Level::WARN); + let executable_name = executable_name(); - let opt = tedge::cli::Opt::parse(); - - if opt.init { - warn!("This --init option has been deprecated and will be removed in a future release. Use the `tedge init` command instead"); - return Ok(()); + if matches!(executable_name.as_deref(), Some("apt" | "tedge-apt-plugin")) { + let try_opt = AptCli::try_parse(); + tedge_apt_plugin::run_and_exit(try_opt); } - let tedge_config_location = tedge_config::TEdgeConfigLocation::from_custom_root(opt.config_dir); - let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); + let opt = parse_multicall_if_known(&executable_name); + match opt { + TEdgeOptMulticall::Component(Component::TedgeMapper(mapper_opt)) => { + block_on(tedge_mapper::run(mapper_opt)) + } + TEdgeOptMulticall::Component(Component::TedgeAgent(opt)) => block_on(tedge_agent::run(opt)), + TEdgeOptMulticall::Component(Component::TedgeLogPlugin(opt)) => { + block_on(tedge_log_plugin::run(opt)) + } + TEdgeOptMulticall::Component(Component::C8yConfigurationPlugin(cp_opt)) => { + block_on(c8y_configuration_plugin::run(cp_opt)) + } + TEdgeOptMulticall::Component(Component::C8yFirmwarePlugin(fp_opt)) => { + block_on(c8y_firmware_plugin::run(fp_opt)) + } + TEdgeOptMulticall::Component(Component::C8yRemoteAccessPlugin(opt)) => { + block_on(c8y_remote_access_plugin::run(opt)).unwrap(); + Ok(()) + } + TEdgeOptMulticall::Component(Component::TedgeWatchdog(opt)) => { + block_on(tedge_watchdog::run(opt)) + } + TEdgeOptMulticall::Tedge { cmd, config_dir } => { + set_log_level(tracing::Level::WARN); - let build_context = BuildContext { - config_repository, - config_location: tedge_config_location, - }; + let tedge_config_location = + tedge_config::TEdgeConfigLocation::from_custom_root(config_dir); + let config_repository = + tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); - if let Some(tedge_opt) = opt.tedge { - let cmd = tedge_opt - .build_command(build_context) - .with_context(|| "missing configuration parameter")?; + let build_context = BuildContext { + config_repository, + config_location: tedge_config_location, + }; + let cmd = cmd + .build_command(build_context) + .with_context(|| "missing configuration parameter")?; - cmd.execute() - .with_context(|| format!("failed to {}", cmd.description())) - } else { - Ok(()) + cmd.execute() + .with_context(|| format!("failed to {}", cmd.description())) + } } } + +fn block_on(future: impl Future) -> T { + tokio::runtime::Runtime::new().unwrap().block_on(future) +} + +fn executable_name() -> Option { + Some( + PathBuf::from(std::env::args_os().next()?) + .file_stem()? + .to_str()? + .to_owned(), + ) +} + +fn parse_multicall_if_known(executable_name: &Option) -> T { + let cmd = T::command(); + + let is_known_subcommand = executable_name + .as_deref() + .map_or(false, |name| cmd.find_subcommand(name).is_some()); + let cmd = cmd.multicall(is_known_subcommand); + T::from_arg_matches(&cmd.get_matches()).expect("get_matches panics if invalid arguments are provided, so we won't have arg matches to convert") +} diff --git a/crates/core/tedge_agent/src/lib.rs b/crates/core/tedge_agent/src/lib.rs new file mode 100644 index 00000000000..19387d2cd14 --- /dev/null +++ b/crates/core/tedge_agent/src/lib.rs @@ -0,0 +1,92 @@ +//! Handles cloud-agnostic operations. +//! +//! The Tedge Agent addresses cloud-agnostic software management operations e.g. +//! listing current installed software list, software update, software removal. +//! Also, the Tedge Agent calls an SM Plugin(s) to execute an action defined by +//! a received operation. +//! +//! It also has following capabilities: +//! +//! - File transfer HTTP server +//! - Restart management +//! - Software management + +use std::sync::Arc; + +use agent::AgentConfig; +use camino::Utf8PathBuf; +use tedge_config::system_services::get_log_level; +use tedge_config::system_services::set_log_level; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tracing::log::warn; + +mod agent; +mod file_transfer_server; +mod restart_manager; +mod software_manager; +mod state_repository; +mod tedge_operation_converter; +mod tedge_to_te_converter; + +#[derive(Debug, Clone, clap::Parser)] +#[clap( +name = clap::crate_name!(), +version = clap::crate_version!(), +about = clap::crate_description!() +)] +pub struct AgentOpt { + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG + #[clap(long)] + pub debug: bool, + + /// Start the agent with clean session off, subscribe to the topics, so that no messages are lost + #[clap(short, long)] + pub init: bool, + + /// Start the agent from custom path + /// + /// WARNING: This is mostly used in testing. + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: Utf8PathBuf, + + /// The device MQTT topic identifier + #[clap(long)] + pub mqtt_device_topic_id: Option>, + + /// MQTT root prefix + #[clap(long)] + pub mqtt_topic_root: Option>, +} + +pub async fn run(agent_opt: AgentOpt) -> Result<(), anyhow::Error> { + let tedge_config_location = + tedge_config::TEdgeConfigLocation::from_custom_root(agent_opt.config_dir.clone()); + + // If `debug` is `false` then only `error!`, `warn!` and `info!` are reported. + // If `debug` is `true` then also `debug!` is reported. + let log_level = if agent_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level("tedge-agent", &tedge_config_location.tedge_config_root_path)? + }; + + set_log_level(log_level); + + let init = agent_opt.init; + + let mut agent = agent::Agent::try_new( + "tedge-agent", + AgentConfig::from_config_and_cliopts(&tedge_config_location, agent_opt)?, + )?; + + if init { + warn!("This --init option has been deprecated and will be removed in a future release"); + return Ok(()); + } else { + agent.start().await?; + } + Ok(()) +} diff --git a/crates/core/tedge_agent/src/main.rs b/crates/core/tedge_agent/src/main.rs index 44641671fc7..0efbe96a006 100644 --- a/crates/core/tedge_agent/src/main.rs +++ b/crates/core/tedge_agent/src/main.rs @@ -1,95 +1,7 @@ -//! Handles cloud-agnostic operations. -//! -//! The Tedge Agent addresses cloud-agnostic software management operations e.g. -//! listing current installed software list, software update, software removal. -//! Also, the Tedge Agent calls an SM Plugin(s) to execute an action defined by -//! a received operation. -//! -//! It also has following capabilities: -//! -//! - File transfer HTTP server -//! - Restart management -//! - Software management - -use std::sync::Arc; - -use agent::AgentConfig; -use camino::Utf8PathBuf; use clap::Parser; -use tedge_config::system_services::get_log_level; -use tedge_config::system_services::set_log_level; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; -use tracing::log::warn; - -mod agent; -mod file_transfer_server; -mod restart_manager; -mod software_manager; -mod state_repository; -mod tedge_operation_converter; -mod tedge_to_te_converter; - -#[derive(Debug, Clone, clap::Parser)] -#[clap( -name = clap::crate_name!(), -version = clap::crate_version!(), -about = clap::crate_description!() -)] -pub struct AgentOpt { - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG - #[clap(long)] - pub debug: bool, - - /// Start the agent with clean session off, subscribe to the topics, so that no messages are lost - #[clap(short, long)] - pub init: bool, - - /// Start the agent from custom path - /// - /// WARNING: This is mostly used in testing. - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: Utf8PathBuf, - - /// The device MQTT topic identifier - #[clap(long)] - pub mqtt_device_topic_id: Option>, - - /// MQTT root prefix - #[clap(long)] - pub mqtt_topic_root: Option>, -} #[tokio::main] async fn main() -> Result<(), anyhow::Error> { - let agent_opt = AgentOpt::parse(); - let tedge_config_location = - tedge_config::TEdgeConfigLocation::from_custom_root(agent_opt.config_dir.clone()); - - // If `debug` is `false` then only `error!`, `warn!` and `info!` are reported. - // If `debug` is `true` then also `debug!` is reported. - let log_level = if agent_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level("tedge-agent", &tedge_config_location.tedge_config_root_path)? - }; - - set_log_level(log_level); - - let init = agent_opt.init; - - let mut agent = agent::Agent::try_new( - "tedge-agent", - AgentConfig::from_config_and_cliopts(&tedge_config_location, agent_opt)?, - )?; - - if init { - warn!("This --init option has been deprecated and will be removed in a future release"); - return Ok(()); - } else { - agent.start().await?; - } - Ok(()) + let agent_opt = tedge_agent::AgentOpt::parse(); + tedge_agent::run(agent_opt).await } diff --git a/crates/core/tedge_mapper/src/az/mapper.rs b/crates/core/tedge_mapper/src/az/mapper.rs index bdeec00bd9f..6a17a0bf341 100644 --- a/crates/core/tedge_mapper/src/az/mapper.rs +++ b/crates/core/tedge_mapper/src/az/mapper.rs @@ -14,13 +14,7 @@ use tracing::warn; const AZURE_MAPPER_NAME: &str = "tedge-mapper-az"; -pub struct AzureMapper {} - -impl AzureMapper { - pub fn new() -> AzureMapper { - AzureMapper {} - } -} +pub struct AzureMapper; #[async_trait] impl TEdgeComponent for AzureMapper { diff --git a/crates/core/tedge_mapper/src/lib.rs b/crates/core/tedge_mapper/src/lib.rs new file mode 100644 index 00000000000..995ee6cbc8d --- /dev/null +++ b/crates/core/tedge_mapper/src/lib.rs @@ -0,0 +1,116 @@ +use crate::aws::mapper::AwsMapper; +use crate::az::mapper::AzureMapper; +use crate::c8y::mapper::CumulocityMapper; +use crate::collectd::mapper::CollectdMapper; +use crate::core::component::TEdgeComponent; +use clap::Parser; +use flockfile::check_another_instance_is_not_running; +use std::fmt; +use std::path::PathBuf; +use tedge_config::system_services::get_log_level; +use tedge_config::system_services::set_log_level; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tracing::log::warn; + +mod aws; +mod az; +mod c8y; +mod collectd; +mod core; + +fn lookup_component(component_name: &MapperName) -> Box { + match component_name { + MapperName::Az => Box::new(AzureMapper), + MapperName::Aws => Box::new(AwsMapper), + MapperName::Collectd => Box::new(CollectdMapper), + MapperName::C8y => Box::new(CumulocityMapper), + } +} + +#[derive(Debug, Parser)] +#[clap( + name = clap::crate_name!(), + version = clap::crate_version!(), + about = clap::crate_description!() +)] +pub struct MapperOpt { + #[clap(subcommand)] + pub name: MapperName, + + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG + #[clap(long, global = true)] + pub debug: bool, + + /// Start the mapper with clean session off, subscribe to the topics, so that no messages are lost + #[clap(short, long)] + pub init: bool, + + /// Start the agent with clean session on, drop the previous session and subscriptions + /// + /// WARNING: All pending messages will be lost. + #[clap(short, long)] + pub clear: bool, + + /// Start the mapper from custom path + /// + /// WARNING: This is mostly used in testing. + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: PathBuf, +} + +#[derive(Debug, clap::Subcommand)] +pub enum MapperName { + Az, + Aws, + C8y, + Collectd, +} + +impl fmt::Display for MapperName { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match self { + MapperName::Az => write!(f, "tedge-mapper-az"), + MapperName::Aws => write!(f, "tedge-mapper-aws"), + MapperName::C8y => write!(f, "tedge-mapper-c8y"), + MapperName::Collectd => write!(f, "tedge-mapper-collectd"), + } + } +} + +pub async fn run(mapper_opt: MapperOpt) -> anyhow::Result<()> { + let component = lookup_component(&mapper_opt.name); + + let tedge_config_location = + tedge_config::TEdgeConfigLocation::from_custom_root(&mapper_opt.config_dir); + let config = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()).load()?; + + let log_level = if mapper_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level( + "tedge-mapper", + &tedge_config_location.tedge_config_root_path, + )? + }; + set_log_level(log_level); + + // Run only one instance of a mapper (if enabled) + let mut _flock = None; + if config.run.lock_files { + let run_dir = config.run.path.as_std_path(); + _flock = check_another_instance_is_not_running(&mapper_opt.name.to_string(), run_dir)?; + } + + if mapper_opt.init { + warn!("This --init option has been deprecated and will be removed in a future release"); + Ok(()) + } else if mapper_opt.clear { + warn!("This --clear option has been deprecated and will be removed in a future release"); + Ok(()) + } else { + component.start(config, &mapper_opt.config_dir).await + } +} diff --git a/crates/core/tedge_mapper/src/main.rs b/crates/core/tedge_mapper/src/main.rs index 3700b1ca18d..a1ccaa45a55 100644 --- a/crates/core/tedge_mapper/src/main.rs +++ b/crates/core/tedge_mapper/src/main.rs @@ -1,119 +1,7 @@ -use crate::aws::mapper::AwsMapper; -use crate::az::mapper::AzureMapper; -use crate::c8y::mapper::CumulocityMapper; -use crate::collectd::mapper::CollectdMapper; -use crate::core::component::TEdgeComponent; use clap::Parser; -use flockfile::check_another_instance_is_not_running; -use std::fmt; -use std::path::PathBuf; -use tedge_config::system_services::get_log_level; -use tedge_config::system_services::set_log_level; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; -use tracing::log::warn; - -mod aws; -mod az; -mod c8y; -mod collectd; -mod core; - -fn lookup_component(component_name: &MapperName) -> Box { - match component_name { - MapperName::Az => Box::new(AzureMapper::new()), - MapperName::Aws => Box::new(AwsMapper), - MapperName::Collectd => Box::new(CollectdMapper), - MapperName::C8y => Box::new(CumulocityMapper), - } -} - -#[derive(Debug, Parser)] -#[clap( - name = clap::crate_name!(), - version = clap::crate_version!(), - about = clap::crate_description!() -)] -pub struct MapperOpt { - #[clap(subcommand)] - pub name: MapperName, - - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG - #[clap(long, global = true)] - pub debug: bool, - - /// Start the mapper with clean session off, subscribe to the topics, so that no messages are lost - #[clap(short, long)] - pub init: bool, - - /// Start the agent with clean session on, drop the previous session and subscriptions - /// - /// WARNING: All pending messages will be lost. - #[clap(short, long)] - pub clear: bool, - - /// Start the mapper from custom path - /// - /// WARNING: This is mostly used in testing. - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, -} - -#[derive(Debug, clap::Subcommand)] -pub enum MapperName { - Az, - Aws, - C8y, - Collectd, -} - -impl fmt::Display for MapperName { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - match self { - MapperName::Az => write!(f, "tedge-mapper-az"), - MapperName::Aws => write!(f, "tedge-mapper-aws"), - MapperName::C8y => write!(f, "tedge-mapper-c8y"), - MapperName::Collectd => write!(f, "tedge-mapper-collectd"), - } - } -} #[tokio::main] async fn main() -> anyhow::Result<()> { - let mapper_opt = MapperOpt::parse(); - - let component = lookup_component(&mapper_opt.name); - - let tedge_config_location = - tedge_config::TEdgeConfigLocation::from_custom_root(&mapper_opt.config_dir); - let config = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()).load()?; - - let log_level = if mapper_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level( - "tedge-mapper", - &tedge_config_location.tedge_config_root_path, - )? - }; - set_log_level(log_level); - - // Run only one instance of a mapper (if enabled) - let mut _flock = None; - if config.run.lock_files { - let run_dir = config.run.path.as_std_path(); - _flock = check_another_instance_is_not_running(&mapper_opt.name.to_string(), run_dir)?; - } - - if mapper_opt.init { - warn!("This --init option has been deprecated and will be removed in a future release"); - Ok(()) - } else if mapper_opt.clear { - warn!("This --clear option has been deprecated and will be removed in a future release"); - Ok(()) - } else { - component.start(config, &mapper_opt.config_dir).await - } + let mapper_opt = tedge_mapper::MapperOpt::parse(); + tedge_mapper::run(mapper_opt).await } diff --git a/crates/core/tedge_watchdog/src/lib.rs b/crates/core/tedge_watchdog/src/lib.rs new file mode 100644 index 00000000000..a86f8031b42 --- /dev/null +++ b/crates/core/tedge_watchdog/src/lib.rs @@ -0,0 +1,56 @@ +use std::path::PathBuf; +use tedge_config::system_services::*; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; + +mod error; + +// on linux, we use systemd +#[cfg(target_os = "linux")] +mod systemd_watchdog; +#[cfg(target_os = "linux")] +use systemd_watchdog as watchdog; + +// on non-linux, we do nothing for now +#[cfg(not(target_os = "linux"))] +mod dummy_watchdog; +#[cfg(not(target_os = "linux"))] +use dummy_watchdog as watchdog; + +#[derive(Debug, clap::Parser)] +#[clap( +name = clap::crate_name!(), +version = clap::crate_version!(), +about = clap::crate_description!() +)] +pub struct WatchdogOpt { + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG + #[clap(long)] + pub debug: bool, + + /// Start the watchdog from custom path + /// + /// WARNING: This is mostly used in testing. + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: PathBuf, +} + +pub async fn run(watchdog_opt: WatchdogOpt) -> Result<(), anyhow::Error> { + let tedge_config_location = + tedge_config::TEdgeConfigLocation::from_custom_root(watchdog_opt.config_dir.clone()); + + let log_level = if watchdog_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level( + "tedge-watchdog", + &tedge_config_location.tedge_config_root_path, + )? + }; + + set_log_level(log_level); + + watchdog::start_watchdog(watchdog_opt.config_dir).await +} diff --git a/crates/core/tedge_watchdog/src/main.rs b/crates/core/tedge_watchdog/src/main.rs index 4bb06b201cd..2134b02d586 100644 --- a/crates/core/tedge_watchdog/src/main.rs +++ b/crates/core/tedge_watchdog/src/main.rs @@ -1,59 +1,7 @@ use clap::Parser; -use std::path::PathBuf; -use tedge_config::system_services::*; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; - -mod error; - -// on linux, we use systemd -#[cfg(target_os = "linux")] -mod systemd_watchdog; -#[cfg(target_os = "linux")] -use systemd_watchdog as watchdog; - -// on non-linux, we do nothing for now -#[cfg(not(target_os = "linux"))] -mod dummy_watchdog; -#[cfg(not(target_os = "linux"))] -use dummy_watchdog as watchdog; - -#[derive(Debug, clap::Parser)] -#[clap( -name = clap::crate_name!(), -version = clap::crate_version!(), -about = clap::crate_description!() -)] -pub struct WatchdogOpt { - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG - #[clap(long)] - pub debug: bool, - - /// Start the watchdog from custom path - /// - /// WARNING: This is mostly used in testing. - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, -} #[tokio::main] -async fn main() -> Result<(), anyhow::Error> { - let watchdog_opt = WatchdogOpt::parse(); - let tedge_config_location = - tedge_config::TEdgeConfigLocation::from_custom_root(watchdog_opt.config_dir.clone()); - - let log_level = if watchdog_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level( - "tedge-watchdog", - &tedge_config_location.tedge_config_root_path, - )? - }; - - set_log_level(log_level); - - watchdog::start_watchdog(watchdog_opt.config_dir).await +async fn main() { + let opt = tedge_watchdog::WatchdogOpt::parse(); + tedge_watchdog::run(opt).await.unwrap(); } diff --git a/plugins/c8y_configuration_plugin/src/lib.rs b/plugins/c8y_configuration_plugin/src/lib.rs new file mode 100644 index 00000000000..b938d983486 --- /dev/null +++ b/plugins/c8y_configuration_plugin/src/lib.rs @@ -0,0 +1,133 @@ +use c8y_config_manager::ConfigManagerBuilder; +use c8y_config_manager::ConfigManagerConfig; +use c8y_http_proxy::credentials::C8YJwtRetriever; +use c8y_http_proxy::C8YHttpProxyBuilder; +use std::path::Path; +use std::path::PathBuf; +use tedge_actors::Runtime; +use tedge_config::system_services::get_log_level; +use tedge_config::system_services::set_log_level; +use tedge_config::CertificateError; +use tedge_config::TEdgeConfig; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tedge_file_system_ext::FsWatchActorBuilder; +use tedge_health_ext::HealthMonitorBuilder; +use tedge_http_ext::HttpActor; +use tedge_mqtt_ext::MqttActorBuilder; +use tedge_mqtt_ext::MqttConfig; +use tedge_signal_ext::SignalActor; +use tedge_timer_ext::TimerActor; +use tracing::log::warn; + +const PLUGIN_NAME: &str = "c8y-configuration-plugin"; + +const AFTER_HELP_TEXT: &str = r#"On start, `c8y-configuration-plugin` notifies the cloud tenant of the managed configuration files, listed in the `CONFIG_FILE`, sending this list with a `119` on `c8y/s/us`. +`c8y-configuration-plugin` subscribes then to `c8y/s/ds` listening for configuration operation requests (messages `524` and `526`). +notifying the Cumulocity tenant of their progress (messages `501`, `502` and `503`). + +The thin-edge `CONFIG_DIR` is used to find where: + * to store temporary files on download: `tedge config get tmp.path`, + * to log operation errors and progress: `tedge config get log.path`, + * to connect the MQTT bus: `tedge config get mqtt.client.port`."#; + +#[derive(Debug, clap::Parser)] +#[clap( +name = clap::crate_name!(), +version = clap::crate_version!(), +about = clap::crate_description!(), +after_help = AFTER_HELP_TEXT +)] +pub struct ConfigPluginOpt { + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG + #[clap(long)] + pub debug: bool, + + /// Create supported operation files + #[clap(short, long)] + pub init: bool, + + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: PathBuf, +} + +pub async fn run(config_plugin_opt: ConfigPluginOpt) -> Result<(), anyhow::Error> { + let config_dir = config_plugin_opt.config_dir; + + // Load tedge config from the provided location + let tedge_config_location = tedge_config::TEdgeConfigLocation::from_custom_root(&config_dir); + let log_level = if config_plugin_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level(PLUGIN_NAME, &tedge_config_location.tedge_config_root_path)? + }; + + set_log_level(log_level); + + let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); + let tedge_config = config_repository.load()?; + + if config_plugin_opt.init { + warn!("This --init option has been deprecated and will be removed in a future release"); + Ok(()) + } else { + run_with(config_dir, tedge_config).await + } +} + +async fn run_with( + config_dir: impl AsRef, + tedge_config: TEdgeConfig, +) -> Result<(), anyhow::Error> { + let runtime_events_logger = None; + let mut runtime = Runtime::try_new(runtime_events_logger).await?; + + // Create actor instances + let mqtt_config = mqtt_config(&tedge_config)?; + let mut jwt_actor = C8YJwtRetriever::builder(mqtt_config.clone()); + let mut http_actor = HttpActor::new().builder(); + let c8y_http_config = (&tedge_config).try_into()?; + let mut c8y_http_proxy_actor = + C8YHttpProxyBuilder::new(c8y_http_config, &mut http_actor, &mut jwt_actor); + + let mut fs_watch_actor = FsWatchActorBuilder::new(); + let mut timer_actor = TimerActor::builder(); + let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(PLUGIN_NAME)); + + // Instantiate health monitor actor + let health_actor = HealthMonitorBuilder::new(PLUGIN_NAME, &mut mqtt_actor); + + // Instantiate config manager actor + let config_manager_config = ConfigManagerConfig::from_tedge_config(config_dir, &tedge_config)?; + let config_actor = ConfigManagerBuilder::try_new( + config_manager_config, + &mut mqtt_actor, + &mut c8y_http_proxy_actor, + &mut timer_actor, + &mut fs_watch_actor, + )?; + + // Shutdown on SIGINT + let signal_actor = SignalActor::builder(&runtime.get_handle()); + + // Run the actors + runtime.spawn(signal_actor).await?; + runtime.spawn(mqtt_actor).await?; + runtime.spawn(jwt_actor).await?; + runtime.spawn(http_actor).await?; + runtime.spawn(c8y_http_proxy_actor).await?; + runtime.spawn(fs_watch_actor).await?; + runtime.spawn(config_actor).await?; + runtime.spawn(timer_actor).await?; + runtime.spawn(health_actor).await?; + + runtime.run_to_completion().await?; + + Ok(()) +} + +fn mqtt_config(tedge_config: &TEdgeConfig) -> Result { + tedge_config.mqtt_config() +} diff --git a/plugins/c8y_configuration_plugin/src/main.rs b/plugins/c8y_configuration_plugin/src/main.rs index f23d8c6146c..26ad03a74b8 100644 --- a/plugins/c8y_configuration_plugin/src/main.rs +++ b/plugins/c8y_configuration_plugin/src/main.rs @@ -1,133 +1,8 @@ -use c8y_config_manager::ConfigManagerBuilder; -use c8y_config_manager::ConfigManagerConfig; -use c8y_http_proxy::credentials::C8YJwtRetriever; -use c8y_http_proxy::C8YHttpProxyBuilder; +use c8y_configuration_plugin::ConfigPluginOpt; use clap::Parser; -use std::path::Path; -use std::path::PathBuf; -use tedge_actors::Runtime; -use tedge_config::system_services::get_log_level; -use tedge_config::system_services::set_log_level; -use tedge_config::CertificateError; -use tedge_config::TEdgeConfig; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; -use tedge_file_system_ext::FsWatchActorBuilder; -use tedge_health_ext::HealthMonitorBuilder; -use tedge_http_ext::HttpActor; -use tedge_mqtt_ext::MqttActorBuilder; -use tedge_mqtt_ext::MqttConfig; -use tedge_signal_ext::SignalActor; -use tedge_timer_ext::TimerActor; -use tracing::log::warn; - -const PLUGIN_NAME: &str = "c8y-configuration-plugin"; - -const AFTER_HELP_TEXT: &str = r#"On start, `c8y-configuration-plugin` notifies the cloud tenant of the managed configuration files, listed in the `CONFIG_FILE`, sending this list with a `119` on `c8y/s/us`. -`c8y-configuration-plugin` subscribes then to `c8y/s/ds` listening for configuration operation requests (messages `524` and `526`). -notifying the Cumulocity tenant of their progress (messages `501`, `502` and `503`). - -The thin-edge `CONFIG_DIR` is used to find where: - * to store temporary files on download: `tedge config get tmp.path`, - * to log operation errors and progress: `tedge config get log.path`, - * to connect the MQTT bus: `tedge config get mqtt.client.port`."#; - -#[derive(Debug, clap::Parser)] -#[clap( -name = clap::crate_name!(), -version = clap::crate_version!(), -about = clap::crate_description!(), -after_help = AFTER_HELP_TEXT -)] -pub struct ConfigPluginOpt { - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG - #[clap(long)] - pub debug: bool, - - /// Create supported operation files - #[clap(short, long)] - pub init: bool, - - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, -} #[tokio::main] -async fn main() -> Result<(), anyhow::Error> { - let config_plugin_opt = ConfigPluginOpt::parse(); - let config_dir = config_plugin_opt.config_dir; - - // Load tedge config from the provided location - let tedge_config_location = tedge_config::TEdgeConfigLocation::from_custom_root(&config_dir); - let log_level = if config_plugin_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level(PLUGIN_NAME, &tedge_config_location.tedge_config_root_path)? - }; - - set_log_level(log_level); - - let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); - let tedge_config = config_repository.load()?; - - if config_plugin_opt.init { - warn!("This --init option has been deprecated and will be removed in a future release"); - Ok(()) - } else { - run(config_dir, tedge_config).await - } -} - -async fn run(config_dir: impl AsRef, tedge_config: TEdgeConfig) -> Result<(), anyhow::Error> { - let runtime_events_logger = None; - let mut runtime = Runtime::try_new(runtime_events_logger).await?; - - // Create actor instances - let mqtt_config = mqtt_config(&tedge_config)?; - let mut jwt_actor = C8YJwtRetriever::builder(mqtt_config.clone()); - let mut http_actor = HttpActor::new().builder(); - let c8y_http_config = (&tedge_config).try_into()?; - let mut c8y_http_proxy_actor = - C8YHttpProxyBuilder::new(c8y_http_config, &mut http_actor, &mut jwt_actor); - - let mut fs_watch_actor = FsWatchActorBuilder::new(); - let mut timer_actor = TimerActor::builder(); - let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(PLUGIN_NAME)); - - // Instantiate health monitor actor - let health_actor = HealthMonitorBuilder::new(PLUGIN_NAME, &mut mqtt_actor); - - // Instantiate config manager actor - let config_manager_config = ConfigManagerConfig::from_tedge_config(config_dir, &tedge_config)?; - let config_actor = ConfigManagerBuilder::try_new( - config_manager_config, - &mut mqtt_actor, - &mut c8y_http_proxy_actor, - &mut timer_actor, - &mut fs_watch_actor, - )?; - - // Shutdown on SIGINT - let signal_actor = SignalActor::builder(&runtime.get_handle()); - - // Run the actors - runtime.spawn(signal_actor).await?; - runtime.spawn(mqtt_actor).await?; - runtime.spawn(jwt_actor).await?; - runtime.spawn(http_actor).await?; - runtime.spawn(c8y_http_proxy_actor).await?; - runtime.spawn(fs_watch_actor).await?; - runtime.spawn(config_actor).await?; - runtime.spawn(timer_actor).await?; - runtime.spawn(health_actor).await?; - - runtime.run_to_completion().await?; - - Ok(()) -} - -fn mqtt_config(tedge_config: &TEdgeConfig) -> Result { - tedge_config.mqtt_config() +async fn main() { + let opt = ConfigPluginOpt::parse(); + c8y_configuration_plugin::run(opt).await.unwrap(); } diff --git a/plugins/c8y_firmware_plugin/src/lib.rs b/plugins/c8y_firmware_plugin/src/lib.rs new file mode 100644 index 00000000000..f00445437ab --- /dev/null +++ b/plugins/c8y_firmware_plugin/src/lib.rs @@ -0,0 +1,114 @@ +use c8y_firmware_manager::FirmwareManagerBuilder; +use c8y_firmware_manager::FirmwareManagerConfig; +use c8y_http_proxy::credentials::C8YJwtRetriever; +use std::path::PathBuf; +use tedge_actors::Runtime; +use tedge_config::system_services::get_log_level; +use tedge_config::system_services::set_log_level; +use tedge_config::TEdgeConfig; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tedge_downloader_ext::DownloaderActor; +use tedge_health_ext::HealthMonitorBuilder; +use tedge_mqtt_ext::MqttActorBuilder; +use tedge_signal_ext::SignalActor; +use tedge_timer_ext::TimerActor; +use tracing::log::warn; + +const PLUGIN_NAME: &str = "c8y-firmware-plugin"; + +const AFTER_HELP_TEXT: &str = r#"`c8y-firmware-plugin` subscribes to `c8y/s/ds` listening for firmware operation requests (message `515`). +Notifying the Cumulocity tenant of their progress (messages `501`, `502` and `503`). +During a successful operation, `c8y-firmware-plugin` updates the installed firmware info in Cumulocity tenant with SmartREST message `115`. + +The thin-edge `CONFIG_DIR` is used to find where: + * to store temporary files on download: `tedge config get tmp.path`, + * to log operation errors and progress: `tedge config get log.path`, + * to connect the MQTT bus: `tedge config get mqtt.bind.port`, + * to timeout pending operations: `tedge config get firmware.child.update.timeout"#; + +#[derive(Debug, clap::Parser)] +#[clap( +name = clap::crate_name!(), +version = clap::crate_version!(), +about = clap::crate_description!(), +after_help = AFTER_HELP_TEXT +)] +pub struct FirmwarePluginOpt { + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG + #[clap(long)] + pub debug: bool, + + /// Create required directories + #[clap(short, long)] + pub init: bool, + + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: PathBuf, +} + +pub async fn run(firmware_plugin_opt: FirmwarePluginOpt) -> Result<(), anyhow::Error> { + // Load tedge config from the provided location + let tedge_config_location = + tedge_config::TEdgeConfigLocation::from_custom_root(&firmware_plugin_opt.config_dir); + let log_level = if firmware_plugin_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level(PLUGIN_NAME, &tedge_config_location.tedge_config_root_path)? + }; + + set_log_level(log_level); + + let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); + let tedge_config = config_repository.load()?; + + if firmware_plugin_opt.init { + warn!("This --init option has been deprecated and will be removed in a future release"); + Ok(()) + } else { + run_with(tedge_config).await + } +} + +async fn run_with(tedge_config: TEdgeConfig) -> Result<(), anyhow::Error> { + let runtime_events_logger = None; + let mut runtime = Runtime::try_new(runtime_events_logger).await?; + + // Create actor instances + let mqtt_config = tedge_config.mqtt_config()?; + let mut jwt_actor = C8YJwtRetriever::builder(mqtt_config.clone()); + let mut timer_actor = TimerActor::builder(); + let mut downloader_actor = DownloaderActor::new().builder(); + let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(PLUGIN_NAME)); + + //Instantiate health monitor actor + let health_actor = HealthMonitorBuilder::new(PLUGIN_NAME, &mut mqtt_actor); + + // Instantiate firmware manager actor + let firmware_manager_config = FirmwareManagerConfig::from_tedge_config(&tedge_config)?; + let firmware_actor = FirmwareManagerBuilder::try_new( + firmware_manager_config, + &mut mqtt_actor, + &mut jwt_actor, + &mut timer_actor, + &mut downloader_actor, + )?; + + // Shutdown on SIGINT + let signal_actor = SignalActor::builder(&runtime.get_handle()); + + // Run the actors + runtime.spawn(signal_actor).await?; + runtime.spawn(mqtt_actor).await?; + runtime.spawn(jwt_actor).await?; + runtime.spawn(downloader_actor).await?; + runtime.spawn(firmware_actor).await?; + runtime.spawn(timer_actor).await?; + runtime.spawn(health_actor).await?; + + runtime.run_to_completion().await?; + + Ok(()) +} diff --git a/plugins/c8y_firmware_plugin/src/main.rs b/plugins/c8y_firmware_plugin/src/main.rs index 3be516ca8ee..891c379e5d8 100644 --- a/plugins/c8y_firmware_plugin/src/main.rs +++ b/plugins/c8y_firmware_plugin/src/main.rs @@ -1,118 +1,7 @@ -use c8y_firmware_manager::FirmwareManagerBuilder; -use c8y_firmware_manager::FirmwareManagerConfig; -use c8y_http_proxy::credentials::C8YJwtRetriever; use clap::Parser; -use std::path::PathBuf; -use tedge_actors::Runtime; -use tedge_config::system_services::get_log_level; -use tedge_config::system_services::set_log_level; -use tedge_config::TEdgeConfig; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; -use tedge_downloader_ext::DownloaderActor; -use tedge_health_ext::HealthMonitorBuilder; -use tedge_mqtt_ext::MqttActorBuilder; -use tedge_signal_ext::SignalActor; -use tedge_timer_ext::TimerActor; -use tracing::log::warn; - -const PLUGIN_NAME: &str = "c8y-firmware-plugin"; - -const AFTER_HELP_TEXT: &str = r#"`c8y-firmware-plugin` subscribes to `c8y/s/ds` listening for firmware operation requests (message `515`). -Notifying the Cumulocity tenant of their progress (messages `501`, `502` and `503`). -During a successful operation, `c8y-firmware-plugin` updates the installed firmware info in Cumulocity tenant with SmartREST message `115`. - -The thin-edge `CONFIG_DIR` is used to find where: - * to store temporary files on download: `tedge config get tmp.path`, - * to log operation errors and progress: `tedge config get log.path`, - * to connect the MQTT bus: `tedge config get mqtt.bind.port`, - * to timeout pending operations: `tedge config get firmware.child.update.timeout"#; - -#[derive(Debug, clap::Parser)] -#[clap( -name = clap::crate_name!(), -version = clap::crate_version!(), -about = clap::crate_description!(), -after_help = AFTER_HELP_TEXT -)] -pub struct FirmwarePluginOpt { - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG - #[clap(long)] - pub debug: bool, - - /// Create required directories - #[clap(short, long)] - pub init: bool, - - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, -} #[tokio::main] -async fn main() -> Result<(), anyhow::Error> { - let firmware_plugin_opt = FirmwarePluginOpt::parse(); - - // Load tedge config from the provided location - let tedge_config_location = - tedge_config::TEdgeConfigLocation::from_custom_root(&firmware_plugin_opt.config_dir); - let log_level = if firmware_plugin_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level(PLUGIN_NAME, &tedge_config_location.tedge_config_root_path)? - }; - - set_log_level(log_level); - - let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location.clone()); - let tedge_config = config_repository.load()?; - - if firmware_plugin_opt.init { - warn!("This --init option has been deprecated and will be removed in a future release"); - Ok(()) - } else { - run(tedge_config).await - } -} - -async fn run(tedge_config: TEdgeConfig) -> Result<(), anyhow::Error> { - let runtime_events_logger = None; - let mut runtime = Runtime::try_new(runtime_events_logger).await?; - - // Create actor instances - let mqtt_config = tedge_config.mqtt_config()?; - let mut jwt_actor = C8YJwtRetriever::builder(mqtt_config.clone()); - let mut timer_actor = TimerActor::builder(); - let mut downloader_actor = DownloaderActor::new().builder(); - let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(PLUGIN_NAME)); - - //Instantiate health monitor actor - let health_actor = HealthMonitorBuilder::new(PLUGIN_NAME, &mut mqtt_actor); - - // Instantiate firmware manager actor - let firmware_manager_config = FirmwareManagerConfig::from_tedge_config(&tedge_config)?; - let firmware_actor = FirmwareManagerBuilder::try_new( - firmware_manager_config, - &mut mqtt_actor, - &mut jwt_actor, - &mut timer_actor, - &mut downloader_actor, - )?; - - // Shutdown on SIGINT - let signal_actor = SignalActor::builder(&runtime.get_handle()); - - // Run the actors - runtime.spawn(signal_actor).await?; - runtime.spawn(mqtt_actor).await?; - runtime.spawn(jwt_actor).await?; - runtime.spawn(downloader_actor).await?; - runtime.spawn(firmware_actor).await?; - runtime.spawn(timer_actor).await?; - runtime.spawn(health_actor).await?; - - runtime.run_to_completion().await?; - - Ok(()) +async fn main() { + let opt = c8y_firmware_plugin::FirmwarePluginOpt::parse(); + c8y_firmware_plugin::run(opt).await.unwrap(); } diff --git a/plugins/c8y_remote_access_plugin/src/input.rs b/plugins/c8y_remote_access_plugin/src/input.rs index 077730ba240..082cb5d7445 100644 --- a/plugins/c8y_remote_access_plugin/src/input.rs +++ b/plugins/c8y_remote_access_plugin/src/input.rs @@ -18,7 +18,7 @@ pub struct RemoteAccessConnect { key: String, } -#[derive(Parser)] +#[derive(Parser, Debug)] #[clap(group(ArgGroup::new("install").args(&["init", "cleanup", "connect_string", "child"])))] #[clap( name = clap::crate_name!(), @@ -26,7 +26,7 @@ version = clap::crate_version!(), about = clap::crate_description!(), arg_required_else_help(true), )] -pub struct Cli { +pub struct C8yRemoteAccessPluginOpt { #[arg(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] config_dir: PathBuf, @@ -47,7 +47,7 @@ pub struct Cli { child: Option, } -impl Cli { +impl C8yRemoteAccessPluginOpt { pub fn get_config_location(&self) -> TEdgeConfigLocation { TEdgeConfigLocation::from_custom_root(&self.config_dir) } @@ -61,21 +61,21 @@ pub enum Command { Connect(RemoteAccessConnect), } -pub fn parse_arguments(cli: Cli) -> miette::Result { +pub fn parse_arguments(cli: C8yRemoteAccessPluginOpt) -> miette::Result { cli.try_into() } -impl TryFrom for Command { +impl TryFrom for Command { type Error = miette::Error; - fn try_from(arguments: Cli) -> Result { + fn try_from(arguments: C8yRemoteAccessPluginOpt) -> Result { match arguments { - Cli { init: true, .. } => Ok(Command::Init), - Cli { cleanup: true, .. } => Ok(Command::Cleanup), - Cli { + C8yRemoteAccessPluginOpt { init: true, .. } => Ok(Command::Init), + C8yRemoteAccessPluginOpt { cleanup: true, .. } => Ok(Command::Cleanup), + C8yRemoteAccessPluginOpt { connect_string: Some(message), .. } => Ok(Command::SpawnChild(message)), - Cli { + C8yRemoteAccessPluginOpt { child: Some(message), .. } => RemoteAccessConnect::deserialize_smartrest(&message).map(Command::Connect), @@ -149,9 +149,11 @@ mod tests { } fn try_parse_arguments(arguments: &[&str]) -> miette::Result { - Cli::try_parse_from(iter::once(&"c8y-remote-access-plugin").chain(arguments)) - .into_diagnostic()? - .try_into() + C8yRemoteAccessPluginOpt::try_parse_from( + iter::once(&"c8y-remote-access-plugin").chain(arguments), + ) + .into_diagnostic()? + .try_into() } #[test] diff --git a/plugins/c8y_remote_access_plugin/src/lib.rs b/plugins/c8y_remote_access_plugin/src/lib.rs new file mode 100644 index 00000000000..c9f5fa4e43a --- /dev/null +++ b/plugins/c8y_remote_access_plugin/src/lib.rs @@ -0,0 +1,249 @@ +use std::path::PathBuf; +use std::process::Stdio; + +use camino::Utf8Path; +use camino::Utf8PathBuf; +use futures::future::try_select; +use futures::future::Either; +use input::parse_arguments; +use miette::miette; +use miette::Context; +use miette::IntoDiagnostic; +use tedge_config::TEdgeConfig; +use tedge_config::TEdgeConfigRepository; +use tedge_utils::file::create_directory_with_user_group; +use tedge_utils::file::create_file_with_user_group; +use tokio::io::AsyncBufReadExt; +use tokio::io::BufReader; +use toml::Table; +use url::Url; + +use crate::auth::Jwt; +pub use crate::input::C8yRemoteAccessPluginOpt; +use crate::input::Command; +use crate::input::RemoteAccessConnect; +use crate::proxy::WebsocketSocketProxy; + +mod auth; +mod csv; +mod input; +mod proxy; + +pub async fn run(opt: C8yRemoteAccessPluginOpt) -> miette::Result<()> { + let config_dir = opt.get_config_location(); + + let tedge_config = TEdgeConfigRepository::new(config_dir.clone()) + .load() + .into_diagnostic() + .context("Reading tedge config")?; + + let command = parse_arguments(opt)?; + + match command { + Command::Init => declare_supported_operation(config_dir.tedge_config_root_path()) + .with_context(|| { + "Failed to initialize c8y-remote-access-plugin. You have to run the command with sudo." + }), + Command::Cleanup => { + remove_supported_operation(config_dir.tedge_config_root_path()); + Ok(()) + } + Command::Connect(command) => proxy(command, tedge_config).await, + Command::SpawnChild(command) => spawn_child(command, config_dir.tedge_config_root_path()).await, + } +} + +fn declare_supported_operation(config_dir: &Utf8Path) -> miette::Result<()> { + let supported_operation_path = supported_operation_path(config_dir); + create_directory_with_user_group( + supported_operation_path.parent().unwrap(), + "tedge", + "tedge", + 0o755, + ) + .into_diagnostic() + .context("Creating supported operations directory")?; + + create_file_with_user_group( + supported_operation_path, + "tedge", + "tedge", + 0o644, + Some( + r#"[exec] +command = "/usr/bin/c8y-remote-access-plugin" +topic = "c8y/s/ds" +on_message = "530" +"#, + ), + ) + .into_diagnostic() + .context("Declaring supported operations") +} + +fn remove_supported_operation(config_dir: &Utf8Path) { + let path = supported_operation_path(config_dir); + // Ignore the error as the file may already have been deleted (#1869) + let _ = std::fs::remove_file(path); +} + +static SUCCESS_MESSAGE: &str = "CONNECTED"; + +#[derive(miette::Diagnostic, Debug, thiserror::Error)] +#[error("Failed while {1}")] +#[diagnostic(help( + "This should never happen. It's very likely a bug in the c8y remote access plugin." +))] +struct Unreachable(#[source] E, &'static str); + +async fn spawn_child(command: String, config_dir: &Utf8Path) -> miette::Result<()> { + let exec_path = get_executable_path(config_dir).await?; + + let mut command = tokio::process::Command::new(exec_path) + .arg("--child") + .arg(command) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .into_diagnostic() + .context("Failed to spawn child process")?; + + let mut stdout = BufReader::new(command.stdout.take().unwrap()); + let mut stderr = BufReader::new(command.stderr.take().unwrap()); + + let copy_error_messages = tokio::task::spawn(async move { + let mut line = String::new(); + while let Ok(amount) = stderr.read_line(&mut line).await { + if amount == 0 { + break; + } + eprint!("{line}"); + line.clear(); + } + }); + + let wait_for_connection = tokio::task::spawn(async move { + let mut line = String::new(); + while stdout.read_line(&mut line).await.is_ok() { + // Copy the output to the parent process stdout to ensure anything we might + // print doesn't get lost before we connect + print!("{line}"); + if line.trim() == SUCCESS_MESSAGE { + break; + } + line.clear(); + } + }); + + let wait_for_failure = tokio::task::spawn(async move { command.wait().await }); + + match try_select(wait_for_connection, wait_for_failure).await { + Ok(Either::Left(_)) => Ok(()), + Ok(Either::Right((Ok(code), _))) => { + copy_error_messages + .await + .map_err(|e| Unreachable(e, "copying stderr from child process"))?; + let code = code.code().unwrap_or(1); + std::process::exit(code) + } + Ok(Either::Right((Err(e), _))) => Err(e) + .into_diagnostic() + .context("Failed to retrieve exit code from child process"), + Err(Either::Left((e, _))) => { + Err(Unreachable(e, "waiting for the connection to be established").into()) + } + Err(Either::Right((e, _))) => Err(Unreachable(e, "waiting for the process to exit").into()), + } +} + +async fn proxy(command: RemoteAccessConnect, config: TEdgeConfig) -> miette::Result<()> { + let host = config + .c8y + .http + .or_config_not_set() + .into_diagnostic()? + .to_string(); + let url = build_proxy_url(host.as_str(), command.key())?; + let jwt = Jwt::retrieve(&config) + .await + .context("Failed when requesting JWT from Cumulocity")?; + + let proxy = WebsocketSocketProxy::connect(&url, command.target_address(), jwt).await?; + + proxy.run().await; + Ok(()) +} + +fn supported_operation_path(config_dir: &Utf8Path) -> Utf8PathBuf { + let mut path = config_dir.to_owned(); + path.push("operations/c8y/c8y_RemoteAccessConnect"); + path +} + +fn build_proxy_url(cumulocity_host: &str, key: &str) -> miette::Result { + format!("wss://{cumulocity_host}/service/remoteaccess/device/{key}") + .parse() + .into_diagnostic() + .context("Creating websocket URL") +} + +async fn get_executable_path(config_dir: &Utf8Path) -> miette::Result { + let operation_path = supported_operation_path(config_dir); + + let content = tokio::fs::read_to_string(&operation_path) + .await + .into_diagnostic() + .with_context(|| { + format!("The operation file {operation_path} does not exist or is not readable.") + })?; + + let operation: Table = content + .parse() + .into_diagnostic() + .with_context(|| format!("Failed to parse {operation_path} file"))?; + + Ok(PathBuf::from( + operation["exec"]["command"] + .as_str() + .ok_or_else(|| miette!("Failed to read command from {operation_path} file"))?, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_supported_operation_path() { + assert_eq!( + supported_operation_path("/etc/tedge".as_ref()), + Utf8PathBuf::from("/etc/tedge/operations/c8y/c8y_RemoteAccessConnect") + ); + } + + #[test] + fn cleanup_existing_operation() { + let dir = tempfile::tempdir().unwrap(); + + let operation_path = create_example_operation(dir.path().try_into().unwrap()); + remove_supported_operation(dir.path().try_into().unwrap()); + + assert!(!operation_path.exists()); + } + + #[test] + fn cleanup_non_existent_operation() { + // Verify that this doesn't panic + remove_supported_operation( + "/tmp/already-deleted-operations-via-removal-of-tedge-agent".into(), + ); + } + + fn create_example_operation(dir: &Utf8Path) -> Utf8PathBuf { + let operation_path = supported_operation_path(dir); + std::fs::create_dir_all(operation_path.parent().unwrap()).unwrap(); + std::fs::File::create(&operation_path).unwrap(); + assert!(operation_path.exists()); + operation_path + } +} diff --git a/plugins/c8y_remote_access_plugin/src/main.rs b/plugins/c8y_remote_access_plugin/src/main.rs index 869adcd0cde..412c45e3a0f 100644 --- a/plugins/c8y_remote_access_plugin/src/main.rs +++ b/plugins/c8y_remote_access_plugin/src/main.rs @@ -1,252 +1,9 @@ -use std::path::PathBuf; -use std::process::Stdio; - -use camino::Utf8Path; -use camino::Utf8PathBuf; +use c8y_remote_access_plugin::C8yRemoteAccessPluginOpt; use clap::Parser; -use futures::future::try_select; -use futures::future::Either; -use input::parse_arguments; -use miette::miette; -use miette::Context; -use miette::IntoDiagnostic; -use tedge_config::TEdgeConfig; -use tedge_config::TEdgeConfigRepository; -use tedge_utils::file::create_directory_with_user_group; -use tedge_utils::file::create_file_with_user_group; -use tokio::io::AsyncBufReadExt; -use tokio::io::BufReader; -use toml::Table; -use url::Url; - -use crate::auth::Jwt; -use crate::input::Command; -use crate::input::RemoteAccessConnect; -use crate::proxy::WebsocketSocketProxy; - -mod auth; -mod csv; -mod input; -mod proxy; #[tokio::main] -async fn main() -> miette::Result<()> { - let opt = input::Cli::parse(); - - let config_dir = opt.get_config_location(); - - let tedge_config = TEdgeConfigRepository::new(config_dir.clone()) - .load() - .into_diagnostic() - .context("Reading tedge config")?; - - let command = parse_arguments(opt)?; - - match command { - Command::Init => declare_supported_operation(config_dir.tedge_config_root_path()) - .with_context(|| { - "Failed to initialize c8y-remote-access-plugin. You have to run the command with sudo." - }), - Command::Cleanup => { - remove_supported_operation(config_dir.tedge_config_root_path()); - Ok(()) - } - Command::Connect(command) => proxy(command, tedge_config).await, - Command::SpawnChild(command) => spawn_child(command, config_dir.tedge_config_root_path()).await, - } -} - -fn declare_supported_operation(config_dir: &Utf8Path) -> miette::Result<()> { - let supported_operation_path = supported_operation_path(config_dir); - create_directory_with_user_group( - supported_operation_path.parent().unwrap(), - "tedge", - "tedge", - 0o755, - ) - .into_diagnostic() - .context("Creating supported operations directory")?; - - create_file_with_user_group( - supported_operation_path, - "tedge", - "tedge", - 0o644, - Some( - r#"[exec] -command = "/usr/bin/c8y-remote-access-plugin" -topic = "c8y/s/ds" -on_message = "530" -"#, - ), - ) - .into_diagnostic() - .context("Declaring supported operations") -} - -fn remove_supported_operation(config_dir: &Utf8Path) { - let path = supported_operation_path(config_dir); - // Ignore the error as the file may already have been deleted (#1869) - let _ = std::fs::remove_file(path); -} - -static SUCCESS_MESSAGE: &str = "CONNECTED"; - -#[derive(miette::Diagnostic, Debug, thiserror::Error)] -#[error("Failed while {1}")] -#[diagnostic(help( - "This should never happen. It's very likely a bug in the c8y remote access plugin." -))] -struct Unreachable(#[source] E, &'static str); - -async fn spawn_child(command: String, config_dir: &Utf8Path) -> miette::Result<()> { - let exec_path = get_executable_path(config_dir).await?; - - let mut command = tokio::process::Command::new(exec_path) - .arg("--child") - .arg(command) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .into_diagnostic() - .context("Failed to spawn child process")?; - - let mut stdout = BufReader::new(command.stdout.take().unwrap()); - let mut stderr = BufReader::new(command.stderr.take().unwrap()); - - let copy_error_messages = tokio::task::spawn(async move { - let mut line = String::new(); - while let Ok(amount) = stderr.read_line(&mut line).await { - if amount == 0 { - break; - } - eprint!("{line}"); - line.clear(); - } - }); - - let wait_for_connection = tokio::task::spawn(async move { - let mut line = String::new(); - while stdout.read_line(&mut line).await.is_ok() { - // Copy the output to the parent process stdout to ensure anything we might - // print doesn't get lost before we connect - print!("{line}"); - if line.trim() == SUCCESS_MESSAGE { - break; - } - line.clear(); - } - }); - - let wait_for_failure = tokio::task::spawn(async move { command.wait().await }); - - match try_select(wait_for_connection, wait_for_failure).await { - Ok(Either::Left(_)) => Ok(()), - Ok(Either::Right((Ok(code), _))) => { - copy_error_messages - .await - .map_err(|e| Unreachable(e, "copying stderr from child process"))?; - let code = code.code().unwrap_or(1); - std::process::exit(code) - } - Ok(Either::Right((Err(e), _))) => Err(e) - .into_diagnostic() - .context("Failed to retrieve exit code from child process"), - Err(Either::Left((e, _))) => { - Err(Unreachable(e, "waiting for the connection to be established").into()) - } - Err(Either::Right((e, _))) => Err(Unreachable(e, "waiting for the process to exit").into()), - } -} - -async fn proxy(command: RemoteAccessConnect, config: TEdgeConfig) -> miette::Result<()> { - let host = config - .c8y - .http - .or_config_not_set() - .into_diagnostic()? - .to_string(); - let url = build_proxy_url(host.as_str(), command.key())?; - let jwt = Jwt::retrieve(&config) - .await - .context("Failed when requesting JWT from Cumulocity")?; - - let proxy = WebsocketSocketProxy::connect(&url, command.target_address(), jwt).await?; - - proxy.run().await; - Ok(()) -} - -fn supported_operation_path(config_dir: &Utf8Path) -> Utf8PathBuf { - let mut path = config_dir.to_owned(); - path.push("operations/c8y/c8y_RemoteAccessConnect"); - path -} - -fn build_proxy_url(cumulocity_host: &str, key: &str) -> miette::Result { - format!("wss://{cumulocity_host}/service/remoteaccess/device/{key}") - .parse() - .into_diagnostic() - .context("Creating websocket URL") -} - -async fn get_executable_path(config_dir: &Utf8Path) -> miette::Result { - let operation_path = supported_operation_path(config_dir); - - let content = tokio::fs::read_to_string(&operation_path) - .await - .into_diagnostic() - .with_context(|| { - format!("The operation file {operation_path} does not exist or is not readable.") - })?; - - let operation: Table = content - .parse() - .into_diagnostic() - .with_context(|| format!("Failed to parse {operation_path} file"))?; - - Ok(PathBuf::from( - operation["exec"]["command"] - .as_str() - .ok_or_else(|| miette!("Failed to read command from {operation_path} file"))?, - )) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn default_supported_operation_path() { - assert_eq!( - supported_operation_path("/etc/tedge".as_ref()), - Utf8PathBuf::from("/etc/tedge/operations/c8y/c8y_RemoteAccessConnect") - ); - } - - #[test] - fn cleanup_existing_operation() { - let dir = tempfile::tempdir().unwrap(); - - let operation_path = create_example_operation(dir.path().try_into().unwrap()); - remove_supported_operation(dir.path().try_into().unwrap()); - - assert!(!operation_path.exists()); - } - - #[test] - fn cleanup_non_existent_operation() { - // Verify that this doesn't panic - remove_supported_operation( - "/tmp/already-deleted-operations-via-removal-of-tedge-agent".into(), - ); - } - - fn create_example_operation(dir: &Utf8Path) -> Utf8PathBuf { - let operation_path = supported_operation_path(dir); - std::fs::create_dir_all(operation_path.parent().unwrap()).unwrap(); - std::fs::File::create(&operation_path).unwrap(); - assert!(operation_path.exists()); - operation_path - } +async fn main() { + miette::set_panic_hook(); + let opt = C8yRemoteAccessPluginOpt::parse(); + c8y_remote_access_plugin::run(opt).await.unwrap(); } diff --git a/plugins/tedge_apt_plugin/src/lib.rs b/plugins/tedge_apt_plugin/src/lib.rs new file mode 100644 index 00000000000..13c3e4bab6c --- /dev/null +++ b/plugins/tedge_apt_plugin/src/lib.rs @@ -0,0 +1,372 @@ +mod error; +mod module_check; + +use crate::error::InternalError; +use crate::module_check::PackageMetadata; +use log::warn; +use regex::Regex; +use serde::Deserialize; +use std::io::{self}; +use std::path::PathBuf; +use std::process::Command; +use std::process::ExitStatus; +use std::process::Stdio; +use tedge_config::TEdgeConfig; +use tedge_config::TEdgeConfigLocation; +use tedge_config::TEdgeConfigRepository; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; + +#[derive(clap::Parser, Debug)] +#[clap( + name = clap::crate_name!(), + version = clap::crate_version!(), + about = clap::crate_description!(), + arg_required_else_help(true) +)] +pub struct AptCli { + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + config_dir: PathBuf, + + #[clap(subcommand)] + operation: PluginOp, +} + +#[derive(clap::Subcommand, Debug)] +pub enum PluginOp { + /// List all the installed modules + List { + /// Filter packages list output by name + #[clap(long, short)] + name: Option, + + /// Filter packages list output by maintainer + #[clap(long, short)] + maintainer: Option, + }, + + /// Install a module + Install { + module: String, + #[clap(short = 'v', long = "module-version")] + version: Option, + #[clap(long = "file")] + file_path: Option, + }, + + /// Uninstall a module + Remove { + module: String, + #[clap(short = 'v', long = "module-version")] + version: Option, + }, + + /// Install or remove multiple modules at once + UpdateList, + + /// Prepare a sequences of install/remove commands + Prepare, + + /// Finalize a sequences of install/remove commands + Finalize, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "lowercase")] +enum UpdateAction { + Install, + Remove, +} +#[derive(Debug, Deserialize)] +struct SoftwareModuleUpdate { + pub action: UpdateAction, + pub name: String, + #[serde(default)] + pub version: Option, + #[serde(default)] + pub path: Option, +} + +fn run_op(operation: PluginOp) -> Result { + let status = match operation { + PluginOp::List { name, maintainer } => { + let dpkg_query = Command::new("dpkg-query") + .args(vec![ + "-f", + "${Package}\t${Version}\t${Maintainer}\t${Status}\n", + "-W", + ]) + .stdout(Stdio::piped()) + .spawn() + .map_err(|err| InternalError::exec_error("dpkg-query", err))? + .wait_with_output() + .map_err(|err| InternalError::exec_error("dpkg-query", err))?; + + let stdout = String::from_utf8(dpkg_query.stdout).unwrap_or_default(); + + let filter = match (&name, &maintainer) { + (None, None) => Regex::new(r"install ok installed").unwrap(), + + _ => match Regex::new( + format!( + r"(^{}\t.*|^\S+\t\S+\t{}\s+.*)install ok installed", + name.unwrap_or_default(), + maintainer.unwrap_or_default() + ) + .as_str(), + ) { + Ok(filter) => filter, + Err(err) => { + eprintln!("tedge-apt-plugin fails to list packages with matching name and maintainer: {err}"); + std::process::exit(1) + } + }, + }; + + for line in stdout.trim_end().lines() { + if filter.is_match(line) { + let (name, version) = get_name_and_version(line); + println!("{name}\t{version}"); + } + } + + dpkg_query.status + } + + PluginOp::Install { + module, + version, + file_path, + } => { + let (installer, _metadata) = get_installer(module, version, file_path)?; + run_cmd( + "apt-get", + &format!("install --quiet --yes --allow-downgrades {}", installer), + )? + } + + PluginOp::Remove { module, version } => { + if let Some(version) = version { + // check the version mentioned present or not + run_cmd( + "apt-get", + &format!("remove --quiet --yes {}={}", module, version), + )? + } else { + run_cmd("apt-get", &format!("remove --quiet --yes {}", module))? + } + } + + PluginOp::UpdateList => { + let mut updates: Vec = Vec::new(); + let mut rdr = csv::ReaderBuilder::new() + .has_headers(false) + .delimiter(b'\t') + .from_reader(io::stdin()); + for result in rdr.deserialize() { + updates.push(result?); + } + + // Maintaining this metadata list to keep the debian package symlinks until the installation is complete, + // which will get cleaned up once it goes out of scope after this block + let mut metadata_vec = Vec::new(); + let mut args: Vec = vec!["install".into(), "--quiet".into(), "--yes".into()]; + for update_module in updates { + match update_module.action { + UpdateAction::Install => { + // if version is `latest` we want to set `version` to an empty value, so + // the apt plugin fetches the most up to date version. + let version = update_module.version.filter(|version| version != "latest"); + + let (installer, metadata) = + get_installer(update_module.name, version, update_module.path)?; + args.push(installer); + metadata_vec.push(metadata); + } + UpdateAction::Remove => { + if let Some(version) = update_module.version { + validate_version(update_module.name.as_str(), version.as_str())? + } + + // Adding a '-' at the end of the package name like 'rolldice-' instructs apt to treat it as removal + args.push(format!("{}-", update_module.name)) + } + }; + } + + println!("apt-get install args: {:?}", args); + let status = Command::new("apt-get") + .args(args) + .stdin(Stdio::null()) + .status() + .map_err(|err| InternalError::exec_error("apt-get", err))?; + + return Ok(status); + } + + PluginOp::Prepare => run_cmd("apt-get", "update --quiet --yes")?, + + PluginOp::Finalize => run_cmd("apt-get", "auto-remove --quiet --yes")?, + }; + + Ok(status) +} + +fn get_installer( + module: String, + version: Option, + file_path: Option, +) -> Result<(String, Option), InternalError> { + match (&version, &file_path) { + (None, None) => Ok((module, None)), + + (Some(version), None) => Ok((format!("{}={}", module, version), None)), + + (None, Some(file_path)) => { + let mut package = PackageMetadata::try_new(file_path)?; + package.validate_package(&[&format!("Package: {}", &module), "Debian package"])?; + Ok((format!("{}", package.file_path().display()), Some(package))) + } + + (Some(version), Some(file_path)) => { + let mut package = PackageMetadata::try_new(file_path)?; + package.validate_package(&[ + &format!("Version: {}", &version), + &format!("Package: {}", &module), + "Debian package", + ])?; + + Ok((format!("{}", package.file_path().display()), Some(package))) + } + } +} + +/// Validate if the provided module version matches the currently installed version +fn validate_version(module_name: &str, module_version: &str) -> Result<(), InternalError> { + // Get the current installed version of the provided package + let output = Command::new("apt") + .arg("list") + .arg("--installed") + .arg(module_name) + .output() + .map_err(|err| InternalError::exec_error("apt-get", err))?; + + let stdout = String::from_utf8(output.stdout)?; + + // Check if the installed version and the provided version match + let second_line = stdout.lines().nth(1); //Ignore line 0 which is always 'Listing...' + if let Some(package_info) = second_line { + if let Some(installed_version) = package_info.split_whitespace().nth(1) + // Value at index 0 is the package name + { + if installed_version != module_version { + return Err(InternalError::MetaDataMismatch { + package: module_name.into(), + expected_key: "Version".into(), + expected_value: installed_version.into(), + provided_value: module_version.into(), + }); + } + } + } + + Ok(()) +} + +fn run_cmd(cmd: &str, args: &str) -> Result { + let args: Vec<&str> = args.split_whitespace().collect(); + let status = Command::new(cmd) + .args(args) + .stdin(Stdio::null()) + .status() + .map_err(|err| InternalError::exec_error(cmd, err))?; + Ok(status) +} + +fn get_name_and_version(line: &str) -> (&str, &str) { + let vec: Vec<&str> = line.split('\t').collect(); + + let name = vec.first().unwrap_or(&"unknown name"); + let version = vec.get(1).unwrap_or(&"unknown version"); + (name, version) +} + +fn get_config(config_dir: PathBuf) -> Option { + let tedge_config_location = TEdgeConfigLocation::from_custom_root(config_dir); + + match TEdgeConfigRepository::new(tedge_config_location).load() { + Ok(config) => Some(config), + Err(err) => { + warn!("Failed to load TEdgeConfig: {}", err); + None + } + } +} + +pub fn run_and_exit(cli: Result) -> ! { + let mut apt = match cli { + Ok(aptcli) => aptcli, + Err(err) => { + err.print().expect("Failed to print help message"); + // re-write the clap exit_status from 2 to 1, if parse fails + std::process::exit(1) + } + }; + + if let PluginOp::List { name, maintainer } = &mut apt.operation { + if let Some(config) = get_config(apt.config_dir) { + if name.is_none() { + *name = config.apt.name.or_none().cloned(); + } + + if maintainer.is_none() { + *maintainer = config.apt.maintainer.or_none().cloned(); + } + } + } + + match run_op(apt.operation) { + Ok(status) if status.success() => { + std::process::exit(0); + } + + Ok(status) => { + if status.code().is_some() { + std::process::exit(2); + } else { + eprintln!("Interrupted by a signal!"); + std::process::exit(4); + } + } + + Err(err) => { + eprintln!("ERROR: {}", err); + std::process::exit(5); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use test_case::test_case; + + #[test_case( + "zsh\t5.8-6+deb11u1\tDebian Zsh Maintainers \tinstall ok installed", + "zsh", "5.8-6+deb11u1" + ; "installed" + )] + fn get_package_name_and_version(line: &str, expected_name: &str, expected_version: &str) { + let (name, version) = get_name_and_version(line); + assert_eq!(name, expected_name); + assert_eq!(version, expected_version); + } + + #[test] + fn both_filters_are_empty_strings() { + let filters = PluginOp::List { + name: Some("".into()), + maintainer: Some("".into()), + }; + assert!(run_op(filters).is_ok()) + } +} diff --git a/plugins/tedge_apt_plugin/src/main.rs b/plugins/tedge_apt_plugin/src/main.rs index 99e734777b8..cb2be35b461 100644 --- a/plugins/tedge_apt_plugin/src/main.rs +++ b/plugins/tedge_apt_plugin/src/main.rs @@ -1,377 +1,6 @@ -mod error; -mod module_check; - -use crate::error::InternalError; -use crate::module_check::PackageMetadata; use clap::Parser; -use log::warn; -use regex::Regex; -use serde::Deserialize; -use std::io::{self}; -use std::path::PathBuf; -use std::process::Command; -use std::process::ExitStatus; -use std::process::Stdio; -use tedge_config::TEdgeConfig; -use tedge_config::TEdgeConfigLocation; -use tedge_config::TEdgeConfigRepository; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; - -#[derive(Parser, Debug)] -#[clap( - name = clap::crate_name!(), - version = clap::crate_version!(), - about = clap::crate_description!(), - arg_required_else_help(true) -)] -struct AptCli { - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - config_dir: PathBuf, - - #[clap(subcommand)] - operation: PluginOp, -} - -#[derive(clap::Subcommand, Debug)] -pub enum PluginOp { - /// List all the installed modules - List { - /// Filter packages list output by name - #[clap(long, short)] - name: Option, - - /// Filter packages list output by maintainer - #[clap(long, short)] - maintainer: Option, - }, - - /// Install a module - Install { - module: String, - #[clap(short = 'v', long = "module-version")] - version: Option, - #[clap(long = "file")] - file_path: Option, - }, - - /// Uninstall a module - Remove { - module: String, - #[clap(short = 'v', long = "module-version")] - version: Option, - }, - - /// Install or remove multiple modules at once - UpdateList, - - /// Prepare a sequences of install/remove commands - Prepare, - - /// Finalize a sequences of install/remove commands - Finalize, -} - -#[derive(Debug, Deserialize)] -#[serde(rename_all = "lowercase")] -enum UpdateAction { - Install, - Remove, -} -#[derive(Debug, Deserialize)] -struct SoftwareModuleUpdate { - pub action: UpdateAction, - pub name: String, - #[serde(default)] - pub version: Option, - #[serde(default)] - pub path: Option, -} - -fn run(operation: PluginOp) -> Result { - let status = match operation { - PluginOp::List { name, maintainer } => { - let dpkg_query = Command::new("dpkg-query") - .args(vec![ - "-f", - "${Package}\t${Version}\t${Maintainer}\t${Status}\n", - "-W", - ]) - .stdout(Stdio::piped()) - .spawn() - .map_err(|err| InternalError::exec_error("dpkg-query", err))? - .wait_with_output() - .map_err(|err| InternalError::exec_error("dpkg-query", err))?; - - let stdout = String::from_utf8(dpkg_query.stdout).unwrap_or_default(); - - let filter = match (&name, &maintainer) { - (None, None) => Regex::new(r"install ok installed").unwrap(), - - _ => match Regex::new( - format!( - r"(^{}\t.*|^\S+\t\S+\t{}\s+.*)install ok installed", - name.unwrap_or_default(), - maintainer.unwrap_or_default() - ) - .as_str(), - ) { - Ok(filter) => filter, - Err(err) => { - eprintln!("tedge-apt-plugin fails to list packages with matching name and maintainer: {err}"); - std::process::exit(1) - } - }, - }; - - for line in stdout.trim_end().lines() { - if filter.is_match(line) { - let (name, version) = get_name_and_version(line); - println!("{name}\t{version}"); - } - } - - dpkg_query.status - } - - PluginOp::Install { - module, - version, - file_path, - } => { - let (installer, _metadata) = get_installer(module, version, file_path)?; - run_cmd( - "apt-get", - &format!("install --quiet --yes --allow-downgrades {}", installer), - )? - } - - PluginOp::Remove { module, version } => { - if let Some(version) = version { - // check the version mentioned present or not - run_cmd( - "apt-get", - &format!("remove --quiet --yes {}={}", module, version), - )? - } else { - run_cmd("apt-get", &format!("remove --quiet --yes {}", module))? - } - } - - PluginOp::UpdateList => { - let mut updates: Vec = Vec::new(); - let mut rdr = csv::ReaderBuilder::new() - .has_headers(false) - .delimiter(b'\t') - .from_reader(io::stdin()); - for result in rdr.deserialize() { - updates.push(result?); - } - - // Maintaining this metadata list to keep the debian package symlinks until the installation is complete, - // which will get cleaned up once it goes out of scope after this block - let mut metadata_vec = Vec::new(); - let mut args: Vec = vec!["install".into(), "--quiet".into(), "--yes".into()]; - for update_module in updates { - match update_module.action { - UpdateAction::Install => { - // if version is `latest` we want to set `version` to an empty value, so - // the apt plugin fetches the most up to date version. - let version = update_module.version.filter(|version| version != "latest"); - - let (installer, metadata) = - get_installer(update_module.name, version, update_module.path)?; - args.push(installer); - metadata_vec.push(metadata); - } - UpdateAction::Remove => { - if let Some(version) = update_module.version { - validate_version(update_module.name.as_str(), version.as_str())? - } - - // Adding a '-' at the end of the package name like 'rolldice-' instructs apt to treat it as removal - args.push(format!("{}-", update_module.name)) - } - }; - } - - println!("apt-get install args: {:?}", args); - let status = Command::new("apt-get") - .args(args) - .stdin(Stdio::null()) - .status() - .map_err(|err| InternalError::exec_error("apt-get", err))?; - - return Ok(status); - } - - PluginOp::Prepare => run_cmd("apt-get", "update --quiet --yes")?, - - PluginOp::Finalize => run_cmd("apt-get", "auto-remove --quiet --yes")?, - }; - - Ok(status) -} - -fn get_installer( - module: String, - version: Option, - file_path: Option, -) -> Result<(String, Option), InternalError> { - match (&version, &file_path) { - (None, None) => Ok((module, None)), - - (Some(version), None) => Ok((format!("{}={}", module, version), None)), - - (None, Some(file_path)) => { - let mut package = PackageMetadata::try_new(file_path)?; - package.validate_package(&[&format!("Package: {}", &module), "Debian package"])?; - Ok((format!("{}", package.file_path().display()), Some(package))) - } - - (Some(version), Some(file_path)) => { - let mut package = PackageMetadata::try_new(file_path)?; - package.validate_package(&[ - &format!("Version: {}", &version), - &format!("Package: {}", &module), - "Debian package", - ])?; - - Ok((format!("{}", package.file_path().display()), Some(package))) - } - } -} - -/// Validate if the provided module version matches the currently installed version -fn validate_version(module_name: &str, module_version: &str) -> Result<(), InternalError> { - // Get the current installed version of the provided package - let output = Command::new("apt") - .arg("list") - .arg("--installed") - .arg(module_name) - .output() - .map_err(|err| InternalError::exec_error("apt-get", err))?; - - let stdout = String::from_utf8(output.stdout)?; - - // Check if the installed version and the provided version match - let second_line = stdout.lines().nth(1); //Ignore line 0 which is always 'Listing...' - if let Some(package_info) = second_line { - if let Some(installed_version) = package_info.split_whitespace().nth(1) - // Value at index 0 is the package name - { - if installed_version != module_version { - return Err(InternalError::MetaDataMismatch { - package: module_name.into(), - expected_key: "Version".into(), - expected_value: installed_version.into(), - provided_value: module_version.into(), - }); - } - } - } - - Ok(()) -} - -fn run_cmd(cmd: &str, args: &str) -> Result { - let args: Vec<&str> = args.split_whitespace().collect(); - let status = Command::new(cmd) - .args(args) - .stdin(Stdio::null()) - .status() - .map_err(|err| InternalError::exec_error(cmd, err))?; - Ok(status) -} - -fn get_name_and_version(line: &str) -> (&str, &str) { - let vec: Vec<&str> = line.split('\t').collect(); - - let name = vec.first().unwrap_or(&"unknown name"); - let version = vec.get(1).unwrap_or(&"unknown version"); - (name, version) -} - -fn get_config(config_dir: PathBuf) -> Option { - let tedge_config_location = TEdgeConfigLocation::from_custom_root(config_dir); - - match TEdgeConfigRepository::new(tedge_config_location).load() { - Ok(config) => Some(config), - Err(err) => { - warn!("Failed to load TEdgeConfig: {}", err); - None - } - } -} fn main() { - let mut apt = match AptCli::try_parse() { - Ok(aptcli) => aptcli, - Err(err) => { - err.print().expect("Failed to print help message"); - // re-write the clap exit_status from 2 to 1, if parse fails - std::process::exit(1) - } - }; - - if let PluginOp::List { - ref mut name, - ref mut maintainer, - } = apt.operation - { - if let Some(config) = get_config(apt.config_dir) { - if name.is_none() { - *name = config.apt.name.or_none().cloned(); - } - - if maintainer.is_none() { - *maintainer = config.apt.maintainer.or_none().cloned(); - } - } - } - - match run(apt.operation) { - Ok(status) if status.success() => { - std::process::exit(0); - } - - Ok(status) => { - if status.code().is_some() { - std::process::exit(2); - } else { - eprintln!("Interrupted by a signal!"); - std::process::exit(4); - } - } - - Err(err) => { - eprintln!("ERROR: {}", err); - std::process::exit(5); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use test_case::test_case; - - #[test_case( - "zsh\t5.8-6+deb11u1\tDebian Zsh Maintainers \tinstall ok installed", - "zsh", "5.8-6+deb11u1" - ; "installed" - )] - fn get_package_name_and_version(line: &str, expected_name: &str, expected_version: &str) { - let (name, version) = get_name_and_version(line); - assert_eq!(name, expected_name); - assert_eq!(version, expected_version); - } - - #[test] - fn both_filters_are_empty_strings() { - let filters = PluginOp::List { - name: Some("".into()), - maintainer: Some("".into()), - }; - assert!(run(filters).is_ok()) - } + let cli = tedge_apt_plugin::AptCli::try_parse(); + tedge_apt_plugin::run_and_exit(cli); } diff --git a/plugins/tedge_log_plugin/src/lib.rs b/plugins/tedge_log_plugin/src/lib.rs new file mode 100644 index 00000000000..3066f9c9a10 --- /dev/null +++ b/plugins/tedge_log_plugin/src/lib.rs @@ -0,0 +1,126 @@ +use clap::Parser; +use std::path::PathBuf; +use std::sync::Arc; +use tedge_actors::Runtime; +use tedge_config::system_services::get_log_level; +use tedge_config::system_services::set_log_level; +use tedge_config::TEdgeConfig; +use tedge_config::TEdgeConfigLocation; +use tedge_config::TEdgeConfigRepository; +use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; +use tedge_file_system_ext::FsWatchActorBuilder; +use tedge_health_ext::HealthMonitorBuilder; +use tedge_http_ext::HttpActor; +use tedge_log_manager::LogManagerBuilder; +use tedge_log_manager::LogManagerConfig; +use tedge_log_manager::LogManagerOptions; +use tedge_mqtt_ext::MqttActorBuilder; +use tedge_signal_ext::SignalActor; +use tracing::info; + +const AFTER_HELP_TEXT: &str = r#"The thin-edge `CONFIG_DIR` is used: +* to find the `tedge.toml` where the following configs are defined: + ** `mqtt.bind.address` and `mqtt.bind.port`: to connect to the tedge MQTT broker + ** `root.topic` and `device.topic`: for the MQTT topics to publish to and subscribe from +* to find/store the `tedge-log-plugin.toml`: the configuration file that specifies which logs to be retrieved"#; + +const TEDGE_LOG_PLUGIN: &str = "tedge-log-plugin"; + +#[derive(Debug, Parser, Clone)] +#[clap( +name = clap::crate_name!(), +version = clap::crate_version!(), +about = clap::crate_description!(), +after_help = AFTER_HELP_TEXT +)] +pub struct LogfilePluginOpt { + /// Turn-on the debug log level. + /// + /// If off only reports ERROR, WARN, and INFO + /// If on also reports DEBUG and TRACE + #[clap(long)] + pub debug: bool, + + #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] + pub config_dir: PathBuf, + + #[clap(long)] + mqtt_topic_root: Option>, + + #[clap(long)] + mqtt_device_topic_id: Option>, +} + +pub async fn run(logfile_opt: LogfilePluginOpt) -> Result<(), anyhow::Error> { + // Load tedge config from the provided location + let tedge_config_location = TEdgeConfigLocation::from_custom_root(&logfile_opt.config_dir); + + let log_level = if logfile_opt.debug { + tracing::Level::DEBUG + } else { + get_log_level( + "tedge-log-plugin", + &tedge_config_location.tedge_config_root_path, + )? + }; + set_log_level(log_level); + + let tedge_config = TEdgeConfigRepository::new(tedge_config_location).load()?; + + run_with(tedge_config, logfile_opt).await +} + +async fn run_with( + tedge_config: TEdgeConfig, + cliopts: LogfilePluginOpt, +) -> Result<(), anyhow::Error> { + let runtime_events_logger = None; + let mut runtime = Runtime::try_new(runtime_events_logger).await?; + + let mqtt_topic_root = cliopts + .mqtt_topic_root + .unwrap_or(tedge_config.mqtt.topic_root.clone().into()); + + let mqtt_device_topic_id = cliopts + .mqtt_device_topic_id + .unwrap_or(tedge_config.mqtt.device_topic_id.clone().into()); + + let mqtt_config = tedge_config.mqtt_config()?; + let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(format!( + "{TEDGE_LOG_PLUGIN}#{mqtt_topic_root}/{mqtt_device_topic_id}", + ))); + + let mut fs_watch_actor = FsWatchActorBuilder::new(); + + let health_actor = HealthMonitorBuilder::new(TEDGE_LOG_PLUGIN, &mut mqtt_actor); + + let mut http_actor = HttpActor::new().builder(); + + // Instantiate log manager actor + let log_manager_config = LogManagerConfig::from_options(LogManagerOptions { + config_dir: cliopts.config_dir, + mqtt_device_topic_id, + mqtt_topic_root, + })?; + let log_actor = LogManagerBuilder::try_new( + log_manager_config, + &mut mqtt_actor, + &mut http_actor, + &mut fs_watch_actor, + )?; + + // Shutdown on SIGINT + let signal_actor = SignalActor::builder(&runtime.get_handle()); + + // Run the actors + runtime.spawn(mqtt_actor).await?; + runtime.spawn(http_actor).await?; + runtime.spawn(fs_watch_actor).await?; + runtime.spawn(log_actor).await?; + runtime.spawn(signal_actor).await?; + runtime.spawn(health_actor).await?; + + info!("Ready to serve log requests"); + runtime.run_to_completion().await?; + Ok(()) +} diff --git a/plugins/tedge_log_plugin/src/main.rs b/plugins/tedge_log_plugin/src/main.rs index d4b26c54533..30880c25ffe 100644 --- a/plugins/tedge_log_plugin/src/main.rs +++ b/plugins/tedge_log_plugin/src/main.rs @@ -1,126 +1,7 @@ use clap::Parser; -use std::path::PathBuf; -use std::sync::Arc; -use tedge_actors::Runtime; -use tedge_config::system_services::get_log_level; -use tedge_config::system_services::set_log_level; -use tedge_config::TEdgeConfig; -use tedge_config::TEdgeConfigLocation; -use tedge_config::TEdgeConfigRepository; -use tedge_config::DEFAULT_TEDGE_CONFIG_PATH; -use tedge_file_system_ext::FsWatchActorBuilder; -use tedge_health_ext::HealthMonitorBuilder; -use tedge_http_ext::HttpActor; -use tedge_log_manager::LogManagerBuilder; -use tedge_log_manager::LogManagerConfig; -use tedge_log_manager::LogManagerOptions; -use tedge_mqtt_ext::MqttActorBuilder; -use tedge_signal_ext::SignalActor; -use tracing::info; - -const AFTER_HELP_TEXT: &str = r#"The thin-edge `CONFIG_DIR` is used: -* to find the `tedge.toml` where the following configs are defined: - ** `mqtt.bind.address` and `mqtt.bind.port`: to connect to the tedge MQTT broker - ** `root.topic` and `device.topic`: for the MQTT topics to publish to and subscribe from -* to find/store the `tedge-log-plugin.toml`: the configuration file that specifies which logs to be retrieved"#; - -const TEDGE_LOG_PLUGIN: &str = "tedge-log-plugin"; - -#[derive(Debug, Parser, Clone)] -#[clap( -name = clap::crate_name!(), -version = clap::crate_version!(), -about = clap::crate_description!(), -after_help = AFTER_HELP_TEXT -)] -pub struct LogfilePluginOpt { - /// Turn-on the debug log level. - /// - /// If off only reports ERROR, WARN, and INFO - /// If on also reports DEBUG and TRACE - #[clap(long)] - pub debug: bool, - - #[clap(long = "config-dir", default_value = DEFAULT_TEDGE_CONFIG_PATH)] - pub config_dir: PathBuf, - - #[clap(long)] - mqtt_topic_root: Option>, - - #[clap(long)] - mqtt_device_topic_id: Option>, -} #[tokio::main] -async fn main() -> Result<(), anyhow::Error> { - let logfile_opt = LogfilePluginOpt::parse(); - - // Load tedge config from the provided location - let tedge_config_location = TEdgeConfigLocation::from_custom_root(&logfile_opt.config_dir); - - let log_level = if logfile_opt.debug { - tracing::Level::DEBUG - } else { - get_log_level( - "tedge-log-plugin", - &tedge_config_location.tedge_config_root_path, - )? - }; - set_log_level(log_level); - - let tedge_config = TEdgeConfigRepository::new(tedge_config_location).load()?; - - run(tedge_config, logfile_opt).await -} - -async fn run(tedge_config: TEdgeConfig, cliopts: LogfilePluginOpt) -> Result<(), anyhow::Error> { - let runtime_events_logger = None; - let mut runtime = Runtime::try_new(runtime_events_logger).await?; - - let mqtt_topic_root = cliopts - .mqtt_topic_root - .unwrap_or(tedge_config.mqtt.topic_root.clone().into()); - - let mqtt_device_topic_id = cliopts - .mqtt_device_topic_id - .unwrap_or(tedge_config.mqtt.device_topic_id.clone().into()); - - let mqtt_config = tedge_config.mqtt_config()?; - let mut mqtt_actor = MqttActorBuilder::new(mqtt_config.clone().with_session_name(format!( - "{TEDGE_LOG_PLUGIN}#{mqtt_topic_root}/{mqtt_device_topic_id}", - ))); - - let mut fs_watch_actor = FsWatchActorBuilder::new(); - - let health_actor = HealthMonitorBuilder::new(TEDGE_LOG_PLUGIN, &mut mqtt_actor); - - let mut http_actor = HttpActor::new().builder(); - - // Instantiate log manager actor - let log_manager_config = LogManagerConfig::from_options(LogManagerOptions { - config_dir: cliopts.config_dir, - mqtt_device_topic_id, - mqtt_topic_root, - })?; - let log_actor = LogManagerBuilder::try_new( - log_manager_config, - &mut mqtt_actor, - &mut http_actor, - &mut fs_watch_actor, - )?; - - // Shutdown on SIGINT - let signal_actor = SignalActor::builder(&runtime.get_handle()); - - // Run the actors - runtime.spawn(mqtt_actor).await?; - runtime.spawn(http_actor).await?; - runtime.spawn(fs_watch_actor).await?; - runtime.spawn(log_actor).await?; - runtime.spawn(signal_actor).await?; - runtime.spawn(health_actor).await?; - - info!("Ready to serve log requests"); - runtime.run_to_completion().await?; - Ok(()) +async fn main() { + let opt = tedge_log_plugin::LogfilePluginOpt::parse(); + tedge_log_plugin::run(opt).await.unwrap(); }