nxd_core/adapters/

plugin_fixture.rs

use crate::plugin_protocol::{
	MAX_EVENTS, MAX_MESSAGE_BYTES, PROTOCOL_VERSION, v1, validate_canonical_json, validate_context,
};
use serde_json::Value;
use std::collections::BTreeMap;
use std::fs;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Mutex;
use std::time::Duration;
use tokio_stream::Stream;
use tonic::{Request, Response, Status};
use v1::provider_service_server::ProviderService;

#[derive(Debug)]
pub struct FixtureProvider {
	mode: String,
	state: Mutex<BTreeMap<String, Value>>,
	/// Optional durable state path so multi-process plan/apply/verify shares state.
	state_file: Option<PathBuf>,
}

impl FixtureProvider {
	pub fn new(mode: impl Into<String>) -> Self {
		Self { mode: mode.into(), state: Mutex::new(BTreeMap::new()), state_file: None }
	}

	pub fn with_state_file(mode: impl Into<String>, state_file: PathBuf) -> Self {
		let state = if state_file.is_file() {
			fs::read_to_string(&state_file)
				.ok()
				.and_then(|content| serde_json::from_str(&content).ok())
				.unwrap_or_default()
		} else {
			BTreeMap::new()
		};
		Self { mode: mode.into(), state: Mutex::new(state), state_file: Some(state_file) }
	}

	async fn before_request(&self) {
		if self.mode == "hang" {
			tokio::time::sleep(Duration::from_secs(30)).await;
		}
	}

	fn context(context: Option<&v1::RequestContext>) -> Result<(), Status> {
		validate_context(context).map_err(|error| Status::failed_precondition(format!("{error:?}")))
	}

	fn resources(input: &[u8]) -> Result<BTreeMap<String, Value>, Status> {
		let value = validate_canonical_json(input)
			.map_err(|error| Status::invalid_argument(format!("{error:?}")))?;
		serde_json::from_value(
			value.get("resources").cloned().unwrap_or_else(|| Value::Object(Default::default())),
		)
		.map_err(|_| Status::invalid_argument("resources must be an object"))
	}

	fn persist_locked(&self, state: &BTreeMap<String, Value>) -> Result<(), Status> {
		let Some(path) = &self.state_file else {
			return Ok(());
		};
		let encoded =
			serde_json::to_string(state).map_err(|_| Status::internal("state serialization failed"))?;
		fs::write(path, encoded).map_err(|_| Status::internal("state file write failed"))?;
		Ok(())
	}

	/// Exercise the confidential Observe/Plan/Verify channel without ever
	/// reflecting the supplied value in a response or status message.
	fn require_session_secret(
		&self,
		secrets: &std::collections::HashMap<String, Vec<u8>>,
	) -> Result<(), Status> {
		if self.mode != "session-secret" {
			return Ok(());
		}
		match secrets.get("secret/session") {
			Some(value) if value.as_slice() == b"fixture-session-secret" => Ok(()),
			_ => Err(Status::permission_denied("required provider session secret was not supplied")),
		}
	}
}

#[tonic::async_trait]
impl ProviderService for FixtureProvider {
	async fn get_metadata(
		&self,
		request: Request<v1::GetMetadataRequest>,
	) -> Result<Response<v1::GetMetadataResponse>, Status> {
		self.before_request().await;
		Self::context(request.get_ref().context.as_ref())?;
		Ok(Response::new(v1::GetMetadataResponse {
			protocol_version: if self.mode == "incompatible" {
				"2.0".to_string()
			} else {
				PROTOCOL_VERSION.to_string()
			},
			provider_kind: "synthetic-fixture".to_string(),
			provider_version: "0.1.0".to_string(),
			capabilities: vec![
				"observe".to_string(),
				"plan".to_string(),
				"apply".to_string(),
				"verify".to_string(),
				"capture".to_string(),
				"cancel".to_string(),
			],
			max_message_bytes: MAX_MESSAGE_BYTES as u32,
			max_events: MAX_EVENTS as u32,
			resource_kinds: vec![],
		}))
	}

