feat(acp): add inbound author gate (--respond-to) (#207)

This commit is contained in:
tlongwell-block
2026-04-01 12:14:30 -04:00
committed by GitHub
parent 8259af0278
commit 9662adf8d2
3 changed files with 488 additions and 36 deletions

View File

@@ -122,9 +122,48 @@ All configuration is via environment variables (or CLI flags — every env var h
| `--heartbeat-prompt` | `SPROUT_ACP_HEARTBEAT_PROMPT` | (built-in) | Custom heartbeat prompt text. Conflicts with `--heartbeat-prompt-file`. |
| `--heartbeat-prompt-file` | `SPROUT_ACP_HEARTBEAT_PROMPT_FILE` | — | Read heartbeat prompt from a file. Conflicts with `--heartbeat-prompt`. |
### Inbound Author Gate
Controls which authors' events the harness forwards to the agent. Events from disallowed authors are silently dropped before reaching subscription rules.
| Flag | Env Var | Default | Description |
|------|---------|---------|-------------|
| `--respond-to` | `SPROUT_ACP_RESPOND_TO` | `owner-only` | Author gate mode: `owner-only`, `allowlist`, `anyone`, `nobody`. |
| `--respond-to-allowlist` | `SPROUT_ACP_RESPOND_TO_ALLOWLIST` | — | Comma-separated 64-char hex pubkeys (required when mode is `allowlist`). Owner is always implicitly included. |
**Modes:**
| Mode | Behavior |
|------|----------|
| `owner-only` | Forward only events from the agent's registered owner. If no owner is set, all events are dropped until the owner is resolved. |
| `allowlist` | Forward events from the listed pubkeys plus the owner. |
| `anyone` | Forward all events (no author filtering). |
| `nobody` | Drop all inbound events. Agent only acts on heartbeat prompts. |
The gate applies to **all** inbound events — @mentions, DMs, thread replies, and any event delivered by the relay. The `!shutdown` command is checked **before** the gate, so the owner can always shut down the agent regardless of mode.
> **Note:** The default mode is `owner-only`. Agents without a registered `agent_owner_pubkey` will not respond to any events until the owner is resolved. Set `--respond-to anyone` to disable the gate entirely.
**Examples:**
```bash
# Default: only respond to owner
sprout-acp
# Respond to a team of three users (owner always included automatically)
sprout-acp --respond-to allowlist \
--respond-to-allowlist "abc123...64hex,def456...64hex,789abc...64hex"
# Respond to anyone (open agent)
sprout-acp --respond-to anyone
# Broadcast-only: post on heartbeat, ignore all inbound events
sprout-acp --respond-to nobody --heartbeat-interval 300
```
### Configuration Examples
**Single agent, no heartbeat (default — backward compatible):**
**Single agent, no heartbeat (default):**
```bash
sprout-acp
```

View File

@@ -3,7 +3,7 @@
//! CLI-first: every option is a CLI flag with env var fallback.
//! Config file (TOML) for complex subscription rules.
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use clap::Parser;
@@ -42,6 +42,32 @@ pub enum DedupMode {
Queue,
}
/// Inbound author gate: which authors' events the harness forwards to the agent.
///
/// - `owner-only` — only the agent's registered owner (default).
/// - `allowlist` — owner + explicit pubkey list (`--respond-to-allowlist`).
/// - `anyone` — all events forwarded (no author filtering).
/// - `nobody` — all events dropped (proactive/heartbeat-only mode).
#[derive(Debug, Clone, Default, PartialEq, clap::ValueEnum)]
pub enum RespondTo {
#[default]
OwnerOnly,
Allowlist,
Anyone,
Nobody,
}
impl std::fmt::Display for RespondTo {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::OwnerOnly => f.write_str("owner-only"),
Self::Allowlist => f.write_str("allowlist"),
Self::Anyone => f.write_str("anyone"),
Self::Nobody => f.write_str("nobody"),
}
}
}
/// Permission mode for agents that support `session/set_config_option` with
/// `configId: "mode"` (e.g. `claude-agent-acp`).
///
@@ -280,6 +306,21 @@ pub struct CliArgs {
value_enum
)]
pub permission_mode: PermissionMode,
/// Inbound author gate: which authors' events the harness forwards.
/// Modes: owner-only (default), allowlist, anyone, nobody.
#[arg(
long,
env = "SPROUT_ACP_RESPOND_TO",
default_value = "owner-only",
value_enum
)]
pub respond_to: RespondTo,
/// Comma-separated 64-char hex pubkeys for allowlist mode.
/// Owner pubkey is always implicitly included.
#[arg(long, env = "SPROUT_ACP_RESPOND_TO_ALLOWLIST", value_delimiter = ',')]
pub respond_to_allowlist: Option<Vec<String>>,
}
// ── Merged NIP-01 filter ──────────────────────────────────────────────────────
@@ -326,6 +367,26 @@ pub struct Config {
pub model: Option<String>,
/// Permission mode to apply after session creation. `Default` = skip.
pub permission_mode: PermissionMode,
/// Inbound author gate mode.
pub respond_to: RespondTo,
/// Validated allowlist of pubkey hex strings (used when respond_to == Allowlist).
pub respond_to_allowlist: HashSet<String>,
}
/// Validate and deduplicate allowlist entries: each must be exactly 64 hex chars.
fn validate_allowlist(entries: &[String]) -> Result<HashSet<String>, ConfigError> {
let mut validated = HashSet::new();
for entry in entries {
let trimmed = entry.trim().to_ascii_lowercase();
if trimmed.len() != 64 || !trimmed.chars().all(|c| c.is_ascii_hexdigit()) {
return Err(ConfigError::ConfigFile(format!(
"invalid pubkey in --respond-to-allowlist: '{entry}' \
(must be exactly 64 hex characters)"
)));
}
validated.insert(trimmed);
}
Ok(validated)
}
fn normalize_agent_command_identity(command: &str) -> String {
@@ -534,6 +595,24 @@ impl Config {
)));
}
// ── Inbound author gate validation ──────────────────────────────────
let respond_to_allowlist = if args.respond_to == RespondTo::Allowlist {
let raw = args.respond_to_allowlist.unwrap_or_default();
if raw.is_empty() {
return Err(ConfigError::ConfigFile(
"--respond-to=allowlist requires --respond-to-allowlist with at least one pubkey".into(),
));
}
validate_allowlist(&raw)?
} else {
if args.respond_to_allowlist.is_some() {
tracing::warn!(
"--respond-to-allowlist is ignored when --respond-to is not 'allowlist'"
);
}
HashSet::new()
};
let config = Config {
keys,
api_token: args.api_token,
@@ -561,6 +640,8 @@ impl Config {
typing_enabled: !args.no_typing,
model: args.model,
permission_mode: args.permission_mode,
respond_to: args.respond_to,
respond_to_allowlist,
};
Ok(config)
@@ -568,8 +649,14 @@ impl Config {
/// Human-readable summary (no secrets).
pub fn summary(&self) -> String {
let respond_to_detail = match &self.respond_to {
RespondTo::Allowlist => {
format!("respond_to=allowlist({})", self.respond_to_allowlist.len())
}
other => format!("respond_to={other}"),
};
format!(
"relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} model={} permission_mode={}",
"relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} model={} permission_mode={} {}",
self.relay_url,
self.keys.public_key().to_hex(),
self.agent_command,
@@ -588,6 +675,7 @@ impl Config {
self.typing_enabled,
self.model.as_deref().unwrap_or("(agent default)"),
self.permission_mode,
respond_to_detail,
)
}
}
@@ -909,6 +997,8 @@ mod tests {
typing_enabled: true,
model: None,
permission_mode: PermissionMode::BypassPermissions,
respond_to: RespondTo::Anyone,
respond_to_allowlist: HashSet::new(),
}
}
@@ -1632,4 +1722,136 @@ channels = "ALL"
"summary should include max_turn: {summary}"
);
}
// ── RespondTo tests ────────────────────────────────────────────────────
#[test]
fn test_respond_to_default_is_owner_only() {
assert_eq!(RespondTo::default(), RespondTo::OwnerOnly);
}
#[test]
fn test_respond_to_display() {
assert_eq!(format!("{}", RespondTo::OwnerOnly), "owner-only");
assert_eq!(format!("{}", RespondTo::Allowlist), "allowlist");
assert_eq!(format!("{}", RespondTo::Anyone), "anyone");
assert_eq!(format!("{}", RespondTo::Nobody), "nobody");
}
#[test]
fn test_respond_to_value_enum_parsing() {
use clap::ValueEnum;
assert_eq!(
RespondTo::from_str("owner-only", true).unwrap(),
RespondTo::OwnerOnly
);
assert_eq!(
RespondTo::from_str("allowlist", true).unwrap(),
RespondTo::Allowlist
);
assert_eq!(
RespondTo::from_str("anyone", true).unwrap(),
RespondTo::Anyone
);
assert_eq!(
RespondTo::from_str("nobody", true).unwrap(),
RespondTo::Nobody
);
}
#[test]
fn test_summary_includes_respond_to() {
let config = test_config(SubscribeMode::Mentions);
let s = config.summary();
assert!(
s.contains("respond_to=anyone"),
"test_config uses Anyone, got: {s}"
);
}
#[test]
fn test_summary_respond_to_allowlist_shows_count() {
let mut config = test_config(SubscribeMode::Mentions);
config.respond_to = RespondTo::Allowlist;
config.respond_to_allowlist = HashSet::from(["ab".repeat(32), "cd".repeat(32)]);
let s = config.summary();
assert!(
s.contains("respond_to=allowlist(2)"),
"should show allowlist count, got: {s}"
);
}
// ── validate_allowlist tests ───────────────────────────────────────────
#[test]
fn test_validate_allowlist_valid_entries() {
let entries = vec!["ab".repeat(32), "cd".repeat(32)];
let result = validate_allowlist(&entries).unwrap();
assert_eq!(result.len(), 2);
}
#[test]
fn test_validate_allowlist_deduplicates() {
let pk = "ab".repeat(32);
let entries = vec![pk.clone(), pk.clone(), pk];
let result = validate_allowlist(&entries).unwrap();
assert_eq!(result.len(), 1);
}
#[test]
fn test_validate_allowlist_normalizes_case() {
let upper = "AB".repeat(32);
let lower = "ab".repeat(32);
let entries = vec![upper, lower];
let result = validate_allowlist(&entries).unwrap();
assert_eq!(result.len(), 1);
assert!(result.contains(&"ab".repeat(32)));
}
#[test]
fn test_validate_allowlist_trims_whitespace() {
let entries = vec![format!(" {} ", "ab".repeat(32))];
let result = validate_allowlist(&entries).unwrap();
assert_eq!(result.len(), 1);
assert!(result.contains(&"ab".repeat(32)));
}
#[test]
fn test_validate_allowlist_rejects_short() {
let entries = vec!["abcd".to_string()];
let err = validate_allowlist(&entries).unwrap_err();
assert!(
err.to_string()
.contains("must be exactly 64 hex characters"),
"got: {err}"
);
}
#[test]
fn test_validate_allowlist_rejects_non_hex() {
let entries = vec!["zz".repeat(32)];
let err = validate_allowlist(&entries).unwrap_err();
assert!(
err.to_string()
.contains("must be exactly 64 hex characters"),
"got: {err}"
);
}
#[test]
fn test_validate_allowlist_rejects_too_long() {
let entries = vec!["ab".repeat(33)]; // 66 chars
let err = validate_allowlist(&entries).unwrap_err();
assert!(
err.to_string()
.contains("must be exactly 64 hex characters"),
"got: {err}"
);
}
#[test]
fn test_validate_allowlist_empty_is_ok() {
let result = validate_allowlist(&[]).unwrap();
assert!(result.is_empty());
}
}

