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

Public dry run dryrun-6341cce4cf9f

Browse source 
Source commit: 6341cce4cf9f801a3bfe2248d6e7e6a119cea715

Public tree identity: sha256:d909cba912925e66131fc55fe33774a2b26b1ff2c12b53ec4ef217c9788d8b7a
This commit is contained in:
Michel Paulissen 2026-07-03 20:50:16 +02:00
parent 84ef57d84b
commit 9f2d6ec010
7 changed files with 654 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,8 +1,9 @@
use std::collections::{BTreeMap, BTreeSet};
use disasmer_core::{
Actor, AuthContext, Authorization, CredentialKind, Digest, EnrollmentError, EnrollmentGrant,
NodeCredential, NodeId, ProcessId, ProjectId, SourceProviderKind, TenantId, UserId,
Actor, AgentId, AuthContext, Authorization, CredentialKind, Digest, EnrollmentError,
EnrollmentGrant, NodeCredential, NodeId, ProcessId, ProjectId, SourceProviderKind, TenantId,
UserId,
};
use serde::{Deserialize, Serialize};
use thiserror::Error;
@ -78,6 +79,21 @@ pub struct ProjectPermissionRecord {
pub can_debug: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AgentPublicKeyRecord {
pub tenant: TenantId,
pub project: ProjectId,
pub user: UserId,
pub agent: AgentId,
pub public_key: String,
pub public_key_fingerprint: Digest,
pub version: u64,
pub revoked: bool,
pub scopes: Vec<String>,
pub human_account_creation_privilege: bool,
pub browser_interaction_required_each_run: bool,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DurableState {
pub tenants: BTreeMap<TenantId, TenantRecord>,
@ -89,6 +105,7 @@ pub struct DurableState {
BTreeMap<(TenantId, ProjectId, String), SourceProviderConfigRecord>,
pub service_policy_records: BTreeMap<(TenantId, String), ServicePolicyRecord>,
pub project_permissions: BTreeMap<(TenantId, ProjectId, UserId), ProjectPermissionRecord>,
pub agent_public_keys: BTreeMap<(TenantId, ProjectId, AgentId), AgentPublicKeyRecord>,
}
pub trait DurableStore {
@ -329,6 +346,105 @@ impl Coordinator {
);
}
pub fn register_agent_public_key(
&mut self,
tenant: TenantId,
project: ProjectId,
user: UserId,
agent: AgentId,
public_key: impl Into<String>,
) -> AgentPublicKeyRecord {
let key = (tenant.clone(), project.clone(), agent.clone());
let version = self
.durable
.agent_public_keys
.get(&key)
.map_or(1, |record| record.version.saturating_add(1));
let public_key = public_key.into();
let public_key_fingerprint = Digest::sha256(&public_key);
let record = AgentPublicKeyRecord {
tenant: tenant.clone(),
project: project.clone(),
user: user.clone(),
agent: agent.clone(),
public_key,
public_key_fingerprint: public_key_fingerprint.clone(),
version,
revoked: false,
scopes: vec!["project:read".to_owned(), "project:run".to_owned()],
human_account_creation_privilege: false,
browser_interaction_required_each_run: false,
};
self.durable.agent_public_keys.insert(key, record.clone());
let subject = format!("agent:{tenant}:{project}:{agent}");
self.durable.credentials.insert(
subject.clone(),
CredentialRecord {
subject,
tenant,
project: Some(project),
kind: CredentialKind::PublicKey,
public_key_fingerprint: Some(public_key_fingerprint),
},
);
record
}
pub fn list_agent_public_keys(&self, context: &AuthContext) -> Vec<AgentPublicKeyRecord> {
self.durable
.agent_public_keys
.values()
.filter(|record| record.tenant == context.tenant && record.project == context.project)
.filter(|record| match &context.actor {
Actor::User(user) => &record.user == user,
Actor::Agent(agent) => &record.agent == agent,
Actor::Node(_) | Actor::Task(_) => false,
})
.cloned()
.collect()
}
pub fn revoke_agent_public_key(
&mut self,
context: &AuthContext,
agent: &AgentId,
) -> Result<AgentPublicKeyRecord, CoordinatorError> {
let key = (
context.tenant.clone(),
context.project.clone(),
agent.clone(),
);
let record = self
.durable
.agent_public_keys
.get_mut(&key)
.ok_or_else(|| {
CoordinatorError::Unauthorized(
"agent public key is not registered for this project".to_owned(),
)
})?;
match &context.actor {
Actor::User(user) if &record.user == user => {}
Actor::User(_) => {
return Err(CoordinatorError::Unauthorized(
"agent public key is outside the signed-in user scope".to_owned(),
));
}
_ => {
return Err(CoordinatorError::Unauthorized(
"agent public-key revocation requires a user identity".to_owned(),
));
}
}
record.revoked = true;
let subject = format!(
"agent:{}:{}:{}",
context.tenant, context.project, record.agent
);
self.durable.credentials.remove(&subject);
Ok(record.clone())
}
pub fn start_process(
&mut self,
tenant: TenantId,
@ -660,6 +776,80 @@ mod tests {
assert_eq!(projects[0].id, ProjectId::from("project-a"));
}
#[test]
fn agent_public_keys_are_project_user_scoped_and_restart_durable() {
let mut store = InMemoryDurableStore::default();
let mut coordinator = Coordinator::boot(&store, 1);
coordinator.upsert_tenant(TenantId::from("tenant"));
coordinator.upsert_user(
TenantId::from("tenant"),
UserId::from("user"),
CredentialKind::CliDeviceSession,
);
coordinator.upsert_project(TenantId::from("tenant"), ProjectId::from("project"), "demo");
let registered = coordinator.register_agent_public_key(
TenantId::from("tenant"),
ProjectId::from("project"),
UserId::from("user"),
AgentId::from("agent-ci"),
"agent-key-v1",
);
assert_eq!(registered.version, 1);
assert!(!registered.revoked);
assert!(!registered.human_account_creation_privilege);
assert_eq!(registered.scopes, vec!["project:read", "project:run"]);
let rotated = coordinator.register_agent_public_key(
TenantId::from("tenant"),
ProjectId::from("project"),
UserId::from("user"),
AgentId::from("agent-ci"),
"agent-key-v2",
);
assert_eq!(rotated.version, 2);
assert_eq!(rotated.public_key, "agent-key-v2");
coordinator.persist(&mut store);
let mut restarted = Coordinator::boot(&store, 2);
let context = AuthContext {
tenant: TenantId::from("tenant"),
project: ProjectId::from("project"),
actor: Actor::User(UserId::from("user")),
};
let listed = restarted.list_agent_public_keys(&context);
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].agent, AgentId::from("agent-ci"));
assert_eq!(listed[0].version, 2);
let foreign_user_context = AuthContext {
tenant: TenantId::from("tenant"),
project: ProjectId::from("project"),
actor: Actor::User(UserId::from("other-user")),
};
assert!(restarted
.list_agent_public_keys(&foreign_user_context)
.is_empty());
let foreign_project_context = AuthContext {
tenant: TenantId::from("tenant"),
project: ProjectId::from("other-project"),
actor: Actor::User(UserId::from("user")),
};
assert!(restarted
.list_agent_public_keys(&foreign_project_context)
.is_empty());
let revoked = restarted
.revoke_agent_public_key(&context, &AgentId::from("agent-ci"))
.unwrap();
assert!(revoked.revoked);
assert!(!restarted
.durable
.credentials
.contains_key("agent:tenant:project:agent-ci"));
}
#[test]
fn node_cannot_claim_process_outside_authorized_scope() {
let store = InMemoryDurableStore::default();

45
crates/disasmer-coordinator/src/postgres_store.rs
View file

@ -3,7 +3,7 @@ use serde::{de::DeserializeOwned, Serialize};
use thiserror::Error;
use crate::{
CredentialRecord, DurableState, FallibleDurableStore, NodeIdentityRecord,
AgentPublicKeyRecord, CredentialRecord, DurableState, FallibleDurableStore, NodeIdentityRecord,
ProjectPermissionRecord, ProjectRecord, ServicePolicyRecord, SourceProviderConfigRecord,
TenantRecord, UserRecord,
};
@ -41,6 +41,11 @@ pub const POSTGRES_DURABLE_TABLES: &[PostgresTable] = &[
durable_record: "credentials",
restart_surviving: true,
},
PostgresTable {
name: "disasmer_agent_public_keys",
durable_record: "agent public keys",
restart_surviving: true,
},
PostgresTable {
name: "disasmer_source_provider_configs",
durable_record: "source-provider configuration",
@ -148,6 +153,18 @@ impl FallibleDurableStore for PostgresDurableStore {
)? {
state.credentials.insert(record.subject.clone(), record);
}
for record in self.query_records::<AgentPublicKeyRecord>(
"SELECT record FROM disasmer_agent_public_keys ORDER BY tenant_id, project_id, agent_id",
)? {
state.agent_public_keys.insert(
(
record.tenant.clone(),
record.project.clone(),
record.agent.clone(),
),
record,
);
}
for record in self.query_records::<SourceProviderConfigRecord>(
"SELECT record FROM disasmer_source_provider_configs ORDER BY tenant_id, project_id, provider_key",
)? {
@ -187,6 +204,7 @@ impl FallibleDurableStore for PostgresDurableStore {
DELETE FROM disasmer_project_permissions;
DELETE FROM disasmer_service_policy_records;
DELETE FROM disasmer_source_provider_configs;
DELETE FROM disasmer_agent_public_keys;
DELETE FROM disasmer_credentials;
DELETE FROM disasmer_node_identities;
DELETE FROM disasmer_projects;
@ -236,6 +254,19 @@ impl FallibleDurableStore for PostgresDurableStore {
&[&record.subject.as_str(), &record.tenant.as_str(), &project_id, &value],
)?;
}
for record in state.agent_public_keys.values() {
let value = Self::record_value(record)?;
tx.execute(
"INSERT INTO disasmer_agent_public_keys (tenant_id, project_id, user_id, agent_id, record) VALUES ($1, $2, $3, $4, $5)",
&[
&record.tenant.as_str(),
&record.project.as_str(),
&record.user.as_str(),
&record.agent.as_str(),
&value,
],
)?;
}
for ((_, _, provider_key), record) in &state.source_provider_configs {
let value = Self::record_value(record)?;
tx.execute(
@ -305,6 +336,15 @@ CREATE TABLE IF NOT EXISTS disasmer_credentials (
record JSONB NOT NULL
);
CREATE TABLE IF NOT EXISTS disasmer_agent_public_keys (
tenant_id TEXT NOT NULL REFERENCES disasmer_tenants(tenant_id) ON DELETE CASCADE,
project_id TEXT NOT NULL REFERENCES disasmer_projects(project_id) ON DELETE CASCADE,
user_id TEXT NOT NULL REFERENCES disasmer_users(user_id) ON DELETE CASCADE,
agent_id TEXT NOT NULL,
record JSONB NOT NULL,
PRIMARY KEY (tenant_id, project_id, agent_id)
);
CREATE TABLE IF NOT EXISTS disasmer_source_provider_configs (
tenant_id TEXT NOT NULL REFERENCES disasmer_tenants(tenant_id) ON DELETE CASCADE,
project_id TEXT NOT NULL REFERENCES disasmer_projects(project_id) ON DELETE CASCADE,
@ -345,12 +385,13 @@ mod tests {
.map(|table| table.name)
.collect::<Vec<_>>();
assert_eq!(names.len(), 8);
assert_eq!(names.len(), 9);
assert!(names.contains(&"disasmer_tenants"));
assert!(names.contains(&"disasmer_users"));
assert!(names.contains(&"disasmer_projects"));
assert!(names.contains(&"disasmer_node_identities"));
assert!(names.contains(&"disasmer_credentials"));
assert!(names.contains(&"disasmer_agent_public_keys"));
assert!(names.contains(&"disasmer_source_provider_configs"));
assert!(names.contains(&"disasmer_service_policy_records"));
assert!(names.contains(&"disasmer_project_permissions"));

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

@ -4,19 +4,21 @@ use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use disasmer_core::{
Actor, ArtifactId, ArtifactRegistry, Capability, CapabilityReportError, CredentialKind,
DataPlaneObject, DataPlaneScope, DefaultScheduler, Digest, DirectBulkTransferPlan,
DownloadLink, DownloadPolicy, EnvironmentRequirements, LimitError, LimitKind,
NativeQuicTransport, NodeCapabilities, NodeDescriptor, NodeEndpoint, NodeId, PanelEvent,
PanelEventKind, PanelState, PanelWidget, PanelWidgetKind, Placement, PlacementRequest,
ProcessId, ProjectId, RateLimit, RendezvousRequest, ResourceLimits, ResourceMeter, Scheduler,
SourcePreparation, SourceProviderKind, StorageLocation, TaskId, TenantId, TransportError,
UserId, VfsPath,
Actor, AgentId, ArtifactId, ArtifactRegistry, Capability, CapabilityReportError,
CredentialKind, DataPlaneObject, DataPlaneScope, DefaultScheduler, Digest,
DirectBulkTransferPlan, DownloadLink, DownloadPolicy, EnvironmentRequirements, LimitError,
LimitKind, NativeQuicTransport, NodeCapabilities, NodeDescriptor, NodeEndpoint, NodeId,
PanelEvent, PanelEventKind, PanelState, PanelWidget, PanelWidgetKind, Placement,
PlacementRequest, ProcessId, ProjectId, RateLimit, RendezvousRequest, ResourceLimits,
ResourceMeter, Scheduler, SourcePreparation, SourceProviderKind, StorageLocation, TaskId,
TenantId, TransportError, UserId, VfsPath,
};
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::{Coordinator, CoordinatorError, InMemoryDurableStore, ProjectRecord};
use crate::{
AgentPublicKeyRecord, Coordinator, CoordinatorError, InMemoryDurableStore, ProjectRecord,
};
const MAX_TASK_LOG_TAIL_BYTES: usize = 256 * 1024;
@ -39,6 +41,31 @@ pub enum CoordinatorRequest {
tenant: String,
actor_user: String,
},
RegisterAgentPublicKey {
tenant: String,
project: String,
user: String,
agent: String,
public_key: String,
},
ListAgentPublicKeys {
tenant: String,
project: String,
user: String,
},
RotateAgentPublicKey {
tenant: String,
project: String,
user: String,
agent: String,
public_key: String,
},
RevokeAgentPublicKey {
tenant: String,
project: String,
user: String,
agent: String,
},
AttachNode {
tenant: String,
project: String,
@ -406,6 +433,14 @@ pub enum CoordinatorResponse {
projects: Vec<ProjectRecord>,
actor: UserId,
},
AgentPublicKey {
record: AgentPublicKeyRecord,
actor: UserId,
},
AgentPublicKeys {
records: Vec<AgentPublicKeyRecord>,
actor: UserId,
},
NodeAttached {
node: NodeId,
tenant: TenantId,
@ -703,6 +738,87 @@ impl CoordinatorService {
actor,
})
}
CoordinatorRequest::RegisterAgentPublicKey {
tenant,
project,
user,
agent,
public_key,
}
| CoordinatorRequest::RotateAgentPublicKey {
tenant,
project,
user,
agent,
public_key,
} => {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let actor = UserId::new(user);
let agent = AgentId::new(agent);
if let Some(existing) = self.coordinator.project(&project) {
if existing.tenant != tenant {
return Err(CoordinatorError::Unauthorized(
"project id is outside the signed-in tenant scope".to_owned(),
)
.into());
}
}
self.coordinator.upsert_tenant(tenant.clone());
self.coordinator.upsert_user(
tenant.clone(),
actor.clone(),
CredentialKind::CliDeviceSession,
);
self.coordinator
.upsert_project(tenant.clone(), project.clone(), "local");
let record = self.coordinator.register_agent_public_key(
tenant,
project,
actor.clone(),
agent,
public_key,
);
self.coordinator.persist(&mut self.store);
Ok(CoordinatorResponse::AgentPublicKey { record, actor })
}
CoordinatorRequest::ListAgentPublicKeys {
tenant,
project,
user,
} => {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let actor = UserId::new(user);
let context = disasmer_core::AuthContext {
tenant,
project,
actor: Actor::User(actor.clone()),
};
Ok(CoordinatorResponse::AgentPublicKeys {
records: self.coordinator.list_agent_public_keys(&context),
actor,
})
}
CoordinatorRequest::RevokeAgentPublicKey {
tenant,
project,
user,
agent,
} => {
let tenant = TenantId::new(tenant);
let project = ProjectId::new(project);
let actor = UserId::new(user);
let agent = AgentId::new(agent);
let context = disasmer_core::AuthContext {
tenant,
project,
actor: Actor::User(actor.clone()),
};
let record = self.coordinator.revoke_agent_public_key(&context, &agent)?;
self.coordinator.persist(&mut self.store);
Ok(CoordinatorResponse::AgentPublicKey { record, actor })
}
CoordinatorRequest::AttachNode {
tenant,
project,
@ -2295,6 +2411,113 @@ mod tests {
assert!(cross_tenant.to_string().contains("tenant scope"));
}
#[test]
fn service_manages_project_scoped_agent_public_keys() {
let mut service = CoordinatorService::new(7);
let CoordinatorResponse::AgentPublicKey { record, actor } = 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()
else {
panic!("expected agent public key registration");
};
assert_eq!(actor, UserId::from("user"));
assert_eq!(record.tenant, TenantId::from("tenant"));
assert_eq!(record.project, ProjectId::from("project"));
assert_eq!(record.user, UserId::from("user"));
assert_eq!(record.agent, AgentId::from("agent-ci"));
assert_eq!(record.version, 1);
assert!(!record.revoked);
assert_eq!(record.scopes, vec!["project:read", "project:run"]);
assert!(!record.human_account_creation_privilege);
assert!(!record.browser_interaction_required_each_run);
let CoordinatorResponse::AgentPublicKey { record, .. } = 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-v2".to_owned(),
})
.unwrap()
else {
panic!("expected agent public key rotation by re-add");
};
assert_eq!(record.version, 2);
assert_eq!(record.public_key, "agent-key-v2");
let CoordinatorResponse::AgentPublicKeys { records, actor } = service
.handle_request(CoordinatorRequest::ListAgentPublicKeys {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
})
.unwrap()
else {
panic!("expected agent public key list");
};
assert_eq!(actor, UserId::from("user"));
assert_eq!(records.len(), 1);
assert_eq!(records[0].agent, AgentId::from("agent-ci"));
let CoordinatorResponse::AgentPublicKeys { records, .. } = service
.handle_request(CoordinatorRequest::ListAgentPublicKeys {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "other-user".to_owned(),
})
.unwrap()
else {
panic!("expected foreign user agent public key list");
};
assert!(records.is_empty());
let CoordinatorResponse::AgentPublicKeys { records, .. } = service
.handle_request(CoordinatorRequest::ListAgentPublicKeys {
tenant: "other-tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
})
.unwrap()
else {
panic!("expected foreign tenant agent public key list");
};
assert!(records.is_empty());
let foreign_user_revoke = service
.handle_request(CoordinatorRequest::RevokeAgentPublicKey {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "other-user".to_owned(),
agent: "agent-ci".to_owned(),
})
.unwrap_err();
assert!(foreign_user_revoke
.to_string()
.contains("signed-in user scope"));
let CoordinatorResponse::AgentPublicKey { record, actor } = service
.handle_request(CoordinatorRequest::RevokeAgentPublicKey {
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
agent: "agent-ci".to_owned(),
})
.unwrap()
else {
panic!("expected agent public key revocation");
};
assert_eq!(actor, UserId::from("user"));
assert!(record.revoked);
}
#[test]
fn service_attaches_node_starts_process_and_records_scoped_task_event() {
let mut service = CoordinatorService::new(7);