	async fn observe(
		&self,
		request: Request<v1::ObserveRequest>,
	) -> Result<Response<v1::ObserveResponse>, Status> {
		self.before_request().await;
		let request = request.into_inner();
		Self::context(request.context.as_ref())?;
		self.require_session_secret(&request.secrets)?;
		if self.mode == "malformed" {
			return Ok(Response::new(v1::ObserveResponse { observations_json: b"{".to_vec() }));
		}
		let state = self.state.lock().map_err(|_| Status::internal("state lock failed"))?;
		let observations_json = serde_json::to_vec(&serde_json::json!({"resources": &*state}))
			.map_err(|_| Status::internal("state serialization failed"))?;
		Ok(Response::new(v1::ObserveResponse { observations_json }))
	}

	async fn capture(
		&self,
		request: Request<v1::CaptureRequest>,
	) -> Result<Response<v1::CaptureResponse>, Status> {
		self.before_request().await;
		let request = request.into_inner();
		Self::context(request.context.as_ref())?;
		self.require_session_secret(&request.secrets)?;
		let selected = request.resource_ids.into_iter().collect::<std::collections::BTreeSet<_>>();
		let state = self.state.lock().map_err(|_| Status::internal("state lock failed"))?;
		let mut artifacts = Vec::new();
		for resource_id in selected {
			let Some(value) = state.get(&resource_id) else {
				continue;
			};
			let content =
				serde_json::to_vec(value).map_err(|_| Status::internal("capture serialization failed"))?;
			artifacts.push(v1::CaptureArtifact {
				resource_id: resource_id.clone(),
				relative_path: format!("resources/{}.json", resource_id.replace('/', "_")),
				content_digest: format!("sha256:{}", crate::plan::sha256_hex_bytes(&content)),
				content,
				mandatory: true,
				media_type: "application/json".into(),
			});
		}
		Ok(Response::new(v1::CaptureResponse { artifacts }))
	}

	async fn plan(
		&self,
		request: Request<v1::PlanRequest>,
	) -> Result<Response<v1::PlanResponse>, Status> {
		self.before_request().await;
		let request = request.into_inner();
		Self::context(request.context.as_ref())?;
		self.require_session_secret(&request.secrets)?;
		let desired = Self::resources(&request.desired_json)?;
		let observed = Self::resources(&request.observations_json)?;
		let mut actions = Vec::new();
		for (resource_id, value) in &desired {
			if observed.get(resource_id) != Some(value) {
				let operation = if observed.contains_key(resource_id) { "update" } else { "create" };
				let details = serde_json::json!({"value": value});
				let outputs = if self.mode == "confidential-output" {
					vec![v1::ActionOutputDecl {
						binding_id: format!("run/{operation}:{resource_id}/generated"),
						output_name: "generated".into(),
						classification: v1::OutputClassification::Confidential.into(),
						max_bytes: 4096,
						value_type: v1::BindingValueType::OpaqueBytes.into(),
						sink_binding: "secret/generated".into(),
						allow_overwrite: false,
					}]
				} else {
					vec![]
				};
				actions.push(v1::PlannedAction {
					action_id: format!("{operation}:{resource_id}"),
					resource_id: resource_id.clone(),
					operation: operation.to_string(),
					risk: v1::Risk::Reversible.into(),
					details_json: serde_json::to_vec(&details)
						.map_err(|_| Status::internal("action serialization failed"))?,
					secret_references: Vec::new(),
					depends_on: Vec::new(),
					contract: Some(v1::ActionContract {
						deployment_target: String::new(),
						target_placement: v1::TargetPlacement::Unspecified.into(),
						outputs,
						inputs: vec![],
						timeout_seconds: 600,
					}),
				});
			}
		}
		for resource_id in observed.keys().filter(|id| !desired.contains_key(*id)) {
			actions.push(v1::PlannedAction {
				action_id: format!("delete:{resource_id}"),
				resource_id: resource_id.clone(),
				operation: "delete".to_string(),
				risk: v1::Risk::Destructive.into(),
				details_json: b"{}".to_vec(),
				secret_references: Vec::new(),
				depends_on: Vec::new(),
				contract: Some(v1::ActionContract {
					deployment_target: String::new(),
					target_placement: v1::TargetPlacement::Unspecified.into(),
					outputs: vec![],
					inputs: vec![],
					timeout_seconds: 600,
				}),
			});
		}
		actions.sort_by(|left, right| left.action_id.cmp(&right.action_id));
		Ok(Response::new(v1::PlanResponse { actions }))
	}

