use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use clap::{Args, Subcommand};
use net_sdk::bootstrap_credential::{Psk, TrustDomainId};
use net_sdk::enrollment::bundle::{MembershipIssuer, MeshContact};
use net_sdk::enrollment::invite::{
EnrollmentEndpoint, EnrollmentKey, InviteSpec, MembershipInvite, Relation, RelayLocator,
};
use net_sdk::enrollment::policy::{ApprovalMode, InvitationPolicy, DEFAULT_INVITATION_TTL};
use net_sdk::enrollment::redeem::Refusal;
use net_sdk::enrollment::service::{
BundleIssuer, EnrollmentService, ServiceConfig, SessionSink, SharedLedger,
};
use net_sdk::enrollment::store::{
EnrollmentLedger, LedgerError, LedgerLimits, OfferId, OfferState,
};
use net_sdk::identity::{EntityId, Identity};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use super::lifecycle::{control_call, now_unix, state_dir, NODE_SUBDIR};
use crate::error::{connection_failure, generic, invalid_args, CliError};
use crate::prelude::{emit_value, OutputFormat};
const LEDGER_SUBDIR: &str = "ledger";
pub(crate) const DEFAULT_RELAY: Option<&str> = None;
pub(crate) struct EnrollPlan {
public: Option<EnrollmentEndpoint>,
relay: Option<EnrollmentEndpoint>,
issuer_path: Option<PathBuf>,
explicit_ledger: Option<PathBuf>,
default_ledger: PathBuf,
domain_name: String,
port_mapping: bool,
}
impl EnrollPlan {
#[allow(clippy::too_many_arguments)]
pub(crate) fn validate(
public_addr: Option<String>,
relay: Option<String>,
issuer_identity: Option<PathBuf>,
ledger: Option<PathBuf>,
domain_name: Option<String>,
port_mapping: bool,
state: &Path,
profile: &str,
) -> Result<Self, CliError> {
let public = match public_addr {
Some(raw) => Some(
EnrollmentEndpoint::parse(&raw)
.map_err(|e| invalid_args(format!("--public-addr: {e}")))?,
),
None => None,
};
let relay = match relay {
Some(raw) => Some(
EnrollmentEndpoint::parse(&raw)
.map_err(|e| invalid_args(format!("--relay: {e}")))?,
),
None => None,
};
if let Some(ledger) = &ledger {
if !ledger.is_dir() {
return Err(invalid_args(format!(
"no enrollment ledger at {}; run `net-mesh enrollment init` or omit --ledger",
ledger.display()
)));
}
}
let domain_name = domain_name.unwrap_or_else(|| profile.to_string());
let usable = !domain_name.is_empty()
&& domain_name.len() <= 64
&& domain_name
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b));
if !usable {
return Err(invalid_args(
"--domain-name must be 1-64 characters of [A-Za-z0-9._-]",
));
}
Ok(Self {
public,
relay,
issuer_path: issuer_identity,
explicit_ledger: ledger,
default_ledger: state.join(LEDGER_SUBDIR),
domain_name,
port_mapping,
})
}
pub(crate) fn port_mapping(&self) -> bool {
self.port_mapping
}
pub(crate) async fn provision(
self,
bind: SocketAddr,
secrets: &mut super::lifecycle::NodeSecrets,
persist: impl FnOnce(&super::lifecycle::NodeSecrets) -> Result<(), CliError>,
) -> Result<(SocketAddr, EnrollOwner), CliError> {
let mut changed = false;
let mut created = Vec::new();
let port = match (bind.port(), secrets.enroll_port) {
(0, Some(port)) => port,
(0, None) => {
let port = free_port(bind.ip())?;
secrets.enroll_port = Some(port);
changed = true;
created.push("port");
port
}
(port, _) => port,
};
let issuer = match &self.issuer_path {
Some(path) => crate::context::load_operator_identity(path).await?,
None => {
let seed = match secrets.issuer {
Some(seed) => seed,
None => {
let mut seed = [0u8; 32];
getrandom::fill(&mut seed)
.map_err(|_| generic("operating-system CSPRNG unavailable"))?;
secrets.issuer = Some(seed);
changed = true;
created.push("issuer");
seed
}
};
Identity::from_seed(seed)
}
};
if changed {
persist(secrets)?;
}
let ledger_dir = self
.explicit_ledger
.clone()
.unwrap_or_else(|| self.default_ledger.clone());
let create = self.explicit_ledger.is_none() && !ledger_dir.exists();
let (dir, entity) = (ledger_dir.clone(), issuer.entity_id().clone());
let ledger = tokio::task::spawn_blocking(move || {
if create {
EnrollmentLedger::create(&dir, entity, LedgerLimits::default())
} else {
EnrollmentLedger::open(&dir, entity, LedgerLimits::default())
}
})
.await
.map_err(|e| generic(format!("ledger task failed: {e}")))?
.map_err(|e| ledger_open_error(&ledger_dir, e))?;
if create {
created.push("ledger");
}
Ok((
SocketAddr::new(bind.ip(), port),
EnrollOwner {
plan: self,
issuer,
ledger: Arc::new(parking_lot::Mutex::new(ledger)),
bind: SocketAddr::new(bind.ip(), port),
created,
org: Arc::new(OrgBook::new(ledger_dir.with_extension("org"))),
},
))
}
}
fn free_port(ip: std::net::IpAddr) -> Result<u16, CliError> {
for attempt in 0..64 {
let candidate = if attempt == 0 {
0
} else {
let r: [u8; 2] = super::lifecycle::random()?;
49_152 + u16::from_be_bytes(r) % 16_384
};
let Ok(tcp) = std::net::TcpListener::bind((ip, candidate)) else {
continue;
};
let port = tcp
.local_addr()
.map_err(|e| generic(format!("choose enrollment port: {e}")))?
.port();
if std::net::UdpSocket::bind((ip, port)).is_ok() {
return Ok(port);
}
}
Err(generic(
"could not find a port free for both UDP and TCP; pass --bind",
))
}
fn ledger_open_error(dir: &Path, e: LedgerError) -> CliError {
match e {
LedgerError::IssuerMismatch => invalid_args(format!(
"ledger {} belongs to a different issuer",
dir.display()
)),
LedgerError::Storage(
net::adapter::net::behavior::enrollment_storage::StorageError::Busy,
) => generic(format!(
"ledger {} is already owned by another process",
dir.display()
)),
other => generic(format!("ledger {}: {other}", dir.display())),
}
}
pub(crate) struct EnrollOwner {
plan: EnrollPlan,
issuer: Identity,
ledger: SharedLedger,
bind: SocketAddr,
created: Vec<&'static str>,
org: Arc<OrgBook>,
}
impl EnrollOwner {
pub(crate) fn issuer(&self) -> &Identity {
&self.issuer
}
}
const MAPPING_WAIT: Duration = Duration::from_secs(4);
pub(crate) fn select_endpoint(
public: Option<&EnrollmentEndpoint>,
mapped_tcp: Option<SocketAddr>,
mapped_udp: Option<SocketAddr>,
bind: SocketAddr,
) -> Option<EnrollmentEndpoint> {
if let Some(public) = public {
return Some(public.clone());
}
if let (Some(tcp), Some(_udp)) = (mapped_tcp, mapped_udp) {
return EnrollmentEndpoint::parse(&tcp.to_string()).ok();
}
if bind.ip().is_unspecified() {
return None;
}
EnrollmentEndpoint::parse(&bind.to_string()).ok()
}
impl EnrollOwner {
pub(crate) async fn start(
self,
mesh: &net_sdk::Mesh,
psk: Psk,
subnet: Option<net_sdk::enrollment::bundle::SubnetLeafIssuer>,
channels: Vec<ChannelIssuer>,
) -> Result<RunningEnrollment, CliError> {
let (tcp_mapping, udp_mapped) = if self.plan.port_mapping {
let udp = async {
let deadline = tokio::time::Instant::now() + MAPPING_WAIT;
loop {
if let Some(addr) = mesh.traversal_stats().port_mapping_external {
return Some(addr);
}
if tokio::time::Instant::now() >= deadline {
return None;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
};
let tcp = tokio::time::timeout(
MAPPING_WAIT,
net_sdk::enrollment::portmap::TcpMapping::establish(self.bind.port()),
);
let (tcp, udp) = tokio::join!(tcp, udp);
(tcp.ok().flatten(), udp)
} else {
(None, None)
};
let tcp_mapped = tcp_mapping.as_ref().and_then(|m| m.external());
let default_endpoint =
select_endpoint(self.plan.public.as_ref(), tcp_mapped, udp_mapped, self.bind);
let trust_domain = psk.trust_domain();
let bundles = Arc::new(NodeBundles {
issuer: self.issuer.clone(),
psk,
noise_pubkey: *mesh.public_key(),
node_id: mesh.node_id(),
tcp_mapped,
udp_mapped,
contacts: parking_lot::Mutex::new(std::collections::HashMap::new()),
subnet: subnet.clone(),
channels: channels.iter().map(|(_, i)| i.clone()).collect(),
org: self.org.clone(),
});
if let Some(addr) = default_endpoint
.as_ref()
.and_then(|e| bundles.contact_addr(e))
{
let _ = mesh.node().set_direct_hint(Some(addr)).await;
}
let service = EnrollmentService::bind(
self.bind,
&self.issuer,
self.ledger.clone(),
bundles.clone(),
ServiceConfig::default(),
)
.await
.map_err(|e| {
connection_failure(format!("enrollment listener on TCP {}: {e}", self.bind))
})?;
let renewal = match &subnet {
Some(issuer) => Some(
net_sdk::enrollment::renew::serve_subnet_renewal(
mesh.node(),
self.ledger.clone(),
issuer.clone(),
)
.map_err(|e| generic(format!("subnet renewal service: {e}")))?,
),
None => None,
};
let standalone = Some(
net_sdk::enrollment::standalone::serve_standalone_redeem(
mesh.node(),
self.ledger.clone(),
subnet.clone(),
Some(self.org.stash.clone()),
channels.iter().map(|(_, i)| i.clone()).collect(),
)
.map_err(|e| generic(format!("standalone redemption service: {e}")))?,
);
let relay = match self.plan.relay {
Some(endpoint) => {
Some(RelayLink::start(mesh.node().clone(), endpoint, service.session_sink()).await)
}
None => None,
};
let context = Arc::new(EnrollContext {
issuer: self.issuer,
ledger: self.ledger,
key: service.enrollment_key(),
default_endpoint,
relay: relay.as_ref().map(|r| r.locator.clone()),
subnet,
channels,
channel_configs: mesh.node().channel_configs().cloned(),
trust_domain,
domain_name: self.plan.domain_name,
bundles,
org: self.org,
});
Ok(RunningEnrollment {
service,
context,
tcp_mapping,
relay,
_renewal: renewal,
_standalone: standalone,
port_mapping: self.plan.port_mapping,
created: self.created,
})
}
}
const RELAY_REGISTER_WAIT: Duration = Duration::from_secs(6);
const RELAY_RETRY: Duration = Duration::from_secs(15);
const RELAY_SPLICE_BACKLOG: usize = 16;
struct RelayLink {
locator: RelayLocator,
state: Arc<parking_lot::Mutex<RelayLinkState>>,
node: Arc<net::adapter::net::MeshNode>,
task: tokio::task::JoinHandle<()>,
}
#[derive(Default)]
struct RelayLinkState {
registered: Option<SocketAddr>,
error: Option<String>,
}
impl RelayLink {
async fn start(
node: Arc<net::adapter::net::MeshNode>,
endpoint: EnrollmentEndpoint,
sink: SessionSink,
) -> Self {
let locator = RelayLocator {
endpoint: endpoint.clone(),
registration: net::adapter::net::traversal::blind_relay::registration_id(
node.entity_id(),
),
};
let state = Arc::new(parking_lot::Mutex::new(RelayLinkState::default()));
let (first_tx, first) = tokio::sync::oneshot::channel();
let task = tokio::spawn(relay_loop(
node.clone(),
endpoint,
sink,
state.clone(),
first_tx,
));
let _ = tokio::time::timeout(RELAY_REGISTER_WAIT, first).await;
Self {
locator,
state,
node,
task,
}
}
}
impl Drop for RelayLink {
fn drop(&mut self) {
self.task.abort();
}
}
async fn relay_loop(
node: Arc<net::adapter::net::MeshNode>,
endpoint: EnrollmentEndpoint,
sink: SessionSink,
state: Arc<parking_lot::Mutex<RelayLinkState>>,
first: tokio::sync::oneshot::Sender<()>,
) {
let mut first = Some(first);
loop {
let attempt = async {
let addr = tokio::net::lookup_host(endpoint.as_str())
.await
.map_err(|e| format!("resolve {}: {e}", endpoint.as_str()))?
.next()
.ok_or_else(|| format!("{} does not resolve", endpoint.as_str()))?;
node.relay_register(addr).await.map_err(|e| e.to_string())
};
match attempt.await {
Ok(mut registration) => {
*state.lock() = RelayLinkState {
registered: Some(registration.relay()),
error: None,
};
if let Some(first) = first.take() {
let _ = first.send(());
}
let mut streams = registration.accept_splices(RELAY_SPLICE_BACKLOG);
while let Some(stream) = streams.recv().await {
sink.serve(stream);
}
return;
}
Err(e) => {
state.lock().error = Some(e);
if let Some(first) = first.take() {
let _ = first.send(());
}
tokio::time::sleep(RELAY_RETRY).await;
}
}
}
}
const CHANNEL_DELEGATION: &str = "none: the device's credential cannot delegate further";
fn channel_json(o: &net_sdk::enrollment::invite::ChannelOffer) -> Value {
json!({
"channel": o.channel.as_str(),
"canonical_hash": format!("{:#018x}", o.channel.hash()),
"root": hex::encode(o.root.as_bytes()),
"rights": super::channel::format_channel_rights(o.rights),
})
}
struct NodeBundles {
issuer: Identity,
psk: Psk,
noise_pubkey: [u8; 32],
node_id: u64,
tcp_mapped: Option<SocketAddr>,
udp_mapped: Option<SocketAddr>,
contacts: parking_lot::Mutex<std::collections::HashMap<String, SocketAddr>>,
subnet: Option<net_sdk::enrollment::bundle::SubnetLeafIssuer>,
channels: Vec<net_sdk::channel_issuer::ChannelLeafIssuer>,
org: Arc<OrgBook>,
}
pub(crate) struct OrgBook {
stash: Arc<net_sdk::enrollment::org::OrgCertStash>,
offers: PathBuf,
relations: PathBuf,
}
impl OrgBook {
fn new(dir: PathBuf) -> Self {
Self {
stash: Arc::new(net_sdk::enrollment::org::OrgCertStash::new(
dir.join("certs"),
)),
offers: dir.join("offers"),
relations: dir.join("relations"),
}
}
fn record_relations(
&self,
offer: &net_sdk::enrollment::store::OfferId,
record: &Value,
) -> std::io::Result<()> {
std::fs::create_dir_all(&self.relations)?;
let path = self.relations.join(format!("{offer}.json"));
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, record.to_string())?;
std::fs::rename(&tmp, &path)
}
fn relations_of(&self, offer: &net_sdk::enrollment::store::OfferId) -> Option<Value> {
let bytes = std::fs::read(self.relations.join(format!("{offer}.json"))).ok()?;
serde_json::from_slice(&bytes).ok()
}
fn record_offer(
&self,
offer: &net_sdk::enrollment::store::OfferId,
org: &net::adapter::net::behavior::org::OrgId,
) -> std::io::Result<()> {
std::fs::create_dir_all(&self.offers)?;
let path = self.offers.join(offer.to_string());
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, hex::encode(org.0))?;
std::fs::rename(&tmp, &path)
}
fn offered_org(
&self,
offer: &net_sdk::enrollment::store::OfferId,
) -> Option<net::adapter::net::behavior::org::OrgId> {
let text = std::fs::read_to_string(self.offers.join(offer.to_string())).ok()?;
let bytes: [u8; 32] = hex::decode(text.trim()).ok()?.try_into().ok()?;
Some(net::adapter::net::behavior::org::OrgId(bytes))
}
}
impl NodeBundles {
fn contact_addr(&self, endpoint: &EnrollmentEndpoint) -> Option<SocketAddr> {
if let Some(addr) = self.contacts.lock().get(endpoint.as_str()) {
return Some(*addr);
}
let addr = match (self.tcp_mapped, self.udp_mapped) {
(Some(tcp), Some(udp)) if tcp.to_string() == endpoint.as_str() => udp,
_ => {
use std::net::ToSocketAddrs as _;
endpoint.as_str().to_socket_addrs().ok()?.next()?
}
};
self.contacts
.lock()
.insert(endpoint.as_str().to_string(), addr);
Some(addr)
}
fn contact_for(&self, invite: &MembershipInvite) -> Option<MeshContact> {
let addr = match invite.endpoint() {
Some(endpoint) => Some(self.contact_addr(endpoint)?),
None => None,
};
let relay = invite.relay().cloned();
if addr.is_none() && relay.is_none() {
return None;
}
Some(MeshContact {
addr,
noise_pubkey: self.noise_pubkey,
node_id: self.node_id,
relay,
})
}
fn issuer_for(&self, contact: MeshContact) -> MembershipIssuer {
let issuer = MembershipIssuer::new(self.issuer.clone(), self.psk.clone(), contact)
.with_org_certs(self.org.stash.clone())
.with_channel_issuers(self.channels.clone());
match &self.subnet {
Some(subnet) => issuer.with_subnet_issuer(subnet.clone()),
None => issuer,
}
}
}
impl BundleIssuer for NodeBundles {
fn issue(
&self,
invite: &MembershipInvite,
intent: &net_sdk::enrollment::invite::RedemptionIntent,
) -> Result<Vec<u8>, Refusal> {
let contact = self.contact_for(invite).ok_or(Refusal::Unavailable)?;
self.issuer_for(contact).issue(invite, intent)
}
fn may_recover(
&self,
invite: &MembershipInvite,
intent: &net_sdk::enrollment::invite::RedemptionIntent,
) -> Result<(), Refusal> {
let unspecified = MeshContact {
addr: Some(SocketAddr::from(([0, 0, 0, 0], 0))),
noise_pubkey: self.noise_pubkey,
node_id: self.node_id,
relay: None,
};
self.issuer_for(unspecified).may_recover(invite, intent)
}
}
pub(crate) struct RunningEnrollment {
service: EnrollmentService,
context: Arc<EnrollContext>,
tcp_mapping: Option<net_sdk::enrollment::portmap::TcpMapping>,
relay: Option<RelayLink>,
_renewal: Option<net::adapter::net::mesh_rpc::ServeHandle>,
_standalone: Option<net::adapter::net::mesh_rpc::ServeHandle>,
port_mapping: bool,
created: Vec<&'static str>,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub(crate) struct EnrollmentReport {
endpoint: Option<String>,
listen: String,
port_mapping: String,
mapped_tcp: Option<String>,
mapped_udp: Option<String>,
enrollment_key: String,
issuer: String,
issuer_fingerprint: String,
domain_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
relay: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
relay_state: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
relay_error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
relay_transport: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
created: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
subnet: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
channels: Vec<serde_json::Value>,
}
impl RunningEnrollment {
pub(crate) fn report(&self) -> EnrollmentReport {
let c = &self.context;
let b = &c.bundles;
let port_mapping = match (self.port_mapping, b.tcp_mapped, b.udp_mapped) {
(false, _, _) => "disabled",
(true, Some(_), Some(_)) => "active",
(true, None, None) => "unavailable",
(true, _, _) => "partial",
};
EnrollmentReport {
endpoint: c.default_endpoint.as_ref().map(|e| e.as_str().to_string()),
listen: self.service.local_addr().to_string(),
port_mapping: port_mapping.to_string(),
mapped_tcp: b.tcp_mapped.map(|a| a.to_string()),
mapped_udp: b.udp_mapped.map(|a| a.to_string()),
enrollment_key: hex::encode(c.key.0),
issuer: hex::encode(c.issuer.entity_id().as_bytes()),
issuer_fingerprint: net_sdk::enrollment::fingerprint(c.issuer.entity_id()),
domain_name: c.domain_name.clone(),
relay: self
.relay
.as_ref()
.map(|r| r.locator.endpoint.as_str().to_string()),
relay_state: self.relay.as_ref().map(|r| {
if r.state.lock().registered.is_some() {
"registered".to_string()
} else {
"unavailable".to_string()
}
}),
relay_error: self
.relay
.as_ref()
.and_then(|r| r.state.lock().error.clone()),
relay_transport: self.relay.as_ref().and_then(|r| {
let registered = r.state.lock().registered?;
Some(
if r.node.relay_tunneled(registered) {
"tcp"
} else {
"udp"
}
.to_string(),
)
}),
created: self.created.iter().map(|s| s.to_string()).collect(),
subnet: c.subnet.as_ref().map(|s| {
let g = s.grant();
serde_json::json!({
"authority": hex::encode(g.authority.as_bytes()),
"issuer_scope": super::subnet::format_subnet(g.scope),
"max_rights": super::subnet::format_subnet_rights(g.maximum_rights),
"topology_epoch": g.topology_epoch,
"verifier": true,
})
}),
channels: c
.channels
.iter()
.map(|(name, issuer)| {
serde_json::json!({
"channel": name.as_str(),
"root": hex::encode(issuer.root().as_bytes()),
"rights": super::channel::format_channel_rights(issuer.grantable()),
"not_after": issuer.not_after(),
})
})
.collect(),
}
}
pub(crate) fn context(&self) -> Arc<EnrollContext> {
self.context.clone()
}
pub(crate) async fn shutdown(self) {
drop(self.relay);
self.service.shutdown().await;
if let Some(mapping) = self.tcp_mapping {
mapping.shutdown().await;
}
}
}
pub(crate) type ChannelIssuer = (
net::adapter::net::channel::ChannelName,
net_sdk::channel_issuer::ChannelLeafIssuer,
);
pub(crate) struct EnrollContext {
issuer: Identity,
ledger: SharedLedger,
key: EnrollmentKey,
default_endpoint: Option<EnrollmentEndpoint>,
relay: Option<RelayLocator>,
subnet: Option<net_sdk::enrollment::bundle::SubnetLeafIssuer>,
channels: Vec<ChannelIssuer>,
channel_configs: Option<Arc<net::adapter::net::channel::ChannelConfigRegistry>>,
trust_domain: TrustDomainId,
domain_name: String,
bundles: Arc<NodeBundles>,
org: Arc<OrgBook>,
}
impl EnrollContext {
pub(crate) fn handle(&self, op: &str, request: &Value) -> Value {
let result = match op {
"invite_create" => self.create(request),
"invite_status" => self.status(request),
"invite_revoke" => self.revoke(request),
"invite_approve" => self.decide(request, true),
"invite_deny" => self.decide(request, false),
"invite_org_pending" => self.org_pending(request),
"invite_org_approve" => self.org_approve(request),
_ => Err("unknown invite operation".to_string()),
};
result.unwrap_or_else(|e| json!({ "error": e }))
}
fn create(&self, request: &Value) -> Result<Value, String> {
let now = now_unix();
let ttl = match request["ttl_secs"].as_u64() {
Some(secs) => Duration::from_secs(secs),
None => DEFAULT_INVITATION_TTL,
};
let mode = if request["require_approval"] == Value::Bool(true) {
ApprovalMode::RequireApproval
} else {
ApprovalMode::Preauthorized
};
let intended = match request["subject"].as_str() {
Some(hex) => Some(parse_entity(hex)?),
None => None,
};
let endpoint = match request["addr"].as_str() {
Some(addr) => {
Some(EnrollmentEndpoint::parse(addr).map_err(|e| format!("--addr: {e}"))?)
}
None => self.default_endpoint.clone(),
};
if endpoint.is_none() && self.relay.is_none() {
return Err(
"no direct address is known for this node (no --public-addr, no router \
mapping, wildcard bind) and no relay is set; pass --addr <host:port> reachable \
by the joiner, or run `up --enroll --relay <host:port>`"
.to_string(),
);
}
if let Some(endpoint) = &endpoint {
self.bundles
.contact_addr(endpoint)
.ok_or_else(|| format!("address {} does not resolve", endpoint.as_str()))?;
}
let standalone = request["standalone"] == Value::Bool(true);
let org = match request["org"].as_str() {
Some(hex) => Some(net_sdk::enrollment::invite::OrgOffer {
org: parse_org_id(hex)?,
}),
None => None,
};
let standalone_relations = usize::from(org.is_some())
+ usize::from(request["subnet"].is_string())
+ usize::from(request["channel"].is_string());
if standalone && standalone_relations > 1 {
return Err(
"a standalone link carries one relation: a subnet, an org or a channel".to_string(),
);
}
let mode = if org.is_some() {
ApprovalMode::RequireApproval
} else {
mode
};
let (mut relations, subnet) = match request["subnet"].as_str() {
None if standalone && (org.is_some() || request["channel"].is_string()) => {
(Vec::new(), None)
}
None if standalone => {
return Err("a standalone link needs a subnet, an org or a channel".to_string())
}
None => (vec![Relation::Mesh], None),
Some(path) => {
let issuer = self.subnet.as_ref().ok_or_else(|| {
"this node has no subnet issuer; start it with --subnet-issuer-grant and \
--subnet-issuer-key"
.to_string()
})?;
let path = super::subnet::parse_subnet_path(path).map_err(|e| e.to_string())?;
let rights = super::subnet::parse_subnet_rights(
request["subnet_rights"].as_str().unwrap_or("attach"),
)
.map_err(|e| e.to_string())?;
let grant = issuer.grant();
let offer = net_sdk::enrollment::invite::SubnetOffer {
scope: net::adapter::net::subnet::SubnetRef {
authority: grant.authority.clone(),
path,
},
topology_epoch: grant.topology_epoch,
rights,
};
if !issuer.covers(&offer) {
return Err(format!(
"subnet {} with {} is outside this node's issuer grant ({} with at most {})",
super::subnet::format_subnet(path),
super::subnet::format_subnet_rights(rights),
super::subnet::format_subnet(grant.scope),
super::subnet::format_subnet_rights(grant.maximum_rights),
));
}
if standalone {
(vec![Relation::Subnet], Some(offer))
} else {
(vec![Relation::Mesh, Relation::Subnet], Some(offer))
}
}
};
if org.is_some() {
relations.push(Relation::Org);
}
for unsupported in ["channel_ttl", "channel_depth", "channel_publisher"] {
if !request[unsupported].is_null() {
return Err(format!(
"{unsupported} is not supported: a channel credential lives exactly as long as this node's grant, cannot delegate, and names this node as publisher"
));
}
}
let channel_expiry = request["channel"]
.as_str()
.and_then(|n| net::adapter::net::channel::ChannelName::new(n).ok())
.and_then(|n| {
self.channels
.iter()
.find(|(c, _)| c == &n)
.map(|(_, i)| i.not_after())
});
let channel = match request["channel"].as_str() {
None => None,
Some(name) => {
let offer = self.channel_offer(name, request["channel_rights"].as_str())?;
relations.push(Relation::Channel);
Some(offer)
}
};
let policy = InvitationPolicy::with_options(now, ttl, mode).map_err(|e| e.to_string())?;
let invite = MembershipInvite::sign(
&self.issuer,
InviteSpec {
trust_domain_name: self.domain_name.clone(),
trust_domain: self.trust_domain,
endpoint: endpoint.clone(),
relay: self.relay.clone(),
enrollment_key: self.key,
subnet: subnet.clone(),
org,
channel: channel.clone(),
relations,
intended_subject: intended,
policy,
},
)
.map_err(|e| e.to_string())?;
let offer = self
.ledger
.lock()
.offer(invite.offer_spec(), now)
.map_err(|e| e.to_string())?;
if let Some(offer_org) = &org {
self.org
.record_offer(&offer, &offer_org.org)
.map_err(|e| format!("org offer record: {e}"))?;
}
self.org
.record_relations(
&offer,
&json!({
"relations": invite.relations().iter().map(|r| format!("{r:?}").to_lowercase()).collect::<Vec<_>>(),
"org": org.as_ref().map(|o| hex::encode(o.org.0)),
"subnet": subnet.as_ref().map(|o| json!({
"authority": hex::encode(o.scope.authority.as_bytes()),
"scope": super::subnet::format_subnet(o.scope.path),
"rights": super::subnet::format_subnet_rights(o.rights),
"topology_epoch": o.topology_epoch,
})),
"standalone": standalone,
"channel": channel.as_ref().map(channel_json),
}),
)
.map_err(|e| format!("offer relation record: {e}"))?;
Ok(json!({
"token": invite.encode(),
"offer_id": offer.to_string(),
"expires_at": policy.expires_at(),
"approval": approval_name(mode),
"bearer": invite.is_bearer(),
"endpoint": endpoint.as_ref().map(|e| e.as_str()),
"relay": self.relay.as_ref().map(|r| r.endpoint.as_str()),
"subnet": subnet.as_ref().map(|o| json!({
"scope": super::subnet::format_subnet(o.scope.path),
"rights": super::subnet::format_subnet_rights(o.rights),
})),
"standalone": standalone,
"org": org.as_ref().map(|o| hex::encode(o.org.0)),
"channel": channel.as_ref().map(|o| {
let mut v = channel_json(o);
v["credential_expires_at"] = json!(channel_expiry);
v["delegation"] = json!(CHANNEL_DELEGATION);
v["publisher"] = json!(if o.rights.contains(net::adapter::net::identity::TokenScope::SUBSCRIBE) {
"this node (the issuer)"
} else {
"none: publish is local to the device"
});
v
}),
"issuer_fingerprint": invite.issuer_fingerprint(),
}))
}
fn channel_offer(
&self,
name: &str,
rights: Option<&str>,
) -> Result<net_sdk::enrollment::invite::ChannelOffer, String> {
let channel = net::adapter::net::channel::ChannelName::new(name)
.map_err(|e| format!("channel `{name}`: {e}"))?;
let rights = super::channel::parse_channel_rights(
rights.ok_or("--channel needs --channel-rights (publish and/or subscribe)")?,
)
.map_err(|e| e.to_string())?;
let (_, issuer) = self
.channels
.iter()
.find(|(n, _)| n == &channel)
.ok_or_else(|| {
format!(
"this node has no grant for channel {}; start it with --channel-grant \
(from `channel issue-grant`)",
channel.as_str()
)
})?;
let offer = net_sdk::enrollment::invite::ChannelOffer {
channel: channel.clone(),
root: issuer.root().clone(),
rights,
};
if !issuer.covers(&offer) {
return Err(format!(
"{} on {} is outside this node's grant (at most {})",
super::channel::format_channel_rights(rights),
channel.as_str(),
super::channel::format_channel_rights(issuer.grantable()),
));
}
if rights.contains(net::adapter::net::identity::TokenScope::SUBSCRIBE) {
let served = self.channel_configs.as_ref().is_some_and(|registry| {
registry
.get_by_name(channel.as_str())
.is_some_and(|config| {
config.channel_id.name() == &channel
&& config.token_roots.contains(issuer.root())
})
});
if !served {
return Err(format!(
"a subscribe link needs {} served here trusting root {}; run \
`net-mesh channel serve {} --token-root {}` first",
channel.as_str(),
hex::encode(issuer.root().as_bytes()),
channel.as_str(),
hex::encode(issuer.root().as_bytes()),
));
}
}
Ok(offer)
}
fn status(&self, request: &Value) -> Result<Value, String> {
let ledger = self.ledger.lock();
let all = ledger.statuses().map_err(|e| e.to_string())?;
let wanted = match request["offer_id"].as_str() {
Some(id) => Some(parse_offer(id)?),
None => None,
};
let rows: Vec<Value> = all
.iter()
.filter(|s| wanted.is_none_or(|w| s.offer_id == w))
.map(|s| {
let (state, subject, receipt) = match &s.state {
OfferState::Offered => ("offered", None, None),
OfferState::PendingApproval { subject } => ("pending_approval", Some(subject), None),
OfferState::Ready { subject } => ("ready", Some(subject), None),
OfferState::Issued {
subject,
receipt_id,
..
} => ("issued", Some(subject), Some(receipt_id.to_string())),
OfferState::Revoked { subject } => ("revoked", subject.as_ref(), None),
OfferState::Denied { subject } => ("denied", Some(subject), None),
};
json!({
"offer_id": s.offer_id.to_string(),
"state": state,
"subject": subject.map(|e| hex::encode(e.as_bytes())),
"receipt_id": receipt,
"approval": approval_name(s.approval),
"expires_at": s.expires_at,
"intended_subject": s.intended_subject.as_ref().map(|e| hex::encode(e.as_bytes())),
})
})
.collect();
if wanted.is_some() && rows.is_empty() {
return Err("unknown offer".to_string());
}
Ok(json!({ "offers": rows }))
}
fn revoke(&self, request: &Value) -> Result<Value, String> {
let offer = parse_offer(request["offer_id"].as_str().unwrap_or_default())?;
self.ledger
.lock()
.revoke(&offer, now_unix())
.map_err(|e| e.to_string())?;
Ok(json!({ "offer_id": offer.to_string(), "state": "revoked" }))
}
fn decide(&self, request: &Value, approve: bool) -> Result<Value, String> {
let offer = parse_offer(request["offer_id"].as_str().unwrap_or_default())?;
if approve && self.org.offered_org(&offer).is_some() {
return Err(
"this invite offers an org membership, which only the org root can sign: \
approve it with `net-mesh org approve <offer> --subject <entity> --org-key <file>`"
.to_string(),
);
}
let subject = parse_entity(request["subject"].as_str().unwrap_or_default())?;
let mut ledger = self.ledger.lock();
let claim = ledger
.pending_claim(&offer)
.map_err(|e| e.to_string())?
.ok_or_else(|| "offer has no claim awaiting approval".to_string())?;
if claim.subject != subject {
return Err("the pending claim is from a different subject".to_string());
}
let now = now_unix();
if approve {
ledger.approve(&offer, &claim, now)
} else {
ledger.deny(&offer, &claim, now)
}
.map_err(|e| e.to_string())?;
Ok(json!({
"offer_id": offer.to_string(),
"state": if approve { "approved" } else { "denied" },
}))
}
}
impl EnrollContext {
fn org_pending(&self, request: &Value) -> Result<Value, String> {
let offer = parse_offer(request["offer_id"].as_str().unwrap_or_default())?;
let org = self
.org
.offered_org(&offer)
.ok_or_else(|| "that offer is not an org invite".to_string())?;
let claim = self
.ledger
.lock()
.pending_claim(&offer)
.map_err(|e| e.to_string())?
.ok_or_else(|| "offer has no claim awaiting approval".to_string())?;
Ok(json!({
"offer_id": offer.to_string(),
"org": hex::encode(org.0),
"subject": hex::encode(claim.subject.as_bytes()),
}))
}
fn org_approve(&self, request: &Value) -> Result<Value, String> {
let offer = parse_offer(request["offer_id"].as_str().unwrap_or_default())?;
let subject = parse_entity(request["subject"].as_str().unwrap_or_default())?;
let cert = hex::decode(request["cert"].as_str().unwrap_or_default())
.ok()
.and_then(|b| net::adapter::net::behavior::org::OrgMembershipCert::from_bytes(&b).ok())
.ok_or_else(|| "malformed membership certificate".to_string())?;
let org = self
.org
.offered_org(&offer)
.ok_or_else(|| "that offer is not an org invite".to_string())?;
if cert.org_id != org {
return Err(
"the certificate is for a different org than the invite offers".to_string(),
);
}
let mut ledger = self.ledger.lock();
let claim = ledger
.pending_claim(&offer)
.map_err(|e| e.to_string())?
.ok_or_else(|| "offer has no claim awaiting approval".to_string())?;
if claim.subject != subject || cert.member != subject {
return Err("the pending claim is from a different subject".to_string());
}
let audience = match request["audience"].as_str() {
Some(hex_a) => {
let bytes = hex::decode(hex_a).map_err(|_| "malformed org audience".to_string())?;
Some(
net::adapter::net::behavior::org_authority::OwnerAudienceCredential::decode_config(&bytes)
.map_err(|e| format!("org audience: {e}"))?,
)
}
None => None,
};
self.org
.stash
.put(&claim, &cert)
.map_err(|e| format!("keeping the certificate: {e}"))?;
if let Some(audience) = &audience {
self.org
.stash
.put_audience(&claim, &org, audience)
.map_err(|e| format!("keeping the org audience: {e}"))?;
}
ledger
.approve(&offer, &claim, now_unix())
.map_err(|e| e.to_string())?;
if let Some(mut record) = self.org.relations_of(&offer) {
record["approved_generation"] = json!(cert.generation);
let _ = self.org.record_relations(&offer, &record);
}
Ok(json!({
"offer_id": offer.to_string(),
"state": "approved",
"org": hex::encode(org.0),
"member": hex::encode(subject.as_bytes()),
"generation": cert.generation,
"not_after": cert.not_after,
"audience": audience.is_some(),
}))
}
}
impl EnrollContext {
pub(crate) fn issued_inventory(&self, kind: &str, wanted: &str) -> Result<Value, String> {
let statuses = self.ledger.lock().statuses().map_err(|e| e.to_string())?;
let mut issued = Vec::new();
let mut unrecorded = 0usize;
for status in statuses {
let Some(record) = self.org.relations_of(&status.offer_id) else {
unrecorded += 1;
continue;
};
let matches = match kind {
"org" => record["org"].as_str() == Some(wanted),
_ => record["subnet"]["scope"].as_str().is_some_and(|scope| {
scope == wanted || scope.starts_with(&format!("{wanted}."))
}),
};
if !matches {
continue;
}
use net_sdk::enrollment::store::OfferState as S;
let (state, subject, issued_at) = match &status.state {
S::Offered => ("offered", None, None),
S::PendingApproval { subject } => ("pending_approval", Some(subject), None),
S::Ready { subject } => ("approved", Some(subject), None),
S::Issued {
subject, issued_at, ..
} => ("issued", Some(subject), Some(*issued_at)),
S::Revoked { subject } => ("revoked_offer", subject.as_ref(), None),
S::Denied { subject } => ("denied", Some(subject), None),
};
issued.push(json!({
"offer_id": status.offer_id.to_string(),
"state": state,
"subject": subject.map(|s| hex::encode(s.as_bytes())),
"issued_at": issued_at,
"relations": record["relations"],
"subnet": record["subnet"],
"approved_generation": record["approved_generation"],
"standalone": record["standalone"],
}));
}
Ok(json!({ "issued": issued, "unrecorded_offers": unrecorded }))
}
}
pub(crate) fn parse_org_id(
hex_id: &str,
) -> Result<net::adapter::net::behavior::org::OrgId, String> {
let bytes: [u8; 32] = hex::decode(hex_id.trim().trim_start_matches("0x"))
.ok()
.and_then(|b| b.try_into().ok())
.ok_or_else(|| "an org id is 64 hex characters".to_string())?;
Ok(net::adapter::net::behavior::org::OrgId(bytes))
}
fn approval_name(mode: ApprovalMode) -> &'static str {
match mode {
ApprovalMode::Preauthorized => "preauthorized",
ApprovalMode::RequireApproval => "require_approval",
}
}
fn parse_entity(hex_id: &str) -> Result<EntityId, String> {
let bytes = hex::decode(hex_id.trim_start_matches("0x"))
.ok()
.and_then(|b| <[u8; 32]>::try_from(b).ok())
.ok_or_else(|| "subject must be a full 64-hex entity id".to_string())?;
Ok(EntityId::from_bytes(bytes))
}
fn parse_offer(hex_id: &str) -> Result<OfferId, String> {
let bytes = hex::decode(hex_id)
.ok()
.and_then(|b| <[u8; 16]>::try_from(b).ok())
.ok_or_else(|| "offer id must be 32 hex characters".to_string())?;
Ok(OfferId::from_bytes(bytes))
}
#[derive(Subcommand, Debug)]
pub enum EnrollmentCommand {
Init(InitArgs),
}
#[derive(Args, Debug)]
pub struct InitArgs {
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
pub issuer_identity: PathBuf,
#[arg(long, value_name = "DIR")]
pub ledger: Option<PathBuf>,
}
pub async fn run_enrollment(
cmd: EnrollmentCommand,
output: Option<OutputFormat>,
profile_name: &str,
) -> Result<(), CliError> {
match cmd {
EnrollmentCommand::Init(args) => {
let state = state_dir(args.state_dir, profile_name)?;
let issuer = crate::context::load_operator_identity(&args.issuer_identity).await?;
let ledger = args.ledger.unwrap_or_else(|| state.join(LEDGER_SUBDIR));
if let Some(parent) = ledger.parent() {
std::fs::create_dir_all(parent)
.map_err(|e| generic(format!("create {}: {e}", parent.display())))?;
}
let (dir, entity) = (ledger.clone(), issuer.entity_id().clone());
tokio::task::spawn_blocking(move || {
EnrollmentLedger::create(&dir, entity, LedgerLimits::default()).map(drop)
})
.await
.map_err(|e| generic(format!("ledger task failed: {e}")))?
.map_err(|e| match e {
LedgerError::Storage(
net::adapter::net::behavior::enrollment_storage::StorageError::AlreadyExists,
) => invalid_args(format!("ledger {} already exists", ledger.display())),
other => generic(format!("ledger {}: {other}", ledger.display())),
})?;
emit_value(
OutputFormat::resolve_oneshot(output),
&json!({
"ledger": ledger.display().to_string(),
"issuer": hex::encode(issuer.entity_id().as_bytes()),
"issuer_fingerprint": net_sdk::enrollment::fingerprint(issuer.entity_id()),
}),
)
.map_err(|e| generic(format!("write result: {e}")))
}
}
}
#[derive(Subcommand, Debug)]
pub enum InviteCommand {
Create(CreateArgs),
Inspect(InspectArgs),
Status(OfferQueryArgs),
Revoke(OfferArgs),
Approve(DecideArgs),
Deny(DecideArgs),
}
#[derive(Args, Debug)]
pub struct CreateArgs {
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
#[arg(long, value_name = "DURATION", value_parser = crate::humantime::parse_duration)]
pub ttl: Option<Duration>,
#[arg(long)]
pub require_approval: bool,
#[arg(long = "for", value_name = "ENTITY")]
pub for_subject: Option<String>,
#[arg(long, value_name = "PATH")]
pub out: Option<PathBuf>,
#[arg(long, value_name = "HOST:PORT")]
pub addr: Option<String>,
#[arg(long, value_name = "PATH")]
pub subnet: Option<String>,
#[arg(long, value_name = "RIGHTS", requires = "subnet")]
pub subnet_rights: Option<String>,
#[arg(long, value_name = "ORG")]
pub org: Option<String>,
#[arg(long, value_name = "NAME", requires = "channel_rights")]
pub channel: Option<String>,
#[arg(long, value_name = "RIGHTS", requires = "channel")]
pub channel_rights: Option<String>,
#[arg(skip)]
pub standalone: bool,
}
#[derive(Args, Debug)]
pub struct InspectArgs {
pub token: String,
}
#[derive(Args, Debug)]
pub struct OfferQueryArgs {
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
pub offer_id: Option<String>,
}
#[derive(Args, Debug)]
pub struct OfferArgs {
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
pub offer_id: String,
}
#[derive(Args, Debug)]
pub struct DecideArgs {
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
pub offer_id: String,
#[arg(long, value_name = "ENTITY")]
pub subject: String,
}
pub(crate) async fn node_request(
state_dir_arg: Option<PathBuf>,
profile: &str,
request: Value,
) -> Result<Value, CliError> {
let dir = state_dir(state_dir_arg, profile)?.join(NODE_SUBDIR);
let (_, reply) = control_call(&dir, request)
.await
.map_err(|e| connection_failure(format!("{e}; is `net-mesh up --enroll` running?")))?;
if let Some(err) = reply["error"].as_str() {
return Err(generic(err.to_string()));
}
Ok(reply)
}
pub async fn run_invite(
cmd: InviteCommand,
output: Option<OutputFormat>,
profile_name: &str,
) -> Result<(), CliError> {
let fmt = OutputFormat::resolve_oneshot(output);
let emit = |v: &Value| emit_value(fmt, v).map_err(|e| generic(format!("write result: {e}")));
match cmd {
InviteCommand::Create(args) => {
if let Some(subject) = &args.for_subject {
parse_entity(subject).map_err(invalid_args)?;
}
if let Some(out) = &args.out {
if out.exists() {
return Err(invalid_args(format!("{} already exists", out.display())));
}
}
let mut request = json!({
"op": "invite_create",
"require_approval": args.require_approval,
});
if let Some(ttl) = args.ttl {
request["ttl_secs"] = json!(ttl.as_secs());
}
if let Some(subject) = &args.for_subject {
request["subject"] = json!(subject.trim_start_matches("0x"));
}
if let Some(addr) = &args.addr {
EnrollmentEndpoint::parse(addr).map_err(|e| invalid_args(format!("--addr: {e}")))?;
request["addr"] = json!(addr);
}
if let Some(subnet) = &args.subnet {
super::subnet::parse_subnet_path(subnet)?;
request["subnet"] = json!(subnet);
}
if let Some(rights) = &args.subnet_rights {
super::subnet::parse_subnet_rights(rights)?;
request["subnet_rights"] = json!(rights);
}
if args.standalone {
request["standalone"] = json!(true);
}
if let Some(org) = &args.org {
parse_org_id(org).map_err(invalid_args)?;
request["org"] = json!(org.trim_start_matches("0x"));
}
if let (Some(channel), Some(rights)) = (&args.channel, &args.channel_rights) {
super::channel::parse_channel_name(channel)?;
super::channel::parse_channel_rights(rights)?;
request["channel"] = json!(channel);
request["channel_rights"] = json!(rights);
}
let mut reply = node_request(args.state_dir, profile_name, request).await?;
if reply["bearer"] == Value::Bool(true) {
eprintln!(
"note: this token is bearer authorization — whoever redeems it first joins. \
Send it privately, or bind it to a device with --for."
);
}
if let Some(out) = &args.out {
let token = crate::secret::ScrubbedString::new(
reply["token"].as_str().unwrap_or_default().to_string(),
);
let tmp = out.with_extension("tmp-netmesh-join");
crate::commands::identity::write_identity_atomically(&tmp, out, token.as_bytes())
.await?;
reply["token"] = Value::Null;
reply["token_file"] = json!(out.display().to_string());
}
emit(&reply)
}
InviteCommand::Inspect(args) => {
let token = if args.token == "-" {
let mut buf = String::new();
use tokio::io::AsyncReadExt as _;
tokio::io::stdin()
.take(4096)
.read_to_string(&mut buf)
.await
.map_err(|e| generic(format!("read token from stdin: {e}")))?;
crate::secret::ScrubbedString::new(buf)
} else {
crate::secret::ScrubbedString::new(args.token)
};
let invite = MembershipInvite::decode(token.as_str())
.map_err(|e| invalid_args(format!("not a valid join token: {e}")))?;
let policy = invite.policy();
emit(&json!({
"issuer": hex::encode(invite.issuer().as_bytes()),
"issuer_fingerprint": invite.issuer_fingerprint(),
"endpoint": invite.endpoint().map(|e| e.as_str()),
"relay": invite.relay().map(|r| r.endpoint.as_str()),
"subnet": invite.subnet().map(|o| json!({
"authority": hex::encode(o.scope.authority.as_bytes()),
"scope": super::subnet::format_subnet(o.scope.path),
"rights": super::subnet::format_subnet_rights(o.rights),
})),
"channel": invite.channel().map(channel_json),
"enrollment_key": hex::encode(invite.enrollment_key().0),
"domain_name": invite.trust_domain_name(),
"trust_domain": invite.trust_domain().to_string(),
"relations": invite.relations().iter().map(|r| format!("{r:?}").to_lowercase()).collect::<Vec<_>>(),
"approval": approval_name(policy.approval_mode()),
"created_at": policy.created_at(),
"expires_at": policy.expires_at(),
"expired": now_unix() >= policy.expires_at(),
"bearer": invite.is_bearer(),
"intended_subject": invite.intended_subject().map(|e| hex::encode(e.as_bytes())),
"signature": "valid for the embedded issuer; confirm the fingerprint out of band",
}))
}
InviteCommand::Status(args) => {
let mut request = json!({ "op": "invite_status" });
if let Some(id) = args.offer_id {
request["offer_id"] = json!(id);
}
emit(&node_request(args.state_dir, profile_name, request).await?)
}
InviteCommand::Revoke(args) => emit(
&node_request(
args.state_dir,
profile_name,
json!({ "op": "invite_revoke", "offer_id": args.offer_id }),
)
.await?,
),
InviteCommand::Approve(args) => emit(
&node_request(
args.state_dir,
profile_name,
json!({ "op": "invite_approve", "offer_id": args.offer_id, "subject": args.subject.trim_start_matches("0x") }),
)
.await?,
),
InviteCommand::Deny(args) => emit(
&node_request(
args.state_dir,
profile_name,
json!({ "op": "invite_deny", "offer_id": args.offer_id, "subject": args.subject.trim_start_matches("0x") }),
)
.await?,
),
}
}
#[derive(Args, Debug)]
pub struct JoinArgs {
pub token: String,
#[arg(long, value_name = "DIR")]
pub state_dir: Option<PathBuf>,
#[arg(long)]
pub yes: bool,
#[arg(long, value_name = "DURATION", default_value = "15s", value_parser = crate::humantime::parse_duration)]
pub wait: Duration,
#[arg(long)]
pub rejoin: bool,
}
pub(crate) const JOIN_SUBDIR: &str = "join";
const DIRECT_ATTACH_WAIT: Duration = Duration::from_secs(5);
const DIRECT_RETRY_PAUSE: Duration = Duration::from_millis(500);
pub(crate) async fn attach_contact(
node: &Arc<net::adapter::net::MeshNode>,
contact: &MeshContact,
wait: Duration,
) -> Result<&'static str, String> {
let deadline = tokio::time::Instant::now() + wait;
let mut direct_failure = None;
if let Some(addr) = contact.addr {
let budget = if contact.relay.is_some() {
DIRECT_ATTACH_WAIT.min(wait)
} else {
wait
};
let direct_deadline = tokio::time::Instant::now() + budget;
loop {
let attempt = tokio::time::timeout_at(
direct_deadline,
node.connect_via(addr, &contact.noise_pubkey, contact.node_id),
)
.await;
match attempt {
Ok(Ok(_)) => return Ok("direct"),
Ok(Err(e)) => direct_failure = Some(format!("direct {addr}: {e}")),
Err(_) => {
direct_failure = Some(format!("direct {addr}: timed out"));
break;
}
}
if direct_deadline.saturating_duration_since(tokio::time::Instant::now())
< DIRECT_RETRY_PAUSE
{
break;
}
tokio::time::sleep(DIRECT_RETRY_PAUSE).await;
}
}
let Some(relay) = &contact.relay else {
return Err(direct_failure.unwrap_or_else(|| "the contact names no address".to_string()));
};
let relayed = tokio::time::timeout_at(deadline, async {
let addr = tokio::net::lookup_host(relay.endpoint.as_str())
.await
.map_err(|e| format!("resolve {}: {e}", relay.endpoint.as_str()))?
.next()
.ok_or_else(|| format!("{} does not resolve", relay.endpoint.as_str()))?;
let via = node
.relay_bind(addr, relay.registration)
.await
.map_err(|e| e.to_string())?;
node.connect_via_endpoint(via, &contact.noise_pubkey, contact.node_id)
.await
.map(|_| addr)
.map_err(|e| e.to_string())
})
.await;
let relay_failure = match relayed {
Ok(Ok(addr)) if node.relay_tunneled(addr) => return Ok("relay_tcp"),
Ok(Ok(_)) => return Ok("relay"),
Ok(Err(e)) => format!("relay {}: {e}", relay.endpoint.as_str()),
Err(_) => format!("relay {}: timed out", relay.endpoint.as_str()),
};
Err(match direct_failure {
Some(direct) => format!("{direct}; {relay_failure}"),
None => relay_failure,
})
}
pub(crate) async fn attach_mesh(
identity: Identity,
psk: &Psk,
contact: &MeshContact,
bind: Option<SocketAddr>,
wait: Duration,
) -> Result<(net_sdk::Mesh, &'static str), String> {
let hint = contact.addr.map(|a| a.ip()).or_else(|| {
contact
.relay
.as_ref()
.and_then(|r| r.endpoint.as_str().parse::<SocketAddr>().ok())
.map(|a| a.ip())
});
let bind = bind.unwrap_or_else(|| match hint {
Some(ip) if ip.is_loopback() => SocketAddr::new(ip, 0),
Some(ip) if ip.is_ipv6() => SocketAddr::from(([0u16; 8], 0)),
_ => SocketAddr::from(([0, 0, 0, 0], 0)),
});
let mesh = net_sdk::MeshBuilder::new(&bind.to_string(), psk.expose_bytes())
.map_err(|e| e.to_string())?
.identity(identity)
.build()
.await
.map_err(|e| e.to_string())?;
mesh.start();
match attach_contact(mesh.node(), contact, wait).await {
Ok(path) => Ok((mesh, path)),
Err(e) => {
let _ = mesh.shutdown().await;
Err(e)
}
}
}
pub async fn run_join(
args: JoinArgs,
output: Option<OutputFormat>,
profile_name: &str,
) -> Result<(), CliError> {
use net_sdk::enrollment::device::{DeviceJoin, DeviceJoinError, JoinStatus};
use net_sdk::enrollment::redeem::RedeemError;
let state = state_dir(args.state_dir, profile_name)?;
let from_stdin = args.token == "-";
let token = if from_stdin {
let mut buf = String::new();
use tokio::io::AsyncReadExt as _;
tokio::io::stdin()
.take(4096)
.read_to_string(&mut buf)
.await
.map_err(|e| generic(format!("read token from stdin: {e}")))?;
crate::secret::ScrubbedString::new(buf)
} else {
crate::secret::ScrubbedString::new(args.token)
};
let invite = MembershipInvite::decode(token.as_str())
.map_err(|e| invalid_args(format!("not a valid join token: {e}")))?;
let policy = invite.policy();
if now_unix() >= policy.expires_at() {
return Err(invalid_args(
"this join token has expired; ask for a new one",
));
}
let direct = invite.endpoint().map_or("none", |e| e.as_str());
let enroll_via = match invite.relay() {
Some(relay) => format!("{direct} (relay fallback {})", relay.endpoint.as_str()),
None => direct.to_string(),
};
eprintln!(
"Joining trust domain '{}' (id {}) run by issuer {}\n enrollment address: {}\n token: {}, expires at unix {}",
invite.trust_domain_name(),
invite.trust_domain(),
invite.issuer_fingerprint(),
enroll_via,
if invite.is_bearer() {
"bearer (the first redeemer joins)"
} else {
"bound to one device"
},
policy.expires_at(),
);
let tty = {
use std::io::IsTerminal as _;
std::io::stdin().is_terminal() && !from_stdin
};
let yes = args.yes;
tokio::task::spawn_blocking(move || {
super::ice::check_confirm_gate(tty, yes, || {
use std::io::{BufRead as _, Write as _};
let mut err = std::io::stderr();
write!(
err,
"Confirm the issuer fingerprint is the one you expect. Type YES to join: "
)
.and_then(|()| err.flush())
.map_err(|e| generic(format!("prompt: {e}")))?;
let mut line = String::new();
std::io::stdin()
.lock()
.read_line(&mut line)
.map_err(|e| generic(format!("prompt: {e}")))?;
Ok(line.trim() == "YES")
})
})
.await
.map_err(|e| generic(format!("confirmation task failed: {e}")))??;
std::fs::create_dir_all(&state)
.map_err(|e| generic(format!("create state directory {}: {e}", state.display())))?;
let dir = state.join(JOIN_SUBDIR);
let storage_err = |e: DeviceJoinError| match e {
DeviceJoinError::Storage(
net::adapter::net::behavior::enrollment_storage::StorageError::Busy,
) => generic(format!(
"join state {} is in use (is `net-mesh up` running with it?)",
dir.display()
)),
other => generic(format!("join state {}: {other}", dir.display())),
};
let mut join = if dir.exists() {
let mut join = DeviceJoin::open(&dir).map_err(storage_err)?;
if join.invite().digest() != invite.digest() {
return Err(invalid_args(format!(
"{} already holds a join from a different token; use another --state-dir",
state.display()
)));
}
if let Some(at) = join.left_at() {
if !args.rejoin {
return Err(invalid_args(format!(
"this device left the mesh at unix {at}; pass --rejoin to join again \
(the issuer must still authorize it)"
)));
}
join.rejoin().map_err(storage_err)?;
}
join
} else {
DeviceJoin::begin(&dir, &invite, Identity::generate()).map_err(storage_err)?
};
let device = hex::encode(join.identity().entity_id().as_bytes());
let fmt = OutputFormat::resolve_oneshot(output);
let status = join.redeem(args.wait).await.map_err(|e| match e {
DeviceJoinError::Redeem(RedeemError::Refused(r)) => {
generic(format!("the issuer refused this join: {r}"))
}
DeviceJoinError::Redeem(
RedeemError::Io(_)
| RedeemError::Timeout
| RedeemError::Handshake
| RedeemError::Relay(_),
) => connection_failure(format!(
"could not complete enrollment via {enroll_via}: {e}"
)),
other => generic(format!("join failed: {other}")),
})?;
if status == JoinStatus::PendingApproval {
return emit_value(
fmt,
&json!({
"state": "pending_approval",
"device": device,
"detail": "the operator must approve this device; run the same join again afterwards",
}),
)
.map_err(|e| generic(format!("write result: {e}")));
}
let Some(bundle) = join.bundle() else {
return Err(generic("join reported installed without a bundle"));
};
let contact = bundle.contact().clone();
let trust_domain = bundle.psk().trust_domain().to_string();
let channel_expires_at = bundle
.channel_chain()
.and_then(|chain| chain.tokens.last().map(|leaf| leaf.not_after));
let enroll_path = join.last_path().map(|p| p.as_str());
let authority_dir = state.join(super::lifecycle::AUTHORITY_SUBDIR);
let org = match bundle.org_membership() {
Some(_) if super::lifecycle::read_org_left(&authority_dir).is_some() => Some(json!({
"state": "left",
"detail": "this device left the org; a new approved org link (`org join`) rejoins",
})),
Some(cert) => {
let (cert, entity, root, audience) = (
cert.clone(),
join.identity().entity_id().clone(),
state.clone(),
bundle.org_audience(),
);
let adopted = tokio::task::spawn_blocking(move || {
super::lifecycle::adopt_org_membership(&root, cert, &entity, audience.as_ref())
})
.await
.map_err(|e| generic(format!("org adoption task failed: {e}")))?
.map_err(|e| {
generic(format!(
"credentials are installed, but adopting the org membership failed: {e}"
))
})?;
Some(adopted)
}
None => None,
};
let attach_path = match attach_mesh(
join.identity().clone(),
bundle.psk(),
&contact,
None,
args.wait,
)
.await
{
Ok((mesh, path)) => {
let _ = mesh.shutdown().await;
path
}
Err(e) => {
return Err(connection_failure(format!(
"credentials are installed, but the live attach failed ({e}); `net-mesh up` retries it"
)))
}
};
emit_value(
fmt,
&json!({
"state": "joined",
"device": device,
"issuer_fingerprint": invite.issuer_fingerprint(),
"domain_name": invite.trust_domain_name(),
"trust_domain": trust_domain,
"contact": contact.addr.map(|a| a.to_string()),
"relay": contact.relay.as_ref().map(|r| r.endpoint.as_str()),
"subnet": invite.subnet().map(|o| json!({
"scope": super::subnet::format_subnet(o.scope.path),
"rights": super::subnet::format_subnet_rights(o.rights),
"credentials": "installed",
})),
"channel": invite.channel().map(|o| {
let mut v = channel_json(o);
v["credential"] = json!("stored");
v["credential_expires_at"] = json!(channel_expires_at);
v["delegation"] = json!(CHANNEL_DELEGATION);
v
}),
"org": org,
"installed": true,
"attached": true,
"enroll_path": enroll_path,
"attach_path": attach_path,
}),
)
.map_err(|e| generic(format!("write result: {e}")))
}
#[cfg(test)]
mod tests {
use super::*;
fn ep(s: &str) -> EnrollmentEndpoint {
EnrollmentEndpoint::parse(s).unwrap()
}
#[test]
fn default_endpoint_prefers_public_then_full_mapping_then_concrete_bind() {
let bind_any: SocketAddr = "0.0.0.0:7443".parse().unwrap();
let bind_lan: SocketAddr = "192.168.1.20:7443".parse().unwrap();
let tcp: SocketAddr = "203.0.113.9:40001".parse().unwrap();
let udp: SocketAddr = "203.0.113.9:40002".parse().unwrap();
let public = ep("node.example.net:7443");
let pick = |p: Option<&EnrollmentEndpoint>, t, u, b| {
select_endpoint(p, t, u, b).map(|e| e.as_str().to_string())
};
assert_eq!(
pick(Some(&public), Some(tcp), Some(udp), bind_lan).as_deref(),
Some("node.example.net:7443")
);
assert_eq!(
pick(None, Some(tcp), Some(udp), bind_any).as_deref(),
Some("203.0.113.9:40001")
);
assert_eq!(pick(None, Some(tcp), None, bind_any), None);
assert_eq!(
pick(None, None, Some(udp), bind_lan).as_deref(),
Some("192.168.1.20:7443")
);
assert_eq!(pick(None, None, None, bind_any), None);
}
}