michel/clusterflux-public
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Attribute agent workflow dispatch

Browse source
This commit is contained in:
Michel Paulissen 2026-07-03 21:56:48 +02:00
parent 28779a2866
commit c514ab7f81
5 changed files with 503 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

34
crates/disasmer-coordinator/src/lib.rs
View file

@ -473,6 +473,40 @@ impl Coordinator {
Ok(record.clone())
}
pub fn authorize_agent_project_run(
&self,
tenant: &TenantId,
project: &ProjectId,
agent: &AgentId,
public_key_fingerprint: &Digest,
) -> Result<AgentPublicKeyRecord, CoordinatorError> {
let record = self
.durable
.agent_public_keys
.get(&(tenant.clone(), project.clone(), agent.clone()))
.ok_or_else(|| {
CoordinatorError::Unauthorized(
"agent public key is not registered for this tenant/project".to_owned(),
)
})?;
if record.revoked {
return Err(CoordinatorError::Unauthorized(
"agent public key has been revoked".to_owned(),
));
}
if &record.public_key_fingerprint != public_key_fingerprint {
return Err(CoordinatorError::Unauthorized(
"agent public key fingerprint does not match the registered key".to_owned(),
));
}
if !record.scopes.iter().any(|scope| scope == "project:run") {
return Err(CoordinatorError::Unauthorized(
"agent public key is not scoped for project runs".to_owned(),
));
}
Ok(record.clone())
}
pub fn start_process(
&mut self,
tenant: TenantId,

352
crates/disasmer-coordinator/src/service.rs
View file

@ -147,7 +147,12 @@ pub enum CoordinatorRequest {
LaunchTask {
tenant: String,
project: String,
actor_user: String,
#[serde(default)]
actor_user: Option<String>,
#[serde(default)]
actor_agent: Option<String>,
#[serde(default)]
agent_public_key_fingerprint: Option<Digest>,
process: String,
task: String,
environment: Option<EnvironmentRequirements>,
@ -193,6 +198,12 @@ pub enum CoordinatorRequest {
StartProcess {
tenant: String,
project: String,
#[serde(default)]
actor_user: Option<String>,
#[serde(default)]
actor_agent: Option<String>,
#[serde(default)]
agent_public_key_fingerprint: Option<Digest>,
process: String,
#[serde(default)]
restart: bool,
@ -426,6 +437,17 @@ pub struct DebugAuditEvent {
pub used_debug_read_bytes: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkflowActor {
pub kind: String,
pub user: Option<UserId>,
pub agent: Option<AgentId>,
pub credential_kind: CredentialKind,
pub public_key_fingerprint: Option<Digest>,
pub authenticated_without_browser: bool,
pub scopes: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TaskAssignment {
pub tenant: TenantId,
@ -539,6 +561,7 @@ pub enum CoordinatorResponse {
TaskLaunched {
process: ProcessId,
task: TaskId,
actor: WorkflowActor,
placement: Placement,
assignment: TaskAssignment,
},
@ -560,6 +583,7 @@ pub enum CoordinatorResponse {
ProcessStarted {
process: ProcessId,
epoch: u64,
actor: WorkflowActor,
},
NodeReconnected {
node: NodeId,
@ -1222,6 +1246,8 @@ impl CoordinatorService {
tenant,
project,
actor_user,
actor_agent,
agent_public_key_fingerprint,
process,
task,
environment,
@ -1240,9 +1266,13 @@ impl CoordinatorService {
let project = ProjectId::new(project);
let process = ProcessId::new(process);
let task = TaskId::new(task);
let actor = UserId::new(actor_user);
self.coordinator
.upsert_user(tenant.clone(), actor, CredentialKind::BrowserSession);
let actor = self.workflow_actor(
&tenant,
&project,
actor_user,
actor_agent,
agent_public_key_fingerprint,
)?;
let active = self.coordinator.active_process(&process).ok_or_else(|| {
CoordinatorError::Unauthorized(
"task launch requires an active coordinator-side virtual process"
@ -1309,6 +1339,7 @@ impl CoordinatorService {
Ok(CoordinatorResponse::TaskLaunched {
process,
task,
actor,
placement,
assignment,
})
@ -1445,6 +1476,9 @@ impl CoordinatorService {
CoordinatorRequest::StartProcess {
tenant,
project,
actor_user,
actor_agent,
agent_public_key_fingerprint,
process,
restart,
} => {
@ -1452,6 +1486,13 @@ impl CoordinatorService {
let project = ProjectId::new(project);
let process = ProcessId::new(process);
self.coordinator.ensure_tenant_active(&tenant)?;
let actor = self.workflow_actor(
&tenant,
&project,
actor_user,
actor_agent,
agent_public_key_fingerprint,
)?;
if let Some(active) = self
.coordinator
.active_process_for_project(&tenant, &project)
@ -1492,6 +1533,7 @@ impl CoordinatorService {
Ok(CoordinatorResponse::ProcessStarted {
process,
epoch: self.coordinator.coordinator_epoch(),
actor,
})
}
CoordinatorRequest::ReconnectNode {
@ -2245,6 +2287,55 @@ impl CoordinatorService {
Ok(event)
}
fn workflow_actor(
&mut self,
tenant: &TenantId,
project: &ProjectId,
actor_user: Option<String>,
actor_agent: Option<String>,
agent_public_key_fingerprint: Option<Digest>,
) -> Result<WorkflowActor, CoordinatorServiceError> {
if let Some(agent) = actor_agent {
let agent = AgentId::new(agent);
let public_key_fingerprint = agent_public_key_fingerprint.ok_or_else(|| {
CoordinatorError::Unauthorized(
"agent workflow dispatch requires public key fingerprint".to_owned(),
)
})?;
let record = self.coordinator.authorize_agent_project_run(
tenant,
project,
&agent,
&public_key_fingerprint,
)?;
return Ok(WorkflowActor {
kind: "agent".to_owned(),
user: Some(record.user),
agent: Some(agent),
credential_kind: CredentialKind::PublicKey,
public_key_fingerprint: Some(public_key_fingerprint),
authenticated_without_browser: true,
scopes: record.scopes,
});
}
let actor = UserId::new(actor_user.unwrap_or_else(|| "user".to_owned()));
self.coordinator.upsert_user(
tenant.clone(),
actor.clone(),
CredentialKind::BrowserSession,
);
Ok(WorkflowActor {
kind: "user".to_owned(),
user: Some(actor),
agent: None,
credential_kind: CredentialKind::BrowserSession,
public_key_fingerprint: None,
authenticated_without_browser: false,
scopes: vec!["project:read".to_owned(), "project:run".to_owned()],
})
}
fn export_endpoint(
&self,
node: &NodeId,
@ -2681,6 +2772,16 @@ mod tests {
}
}
fn assert_agent_workflow_actor(actor: &WorkflowActor, fingerprint: &Digest) {
assert_eq!(actor.kind, "agent");
assert_eq!(actor.user, Some(UserId::from("user")));
assert_eq!(actor.agent, Some(AgentId::from("agent-ci")));
assert_eq!(actor.credential_kind, CredentialKind::PublicKey);
assert_eq!(actor.public_key_fingerprint.as_ref(), Some(fingerprint));
assert!(actor.authenticated_without_browser);
assert!(actor.scopes.iter().any(|scope| scope == "project:run"));
}
#[test]
fn service_creates_selects_and_lists_signed_in_user_projects() {
let mut service = CoordinatorService::new(7);
@ -2812,6 +2913,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -2926,6 +3030,160 @@ mod tests {
assert!(record.revoked);
}
#[test]
fn service_runs_agent_workflows_with_scoped_key_attribution() {
let mut service = CoordinatorService::new(7);
let agent_fingerprint = Digest::sha256("agent-key-v1");
service
.handle_request(CoordinatorRequest::RegisterAgentPublicKey {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
agent: "agent-ci".to_owned(),
public_key: "agent-key-v1".to_owned(),
})
.unwrap();
let bad_key = service
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: Some("agent-ci".to_owned()),
agent_public_key_fingerprint: Some(Digest::sha256("other-key")),
process: "vp-agent-bad-key".to_owned(),
restart: false,
})
.unwrap_err();
assert!(bad_key.to_string().contains("fingerprint"));
service
.handle_request(CoordinatorRequest::AttachNode {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
node: "worker-linux".to_owned(),
public_key: "worker-linux-public-key".to_owned(),
})
.unwrap();
service
.handle_request(CoordinatorRequest::ReportNodeCapabilities {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
node: "worker-linux".to_owned(),
capabilities: linux_capabilities(),
cached_environment_digests: Vec::new(),
dependency_cache_digests: Vec::new(),
source_snapshots: Vec::new(),
artifact_locations: Vec::new(),
direct_connectivity: true,
online: true,
})
.unwrap();
let CoordinatorResponse::ProcessStarted {
actor: start_actor, ..
} = service
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: Some("agent-ci".to_owned()),
agent_public_key_fingerprint: Some(agent_fingerprint.clone()),
process: "vp-agent".to_owned(),
restart: false,
})
.unwrap()
else {
panic!("expected agent-authenticated process start");
};
assert_agent_workflow_actor(&start_actor, &agent_fingerprint);
let CoordinatorResponse::TaskLaunched {
actor: launch_actor,
placement,
assignment,
..
} = service
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: Some("agent-ci".to_owned()),
agent_public_key_fingerprint: Some(agent_fingerprint.clone()),
process: "vp-agent".to_owned(),
task: "compile-linux".to_owned(),
environment: None,
environment_digest: None,
required_capabilities: vec![Capability::Command],
dependency_cache: None,
source_snapshot: None,
required_artifacts: Vec::new(),
quota_available: true,
policy_allowed: true,
command: "cargo".to_owned(),
command_args: vec!["test".to_owned()],
artifact_path: "/vfs/artifacts/dap-output.txt".to_owned(),
})
.unwrap()
else {
panic!("expected agent-authenticated task launch");
};
assert_agent_workflow_actor(&launch_actor, &agent_fingerprint);
assert_eq!(placement.node, NodeId::from("worker-linux"));
assert_eq!(assignment.node, NodeId::from("worker-linux"));
let quota_error = service
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: Some("agent-ci".to_owned()),
agent_public_key_fingerprint: Some(agent_fingerprint.clone()),
process: "vp-agent".to_owned(),
task: "compile-linux-quota".to_owned(),
environment: None,
environment_digest: None,
required_capabilities: vec![Capability::Command],
dependency_cache: None,
source_snapshot: None,
required_artifacts: Vec::new(),
quota_available: false,
policy_allowed: true,
command: "cargo".to_owned(),
command_args: vec!["test".to_owned()],
artifact_path: "/vfs/artifacts/quota.txt".to_owned(),
})
.unwrap_err();
assert!(quota_error
.to_string()
.contains("quota unavailable for placement"));
let policy_error = service
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: Some("agent-ci".to_owned()),
agent_public_key_fingerprint: Some(agent_fingerprint),
process: "vp-agent".to_owned(),
task: "compile-linux-policy".to_owned(),
environment: None,
environment_digest: None,
required_capabilities: vec![Capability::Command],
dependency_cache: None,
source_snapshot: None,
required_artifacts: Vec::new(),
quota_available: true,
policy_allowed: false,
command: "cargo".to_owned(),
command_args: vec!["test".to_owned()],
artifact_path: "/vfs/artifacts/policy.txt".to_owned(),
})
.unwrap_err();
assert!(policy_error.to_string().contains("policy denied placement"));
}
#[test]
fn service_attaches_node_starts_process_and_records_scoped_task_event() {
let mut service = CoordinatorService::new(7);
@ -2944,6 +3202,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -2952,7 +3213,16 @@ mod tests {
started,
CoordinatorResponse::ProcessStarted {
process: ProcessId::from("process"),
epoch: 7
epoch: 7,
actor: WorkflowActor {
kind: "user".to_owned(),
user: Some(UserId::from("user")),
agent: None,
credential_kind: CredentialKind::BrowserSession,
public_key_fingerprint: None,
authenticated_without_browser: false,
scopes: vec!["project:read".to_owned(), "project:run".to_owned()],
}
}
);
@ -3178,6 +3448,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -3294,6 +3567,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -3390,6 +3666,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -3467,7 +3746,9 @@ mod tests {
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
actor_user: Some("user".to_owned()),
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
task: "task".to_owned(),
environment: None,
@ -3599,6 +3880,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -3621,7 +3905,9 @@ mod tests {
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
actor_user: Some("user".to_owned()),
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
task: task.to_owned(),
environment: None,
@ -3687,7 +3973,9 @@ mod tests {
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
actor_user: Some("user".to_owned()),
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
task: "package-linux".to_owned(),
environment: None,
@ -3715,6 +4003,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process-a".to_owned(),
restart: false,
})
@ -3728,6 +4019,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process-a".to_owned(),
restart: false,
})
@ -3740,6 +4034,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process-b".to_owned(),
restart: true,
})
@ -3752,6 +4049,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process-a".to_owned(),
restart: true,
})
@ -3761,6 +4061,15 @@ mod tests {
CoordinatorResponse::ProcessStarted {
process: ProcessId::from("process-a"),
epoch: 7,
actor: WorkflowActor {
kind: "user".to_owned(),
user: Some(UserId::from("user")),
agent: None,
credential_kind: CredentialKind::BrowserSession,
public_key_fingerprint: None,
authenticated_without_browser: false,
scopes: vec!["project:read".to_owned(), "project:run".to_owned()],
}
}
);
}
@ -3780,6 +4089,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -4129,6 +4441,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -4304,6 +4619,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})
@ -4556,6 +4874,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "vp-control".to_owned(),
restart: false,
})
@ -4569,11 +4890,14 @@ mod tests {
task,
placement,
assignment,
..
} = service
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
actor_user: Some("user".to_owned()),
actor_agent: None,
agent_public_key_fingerprint: None,
process: "vp-control".to_owned(),
task: "compile-linux".to_owned(),
environment: None,
@ -4634,6 +4958,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "vp-control".to_owned(),
restart: false,
})
@ -4643,7 +4970,9 @@ mod tests {
.handle_request(CoordinatorRequest::LaunchTask {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
actor_user: "user".to_owned(),
actor_user: Some("user".to_owned()),
actor_agent: None,
agent_public_key_fingerprint: None,
process: "vp-control".to_owned(),
task: "compile-linux".to_owned(),
environment: None,
@ -4768,6 +5097,9 @@ mod tests {
.handle_request(CoordinatorRequest::StartProcess {
tenant: "tenant-b".to_owned(),
project: "project-b".to_owned(),
actor_user: None,
actor_agent: None,
agent_public_key_fingerprint: None,
process: "process".to_owned(),
restart: false,
})