nxd_plugin_protocol/
lib.rs

1mod progress;
2pub use progress::{EventSequencer, ProgressBatcher, ProgressReporter};
3
4pub const PROTOCOL_VERSION: &str = "1.1";
5pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024;
6/// Hard cap on retained apply events. Progress is streamed/logged; only the
7/// terminal Completed event must be retained for confidential outputs.
8pub const MAX_EVENTS: usize = 128;
9/// gRPC status/trailers live in HTTP/2 HEADERS frames. Keep well under the
10/// default SETTINGS_MAX_HEADER_LIST_SIZE (~16–64 KiB) so large Nix stderr cannot
11/// GOAWAY the plugin connection with `header_list_way_too_large`.
12pub const MAX_STATUS_MESSAGE_BYTES: usize = 4 * 1024;
13/// Progress/detail body payloads (DATA frames). Smaller than MAX_MESSAGE_BYTES
14/// so one event stays cheap to buffer on the wire.
15pub const MAX_PROGRESS_MESSAGE_BYTES: usize = 64 * 1024;
16
17/// Truncate a human message for gRPC Status / trailers (HEADERS).
18///
19/// Prefer the **tail**: Nix/process failures print progress first and the real
20/// error last. Keeping the head (derivation lists) hides the failure.
21pub fn truncate_status_message(message: &str) -> String {
22	truncate_utf8_keep_tail(message, MAX_STATUS_MESSAGE_BYTES)
23}
24
25/// Truncate a progress/detail message for ApplyResponse DATA frames.
26/// Progress chunks already batch recent lines; keep the head of each chunk.
27pub fn truncate_progress_message(message: &str) -> String {
28	truncate_utf8_keep_head(message, MAX_PROGRESS_MESSAGE_BYTES)
29}
30
31fn truncate_utf8_keep_head(message: &str, max_bytes: usize) -> String {
32	if message.len() <= max_bytes {
33		return message.to_string();
34	}
35	let mut end = max_bytes.saturating_sub(32);
36	while end > 0 && !message.is_char_boundary(end) {
37		end -= 1;
38	}
39	format!("{}…[truncated {} bytes]", &message[..end], message.len().saturating_sub(end))
40}
41
42fn truncate_utf8_keep_tail(message: &str, max_bytes: usize) -> String {
43	if message.len() <= max_bytes {
44		return message.to_string();
45	}
46	let budget = max_bytes.saturating_sub(40);
47	let mut start = message.len().saturating_sub(budget);
48	while start < message.len() && !message.is_char_boundary(start) {
49		start += 1;
50	}
51	format!("[truncated {start} leading bytes]…{}", &message[start..])
52}
53
54const SENSITIVE_KEYS: &[&str] = &[
55	"authorization",
56	"credential",
57	"private-key",
58	"private_key",
59	"password",
60	"passwd",
61	"api-key",
62	"api_key",
63	"apikey",
64	"secret",
65	"token",
66	"cookie",
67];
68
69/// Language-neutral `safe_message` policy helper for bundled Rust providers and core.
70///
71/// This is deliberately content preserving: it removes control characters and
72/// sensitive values, not ordinary operational lines.
73#[derive(Clone, Debug, Default)]
74pub struct SensitiveTextRedactor {
75	in_private_key: bool,
76	exact_values: Vec<String>,
77}
78
79impl SensitiveTextRedactor {
80	pub fn with_sensitive_values<'a>(values: impl IntoIterator<Item = &'a [u8]>) -> Self {
81		let mut exact_values = values
82			.into_iter()
83			.filter_map(|value| std::str::from_utf8(value).ok())
84			.filter(|value| !value.is_empty())
85			.map(ToOwned::to_owned)
86			.collect::<Vec<_>>();
87		exact_values.sort_by_key(|value| std::cmp::Reverse(value.len()));
88		exact_values.dedup();
89		Self { in_private_key: false, exact_values }
90	}
91
92	pub fn redact(&mut self, input: &str) -> String {
93		let mut exact_redacted = input.to_string();
94		for value in &self.exact_values {
95			exact_redacted = exact_redacted.replace(value, "[REDACTED]");
96		}
97		let mut output = String::with_capacity(exact_redacted.len());
98		for segment in exact_redacted.split_inclusive('\n') {
99			let (line, newline) = segment.strip_suffix('\n').map_or((segment, ""), |line| (line, "\n"));
100			output.push_str(&self.redact_line(line.trim_end_matches('\r')));
101			output.push_str(newline);
102		}
103		output
104	}
105
106	fn redact_line(&mut self, input: &str) -> String {
107		let mut line = input.chars().filter(|ch| *ch == '\t' || !ch.is_control()).collect::<String>();
108		let upper = line.to_ascii_uppercase();
109		if upper.contains("-----BEGIN ") && upper.contains("PRIVATE KEY-----") {
110			self.in_private_key = true;
111			return "[REDACTED PRIVATE KEY MATERIAL]".into();
112		}
113		if self.in_private_key {
114			if upper.contains("-----END ") && upper.contains("PRIVATE KEY-----") {
115				self.in_private_key = false;
116			}
117			return "[REDACTED PRIVATE KEY MATERIAL]".into();
118		}
119
120		line = redact_uri_credentials(&line);
121		redact_key_values(&line)
122	}
123}
124
125pub fn redact_sensitive_text(input: &str) -> String {
126	SensitiveTextRedactor::default().redact(input)
127}
128
129fn redact_uri_credentials(input: &str) -> String {
130	let mut output = input.to_string();
131	let mut search_from = 0;
132	while let Some(relative_scheme) = output[search_from..].find("://") {
133		let authority_start = search_from + relative_scheme + 3;
134		let authority_end = output[authority_start..]
135			.find(|ch: char| ch == '/' || ch.is_whitespace())
136			.map_or(output.len(), |relative| authority_start + relative);
137		let Some(relative_at) = output[authority_start..authority_end].rfind('@') else {
138			search_from = authority_end;
139			continue;
140		};
141		let at = authority_start + relative_at;
142		if !output[authority_start..at].contains(':') {
143			search_from = authority_end;
144			continue;
145		}
146		output.replace_range(authority_start..at, "[REDACTED]");
147		search_from = authority_start + "[REDACTED]@".len();
148	}
149	output
150}
151
152fn redact_key_values(input: &str) -> String {
153	let lower = input.to_ascii_lowercase();
154	let mut ranges = Vec::new();
155	for key in SENSITIVE_KEYS {
156		let mut offset = 0;
157		while let Some(relative) = lower[offset..].find(key) {
158			let start = offset + relative;
159			let end_key = start + key.len();
160			let before = lower[..start].chars().next_back();
161			let cli_key = lower[..start].ends_with("--");
162			let before_boundary =
163				cli_key || before.is_none_or(|ch| !ch.is_ascii_alphanumeric() && ch != '_' && ch != '-');
164			let after_boundary = lower[end_key..]
165				.chars()
166				.next()
167				.is_none_or(|ch| !ch.is_ascii_alphanumeric() && ch != '_' && ch != '-');
168			if !before_boundary || !after_boundary {
169				offset = end_key;
170				continue;
171			}
172
173			let bytes = lower.as_bytes();
174			let mut cursor = end_key;
175			if bytes.get(cursor) == Some(&b'\"') {
176				cursor += 1;
177			}
178			let whitespace_start = cursor;
179			while bytes.get(cursor).is_some_and(u8::is_ascii_whitespace) {
180				cursor += 1;
181			}
182			let has_whitespace = cursor > whitespace_start;
183			let has_separator = bytes.get(cursor).is_some_and(|byte| *byte == b'=' || *byte == b':');
184			if has_separator {
185				cursor += 1;
186				while bytes.get(cursor).is_some_and(u8::is_ascii_whitespace) {
187					cursor += 1;
188				}
189			} else if !has_whitespace {
190				offset = end_key;
191				continue;
192			}
193			if cursor >= input.len() {
194				offset = end_key;
195				continue;
196			}
197
198			let value_start = cursor;
199			let value_end = if bytes[value_start] == b'\"' || bytes[value_start] == b'\'' {
200				let quote = bytes[value_start];
201				let content_start = value_start + 1;
202				let content_end = bytes[content_start..]
203					.iter()
204					.position(|byte| *byte == quote)
205					.map_or(input.len(), |relative| content_start + relative);
206				ranges.push((content_start, content_end));
207				offset = content_end;
208				continue;
209			} else if matches!(*key, "authorization" | "cookie" | "credential") {
210				input.len()
211			} else {
212				bytes[value_start..]
213					.iter()
214					.position(|byte| byte.is_ascii_whitespace() || b",;&}".contains(byte))
215					.map_or(input.len(), |relative| value_start + relative)
216			};
217			if value_end > value_start {
218				ranges.push((value_start, value_end));
219			}
220			offset = value_end.max(end_key);
221		}
222	}
223	ranges.sort_unstable();
224	ranges.dedup();
225	let mut output = input.to_string();
226	for (start, end) in ranges.into_iter().rev() {
227		if end <= output.len() && start < end {
228			output.replace_range(start..end, "[REDACTED]");
229		}
230	}
231	output
232}
233
234pub mod v1 {
235	tonic::include_proto!("nxd.plugin.v1");
236}
237
238#[derive(Clone, Debug, PartialEq, Eq)]
239pub enum ProtocolError {
240	MissingContext,
241	IncompatibleVersion(String),
242	InvalidIdentity,
243	InvalidDeadline,
244	MessageTooLarge,
245	InvalidCanonicalJson,
246	InvalidMetadata,
247}
248
249pub fn validate_context(context: Option<&v1::RequestContext>) -> Result<(), ProtocolError> {
250	let context = context.ok_or(ProtocolError::MissingContext)?;
251	if context.protocol_version != PROTOCOL_VERSION {
252		return Err(ProtocolError::IncompatibleVersion(context.protocol_version.clone()));
253	}
254	if context.provider_instance.is_empty()
255		|| context.operation_id.is_empty()
256		|| context.provider_instance.chars().chain(context.operation_id.chars()).any(char::is_control)
257	{
258		return Err(ProtocolError::InvalidIdentity);
259	}
260	if context.deadline_unix_ms == 0 {
261		return Err(ProtocolError::InvalidDeadline);
262	}
263	Ok(())
264}
265
266pub fn validate_canonical_json(input: &[u8]) -> Result<serde_json::Value, ProtocolError> {
267	if input.len() > MAX_MESSAGE_BYTES {
268		return Err(ProtocolError::MessageTooLarge);
269	}
270	serde_json::from_slice(input).map_err(|_| ProtocolError::InvalidCanonicalJson)
271}
272
273pub fn validate_metadata(metadata: &v1::GetMetadataResponse) -> Result<(), ProtocolError> {
274	if metadata.protocol_version != PROTOCOL_VERSION
275		|| metadata.provider_kind.is_empty()
276		|| metadata.provider_version.is_empty()
277		|| metadata.max_message_bytes == 0
278		|| metadata.max_message_bytes as usize > MAX_MESSAGE_BYTES
279		|| metadata.max_events == 0
280		|| metadata.max_events as usize > MAX_EVENTS
281	{
282		return Err(ProtocolError::InvalidMetadata);
283	}
284	let mut seen_kinds = std::collections::BTreeSet::new();
285	for descriptor in &metadata.resource_kinds {
286		if descriptor.kind.is_empty()
287			|| descriptor.kind.chars().any(char::is_control)
288			|| !seen_kinds.insert(descriptor.kind.as_str())
289			|| descriptor.schema_json.is_empty()
290			|| descriptor.schema_json.len() > MAX_MESSAGE_BYTES
291			|| !descriptor.schema_digest.starts_with("sha256:")
292		{
293			return Err(ProtocolError::InvalidMetadata);
294		}
295		let digest = schema_digest(&descriptor.schema_json);
296		if descriptor.schema_digest != digest {
297			return Err(ProtocolError::InvalidMetadata);
298		}
299		// Schema must be parseable JSON object.
300		let value: serde_json::Value = serde_json::from_slice(&descriptor.schema_json)
301			.map_err(|_| ProtocolError::InvalidMetadata)?;
302		if !value.is_object() {
303			return Err(ProtocolError::InvalidMetadata);
304		}
305	}
306	Ok(())
307}
308
309/// Stable content digest for provider resource schema bytes.
310pub fn schema_digest(schema_json: &[u8]) -> String {
311	use sha2::{Digest, Sha256};
312	let hash = Sha256::digest(schema_json);
313	format!("sha256:{hash:x}")
314}
315
316#[cfg(test)]
317mod tests {
318	use super::*;
319
320	fn context() -> v1::RequestContext {
321		v1::RequestContext {
322			protocol_version: PROTOCOL_VERSION.to_string(),
323			provider_instance: "provider/synthetic".to_string(),
324			operation_id: "operation/test".to_string(),
325			deadline_unix_ms: 1,
326		}
327	}
328
329	#[test]
330	fn validates_protocol_context_json_and_metadata_bounds() {
331		validate_context(Some(&context())).expect("context validates");
332		validate_canonical_json(br#"{"kind":"synthetic"}"#).expect("JSON validates");
333		validate_metadata(&v1::GetMetadataResponse {
334			protocol_version: PROTOCOL_VERSION.to_string(),
335			provider_kind: "synthetic-fixture".to_string(),
336			provider_version: "0.1.0".to_string(),
337			capabilities: vec!["observe".to_string()],
338			max_message_bytes: MAX_MESSAGE_BYTES as u32,
339			max_events: MAX_EVENTS as u32,
340			resource_kinds: vec![],
341		})
342		.expect("metadata validates");
343	}
344
345	#[test]
346	fn rejects_incompatible_unbounded_or_malformed_inputs() {
347		let mut incompatible = context();
348		incompatible.protocol_version = "2.0".to_string();
349		assert!(matches!(
350			validate_context(Some(&incompatible)),
351			Err(ProtocolError::IncompatibleVersion(version)) if version == "2.0"
352		));
353		assert!(matches!(
354			validate_canonical_json(&vec![b'x'; MAX_MESSAGE_BYTES + 1]),
355			Err(ProtocolError::MessageTooLarge)
356		));
357		assert!(matches!(validate_canonical_json(b"{"), Err(ProtocolError::InvalidCanonicalJson)));
358	}
359
360	#[test]
361	fn redacts_sensitive_values_without_filtering_operational_output() {
362		let mut redactor = SensitiveTextRedactor::default();
363		let rendered = redactor.redact(
364			"setting up secrets...\n\
365			 copying path '/nix/store/abc' from 'https://cache.example'...\n\
366			 token=abc123 password: \"hunter2\"\n\
367			 source https://user:[email protected]/path\n\
368			 -----BEGIN OPENSSH PRIVATE KEY-----\n\
369			 private-body\n\
370			 -----END OPENSSH PRIVATE KEY-----\n",
371		);
372		assert!(rendered.contains("setting up secrets..."));
373		assert!(rendered.contains("copying path '/nix/store/abc' from 'https://cache.example'..."));
374		assert!(rendered.contains("token=[REDACTED] password: \"[REDACTED]\""));
375		assert!(rendered.contains("https://[REDACTED]@example.test/path"));
376		assert_eq!(redact_sensitive_text("ssh://deploy@builder"), "ssh://deploy@builder");
377		assert_eq!(rendered.matches("[REDACTED PRIVATE KEY MATERIAL]").count(), 3);
378		for secret in ["abc123", "hunter2", "user:pass", "private-body"] {
379			assert!(!rendered.contains(secret));
380		}
381	}
382}
383
384#[cfg(test)]
385mod truncation_tests {
386	use super::*;
387
388	#[test]
389	fn truncate_status_message_fits_header_budget() {
390		let huge = "x".repeat(MAX_STATUS_MESSAGE_BYTES * 4);
391		let out = truncate_status_message(&huge);
392		assert!(out.len() <= MAX_STATUS_MESSAGE_BYTES + 64);
393		assert!(out.contains("truncated"));
394	}
395
396	#[test]
397	fn truncate_status_keeps_tail_for_nix_style_logs() {
398		let head =
399			"these 266 derivations will be built:\n".to_string() + &"/nix/store/aaa.drv\n".repeat(500);
400		let tail = "error: build of '/nix/store/fail.drv' failed on 'ssh-ng://deploy@builder'";
401		let message = format!("{head}{tail}");
402		let out = truncate_status_message(&message);
403		assert!(out.contains("error: build of"), "{out}");
404		assert!(!out.starts_with("these 266"), "{out}");
405	}
406
407	#[test]
408	fn truncate_progress_preserves_short_messages() {
409		assert_eq!(truncate_progress_message("ok"), "ok");
410	}
411}
nxd_plugin_protocol/lib.rs

nxd_plugin_protocol/
lib.rs
