feat: issue native checkout credentials #48
@@ -1,27 +1,28 @@
|
||||
pub mod action_delivery;
|
||||
pub mod action_oci;
|
||||
pub mod action_repository;
|
||||
pub mod action_store;
|
||||
pub mod actions;
|
||||
pub mod actions_read;
|
||||
pub mod actions_read_identity;
|
||||
pub mod admin;
|
||||
pub mod check_events;
|
||||
pub mod checks;
|
||||
pub mod events;
|
||||
pub mod intake;
|
||||
pub mod maintenance;
|
||||
mod native_events;
|
||||
pub mod projection;
|
||||
#[path = "forge.rs"]
|
||||
pub mod repository;
|
||||
pub mod repository_grpc;
|
||||
pub mod repository_sources;
|
||||
pub mod reusable;
|
||||
mod secret_reference_syntax;
|
||||
pub mod secret_references;
|
||||
pub mod secrets;
|
||||
pub mod sources;
|
||||
pub mod token;
|
||||
pub mod trigger;
|
||||
pub mod webhook;
|
||||
pub mod action_delivery;
|
||||
pub mod action_oci;
|
||||
pub mod action_repository;
|
||||
pub mod action_store;
|
||||
pub mod actions;
|
||||
pub mod actions_read;
|
||||
pub mod actions_read_identity;
|
||||
pub mod admin;
|
||||
pub mod check_events;
|
||||
pub mod checks;
|
||||
pub mod events;
|
||||
pub mod intake;
|
||||
pub mod maintenance;
|
||||
mod native_events;
|
||||
pub mod projection;
|
||||
#[path = "forge.rs"]
|
||||
pub mod repository;
|
||||
pub mod repository_credentials;
|
||||
pub mod repository_grpc;
|
||||
pub mod repository_sources;
|
||||
pub mod reusable;
|
||||
mod secret_reference_syntax;
|
||||
pub mod secret_references;
|
||||
pub mod secrets;
|
||||
pub mod sources;
|
||||
pub mod token;
|
||||
pub mod trigger;
|
||||
pub mod webhook;
|
||||
@@ -1,386 +1,392 @@
|
||||
use std::error::Error;
|
||||
use std::net::SocketAddr;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use clap::{Parser, Subcommand, ValueEnum};
|
||||
use syncode_control::action_delivery::router as action_delivery_router;
|
||||
use syncode_control::action_oci::PinnedOciResolver;
|
||||
use syncode_control::action_repository::NativeActionRepository;
|
||||
use syncode_control::action_store::FileActionStore;
|
||||
use syncode_control::actions::ActionResolver;
|
||||
use syncode_control::actions_read::ActionsRead;
|
||||
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
|
||||
use syncode_control::admin::{Admin, router as admin_router};
|
||||
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
|
||||
use syncode_control::checks::Checks;
|
||||
use syncode_control::maintenance;
|
||||
use syncode_control::projection;
|
||||
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
|
||||
use syncode_control::repository_grpc::NativeRepositoryContents;
|
||||
use syncode_control::repository_sources::RepositorySources;
|
||||
use syncode_control::secrets::RuntimeSecrets;
|
||||
use syncode_control::token::EnrolmentScope;
|
||||
use syncode_control::webhook::{Intake, router};
|
||||
use syncode_control_node::{
|
||||
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
|
||||
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_URL")]
|
||||
projection_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
|
||||
projection_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
|
||||
projection_min_run_number: u64,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
projection_min_run_number: arguments.projection_min_run_number,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
|
||||
let repository_sources = RepositorySources::new(
|
||||
native_repository,
|
||||
RepositoryContents::new(
|
||||
arguments.repository_source,
|
||||
arguments.repository_source_token,
|
||||
),
|
||||
);
|
||||
let intake = Arc::new(Intake::new(
|
||||
repository_sources,
|
||||
runs.clone(),
|
||||
action_resolver,
|
||||
arguments.event_feed_secret,
|
||||
));
|
||||
let admin_token = arguments
|
||||
.admin_token
|
||||
.filter(|token| !token.is_empty())
|
||||
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
|
||||
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
|
||||
let http = router(intake)
|
||||
.merge(admin_router(Arc::new(
|
||||
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
|
||||
)))
|
||||
.merge(action_delivery_router(
|
||||
action_store,
|
||||
artifact_authority.clone(),
|
||||
runs.clone(),
|
||||
));
|
||||
let events = TcpListener::bind(arguments.listen_events).await?;
|
||||
let check_events = CheckEventPublisher::new(
|
||||
store.clone(),
|
||||
CheckEventPublisherConfig {
|
||||
endpoint: arguments.collaboration_event_url.as_str(),
|
||||
secret: arguments.collaboration_event_secret,
|
||||
source_node_id: arguments.source_node_id,
|
||||
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
|
||||
batch: arguments.check_event_batch,
|
||||
},
|
||||
)?;
|
||||
|
||||
let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
|
||||
let projection = ProjectionClient::new(
|
||||
arguments.projection_url,
|
||||
arguments.projection_token,
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)?;
|
||||
let secret_service = RuntimeSecrets::new(
|
||||
runs.clone(),
|
||||
nodes.clone(),
|
||||
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
|
||||
authority.clone(),
|
||||
);
|
||||
let actions_read = ActionsRead::new(
|
||||
runs.clone(),
|
||||
IdentityActionsAuthorization::connect(
|
||||
arguments.identity_grpc,
|
||||
arguments.identity_shared_secret,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
// Both ends of the control plane run for as long as the other does: without
|
||||
// events there is nothing to assign, and without sessions there is nobody to
|
||||
// assign it to. Whichever stops first takes the process down with it.
|
||||
let node_service = match arguments.mode {
|
||||
Mode::Shadow => NodeSessionServer::shadow(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
projection.clone(),
|
||||
),
|
||||
Mode::Active => NodeSessionServer::new(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
projection.clone(),
|
||||
),
|
||||
};
|
||||
tokio::select! {
|
||||
served = axum::serve(events, http).into_future() => served?,
|
||||
served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?,
|
||||
published = check_events.run() => published?,
|
||||
served = Server::builder()
|
||||
.add_service(GeneratedServer::new(node_service))
|
||||
.add_service(GeneratedSecretsServer::new(secret_service))
|
||||
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
|
||||
.add_service(GeneratedActionsReadServer::new(actions_read))
|
||||
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn shutdown() {
|
||||
if let Err(error) = tokio::signal::ctrl_c().await {
|
||||
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
}
|
||||
use std::error::Error;
|
||||
use std::net::SocketAddr;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use clap::{Parser, Subcommand, ValueEnum};
|
||||
use syncode_control::action_delivery::router as action_delivery_router;
|
||||
use syncode_control::action_oci::PinnedOciResolver;
|
||||
use syncode_control::action_repository::NativeActionRepository;
|
||||
use syncode_control::action_store::FileActionStore;
|
||||
use syncode_control::actions::ActionResolver;
|
||||
use syncode_control::actions_read::ActionsRead;
|
||||
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
|
||||
use syncode_control::admin::{Admin, router as admin_router};
|
||||
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
|
||||
use syncode_control::checks::Checks;
|
||||
use syncode_control::maintenance;
|
||||
use syncode_control::projection;
|
||||
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
|
||||
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
|
||||
use syncode_control::repository_grpc::NativeRepositoryContents;
|
||||
use syncode_control::repository_sources::RepositorySources;
|
||||
use syncode_control::secrets::RuntimeSecrets;
|
||||
use syncode_control::token::EnrolmentScope;
|
||||
use syncode_control::webhook::{Intake, router};
|
||||
use syncode_control_node::{
|
||||
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
|
||||
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_URL")]
|
||||
projection_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
|
||||
projection_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
|
||||
projection_min_run_number: u64,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
projection_min_run_number: arguments.projection_min_run_number,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
|
||||
let repository_sources = RepositorySources::new(
|
||||
native_repository,
|
||||
RepositoryContents::new(
|
||||
arguments.repository_source,
|
||||
arguments.repository_source_token,
|
||||
),
|
||||
);
|
||||
let intake = Arc::new(Intake::new(
|
||||
repository_sources,
|
||||
runs.clone(),
|
||||
action_resolver,
|
||||
arguments.event_feed_secret,
|
||||
));
|
||||
let admin_token = arguments
|
||||
.admin_token
|
||||
.filter(|token| !token.is_empty())
|
||||
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
|
||||
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
|
||||
let http = router(intake)
|
||||
.merge(admin_router(Arc::new(
|
||||
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
|
||||
)))
|
||||
.merge(action_delivery_router(
|
||||
action_store,
|
||||
artifact_authority.clone(),
|
||||
runs.clone(),
|
||||
));
|
||||
let events = TcpListener::bind(arguments.listen_events).await?;
|
||||
let check_events = CheckEventPublisher::new(
|
||||
store.clone(),
|
||||
CheckEventPublisherConfig {
|
||||
endpoint: arguments.collaboration_event_url.as_str(),
|
||||
secret: arguments.collaboration_event_secret,
|
||||
source_node_id: arguments.source_node_id,
|
||||
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
|
||||
batch: arguments.check_event_batch,
|
||||
},
|
||||
)?;
|
||||
|
||||
let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
|
||||
let projection = ProjectionClient::new(
|
||||
arguments.projection_url,
|
||||
arguments.projection_token,
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)?;
|
||||
let secret_service = RuntimeSecrets::new(
|
||||
runs.clone(),
|
||||
nodes.clone(),
|
||||
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
|
||||
IdentityRepositoryCredentials::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
authority.clone(),
|
||||
);
|
||||
let actions_read = ActionsRead::new(
|
||||
runs.clone(),
|
||||
IdentityActionsAuthorization::connect(
|
||||
arguments.identity_grpc,
|
||||
arguments.identity_shared_secret,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
|
||||
// Both ends of the control plane run for as long as the other does: without
|
||||
// events there is nothing to assign, and without sessions there is nobody to
|
||||
// assign it to. Whichever stops first takes the process down with it.
|
||||
let node_service = match arguments.mode {
|
||||
Mode::Shadow => NodeSessionServer::shadow(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
projection.clone(),
|
||||
),
|
||||
Mode::Active => NodeSessionServer::new(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
projection.clone(),
|
||||
),
|
||||
};
|
||||
tokio::select! {
|
||||
served = axum::serve(events, http).into_future() => served?,
|
||||
served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?,
|
||||
published = check_events.run() => published?,
|
||||
served = Server::builder()
|
||||
.add_service(GeneratedServer::new(node_service))
|
||||
.add_service(GeneratedSecretsServer::new(secret_service))
|
||||
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
|
||||
.add_service(GeneratedActionsReadServer::new(actions_read))
|
||||
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn shutdown() {
|
||||
if let Err(error) = tokio::signal::ctrl_c().await {
|
||||
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
}
|
||||
+56
-16
@@ -1,142 +1,182 @@
|
||||
use std::error::Error;
|
||||
use std::future::Future;
|
||||
use std::time::SystemTime;
|
||||
|
||||
use syncode_control_node::wire::{SecretRequest, SecretResponse};
|
||||
use syncode_control_node::{CapabilityAuthority, RuntimeSecretsService};
|
||||
use syncode_control_nodes::{Lifecycle, NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs, SecretReadOutcome};
|
||||
use syncode_workflow::SecretName;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
pub trait SecretSource: Send + Sync + 'static {
|
||||
type Error: Error + Send + Sync + 'static;
|
||||
|
||||
fn resolve(
|
||||
&self,
|
||||
repository: &str,
|
||||
name: &str,
|
||||
) -> impl Future<Output = Result<Option<String>, Self::Error>> + Send;
|
||||
}
|
||||
|
||||
pub struct RuntimeSecrets<L, S, F> {
|
||||
runs: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
source: F,
|
||||
authority: CapabilityAuthority,
|
||||
}
|
||||
|
||||
impl<L, S, F> RuntimeSecrets<L, S, F> {
|
||||
pub const fn new(
|
||||
runs: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
source: F,
|
||||
authority: CapabilityAuthority,
|
||||
) -> Self {
|
||||
Self {
|
||||
runs,
|
||||
nodes,
|
||||
source,
|
||||
authority,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl<L, S, F> RuntimeSecretsService for RuntimeSecrets<L, S, F>
|
||||
where
|
||||
L: RunLog + 'static,
|
||||
S: NodeStore + 'static,
|
||||
F: SecretSource,
|
||||
{
|
||||
async fn resolve(
|
||||
&self,
|
||||
request: Request<SecretRequest>,
|
||||
) -> Result<Response<SecretResponse>, Status> {
|
||||
let request = request.into_inner();
|
||||
let claims = self
|
||||
.authority
|
||||
.verify(&request.capability, SystemTime::now())
|
||||
.map_err(|error| Status::unauthenticated(error.to_string()))?;
|
||||
let name = request.name.to_ascii_uppercase();
|
||||
if name.parse::<SecretName>().is_err() {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(Status::invalid_argument("invalid secret name"));
|
||||
}
|
||||
let origin = match self.authorize(claims, &name).await {
|
||||
Ok(origin) => origin,
|
||||
Err(error) => {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
let value = match self.source.resolve(origin.repository(), &name).await {
|
||||
Ok(Some(value)) => value,
|
||||
Ok(None) => {
|
||||
self.audit(claims, name, SecretReadOutcome::Missing).await?;
|
||||
return Err(Status::not_found("secret not found"));
|
||||
}
|
||||
Err(error) => {
|
||||
self.audit(claims, name, SecretReadOutcome::SourceFailure)
|
||||
.await?;
|
||||
return Err(Status::unavailable(format!(
|
||||
"secret source unavailable: {error}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
if let Err(error) = self.authorize(claims, &name).await {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(error);
|
||||
}
|
||||
self.audit(claims, name, SecretReadOutcome::Granted).await?;
|
||||
Ok(Response::new(SecretResponse { value }))
|
||||
}
|
||||
}
|
||||
|
||||
impl<L: RunLog, S: NodeStore, F> RuntimeSecrets<L, S, F> {
|
||||
async fn authorize(
|
||||
&self,
|
||||
claims: syncode_control_node::CapabilityClaims,
|
||||
name: &str,
|
||||
) -> Result<syncode_control_runs::Origin, Status> {
|
||||
let origin = self
|
||||
.runs
|
||||
.authorize_secret(
|
||||
claims.run(),
|
||||
claims.job(),
|
||||
claims.node(),
|
||||
claims.fence(),
|
||||
name,
|
||||
)
|
||||
.await
|
||||
.map_err(|error| Status::permission_denied(error.to_string()))?;
|
||||
if !origin.secrets_allowed() {
|
||||
return Err(Status::permission_denied(
|
||||
"secrets are denied for an untrusted pull request",
|
||||
));
|
||||
}
|
||||
let lifecycle = self
|
||||
.nodes
|
||||
.lifecycle(claims.node())
|
||||
.await
|
||||
.map_err(|error| Status::permission_denied(error.to_string()))?;
|
||||
if !matches!(lifecycle, Lifecycle::Active | Lifecycle::Draining) {
|
||||
return Err(Status::permission_denied(format!(
|
||||
"node is not allowed to read secrets while {lifecycle:?}"
|
||||
)));
|
||||
}
|
||||
Ok(origin)
|
||||
}
|
||||
|
||||
async fn audit(
|
||||
&self,
|
||||
claims: syncode_control_node::CapabilityClaims,
|
||||
name: String,
|
||||
outcome: SecretReadOutcome,
|
||||
) -> Result<(), Status> {
|
||||
self.runs
|
||||
.audit_secret(claims.run(), claims.job(), claims.node(), name, outcome)
|
||||
.await
|
||||
.map_err(|error| Status::internal(format!("cannot audit secret read: {error}")))
|
||||
}
|
||||
}
|
||||
use std::error::Error;
|
||||
use std::future::Future;
|
||||
use std::time::SystemTime;
|
||||
|
||||
use syncode_control_node::wire::{SecretRequest, SecretResponse};
|
||||
use syncode_control_node::{CapabilityAuthority, REPOSITORY_TOKEN_SECRET, RuntimeSecretsService};
|
||||
use syncode_control_nodes::{Lifecycle, NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs, SecretReadOutcome};
|
||||
use syncode_workflow::SecretName;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
pub trait SecretSource: Send + Sync + 'static {
|
||||
type Error: Error + Send + Sync + 'static;
|
||||
|
||||
fn resolve(
|
||||
&self,
|
||||
repository: &str,
|
||||
name: &str,
|
||||
) -> impl Future<Output = Result<Option<String>, Self::Error>> + Send;
|
||||
}
|
||||
|
||||
pub trait RepositoryCredentialSource: Send + Sync + 'static {
|
||||
type Error: Error + Send + Sync + 'static;
|
||||
|
||||
fn issue(
|
||||
&self,
|
||||
user: &str,
|
||||
repository: &str,
|
||||
) -> impl Future<Output = Result<String, Self::Error>> + Send;
|
||||
}
|
||||
|
||||
pub struct RuntimeSecrets<L, S, F, R> {
|
||||
runs: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
source: F,
|
||||
repository_credentials: R,
|
||||
authority: CapabilityAuthority,
|
||||
}
|
||||
|
||||
impl<L, S, F, R> RuntimeSecrets<L, S, F, R> {
|
||||
pub const fn new(
|
||||
runs: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
source: F,
|
||||
repository_credentials: R,
|
||||
authority: CapabilityAuthority,
|
||||
) -> Self {
|
||||
Self {
|
||||
runs,
|
||||
nodes,
|
||||
source,
|
||||
repository_credentials,
|
||||
authority,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl<L, S, F, R> RuntimeSecretsService for RuntimeSecrets<L, S, F, R>
|
||||
where
|
||||
L: RunLog + 'static,
|
||||
S: NodeStore + 'static,
|
||||
F: SecretSource,
|
||||
R: RepositoryCredentialSource,
|
||||
{
|
||||
async fn resolve(
|
||||
&self,
|
||||
request: Request<SecretRequest>,
|
||||
) -> Result<Response<SecretResponse>, Status> {
|
||||
let request = request.into_inner();
|
||||
let claims = self
|
||||
.authority
|
||||
.verify(&request.capability, SystemTime::now())
|
||||
.map_err(|error| Status::unauthenticated(error.to_string()))?;
|
||||
let name = request.name.to_ascii_uppercase();
|
||||
if name.parse::<SecretName>().is_err() {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(Status::invalid_argument("invalid secret name"));
|
||||
}
|
||||
let origin = match self.authorize(claims, &name).await {
|
||||
Ok(origin) => origin,
|
||||
Err(error) => {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
let value = if name == REPOSITORY_TOKEN_SECRET {
|
||||
let user = match origin.principal() {
|
||||
Some(user) => user,
|
||||
None => {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(Status::permission_denied(
|
||||
"workflow repository credentials require an originating user",
|
||||
));
|
||||
}
|
||||
};
|
||||
match self
|
||||
.repository_credentials
|
||||
.issue(user, origin.repository())
|
||||
.await
|
||||
{
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
self.audit(claims, name, SecretReadOutcome::SourceFailure)
|
||||
.await?;
|
||||
return Err(Status::unavailable(format!(
|
||||
"repository credential source unavailable: {error}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
match self.source.resolve(origin.repository(), &name).await {
|
||||
Ok(Some(value)) => value,
|
||||
Ok(None) => {
|
||||
self.audit(claims, name, SecretReadOutcome::Missing).await?;
|
||||
return Err(Status::not_found("secret not found"));
|
||||
}
|
||||
Err(error) => {
|
||||
self.audit(claims, name, SecretReadOutcome::SourceFailure)
|
||||
.await?;
|
||||
return Err(Status::unavailable(format!(
|
||||
"secret source unavailable: {error}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
};
|
||||
if let Err(error) = self.authorize(claims, &name).await {
|
||||
self.audit(claims, name, SecretReadOutcome::Denied).await?;
|
||||
return Err(error);
|
||||
}
|
||||
self.audit(claims, name, SecretReadOutcome::Granted).await?;
|
||||
Ok(Response::new(SecretResponse { value }))
|
||||
}
|
||||
}
|
||||
|
||||
impl<L: RunLog, S: NodeStore, F, R> RuntimeSecrets<L, S, F, R> {
|
||||
async fn authorize(
|
||||
&self,
|
||||
claims: syncode_control_node::CapabilityClaims,
|
||||
name: &str,
|
||||
) -> Result<syncode_control_runs::Origin, Status> {
|
||||
let origin = self
|
||||
.runs
|
||||
.authorize_secret(
|
||||
claims.run(),
|
||||
claims.job(),
|
||||
claims.node(),
|
||||
claims.fence(),
|
||||
name,
|
||||
)
|
||||
.await
|
||||
.map_err(|error| Status::permission_denied(error.to_string()))?;
|
||||
if !origin.secrets_allowed() {
|
||||
return Err(Status::permission_denied(
|
||||
"secrets are denied for an untrusted pull request",
|
||||
));
|
||||
}
|
||||
let lifecycle = self
|
||||
.nodes
|
||||
.lifecycle(claims.node())
|
||||
.await
|
||||
.map_err(|error| Status::permission_denied(error.to_string()))?;
|
||||
if !matches!(lifecycle, Lifecycle::Active | Lifecycle::Draining) {
|
||||
return Err(Status::permission_denied(format!(
|
||||
"node is not allowed to read secrets while {lifecycle:?}"
|
||||
)));
|
||||
}
|
||||
Ok(origin)
|
||||
}
|
||||
|
||||
async fn audit(
|
||||
&self,
|
||||
claims: syncode_control_node::CapabilityClaims,
|
||||
name: String,
|
||||
outcome: SecretReadOutcome,
|
||||
) -> Result<(), Status> {
|
||||
self.runs
|
||||
.audit_secret(claims.run(), claims.job(), claims.node(), name, outcome)
|
||||
.await
|
||||
.map_err(|error| Status::internal(format!("cannot audit secret read: {error}")))
|
||||
}
|
||||
}
|
||||
@@ -1,129 +1,173 @@
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
#[path = "support/repository.rs"]
|
||||
#[allow(dead_code)]
|
||||
mod repository;
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control::action_repository::NativeActionRepository;
|
||||
use syncode_control::actions::ActionRepositoryPort;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
#[derive(Default)]
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
_request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
Err(Status::unimplemented("get_repository_coordinates"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
let request = request.into_inner();
|
||||
if request.owner != "actions" || request.name != "checkout" {
|
||||
return Err(Status::not_found("repository"));
|
||||
}
|
||||
Ok(Response::new(ResolveRepositoryResponse {
|
||||
repository_id: repository::REPOSITORY.to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
async fn source() -> TestResult<NativeActionRepository> {
|
||||
let repository_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let repository_endpoint = format!("http://{}", repository_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(repository::service())
|
||||
.serve_with_incoming(TcpListenerStream::new(repository_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
Ok(NativeActionRepository::connect(
|
||||
repository_endpoint,
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)
|
||||
.await?)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolves_coordinates_and_fetches_an_immutable_native_archive() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
"https://dev.syncode.sh/actions/checkout".parse()?,
|
||||
"v4".to_owned(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
let paths = tar::Archive::new(snapshot.archive.as_slice())
|
||||
.entries()?
|
||||
.map(|entry| entry.and_then(|entry| entry.path().map(|path| path.into_owned())))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
assert_eq!(paths, [std::path::PathBuf::from("source/action.yml")]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn accepts_a_native_repository_id_without_coordinate_resolution() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
format!("https://dev.syncode.sh/{}", repository::REPOSITORY).parse()?,
|
||||
repository::COMMIT.to_owned(),
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
#[path = "support/repository.rs"]
|
||||
#[allow(dead_code)]
|
||||
mod repository;
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control::action_repository::NativeActionRepository;
|
||||
use syncode_control::actions::ActionRepositoryPort;
|
||||
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
|
||||
use syncode_control::secrets::RepositoryCredentialSource;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
#[derive(Default)]
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
let request = request.into_inner();
|
||||
if request.user_id != "user-id" || request.repository_id != repository::REPOSITORY {
|
||||
return Err(Status::invalid_argument("unexpected token scope"));
|
||||
}
|
||||
Ok(Response::new(IssueWorkflowRepositoryTokenResponse {
|
||||
token: "workflow-repository-token".to_owned(),
|
||||
expires_at_unix: 1,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
_request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
Err(Status::unimplemented("get_repository_coordinates"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
let request = request.into_inner();
|
||||
if request.owner != "actions" || request.name != "checkout" {
|
||||
return Err(Status::not_found("repository"));
|
||||
}
|
||||
Ok(Response::new(ResolveRepositoryResponse {
|
||||
repository_id: repository::REPOSITORY.to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn issues_a_scoped_repository_credential_over_identity_grpc() -> TestResult {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let endpoint = format!("http://{}", listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(listener))
|
||||
.await;
|
||||
});
|
||||
let credentials =
|
||||
IdentityRepositoryCredentials::connect(endpoint, "shared-secret".to_owned()).await?;
|
||||
|
||||
let token = credentials.issue("user-id", repository::REPOSITORY).await?;
|
||||
|
||||
assert_eq!(token, "workflow-repository-token");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn source() -> TestResult<NativeActionRepository> {
|
||||
let repository_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let repository_endpoint = format!("http://{}", repository_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(repository::service())
|
||||
.serve_with_incoming(TcpListenerStream::new(repository_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
Ok(NativeActionRepository::connect(
|
||||
repository_endpoint,
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)
|
||||
.await?)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolves_coordinates_and_fetches_an_immutable_native_archive() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
"https://dev.syncode.sh/actions/checkout".parse()?,
|
||||
"v4".to_owned(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
let paths = tar::Archive::new(snapshot.archive.as_slice())
|
||||
.entries()?
|
||||
.map(|entry| entry.and_then(|entry| entry.path().map(|path| path.into_owned())))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
assert_eq!(paths, [std::path::PathBuf::from("source/action.yml")]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn accepts_a_native_repository_id_without_coordinate_resolution() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
format!("https://dev.syncode.sh/{}", repository::REPOSITORY).parse()?,
|
||||
repository::COMMIT.to_owned(),
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,130 +1,138 @@
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use axum::Json;
|
||||
use axum::http::StatusCode;
|
||||
use axum::routing::post;
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use syncode_control_runs::{Forgotten, JobId, Origin, Runs};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let projection_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?;
|
||||
let (sent, mut received) = mpsc::unbounded_channel();
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/projection",
|
||||
post(move |Json(body): Json<serde_json::Value>| {
|
||||
let sent = sent.clone();
|
||||
async move {
|
||||
let _ = sent.send(body);
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(projection_listener, app).await;
|
||||
});
|
||||
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = runs
|
||||
.queue(
|
||||
JobId::fresh(),
|
||||
Origin::new(
|
||||
"76128383-1df5-4979-9b13-c048a5287e9a".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned())),
|
||||
b"plan".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
let projection = runs.projection(run).await?.expect("projection");
|
||||
let client = ProjectionClient::new(
|
||||
projection_base,
|
||||
String::new(),
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)?;
|
||||
assert_eq!(
|
||||
client
|
||||
.repository_coordinates("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
client.send(&projection).await?;
|
||||
|
||||
let body = received.recv().await.expect("projection body");
|
||||
assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo");
|
||||
assert_eq!(body["origin"]["native"], true);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use axum::Json;
|
||||
use axum::http::StatusCode;
|
||||
use axum::routing::post;
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use syncode_control_runs::{Forgotten, JobId, Origin, Runs};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
_request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
Err(Status::unimplemented("issue_workflow_repository_token"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let projection_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?;
|
||||
let (sent, mut received) = mpsc::unbounded_channel();
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/projection",
|
||||
post(move |Json(body): Json<serde_json::Value>| {
|
||||
let sent = sent.clone();
|
||||
async move {
|
||||
let _ = sent.send(body);
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(projection_listener, app).await;
|
||||
});
|
||||
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = runs
|
||||
.queue(
|
||||
JobId::fresh(),
|
||||
Origin::new(
|
||||
"76128383-1df5-4979-9b13-c048a5287e9a".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned())),
|
||||
b"plan".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
let projection = runs.projection(run).await?.expect("projection");
|
||||
let client = ProjectionClient::new(
|
||||
projection_base,
|
||||
String::new(),
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)?;
|
||||
assert_eq!(
|
||||
client
|
||||
.repository_coordinates("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
client.send(&projection).await?;
|
||||
|
||||
let body = received.recv().await.expect("projection body");
|
||||
assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo");
|
||||
assert_eq!(body["origin"]["native"], true);
|
||||
Ok(())
|
||||
}
|
||||
+15
-2
File diff suppressed because it is too large
Load Diff
@@ -1,70 +1,81 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package syncode.identity.v1;
|
||||
|
||||
service Identity {
|
||||
rpc ValidateSession(ValidateSessionRequest) returns (ValidateSessionResponse);
|
||||
rpc CheckCapability(CheckCapabilityRequest) returns (CheckCapabilityResponse);
|
||||
rpc ResolveRepository(ResolveRepositoryRequest) returns (ResolveRepositoryResponse);
|
||||
rpc GetRepositoryCoordinates(GetRepositoryCoordinatesRequest) returns (GetRepositoryCoordinatesResponse);
|
||||
}
|
||||
|
||||
message ResolveRepositoryRequest {
|
||||
string owner = 1;
|
||||
string name = 2;
|
||||
}
|
||||
|
||||
message ResolveRepositoryResponse {
|
||||
string repository_id = 1;
|
||||
}
|
||||
|
||||
message GetRepositoryCoordinatesRequest {
|
||||
string repository_id = 1;
|
||||
}
|
||||
|
||||
message GetRepositoryCoordinatesResponse {
|
||||
string owner = 1;
|
||||
string name = 2;
|
||||
}
|
||||
|
||||
enum PrincipalKind {
|
||||
PRINCIPAL_KIND_UNSPECIFIED = 0;
|
||||
PRINCIPAL_KIND_USER = 1;
|
||||
PRINCIPAL_KIND_PLATFORM_AGENT = 2;
|
||||
PRINCIPAL_KIND_LOCAL_AGENT = 3;
|
||||
PRINCIPAL_KIND_ACCESS_TOKEN = 4;
|
||||
}
|
||||
|
||||
enum ResourceKind {
|
||||
RESOURCE_KIND_UNSPECIFIED = 0;
|
||||
RESOURCE_KIND_ORGANIZATION = 1;
|
||||
RESOURCE_KIND_REPOSITORY = 2;
|
||||
RESOURCE_KIND_INSTANCE = 3;
|
||||
}
|
||||
|
||||
message ValidateSessionRequest {
|
||||
string session_token = 1;
|
||||
}
|
||||
|
||||
message ValidateSessionResponse {
|
||||
string principal_id = 1;
|
||||
PrincipalKind principal_kind = 2;
|
||||
int64 expires_at_unix = 3;
|
||||
string owner_user_id = 4;
|
||||
repeated string capabilities = 5;
|
||||
string audience = 6;
|
||||
ResourceKind resource_kind = 7;
|
||||
string resource_id = 8;
|
||||
}
|
||||
|
||||
message CheckCapabilityRequest {
|
||||
string principal_id = 1;
|
||||
PrincipalKind principal_kind = 2;
|
||||
ResourceKind resource_kind = 3;
|
||||
string resource_id = 4;
|
||||
string capability = 5;
|
||||
}
|
||||
|
||||
message CheckCapabilityResponse {
|
||||
bool allowed = 1;
|
||||
}
|
||||
syntax = "proto3";
|
||||
|
||||
package syncode.identity.v1;
|
||||
|
||||
service Identity {
|
||||
rpc ValidateSession(ValidateSessionRequest) returns (ValidateSessionResponse);
|
||||
rpc CheckCapability(CheckCapabilityRequest) returns (CheckCapabilityResponse);
|
||||
rpc ResolveRepository(ResolveRepositoryRequest) returns (ResolveRepositoryResponse);
|
||||
rpc GetRepositoryCoordinates(GetRepositoryCoordinatesRequest) returns (GetRepositoryCoordinatesResponse);
|
||||
rpc IssueWorkflowRepositoryToken(IssueWorkflowRepositoryTokenRequest) returns (IssueWorkflowRepositoryTokenResponse);
|
||||
}
|
||||
|
||||
message ResolveRepositoryRequest {
|
||||
string owner = 1;
|
||||
string name = 2;
|
||||
}
|
||||
|
||||
message ResolveRepositoryResponse {
|
||||
string repository_id = 1;
|
||||
}
|
||||
|
||||
message GetRepositoryCoordinatesRequest {
|
||||
string repository_id = 1;
|
||||
}
|
||||
|
||||
message GetRepositoryCoordinatesResponse {
|
||||
string owner = 1;
|
||||
string name = 2;
|
||||
}
|
||||
|
||||
message IssueWorkflowRepositoryTokenRequest {
|
||||
string user_id = 1;
|
||||
string repository_id = 2;
|
||||
}
|
||||
|
||||
message IssueWorkflowRepositoryTokenResponse {
|
||||
string token = 1;
|
||||
int64 expires_at_unix = 2;
|
||||
}
|
||||
|
||||
enum PrincipalKind {
|
||||
PRINCIPAL_KIND_UNSPECIFIED = 0;
|
||||
PRINCIPAL_KIND_USER = 1;
|
||||
PRINCIPAL_KIND_PLATFORM_AGENT = 2;
|
||||
PRINCIPAL_KIND_LOCAL_AGENT = 3;
|
||||
PRINCIPAL_KIND_ACCESS_TOKEN = 4;
|
||||
}
|
||||
|
||||
enum ResourceKind {
|
||||
RESOURCE_KIND_UNSPECIFIED = 0;
|
||||
RESOURCE_KIND_ORGANIZATION = 1;
|
||||
RESOURCE_KIND_REPOSITORY = 2;
|
||||
RESOURCE_KIND_INSTANCE = 3;
|
||||
}
|
||||
|
||||
message ValidateSessionRequest {
|
||||
string session_token = 1;
|
||||
}
|
||||
|
||||
message ValidateSessionResponse {
|
||||
string principal_id = 1;
|
||||
PrincipalKind principal_kind = 2;
|
||||
int64 expires_at_unix = 3;
|
||||
string owner_user_id = 4;
|
||||
repeated string capabilities = 5;
|
||||
string audience = 6;
|
||||
ResourceKind resource_kind = 7;
|
||||
string resource_id = 8;
|
||||
}
|
||||
|
||||
message CheckCapabilityRequest {
|
||||
string principal_id = 1;
|
||||
PrincipalKind principal_kind = 2;
|
||||
ResourceKind resource_kind = 3;
|
||||
string resource_id = 4;
|
||||
string capability = 5;
|
||||
}
|
||||
|
||||
message CheckCapabilityResponse {
|
||||
bool allowed = 1;
|
||||
}
|
||||
@@ -1,41 +1,43 @@
|
||||
mod artifact;
|
||||
mod capability;
|
||||
mod declaration;
|
||||
mod error;
|
||||
mod identity;
|
||||
mod outbound;
|
||||
mod projection;
|
||||
mod reports;
|
||||
mod session;
|
||||
|
||||
pub mod wire {
|
||||
//! Generated from `proto/node.proto`.
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs"));
|
||||
}
|
||||
|
||||
pub mod identity_wire {
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.identity.v1.rs"));
|
||||
}
|
||||
|
||||
pub mod actions_wire {
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.control.v1.rs"));
|
||||
}
|
||||
|
||||
pub use actions_wire::actions_read_server::ActionsRead as ActionsReadService;
|
||||
pub use actions_wire::actions_read_server::ActionsReadServer as GeneratedActionsReadServer;
|
||||
pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError};
|
||||
pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError};
|
||||
pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError};
|
||||
pub use session::NodeSessionServer;
|
||||
pub use wire::checks_server::Checks as ChecksService;
|
||||
pub use wire::checks_server::ChecksServer as GeneratedChecksServer;
|
||||
pub use wire::node_session_client::NodeSessionClient;
|
||||
pub use wire::node_session_server::NodeSessionServer as GeneratedServer;
|
||||
pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService;
|
||||
pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer;
|
||||
mod artifact;
|
||||
mod capability;
|
||||
mod declaration;
|
||||
mod error;
|
||||
mod identity;
|
||||
mod outbound;
|
||||
mod projection;
|
||||
mod reports;
|
||||
mod session;
|
||||
|
||||
pub mod wire {
|
||||
//! Generated from `proto/node.proto`.
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs"));
|
||||
}
|
||||
|
||||
pub mod identity_wire {
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.identity.v1.rs"));
|
||||
}
|
||||
|
||||
pub mod actions_wire {
|
||||
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]
|
||||
|
||||
include!(concat!(env!("OUT_DIR"), "/syncode.control.v1.rs"));
|
||||
}
|
||||
|
||||
pub use actions_wire::actions_read_server::ActionsRead as ActionsReadService;
|
||||
pub use actions_wire::actions_read_server::ActionsReadServer as GeneratedActionsReadServer;
|
||||
pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError};
|
||||
pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError};
|
||||
pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError};
|
||||
pub use session::NodeSessionServer;
|
||||
pub use wire::checks_server::Checks as ChecksService;
|
||||
pub use wire::checks_server::ChecksServer as GeneratedChecksServer;
|
||||
pub use wire::node_session_client::NodeSessionClient;
|
||||
pub use wire::node_session_server::NodeSessionServer as GeneratedServer;
|
||||
pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService;
|
||||
pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer;
|
||||
|
||||
pub const REPOSITORY_TOKEN_SECRET: &str = "SYNCODE_REPOSITORY_TOKEN";
|
||||
@@ -1,70 +1,103 @@
|
||||
use std::time::SystemTime;
|
||||
|
||||
use syncode_control_runs::NodeId;
|
||||
|
||||
use crate::error::SessionError;
|
||||
use crate::wire::{Dependency, JobAssignment, control_message};
|
||||
use crate::{declaration, wire};
|
||||
|
||||
pub(super) fn assignment_body(
|
||||
assignment: &syncode_control_runs::Assignment,
|
||||
repository: String,
|
||||
node: NodeId,
|
||||
authority: &crate::CapabilityAuthority,
|
||||
artifact_authority: &crate::ArtifactTokenAuthority,
|
||||
action_artifact_url: &url::Url,
|
||||
) -> Result<control_message::Body, SessionError> {
|
||||
let secret_capability = if assignment.secrets().is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
authority.issue(
|
||||
assignment.run(),
|
||||
assignment.job(),
|
||||
node,
|
||||
assignment.fence(),
|
||||
SystemTime::now(),
|
||||
)?
|
||||
};
|
||||
let actions_runtime_token = artifact_authority.issue(
|
||||
assignment.run(),
|
||||
assignment.number(),
|
||||
assignment.job(),
|
||||
node,
|
||||
assignment.fence(),
|
||||
SystemTime::now(),
|
||||
)?;
|
||||
Ok(control_message::Body::Assignment(JobAssignment {
|
||||
run: assignment.run().to_string(),
|
||||
job: assignment.job().to_string(),
|
||||
plan: assignment.plan().to_vec(),
|
||||
fence: assignment.fence().get(),
|
||||
origin: Some(declaration::origin(
|
||||
assignment.number(),
|
||||
assignment.origin(),
|
||||
repository,
|
||||
)),
|
||||
needs: assignment
|
||||
.needs()
|
||||
.iter()
|
||||
.map(|dependency| Dependency {
|
||||
key: dependency.key().to_owned(),
|
||||
conclusion: wire_conclusion(dependency.conclusion()) as i32,
|
||||
outputs: dependency.outputs().clone().into_iter().collect(),
|
||||
})
|
||||
.collect(),
|
||||
secrets: assignment.secrets().to_vec(),
|
||||
secret_capability,
|
||||
actions_runtime_token: actions_runtime_token.clone(),
|
||||
action_artifact_url: action_artifact_url.to_string(),
|
||||
action_artifact_capability: actions_runtime_token,
|
||||
}))
|
||||
}
|
||||
|
||||
const fn wire_conclusion(conclusion: syncode_control_runs::Conclusion) -> wire::Conclusion {
|
||||
match conclusion {
|
||||
syncode_control_runs::Conclusion::Success => wire::Conclusion::Success,
|
||||
syncode_control_runs::Conclusion::Failure => wire::Conclusion::Failure,
|
||||
syncode_control_runs::Conclusion::Cancelled => wire::Conclusion::Cancelled,
|
||||
syncode_control_runs::Conclusion::Skipped => wire::Conclusion::Skipped,
|
||||
}
|
||||
}
|
||||
use std::time::SystemTime;
|
||||
|
||||
use syncode_control_runs::NodeId;
|
||||
|
||||
use crate::error::SessionError;
|
||||
use crate::wire::{Dependency, JobAssignment, control_message};
|
||||
use crate::{declaration, wire};
|
||||
|
||||
pub(super) fn assignment_body(
|
||||
assignment: &syncode_control_runs::Assignment,
|
||||
repository: String,
|
||||
node: NodeId,
|
||||
authority: &crate::CapabilityAuthority,
|
||||
artifact_authority: &crate::ArtifactTokenAuthority,
|
||||
action_artifact_url: &url::Url,
|
||||
) -> Result<control_message::Body, SessionError> {
|
||||
let secrets = assignment_secrets(assignment.origin().repository(), assignment.secrets());
|
||||
let secret_capability = if secrets.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
authority.issue(
|
||||
assignment.run(),
|
||||
assignment.job(),
|
||||
node,
|
||||
assignment.fence(),
|
||||
SystemTime::now(),
|
||||
)?
|
||||
};
|
||||
let actions_runtime_token = artifact_authority.issue(
|
||||
assignment.run(),
|
||||
assignment.number(),
|
||||
assignment.job(),
|
||||
node,
|
||||
assignment.fence(),
|
||||
SystemTime::now(),
|
||||
)?;
|
||||
Ok(control_message::Body::Assignment(JobAssignment {
|
||||
run: assignment.run().to_string(),
|
||||
job: assignment.job().to_string(),
|
||||
plan: assignment.plan().to_vec(),
|
||||
fence: assignment.fence().get(),
|
||||
origin: Some(declaration::origin(
|
||||
assignment.number(),
|
||||
assignment.origin(),
|
||||
repository,
|
||||
)),
|
||||
needs: assignment
|
||||
.needs()
|
||||
.iter()
|
||||
.map(|dependency| Dependency {
|
||||
key: dependency.key().to_owned(),
|
||||
conclusion: wire_conclusion(dependency.conclusion()) as i32,
|
||||
outputs: dependency.outputs().clone().into_iter().collect(),
|
||||
})
|
||||
.collect(),
|
||||
secrets,
|
||||
secret_capability,
|
||||
actions_runtime_token: actions_runtime_token.clone(),
|
||||
action_artifact_url: action_artifact_url.to_string(),
|
||||
action_artifact_capability: actions_runtime_token,
|
||||
}))
|
||||
}
|
||||
|
||||
fn assignment_secrets(repository: &str, requested: &[String]) -> Vec<String> {
|
||||
let mut secrets = requested.to_vec();
|
||||
if uuid::Uuid::parse_str(repository).is_ok()
|
||||
&& !secrets
|
||||
.iter()
|
||||
.any(|name| name == crate::REPOSITORY_TOKEN_SECRET)
|
||||
{
|
||||
secrets.push(crate::REPOSITORY_TOKEN_SECRET.to_owned());
|
||||
}
|
||||
secrets
|
||||
}
|
||||
|
||||
const fn wire_conclusion(conclusion: syncode_control_runs::Conclusion) -> wire::Conclusion {
|
||||
match conclusion {
|
||||
syncode_control_runs::Conclusion::Success => wire::Conclusion::Success,
|
||||
syncode_control_runs::Conclusion::Failure => wire::Conclusion::Failure,
|
||||
syncode_control_runs::Conclusion::Cancelled => wire::Conclusion::Cancelled,
|
||||
syncode_control_runs::Conclusion::Skipped => wire::Conclusion::Skipped,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::assignment_secrets;
|
||||
use crate::REPOSITORY_TOKEN_SECRET;
|
||||
|
||||
#[test]
|
||||
fn native_assignments_receive_one_repository_token() {
|
||||
let repository = uuid::Uuid::new_v4().to_string();
|
||||
let requested = vec![REPOSITORY_TOKEN_SECRET.to_owned()];
|
||||
|
||||
assert_eq!(assignment_secrets(&repository, &[]), requested);
|
||||
assert_eq!(assignment_secrets(&repository, &requested), requested);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_assignments_do_not_receive_repository_tokens() {
|
||||
assert!(assignment_secrets("syncode/control", &[]).is_empty());
|
||||
}
|
||||
}
|
||||
@@ -1,0 +1,67 @@
|
||||
use syncode_control_node::identity_wire::IssueWorkflowRepositoryTokenRequest;
|
||||
use syncode_control_node::identity_wire::identity_client::IdentityClient;
|
||||
use thiserror::Error;
|
||||
use tonic::Request;
|
||||
use tonic::metadata::{Ascii, MetadataValue};
|
||||
use tonic::transport::Channel;
|
||||
|
||||
use crate::secrets::RepositoryCredentialSource;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct IdentityRepositoryCredentials {
|
||||
client: IdentityClient<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum IdentityRepositoryCredentialsError {
|
||||
#[error("cannot authorize identity requests")]
|
||||
Authorization,
|
||||
#[error("cannot connect to identity: {0}")]
|
||||
Connect(#[from] tonic::transport::Error),
|
||||
#[error("identity rejected the workflow repository credential: {0}")]
|
||||
Request(#[from] tonic::Status),
|
||||
#[error("identity returned an empty workflow repository credential")]
|
||||
Empty,
|
||||
}
|
||||
|
||||
impl IdentityRepositoryCredentials {
|
||||
pub async fn connect(
|
||||
endpoint: String,
|
||||
shared_secret: String,
|
||||
) -> Result<Self, IdentityRepositoryCredentialsError> {
|
||||
let client = IdentityClient::connect(endpoint).await?;
|
||||
let authorization = format!("Bearer {shared_secret}")
|
||||
.parse()
|
||||
.map_err(|_| IdentityRepositoryCredentialsError::Authorization)?;
|
||||
Ok(Self {
|
||||
client,
|
||||
authorization,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl RepositoryCredentialSource for IdentityRepositoryCredentials {
|
||||
type Error = IdentityRepositoryCredentialsError;
|
||||
|
||||
async fn issue(&self, user: &str, repository: &str) -> Result<String, Self::Error> {
|
||||
let mut request = Request::new(IssueWorkflowRepositoryTokenRequest {
|
||||
user_id: user.to_owned(),
|
||||
repository_id: repository.to_owned(),
|
||||
});
|
||||
request
|
||||
.metadata_mut()
|
||||
.insert("authorization", self.authorization.clone());
|
||||
let token = self
|
||||
.client
|
||||
.clone()
|
||||
.issue_workflow_repository_token(request)
|
||||
.await?
|
||||
.into_inner()
|
||||
.token;
|
||||
if token.is_empty() {
|
||||
return Err(IdentityRepositoryCredentialsError::Empty);
|
||||
}
|
||||
Ok(token)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user