	type ApplyStream =
		Pin<Box<dyn Stream<Item = Result<v1::ApplyResponse, Status>> + Send + 'static>>;

	async fn apply(
		&self,
		request: Request<v1::ApplyRequest>,
	) -> Result<Response<Self::ApplyStream>, Status> {
		self.before_request().await;
		let request = request.into_inner();
		Self::context(request.context.as_ref())?;
		let action = request.action.ok_or_else(|| Status::invalid_argument("action is required"))?;
		let mut state = self.state.lock().map_err(|_| Status::internal("state lock failed"))?;
		match action.operation.as_str() {
			"create" | "update" => {
				let details = validate_canonical_json(&action.details_json)
					.map_err(|error| Status::invalid_argument(format!("{error:?}")))?;
				let value = details
					.get("value")
					.cloned()
					.ok_or_else(|| Status::invalid_argument("action value is required"))?;
				state.insert(action.resource_id.clone(), value);
			}
			"delete" => {
				state.remove(&action.resource_id);
			}
			_ => return Err(Status::invalid_argument("unsupported operation")),
		}
		self.persist_locked(&state)?;
		drop(state);
		// Flood past the supervisor streamed-progress guard (MAX_EVENTS * 64).
		let count =
			if self.mode == "flood" { MAX_EVENTS.saturating_mul(64).saturating_add(2) } else { 2 };
		let mode = self.mode.clone();
		let events = (0..count).map(move |index| {
			let confidential_outputs = if mode == "confidential-output" && index + 1 == count {
				[("generated".to_string(), b"fixture-confidential".to_vec())].into_iter().collect()
			} else {
				Default::default()
			};
			let safe_message = if mode == "unsafe-progress" {
				"ordinary provider detail\ntoken=field-secret\nopaque-action-secret".to_string()
			} else {
				"synthetic fixture event".to_string()
			};
			Ok(v1::ApplyResponse {
				sequence: (index + 1) as u32,
				phase: if index + 1 == count {
					v1::EventPhase::Completed.into()
				} else {
					v1::EventPhase::Progress.into()
				},
				action_id: action.action_id.clone(),
				safe_message,
				error: None,
				detail: false,
				confidential_outputs,
				public_outputs: Default::default(),
			})
		});
		Ok(Response::new(Box::pin(tokio_stream::iter(events))))
	}

	async fn verify(
		&self,
		request: Request<v1::VerifyRequest>,
	) -> Result<Response<v1::VerifyResponse>, Status> {
		self.before_request().await;
		let request = request.into_inner();
		Self::context(request.context.as_ref())?;
		self.require_session_secret(&request.secrets)?;
		let action = request.action.ok_or_else(|| Status::invalid_argument("action is required"))?;
		let state = self.state.lock().map_err(|_| Status::internal("state lock failed"))?;
		let satisfied = match action.operation.as_str() {
			"create" | "update" => state.contains_key(&action.resource_id),
			"delete" => !state.contains_key(&action.resource_id),
			_ => false,
		};
		Ok(Response::new(v1::VerifyResponse {
			satisfied,
			observation_json: serde_json::to_vec(&serde_json::json!({
				"resource": state.get(&action.resource_id)
			}))
			.map_err(|_| Status::internal("verification serialization failed"))?,
			error: None,
		}))
	}

	async fn cancel(
		&self,
		request: Request<v1::CancelRequest>,
	) -> Result<Response<v1::CancelResponse>, Status> {
		Self::context(request.get_ref().context.as_ref())?;
		Ok(Response::new(v1::CancelResponse { accepted: true }))
	}
}