nxd_core/adapters/
secret_supervisor.rs

1//! Supervised client for external secret resolver/sink executables.
2//!
3//! Secrets travel only over a private local channel. Values are never put in
4//! argv, environment, plans, or logs. Sessions are action-scoped: every resolve
5//! /store/delete carries the approved action identity.
6
7use crate::adapters::external_runtime::SecretExecutable;
8use crate::contract::SecretBinding;
9use hyper_util::rt::TokioIo;
10use nxd_secret_protocol::v1::secret_service_client::SecretServiceClient;
11use nxd_secret_protocol::{MAX_SECRET_BYTES, PROTOCOL_VERSION, v1};
12use std::fs;
13use std::os::unix::fs::PermissionsExt;
14use std::path::{Path, PathBuf};
15use std::process::Stdio;
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::time::{Duration, SystemTime, UNIX_EPOCH};
18use tokio::io::AsyncReadExt;
19use tokio::net::UnixStream;
20use tokio::process::{Child, Command};
21use tonic::transport::{Channel, Endpoint};
22use tower::service_fn;
23use zeroize::Zeroizing;
24
25static SESSION_SEQUENCE: AtomicU64 = AtomicU64::new(0);
26
27#[derive(Clone, Debug, PartialEq, Eq)]
28pub enum SecretRuntimeError {
29	InvalidCommand,
30	Spawn,
31	StartupTimeout,
32	ProcessExited,
33	Transport,
34	Incompatible,
35	Protocol,
36	RpcTimeout,
37	NotFound,
38	AlreadyExists,
39	PermissionDenied,
40	Unsupported,
41	InvalidSecret,
42	CapabilityUnavailable(String),
43}
44
45impl std::fmt::Display for SecretRuntimeError {
46	fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47		match self {
48			Self::CapabilityUnavailable(message) => write!(formatter, "{message}"),
49			Self::AlreadyExists => {
50				write!(formatter, "secret binding already exists and overwrite was not approved")
51			}
52			other => write!(formatter, "secret runtime error: {other:?}"),
53		}
54	}
55}
56
57impl std::error::Error for SecretRuntimeError {}
58
59pub struct SecretSession {
60	child: Child,
61	client: SecretServiceClient<Channel>,
62	_run_dir: PrivateRunDirectory,
63	stderr_task: tokio::task::JoinHandle<()>,
64	request_timeout: Duration,
65	resolver_instance: String,
66	capabilities: Vec<String>,
67}
68
69impl SecretSession {
70	pub async fn start(
71		executable: &SecretExecutable,
72		startup_timeout: Duration,
73		request_timeout: Duration,
74	) -> Result<Self, SecretRuntimeError> {
75		if !executable.identity.path.is_absolute() || !executable.identity.path.is_file() {
76			return Err(SecretRuntimeError::InvalidCommand);
77		}
78		let run_dir = create_run_dir()?;
79		let socket = run_dir.0.join("secret.sock");
80		let mut child_command = Command::new(&executable.identity.path);
81		child_command
82			.env_clear()
83			.args(&executable.arguments)
84			.arg("--socket")
85			.arg(&socket)
86			.stdin(Stdio::null())
87			.stdout(Stdio::null())
88			.stderr(Stdio::piped())
89			.kill_on_drop(true);
90		let mut child = child_command.spawn().map_err(|_| SecretRuntimeError::Spawn)?;
91		let mut stderr = child.stderr.take().ok_or(SecretRuntimeError::Spawn)?;
92		let stderr_task = tokio::spawn(async move {
93			let mut output = Vec::new();
94			let _ = (&mut stderr).take(8192).read_to_end(&mut output).await;
95			output.fill(0);
96		});
97		let channel = connect(&socket, &mut child, startup_timeout).await?;
98		let mut session = Self {
99			child,
100			client: SecretServiceClient::new(channel),
101			_run_dir: run_dir,
102			stderr_task,
103			request_timeout,
104			resolver_instance: executable.resolver_id.clone(),
105			capabilities: Vec::new(),
106		};
107		let describe = tokio::time::timeout(
108			request_timeout,
109			session.client.describe(v1::DescribeRequest {
110				context: Some(request_context(&session.resolver_instance, "describe", request_timeout)),
111			}),
112		)
113		.await
114		.map_err(|_| SecretRuntimeError::RpcTimeout)?
115		.map_err(|status| {
116			SecretRuntimeError::CapabilityUnavailable(format!(
117				"secret resolver describe failed: {}",
118				status.message()
119			))
120		})?
121		.into_inner();
122		if describe.protocol_version != PROTOCOL_VERSION {
123			return Err(SecretRuntimeError::Incompatible);
124		}
125		session.capabilities = describe.capabilities;
126		Ok(session)
127	}
128
129	pub fn capabilities(&self) -> &[String] {
130		&self.capabilities
131	}
132
133	pub async fn resolve(
134		&mut self,
135		binding: &SecretBinding,
136		action_id: &str,
137	) -> Result<Zeroizing<Vec<u8>>, SecretRuntimeError> {
138		if !self.capabilities.iter().any(|capability| capability == "resolve") {
139			return Err(SecretRuntimeError::CapabilityUnavailable(
140				"secret executable does not advertise resolve".to_string(),
141			));
142		}
143		let response = tokio::time::timeout(
144			self.request_timeout,
145			self.client.resolve(v1::ResolveRequest {
146				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
147				binding: Some(wire_binding(binding)),
148			}),
149		)
150		.await
151		.map_err(|_| SecretRuntimeError::RpcTimeout)?
152		.map_err(map_status)?
153		.into_inner();
154		if response.value.is_empty() || response.value.len() > MAX_SECRET_BYTES {
155			return Err(SecretRuntimeError::InvalidSecret);
156		}
157		Ok(Zeroizing::new(response.value))
158	}
159
160	pub async fn inspect_binding(
161		&mut self,
162		binding: &SecretBinding,
163		action_id: &str,
164	) -> Result<v1::InspectBindingResponse, SecretRuntimeError> {
165		if !self.capabilities.iter().any(|capability| capability == "inspect-binding-v1") {
166			return Err(SecretRuntimeError::CapabilityUnavailable(
167				"secret executable does not advertise inspect-binding-v1".to_string(),
168			));
169		}
170		let response = tokio::time::timeout(
171			self.request_timeout,
172			self.client.inspect_binding(v1::InspectBindingRequest {
173				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
174				binding: Some(wire_binding(binding)),
175			}),
176		)
177		.await
178		.map_err(|_| SecretRuntimeError::RpcTimeout)?
179		.map_err(map_status)?
180		.into_inner();
181		if response.present {
182			if response.document_sha256.len() != 64
183				|| !response.document_sha256.bytes().all(|byte| byte.is_ascii_hexdigit())
184			{
185				return Err(SecretRuntimeError::CapabilityUnavailable(
186					"secret resolver returned an invalid binding inspection digest".into(),
187				));
188			}
189		} else if !response.document_sha256.is_empty()
190			&& (response.document_sha256.len() != 64
191				|| !response.document_sha256.bytes().all(|byte| byte.is_ascii_hexdigit()))
192		{
193			return Err(SecretRuntimeError::CapabilityUnavailable(
194				"secret resolver returned an invalid absent-binding document digest".into(),
195			));
196		}
197		Ok(response)
198	}
199
200	pub async fn store(
201		&mut self,
202		binding: &SecretBinding,
203		action_id: &str,
204		value: &[u8],
205		allow_overwrite: bool,
206		allow_create: bool,
207		expected_previous_document_sha256: &str,
208	) -> Result<String, SecretRuntimeError> {
209		if !self.capabilities.iter().any(|capability| capability == "store") {
210			return Err(SecretRuntimeError::CapabilityUnavailable(
211				"secret executable does not advertise store (sink)".to_string(),
212			));
213		}
214		if value.is_empty() || value.len() > MAX_SECRET_BYTES {
215			return Err(SecretRuntimeError::InvalidSecret);
216		}
217		let response = tokio::time::timeout(
218			self.request_timeout,
219			self.client.store(v1::StoreRequest {
220				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
221				binding: Some(wire_binding(binding)),
222				value: value.to_vec(),
223				allow_overwrite,
224				allow_create,
225				expected_previous_document_sha256: expected_previous_document_sha256.to_string(),
226			}),
227		)
228		.await
229		.map_err(|_| SecretRuntimeError::RpcTimeout)?
230		.map_err(map_status)?
231		.into_inner();
232		if response.receipt.is_empty() {
233			return Err(SecretRuntimeError::Protocol);
234		}
235		Ok(response.receipt)
236	}
237
238	pub async fn delete(
239		&mut self,
240		binding: &SecretBinding,
241		action_id: &str,
242	) -> Result<String, SecretRuntimeError> {
243		if !self.capabilities.iter().any(|capability| capability == "delete") {
244			return Err(SecretRuntimeError::CapabilityUnavailable(
245				"secret executable does not advertise delete (sink)".to_string(),
246			));
247		}
248		let response = tokio::time::timeout(
249			self.request_timeout,
250			self.client.delete(v1::DeleteRequest {
251				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
252				binding: Some(wire_binding(binding)),
253			}),
254		)
255		.await
256		.map_err(|_| SecretRuntimeError::RpcTimeout)?
257		.map_err(map_status)?
258		.into_inner();
259		if response.receipt.is_empty() {
260			return Err(SecretRuntimeError::Protocol);
261		}
262		Ok(response.receipt)
263	}
264
265	pub async fn resolve_public_artifact(
266		&mut self,
267		binding: &SecretBinding,
268		action_id: &str,
269	) -> Result<v1::ResolvePublicArtifactResponse, SecretRuntimeError> {
270		if !self.capabilities.iter().any(|capability| capability == "public-artifact-resolve-v1") {
271			return Err(SecretRuntimeError::CapabilityUnavailable(
272				"secret executable does not advertise public-artifact-resolve-v1".to_string(),
273			));
274		}
275		let response = tokio::time::timeout(
276			self.request_timeout,
277			self.client.resolve_public_artifact(v1::ResolvePublicArtifactRequest {
278				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
279				binding: Some(wire_public_binding(binding)),
280			}),
281		)
282		.await
283		.map_err(|_| SecretRuntimeError::RpcTimeout)?
284		.map_err(map_status)?
285		.into_inner();
286		if response.value.is_empty() || response.sha256.len() != 64 {
287			return Err(SecretRuntimeError::CapabilityUnavailable(
288				"secret resolver returned an invalid public artifact".into(),
289			));
290		}
291		Ok(response)
292	}
293
294	pub async fn store_public_artifact(
295		&mut self,
296		binding: &SecretBinding,
297		action_id: &str,
298		value: &[u8],
299		expected_previous_sha256: &str,
300	) -> Result<v1::StorePublicArtifactResponse, SecretRuntimeError> {
301		if !self.capabilities.iter().any(|capability| capability == "public-artifact-store-v1") {
302			return Err(SecretRuntimeError::CapabilityUnavailable(
303				"secret executable does not advertise public-artifact-store-v1".to_string(),
304			));
305		}
306		let response = tokio::time::timeout(
307			self.request_timeout,
308			self.client.store_public_artifact(v1::StorePublicArtifactRequest {
309				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
310				binding: Some(wire_public_binding(binding)),
311				value: value.to_vec(),
312				expected_previous_sha256: expected_previous_sha256.to_string(),
313			}),
314		)
315		.await
316		.map_err(|_| SecretRuntimeError::RpcTimeout)?
317		.map_err(map_status)?
318		.into_inner();
319		if response.receipt.is_empty() || response.sha256.len() != 64 {
320			return Err(SecretRuntimeError::Protocol);
321		}
322		Ok(response)
323	}
324
325	pub async fn inspect_recipient_policy(
326		&mut self,
327		action_id: &str,
328		identity_id: &str,
329		age_recipient: &str,
330		recipient_alias: &str,
331		creation_rule_path_regex: &str,
332		document_references: Vec<String>,
333	) -> Result<v1::InspectRecipientPolicyResponse, SecretRuntimeError> {
334		if !self.capabilities.iter().any(|capability| capability == "recipient-policy-v1") {
335			return Err(SecretRuntimeError::CapabilityUnavailable(
336				"secret executable does not advertise recipient-policy-v1".to_string(),
337			));
338		}
339		let response = tokio::time::timeout(
340			self.request_timeout,
341			self.client.inspect_recipient_policy(v1::InspectRecipientPolicyRequest {
342				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
343				identity_id: identity_id.to_string(),
344				age_recipient: age_recipient.to_string(),
345				creation_rule_path_regex: creation_rule_path_regex.to_string(),
346				document_references,
347				recipient_alias: recipient_alias.to_string(),
348			}),
349		)
350		.await
351		.map_err(|_| SecretRuntimeError::RpcTimeout)?
352		.map_err(map_status)?
353		.into_inner();
354		if response.policy_sha256.is_empty() {
355			return Err(SecretRuntimeError::CapabilityUnavailable(
356				"secret resolver returned an incomplete recipient-policy inspection".into(),
357			));
358		}
359		Ok(response)
360	}
361
362	#[allow(clippy::too_many_arguments)]
363	pub async fn apply_recipient_policy(
364		&mut self,
365		action_id: &str,
366		identity_id: &str,
367		age_recipient: &str,
368		recipient_alias: &str,
369		creation_rule_path_regex: &str,
370		expected_policy_sha256: &str,
371		documents: Vec<v1::RecipientPolicyDocument>,
372		expected_previous_age_recipient: Option<&str>,
373	) -> Result<v1::ApplyRecipientPolicyResponse, SecretRuntimeError> {
374		if !self.capabilities.iter().any(|capability| capability == "recipient-policy-v1") {
375			return Err(SecretRuntimeError::CapabilityUnavailable(
376				"secret executable does not advertise recipient-policy-v1".to_string(),
377			));
378		}
379		let response = tokio::time::timeout(
380			self.request_timeout,
381			self.client.apply_recipient_policy(v1::ApplyRecipientPolicyRequest {
382				context: Some(request_context(&self.resolver_instance, action_id, self.request_timeout)),
383				identity_id: identity_id.to_string(),
384				age_recipient: age_recipient.to_string(),
385				creation_rule_path_regex: creation_rule_path_regex.to_string(),
386				expected_policy_sha256: expected_policy_sha256.to_string(),
387				documents,
388				recipient_alias: recipient_alias.to_string(),
389				expected_previous_age_recipient: expected_previous_age_recipient
390					.unwrap_or_default()
391					.to_string(),
392			}),
393		)
394		.await
395		.map_err(|_| SecretRuntimeError::RpcTimeout)?
396		.map_err(map_status)?
397		.into_inner();
398		if response.receipt.is_empty()
399			|| response.policy_sha256.is_empty()
400			|| response.documents.is_empty()
401		{
402			return Err(SecretRuntimeError::Protocol);
403		}
404		Ok(response)
405	}
406}
407
408impl Drop for SecretSession {
409	fn drop(&mut self) {
410		let _ = self.child.start_kill();
411		self.stderr_task.abort();
412	}
413}
414
415struct PrivateRunDirectory(PathBuf);
416
417impl Drop for PrivateRunDirectory {
418	fn drop(&mut self) {
419		let _ = fs::remove_dir_all(&self.0);
420	}
421}
422
423fn create_run_dir() -> Result<PrivateRunDirectory, SecretRuntimeError> {
424	let sequence = SESSION_SEQUENCE.fetch_add(1, Ordering::Relaxed);
425	let run_dir = PathBuf::from("/tmp").join(format!("nxd-secret-{}-{sequence}", std::process::id()));
426	fs::create_dir(&run_dir).map_err(|_| SecretRuntimeError::Spawn)?;
427	fs::set_permissions(&run_dir, fs::Permissions::from_mode(0o700))
428		.map_err(|_| SecretRuntimeError::Spawn)?;
429	Ok(PrivateRunDirectory(run_dir))
430}
431
432async fn connect(
433	socket: &Path,
434	child: &mut Child,
435	timeout: Duration,
436) -> Result<Channel, SecretRuntimeError> {
437	let socket = socket.to_path_buf();
438	tokio::time::timeout(timeout, async {
439		loop {
440			if child.try_wait().map_err(|_| SecretRuntimeError::ProcessExited)?.is_some() {
441				return Err(SecretRuntimeError::ProcessExited);
442			}
443			let path = socket.clone();
444			let endpoint =
445				Endpoint::from_static("http://[::]:50051").connect_timeout(Duration::from_millis(100));
446			if let Ok(channel) = endpoint
447				.connect_with_connector(service_fn(move |_| {
448					let path = path.clone();
449					async move { UnixStream::connect(path).await.map(TokioIo::new) }
450				}))
451				.await
452			{
453				return Ok(channel);
454			}
455			tokio::time::sleep(Duration::from_millis(20)).await;
456		}
457	})
458	.await
459	.map_err(|_| SecretRuntimeError::StartupTimeout)?
460}
461
462fn request_context(
463	resolver_instance: &str,
464	action_id: &str,
465	request_timeout: Duration,
466) -> v1::RequestContext {
467	let deadline_unix_ms = SystemTime::now()
468		.duration_since(UNIX_EPOCH)
469		.unwrap_or_default()
470		.saturating_add(request_timeout)
471		.as_millis() as u64;
472	let action_id = if action_id.starts_with("action/") || action_id == "describe" {
473		action_id.to_string()
474	} else {
475		format!("action/{action_id}")
476	};
477	v1::RequestContext {
478		protocol_version: PROTOCOL_VERSION.to_string(),
479		resolver_instance: resolver_instance.to_string(),
480		action_id,
481		deadline_unix_ms,
482	}
483}
484
485fn wire_binding(binding: &SecretBinding) -> v1::SecretBinding {
486	v1::SecretBinding {
487		id: binding.id.clone(),
488		resolver: binding.resolver.clone(),
489		reference: binding.reference.clone(),
490	}
491}
492
493fn wire_public_binding(binding: &SecretBinding) -> v1::PublicArtifactBinding {
494	v1::PublicArtifactBinding {
495		id: binding.id.clone(),
496		resolver: binding.resolver.clone(),
497		reference: binding.reference.clone(),
498	}
499}
500
501fn map_status(status: tonic::Status) -> SecretRuntimeError {
502	match status.code() {
503		tonic::Code::NotFound => SecretRuntimeError::NotFound,
504		tonic::Code::AlreadyExists => SecretRuntimeError::AlreadyExists,
505		tonic::Code::PermissionDenied => SecretRuntimeError::PermissionDenied,
506		tonic::Code::DeadlineExceeded => SecretRuntimeError::RpcTimeout,
507		tonic::Code::Unimplemented => SecretRuntimeError::Unsupported,
508		tonic::Code::InvalidArgument => {
509			SecretRuntimeError::CapabilityUnavailable(status.message().to_string())
510		}
511		tonic::Code::FailedPrecondition => {
512			SecretRuntimeError::CapabilityUnavailable(status.message().to_string())
513		}
514		tonic::Code::Internal | tonic::Code::DataLoss => {
515			SecretRuntimeError::CapabilityUnavailable(status.message().to_string())
516		}
517		_ => SecretRuntimeError::Transport,
518	}
519}
nxd_core/adapters/secret_supervisor.rs

nxd_core/adapters/
secret_supervisor.rs
