holger-cli/src/run.rs
//! **Run one command**: dial `--at`, make the one call, say the answer.
//!
//! The answer is printed as the server sent it (every field, by its proto name): the
//! client adds no judgement of its own, exactly as the console decides nothing.
use std::io::Read as _;
use std::path::Path;
use crate::cli::*;
use crate::conn::Authed;
use crate::error::{existing, CliError};
use crate::pb;
/// What a client may receive in one message (an artifact, an SBOM, a snapshot).
pub const MAX_MESSAGE: usize = 256 * 1024 * 1024;
/// Pure: lowercase hex SHA-256 of `bytes`.
pub fn sha256_hex(bytes: &[u8]) -> String {
ring::digest::digest(&ring::digest::SHA256, bytes)
.as_ref()
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
/// Pure: one answer, every field shown.
pub fn show<T: std::fmt::Debug>(answer: &T) -> String {
format!("{answer:#?}\n")
}
fn id(name: &str, version: &str, namespace: &str) -> pb::v1::ArtifactId {
pb::v1::ArtifactId {
namespace: namespace.to_owned(),
name: name.to_owned(),
version: version.to_owned(),
}
}
fn coord_id(c: &Coord) -> Option<pb::v1::ArtifactId> {
Some(id(&c.name, &c.version, &c.namespace))
}
fn plain_id(i: &Id) -> Option<pb::v1::ArtifactId> {
Some(id(&i.name, &i.version, &i.namespace))
}
/// Read a whole input file, or [`CliError::PathMissing`].
fn read_input(path: &Path) -> Result<Vec<u8>, CliError> {
let path = existing(path)?;
std::fs::read(&path).map_err(|e| CliError::Local(format!("reading {}: {e}", path.display())))
}
fn audit_filter(f: &crate::cli::AuditFilterArgs) -> pb::v1::AuditFilter {
pb::v1::AuditFilter {
ident: f.who.clone(),
repo: f.repository.clone(),
artifact_prefix: f.artifact.clone(),
action: f.action.clone(),
source_ip: f.source_ip.clone(),
status: f.status,
since_nanos: f.since,
until_nanos: f.until,
}
}
/// Write `bytes` to `out` ("-": stdout, returned to print), answering what to say.
fn write_output(out: &Path, bytes: &[u8], what: &str) -> Result<String, CliError> {
if out.as_os_str() == "-" {
return Ok(String::from_utf8_lossy(bytes).into_owned());
}
std::fs::write(out, bytes)
.map_err(|e| CliError::Local(format!("writing {}: {e}", out.display())))?;
Ok(format!(
"{what}: {} bytes -> {}\n",
bytes.len(),
out.display()
))
}
/// Pure: `key=value` as a property match.
pub fn property_match(kv: &str) -> Result<pb::v1::PropertyMatch, CliError> {
match kv.split_once('=') {
Some((k, v)) if !k.trim().is_empty() => Ok(pb::v1::PropertyMatch {
key: k.trim().to_owned(),
value: v.to_owned(),
}),
_ => Err(CliError::InvalidArgs(format!(
"--property {kv}: write it key=value"
))),
}
}
fn read_stdin_line(what: &str) -> Result<String, CliError> {
let mut s = String::new();
std::io::stdin()
.read_to_string(&mut s)
.map_err(|e| CliError::Local(format!("reading {what} from stdin: {e}")))?;
let s = s.trim().to_owned();
if s.is_empty() {
return Err(CliError::InvalidArgs(format!("no {what} on stdin")));
}
Ok(s)
}
fn source(s: &Source, password: Option<String>) -> pb::v1::MigrationSource {
pb::v1::MigrationSource {
kind: match s.kind {
SourceKind::Nexus => pb::v1::migration_source::Kind::Nexus as i32,
SourceKind::Artifactory => pb::v1::migration_source::Kind::Artifactory as i32,
},
base_url: s.base_url.clone(),
username: s.username.clone(),
password: password.unwrap_or_default(),
repository: s.repository.clone(),
}
}
fn password(s: &Source) -> Result<Option<String>, CliError> {
if s.password_stdin {
read_stdin_line("password").map(Some)
} else {
Ok(None)
}
}
macro_rules! client {
($module:path, $client:ident, $ch:expr) => {{
use $module as m;
m::$client::new($ch.clone())
.max_decoding_message_size(MAX_MESSAGE)
.max_encoding_message_size(MAX_MESSAGE)
}};
}
/// One call, within [`crate::conn::CALL_DEADLINE`] (RA2-4); a stream's answer is the stream,
/// whose every message then has [`crate::conn::STREAM_IDLE`] ([`next`]).
macro_rules! call {
($verb:expr, $fut:expr) => {
call!($verb, $fut, crate::conn::CALL_DEADLINE)
};
($verb:expr, $fut:expr, $deadline:expr) => {
crate::conn::within($verb, $deadline, $fut)
.await?
.map(|r| r.into_inner())
.map_err(|s| CliError::from_status($verb, s))?
};
}
/// The next message of a stream, within [`crate::conn::STREAM_IDLE`].
async fn next<T>(verb: &str, stream: &mut tonic::Streaming<T>) -> Result<Option<T>, CliError> {
crate::conn::within(verb, crate::conn::STREAM_IDLE, stream.message())
.await?
.map_err(|s| CliError::from_status(verb, s))
}
/// **Run `cli`** against its `--at`, with `env` for the credential lookup. Answers what to
/// print. `holger-cli`'s `main` is this with the process environment.
pub async fn run(cli: Cli, env: impl Fn(&str) -> Option<String>) -> Result<String, CliError> {
let at = cli.at.clone();
// The local verbs first: they touch only the credential store.
match &cli.command {
Command::Logout => {
let dir = crate::cred::store_dir(&env).ok_or_else(|| {
CliError::Local("no home to keep credentials in ($HOLGER_CLI_HOME or $HOME)".into())
})?;
crate::cred::store(&dir, &at, None)?;
return Ok(format!(
"forgot the credential for {}\n",
crate::cred::key(&at)
));
}
Command::Login => {
let token = match &cli.token_file {
Some(f) => crate::cred::read_token_file(f)?,
None => read_stdin_line("credential")?,
};
return login(&at, cli.ca.as_deref(), cli.connect.as_deref(), &token, &env).await;
}
_ => {}
}
// The local tools: they dial no server (`--at` is not theirs).
match cli.command {
Command::Bundle(BundleCmd::Tool(t)) => {
local(move || holger_update_bundle::cli::bundle::run(t)).await
}
Command::Verify(a) => local(move || holger_update_bundle::cli::verify::verify(a)).await,
Command::Agent(a) => holger_agent_cli::run(a)
.await
.map(|()| String::new())
.map_err(|e| CliError::Local(format!("{e:#}"))),
command => run_remote(Cli { command, ..cli }, env).await,
}
}
/// A local tool's verb on a blocking thread; it prints its own answer.
async fn local(
f: impl FnOnce() -> anyhow::Result<()> + Send + 'static,
) -> Result<String, CliError> {
tokio::task::spawn_blocking(f)
.await
.map_err(|e| CliError::Local(format!("the tool's thread: {e}")))?
.map(|()| String::new())
.map_err(|e| CliError::Local(format!("{e:#}")))
}
async fn run_remote(cli: Cli, env: impl Fn(&str) -> Option<String>) -> Result<String, CliError> {
let at = cli.at.clone();
let token = crate::cred::resolve(cli.token_file.as_deref(), &env, &at)?;
let ch = crate::conn::dial(
&at,
cli.ca.as_deref(),
cli.connect.as_deref(),
token.as_deref(),
)
.await?;
dispatch(cli.command, &ch).await
}
/// `login`: the credential is checked with the server (it must authenticate), then stored
/// for this `--at` only.
async fn login(
at: &str,
ca: Option<&Path>,
connect: Option<&str>,
token: &str,
env: &impl Fn(&str) -> Option<String>,
) -> Result<String, CliError> {
let ch = crate::conn::dial(at, ca, connect, Some(token)).await?;
let mut admin = client!(pb::v1::admin_service_client, AdminServiceClient, ch);
let me = call!(
"login",
admin.who_am_i(pb::v1::WhoAmIRequest {
repository: String::new()
})
);
if !me.authenticated {
return Err(CliError::Local(format!("the server at {at} does not accept this credential (it reads as anonymous); nothing was stored")));
}
let dir = crate::cred::store_dir(env).ok_or_else(|| {
CliError::Local("no home to keep credentials in ($HOLGER_CLI_HOME or $HOME)".into())
})?;
let file = crate::cred::store(&dir, at, Some(token))?;
Ok(format!(
"logged in to {} as {} ({}, {}); the credential is in {} (0600)\n",
crate::cred::key(at),
me.subject,
me.method,
me.global_role,
file.display()
))
}
async fn dispatch(command: Command, ch: &Authed) -> Result<String, CliError> {
use pb::v1 as h;
Ok(match command {
Command::Login
| Command::Logout
| Command::Verify(_)
| Command::Agent(_)
| Command::Bundle(BundleCmd::Tool(_)) => unreachable!("handled before dialling"),
Command::Health => show(&call!(
"health",
client!(h::admin_service_client, AdminServiceClient, ch).health(h::HealthRequest {})
)),
Command::Whoami { repository } => show(&call!(
"whoami",
client!(h::admin_service_client, AdminServiceClient, ch)
.who_am_i(h::WhoAmIRequest { repository })
)),
Command::Repos => show(&call!(
"repos",
client!(h::admin_service_client, AdminServiceClient, ch)
.list_repositories(h::ListRepositoriesRequest {})
)),
Command::Profile => show(&call!(
"profile",
client!(h::admin_service_client, AdminServiceClient, ch)
.server_profile(h::ServerProfileRequest {})
)),
Command::Snapshot { log_lines } => show(&call!(
"snapshot",
client!(h::admin_service_client, AdminServiceClient, ch)
.admin_snapshot(h::AdminSnapshotRequest { log_lines })
)),
Command::Board => show(&call!(
"board",
client!(h::admin_service_client, AdminServiceClient, ch)
.topology(h::TopologyRequest {})
)),
Command::Flow { json, open } => {
let r = call!(
"flow",
client!(h::admin_service_client, AdminServiceClient, ch)
.flow_svg(h::FlowSvgRequest { open })
);
if json {
format!("{}\n", r.graph_json)
} else {
r.svg
}
}
Command::Security => show(&call!(
"security",
client!(h::admin_service_client, AdminServiceClient, ch)
.security_summary(h::SecuritySummaryRequest {})
)),
Command::Artifacts(a) => artifacts(a, ch).await?,
Command::Archive(ArchiveCmd::Files { repository, prefix }) => show(&call!(
"archive files",
client!(h::archive_service_client, ArchiveServiceClient, ch)
.list_archive_files(h::ListArchiveFilesRequest { repository, prefix })
)),
Command::Archive(ArchiveCmd::Info { repository }) => show(&call!(
"archive info",
client!(h::archive_service_client, ArchiveServiceClient, ch)
.archive_info(h::ArchiveInfoRequest { repository })
)),
Command::Search(s) => {
let properties = s
.properties
.iter()
.map(|p| property_match(p))
.collect::<Result<Vec<_>, _>>()?;
let req = h::SearchRequest {
name: s.name,
namespace: s.namespace,
version: s.version,
checksum: s.checksum,
path: s.path,
repositories: s.repositories,
limit: s.limit,
properties,
};
show(&call!(
"search",
client!(h::search_service_client, SearchServiceClient, ch).search(req)
))
}
Command::Sbom(SbomCmd::Attach { coord, file }) => {
let document = read_input(&file)?;
let req = h::AttachSbomRequest {
repository: coord.repository.clone(),
artifact: coord_id(&coord),
document,
};
show(&call!(
"sbom attach",
client!(h::sbom_service_client, SbomServiceClient, ch).attach_sbom(req)
))
}
Command::Sbom(SbomCmd::Fetch { coord, out }) => {
let r = call!(
"sbom fetch",
client!(h::sbom_service_client, SbomServiceClient, ch).fetch_sbom(
h::FetchSbomRequest {
repository: coord.repository.clone(),
artifact: coord_id(&coord)
}
)
);
write_output(&out, &r.document, &format!("sbom {}", r.content_address))?
}
Command::Property(PropertyCmd::Get { coord }) => show(&call!(
"property get",
client!(h::property_service_client, PropertyServiceClient, ch).get_properties(
h::GetPropertiesRequest {
repository: coord.repository.clone(),
artifact: coord_id(&coord)
}
)
)),
Command::Property(PropertyCmd::Set { coord, key, values }) => {
let req = h::SetPropertyRequest {
repository: coord.repository.clone(),
artifact: coord_id(&coord),
key,
values,
};
show(&call!(
"property set",
client!(h::property_service_client, PropertyServiceClient, ch).set_property(req)
))
}
Command::Promote {
from,
to,
id: i,
r#async,
} => {
let req = h::PromoteArtifactRequest {
from_repo: from,
to_repo: to,
artifact: plain_id(&i),
async_job: r#async,
};
show(&call!(
"promote",
client!(h::promotion_service_client, PromotionServiceClient, ch)
.promote_artifact(req)
))
}
Command::Intake(c) => intake(c, ch).await?,
Command::Audit(AuditCmd::Trail { sha256 }) => show(&call!(
"audit trail",
client!(h::audit_service_client, AuditServiceClient, ch)
.get_artifact_trail(h::GetArtifactTrailRequest { sha256 })
)),
Command::Audit(AuditCmd::Downloads {
who,
repository,
since,
until,
limit,
}) => {
let req = h::QueryDownloadsRequest {
ident: who,
repo: repository,
since_nanos: since,
until_nanos: until,
limit,
};
show(&call!(
"audit downloads",
client!(h::audit_service_client, AuditServiceClient, ch).query_downloads(req)
))
}
Command::Audit(AuditCmd::Query {
filter,
offset,
limit,
}) => {
let req = h::AuditQueryRequest {
filter: Some(audit_filter(&filter)),
offset,
limit,
};
show(&call!(
"audit query",
client!(h::audit_service_client, AuditServiceClient, ch).query_audit(req)
))
}
Command::Audit(AuditCmd::Aggregate {
filter,
by,
ok_only,
}) => {
let req = h::AuditAggregateRequest {
filter: Some(audit_filter(&filter)),
group_by: by,
ok_only,
max_groups: 0,
};
show(&call!(
"audit aggregate",
client!(h::audit_service_client, AuditServiceClient, ch).aggregate_audit(req)
))
}
Command::Audit(AuditCmd::Export {
filter,
out,
since_seq,
limit,
}) => {
let req = h::AuditExportRequest {
filter: Some(audit_filter(&filter)),
since_seq,
limit,
};
let x = call!(
"audit export",
client!(h::audit_service_client, AuditServiceClient, ch).export_audit(req)
);
let mut said = write_output(
&out,
&x.arrow_ipc,
&format!("audit export ({} rows, {})", x.rows, x.layout),
)?;
if x.truncated {
said.push_str(&format!(
"more rows match: run again with --since-seq {}\n",
x.next_seq
));
}
said
}
Command::Audit(AuditCmd::Verify) => show(&call!(
"audit verify",
client!(h::audit_service_client, AuditServiceClient, ch)
.verify_audit_chain(h::VerifyAuditChainRequest {})
)),
Command::Migration(c) => migration(c, ch).await?,
Command::Replicate {
from,
to,
execute,
limit,
r#async,
} => {
let req = h::ReplicateRequest {
from_repo: from,
to_repo: to,
execute,
limit,
async_job: r#async,
};
show(&call!(
"replicate",
client!(h::replication_service_client, ReplicationServiceClient, ch).replicate(req)
))
}
Command::Jobs { kind, state, limit } => {
show(&call!(
"jobs",
client!(h::job_service_client, JobServiceClient, ch)
.list_jobs(h::ListJobsRequest { kind, limit, state })
))
}
Command::Admission(AdmissionCmd::Usage { limit }) => show(&call!(
"admission usage",
client!(h::admission_service_client, AdmissionServiceClient, ch)
.list_usage(h::ListUsageRequest { limit })
)),
Command::Admission(AdmissionCmd::Plan {
consumer,
work,
size_hint,
}) => show(&call!(
"admission plan",
client!(h::admission_service_client, AdmissionServiceClient, ch).admit_plan(
h::AdmitPlanRequest {
consumer,
work,
size_hint
}
)
)),
Command::Usage { repository } => show(&call!(
"usage",
client!(h::usage_service_client, UsageServiceClient, ch)
.read_usage(h::ReadUsageRequest { repository })
)),
Command::Room(RoomCmd::Show { repository }) => show(&call!(
"room show",
client!(pb::room::room_service_client, RoomServiceClient, ch)
.room(pb::room::RoomRequest { repository })
)),
Command::Room(RoomCmd::PrivateEstate) => show(&call!(
"room private-estate",
client!(pb::room::room_service_client, RoomServiceClient, ch)
.private_estate(pb::room::PrivateEstateRequest {})
)),
Command::Account(AccountCmd::Mine) => show(&call!(
"account mine",
client!(
pb::account::account_service_client,
AccountServiceClient,
ch
)
.my_repositories(pb::account::MyRepositoriesRequest {})
)),
Command::Account(AccountCmd::Create { name, format }) => show(&call!(
"account create",
client!(
pb::account::account_service_client,
AccountServiceClient,
ch
)
.create_repository(pb::account::CreateRepositoryRequest { name, format })
)),
Command::Account(AccountCmd::Grant {
repository,
person,
access,
}) => show(&call!(
"account grant",
client!(
pb::account::account_service_client,
AccountServiceClient,
ch
)
.grant(pb::account::GrantRequest {
repository,
person,
access
})
)),
Command::Token(c) => token(c, ch).await?,
Command::Bundle(c) => bundle(c, ch).await?,
Command::Intel(IntelCmd::Import { file }) => {
let path = existing(&file)?;
let bytes = std::fs::metadata(&path)
.map_err(|e| CliError::Local(format!("reading {}: {e}", path.display())))?
.len();
let mut admin = client!(h::admin_service_client, AdminServiceClient, ch);
show(&call!(
"intel import",
admin.import_intel_bundle(import_chunks(path)),
import_deadline(bytes)
))
}
Command::Downstream(DownstreamCmd::Invite {
name,
door,
hub_key,
endpoint,
out,
}) => {
if out.exists() {
return Err(CliError::InvalidArgs(format!(
"{} already exists; the invite is written to a fresh file",
out.display()
)));
}
let r = call!(
"downstream invite",
client!(
pb::doors::downstream_service_client,
DownstreamServiceClient,
ch
)
.invite(pb::doors::DownstreamInviteRequest {
name: name.clone(),
door,
hub_key,
endpoint
})
);
// 0600: a WireGuard invite (schema v2) holds the link's pre-shared key.
use std::io::Write as _;
use std::os::unix::fs::OpenOptionsExt as _;
let mut f = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&out)
.map_err(|e| CliError::Local(format!("writing {}: {e}", out.display())))?;
f.write_all(&r.invite)
.and_then(|()| f.sync_all())
.map_err(|e| CliError::Local(format!("writing {}: {e}", out.display())))?;
let mut said = format!(
"INVITED {name} -> {} ({})\n door: {}\n",
out.display(),
r.schema,
r.door
);
if !r.address.is_empty() {
said.push_str(&format!(
" address: {} (on the front's downstream interface)\n",
r.address
));
}
said.push_str(&format!(
" its RON trusts: (name: \"{}\", key: \"{}\", invite: Some(\"{}\"))\n",
r.upstream,
r.key,
out.display()
));
said
}
Command::FirstKey { display } => {
let r = call!(
"first-key",
client!(
pb::doors::first_key_service_client,
FirstKeyServiceClient,
ch
)
.invite(pb::doors::FirstKeyRequest { display })
);
// The secret on its own line, once, so it can be taken without the rest.
format!(
"{}\nid {} expires {} groups [{}]\nopen on the console: {}\n{}\n",
r.invite,
r.id,
r.expires,
r.groups.join(", "),
r.path,
r.words
)
}
})
}
async fn artifacts(c: ArtifactsCmd, ch: &Authed) -> Result<String, CliError> {
use pb::v1 as h;
let mut repo = client!(h::repository_service_client, RepositoryServiceClient, ch);
Ok(match c {
ArtifactsCmd::List {
repository,
filter,
limit,
page_token,
} => show(&call!(
"artifacts list",
repo.list_artifacts(h::ListArtifactsRequest {
repository,
name_filter: filter,
limit,
page_token
})
)),
ArtifactsCmd::Fetch { coord, out } => {
let r = call!(
"artifacts fetch",
repo.fetch_artifact(h::FetchArtifactRequest {
repository: coord.repository.clone(),
id: coord_id(&coord)
})
);
write_output(
&out,
&r.data,
&format!("fetched {} ({})", coord.name, r.content_type),
)?
}
ArtifactsCmd::Put {
coord,
file,
r#async,
} => {
// Streamed (stinky 14): read from the file as it is sent, never whole in memory.
let path = existing(&file)?;
let bytes = std::fs::metadata(&path)
.map_err(|e| CliError::Local(format!("reading {}: {e}", path.display())))?
.len();
let first = h::PutArtifactChunk {
repository: coord.repository.clone(),
id: coord_id(&coord),
data: Vec::new(),
async_job: r#async,
};
show(&call!(
"artifacts put",
repo.put_artifact_stream(upload_chunks(
std::fs::File::open(&path)
.map_err(|e| CliError::Local(format!("reading {}: {e}", path.display())))?,
first
)),
import_deadline(bytes)
))
}
})
}
async fn intake(c: IntakeCmd, ch: &Authed) -> Result<String, CliError> {
use pb::v1 as h;
let mut p = client!(h::promotion_service_client, PromotionServiceClient, ch);
Ok(match c {
IntakeCmd::List { queue, limit } => show(&call!(
"intake list",
p.list_intake_artifacts(h::ListIntakeArtifactsRequest { queue, limit })
)),
IntakeCmd::State { id: i } => show(&call!(
"intake state",
p.get_artifact_state(h::GetArtifactStateRequest {
artifact: plain_id(&i)
})
)),
IntakeCmd::Evidence { id: i } => show(&call!(
"intake evidence",
p.get_artifact_evidence(h::GetArtifactEvidenceRequest {
artifact: plain_id(&i)
})
)),
IntakeCmd::Advance {
id: i,
verb,
to_state,
reason,
} => {
let verb = match verb {
Verb::Advance => h::IntakeVerb::Advance,
Verb::Quarantine => h::IntakeVerb::Quarantine,
Verb::Release => h::IntakeVerb::Release,
Verb::Block => h::IntakeVerb::Block,
} as i32;
show(&call!(
"intake advance",
p.advance_artifact(h::AdvanceArtifactRequest {
artifact: plain_id(&i),
verb,
to_state,
reason,
evidence: Vec::new()
})
))
}
IntakeCmd::Sets { status } => show(&call!(
"intake sets",
p.list_dependency_sets(h::ListDependencySetsRequest { status })
)),
IntakeCmd::Set(SetCmd {
verb:
Some(SetVerb::Approve {
id,
revision,
reason,
}),
..
}) => show(&call!(
"intake set approve",
p.approve_dependency_set(h::JudgeDependencySetRequest {
id,
reason,
revision
})
)),
IntakeCmd::Set(SetCmd {
verb:
Some(SetVerb::Refuse {
id,
revision,
reason,
}),
..
}) => show(&call!(
"intake set refuse",
p.refuse_dependency_set(h::JudgeDependencySetRequest {
id,
reason,
revision
})
)),
IntakeCmd::Set(SetCmd { id, verb: None }) => show(&call!(
"intake set",
p.get_dependency_set(h::GetDependencySetRequest {
id: id.unwrap_or_default()
})
)),
})
}
async fn migration(c: MigrationCmd, ch: &Authed) -> Result<String, CliError> {
use pb::v1 as h;
let mut m = client!(h::migration_service_client, MigrationServiceClient, ch);
Ok(match c {
MigrationCmd::Discover(s) => {
let pw = password(&s)?;
show(&call!(
"migration discover",
m.discover(h::DiscoverMigrationRequest {
source: Some(source(&s, pw))
})
))
}
MigrationCmd::Start {
source: s,
concurrency,
} => {
let pw = password(&s)?;
show(&call!(
"migration start",
m.start(h::StartMigrationRequest {
source: Some(source(&s, pw)),
concurrency,
skip_digest_verification: false
})
))
}
MigrationCmd::Watch { job } => {
let mut stream = call!(
"migration watch",
m.watch(h::WatchMigrationRequest { job_id: job })
);
let mut out = String::new();
while let Some(ev) = next("migration watch", &mut stream).await? {
out.push_str(&format!(
"{} {} done {} failed {} skipped {} of {} ({} bytes){}{}\n",
ev.job_id,
ev.repo,
ev.done,
ev.failed,
ev.skipped,
ev.total,
ev.bytes,
if ev.job_finished { " FINISHED" } else { "" },
if ev.message.is_empty() {
String::new()
} else {
format!(": {}", ev.message)
}
));
if ev.job_finished {
break;
}
}
out
}
MigrationCmd::Cancel { job } => show(&call!(
"migration cancel",
m.cancel(h::CancelMigrationRequest { job_id: job })
)),
})
}
async fn token(c: TokenCmd, ch: &Authed) -> Result<String, CliError> {
use pb::token as t;
let mut tok = client!(t::token_service_client, TokenServiceClient, ch);
Ok(match c {
TokenCmd::Mint {
label,
repositories,
scope,
days,
} => {
let scope = match scope {
Scope::Read => t::TokenScope::Read,
Scope::ReadWrite => t::TokenScope::ReadWrite,
} as i32;
let r = call!(
"token mint",
tok.mint(t::MintTokenRequest {
label,
repositories,
scope,
days
})
);
// The secret on its own line, once.
format!(
"{}\n{}{}\n",
r.secret,
r.token.as_ref().map(show).unwrap_or_default(),
r.words
)
}
TokenCmd::List => show(&call!("token list", tok.list(t::ListTokensRequest {}))),
TokenCmd::Revoke { id } => show(&call!(
"token revoke",
tok.revoke(t::RevokeTokenRequest { id })
)),
})
}
/// **A streamed export written to disk as it arrives** (RA2-5): each member lands in a fresh
/// directory beside `out` and the whole is renamed to `out` only once the summary's
/// `candidate_sha256` matches the `core.json` that arrived and every member is there.
pub struct ExportWriter {
out: std::path::PathBuf,
/// Where the members land until the checks pass (beside `out`, never in it).
partial: std::path::PathBuf,
current: Option<(String, std::fs::File)>,
members: std::collections::BTreeSet<String>,
core: Option<ring::digest::Context>,
done: bool,
}
/// The member whose digest the summary names (`candidate_sha256` = sha256 of core.json).
pub const CORE_JSON: &str = "core.json";
impl ExportWriter {
pub fn new(out: &Path) -> Result<Self, CliError> {
let name = out
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_else(|| "export".to_owned());
let partial = out.with_file_name(format!(".{name}.partial-{}", std::process::id()));
std::fs::create_dir(&partial)
.map_err(|e| CliError::Local(format!("creating {}: {e}", partial.display())))?;
Ok(ExportWriter {
out: out.to_owned(),
partial,
current: None,
members: Default::default(),
core: None,
done: false,
})
}
fn fail(&mut self, why: String) -> CliError {
self.current = None;
let _ = std::fs::remove_dir_all(&self.partial);
self.done = true;
CliError::Local(format!("{why}; nothing was kept"))
}
/// One chunk, written as it arrives; the summary ends it (`Some`, once the digest and
/// the member count match and the candidate is at `out`).
pub fn take(
&mut self,
c: pb::doors::ExportBundleChunk,
) -> Result<Option<pb::doors::ExportBundleSummary>, CliError> {
use std::io::Write as _;
if let Some(s) = c.summary {
self.current = None;
let got = self.core.take().map(|h| {
h.finish()
.as_ref()
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
});
if s.candidate_sha256.is_empty()
|| got.as_deref() != Some(s.candidate_sha256.to_ascii_lowercase().as_str())
{
return Err(self.fail(format!("the export's {CORE_JSON} hashes to sha256 {}, not the summary's candidate_sha256 {:?}", got.unwrap_or_else(|| "(no core.json arrived)".into()), s.candidate_sha256)));
}
if s.members as usize != self.members.len() {
return Err(self.fail(format!(
"the summary says {} member(s) and {} arrived",
s.members,
self.members.len()
)));
}
std::fs::rename(&self.partial, &self.out).map_err(|e| {
self.fail(format!(
"moving the candidate to {}: {e}",
self.out.display()
))
})?;
self.done = true;
return Ok(Some(s));
}
if c.path.is_empty()
|| c.path.starts_with('/')
|| c.path
.split('/')
.any(|p| p == ".." || p.is_empty() || p == ".")
{
return Err(self.fail(format!(
"the server sent a member outside the candidate ({:?})",
c.path
)));
}
if self.current.as_ref().map(|(p, _)| p.as_str()) != Some(c.path.as_str()) {
let dest = self.partial.join(&c.path);
if let Some(parent) = dest.parent() {
std::fs::create_dir_all(parent)
.map_err(|e| self.fail(format!("creating {}: {e}", parent.display())))?;
}
// A member that comes back after another is appended to, as the old collector did.
let file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&dest)
.map_err(|e| self.fail(format!("writing {}: {e}", dest.display())))?;
self.members.insert(c.path.clone());
self.current = Some((c.path.clone(), file));
}
if c.path == CORE_JSON {
self.core
.get_or_insert_with(|| ring::digest::Context::new(&ring::digest::SHA256))
.update(&c.data);
}
let write = self.current.as_mut().map(|(_, f)| f.write_all(&c.data));
if let Some(Err(e)) = write {
return Err(self.fail(format!(
"writing {}: {e}",
self.partial.join(&c.path).display()
)));
}
Ok(None)
}
/// How many members were written.
pub fn members(&self) -> usize {
self.members.len()
}
}
impl Drop for ExportWriter {
/// An export that ended without its summary (the stream broke, a deadline) leaves nothing.
fn drop(&mut self) {
if !self.done {
self.current = None;
let _ = std::fs::remove_dir_all(&self.partial);
}
}
}
/// **The import's chunks, read from disk as they are sent** (RA2-5): 1 MiB at a time,
/// never the whole file in memory. No chunk asks for a first import: that is an admin's
/// authorization on the secure side (lane HOL-CDS), and the server refuses the flag.
fn import_chunks(
path: std::path::PathBuf,
) -> impl tonic::codegen::tokio_stream::Stream<Item = pb::v1::ImportBundleChunk> + Send + 'static {
let (tx, rx) = tokio::sync::mpsc::channel::<pb::v1::ImportBundleChunk>(4);
tokio::task::spawn_blocking(move || {
use std::io::Read as _;
let Ok(mut f) = std::fs::File::open(&path) else {
return;
};
let mut first = true;
loop {
let mut data = vec![0u8; IMPORT_CHUNK];
let mut n = 0;
while n < data.len() {
match f.read(&mut data[n..]) {
Ok(0) => break,
Ok(k) => n += k,
Err(_) => return,
}
}
data.truncate(n);
if n == 0 && !first {
return;
}
let last = n < IMPORT_CHUNK;
if tx
.blocking_send(pb::v1::ImportBundleChunk {
data,
first_import: false,
})
.is_err()
{
return;
}
first = false;
if last {
return;
}
}
});
tonic::codegen::tokio_stream::wrappers::ReceiverStream::new(rx)
}
/// One import chunk.
pub const IMPORT_CHUNK: usize = 1024 * 1024;
/// One upload chunk (`artifacts put`).
pub const UPLOAD_CHUNK: usize = 1024 * 1024;
/// How many upload chunks wait to be sent at most.
pub const UPLOAD_DEPTH: usize = 4;
/// **What an upload holds in memory at most**, whatever the file's size: the chunks waiting
/// to be sent, the one being sent and the one being read.
pub const UPLOAD_MEMORY_BOUND: usize = (UPLOAD_DEPTH + 2) * UPLOAD_CHUNK;
/// **An upload's chunks, read from `body` as they are sent** (stinky 14): the first names
/// the repository and the artifact (`first`'s fields) and carries the first bytes; every
/// later one carries bytes only, [`UPLOAD_CHUNK`] at a time. At most [`UPLOAD_DEPTH`] chunks
/// wait, so the reader stops while the server is slower: never the whole body in memory. A
/// read error ends the stream short, and the server then has fewer bytes than the file: the
/// caller's answer is the server's.
pub fn upload_chunks(
mut body: impl std::io::Read + Send + 'static,
first: pb::v1::PutArtifactChunk,
) -> impl tonic::codegen::tokio_stream::Stream<Item = pb::v1::PutArtifactChunk> + Send + 'static {
let (tx, rx) = tokio::sync::mpsc::channel::<pb::v1::PutArtifactChunk>(UPLOAD_DEPTH);
tokio::task::spawn_blocking(move || {
let mut head = Some(first);
loop {
let mut data = vec![0u8; UPLOAD_CHUNK];
let mut n = 0;
while n < data.len() {
match body.read(&mut data[n..]) {
Ok(0) => break,
Ok(k) => n += k,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(_) => return,
}
}
data.truncate(n);
let chunk = match head.take() {
Some(f) => pb::v1::PutArtifactChunk { data, ..f },
None if n == 0 => return,
None => pb::v1::PutArtifactChunk {
data,
..Default::default()
},
};
if tx.blocking_send(chunk).is_err() || n < UPLOAD_CHUNK {
return;
}
}
});
tonic::codegen::tokio_stream::wrappers::ReceiverStream::new(rx)
}
/// An import's deadline: the call's own, and a second per MiB on top.
fn import_deadline(bytes: u64) -> std::time::Duration {
crate::conn::CALL_DEADLINE + std::time::Duration::from_secs(bytes / IMPORT_CHUNK as u64)
}
async fn bundle(c: BundleCmd, ch: &Authed) -> Result<String, CliError> {
match c {
BundleCmd::Tool(_) => unreachable!("a local tool, handled before dialling"),
BundleCmd::Export {
out,
bundle_id,
sequence,
supplier,
product,
release_version,
expires_days,
artifacts,
} => {
if out.exists() {
return Err(CliError::InvalidArgs(format!(
"{} already exists; the export writes a fresh candidate directory",
out.display()
)));
}
let req = pb::doors::ExportBundleRequest {
bundle_id,
sequence,
supplier,
product,
release_version,
expires_days,
artifacts,
};
let mut stream = call!(
"bundle export",
client!(pb::doors::bundle_service_client, BundleServiceClient, ch).export(req)
);
// Written as it arrives, checked against the summary's digest (RA2-5).
let mut w = ExportWriter::new(&out)?;
let summary = loop {
match next("bundle export", &mut stream).await? {
Some(c) => {
if let Some(s) = w.take(c)? {
break s;
}
}
None => {
return Err(CliError::Local(
"the export ended without its summary; nothing was kept".into(),
))
}
}
};
let members = w.members();
let mut said = format!(
"EXPORTED {} seq {} -> {}\n verdict: {} (computed from the stored evidence)\n",
summary.bundle_id,
summary.sequence,
out.display(),
summary.verdict
);
for line in &summary.artifacts {
said.push_str(&format!(" {line}\n"));
}
said.push_str(&format!(
" candidate sha256: {} (checked)\n {} member(s)\n",
summary.candidate_sha256, members
));
Ok(said)
}
BundleCmd::AllowFirstImport {
vendor,
product,
sequence,
} => {
let req = pb::doors::AllowFirstImportRequest {
vendor,
product,
sequence,
};
let r = call!(
"bundle allow-first-import",
client!(pb::doors::bundle_service_client, BundleServiceClient, ch)
.allow_first_import(req)
);
Ok(show(&r))
}
BundleCmd::Import { file } => {
let path = existing(&file)?;
let bytes = std::fs::metadata(&path)
.map_err(|e| CliError::Local(format!("reading {}: {e}", path.display())))?
.len();
let mut admin = client!(pb::v1::admin_service_client, AdminServiceClient, ch);
let r = call!(
"bundle import",
admin.import_bundle(import_chunks(path)),
import_deadline(bytes)
);
Ok(show(&r))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_property_match_is_key_equals_value() {
assert_eq!(
property_match("team=core").unwrap(),
pb::v1::PropertyMatch {
key: "team".into(),
value: "core".into()
}
);
assert_eq!(property_match("k=").unwrap().value, "");
assert_eq!(
property_match("novalue").unwrap_err().code().as_deref(),
Some("holger-cli-1")
);
assert_eq!(
property_match("=v").unwrap_err().code().as_deref(),
Some("holger-cli-1")
);
}
#[test]
fn a_streamed_export_lands_member_by_member_and_never_outside() {
let c = |p: &str, d: &[u8]| pb::doors::ExportBundleChunk {
path: p.into(),
data: d.to_vec(),
summary: None,
};
for evil in ["../x", "/etc/passwd", "a//b", "", "./a"] {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("cand");
let mut w = ExportWriter::new(&out).unwrap();
assert!(w.take(c(evil, b"x")).is_err(), "{evil}");
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
0,
"{evil}: nothing is left"
);
}
// No summary (the stream broke): nothing is left either.
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("cand");
let mut w = ExportWriter::new(&out).unwrap();
w.take(c("a", b"1")).unwrap();
drop(w);
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
0,
"an export with no summary leaves nothing"
);
}
/// The import is read in chunks as it is sent: 1 MiB each, never a first-import mark, an
/// empty file one empty chunk.
#[tokio::test]
async fn an_import_is_read_in_chunks_as_it_is_sent() {
use tonic::codegen::tokio_stream::StreamExt as _;
let dir = tempfile::tempdir().unwrap();
let f = dir.path().join("b.znippy");
std::fs::write(&f, vec![7u8; IMPORT_CHUNK * 2 + 5]).unwrap();
let got: Vec<_> = import_chunks(f).collect().await;
assert_eq!(
got.iter()
.map(|c| (c.data.len(), c.first_import))
.collect::<Vec<_>>(),
vec![(IMPORT_CHUNK, false), (IMPORT_CHUNK, false), (5, false)]
);
let empty = dir.path().join("e");
std::fs::write(&empty, b"").unwrap();
let got: Vec<_> = import_chunks(empty).collect().await;
assert_eq!(
got.iter().map(|c| c.data.len()).collect::<Vec<_>>(),
vec![0]
);
}
/// A reader that makes up `len` bytes as they are asked for, and counts them.
struct Counted {
left: u64,
read: std::sync::Arc<std::sync::atomic::AtomicU64>,
}
impl std::io::Read for Counted {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = (buf.len() as u64).min(self.left) as usize;
buf[..n].fill(7);
self.left -= n as u64;
self.read
.fetch_add(n as u64, std::sync::atomic::Ordering::SeqCst);
Ok(n)
}
}
/// ★ **An upload never holds more than [`UPLOAD_MEMORY_BOUND`]** (stinky 14): a body 64
/// times the bound is read only as fast as it is sent. With the sender stopped after the
/// first chunk, what has been read stays under the bound; drained, every byte arrives once,
/// in chunks of [`UPLOAD_CHUNK`], and only the first names the artifact.
#[tokio::test(flavor = "multi_thread")]
async fn an_upload_is_read_only_as_fast_as_it_is_sent() {
use tonic::codegen::tokio_stream::StreamExt as _;
let len = 64 * UPLOAD_MEMORY_BOUND as u64 + 5;
let read = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
let first = pb::v1::PutArtifactChunk {
repository: "files".into(),
id: Some(id("big", "1", "")),
data: Vec::new(),
async_job: false,
};
let mut s = Box::pin(upload_chunks(
Counted {
left: len,
read: read.clone(),
},
first,
));
let head = s.next().await.expect("a first chunk");
assert_eq!(
(
head.repository.as_str(),
head.id.as_ref().map(|i| i.name.as_str()),
head.data.len()
),
("files", Some("big"), UPLOAD_CHUNK)
);
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
let held = read.load(std::sync::atomic::Ordering::SeqCst);
assert!(
held <= UPLOAD_MEMORY_BOUND as u64,
"read {held} bytes ahead of the sender; the bound is {UPLOAD_MEMORY_BOUND}"
);
let (mut total, mut chunks) = (head.data.len() as u64, 1u64);
while let Some(c) = s.next().await {
assert!(
c.repository.is_empty() && c.id.is_none(),
"only the first chunk names the artifact"
);
assert!(c.data.len() <= UPLOAD_CHUNK);
total += c.data.len() as u64;
chunks += 1;
}
assert_eq!(total, len, "every byte, once");
assert_eq!(chunks, len.div_ceil(UPLOAD_CHUNK as u64));
// An empty file is one chunk that names the artifact and carries nothing.
let first = pb::v1::PutArtifactChunk {
repository: "files".into(),
id: Some(id("e", "", "")),
data: Vec::new(),
async_job: true,
};
let got: Vec<_> = upload_chunks(std::io::empty(), first).collect().await;
assert_eq!(
got.iter()
.map(|c| (c.data.len(), c.async_job, c.repository.as_str()))
.collect::<Vec<_>>(),
vec![(0, true, "files")]
);
}
#[test]
fn an_answer_shows_every_field() {
let s = show(&pb::v1::HealthResponse {
status: "ok".into(),
version: "2.0.0".into(),
uptime_seconds: 3,
});
assert!(
s.contains("status: \"ok\"") && s.contains("uptime_seconds: 3"),
"{s}"
);
}
fn core_and_payload() -> (Vec<pb::doors::ExportBundleChunk>, String) {
let core = br#"{"bundle_id":"b"}"#.to_vec();
let digest = sha256_hex(&core);
let c = |p: &str, d: &[u8]| pb::doors::ExportBundleChunk {
path: p.into(),
data: d.to_vec(),
summary: None,
};
(
vec![
c("core.json", &core[..5]),
c("core.json", &core[5..]),
c("payload/a.bin", b"12"),
c("payload/a.bin", b"3"),
],
digest,
)
}
fn summary(digest: &str, members: u32) -> pb::doors::ExportBundleChunk {
pb::doors::ExportBundleChunk {
summary: Some(pb::doors::ExportBundleSummary {
bundle_id: "b".into(),
candidate_sha256: digest.into(),
members,
..Default::default()
}),
..Default::default()
}
}
/// ★ RA2-5: **an export goes to disk as it arrives, never whole in memory, and only a
/// candidate whose core.json hashes to the summary's candidate_sha256 is kept.** Before the
/// summary, the members are on disk in a fresh directory beside `out` (not in `out`).
#[test]
fn an_export_is_streamed_to_disk_and_its_digest_is_checked() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("cand");
let (chunks, digest) = core_and_payload();
let mut w = ExportWriter::new(&out).unwrap();
for c in chunks {
assert!(w.take(c).unwrap().is_none());
}
let on_disk: u64 = walk_bytes(dir.path());
assert_eq!(
on_disk,
17 + 3,
"the members are on disk before the summary arrives"
);
assert!(
!out.exists(),
"nothing is at `out` before the digest is checked"
);
let s = w
.take(summary(&digest, 2))
.unwrap()
.expect("the summary ends it");
assert_eq!(s.bundle_id, "b");
assert_eq!(std::fs::read(out.join("payload/a.bin")).unwrap(), b"123");
assert_eq!(
std::fs::read(out.join("core.json")).unwrap(),
br#"{"bundle_id":"b"}"#
);
}
#[test]
fn an_export_whose_digest_or_members_do_not_match_is_not_kept() {
for (digest, members, why) in [
("00".repeat(32), 2, "sha256"),
(String::new(), 2, "sha256"),
(core_and_payload().1, 3, "member"),
] {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("cand");
let mut w = ExportWriter::new(&out).unwrap();
for c in core_and_payload().0 {
w.take(c).unwrap();
}
let e = w.take(summary(&digest, members)).unwrap_err();
assert!(
e.to_string().contains(why) && e.to_string().contains("nothing was kept"),
"{e}"
);
assert!(!out.exists(), "a refused candidate is not at `out`");
assert_eq!(
walk_bytes(dir.path()),
0,
"and nothing of it is left on disk"
);
}
}
fn walk_bytes(dir: &Path) -> u64 {
let mut n = 0;
for e in std::fs::read_dir(dir).unwrap().flatten() {
let p = e.path();
n += if p.is_dir() {
walk_bytes(&p)
} else {
std::fs::metadata(&p).unwrap().len()
};
}
n
}
}