onetaskgraph_core/subprocess/
plugin.rs

1//! Configuring a source that is another program.
2
3use std::collections::BTreeMap;
4use std::num::NonZeroU64;
5
6use onetaskgraph_plugin_api::{SecretResolver, SourceError, SourceName, SourcePlugin, TaskSource};
7use schemars::{JsonSchema, Schema, schema_for};
8use secrecy::ExposeSecret;
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11
12use super::source::{RequestDeadline, SubprocessSource};
13use crate::secrets::CredentialName;
14
15/// The name a configuration document's `plugin:` field names this kind by.
16pub(crate) const KIND: &str = "subprocess";
17
18/// How to run a plugin that speaks `docs/plugin-protocol.md`.
19///
20/// `settings` is the seam that keeps this one plugin general: it is handed to the child
21/// as its `config:` block verbatim, so what a Python source needs and what a Rust one
22/// needs are that child's business rather than a field of this schema. Which is also why
23/// the credentials a child may see are *named* here rather than inherited: §3.1 forbids a
24/// plugin reading credentials from its own environment, and forwarding the engine's whole
25/// environment would hand every plugin every secret on the host.
26#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
27#[serde(deny_unknown_fields)]
28pub struct SubprocessConfig {
29    /// The program to run.
30    pub command: Program,
31    /// Arguments to run it with.
32    #[serde(default)]
33    pub args: Vec<String>,
34    /// The environment variables whose resolved values the handshake forwards.
35    #[serde(default)]
36    pub secrets: Vec<CredentialName>,
37    /// This source's own settings, handed to the child verbatim.
38    #[serde(default)]
39    pub settings: Value,
40    /// Defaults to 30 seconds and cannot be zero.
41    #[serde(default = "default_deadline_ms")]
42    pub deadline_ms: NonZeroU64,
43}
44
45fn default_deadline_ms() -> NonZeroU64 {
46    RequestDeadline::DEFAULT.milliseconds()
47}
48
49/// The program that serves a source: a name that is not blank.
50///
51/// A newtype rather than a `String` checked on the way past, because the check has to hold
52/// wherever one of these comes from. A blank command is not a source that fails later; it
53/// is a source that was never configured, and the difference is the difference between a
54/// sentence naming the field to fill in and a spawn error about an empty path.
55#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
56#[serde(into = "String", try_from = "String")]
57// schemars does not read `serde(into)`, and this is a string on the wire.
58#[schemars(with = "String")]
59pub struct Program(String);
60
61impl Program {
62    /// `command` when it names something, and nothing otherwise.
63    #[must_use]
64    pub fn new(command: &str) -> Option<Self> {
65        (!command.trim().is_empty()).then(|| Self(command.to_owned()))
66    }
67
68    /// The program itself.
69    #[must_use]
70    pub fn as_str(&self) -> &str {
71        &self.0
72    }
73}
74
75impl TryFrom<String> for Program {
76    type Error = String;
77
78    fn try_from(value: String) -> Result<Self, Self::Error> {
79        Self::new(&value)
80            .ok_or_else(|| "`command` must name the program that serves this source".to_owned())
81    }
82}
83
84impl From<Program> for String {
85    fn from(value: Program) -> Self {
86        value.0
87    }
88}
89
90/// The factory a configuration reaches through `plugin: subprocess`.
91#[derive(Debug, Clone, Copy, Default)]
92pub struct Plugin;
93
94impl SourcePlugin for Plugin {
95    fn kind(&self) -> &'static str {
96        KIND
97    }
98
99    fn config_schema(&self) -> Schema {
100        schema_for!(SubprocessConfig)
101    }
102
103    fn build(
104        &self,
105        name: &SourceName,
106        config: &Value,
107        secrets: &dyn SecretResolver,
108    ) -> Result<Box<dyn TaskSource>, SourceError> {
109        let config: SubprocessConfig =
110            serde_json::from_value(config.clone()).map_err(|error| SourceError::Config {
111                message: format!("source {name}: {error}"),
112            })?;
113        let forwarded = resolve_named(name, &config.secrets, secrets)?;
114        SubprocessSource::connect_with_deadline(
115            config.command.as_str(),
116            &config.args,
117            name,
118            &config.settings,
119            forwarded,
120            RequestDeadline::from_millis(config.deadline_ms),
121        )
122        .map(|source| Box::new(source) as Box<dyn TaskSource>)
123    }
124}
125
126/// The values of exactly the variables this configuration names, and nothing else.
127///
128/// A named variable nothing defines is refused here rather than forwarded as absent: the
129/// plugin asked for it, so a run that spawned anyway would fail later inside the child
130/// with a message about a credential, and the thing the user has to fix is on this side.
131fn resolve_named(
132    name: &SourceName,
133    named: &[CredentialName],
134    secrets: &dyn SecretResolver,
135) -> Result<BTreeMap<String, String>, SourceError> {
136    let mut forwarded = BTreeMap::new();
137    for variable in named {
138        let value = secrets
139            .get(variable.as_str())
140            .ok_or_else(|| SourceError::Auth {
141                message: format!(
142                    "source {name}: nothing defines {variable}, which this source's \
143                     `secrets` names; export it, or add it to the credentials file"
144                ),
145            })?;
146        forwarded.insert(
147            variable.as_str().to_owned(),
148            value.expose_secret().to_owned(),
149        );
150    }
151    Ok(forwarded)
152}
onetaskgraph_core/subprocess/plugin.rs

onetaskgraph_core/subprocess/
plugin.rs
