feat(tools): the ingest credential — the event door, write-only

The agent token is deliberately barred from /v1/ingest: the events
are the inbound channel, and the model must not feed it itself. But
the same bar closes the door on every honest integration too — a
sidecar poller with events to hand in had to hold the human token,
which is wider than its job. This adds the missing class: a third
bearer, the ingest token, answering the event door and nothing else.

`ingest_allowed` is the whole credential in one function — one
route row, `POST /v1/ingest` — symmetric with `agent_allowed` and
suite-covered the same way: the mock and the daemon's real router
both answer the new checks (a push succeeds with the contract header;
a watch read and a consent answer are 403). The middleware classifier
gains the third arm, boot refuses an ingest token equal to either
other credential — one value playing two roles silently widens the
scoped one — and the daemon provisions it to tool shells as
`RHEMA_INGEST_TOKEN`, so a tool that only pushes events never sees
the wider agent token at all.

The door is the first half of the extensions story (rhema-tools spec
§8): a self-timed sidecar is now a tier-1 tool with a credential
narrow enough to hold.
This commit is contained in:
vhaudiquet committed 2026-10-24 01:17:43 +02:00
1 parent 98bc0fd623
commit f96f8c4c1d
10 files changed
+210 -18

No files matched your search

+61
View File
@@ -62,6 +62,14 @@ pub enum ConfigError {
"agent_token is the same value as token; the agent token must be its own secret, scoped to the tool routes" "agent_token is the same value as token; the agent token must be its own secret, scoped to the tool routes"
)] )]
SameAgentToken, SameAgentToken,
/// The ingest credential is the same value as the human token or
/// the agent token: boot refuses, because the ingest credential's
/// safety is that it is write-only and the event door only — a
/// value shared with a wider class silently widens it.
#[error(
"ingest_token is the same value as another credential; the ingest token must be its own secret, write-only on the event door"
)]
IngestTokenCollision,
/// One watch is unusable; the daemon refuses to start rather than /// One watch is unusable; the daemon refuses to start rather than
/// poll a half-configured watch set. /// poll a half-configured watch set.
#[error("watch #{index} ({forge} {repo}): {reason}")] #[error("watch #{index} ({forge} {repo}): {reason}")]
@@ -304,6 +312,14 @@ pub struct ClawConfig {
/// A directory of client-tool binaries (the rhema-tools project's) /// A directory of client-tool binaries (the rhema-tools project's)
/// layered onto the agent's shell PATH. Unset leaves PATH alone. /// layered onto the agent's shell PATH. Unset leaves PATH alone.
pub tools_bin: Option<PathBuf>, pub tools_bin: Option<PathBuf>,
/// The ingest credential: a third bearer class, handed to
/// integrations (a sidecar poller, an OS cron job) so they can hand
/// events to `/v1/ingest` — and do nothing else. Write-only by
/// construction: the middleware keeps it to the event door, and it
/// must differ from both other tokens (boot refuses otherwise).
/// Unset closes the door to credentials; the human token can still
/// ingest.
pub ingest_token: Option<String>,
/// What to watch — the optional seed for the persistent watch table. /// What to watch — the optional seed for the persistent watch table.
/// Imported once, when the table is still empty; after that the /// Imported once, when the table is still empty; after that the
/// table rules, and watches are managed in chat or over HTTP. /// table rules, and watches are managed in chat or over HTTP.
@@ -337,6 +353,7 @@ impl Default for ClawConfig {
token: None, token: None,
agent_token: None, agent_token: None,
tools_bin: None, tools_bin: None,
ingest_token: None,
transcript: None, transcript: None,
watches: Vec::new(), watches: Vec::new(),
calendars: Vec::new(), calendars: Vec::new(),
@@ -471,6 +488,16 @@ impl ClawConfig {
return Err(ConfigError::SameAgentToken); return Err(ConfigError::SameAgentToken);
} }
} }
// The ingest credential's safety is that it is write-only and
// one route deep; a value shared with a wider class silently
// widens it, so boot refuses the collision with either token.
if let Some(ingest_token) = self.ingest_token.as_deref().filter(|t| !t.is_empty()) {
if Some(ingest_token) == self.token.as_deref()
|| Some(ingest_token) == self.agent_token.as_deref()
{
return Err(ConfigError::IngestTokenCollision);
}
}
for (index, watch) in self.watches.iter().enumerate() { for (index, watch) in self.watches.iter().enumerate() {
let fail = |reason: String| ConfigError::Watch { let fail = |reason: String| ConfigError::Watch {
index, index,
@@ -609,6 +636,40 @@ impl ClawConfig {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
#[test]
fn the_ingest_token_must_differ_from_both_other_credentials() {
let config = ClawConfig {
token: Some("same".to_owned()),
ingest_token: Some("same".to_owned()),
..ClawConfig::default()
};
assert!(
matches!(config.validate(), Err(ConfigError::IngestTokenCollision)),
"an ingest credential sharing the human token is refused"
);
let config = ClawConfig {
token: Some("human".to_owned()),
agent_token: Some("same".to_owned()),
..config
};
assert!(
matches!(config.validate(), Err(ConfigError::IngestTokenCollision)),
"an ingest credential sharing the agent token is refused"
);
let config = ClawConfig {
agent_token: Some("agent".to_owned()),
ingest_token: Some("ingest".to_owned()),
..config
};
assert!(config.validate().is_ok(), "three distinct tokens pass");
// An empty ingest token is no token: no collision to refuse.
let blank = ClawConfig {
ingest_token: Some(String::new()),
..ClawConfig::default()
};
assert!(blank.validate().is_ok(), "a blank ingest token is unset");
}
#[test] #[test]
fn the_agent_token_must_differ_from_the_human_token() { fn the_agent_token_must_differ_from_the_human_token() {
let config = ClawConfig { let config = ClawConfig {
+7
View File
@@ -242,6 +242,12 @@ impl Daemon {
); );
tool_env.insert("RHEMA_TOKEN".to_owned(), agent_token.clone()); tool_env.insert("RHEMA_TOKEN".to_owned(), agent_token.clone());
} }
// The ingest credential rides its own variable: a tool asks
// for exactly the class it needs, and a tool that only pushes
// events never sees the wider agent token at all.
if let Some(ingest_token) = &config.ingest_token {
tool_env.insert("RHEMA_INGEST_TOKEN".to_owned(), ingest_token.clone());
}
if let Some(tools_bin) = &config.tools_bin { if let Some(tools_bin) = &config.tools_bin {
let inherited = std::env::var("PATH").unwrap_or_default(); let inherited = std::env::var("PATH").unwrap_or_default();
tool_env.insert( tool_env.insert(
@@ -399,6 +405,7 @@ impl Daemon {
phone, phone,
config.token.clone(), config.token.clone(),
config.agent_token.clone(), config.agent_token.clone(),
config.ingest_token.clone(),
); );
Ok(Self { Ok(Self {
+50 -8
View File
@@ -95,6 +95,9 @@ pub struct ClawState {
/// answering only the contract's data routes (see /// answering only the contract's data routes (see
/// [`require_auth`]). /// [`require_auth`]).
agent_token: Option<String>, agent_token: Option<String>,
/// The ingest credential integrations hold: a third bearer
/// answering the event door only (see [`require_auth`]).
ingest_token: Option<String>,
/// The generic sources `/v1/ingest` has registered on the /// The generic sources `/v1/ingest` has registered on the
/// dispatcher, so a source name registers exactly once however many /// dispatcher, so a source name registers exactly once however many
/// events race in through it. /// events race in through it.
@@ -128,6 +131,7 @@ impl ClawState {
phone: Arc<crate::phone::PhoneStore>, phone: Arc<crate::phone::PhoneStore>,
token: Option<String>, token: Option<String>,
agent_token: Option<String>, agent_token: Option<String>,
ingest_token: Option<String>,
) -> Self { ) -> Self {
Self { Self {
dispatcher, dispatcher,
@@ -148,6 +152,7 @@ impl ClawState {
phone, phone,
token, token,
agent_token, agent_token,
ingest_token,
generic_sources: Arc::new(Mutex::new(BTreeSet::new())), generic_sources: Arc::new(Mutex::new(BTreeSet::new())),
} }
} }
@@ -312,28 +317,39 @@ fn tokens_match(presented: &str, expected: &str) -> bool {
diff == 0 diff == 0
} }
/// The bearer check, in two classes. With neither token configured the /// The bearer check, in three classes. With no token configured the
/// surface is open, as before. The human token answers every `/v1` /// surface is open, as before. The human token answers every `/v1`
/// route. The agent token — what the client tools hold — answers only /// route. The agent token — what the client tools hold — answers only
/// the data routes the tool contract allowlists, and only with the /// the data routes the tool contract allowlists, and only with the
/// contract header at the matching version: a tool cannot reach /// contract header at the matching version: a tool cannot reach
/// consents, sessions, or ingest, so a leaked agent token cannot /// consents, sessions, or ingest, so a leaked agent token cannot
/// answer the agent's own `ask_secret` (rhema-tools spec §5). Any /// answer the agent's own `ask_secret` (rhema-tools spec §5). The
/// other bearer is a 401. `/healthz` never reaches this. /// ingest credential — what integrations hold — answers the event
/// door and nothing else: write-only by construction, it can hand
/// events in and read nothing back. Any other bearer is a 401.
/// `/healthz` never reaches this.
async fn require_auth(State(state): State<ClawState>, request: Request, next: Next) -> Response { async fn require_auth(State(state): State<ClawState>, request: Request, next: Next) -> Response {
let presented = request let presented = request
.headers() .headers()
.get(axum::http::header::AUTHORIZATION) .get(axum::http::header::AUTHORIZATION)
.and_then(|value| value.to_str().ok()) .and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer ")); .and_then(|value| value.strip_prefix("Bearer "));
let class = match (&state.token, &state.agent_token, presented) { let class = match (
&state.token,
&state.agent_token,
&state.ingest_token,
presented,
) {
// Nothing configured: the open surface, bearers ignored. // Nothing configured: the open surface, bearers ignored.
(None, None, _) => Auth::Human, (None, None, None, _) => Auth::Human,
(Some(expected), _, Some(presented)) if tokens_match(presented, expected) => Auth::Human, (Some(expected), _, _, Some(presented)) if tokens_match(presented, expected) => Auth::Human,
(_, Some(expected), Some(presented)) if tokens_match(presented, expected) => Auth::Agent, (_, Some(expected), _, Some(presented)) if tokens_match(presented, expected) => Auth::Agent,
(_, _, Some(expected), Some(presented)) if tokens_match(presented, expected) => {
Auth::Ingest
}
// A token configured and no bearer, or a bearer matching // A token configured and no bearer, or a bearer matching
// neither. // neither.
(_, _, _) => Auth::Invalid, (_, _, _, _) => Auth::Invalid,
}; };
match class { match class {
Auth::Human => next.run(request).await, Auth::Human => next.run(request).await,
@@ -370,6 +386,32 @@ async fn require_auth(State(state): State<ClawState>, request: Request, next: Ne
.into_response() .into_response()
} }
} }
Auth::Ingest => {
let version = request
.headers()
.get(CONTRACT_HEADER)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| value == CONTRACT_VERSION.to_string());
if !version {
return (
StatusCode::CONFLICT,
Json(json!({"error": format!(
"ingest calls carry the {CONTRACT_HEADER} header at version {CONTRACT_VERSION}"
)})),
)
.into_response();
}
if rhema_tools_contract::ingest_allowed(request.method().as_str(), request.uri().path())
{
next.run(request).await
} else {
(
StatusCode::FORBIDDEN,
Json(json!({"error": "this route is not the ingest token's"})),
)
.into_response()
}
}
} }
} }
+1
View File
@@ -70,6 +70,7 @@ fn config(server: &MockServer, dir: &tempfile::TempDir) -> ClawConfig {
agent_token: None, agent_token: None,
tools_bin: None, tools_bin: None,
transcript: None, transcript: None,
ingest_token: None,
watches: vec![Watch { watches: vec![Watch {
forge: ForgeKind::GitHub, forge: ForgeKind::GitHub,
repo: "acme/widgets".to_owned(), repo: "acme/widgets".to_owned(),
+1
View File
@@ -58,6 +58,7 @@ fn config(server: &MockServer, dir: &tempfile::TempDir) -> ClawConfig {
agent_token: None, agent_token: None,
tools_bin: None, tools_bin: None,
transcript: None, transcript: None,
ingest_token: None,
watches: vec![Watch { watches: vec![Watch {
forge: ForgeKind::GitHub, forge: ForgeKind::GitHub,
repo: "acme/widgets".to_owned(), repo: "acme/widgets".to_owned(),
+5
View File
@@ -63,6 +63,7 @@ fn config(dir: &tempfile::TempDir, token: Option<&str>) -> ClawConfig {
token: token.map(str::to_owned), token: token.map(str::to_owned),
agent_token: None, agent_token: None,
tools_bin: None, tools_bin: None,
ingest_token: None,
transcript: None, transcript: None,
watches: Vec::new(), watches: Vec::new(),
calendars: Vec::new(), calendars: Vec::new(),
@@ -1200,6 +1201,7 @@ async fn the_surface_speaks_the_tool_contract() {
let dir = tempfile::tempdir().expect("tempdir"); let dir = tempfile::tempdir().expect("tempdir");
let mut config = config(&dir, Some("human-token")); let mut config = config(&dir, Some("human-token"));
config.agent_token = Some("agent-token".to_owned()); config.agent_token = Some("agent-token".to_owned());
config.ingest_token = Some("ingest-token".to_owned());
let daemon = Daemon::build( let daemon = Daemon::build(
&config, &config,
mock_registry(Arc::new(MockProvider::new(Vec::new()))), mock_registry(Arc::new(MockProvider::new(Vec::new()))),
@@ -1226,6 +1228,9 @@ async fn the_surface_speaks_the_tool_contract() {
rhema_tools_contract::Auth::Invalid => { rhema_tools_contract::Auth::Invalid => {
builder.header(reqwest::header::AUTHORIZATION, "Bearer a-stranger") builder.header(reqwest::header::AUTHORIZATION, "Bearer a-stranger")
} }
rhema_tools_contract::Auth::Ingest => {
builder.header(reqwest::header::AUTHORIZATION, "Bearer ingest-token")
}
}; };
let builder = match request.contract { let builder = match request.contract {
Some(version) => builder.header("x-rhema-contract", version.to_string()), Some(version) => builder.header("x-rhema-contract", version.to_string()),
+39 -6
View File
@@ -9,12 +9,12 @@ use crate::{ApiError, Auth, CallRequest, CallResponse, now_millis};
/// The first failure is returned, named by the check that failed; `Ok` /// The first failure is returned, named by the check that failed; `Ok`
/// means the transport speaks the contract. /// means the transport speaks the contract.
/// ///
/// The suite assumes a daemon configured with both tokens — a human /// The suite assumes a daemon configured with the tokens — the human
/// one and an agent one — because the agent-token class is the /// one, the agent one, and the ingest one — because the scoped
/// contract's load-bearing guarantee. A daemon running an open surface /// credential classes are the contract's load-bearing guarantee. A
/// (no tokens) cannot pass it, by design: the credential section of /// daemon running an open surface (no tokens) cannot pass it, by
/// the spec (§5) makes the scoped agent token a precondition of the /// design: the credential section of the spec (§5) makes the scoped
/// client tools. /// agent token a precondition of the client tools.
pub async fn conformance<F, Fut>(call: F) -> Result<(), String> pub async fn conformance<F, Fut>(call: F) -> Result<(), String>
where where
F: Fn(CallRequest) -> Fut, F: Fn(CallRequest) -> Fut,
@@ -335,6 +335,39 @@ where
) )
.await?; .await?;
// — The ingest credential: the event door, write-only —
ctx.expect_success(
"the ingest credential hands one event in",
req(
"POST",
"/v1/ingest",
Auth::Ingest,
Some(1),
Some(serde_json::json!({
"source": "ciwatch", "key": "ci:door/1", "payload": {"text": "ping"}
})),
),
)
.await?;
ctx.expect(
"the ingest credential cannot read the watches",
req("GET", "/v1/watches", Auth::Ingest, Some(1), None),
403,
)
.await?;
ctx.expect(
"the ingest credential cannot answer a consent",
req(
"POST",
"/v1/consents/c-1",
Auth::Ingest,
Some(1),
Some(serde_json::json!({"grant": true})),
),
403,
)
.await?;
// — The one error shape — // — The one error shape —
let missing = ctx let missing = ctx
.expect( .expect(
+20
View File
@@ -77,6 +77,23 @@ pub fn agent_allowed(method: &str, path: &str) -> bool {
.any(|(route_method, pattern)| *route_method == method && match_pattern(pattern, &segments)) .any(|(route_method, pattern)| *route_method == method && match_pattern(pattern, &segments))
} }
/// Whether the ingest credential may call this route: the event door
/// and nothing else. The ingest class is write-only by construction —
/// it can hand events in, and read nothing back — so its allowlist is
/// one row, and `false` for every method it does not name.
pub fn ingest_allowed(method: &str, path: &str) -> bool {
// The routes are daemon-side (`/v1/...`), but a daemon whose auth
// middleware runs inside its `/v1` nest sees the stripped path —
// accept both spellings, one allowlist.
let path = if path.starts_with("/v1") {
path
} else {
&format!("/v1{path}")
};
let segments: Vec<&str> = path.trim_matches('/').split('/').collect();
method == "POST" && match_pattern(&["v1", "ingest"], &segments)
}
/// Matches one allowlist pattern against a request's path segments. /// Matches one allowlist pattern against a request's path segments.
fn match_pattern(pattern: &[&str], segments: &[&str]) -> bool { fn match_pattern(pattern: &[&str], segments: &[&str]) -> bool {
match (pattern.first(), segments.first()) { match (pattern.first(), segments.first()) {
@@ -192,6 +209,9 @@ pub enum Auth {
Human, Human,
/// The agent token: the [`agent_allowed`] routes only. /// The agent token: the [`agent_allowed`] routes only.
Agent, Agent,
/// The ingest credential: the event door ([`ingest_allowed`])
/// only, and nothing readable.
Ingest,
/// A bearer that matches neither configured token. /// A bearer that matches neither configured token.
Invalid, Invalid,
} }
+20 -2
View File
@@ -8,6 +8,7 @@ use serde_json::{Value, json};
use std::sync::PoisonError; use std::sync::PoisonError;
use crate::ingest_allowed;
use crate::{ use crate::{
Auth, CONTRACT_VERSION, CallRequest, CallResponse, NotificationOut, NotificationPush, TaskAdd, Auth, CONTRACT_VERSION, CallRequest, CallResponse, NotificationOut, NotificationPush, TaskAdd,
TaskOut, agent_allowed, now_millis, TaskOut, agent_allowed, now_millis,
@@ -19,6 +20,7 @@ use crate::{
pub struct MockClaw { pub struct MockClaw {
human_token: Option<String>, human_token: Option<String>,
agent_token: Option<String>, agent_token: Option<String>,
ingest_token: Option<String>,
notifications: Mutex<Vec<NotificationOut>>, notifications: Mutex<Vec<NotificationOut>>,
tasks: Mutex<Vec<TaskOut>>, tasks: Mutex<Vec<TaskOut>>,
watches: Mutex<Vec<Value>>, watches: Mutex<Vec<Value>>,
@@ -29,10 +31,15 @@ pub struct MockClaw {
impl MockClaw { impl MockClaw {
/// A mock with the daemon's two tokens; `None` means that class is /// A mock with the daemon's two tokens; `None` means that class is
/// unconfigured and its bearers are strangers. /// unconfigured and its bearers are strangers.
pub fn new(human_token: Option<&str>, agent_token: Option<&str>) -> Self { pub fn new(
human_token: Option<&str>,
agent_token: Option<&str>,
ingest_token: Option<&str>,
) -> Self {
Self { Self {
human_token: human_token.map(str::to_owned), human_token: human_token.map(str::to_owned),
agent_token: agent_token.map(str::to_owned), agent_token: agent_token.map(str::to_owned),
ingest_token: ingest_token.map(str::to_owned),
notifications: Mutex::new(Vec::new()), notifications: Mutex::new(Vec::new()),
tasks: Mutex::new(Vec::new()), tasks: Mutex::new(Vec::new()),
watches: Mutex::new(Vec::new()), watches: Mutex::new(Vec::new()),
@@ -46,7 +53,8 @@ impl MockClaw {
pub async fn call(&self, request: CallRequest) -> Result<CallResponse, String> { pub async fn call(&self, request: CallRequest) -> Result<CallResponse, String> {
// The bearer gate. With no tokens configured the surface is // The bearer gate. With no tokens configured the surface is
// open; with either, a stranger is a 401. // open; with either, a stranger is a 401.
let open = self.human_token.is_none() && self.agent_token.is_none(); let open =
self.human_token.is_none() && self.agent_token.is_none() && self.ingest_token.is_none();
if !open && matches!(request.auth, Auth::None | Auth::Invalid) { if !open && matches!(request.auth, Auth::None | Auth::Invalid) {
return Ok(self.error(401, "unauthorized")); return Ok(self.error(401, "unauthorized"));
} }
@@ -59,6 +67,16 @@ impl MockClaw {
return Ok(self.error(403, "this route is not the agent token's")); return Ok(self.error(403, "this route is not the agent token's"));
} }
} }
// The ingest class: the event door, contract version required —
// an integration speaks the same handshake the tools do.
if request.auth == Auth::Ingest {
if request.contract != Some(CONTRACT_VERSION) {
return Ok(self.error(409, "contract version mismatch"));
}
if !ingest_allowed(request.method, &request.path) {
return Ok(self.error(403, "this route is not the ingest token's"));
}
}
let segments: Vec<&str> = request.path.trim_matches('/').split('/').collect(); let segments: Vec<&str> = request.path.trim_matches('/').split('/').collect();
match (request.method, segments.as_slice()) { match (request.method, segments.as_slice()) {
+6 -2
View File
@@ -6,7 +6,11 @@ use rhema_tools_contract::{CallRequest, MockClaw, conformance};
#[tokio::test] #[tokio::test]
async fn the_suite_passes_against_the_mock_router() { async fn the_suite_passes_against_the_mock_router() {
let mock = MockClaw::new(Some("human-token"), Some("agent-token")); let mock = MockClaw::new(
Some("human-token"),
Some("agent-token"),
Some("ingest-token"),
);
let result = let result =
conformance(|request: CallRequest| async { Ok(mock.call(request).await.unwrap()) }).await; conformance(|request: CallRequest| async { Ok(mock.call(request).await.unwrap()) }).await;
assert_eq!(result, Ok(()), "the mock router speaks the contract"); assert_eq!(result, Ok(()), "the mock router speaks the contract");
@@ -16,7 +20,7 @@ async fn the_suite_passes_against_the_mock_router() {
async fn an_open_mock_fails_the_suite() { async fn an_open_mock_fails_the_suite() {
// No tokens: the agent-token guarantees the suite exists to check // No tokens: the agent-token guarantees the suite exists to check
// are gone. The suite must say so, not sail through. // are gone. The suite must say so, not sail through.
let mock = MockClaw::new(None, None); let mock = MockClaw::new(None, None, None);
let result = let result =
conformance(|request: CallRequest| async { Ok(mock.call(request).await.unwrap()) }).await; conformance(|request: CallRequest| async { Ok(mock.call(request).await.unwrap()) }).await;
assert!(result.is_err(), "an open surface cannot pass the suite"); assert!(result.is_err(), "an open surface cannot pass the suite");