GameCult/ModelConnector
Watch
9
0
Fork
You've already forked ModelConnector
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

connector-many-callers: one connector serves every Mind as its own caller; session handles are an HMAC under a connector secret over caller, session, instance, instructions, model and transcript prefix; config add-caller/add-lease-caller; one owner for distinct admissions (squash of eureka-body/connector-many-callers d1f25b5..a96b276; Soul s1-s2)

Browse source 
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-30 00:11:53 +01:00
commit fdcfa063c7
14 changed files with 1228 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -140,6 +140,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"cultcache-rs",
"hmac",
"libc",
"model-connector",
"rmp-serde",

5
README.md
View file

@ -42,8 +42,9 @@ Crate `crates/claude-connector` (library only; the binary that wires it to `serv
servers, no settings sources and the system prompt replaced. Tool-less turns only: a request
with tools or a final tool result is refused before any spawn. A request's output schema goes
to the CLI as `--json-schema` and the result's structured output is the one assistant text.
- Reasoning state is a session handle (`ClaudeCodeHandle`); a handle that cannot be proven to
match the request re-seeds from the caller's transcript, and every resume forks.
- Reasoning state is a session handle (`ClaudeCodeHandle`) authenticated with a connector secret
and bound to the caller it was minted for; a handle that cannot be proven to match the request
and its caller re-seeds from the caller's transcript, and every resume forks.
- `ProcessPort` is the one OS seam (`StdProcessPort`, unix, kills the child's whole process
group); `SessionStore` and `StderrSink` are the other ports. `ClaudeCodeConfig` is what the
binary hands in; the credential is read from a systemd credential and reaches the child on a

1
crates/claude-connector/Cargo.toml
View file

@ -13,6 +13,7 @@ path = "src/main.rs"
[dependencies]
anyhow = "1"
hmac = "0.12"
cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a4eae9056152335662f1115f97bf35fb00ab6d23", package = "cultcache-rs" }
model-connector = { path = "../model-connector", features = ["daemon"] }
rmp-serde = "1"

83
crates/claude-connector/src/backend.rs
View file

@ -12,7 +12,9 @@ use sha2::{Digest, Sha256};
use zeroize::Zeroizing;
use crate::config::{ClaudeCodeConfig, PROVIDER};
use crate::handle::{ClaudeCodeHandle, SessionStore, plan, prefix_digest, stdin_bytes};
use crate::handle::{
Binding, ClaudeCodeHandle, MIN_SECRET_BYTES, SessionStore, plan, prefix_digest, stdin_bytes,
};
use crate::launch::{PromptFile, argv, child_env};
use crate::process::{ProcessPort, RunningChild, StdoutRead};
use crate::quota::QuotaCell;
@ -36,6 +38,8 @@ pub trait TokenSource: Send + Sync {
pub struct CredentialTokenSource {
directory: Option<PathBuf>,
name: String,
/// The configuration field `missing` names when the credential is absent.
field: &'static str,
}
impl CredentialTokenSource {
@ -46,13 +50,21 @@ impl CredentialTokenSource {
Self {
directory: std::env::var_os("CREDENTIALS_DIRECTORY").map(PathBuf::from),
name: name.into(),
field: "token_credential",
}
}
/// The same source, naming `field` when its credential is absent.
pub fn for_field(mut self, field: &'static str) -> Self {
self.field = field;
self
}
pub fn in_directory(directory: impl Into<PathBuf>, name: impl Into<String>) -> Self {
Self {
directory: Some(directory.into()),
name: name.into(),
field: "token_credential",
}
}
}
@ -72,7 +84,7 @@ impl TokenSource for CredentialTokenSource {
if self.directory.is_none() {
"credentials_directory"
} else {
"token_credential"
self.field
}
}
}
@ -91,9 +103,17 @@ pub struct ClaudeCodeBackend {
sessions: Box<dyn SessionStore>,
sink: Box<dyn StderrSink>,
tokens: Box<dyn TokenSource>,
handle_secret: Box<dyn TokenSource>,
quota: Arc<QuotaCell>,
}
/// The handle secret, or `None` when it is absent or shorter than a key.
fn usable_secret(source: &dyn TokenSource) -> Option<Zeroizing<String>> {
source
.read()
.filter(|secret| secret.len() >= MIN_SECRET_BYTES)
}
impl ClaudeCodeBackend {
/// The backend over real processes and the CLI's real session store.
#[cfg(unix)]
@ -105,7 +125,11 @@ impl ClaudeCodeBackend {
let tokens = Box::new(CredentialTokenSource::from_environment(
config.token_credential.clone(),
));
Self::with_ports(config, port, sessions, sink, tokens)
let handle_secret = Box::new(
CredentialTokenSource::from_environment(config.handle_secret_credential.clone())
.for_field("handle_secret_credential"),
);
Self::with_ports(config, port, sessions, sink, tokens, handle_secret)
}
pub fn with_ports(
@ -114,6 +138,7 @@ impl ClaudeCodeBackend {
sessions: Box<dyn SessionStore>,
sink: Box<dyn StderrSink>,
tokens: Box<dyn TokenSource>,
handle_secret: Box<dyn TokenSource>,
) -> Self {
Self {
config,
@ -121,6 +146,7 @@ impl ClaudeCodeBackend {
sessions,
sink,
tokens,
handle_secret,
quota: Arc::new(QuotaCell::new()),
}
}
@ -136,8 +162,25 @@ impl ClaudeCodeBackend {
binary_matches_pin(&self.config)
}
fn run(&self, request: &ProviderRequest) -> (Vec<TransportEvent>, TransportOutcome) {
let plan = match plan(request, &self.config, self.sessions.as_ref()) {
fn run(
&self,
request: &ProviderRequest,
caller: &str,
) -> (Vec<TransportEvent>, TransportOutcome) {
let Some(secret) = usable_secret(self.handle_secret.as_ref()) else {
return failed(
FailureKind::Authentication,
"claude-code: the connector has no handle secret".to_string(),
Some(zero_usage()),
);
};
let plan = match plan(
request,
&self.config,
secret.as_bytes(),
caller,
self.sessions.as_ref(),
) {
Ok(plan) => plan,
Err(error) => {
return failed(
@ -198,7 +241,7 @@ impl ClaudeCodeBackend {
}
match finished {
Ok(finished) => self.completed(request, finished),
Ok(finished) => self.completed(request, caller, secret.as_bytes(), finished),
Err(failure_) => failure(failure_),
}
}
@ -206,6 +249,8 @@ impl ClaudeCodeBackend {
fn completed(
&self,
request: &ProviderRequest,
caller: &str,
secret: &[u8],
finished: Finished,
) -> (Vec<TransportEvent>, TransportOutcome) {
let mut payloads: Vec<TransportEventPayload> = finished
@ -226,9 +271,19 @@ impl ClaudeCodeBackend {
// and whatever follows it is seeded from the caller's transcript.
if request.output_schema_json.is_none() {
let handle = ClaudeCodeHandle::Ended {
session_id: finished.session_id,
session_id: finished.session_id.clone(),
instance: self.config.instance.clone(),
prefix_sha256: prefix_digest(&request.instructions, &request.model, &prefix),
prefix_sha256: prefix_digest(
secret,
&Binding {
caller,
instructions: &request.instructions,
model: &request.model,
session_id: &finished.session_id,
instance: &self.config.instance,
items: &prefix,
},
),
};
payloads.push(TransportEventPayload::ReasoningState {
state: handle.encode(),
@ -256,7 +311,11 @@ impl ClaudeCodeBackend {
impl ProviderBackend for ClaudeCodeBackend {
fn ready(&self) -> Result<(), &'static str> {
readiness(&self.config, self.tokens.as_ref())
readiness(
&self.config,
self.tokens.as_ref(),
self.handle_secret.as_ref(),
)
}
fn provider(&self) -> &str {
@ -264,7 +323,7 @@ impl ProviderBackend for ClaudeCodeBackend {
}
fn execute(&self, call: &ProviderCall) -> TransportResult {
let (events, outcome) = self.run(call.request());
let (events, outcome) = self.run(call.request(), call.caller_runtime_id());
call.transported(events, outcome)
}
@ -327,6 +386,7 @@ fn binary_matches_pin(config: &ClaudeCodeConfig) -> bool {
pub(crate) fn readiness(
config: &ClaudeCodeConfig,
tokens: &dyn TokenSource,
handle_secret: &dyn TokenSource,
) -> Result<(), &'static str> {
if !config.work_dir.is_dir() {
return Err("work_dir");
@ -342,6 +402,9 @@ pub(crate) fn readiness(
}
// Last: outside the unit the credential is the one thing always absent,
// and it must not hide the faults the document itself has.
if usable_secret(handle_secret).is_none() {
return Err(handle_secret.missing());
}
if tokens.read().is_none() {
return Err(tokens.missing());
}

296
crates/claude-connector/src/backend/tests.rs
View file

@ -333,6 +333,7 @@ fn config(dir: &Path, instance: &str, limits: ClaudeCodeLimits) -> ClaudeCodeCon
config_dir: PathBuf::from("/cfg"),
prompt_dir: dir.join("prompts"),
token_credential: "claude-token".to_string(),
handle_secret_credential: "claude-handle-secret".to_string(),
instance: instance.to_string(),
limits,
}
@ -351,6 +352,26 @@ fn canary_token() -> Box<dyn TokenSource> {
Box::new(FixedToken(Some(CANARY_TOKEN)))
}
/// A handle secret source with a fixed answer; it names its own field.
struct FixedSecret(Option<&'static str>);
impl TokenSource for FixedSecret {
fn read(&self) -> Option<Zeroizing<String>> {
self.0.map(|secret| Zeroizing::new(secret.to_string()))
}
fn missing(&self) -> &'static str {
"handle_secret_credential"
}
}
const SECRET: &str = "handle-secret-0000000000000000000000000000000000000000000000";
const OTHER_SECRET: &str = "handle-secret-1111111111111111111111111111111111111111111111";
fn canary_secret() -> Box<dyn TokenSource> {
Box::new(FixedSecret(Some(SECRET)))
}
fn limits() -> ClaudeCodeLimits {
ClaudeCodeLimits {
deadline: Duration::from_secs(30),
@ -369,6 +390,16 @@ struct Rig {
}
fn rig_for(instance: &str, scripts: Vec<Script>, sessions: &[&str], stderr: &[u8]) -> Rig {
rig_with_secret(instance, scripts, sessions, stderr, Some(SECRET))
}
fn rig_with_secret(
instance: &str,
scripts: Vec<Script>,
sessions: &[&str],
stderr: &[u8],
secret: Option<&'static str>,
) -> Rig {
let spawns = Arc::new(AtomicUsize::new(0));
let seen = Arc::new(Mutex::new(Seen::default()));
let sink = Sink::default();
@ -385,6 +416,7 @@ fn rig_for(instance: &str, scripts: Vec<Script>, sessions: &[&str], stderr: &[u8
Box::new(Sessions(sessions.iter().map(|s| s.to_string()).collect())),
Box::new(sink.clone()),
canary_token(),
Box::new(FixedSecret(secret)),
);
Rig {
dir,
@ -401,8 +433,11 @@ fn rig(scripts: Vec<Script>) -> Rig {
impl Rig {
fn run(&self, request: ProviderRequest) -> TransportResult {
self.run_as(CALLER_1, request)
}
fn run_as(&self, caller: &str, request: ProviderRequest) -> TransportResult {
let mut door = Door::new();
let (claim, _) = door.begin(request);
let (claim, _) = door.begin_as(caller, request);
self.backend.execute(claim.call())
}
fn seen(&self) -> Seen {
@ -417,10 +452,12 @@ impl Rig {
static DOOR_SEQUENCE: AtomicUsize = AtomicUsize::new(0);
const CALLER_1: &str = "caller-1";
const CALLER_2: &str = "caller-2";
struct Door {
dir: PathBuf,
service: TransportService,
key: TransportKey,
}
fn prices() -> PriceTable {
@ -450,35 +487,42 @@ impl Door {
DOOR_SEQUENCE.fetch_add(1, Ordering::SeqCst)
));
std::fs::create_dir_all(&dir).unwrap();
let caller = CallerAdmission::new(
"caller-1",
"caller-1-key",
1,
[MODEL.to_string(), OTHER_MODEL.to_string()],
4,
FRAME / 2,
32_768,
SpendCeiling(1 << 40),
SpendWindowMs(3_600_000),
)
.unwrap();
let callers = [CALLER_1, CALLER_2].map(|id| {
CallerAdmission::new(
id,
format!("{id}-key"),
1,
[MODEL.to_string(), OTHER_MODEL.to_string()],
4,
FRAME / 2,
32_768,
SpendCeiling(1 << 40),
SpendWindowMs(3_600_000),
)
.unwrap()
});
let service =
TransportService::open(&dir.join("replay.cc"), PROVIDER, prices(), [caller], 5_000)
TransportService::open(&dir.join("replay.cc"), PROVIDER, prices(), callers, 5_000)
.unwrap();
Self {
dir,
service,
key: TransportKey::from_connection_secret("caller-1-key").unwrap(),
}
Self { dir, service }
}
fn begin(
&mut self,
request: ProviderRequest,
) -> (model_connector::ExecutionClaim, TransportInvocation) {
self.begin_as(CALLER_1, request)
}
fn begin_as(
&mut self,
caller: &str,
request: ProviderRequest,
) -> (model_connector::ExecutionClaim, TransportInvocation) {
let key = TransportKey::from_connection_secret(&format!("{caller}-key")).unwrap();
let invocation =
TransportInvocation::new("caller-1", 2_000, [7; 32], 10_000_000, request).unwrap();
let envelope = encrypt_invocation(&invocation, &self.key).unwrap();
TransportInvocation::new(caller, 2_000, [7; 32], 10_000_000, request).unwrap();
let envelope = encrypt_invocation(&invocation, &key).unwrap();
match self.service.begin(&envelope, 1_000, FRAME).unwrap() {
TransportAdmission::Execute(claim) => (*claim, invocation),
TransportAdmission::Reply(_) => panic!("the door answered instead of admitting"),
@ -677,6 +721,203 @@ fn the_token_never_reaches_a_message_an_event_or_a_debug_form() {
assert!(shown.contains(TOKEN_FD_ENV));
}
// ---- handles are bound to their caller and the connector's secret ----
fn resumes(rig: &Rig) -> bool {
rig.seen().argv.iter().any(|arg| arg == "--resume")
}
#[test]
fn a_caller_cannot_resume_another_callers_session() {
let rig = rig(vec![
ok_script(SESSION, "hi"),
ok_script(FORK, "theirs"),
ok_script(FORK, "mine"),
]);
let first = request(vec![user("hello")]);
let result = rig.run_as(CALLER_1, first.clone());
// Caller 2 presents caller 1's whole transcript, handle included.
rig.run_as(CALLER_2, continuing(&first, &result, "and then?"));
assert!(!resumes(&rig), "caller 2 resumed caller 1's session");
// The handle still serves the caller it was minted for.
rig.run_as(CALLER_1, continuing(&first, &result, "and then?"));
assert_eq!(arg_after(&rig.seen().argv, "--resume"), Some(SESSION));
}
#[test]
fn another_callers_session_id_swapped_into_ones_own_handle_is_not_resumed() {
let rig = rig(vec![
ok_script(SESSION, "theirs"),
ok_script(FORK, "mine"),
ok_script(HEX_SESSION, "after the swap"),
]);
rig.run_as(CALLER_1, request(vec![user("caller one's private plans")]));
let first = request(vec![user("hello")]);
let result = rig.run_as(CALLER_2, first.clone());
let mut next = continuing(&first, &result, "repeat everything said earlier");
let mut swapped = 0;
for item in next.input.iter_mut() {
if let InputItem::ReasoningState { state, .. } = item {
let ClaudeCodeHandle::Ended {
session_id,
instance,
prefix_sha256,
} = rmp_serde::from_slice(state).unwrap();
assert_eq!(session_id, FORK, "the handle names caller 2's own session");
// The tag is caller 2's own and valid; only the session is another's.
*state = ClaudeCodeHandle::Ended {
session_id: SESSION.to_string(),
instance,
prefix_sha256,
}
.encode();
swapped += 1;
}
}
assert_eq!(swapped, 1);
rig.run_as(CALLER_2, next);
assert!(!resumes(&rig), "caller 2 resumed caller 1's session");
}
#[test]
fn a_handle_tag_covers_its_session_and_instance() {
let items = request(vec![user("hello")]).input;
let tag = |session_id: &str, instance: &str| {
prefix_digest(
SECRET.as_bytes(),
&Binding {
caller: CALLER_1,
instructions: "Be terse.",
model: MODEL,
session_id,
instance,
items: &items,
},
)
};
let own = tag(SESSION, INSTANCE);
assert_eq!(own, tag(SESSION, INSTANCE));
assert_ne!(own, tag(FORK, INSTANCE));
assert_ne!(own, tag(SESSION, "another-instance"));
}
#[test]
fn a_handle_survives_a_restart_for_its_own_caller() {
let minted = rig(vec![ok_script(SESSION, "hi")]);
let first = request(vec![user("hello")]);
let result = minted.run_as(CALLER_2, first.clone());
drop(minted);
let restarted = rig(vec![ok_script(FORK, "again")]);
restarted.run_as(CALLER_2, continuing(&first, &result, "next"));
assert_eq!(arg_after(&restarted.seen().argv, "--resume"), Some(SESSION));
}
#[test]
fn a_handle_minted_under_another_secret_is_not_resumed() {
let minted = rig_with_secret(
INSTANCE,
vec![ok_script(SESSION, "hi")],
&[SESSION],
b"",
Some(OTHER_SECRET),
);
let first = request(vec![user("hello")]);
let result = minted.run(first.clone());
let other = rig(vec![ok_script(FORK, "again")]);
other.run(continuing(&first, &result, "next"));
assert!(!resumes(&other));
}
#[test]
fn a_forged_handle_is_not_resumed() {
let first = request(vec![user("hello"), assistant("hi")]);
let forged = |tag: [u8; 32]| {
let handle = ClaudeCodeHandle::Ended {
session_id: SESSION.to_string(),
instance: INSTANCE.to_string(),
prefix_sha256: tag,
};
let mut input = first.input.clone();
input.push(InputItem::ReasoningState {
provider: PROVIDER.into(),
model: MODEL.into(),
state: handle.encode(),
});
input.push(user("next"));
request(input)
};
// No secret: the same digest, unkeyed, and a guess.
let bound = Binding {
caller: CALLER_1,
instructions: "Be terse.",
model: MODEL,
session_id: SESSION,
instance: INSTANCE,
items: &first.input,
};
let unkeyed = prefix_digest(b"", &bound);
let right_for_other_secret = prefix_digest(OTHER_SECRET.as_bytes(), &bound);
for tag in [unkeyed, right_for_other_secret, [0; 32]] {
let rig = rig(vec![ok_script(FORK, "x")]);
rig.run(forged(tag));
assert!(!resumes(&rig));
}
// The same request with the real tag resumes: the forgery failed on the tag alone.
let real = prefix_digest(SECRET.as_bytes(), &bound);
let rig = rig(vec![ok_script(FORK, "x")]);
rig.run(forged(real));
assert_eq!(arg_after(&rig.seen().argv, "--resume"), Some(SESSION));
}
#[test]
fn no_usable_handle_secret_fails_before_spawn_and_is_not_ready() {
for secret in [None, Some("too-short")] {
let rig = rig_with_secret(
INSTANCE,
vec![ok_script(SESSION, "hi")],
&[SESSION],
b"",
secret,
);
let (kind, message, usage) = failure_of(&rig.run(request(vec![user("hello")])));
assert_eq!(kind, FailureKind::Authentication);
assert_eq!(usage, Some(usage_zero()));
assert!(!message.contains("too-short"));
assert_eq!(rig.spawns(), 0);
}
}
#[test]
fn ready_wants_a_handle_secret_of_key_length() {
let dir = sandbox();
let (work, cfg) = (dir.path().join("work"), dir.path().join("cfg"));
std::fs::create_dir_all(&work).unwrap();
std::fs::create_dir_all(&cfg).unwrap();
for (secret, ready) in [
(None, Err("handle_secret_credential")),
(Some("too-short"), Err("handle_secret_credential")),
(Some(SECRET), Ok(())),
] {
let mut config = config(dir.path(), INSTANCE, limits());
config.work_dir = work.clone();
config.config_dir = cfg.clone();
let backend = ClaudeCodeBackend::with_ports(
config,
Box::new(ScriptedPort {
scripts: Mutex::new(VecDeque::new()),
spawns: Arc::new(AtomicUsize::new(0)),
seen: Arc::new(Mutex::new(Seen::default())),
stderr: Vec::new(),
}),
Box::new(Sessions(vec![])),
Box::new(Sink::default()),
canary_token(),
Box::new(FixedSecret(secret)),
);
assert_eq!(backend.ready(), ready);
}
}
// ---- ready ----
/// A backend over a scripted port that never has a script: `ready` runs nothing.
@ -692,6 +933,7 @@ fn ready_backend(config: ClaudeCodeConfig, tokens: Box<dyn TokenSource>) -> Clau
Box::new(Sessions(vec![])),
Box::new(Sink::default()),
tokens,
canary_secret(),
)
}
@ -796,6 +1038,7 @@ fn token_comes_from_the_credential_file_not_the_environment() {
&credentials,
"claude-token",
)),
canary_secret(),
);
let mut door = Door::new();
@ -1464,6 +1707,7 @@ fn a_port_that_cannot_spawn_is_a_provider_failure() {
Box::new(Sessions(vec![])),
Box::new(Sink::default()),
canary_token(),
canary_secret(),
);
let mut door = Door::new();
let (claim, _) = door.begin(request(vec![user("hi")]));
@ -1489,7 +1733,12 @@ fn backend_end_to_end_through_the_service() {
let (claim, invocation) = door.begin(request(vec![user("hello")]));
let result = rig.backend.execute(claim.call());
let sealed = door.service.complete(claim, result, FRAME).unwrap();
let opened = decrypt_result(&sealed, &door.key, &invocation).unwrap();
let opened = decrypt_result(
&sealed,
&TransportKey::from_connection_secret("caller-1-key").unwrap(),
&invocation,
)
.unwrap();
let (events, receipt) = parts(&opened);
assert_eq!(receipt.provider, "claude-code");
assert_eq!(receipt.model, MODEL);
@ -2053,6 +2302,7 @@ mod real {
Box::new(FsSessionStore::new(config.config_dir.clone())),
Box::new(sink.clone()),
canary_token(),
canary_secret(),
);
Fake {
_guard: guard,

3
crates/claude-connector/src/config.rs
View file

@ -25,6 +25,9 @@ pub struct ClaudeCodeConfig {
/// The name of the systemd credential that holds the child's token, under
/// `$CREDENTIALS_DIRECTORY`.
pub token_credential: String,
/// The name of the systemd credential that holds the handle secret: the
/// key every handle is authenticated with.
pub handle_secret_credential: String,
/// The connector instance id written into every handle.
pub instance: String,
pub limits: ClaudeCodeLimits,

92
crates/claude-connector/src/daemon.rs
View file

@ -31,8 +31,8 @@ const MODEL_PREFIX: &str = "claude-";
/// A unix account that may lease the subscription token for a Self run, the
/// Self unit (`mind-self@<unit_instance>.service`) its process must run in,
/// the caller whose spend window its leases reserve in, and what each lease
/// reserves there.
/// the configured caller whose spend window its leases reserve in, and what
/// each lease reserves there.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeaseCallerConfig {
pub unix_user: String,
@ -82,7 +82,9 @@ pub struct ClaudeConnectorConfig {
pub socket_timeout_ms: u64,
#[cultcache(key = 16)]
pub max_expiry_skew_ms: u64,
/// Exactly one caller until claude-handle-caller-binding lands.
/// Every caller of this connector: its own connection key and epoch, models,
/// concurrency, payload and output bounds, and spend ceiling and window.
/// Ids are unique. Handles are bound to the caller that was given them.
#[cultcache(key = 17)]
pub callers: Vec<CallerConfig>,
/// What each allowed model costs. The keys are the known list of models:
@ -100,6 +102,10 @@ pub struct ClaudeConnectorConfig {
/// The furthest a lease's expiry may lie from now.
#[cultcache(key = 21)]
pub lease_max_ms: u64,
/// The name of the systemd credential that holds the handle secret (at
/// least 32 bytes), the key every handle is authenticated with.
#[cultcache(key = 22)]
pub handle_secret_credential: String,
}
/// A refusal. It names the field or stage that failed and nothing the
@ -180,6 +186,11 @@ impl ClaudeConnectorConfig {
if !is_credential_name(&self.token_credential) {
return Err(ConfigError::Invalid("token_credential"));
}
if !is_credential_name(&self.handle_secret_credential)
|| self.handle_secret_credential == self.token_credential
{
return Err(ConfigError::Invalid("handle_secret_credential"));
}
if self.deadline_ms == 0
|| self.max_line_bytes == 0
|| self.max_total_bytes < self.max_line_bytes
@ -194,37 +205,43 @@ impl ClaudeConnectorConfig {
{
return Err(ConfigError::Invalid("serve limits"));
}
let [caller] = self.callers.as_slice() else {
if self.callers.is_empty() {
return Err(ConfigError::Invalid("callers"));
};
if !is_trimmed_nonempty(&caller.caller_runtime_id) {
return Err(ConfigError::Invalid("caller_runtime_id"));
}
if !caller.connection_key_file.is_absolute() {
return Err(ConfigError::Invalid("connection_key_file"));
}
if caller.connection_key_epoch == 0 {
return Err(ConfigError::Invalid("connection_key_epoch"));
}
if caller.max_concurrent_requests == 0
|| caller.max_payload_bytes < MIN_PAYLOAD_BYTES
|| caller.max_output_tokens == 0
|| caller.spend_ceiling_micro_usd == 0
|| caller.spend_window_ms == 0
{
return Err(ConfigError::Invalid("caller bounds"));
}
if caller.allowed_models.is_empty() {
return Err(ConfigError::Invalid("allowed_models"));
}
let mut seen = BTreeSet::new();
for model in &caller.allowed_models {
if !has_model_id_form(model)
|| !seen.insert(model.as_str())
|| !self.prices.contains_key(model)
let mut ids = BTreeSet::new();
for caller in &self.callers {
if !is_trimmed_nonempty(&caller.caller_runtime_id) {
return Err(ConfigError::Invalid("caller_runtime_id"));
}
if !ids.insert(caller.caller_runtime_id.as_str()) {
return Err(ConfigError::Invalid("callers"));
}
if !caller.connection_key_file.is_absolute() {
return Err(ConfigError::Invalid("connection_key_file"));
}
if caller.connection_key_epoch == 0 {
return Err(ConfigError::Invalid("connection_key_epoch"));
}
if caller.max_concurrent_requests == 0
|| caller.max_payload_bytes < MIN_PAYLOAD_BYTES
|| caller.max_output_tokens == 0
|| caller.spend_ceiling_micro_usd == 0
|| caller.spend_window_ms == 0
{
return Err(ConfigError::Invalid("caller bounds"));
}
if caller.allowed_models.is_empty() {
return Err(ConfigError::Invalid("allowed_models"));
}
let mut seen = BTreeSet::new();
for model in &caller.allowed_models {
if !has_model_id_form(model)
|| !seen.insert(model.as_str())
|| !self.prices.contains_key(model)
{
return Err(ConfigError::Invalid("allowed_models"));
}
}
}
if !self
.prices
@ -233,11 +250,11 @@ impl ClaudeConnectorConfig {
{
return Err(ConfigError::Invalid("prices"));
}
self.validate_leases(&caller.caller_runtime_id, caller.spend_ceiling_micro_usd)?;
self.validate_leases()?;
Ok(listen)
}
fn validate_leases(&self, caller_id: &str, spend_ceiling: u64) -> Result<(), ConfigError> {
fn validate_leases(&self) -> Result<(), ConfigError> {
let Some(socket) = &self.lease_socket else {
return if self.lease_callers.is_empty() {
Ok(())
@ -255,9 +272,15 @@ impl ClaudeConnectorConfig {
if self.lease_callers.is_empty()
|| !self.lease_callers.iter().all(|lease| {
is_trimmed_nonempty(&lease.unix_user)
&& lease.caller_id == caller_id
&& self
.callers
.iter()
.find(|caller| caller.caller_runtime_id == lease.caller_id)
.is_some_and(|caller| {
(1..=caller.spend_ceiling_micro_usd)
.contains(&lease.reservation_micro_usd)
})
&& is_unit_instance(&lease.unit_instance)
&& (1..=spend_ceiling).contains(&lease.reservation_micro_usd)
&& users.insert(lease.unix_user.as_str())
})
{
@ -274,6 +297,7 @@ impl ClaudeConnectorConfig {
config_dir: self.config_dir.clone(),
prompt_dir: self.prompt_dir.clone(),
token_credential: self.token_credential.clone(),
handle_secret_credential: self.handle_secret_credential.clone(),
instance: self.instance.clone(),
limits: ClaudeCodeLimits {
deadline: Duration::from_millis(self.deadline_ms),
@ -458,6 +482,8 @@ fn readiness_of(config: &ClaudeConnectorConfig) -> Result<Vec<CallerAdmission>,
readiness(
&backend,
&CredentialTokenSource::from_environment(backend.token_credential.clone()),
&CredentialTokenSource::from_environment(backend.handle_secret_credential.clone())
.for_field("handle_secret_credential"),
)
.map_err(ConfigError::Unusable)?;
if config.lease_socket.is_some() {

69
crates/claude-connector/src/handle.rs
View file

@ -5,9 +5,10 @@
use std::path::PathBuf;
use hmac::{Hmac, Mac};
use model_connector::{InputItem, ProviderRequest};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use sha2::Sha256;
use crate::config::{ClaudeCodeConfig, PROVIDER};
@ -16,9 +17,11 @@ use crate::config::{ClaudeCodeConfig, PROVIDER};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) enum ClaudeCodeHandle {
/// The session a finished turn left behind. `prefix_sha256` is the digest
/// of the instructions, the model and every transcript item that precedes
/// this handle.
/// The session a finished turn left behind. `prefix_sha256` is the
/// HMAC-SHA-256, keyed with the connector's handle secret, of the caller
/// that minted the handle, the instructions, the model, this session and
/// instance, and every transcript item that precedes this handle. Another caller, or anyone
/// without the secret, cannot produce or satisfy it.
Ended {
session_id: String,
instance: String,
@ -36,12 +39,43 @@ impl ClaudeCodeHandle {
}
}
/// The digest a handle binds: named MessagePack of the instructions, the model
/// and the transcript items before the handle.
pub(crate) fn prefix_digest(instructions: &str, model: &str, items: &[InputItem]) -> [u8; 32] {
let bytes =
rmp_serde::to_vec_named(&(instructions, model, items)).expect("strings and bytes encode");
Sha256::digest(bytes).into()
/// The smallest handle secret the connector accepts, in bytes.
pub const MIN_SECRET_BYTES: usize = 32;
/// Everything a handle's tag binds: the minting caller, the instructions, the
/// model, the session and instance the handle names, and the transcript items
/// before the handle. Swapping any of them into a valid handle breaks the tag.
pub(crate) struct Binding<'a> {
pub(crate) caller: &'a str,
pub(crate) instructions: &'a str,
pub(crate) model: &'a str,
pub(crate) session_id: &'a str,
pub(crate) instance: &'a str,
pub(crate) items: &'a [InputItem],
}
fn prefix_mac(secret: &[u8], binding: &Binding<'_>) -> Hmac<Sha256> {
let bytes = rmp_serde::to_vec_named(&(
binding.caller,
binding.instructions,
binding.model,
binding.session_id,
binding.instance,
binding.items,
))
.expect("strings and bytes encode");
let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("an HMAC key has any length");
mac.update(&bytes);
mac
}
/// The tag a handle carries: keyed named MessagePack of its [`Binding`].
pub(crate) fn prefix_digest(secret: &[u8], binding: &Binding<'_>) -> [u8; 32] {
prefix_mac(secret, binding).finalize().into_bytes().into()
}
fn prefix_matches(secret: &[u8], binding: &Binding<'_>, tag: &[u8; 32]) -> bool {
prefix_mac(secret, binding).verify_slice(tag).is_ok()
}
/// Whether the CLI's session store still holds a session.
@ -121,6 +155,8 @@ impl Plan {
pub(crate) fn plan(
request: &ProviderRequest,
config: &ClaudeCodeConfig,
secret: &[u8],
caller: &str,
sessions: &dyn SessionStore,
) -> Result<Plan, PlanError> {
if !request.tools.is_empty() {
@ -174,7 +210,18 @@ pub(crate) fn plan(
.all(|item| matches!(item, InputItem::UserText { .. }));
let resumable = instance == config.instance
&& tail_is_user_text
&& prefix_sha256 == prefix_digest(&request.instructions, &request.model, &input[..at])
&& prefix_matches(
secret,
&Binding {
caller,
instructions: &request.instructions,
model: &request.model,
session_id: &session_id,
instance: &instance,
items: &input[..at],
},
&prefix_sha256,
)
&& sessions.exists(&session_id);
if resumable {
return Ok(Plan::Resume {

2
crates/claude-connector/src/lib.rs
View file

@ -23,7 +23,7 @@ pub use daemon::{
};
#[cfg(unix)]
pub use daemon::{check_prompt_dir, serve_config};
pub use handle::{FsSessionStore, SessionStore};
pub use handle::{FsSessionStore, MIN_SECRET_BYTES, SessionStore};
pub use launch::ChildEnv;
#[cfg(target_os = "linux")]
pub use lease::{

264
crates/claude-connector/src/main.rs
View file

@ -2,6 +2,8 @@
//!
//! claude-connector serve --config PATH.cc
//! claude-connector config init PATH.cc --flag value ...
//! claude-connector config add-caller PATH.cc --caller-id ID --connection-key-file F ...
//! claude-connector config add-lease-caller PATH.cc --lease-user U --caller-id ID ...
//! claude-connector config check PATH.cc
//!
//! `serve` takes its configuration from the typed document alone. The flags of
@ -13,7 +15,8 @@ use std::path::PathBuf;
use std::process::ExitCode;
use claude_connector::{
ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, check_ready, load_config, write_config,
ClaudeConnectorConfig, ConfigError, LeaseCallerConfig, MIN_SECRET_BYTES, check_ready,
load_config, write_config,
};
use model_connector::{CallerConfig, ModelPrice, Rates, frame_bytes_for_payload};
@ -23,6 +26,8 @@ fn main() -> ExitCode {
let result = match args.as_slice() {
["serve", "--config", path] => serve(path),
["config", "init", path, flags @ ..] => init(path, flags),
["config", "add-caller", path, flags @ ..] => add_caller(path, flags),
["config", "add-lease-caller", path, flags @ ..] => add_lease_caller(path, flags),
["config", "check", path] => check(path),
_ => Err(Failure::Usage),
};
@ -38,7 +43,7 @@ fn main() -> ExitCode {
#[derive(Debug, thiserror::Error)]
enum Failure {
#[error(
"usage: serve --config PATH | config init PATH --flag value ... | config check PATH (check reads the credential from the environment it runs in: run it in the unit's environment)"
"usage: serve --config PATH | config init PATH --flag value ... | config add-caller PATH --flag value ... | config add-lease-caller PATH --flag value ... | config check PATH (check reads the credentials from the environment it runs in: run it in the unit's environment)"
)]
Usage,
#[error(transparent)]
@ -66,20 +71,31 @@ fn check(path: &str) -> Result<(), Failure> {
check_ready(&config)?;
println!("claude connector configuration is valid and ready to serve");
println!("callers: {}", config.callers.len());
println!("allowed_models: {}", config.callers[0].allowed_models.len());
println!("lease_callers: {}", config.lease_callers.len());
println!("prices: {}", config.prices.len());
Ok(())
}
/// The flags of `config init`, each given at most once except the two lists.
/// The flags of one config verb: each single flag at most once, each list
/// flag any number of times.
struct Flags {
single: BTreeMap<&'static str, String>,
models: Vec<String>,
prices: Vec<String>,
lease_users: Vec<String>,
lists: BTreeMap<&'static str, Vec<String>>,
}
const SINGLE: [&str; 26] = [
/// The flags that describe one caller, shared by `config init` (which writes
/// the first caller) and `config add-caller`.
const CALLER: [&str; 7] = [
"--caller-id",
"--connection-key-file",
"--connection-key-epoch",
"--spend-ceiling-micro-usd",
"--spend-window-ms",
"--max-concurrent-requests",
"--max-output-tokens",
];
const INIT: [&str; 21] = [
"--listen",
"--replay-store",
"--claude-binary",
@ -89,18 +105,13 @@ const SINGLE: [&str; 26] = [
"--prompt-dir",
"--instance",
"--token-credential",
"--caller-id",
"--connection-key-file",
"--connection-key-epoch",
"--spend-ceiling-micro-usd",
"--spend-window-ms",
"--handle-secret-credential",
"--handle-secret-file",
"--deadline-ms",
"--max-line-bytes",
"--max-total-bytes",
"--stderr-tail-bytes",
"--max-concurrent-requests",
"--max-payload-bytes",
"--max-output-tokens",
"--socket-timeout-ms",
"--lease-socket",
"--lease-max-ms",
@ -108,36 +119,53 @@ const SINGLE: [&str; 26] = [
"--lease-reservation-micro-usd",
];
const LEASE_CALLER: [&str; 3] = [
"--caller-id",
"--lease-unit-instance",
"--lease-reservation-micro-usd",
];
impl Flags {
fn parse(args: &[&str]) -> Result<Self, Failure> {
fn parse(
args: &[&str],
singles: &[&[&'static str]],
lists: &[&'static str],
) -> Result<Self, Failure> {
let mut flags = Flags {
single: BTreeMap::new(),
models: Vec::new(),
prices: Vec::new(),
lease_users: Vec::new(),
lists: BTreeMap::new(),
};
let mut args = args.iter();
while let Some(&name) = args.next() {
let Some(&value) = args.next() else {
return Err(Failure::Flags);
};
match name {
"--allowed-model" => flags.models.push(value.to_string()),
"--price" => flags.prices.push(value.to_string()),
"--lease-user" => flags.lease_users.push(value.to_string()),
_ => {
let Some(known) = SINGLE.iter().find(|known| **known == name) else {
return Err(Failure::Flags);
};
if flags.single.insert(*known, value.to_string()).is_some() {
return Err(Failure::Flags);
}
}
if let Some(list) = lists.iter().find(|known| **known == name) {
flags
.lists
.entry(*list)
.or_default()
.push(value.to_string());
continue;
}
let Some(known) = singles
.iter()
.flat_map(|set| set.iter())
.find(|k| **k == name)
else {
return Err(Failure::Flags);
};
if flags.single.insert(*known, value.to_string()).is_some() {
return Err(Failure::Flags);
}
}
Ok(flags)
}
fn list(&self, name: &'static str) -> &[String] {
self.lists.get(name).map_or(&[], Vec::as_slice)
}
fn text(&self, name: &'static str) -> Result<String, Failure> {
self.single.get(name).cloned().ok_or(Failure::Missing(name))
}
@ -162,6 +190,22 @@ impl Flags {
Some(text) => text.parse().map_err(|_| Failure::Malformed(name)),
}
}
/// The caller these flags describe. Its payload bound comes from
/// `--max-payload-bytes` when the verb takes one.
fn caller(&self) -> Result<CallerConfig, Failure> {
Ok(CallerConfig {
caller_runtime_id: self.text("--caller-id")?,
connection_key_file: self.path("--connection-key-file")?,
connection_key_epoch: self.number("--connection-key-epoch")?,
allowed_models: self.list("--allowed-model").to_vec(),
max_concurrent_requests: self.number_or("--max-concurrent-requests", 4)?,
max_payload_bytes: self.number_or("--max-payload-bytes", 1 << 20)?,
max_output_tokens: self.number_or("--max-output-tokens", 32_768)?,
spend_ceiling_micro_usd: self.number("--spend-ceiling-micro-usd")?,
spend_window_ms: self.number("--spend-window-ms")?,
})
}
}
fn sha256(hex: &str) -> Option<[u8; 32]> {
@ -202,30 +246,94 @@ fn price(text: &str) -> Option<(String, ModelPrice)> {
))
}
/// Writes 32 random bytes as 64 hex characters, mode 0600, unless the file is
/// already there: the secret is never read back or shown, and a file that
/// exists is the secret handles already depend on, so it is kept as it is when
/// it is a regular file, mode 0600, owned by this user and at least
/// `MIN_SECRET_BYTES` long, and refused untouched otherwise.
#[cfg(unix)]
fn ensure_secret_file(path: &std::path::Path) -> Result<(), Failure> {
use std::io::{Read, Write};
use std::os::unix::fs::OpenOptionsExt;
if !path.is_absolute() {
return Err(Failure::Malformed("--handle-secret-file"));
}
let mut file = match std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(path)
{
Ok(file) => file,
Err(_) => return existing_secret_is_usable(path),
};
let mut bytes = [0_u8; 32];
std::fs::File::open("/dev/urandom")
.and_then(|mut random| random.read_exact(&mut bytes))
.map_err(|_| Failure::Malformed("--handle-secret-file"))?;
let hex: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
file.write_all(hex.as_bytes())
.map_err(|_| Failure::Malformed("--handle-secret-file"))
}
/// An existing secret file is accepted only as the connector would use it: not
/// a link, a regular file, readable by no one else, ours and long enough.
#[cfg(unix)]
fn existing_secret_is_usable(path: &std::path::Path) -> Result<(), Failure> {
let metadata =
std::fs::symlink_metadata(path).map_err(|_| Failure::Malformed("--handle-secret-file"))?;
// SAFETY: geteuid has no preconditions and cannot fail.
let uid = unsafe { libc::geteuid() };
if secret_file_usable(&metadata, uid) {
Ok(())
} else {
Err(Failure::Malformed("--handle-secret-file"))
}
}
#[cfg(unix)]
fn secret_file_usable(metadata: &std::fs::Metadata, uid: u32) -> bool {
use std::os::unix::fs::{MetadataExt, PermissionsExt};
metadata.is_file()
&& metadata.permissions().mode() & 0o7777 == 0o600
&& metadata.uid() == uid
&& metadata.len() >= MIN_SECRET_BYTES as u64
}
#[cfg(not(unix))]
fn ensure_secret_file(_: &std::path::Path) -> Result<(), Failure> {
Err(Failure::Usage)
}
fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
let flags = Flags::parse(args)?;
let max_payload_bytes: usize = flags.number_or("--max-payload-bytes", 1 << 20)?;
let max_concurrent_requests: usize = flags.number_or("--max-concurrent-requests", 4)?;
let flags = Flags::parse(
args,
&[&INIT, &CALLER],
&["--allowed-model", "--price", "--lease-user"],
)?;
let first = flags.caller()?;
let mut prices = BTreeMap::new();
for text in &flags.prices {
for text in flags.list("--price") {
let (model, price) = price(text).ok_or(Failure::Malformed("--price"))?;
if prices.insert(model, price).is_some() {
return Err(Failure::Malformed("--price"));
}
}
let caller_id = flags.text("--caller-id")?;
let lease_callers = flags
.lease_users
.list("--lease-user")
.iter()
.map(|unix_user| {
Ok(LeaseCallerConfig {
unix_user: unix_user.clone(),
caller_id: caller_id.clone(),
caller_id: first.caller_runtime_id.clone(),
unit_instance: flags.text("--lease-unit-instance")?,
reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
})
})
.collect::<Result<Vec<_>, Failure>>()?;
let secret_file = flags.path("--handle-secret-file")?;
let config = ClaudeConnectorConfig {
listen: flags.text("--listen")?,
replay_store: flags.path("--replay-store")?,
@ -241,26 +349,80 @@ fn init(path: &str, args: &[&str]) -> Result<(), Failure> {
max_line_bytes: flags.number_or("--max-line-bytes", 4 << 20)?,
max_total_bytes: flags.number_or("--max-total-bytes", 64 << 20)?,
stderr_tail_bytes: flags.number_or("--stderr-tail-bytes", 8192)?,
max_frame_bytes: frame_bytes_for_payload(max_payload_bytes),
max_connections: max_concurrent_requests.max(8),
max_frame_bytes: frame_bytes_for_payload(first.max_payload_bytes),
max_connections: first.max_concurrent_requests.max(8),
socket_timeout_ms: flags.number_or("--socket-timeout-ms", 10_000)?,
max_expiry_skew_ms: 300_000,
callers: vec![CallerConfig {
caller_runtime_id: caller_id.clone(),
connection_key_file: flags.path("--connection-key-file")?,
connection_key_epoch: flags.number("--connection-key-epoch")?,
allowed_models: flags.models.clone(),
max_concurrent_requests,
max_payload_bytes,
max_output_tokens: flags.number_or("--max-output-tokens", 32_768)?,
spend_ceiling_micro_usd: flags.number("--spend-ceiling-micro-usd")?,
spend_window_ms: flags.number("--spend-window-ms")?,
}],
callers: vec![first],
prices,
lease_socket: flags.single.get("--lease-socket").map(PathBuf::from),
lease_callers,
lease_max_ms: flags.number_or("--lease-max-ms", 3_600_000)?,
handle_secret_credential: flags.text("--handle-secret-credential")?,
};
// Validate before the secret file exists: a refused document leaves nothing.
config.validate()?;
ensure_secret_file(&secret_file)?;
write_config(path.as_ref(), &config)?;
Ok(())
}
/// Appends one caller to an existing document. The serve limits grow to carry
/// it: the frame to its payload bound, the connections to every caller's slots.
fn add_caller(path: &str, args: &[&str]) -> Result<(), Failure> {
let flags = Flags::parse(
args,
&[&CALLER, &["--max-payload-bytes"]],
&["--allowed-model"],
)?;
let caller = flags.caller()?;
let mut config = load_config(path.as_ref())?;
config.max_frame_bytes = config
.max_frame_bytes
.max(frame_bytes_for_payload(caller.max_payload_bytes));
config.callers.push(caller);
let slots: usize = config
.callers
.iter()
.map(|caller| caller.max_concurrent_requests)
.sum();
config.max_connections = config.max_connections.max(slots);
write_config(path.as_ref(), &config)?;
Ok(())
}
/// Appends one leasing account to an existing document that lends a token.
fn add_lease_caller(path: &str, args: &[&str]) -> Result<(), Failure> {
let flags = Flags::parse(args, &[&LEASE_CALLER, &["--lease-user"]], &[])?;
let mut config = load_config(path.as_ref())?;
config.lease_callers.push(LeaseCallerConfig {
unix_user: flags.text("--lease-user")?,
caller_id: flags.text("--caller-id")?,
unit_instance: flags.text("--lease-unit-instance")?,
reservation_micro_usd: flags.number("--lease-reservation-micro-usd")?,
});
write_config(path.as_ref(), &config)?;
Ok(())
}
#[cfg(all(test, unix))]
mod tests {
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use super::*;
#[test]
fn a_secret_file_owned_by_someone_else_is_not_usable() {
let dir =
std::env::temp_dir().join(format!("claude-connector-main-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("handle.secret");
std::fs::write(&path, "k".repeat(64)).unwrap();
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();
let metadata = std::fs::symlink_metadata(&path).unwrap();
let owner = metadata.uid();
assert!(secret_file_usable(&metadata, owner));
assert!(!secret_file_usable(&metadata, owner.wrapping_add(1)));
std::fs::remove_dir_all(&dir).unwrap();
}
}

2
crates/claude-connector/src/quota.rs
View file

@ -76,7 +76,7 @@ fn unix_ms() -> u64 {
mod tests {
use super::*;
const RESET: u64 = 1_793_307_000_000;
const RESET: u64 = 4_102_444_800_000;
fn quota(window: &str, permille: u32, observed_at: u64) -> PlanQuota {
PlanQuota {

474
crates/claude-connector/tests/config_document.rs
View file

@ -74,6 +74,7 @@ fn valid() -> ClaudeConnectorConfig {
prompt_dir: "/var/lib/claude-connector/prompts".into(),
instance: "persona-a".to_string(),
token_credential: "claude-token".to_string(),
handle_secret_credential: "claude-handle-secret".to_string(),
deadline_ms: 600_000,
max_line_bytes: 4 << 20,
max_total_bytes: 64 << 20,
@ -326,11 +327,10 @@ fn config_refuses_aliases_unpriced_models_and_second_callers() {
Box::new(|c| c.callers[0].allowed_models.clear()),
),
(
"second caller",
"duplicate caller id",
Box::new(|c| {
let mut second = c.callers[0].clone();
second.caller_runtime_id = "other".to_string();
c.callers.push(second);
let again = c.callers[0].clone();
c.callers.push(again);
}),
),
("no caller", Box::new(|c| c.callers.clear())),
@ -538,7 +538,7 @@ fn no_refusal_echoes_a_value() {
}),
Box::new(|c| {
let mut second = c.callers[0].clone();
second.caller_runtime_id = CANARY.to_string();
second.caller_runtime_id = format!(" {CANARY}");
c.callers.push(second);
}),
];
@ -646,6 +646,10 @@ fn init_flags<'a>(scratch: &'a Scratch, extra: &[&'a str]) -> Vec<String> {
"persona-a",
"--token-credential",
"claude-token",
"--handle-secret-credential",
"claude-handle-secret",
"--handle-secret-file",
&path("handle.secret"),
"--caller-id",
"persona-a-caller",
"--connection-key-file",
@ -704,6 +708,11 @@ fn prepare(scratch: &Scratch) -> (Vec<String>, PathBuf) {
)
.unwrap();
std::fs::write(scratch.path("credentials/claude-token"), CANARY).unwrap();
std::fs::write(
scratch.path("credentials/claude-handle-secret"),
"s".repeat(64),
)
.unwrap();
std::fs::write(scratch.path("caller.key"), "caller-secret").unwrap();
std::fs::write(scratch.path("claude"), b"pinned claude").unwrap();
let sha: String = Sha256::digest(b"pinned claude")
@ -749,7 +758,7 @@ fn check_passes_a_document_serve_can_start_and_reports_counts_only() {
);
assert!(checked.status.success(), "{}", shown(&checked));
let text = shown(&checked);
assert!(text.contains("callers: 1") && text.contains("allowed_models: 1"));
assert!(text.contains("callers: 1") && text.contains("lease_callers: 0"));
assert!(text.contains("prices: 1"));
for secret in ["persona-a", HAIKU, CANARY, "caller-secret"] {
assert!(!text.contains(secret), "{text}");
@ -788,6 +797,32 @@ fn check_refuses_exactly_what_serve_refuses_and_names_the_field() {
config.max_frame_bytes = 4099;
write_config(&path, &config).unwrap();
}),
("two callers on one key", "callers", |s, _| {
std::fs::write(s.path("twin.key"), "caller-secret").unwrap();
let config = s.path("connector.cc");
let twin = s.path("twin.key");
let added = run(
&[
"config",
"add-caller",
config.to_str().unwrap(),
"--caller-id",
"twin",
"--connection-key-file",
twin.to_str().unwrap(),
"--connection-key-epoch",
"1",
"--allowed-model",
HAIKU,
"--spend-ceiling-micro-usd",
"1000",
"--spend-window-ms",
"60000",
],
&[],
);
assert!(added.status.success(), "{}", shown(&added));
}),
("too few connections", "connection capacity", |s, _| {
let path = s.path("connector.cc");
let mut config = load_config(&path).unwrap();
@ -1206,3 +1241,430 @@ fn serve_binds_the_lease_socket_and_answers_a_stranger_with_a_refusal() {
outcome.unwrap();
assert!(running);
}
// ---- many callers ----
fn second_caller() -> CallerConfig {
CallerConfig {
caller_runtime_id: "heartbeat".to_string(),
connection_key_file: "/etc/claude-connector/heartbeat.key".into(),
connection_key_epoch: 3,
allowed_models: vec![HAIKU.to_string()],
max_concurrent_requests: 1,
max_payload_bytes: 64 << 10,
max_output_tokens: 1024,
spend_ceiling_micro_usd: 1_000_000,
spend_window_ms: 60_000,
}
}
#[test]
fn several_callers_round_trip_and_each_is_validated_alone() {
let scratch = Scratch::new();
let path = scratch.path("connector.cc");
let mut config = valid();
config.callers.push(second_caller());
write_config(&path, &config).unwrap();
assert_eq!(load_config(&path).unwrap(), config);
let broken: Vec<(&str, Change)> = vec![
(
"caller bounds",
Box::new(|c| c.callers[1].spend_ceiling_micro_usd = 0),
),
(
"connection_key_epoch",
Box::new(|c| c.callers[1].connection_key_epoch = 0),
),
(
"connection_key_file",
Box::new(|c| c.callers[1].connection_key_file = "heartbeat.key".into()),
),
(
"allowed_models",
Box::new(|c| c.callers[1].allowed_models = vec!["claude-opus-9".to_string()]),
),
(
"callers",
Box::new(|c| c.callers[1].caller_runtime_id = c.callers[0].caller_runtime_id.clone()),
),
];
for (field, change) in &broken {
let mut config = valid();
config.callers.push(second_caller());
change(&mut config);
assert_eq!(refused(&config), *field);
}
}
#[test]
fn a_lease_caller_bills_one_configured_caller_within_that_callers_ceiling() {
let mut config = valid();
config.callers.push(second_caller());
let lease = |user: &str, caller: &str, instance: &str, reservation: u64| LeaseCallerConfig {
unix_user: user.to_string(),
caller_id: caller.to_string(),
unit_instance: instance.to_string(),
reservation_micro_usd: reservation,
};
// A lease caller of each caller, each within its own caller's ceiling.
config
.lease_callers
.push(lease("mind-heartbeat", "heartbeat", "heartbeat", 1_000_000));
config.validate().unwrap();
// The first caller's ceiling does not stretch to the second's lease, nor the reverse.
config.lease_callers[1].reservation_micro_usd = 1_000_001;
assert_eq!(refused(&config), "lease_callers");
config.lease_callers[1].reservation_micro_usd = 1;
config.lease_callers[0].reservation_micro_usd = 50_000_001;
assert_eq!(refused(&config), "lease_callers");
config.lease_callers[0].reservation_micro_usd = 50_000_000;
// A lease caller naming no configured caller, or sharing another's account.
config.lease_callers[1].caller_id = "nobody".to_string();
assert_eq!(refused(&config), "lease_callers");
config.lease_callers[1].caller_id = "heartbeat".to_string();
config.lease_callers[1].unix_user = config.lease_callers[0].unix_user.clone();
assert_eq!(refused(&config), "lease_callers");
}
#[test]
fn add_caller_and_add_lease_caller_extend_the_document_and_grow_the_serve_limits() {
let scratch = Scratch::new();
let made = init(
&scratch,
&init_flags(
&scratch,
&[
"--lease-socket",
"/run/claude-connector/lease.sock",
"--lease-user",
"mind-eureka-self",
"--lease-unit-instance",
"eureka",
"--lease-reservation-micro-usd",
"5000000",
],
),
);
assert!(made.status.success(), "{}", shown(&made));
let path = scratch.path("connector.cc");
let path_text = path.to_str().unwrap();
let before = load_config(&path).unwrap();
let key = scratch.path("heartbeat.key");
let added = run(
&[
"config",
"add-caller",
path_text,
"--caller-id",
"heartbeat",
"--connection-key-file",
key.to_str().unwrap(),
"--connection-key-epoch",
"2",
"--allowed-model",
HAIKU,
"--spend-ceiling-micro-usd",
"2000000",
"--spend-window-ms",
"60000",
"--max-concurrent-requests",
"30",
"--max-payload-bytes",
"4194304",
],
&[],
);
assert!(added.status.success(), "{}", shown(&added));
let leased = run(
&[
"config",
"add-lease-caller",
path_text,
"--lease-user",
"mind-heartbeat",
"--caller-id",
"heartbeat",
"--lease-unit-instance",
"heartbeat",
"--lease-reservation-micro-usd",
"1000000",
],
&[],
);
assert!(leased.status.success(), "{}", shown(&leased));
let after = load_config(&path).unwrap();
assert_eq!(after.callers[0], before.callers[0]);
assert_eq!(after.callers.len(), 2);
let heartbeat = &after.callers[1];
assert_eq!(heartbeat.caller_runtime_id, "heartbeat");
assert_eq!(heartbeat.connection_key_epoch, 2);
assert_eq!(heartbeat.spend_ceiling_micro_usd, 2_000_000);
assert_eq!(heartbeat.max_payload_bytes, 4 << 20);
assert_eq!(after.lease_callers[0], before.lease_callers[0]);
assert_eq!(after.lease_callers[1].caller_id, "heartbeat");
// The frame carries the largest payload and the connections every slot.
assert!(after.max_frame_bytes > before.max_frame_bytes);
assert!(after.max_connections >= 34);
assert_eq!(after.prices, before.prices);
assert_eq!(
after.handle_secret_credential,
before.handle_secret_credential
);
// Repeating either verb is refused and leaves the document as it was.
let again = run(
&[
"config",
"add-caller",
path_text,
"--caller-id",
"heartbeat",
"--connection-key-file",
key.to_str().unwrap(),
"--connection-key-epoch",
"2",
"--allowed-model",
HAIKU,
"--spend-ceiling-micro-usd",
"2000000",
"--spend-window-ms",
"60000",
],
&[],
);
assert!(!again.status.success());
assert_eq!(load_config(&path).unwrap(), after);
let unleased = run(
&[
"config",
"add-lease-caller",
path_text,
"--lease-user",
"mind-heartbeat",
"--caller-id",
"heartbeat",
"--lease-unit-instance",
"heartbeat",
"--lease-reservation-micro-usd",
"1000000",
],
&[],
);
assert!(!unleased.status.success());
assert_eq!(load_config(&path).unwrap(), after);
}
#[test]
fn add_lease_caller_needs_a_document_that_lends_a_token() {
let scratch = Scratch::new();
let made = init(&scratch, &init_flags(&scratch, &[]));
assert!(made.status.success(), "{}", shown(&made));
let path = scratch.path("connector.cc");
let leased = run(
&[
"config",
"add-lease-caller",
path.to_str().unwrap(),
"--lease-user",
"mind-heartbeat",
"--caller-id",
"persona-a-caller",
"--lease-unit-instance",
"heartbeat",
"--lease-reservation-micro-usd",
"1000000",
],
&[],
);
assert!(!leased.status.success());
assert!(load_config(&path).unwrap().lease_callers.is_empty());
}
#[cfg(unix)]
#[test]
fn init_creates_the_handle_secret_once_private_and_never_shows_it() {
use std::os::unix::fs::PermissionsExt;
let scratch = Scratch::new();
let secret = scratch.path("handle.secret");
let made = init(&scratch, &init_flags(&scratch, &[]));
assert!(made.status.success(), "{}", shown(&made));
let metadata = std::fs::metadata(&secret).unwrap();
assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
let text = std::fs::read_to_string(&secret).unwrap();
assert_eq!(text.len(), 64);
assert!(text.bytes().all(|b| b.is_ascii_hexdigit()));
assert!(!shown(&made).contains(&text));
// A second init keeps the secret the first one made, since handles depend on it.
std::fs::remove_file(scratch.path("connector.cc")).unwrap();
let again = init(&scratch, &init_flags(&scratch, &[]));
assert!(again.status.success(), "{}", shown(&again));
assert_eq!(std::fs::read_to_string(&secret).unwrap(), text);
// A refused document leaves no secret file behind.
let fresh = Scratch::new();
let flags = init_flags(&fresh, &["--max-payload-bytes", "1"]);
let refused = init(&fresh, &flags);
assert!(!refused.status.success());
assert!(!fresh.path("handle.secret").exists());
}
#[cfg(unix)]
#[test]
fn check_wants_the_handle_secret_credential_and_names_it() {
let scratch = Scratch::new();
let (flags, credentials) = prepare(&scratch);
let made = init(&scratch, &flags);
assert!(made.status.success(), "{}", shown(&made));
let config = scratch.path("connector.cc");
let path = config.to_str().unwrap();
for body in [None, Some("too-short")] {
let secret = credentials.join("claude-handle-secret");
match body {
None => std::fs::remove_file(&secret).unwrap(),
Some(text) => std::fs::write(&secret, text).unwrap(),
}
let checked = run(
&["config", "check", path],
&[("CREDENTIALS_DIRECTORY", &credentials)],
);
assert!(!checked.status.success());
let text = shown(&checked);
assert!(text.contains("handle_secret_credential"), "{text}");
assert!(!text.contains("too-short"), "{text}");
}
}
/// A document written before the handle secret had a field cannot name where
/// the secret is, so it is not a document this connector serves from: it is
/// refused as a store fault, to be authored again with `config init`.
#[test]
fn a_document_without_the_handle_secret_field_is_refused_as_a_store() {
let scratch = Scratch::new();
let path = scratch.path("old.cc");
let mut payload = rmp_serde::to_vec(&valid()).unwrap();
// array16(23) -> array16(22), and drop the trailing 20-byte fixstr.
assert_eq!(payload[..3], [0xdc, 0, 23]);
payload[2] = 22;
payload.truncate(payload.len() - 21);
store_raw(
&path,
"runtime",
ClaudeConnectorConfig::TYPE,
Some(ClaudeConnectorConfig::TYPE),
payload,
);
assert!(matches!(load_config(&path), Err(ConfigError::Store)));
}
#[test]
fn the_handle_secret_credential_is_a_name_of_its_own() {
for name in ["claude-token", "a/b", "", ".."] {
let mut config = valid();
config.handle_secret_credential = name.to_string();
assert_eq!(refused(&config), "handle_secret_credential", "{name:?}");
}
}
#[cfg(unix)]
#[test]
fn init_refuses_a_secret_file_it_cannot_create_and_writes_no_document() {
let scratch = Scratch::new();
let mut flags = init_flags(&scratch, &[]);
let at = flags
.iter()
.position(|f| f == "--handle-secret-file")
.unwrap();
flags[at + 1] = scratch
.path("absent/handle.secret")
.to_string_lossy()
.into_owned();
let made = init(&scratch, &flags);
assert!(!made.status.success());
assert!(!scratch.path("connector.cc").exists());
}
#[cfg(unix)]
#[test]
fn config_init_keeps_an_existing_secret_file_only_if_it_is_usable() {
use std::os::unix::fs::{PermissionsExt, symlink};
let write = |path: &Path, bytes: usize, mode: u32| {
std::fs::write(path, "k".repeat(bytes)).unwrap();
std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode)).unwrap();
};
type Plant = fn(&Scratch, &Path);
let refused: Vec<(&str, Plant)> = vec![
("world readable", |_, p| {
std::fs::write(p, "k".repeat(64)).unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o644)).unwrap();
}),
("group writable", |_, p| {
std::fs::write(p, "k".repeat(64)).unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o660)).unwrap();
}),
("read only", |_, p| {
std::fs::write(p, "k".repeat(64)).unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o400)).unwrap();
}),
("one byte short", |_, p| {
std::fs::write(p, "k".repeat(31)).unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
}),
("empty", |_, p| {
std::fs::write(p, "").unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
}),
("a directory", |_, p| {
std::fs::create_dir(p).unwrap();
std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600)).unwrap();
}),
("a link to a good file", |s, p| {
let target = s.path("elsewhere.secret");
std::fs::write(&target, "k".repeat(64)).unwrap();
std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o600)).unwrap();
symlink(&target, p).unwrap();
}),
];
for (name, plant) in refused {
let scratch = Scratch::new();
let flags = init_flags(&scratch, &[]);
let secret = scratch.path("handle.secret");
plant(&scratch, &secret);
let before = std::fs::symlink_metadata(&secret).unwrap();
let made = init(&scratch, &flags);
assert!(!made.status.success(), "{name} was accepted");
assert!(shown(&made).contains("--handle-secret-file"), "{name}");
assert!(!shown(&made).contains("kkkk"), "{name}");
assert!(
!scratch.path("connector.cc").exists(),
"{name} left a document"
);
let after = std::fs::symlink_metadata(&secret).unwrap();
assert_eq!(
before.permissions().mode(),
after.permissions().mode(),
"{name}"
);
assert_eq!(before.len(), after.len(), "{name}");
assert_eq!(
before.file_type().is_symlink(),
after.file_type().is_symlink(),
"{name}"
);
}
// The smallest usable file is accepted and left exactly as it was.
let scratch = Scratch::new();
let flags = init_flags(&scratch, &[]);
let secret = scratch.path("handle.secret");
write(&secret, 32, 0o600);
let made = init(&scratch, &flags);
assert!(made.status.success(), "{}", shown(&made));
assert_eq!(std::fs::read_to_string(&secret).unwrap(), "k".repeat(32));
assert!(scratch.path("connector.cc").exists());
}

23
crates/model-connector/src/serve.rs
View file

@ -19,7 +19,7 @@ use crate::envelope::{
use crate::readout::PlanQuotaSnapshot;
use crate::service::{
CallerAdmission, PriceTable, ProviderCall, SpendCeiling, SpendWindowMs, TransportAdmission,
TransportService,
TransportService, distinct_admissions,
};
/// The only provider seam of the connector kit. A connector binary implements
@ -87,6 +87,7 @@ impl ServeLimits {
if admissions.is_empty() {
return Err(ServeError::InvalidConfig("callers"));
}
distinct_admissions(admissions)?;
let slots = admissions
.iter()
.try_fold(0_usize, |total, caller| {
@ -950,6 +951,26 @@ mod tests {
));
}
#[test]
fn the_limits_refuse_callers_that_share_a_key_or_an_id() {
let limits = limits();
let shared_key = [
caller("one", "shared-key", 1),
caller("two", "shared-key", 1),
];
assert!(matches!(
limits.validate(&shared_key),
Err(ServeError::Service(ServiceError::SharedCallerKey))
));
let shared_id = [caller("one", "key-a", 1), caller("one", "key-b", 1)];
assert!(matches!(
limits.validate(&shared_id),
Err(ServeError::Service(ServiceError::DuplicateCaller))
));
let distinct = [caller("one", "key-a", 1), caller("two", "key-b", 1)];
assert!(limits.validate(&distinct).is_ok());
}
#[test]
fn only_loopback_peers_are_admitted() {
assert!(peer_admitted(IpAddr::from([127, 0, 0, 1])));

63
crates/model-connector/src/service.rs
View file

@ -47,6 +47,24 @@ pub struct SpendCeiling(pub u64);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SpendWindowMs(pub u64);
/// The rule that admitted callers are distinct: no two share a connection key
/// (a key names its caller), and no two share a caller id. `TransportService`
/// opens only under it, and a readiness check that must agree with opening
/// asks the same function.
pub fn distinct_admissions(admissions: &[CallerAdmission]) -> Result<(), ServiceError> {
let mut keys = HashSet::new();
let mut ids = HashSet::new();
for admission in admissions {
if !keys.insert(admission.security.0) {
return Err(ServiceError::SharedCallerKey);
}
if !ids.insert(admission.caller_runtime_id.as_str()) {
return Err(ServiceError::DuplicateCaller);
}
}
Ok(())
}
pub struct CallerAdmission {
caller_runtime_id: String,
connection_key_epoch: u32,
@ -388,6 +406,12 @@ impl ProviderCall {
&self.request
}
/// The caller this call was admitted for: the caller whose key sealed the
/// invocation, never a name the request carries.
pub fn caller_runtime_id(&self) -> &str {
&self.receipt.caller_runtime_id
}
/// How many foreign reasoning items the service removed from the request.
pub fn foreign_reasoning_dropped(&self) -> u32 {
self.receipt.foreign_reasoning_dropped
@ -533,17 +557,12 @@ impl TransportService {
{
return Err(ServiceError::InvalidAdmission);
}
let mut callers = HashMap::new();
let mut keys = HashSet::new();
for admission in admissions {
if !keys.insert(admission.security.0) {
return Err(ServiceError::SharedCallerKey);
}
let caller_runtime_id = admission.caller_runtime_id.clone();
if callers.insert(caller_runtime_id, admission).is_some() {
return Err(ServiceError::DuplicateCaller);
}
}
let admissions: Vec<CallerAdmission> = admissions.into_iter().collect();
distinct_admissions(&admissions)?;
let callers: HashMap<String, CallerAdmission> = admissions
.into_iter()
.map(|admission| (admission.caller_runtime_id.clone(), admission))
.collect();
if callers.is_empty() {
return Err(ServiceError::InvalidAdmission);
}
@ -3800,6 +3819,28 @@ mod tests {
assert_eq!((idle.spent_micro_usd, idle.reserved_micro_usd), (0, 0));
}
#[test]
fn any_admitted_caller_reads_every_callers_row() {
let store = TestStore::new();
let mut service = open_service(&store, 4);
let claim = claim_at(&mut service, "request-1", 1_000);
let result = text_result(claim.call(), "done");
service.complete(*claim, result, FRAME).unwrap();
// Ghostlight asks, with its own key, and sees epiphany's spend.
let readout = readout_as(
&service,
"ghostlight-yggdrasil",
"ghostlight-distinct-test-key",
1_500,
FRAME,
)
.unwrap();
assert_eq!(row(&readout, CALLER).spent_micro_usd, FIXTURE_USAGE_COST);
assert_eq!(row(&readout, "ghostlight-yggdrasil").spent_micro_usd, 0);
// A key that belongs to no admitted caller reads nothing.
assert!(readout_as(&service, "ghostlight-yggdrasil", KEY, 1_500, FRAME).is_err());
}
#[test]
fn a_claim_a_crash_left_reads_as_spent_not_reserved() {
let store = TestStore::new();