View File

@@ -14,7 +14,7 @@ use std::time::Duration;
use acp::{AcpClient, EnvVar, McpServer};
use anyhow::Result;
use clap::Parser;
use config::{Config, DedupMode, ModelsArgs, SubscribeMode};
use config::{Config, DedupMode, ModelsArgs, RespondTo, SubscribeMode};
use filter::SubscriptionRule;
use futures_util::FutureExt;
use nostr::ToBech32;
@@ -48,6 +48,73 @@ fn is_subcommand(name: &str) -> bool {
/// Timeout for the `sprout-acp models` subcommand (spawn + init + session/new).
const MODELS_TIMEOUT: Duration = Duration::from_secs(10);
// ── Owner cache ───────────────────────────────────────────────────────────────
/// Lazy-resolving cache for the agent's owner pubkey.
///
/// Replaces the bare `Option<String>` that previously lived in the event loop.
/// On success, the owner pubkey is cached for the process lifetime (owner
/// changes require a harness restart). On failure/miss, retries after 60s
/// to avoid hammering the API.
struct OwnerCache {
pubkey: Option<String>,
last_attempt: Option<std::time::Instant>,
}
/// How long to cache a failed owner lookup before retrying.
const OWNER_CACHE_TTL: Duration = Duration::from_secs(60);
impl OwnerCache {
fn new(initial: Option<String>) -> Self {
let last_attempt = if initial.is_some() {
Some(std::time::Instant::now())
} else {
None
};
Self {
pubkey: initial,
last_attempt,
}
}
/// Return the cached owner pubkey, or attempt a lazy resolution if stale.
async fn get_or_resolve(
&mut self,
rest_client: &relay::RestClient,
agent_pubkey_hex: &str,
) -> Option<&str> {
if self.pubkey.is_some() {
return self.pubkey.as_deref();
}
let stale = self
.last_attempt
.map(|t| t.elapsed() >= OWNER_CACHE_TTL)
.unwrap_or(true);
if !stale {
return None;
}
self.last_attempt = Some(std::time::Instant::now());
let profile_url = format!("/api/users/{agent_pubkey_hex}/profile");
match rest_client.get_json(&profile_url).await {
Ok(v) => {
// Normalize to lowercase hex for consistent comparison with
// nostr PublicKey::to_hex() and validated allowlist entries.
self.pubkey = v
.get("agent_owner_pubkey")
.and_then(|v| v.as_str())
.map(|s| s.to_ascii_lowercase());
if let Some(ref o) = self.pubkey {
tracing::info!("lazy owner resolution succeeded: {o}");
}
}
Err(e) => {
tracing::warn!("lazy owner lookup failed: {e}");
}
}
self.pubkey.as_deref()
}
}
/// Maximum crashes in a 60-second window before a slot's circuit opens.
const CIRCUIT_BREAKER_THRESHOLD: usize = 3;
/// Window for circuit-breaker crash counting.
@@ -399,31 +466,47 @@ async fn tokio_main() -> Result<()> {
}
}
// ── Step 2d: Query agent owner (with retry) ─────────────────────────────
// Owner lookup is critical for !shutdown — a transient failure here would
// permanently disable remote shutdown for this process. Retry a few times
// with backoff so a brief relay hiccup doesn't leave us uncontrollable.
// Owner lookup: try at startup, but if it fails the shutdown handler will
// retry lazily when a candidate !shutdown message arrives. This means a
// relay outage during startup doesn't permanently disable remote shutdown.
let mut owner_pubkey: Option<String> = {
// ── Step 2d: Query agent owner ──────────────────────────────────────────
// Owner lookup is used by !shutdown and the inbound author gate.
// Try at startup; OwnerCache retries lazily on cache miss.
let startup_owner: Option<String> = {
let profile_url = format!("/api/users/{pubkey_hex}/profile");
match rest_client_for_presence.get_json(&profile_url).await {
Ok(v) => v
.get("agent_owner_pubkey")
.and_then(|v| v.as_str())
.map(String::from),
.map(|s| s.to_ascii_lowercase()),
Err(e) => {
tracing::warn!("startup owner lookup failed (will retry lazily): {e}");
None
}
}
};
if let Some(ref owner) = owner_pubkey {
if let Some(ref owner) = startup_owner {
tracing::info!("agent owner: {owner}");
} else {
tracing::info!("no agent owner set at startup — will resolve lazily on !shutdown");
tracing::info!("no agent owner set at startup — will resolve lazily");
}
// Warn if owner-dependent mode but no owner resolved yet.
if startup_owner.is_none() {
match &config.respond_to {
RespondTo::OwnerOnly => {
tracing::warn!(
"respond-to=owner-only but no owner is set — all events will be \
dropped until owner is resolved. Set --respond-to=anyone to override."
);
}
RespondTo::Allowlist => {
tracing::warn!(
"respond-to=allowlist but no owner is set — allowlisted pubkeys \
will still be accepted, but owner-based matching is unavailable \
until owner is resolved."
);
}
_ => {} // anyone/nobody don't depend on owner
}
}
let mut owner_cache = OwnerCache::new(startup_owner);
// ── Step 3: Discover channels and build subscription rules ────────────────
let channel_info_map = relay
@@ -849,28 +932,13 @@ async fn tokio_main() -> Result<()> {
&& t.as_slice().get(1).map(|s| s.as_str()) == Some(pubkey_hex.as_str())
});
if is_shutdown {
// Lazy owner resolution: if we don't have the owner
// yet (startup lookup failed), try now. This ensures
// a relay outage during startup doesn't permanently
// Lazy owner resolution via OwnerCache — a relay
// outage during startup doesn't permanently
// disable remote shutdown.
if owner_pubkey.is_none() {
let profile_url = format!("/api/users/{pubkey_hex}/profile");
match rest_client_for_presence.get_json(&profile_url).await {
Ok(v) => {
owner_pubkey = v
.get("agent_owner_pubkey")
.and_then(|v| v.as_str())
.map(String::from);
if let Some(ref o) = owner_pubkey {
tracing::info!("lazy owner resolution succeeded: {o}");
}
}
Err(e) => {
tracing::warn!("lazy owner lookup failed: {e}");
}
}
}
if let Some(ref owner) = owner_pubkey {
let owner = owner_cache
.get_or_resolve(&rest_client_for_presence, &pubkey_hex)
.await;
if let Some(owner) = owner {
if sprout_event.event.pubkey.to_hex() == *owner {
tracing::info!(
channel_id = %sprout_event.channel_id,
@@ -887,6 +955,48 @@ async fn tokio_main() -> Result<()> {
}
// ── End shutdown command handling ──────────────────
// ── Inbound author gate ──────────────────────────
// Coarse security policy: drop events from disallowed
// authors before they reach subscription rules or the
// agent. Must be AFTER !shutdown (owner can always
// shut down regardless of gate mode).
{
let author = sprout_event.event.pubkey.to_hex();
let allowed = match &config.respond_to {
RespondTo::Anyone => true,
RespondTo::Nobody => false,
RespondTo::OwnerOnly => {
let owner = owner_cache
.get_or_resolve(
&rest_client_for_presence,
&pubkey_hex,
)
.await;
owner == Some(author.as_str())
}
RespondTo::Allowlist => {
let owner = owner_cache
.get_or_resolve(
&rest_client_for_presence,
&pubkey_hex,
)
.await;
config.respond_to_allowlist.contains(&author)
|| owner == Some(author.as_str())
}
};
if !allowed {
tracing::debug!(
channel_id = %sprout_event.channel_id,
author = %sprout_event.event.pubkey.to_hex(),
mode = %config.respond_to,
"inbound author gate — dropping event"
);
continue;
}
}
// ── End inbound author gate ──────────────────────
let matched = filter::match_event(&sprout_event.event, sprout_event.channel_id, &rules, &pubkey_hex).await;
let prompt_tag = match matched {
Some(m) => m.prompt_tag,
@@ -1796,3 +1906,84 @@ fn build_mcp_servers(config: &Config) -> Vec<McpServer> {
},
}]
}
// ── Tests ─────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod owner_cache_tests {
use super::*;
#[test]
fn new_with_some_caches_immediately() {
let cache = OwnerCache::new(Some("abcd".into()));
assert_eq!(cache.pubkey.as_deref(), Some("abcd"));
assert!(cache.last_attempt.is_some());
}
#[test]
fn new_with_none_has_no_attempt() {
let cache = OwnerCache::new(None);
assert!(cache.pubkey.is_none());
assert!(cache.last_attempt.is_none());
}
#[test]
fn get_or_resolve_returns_cached_immediately() {
// When pubkey is already cached, get_or_resolve should return it
// without any API call. We can verify this by checking the return
// value without providing a real RestClient (the method short-circuits
// before using it).
let cache = OwnerCache::new(Some("ab".repeat(32)));
// We can't call get_or_resolve without a RestClient in a sync test,
// but we can verify the cache state directly.
assert_eq!(cache.pubkey.as_deref(), Some("ab".repeat(32)).as_deref());
}
#[test]
fn success_cached_for_process_lifetime() {
// After a successful resolution, pubkey stays cached even if
// last_attempt is old. The early return `if self.pubkey.is_some()`
// means the TTL check is never reached.
let mut cache = OwnerCache::new(Some("ab".repeat(32)));
// Simulate time passing by backdating last_attempt
cache.last_attempt = Some(std::time::Instant::now() - Duration::from_secs(3600));
// pubkey is still cached — success is permanent
assert!(cache.pubkey.is_some());
}
#[test]
fn failure_respects_ttl() {
// After a failed lookup, last_attempt is set. A subsequent call
// within the TTL window should NOT retry (stale == false).
let mut cache = OwnerCache::new(None);
cache.last_attempt = Some(std::time::Instant::now());
// Within TTL: stale check returns false, so no retry
let stale = cache
.last_attempt
.map(|t| t.elapsed() >= OWNER_CACHE_TTL)
.unwrap_or(true);
assert!(!stale, "should not be stale within TTL");
}
#[test]
fn failure_retries_after_ttl() {
// After TTL expires, the cache should consider itself stale.
let mut cache = OwnerCache::new(None);
cache.last_attempt = Some(std::time::Instant::now() - Duration::from_secs(61));
let stale = cache
.last_attempt
.map(|t| t.elapsed() >= OWNER_CACHE_TTL)
.unwrap_or(true);
assert!(stale, "should be stale after TTL");
}
#[test]
fn no_attempt_is_always_stale() {
let cache = OwnerCache::new(None);
let stale = cache
.last_attempt
.map(|t| t.elapsed() >= OWNER_CACHE_TTL)
.unwrap_or(true);
assert!(stale, "no prior attempt should be considered stale");
}
}