diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -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; diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -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, - - /// 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, - - #[arg( - long, - env = "SYNCODE_ACTION_OCI_REGISTRIES", - value_delimiter = ',', - num_args = 1.. - )] - action_oci_registries: Vec, - - #[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, - - /// 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:` or - /// `repository:/`. - #[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> { - 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::()?.into(); - let token = nodes.issue_token(scope).await?; - println!("{}", token.secret().expose()); - } - Command::RevokeNode { node } => { - nodes.revoke(node.parse::()?).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::()?; - let action_allowlist = arguments - .action_allowlist - .into_iter() - .map(|repository| { - repository - .to_string() - .parse::() - }) - .collect::, _>>()?; - 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, + + /// 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, + + #[arg( + long, + env = "SYNCODE_ACTION_OCI_REGISTRIES", + value_delimiter = ',', + num_args = 1.. + )] + action_oci_registries: Vec, + + #[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, + + /// 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:` or + /// `repository:/`. + #[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> { + 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::()?.into(); + let token = nodes.issue_token(scope).await?; + println!("{}", token.secret().expose()); + } + Command::RevokeNode { node } => { + nodes.revoke(node.parse::()?).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::()?; + let action_allowlist = arguments + .action_allowlist + .into_iter() + .map(|repository| { + repository + .to_string() + .parse::() + }) + .collect::, _>>()?; + 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; + } +} diff --git a/src/secrets.rs b/src/secrets.rs --- a/src/secrets.rs +++ b/src/secrets.rs @@ -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, Self::Error>> + Send; -} - -pub struct RuntimeSecrets { - runs: Runs, - nodes: Nodes, - source: F, - authority: CapabilityAuthority, -} - -impl RuntimeSecrets { - pub const fn new( - runs: Runs, - nodes: Nodes, - source: F, - authority: CapabilityAuthority, - ) -> Self { - Self { - runs, - nodes, - source, - authority, - } - } -} - -#[tonic::async_trait] -impl RuntimeSecretsService for RuntimeSecrets -where - L: RunLog + 'static, - S: NodeStore + 'static, - F: SecretSource, -{ - async fn resolve( - &self, - request: Request, - ) -> Result, 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::().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 RuntimeSecrets { - async fn authorize( - &self, - claims: syncode_control_node::CapabilityClaims, - name: &str, - ) -> Result { - 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, Self::Error>> + Send; +} + +pub trait RepositoryCredentialSource: Send + Sync + 'static { + type Error: Error + Send + Sync + 'static; + + fn issue( + &self, + user: &str, + repository: &str, + ) -> impl Future> + Send; +} + +pub struct RuntimeSecrets { + runs: Runs, + nodes: Nodes, + source: F, + repository_credentials: R, + authority: CapabilityAuthority, +} + +impl RuntimeSecrets { + pub const fn new( + runs: Runs, + nodes: Nodes, + source: F, + repository_credentials: R, + authority: CapabilityAuthority, + ) -> Self { + Self { + runs, + nodes, + source, + repository_credentials, + authority, + } + } +} + +#[tonic::async_trait] +impl RuntimeSecretsService for RuntimeSecrets +where + L: RunLog + 'static, + S: NodeStore + 'static, + F: SecretSource, + R: RepositoryCredentialSource, +{ + async fn resolve( + &self, + request: Request, + ) -> Result, 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::().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 RuntimeSecrets { + async fn authorize( + &self, + claims: syncode_control_node::CapabilityClaims, + name: &str, + ) -> Result { + 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}"))) + } +} diff --git a/tests/action_repository.rs b/tests/action_repository.rs --- a/tests/action_repository.rs +++ b/tests/action_repository.rs @@ -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 = Result>; - -#[derive(Default)] -struct FixtureIdentity; - -#[tonic::async_trait] -impl Identity for FixtureIdentity { - async fn get_repository_coordinates( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("get_repository_coordinates")) - } - - async fn validate_session( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("validate_session")) - } - - async fn check_capability( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("check_capability")) - } - - async fn resolve_repository( - &self, - request: Request, - ) -> Result, 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 { - 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::, _>>()?; - 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 = Result>; + +#[derive(Default)] +struct FixtureIdentity; + +#[tonic::async_trait] +impl Identity for FixtureIdentity { + async fn issue_workflow_repository_token( + &self, + request: Request, + ) -> Result, 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, + ) -> Result, Status> { + Err(Status::unimplemented("get_repository_coordinates")) + } + + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + request: Request, + ) -> Result, 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 { + 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::, _>>()?; + 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(()) +} diff --git a/tests/projection_client.rs b/tests/projection_client.rs --- a/tests/projection_client.rs +++ b/tests/projection_client.rs @@ -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 = Result>; - -struct FixtureIdentity; - -#[tonic::async_trait] -impl Identity for FixtureIdentity { - async fn validate_session( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("validate_session")) - } - - async fn check_capability( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("check_capability")) - } - - async fn resolve_repository( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("resolve_repository")) - } - - async fn get_repository_coordinates( - &self, - request: Request, - ) -> Result, 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| { - 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 = Result>; + +struct FixtureIdentity; + +#[tonic::async_trait] +impl Identity for FixtureIdentity { + async fn issue_workflow_repository_token( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("issue_workflow_repository_token")) + } + + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("resolve_repository")) + } + + async fn get_repository_coordinates( + &self, + request: Request, + ) -> Result, 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| { + 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(()) +} diff --git a/tests/webhook.rs b/tests/webhook.rs --- a/tests/webhook.rs +++ b/tests/webhook.rs @@ -1,735 +1,748 @@ -#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] - -#[path = "support/actions.rs"] -mod actions; -#[path = "support/repository.rs"] -mod repository; - -use std::error::Error; -use std::sync::Arc; -use std::time::Duration; - -use base64::Engine; -use base64::engine::general_purpose::STANDARD; -use hmac::{Hmac, KeyInit, Mac}; -use sha2::Sha256; -use syncode_control::repository::RepositoryContents; -use syncode_control::repository_grpc::NativeRepositoryContents; -use syncode_control::repository_sources::RepositorySources; -use syncode_control::webhook::{Intake, router}; -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_node::wire::{ - Capabilities, Capacity, ControlMessage, Enrol, Hello, NodeMessage, control_message, - node_message, -}; -use syncode_control_node::{ - ArtifactTokenAuthority, CapabilityAuthority, GeneratedServer, NodeSessionClient, - NodeSessionServer, ProjectionClient, -}; -use syncode_control_nodes::{Ephemeral, Nodes, Scope}; -use syncode_control_runs::{Forgotten, NodeId, Runs}; -use syncode_workflow::VersionedPlan; -use syncode_workflow_github_actions::expression::ExpressionProgram; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpListener; -use tokio::sync::mpsc; -use tokio_stream::StreamExt; -use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; -use tonic::transport::Server; -use tonic::{Request, Response, Status, Streaming}; -use url::Url; - -use actions::FIXTURE_ACTIONS; - -type TestResult = Result>; - -struct ProjectionIdentity; - -#[tonic::async_trait] -impl Identity for ProjectionIdentity { - async fn validate_session( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("validate_session")) - } - - async fn check_capability( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("check_capability")) - } - - async fn resolve_repository( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("resolve_repository")) - } - - async fn get_repository_coordinates( - &self, - _request: Request, - ) -> Result, Status> { - Ok(Response::new(GetRepositoryCoordinatesResponse { - owner: "syncode".to_owned(), - name: "fixture".to_owned(), - })) - } -} - -const SECRET: &str = "the secret the hook was configured with"; -const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c"; - -const WORKFLOW: &str = r#" -name: CI -on: - push: - branches: [main] - tags: ["v*"] - pull_request: - branches: [main] - paths: - - "src/**" - workflow_dispatch: - schedule: - - cron: "0 6 * * *" -jobs: - build: - runs-on: [self-hosted, linux] - steps: - - run: echo "the run came from a push" -"#; - -const OTHER_MANUAL_WORKFLOW: &str = r#" -name: Other manual workflow -on: - push: - workflow_dispatch: -jobs: - other: - runs-on: [self-hosted, linux] - steps: - - run: echo "the other workflow must not run" -"#; - -/// A forge that serves one workflow directory and has nothing in the other. -async fn forge() -> TestResult { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let base = Url::parse(&format!("http://{}/", listener.local_addr()?))?; - let listing = format!( - r#"[{{"path":".gitea/workflows/ci.yml","type":"file","content":"{}"}},{{"path":".gitea/workflows/other.yml","type":"file","content":"{}"}}]"#, - STANDARD.encode(WORKFLOW), - STANDARD.encode(OTHER_MANUAL_WORKFLOW) - ); - - tokio::spawn(async move { - // Answers by what was asked for rather than in a fixed order: a push - // never asks for a file list, a pull request does. - let files = r#"[{"filename":"src/main.rs"}]"#.to_owned(); - loop { - let Ok((mut stream, _)) = listener.accept().await else { - return; - }; - let mut buffer = vec![0_u8; 4096]; - let read = stream.read(&mut buffer).await.unwrap_or(0); - let request = String::from_utf8_lossy(&buffer[..read]).to_string(); - let asked = request.lines().next().unwrap_or_default().to_owned(); - - let (status, body) = if asked.contains("/pulls/") { - (200, files.clone()) - } else if asked.contains(".gitea%2Fworkflows") || asked.contains(".gitea/workflows") { - (200, listing.clone()) - } else { - (404, String::from("{}")) - }; - - let response = format!( - "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", - body.len() - ); - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.shutdown().await; - } - }); - - Ok(base) -} - -struct Harness { - runs: Runs, - nodes: Nodes, - session: String, - events: String, -} - -/// Both ends of the control plane over one set of registries, exactly as the -/// binary wires them. -async fn start() -> TestResult { - start_mode(false).await -} - -async fn start_mode(shadow: bool) -> TestResult { - let runs = Runs::restored(Forgotten::default()).await?; - let nodes = Nodes::restored(Ephemeral::default()).await?; - let projection = projection_client().await?; - - let sessions = TcpListener::bind("127.0.0.1:0").await?; - let session = format!("http://{}", sessions.local_addr()?); - let served = (runs.clone(), nodes.clone()); - tokio::spawn(async move { - let session = if shadow { - NodeSessionServer::shadow( - served.0, - served.1, - capability_authority(), - artifact_authority(), - "http://127.0.0.1:1/".parse().expect("artifact URL"), - projection, - ) - } else { - NodeSessionServer::new( - served.0, - served.1, - capability_authority(), - artifact_authority(), - "http://127.0.0.1:1/".parse().expect("artifact URL"), - projection, - ) - }; - let _ = Server::builder() - .add_service(GeneratedServer::new(session)) - .serve_with_incoming(TcpListenerStream::new(sessions)) - .await; - }); - - 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 native = NativeRepositoryContents::connect(repository_endpoint).await?; - - let deliveries = TcpListener::bind("127.0.0.1:0").await?; - let events = format!("http://{}", deliveries.local_addr()?); - let intake = Arc::new(Intake::new( - RepositorySources::new( - native, - RepositoryContents::new(forge().await?, "a repository token".to_owned()), - ), - runs.clone(), - FIXTURE_ACTIONS, - SECRET.to_owned(), - )); - tokio::spawn(async move { - let _ = axum::serve(deliveries, router(intake)).await; - }); - - Ok(Harness { - runs, - nodes, - session, - events, - }) -} - -fn capability_authority() -> CapabilityAuthority { - CapabilityAuthority::new("test-capability-key").expect("capability key") -} - -fn artifact_authority() -> ArtifactTokenAuthority { - ArtifactTokenAuthority::new("test-artifact-key").expect("artifact key") -} - -async fn projection_client() -> TestResult { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let base = Url::parse(&format!("http://{}/", listener.local_addr()?))?; - tokio::spawn(async move { - let app = axum::Router::new().fallback(|| async { axum::http::StatusCode::NO_CONTENT }); - if let Err(error) = axum::serve(listener, app).await { - eprintln!("projection fixture failed: {error}"); - } - }); - 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(ProjectionIdentity)) - .serve_with_incoming(TcpListenerStream::new(identity_listener)) - .await; - }); - Ok(ProjectionClient::new( - base, - String::new(), - identity_endpoint, - String::new(), - )?) -} - -async fn open_node( - harness: &Harness, -) -> TestResult<(mpsc::Sender, Streaming)> { - let (sender, receiver) = mpsc::channel(8); - let mut client = NodeSessionClient::connect(harness.session.clone()).await?; - let mut inbound = client - .open(ReceiverStream::new(receiver)) - .await? - .into_inner(); - let token = harness.nodes.issue_token(Scope::Instance).await?; - sender - .send(NodeMessage { - sequence: 1, - message_id: "node-message-1".to_owned(), - idempotency_key: "node-message-1".to_owned(), - body: Some(node_message::Body::Enrol(Enrol { - token: token.secret().expose().to_owned(), - })), - }) - .await?; - let message = next(&mut inbound).await?; - let Some(control_message::Body::Enrolled(enrolled)) = message.body else { - return Err("expected an enrolled node".into()); - }; - sender - .send(NodeMessage { - sequence: 2, - message_id: "node-message-2".to_owned(), - idempotency_key: "node-message-2".to_owned(), - body: Some(node_message::Body::Hello(Hello { - node: enrolled.node, - credential: enrolled.credential, - capabilities: Some(Capabilities { - architecture: "arm64".to_owned(), - operating_system: "linux".to_owned(), - container_runtime: "docker".to_owned(), - container_runtime_version: "28.6.1".to_owned(), - cores: 2, - memory_bytes: 8 * 1024 * 1024 * 1024, - labels: vec!["self-hosted".to_owned(), "linux".to_owned()], - }), - capacity: Some(Capacity { - build_volume_free_bytes: 60 * 1024 * 1024 * 1024, - layer_store_bytes: 12 * 1024 * 1024 * 1024, - cache_volume_present: true, - cache_volume_total_bytes: 100, - cache_volume_used_bytes: 40, - cache_volume_path: "/var/cache/syncode".to_owned(), - cached_images: Vec::new(), - cached_actions: Vec::new(), - }), - max_parallel: 1, - })), - }) - .await?; - let welcome = next(&mut inbound).await?; - if !matches!(welcome.body, Some(control_message::Body::Welcome(_))) { - return Err("expected a welcome".into()); - } - Ok((sender, inbound)) -} - -fn push_body() -> String { - format!( - r#"{{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"{}","message_type":"repository.ref.updated","sequence":2,"term":1,"payload":{{"request":{{"principal_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89213"}},"changes":[{{"name_hex":"726566732f68656164732f6d61696e","old":{{"kind":"object","object_id":"1111111111111111111111111111111111111111"}},"new":{{"kind":"object","object_id":"{COMMIT}"}}}}]}}}}"#, - repository::REPOSITORY - ) -} - -fn tag_body() -> String { - format!( - r#"{{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"{}","message_type":"repository.ref.updated","sequence":2,"term":1,"payload":{{"request":{{"principal_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89213"}},"changes":[{{"name_hex":"726566732f746167732f76302e352e30","old":null,"new":{{"kind":"object","object_id":"{}"}}}}]}}}}"#, - repository::REPOSITORY, - repository::TAG_OBJECT - ) -} - -fn legacy_push_body() -> String { - format!( - r#"{{"ref":"refs/heads/main","after":"{COMMIT}","repository":{{"full_name":"syncode/demo"}},"commits":[{{"modified":["src/main.rs"]}}]}}"# - ) -} - -fn signature(body: &str) -> String { - let mut mac = Hmac::::new_from_slice(SECRET.as_bytes()).expect("key"); - mac.update(body.as_bytes()); - mac.finalize() - .into_bytes() - .iter() - .map(|byte| format!("{byte:02x}")) - .collect() -} - -async fn deliver(harness: &Harness, event: &str, body: &str, signed: bool) -> TestResult { - let signature = if signed { - signature(body) - } else { - signature("something else entirely") - }; - let response = reqwest::Client::new() - .post(format!("{}/events", harness.events)) - .header("x-syncode-event", event) - .header("x-syncode-signature", signature) - .header("x-syncode-delivery", format!("{event}-delivery")) - .header( - "x-syncode-workflow", - if event == "repository.ref.updated" { - ".gitea/workflows/ci.yml" - } else { - "" - }, - ) - .body(body.to_owned()) - .send() - .await?; - Ok(response.status().as_u16()) -} - -async fn deliver_native_pull_request(harness: &Harness, body: &str) -> TestResult { - let response = reqwest::Client::new() - .post(format!("{}/events", harness.events)) - .header("x-syncode-event", "pull_request") - .header("x-syncode-signature", signature(body)) - .header("x-syncode-delivery", "native-pull-request-delivery") - .header("x-syncode-repository-id", repository::REPOSITORY) - .header( - "x-syncode-before", - "1111111111111111111111111111111111111111", - ) - .header("x-syncode-after", COMMIT) - .body(body.to_owned()) - .send() - .await?; - Ok(response.status().as_u16()) -} - -async fn next(stream: &mut Streaming) -> TestResult { - Ok(stream - .next() - .await - .ok_or("the control plane closed the stream")??) -} - -#[tokio::test] -async fn a_push_becomes_a_plan_the_node_is_handed() -> TestResult { - let harness = start().await?; - - let body = push_body(); - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, true).await?, - 202 - ); - - let (sender, receiver) = mpsc::channel(8); - let mut client = NodeSessionClient::connect(harness.session.clone()).await?; - let mut inbound = client - .open(ReceiverStream::new(receiver)) - .await? - .into_inner(); - - let token = harness.nodes.issue_token(Scope::Instance).await?; - sender - .send(NodeMessage { - sequence: 1, - message_id: "node-message-1".to_owned(), - idempotency_key: "node-message-1".to_owned(), - body: Some(node_message::Body::Enrol(Enrol { - token: token.secret().expose().to_owned(), - })), - }) - .await?; - let message = next(&mut inbound).await?; - let Some(control_message::Body::Enrolled(enrolled)) = message.body else { - panic!("expected an identity, got {:?}", message.body); - }; - - sender - .send(NodeMessage { - sequence: 2, - message_id: "node-message-2".to_owned(), - idempotency_key: "node-message-2".to_owned(), - body: Some(node_message::Body::Hello(Hello { - node: enrolled.node, - credential: enrolled.credential, - capabilities: Some(Capabilities { - architecture: "arm64".to_owned(), - operating_system: "linux".to_owned(), - container_runtime: "docker".to_owned(), - container_runtime_version: "28.6.1".to_owned(), - cores: 2, - memory_bytes: 8 * 1024 * 1024 * 1024, - labels: vec!["self-hosted".to_owned(), "linux".to_owned()], - }), - capacity: Some(Capacity { - build_volume_free_bytes: 60 * 1024 * 1024 * 1024, - layer_store_bytes: 12 * 1024 * 1024 * 1024, - cache_volume_present: true, - cache_volume_total_bytes: 100, - cache_volume_used_bytes: 40, - cache_volume_path: "/var/cache/syncode".to_owned(), - cached_images: Vec::new(), - cached_actions: Vec::new(), - }), - max_parallel: 1, - })), - }) - .await?; - - let welcome = next(&mut inbound).await?; - assert!(matches!( - welcome.body, - Some(control_message::Body::Welcome(_)) - )); - - let assignment = next(&mut inbound).await?; - let Some(control_message::Body::Assignment(assignment)) = assignment.body else { - panic!("expected the plan the push produced, got {assignment:?}"); - }; - // The node is handed a plan, not the workflow file the push carried. - let plan: VersionedPlan = serde_json::from_slice(&assignment.plan)?; - assert_eq!(plan.schema(), syncode_workflow::PlanSchemaVersion::CURRENT); - - // A plan says nothing about what it is being built from, so the assignment - // states it: without this a job cannot check anything out. - let origin = assignment.origin.ok_or("the assignment stated no origin")?; - assert_eq!(origin.number, 1); - assert_eq!(origin.repository, repository::REPOSITORY); - assert_eq!(origin.commit, COMMIT); - assert_eq!(origin.reference, "refs/heads/main"); - assert_eq!(origin.event, "push"); - - Ok(()) -} - -#[tokio::test] -async fn an_annotated_tag_push_uses_the_target_commit_and_tag_reference() -> TestResult { - let harness = start().await?; - let body = tag_body(); - - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, true).await?, - 202 - ); - let assignment = harness - .runs - .take_next(NodeId::fresh()) - .await? - .ok_or("the tag push produced no run")?; - - assert_eq!(assignment.origin().commit(), COMMIT); - assert_eq!(assignment.origin().reference(), "refs/tags/v0.5.0"); - assert_eq!(assignment.origin().event(), "push"); - Ok(()) -} - -#[tokio::test] -async fn a_delivery_nobody_signed_queues_nothing() -> TestResult { - let harness = start().await?; - - let body = push_body(); - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, false).await?, - 401 - ); - assert!( - harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .is_none() - ); - - Ok(()) -} - -#[tokio::test] -async fn a_retried_delivery_does_not_duplicate_its_run() -> TestResult { - let harness = start().await?; - let body = push_body(); - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, true).await?, - 202 - ); - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, true).await?, - 202 - ); - - let first = harness - .runs - .take_next(NodeId::fresh()) - .await? - .ok_or("the delivery produced no run")?; - assert!(harness.runs.take_next(NodeId::fresh()).await?.is_none()); - assert_eq!(first.origin().event(), "push"); - Ok(()) -} - -#[tokio::test] -async fn an_event_the_workflow_does_not_declare_is_accepted_and_ignored() -> TestResult { - let harness = start().await?; - - let body = push_body(); - assert_eq!(deliver(&harness, "issues", &body, true).await?, 202); - assert!( - harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .is_none() - ); - - Ok(()) -} - -#[tokio::test] -async fn a_legacy_forge_push_is_accepted_without_a_duplicate_run() -> TestResult { - let harness = start().await?; - let body = legacy_push_body(); - assert_eq!(deliver(&harness, "push", &body, true).await?, 202); - assert!(harness.runs.take_next(NodeId::fresh()).await?.is_none()); - Ok(()) -} - -fn pull_request_body(action: &str) -> String { - format!( - r#"{{"action":"{action}","number":7,"repository":{{"full_name":"syncode/demo"}},"pull_request":{{"head":{{"sha":"{COMMIT}"}},"base":{{"ref":"main"}}}}}}"# - ) -} - -#[tokio::test] -async fn a_pull_request_becomes_a_run_using_the_files_it_touches() -> TestResult { - let harness = start().await?; - - let body = pull_request_body("opened"); - assert_eq!(deliver(&harness, "pull_request", &body, true).await?, 202); - - // The workflow filters on `paths: src/**`. The webhook body carries no file - // list, so unless the control plane asks the forge for one, this run does - // not exist and nothing says why. - let assignment = harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .ok_or("the pull request produced no run")?; - let plan: VersionedPlan = serde_json::from_slice(assignment.plan())?; - assert_eq!(plan.schema(), syncode_workflow::PlanSchemaVersion::CURRENT); - - // A pull request head sits on no branch this control plane knows, but the - // forge publishes it under the pull request, which is what a checkout can - // fetch. - assert_eq!(assignment.origin().reference(), "refs/pull/7/head"); - assert_eq!(assignment.origin().event(), "pull_request"); - Ok(()) -} - -#[tokio::test] -async fn a_repository_plane_pull_request_reads_content_over_grpc() -> TestResult { - let harness = start().await?; - - let body = pull_request_body("opened"); - assert_eq!(deliver_native_pull_request(&harness, &body).await?, 202); - - let assignment = harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .ok_or("the native pull request produced no run")?; - assert_eq!(assignment.origin().repository(), "syncode/demo"); - assert_eq!(assignment.origin().reference(), "refs/pull/7/head"); - Ok(()) -} - -#[tokio::test] -async fn closing_a_pull_request_starts_nothing() -> TestResult { - let harness = start().await?; - - let body = pull_request_body("closed"); - assert_eq!(deliver(&harness, "pull_request", &body, true).await?, 202); - assert!( - harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .is_none() - ); - Ok(()) -} - -#[tokio::test] -async fn manual_and_scheduled_deliveries_become_runs() -> TestResult { - for event in ["workflow_dispatch", "schedule"] { - let harness = start().await?; - let body = format!( - r#"{{"ref":"refs/heads/main","after":"{COMMIT}","workflow":".gitea/workflows/ci.yml","repository":{{"full_name":"syncode/demo"}}}}"# - ); - assert_eq!(deliver(&harness, event, &body, true).await?, 202); - let assignment = harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .ok_or("event produced no run")?; - assert_eq!(assignment.origin().event(), event); - assert_eq!(assignment.origin().reference(), "refs/heads/main"); - assert!( - harness - .runs - .take_next(syncode_control_runs::NodeId::fresh()) - .await? - .is_none(), - "a requested event started a workflow other than the selected file" - ); - } - Ok(()) -} - -#[tokio::test] -async fn a_manual_delivery_can_select_a_tag() -> TestResult { - let harness = start().await?; - let body = format!( - r#"{{"ref":"refs/tags/v0.4.0","after":"{COMMIT}","workflow":".gitea/workflows/ci.yml","repository":{{"full_name":"syncode/demo"}}}}"# - ); - - assert_eq!( - deliver(&harness, "workflow_dispatch", &body, true).await?, - 202 - ); - let assignment = harness - .runs - .take_next(NodeId::fresh()) - .await? - .ok_or("manual tag event produced no run")?; - assert_eq!(assignment.origin().reference(), "refs/tags/v0.4.0"); - Ok(()) -} - -#[tokio::test] -async fn shadow_mode_persists_the_run_but_never_assigns_it() -> TestResult { - let harness = start_mode(true).await?; - let body = push_body(); - assert_eq!( - deliver(&harness, "repository.ref.updated", &body, true).await?, - 202 - ); - let (_sender, mut inbound) = open_node(&harness).await?; - - assert!( - tokio::time::timeout(Duration::from_millis(100), inbound.next()) - .await - .is_err(), - "shadow mode must not assign the run" - ); - assert!( - harness.runs.take_next(NodeId::fresh()).await?.is_some(), - "the shadow run must still exist in the aggregate" - ); - Ok(()) -} +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] + +#[path = "support/actions.rs"] +mod actions; +#[path = "support/repository.rs"] +mod repository; + +use std::error::Error; +use std::sync::Arc; +use std::time::Duration; + +use base64::Engine; +use base64::engine::general_purpose::STANDARD; +use hmac::{Hmac, KeyInit, Mac}; +use sha2::Sha256; +use syncode_control::repository::RepositoryContents; +use syncode_control::repository_grpc::NativeRepositoryContents; +use syncode_control::repository_sources::RepositorySources; +use syncode_control::webhook::{Intake, router}; +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_node::wire::{ + Capabilities, Capacity, ControlMessage, Enrol, Hello, NodeMessage, control_message, + node_message, +}; +use syncode_control_node::{ + ArtifactTokenAuthority, CapabilityAuthority, GeneratedServer, NodeSessionClient, + NodeSessionServer, ProjectionClient, +}; +use syncode_control_nodes::{Ephemeral, Nodes, Scope}; +use syncode_control_runs::{Forgotten, NodeId, Runs}; +use syncode_workflow::VersionedPlan; +use syncode_workflow_github_actions::expression::ExpressionProgram; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio_stream::StreamExt; +use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream}; +use tonic::transport::Server; +use tonic::{Request, Response, Status, Streaming}; +use url::Url; + +use actions::FIXTURE_ACTIONS; + +type TestResult = Result>; + +struct ProjectionIdentity; + +#[tonic::async_trait] +impl Identity for ProjectionIdentity { + async fn issue_workflow_repository_token( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("issue_workflow_repository_token")) + } + + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("resolve_repository")) + } + + async fn get_repository_coordinates( + &self, + _request: Request, + ) -> Result, Status> { + Ok(Response::new(GetRepositoryCoordinatesResponse { + owner: "syncode".to_owned(), + name: "fixture".to_owned(), + })) + } +} + +const SECRET: &str = "the secret the hook was configured with"; +const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c"; + +const WORKFLOW: &str = r#" +name: CI +on: + push: + branches: [main] + tags: ["v*"] + pull_request: + branches: [main] + paths: + - "src/**" + workflow_dispatch: + schedule: + - cron: "0 6 * * *" +jobs: + build: + runs-on: [self-hosted, linux] + steps: + - run: echo "the run came from a push" +"#; + +const OTHER_MANUAL_WORKFLOW: &str = r#" +name: Other manual workflow +on: + push: + workflow_dispatch: +jobs: + other: + runs-on: [self-hosted, linux] + steps: + - run: echo "the other workflow must not run" +"#; + +/// A forge that serves one workflow directory and has nothing in the other. +async fn forge() -> TestResult { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let base = Url::parse(&format!("http://{}/", listener.local_addr()?))?; + let listing = format!( + r#"[{{"path":".gitea/workflows/ci.yml","type":"file","content":"{}"}},{{"path":".gitea/workflows/other.yml","type":"file","content":"{}"}}]"#, + STANDARD.encode(WORKFLOW), + STANDARD.encode(OTHER_MANUAL_WORKFLOW) + ); + + tokio::spawn(async move { + // Answers by what was asked for rather than in a fixed order: a push + // never asks for a file list, a pull request does. + let files = r#"[{"filename":"src/main.rs"}]"#.to_owned(); + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let mut buffer = vec![0_u8; 4096]; + let read = stream.read(&mut buffer).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buffer[..read]).to_string(); + let asked = request.lines().next().unwrap_or_default().to_owned(); + + let (status, body) = if asked.contains("/pulls/") { + (200, files.clone()) + } else if asked.contains(".gitea%2Fworkflows") || asked.contains(".gitea/workflows") { + (200, listing.clone()) + } else { + (404, String::from("{}")) + }; + + let response = format!( + "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } + }); + + Ok(base) +} + +struct Harness { + runs: Runs, + nodes: Nodes, + session: String, + events: String, +} + +/// Both ends of the control plane over one set of registries, exactly as the +/// binary wires them. +async fn start() -> TestResult { + start_mode(false).await +} + +async fn start_mode(shadow: bool) -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let nodes = Nodes::restored(Ephemeral::default()).await?; + let projection = projection_client().await?; + + let sessions = TcpListener::bind("127.0.0.1:0").await?; + let session = format!("http://{}", sessions.local_addr()?); + let served = (runs.clone(), nodes.clone()); + tokio::spawn(async move { + let session = if shadow { + NodeSessionServer::shadow( + served.0, + served.1, + capability_authority(), + artifact_authority(), + "http://127.0.0.1:1/".parse().expect("artifact URL"), + projection, + ) + } else { + NodeSessionServer::new( + served.0, + served.1, + capability_authority(), + artifact_authority(), + "http://127.0.0.1:1/".parse().expect("artifact URL"), + projection, + ) + }; + let _ = Server::builder() + .add_service(GeneratedServer::new(session)) + .serve_with_incoming(TcpListenerStream::new(sessions)) + .await; + }); + + 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 native = NativeRepositoryContents::connect(repository_endpoint).await?; + + let deliveries = TcpListener::bind("127.0.0.1:0").await?; + let events = format!("http://{}", deliveries.local_addr()?); + let intake = Arc::new(Intake::new( + RepositorySources::new( + native, + RepositoryContents::new(forge().await?, "a repository token".to_owned()), + ), + runs.clone(), + FIXTURE_ACTIONS, + SECRET.to_owned(), + )); + tokio::spawn(async move { + let _ = axum::serve(deliveries, router(intake)).await; + }); + + Ok(Harness { + runs, + nodes, + session, + events, + }) +} + +fn capability_authority() -> CapabilityAuthority { + CapabilityAuthority::new("test-capability-key").expect("capability key") +} + +fn artifact_authority() -> ArtifactTokenAuthority { + ArtifactTokenAuthority::new("test-artifact-key").expect("artifact key") +} + +async fn projection_client() -> TestResult { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let base = Url::parse(&format!("http://{}/", listener.local_addr()?))?; + tokio::spawn(async move { + let app = axum::Router::new().fallback(|| async { axum::http::StatusCode::NO_CONTENT }); + if let Err(error) = axum::serve(listener, app).await { + eprintln!("projection fixture failed: {error}"); + } + }); + 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(ProjectionIdentity)) + .serve_with_incoming(TcpListenerStream::new(identity_listener)) + .await; + }); + Ok(ProjectionClient::new( + base, + String::new(), + identity_endpoint, + String::new(), + )?) +} + +async fn open_node( + harness: &Harness, +) -> TestResult<(mpsc::Sender, Streaming)> { + let (sender, receiver) = mpsc::channel(8); + let mut client = NodeSessionClient::connect(harness.session.clone()).await?; + let mut inbound = client + .open(ReceiverStream::new(receiver)) + .await? + .into_inner(); + let token = harness.nodes.issue_token(Scope::Instance).await?; + sender + .send(NodeMessage { + sequence: 1, + message_id: "node-message-1".to_owned(), + idempotency_key: "node-message-1".to_owned(), + body: Some(node_message::Body::Enrol(Enrol { + token: token.secret().expose().to_owned(), + })), + }) + .await?; + let message = next(&mut inbound).await?; + let Some(control_message::Body::Enrolled(enrolled)) = message.body else { + return Err("expected an enrolled node".into()); + }; + sender + .send(NodeMessage { + sequence: 2, + message_id: "node-message-2".to_owned(), + idempotency_key: "node-message-2".to_owned(), + body: Some(node_message::Body::Hello(Hello { + node: enrolled.node, + credential: enrolled.credential, + capabilities: Some(Capabilities { + architecture: "arm64".to_owned(), + operating_system: "linux".to_owned(), + container_runtime: "docker".to_owned(), + container_runtime_version: "28.6.1".to_owned(), + cores: 2, + memory_bytes: 8 * 1024 * 1024 * 1024, + labels: vec!["self-hosted".to_owned(), "linux".to_owned()], + }), + capacity: Some(Capacity { + build_volume_free_bytes: 60 * 1024 * 1024 * 1024, + layer_store_bytes: 12 * 1024 * 1024 * 1024, + cache_volume_present: true, + cache_volume_total_bytes: 100, + cache_volume_used_bytes: 40, + cache_volume_path: "/var/cache/syncode".to_owned(), + cached_images: Vec::new(), + cached_actions: Vec::new(), + }), + max_parallel: 1, + })), + }) + .await?; + let welcome = next(&mut inbound).await?; + if !matches!(welcome.body, Some(control_message::Body::Welcome(_))) { + return Err("expected a welcome".into()); + } + Ok((sender, inbound)) +} + +fn push_body() -> String { + format!( + r#"{{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"{}","message_type":"repository.ref.updated","sequence":2,"term":1,"payload":{{"request":{{"principal_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89213"}},"changes":[{{"name_hex":"726566732f68656164732f6d61696e","old":{{"kind":"object","object_id":"1111111111111111111111111111111111111111"}},"new":{{"kind":"object","object_id":"{COMMIT}"}}}}]}}}}"#, + repository::REPOSITORY + ) +} + +fn tag_body() -> String { + format!( + r#"{{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"{}","message_type":"repository.ref.updated","sequence":2,"term":1,"payload":{{"request":{{"principal_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89213"}},"changes":[{{"name_hex":"726566732f746167732f76302e352e30","old":null,"new":{{"kind":"object","object_id":"{}"}}}}]}}}}"#, + repository::REPOSITORY, + repository::TAG_OBJECT + ) +} + +fn legacy_push_body() -> String { + format!( + r#"{{"ref":"refs/heads/main","after":"{COMMIT}","repository":{{"full_name":"syncode/demo"}},"commits":[{{"modified":["src/main.rs"]}}]}}"# + ) +} + +fn signature(body: &str) -> String { + let mut mac = Hmac::::new_from_slice(SECRET.as_bytes()).expect("key"); + mac.update(body.as_bytes()); + mac.finalize() + .into_bytes() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +async fn deliver(harness: &Harness, event: &str, body: &str, signed: bool) -> TestResult { + let signature = if signed { + signature(body) + } else { + signature("something else entirely") + }; + let response = reqwest::Client::new() + .post(format!("{}/events", harness.events)) + .header("x-syncode-event", event) + .header("x-syncode-signature", signature) + .header("x-syncode-delivery", format!("{event}-delivery")) + .header( + "x-syncode-workflow", + if event == "repository.ref.updated" { + ".gitea/workflows/ci.yml" + } else { + "" + }, + ) + .body(body.to_owned()) + .send() + .await?; + Ok(response.status().as_u16()) +} + +async fn deliver_native_pull_request(harness: &Harness, body: &str) -> TestResult { + let response = reqwest::Client::new() + .post(format!("{}/events", harness.events)) + .header("x-syncode-event", "pull_request") + .header("x-syncode-signature", signature(body)) + .header("x-syncode-delivery", "native-pull-request-delivery") + .header("x-syncode-repository-id", repository::REPOSITORY) + .header( + "x-syncode-before", + "1111111111111111111111111111111111111111", + ) + .header("x-syncode-after", COMMIT) + .body(body.to_owned()) + .send() + .await?; + Ok(response.status().as_u16()) +} + +async fn next(stream: &mut Streaming) -> TestResult { + Ok(stream + .next() + .await + .ok_or("the control plane closed the stream")??) +} + +#[tokio::test] +async fn a_push_becomes_a_plan_the_node_is_handed() -> TestResult { + let harness = start().await?; + + let body = push_body(); + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, true).await?, + 202 + ); + + let (sender, receiver) = mpsc::channel(8); + let mut client = NodeSessionClient::connect(harness.session.clone()).await?; + let mut inbound = client + .open(ReceiverStream::new(receiver)) + .await? + .into_inner(); + + let token = harness.nodes.issue_token(Scope::Instance).await?; + sender + .send(NodeMessage { + sequence: 1, + message_id: "node-message-1".to_owned(), + idempotency_key: "node-message-1".to_owned(), + body: Some(node_message::Body::Enrol(Enrol { + token: token.secret().expose().to_owned(), + })), + }) + .await?; + let message = next(&mut inbound).await?; + let Some(control_message::Body::Enrolled(enrolled)) = message.body else { + panic!("expected an identity, got {:?}", message.body); + }; + + sender + .send(NodeMessage { + sequence: 2, + message_id: "node-message-2".to_owned(), + idempotency_key: "node-message-2".to_owned(), + body: Some(node_message::Body::Hello(Hello { + node: enrolled.node, + credential: enrolled.credential, + capabilities: Some(Capabilities { + architecture: "arm64".to_owned(), + operating_system: "linux".to_owned(), + container_runtime: "docker".to_owned(), + container_runtime_version: "28.6.1".to_owned(), + cores: 2, + memory_bytes: 8 * 1024 * 1024 * 1024, + labels: vec!["self-hosted".to_owned(), "linux".to_owned()], + }), + capacity: Some(Capacity { + build_volume_free_bytes: 60 * 1024 * 1024 * 1024, + layer_store_bytes: 12 * 1024 * 1024 * 1024, + cache_volume_present: true, + cache_volume_total_bytes: 100, + cache_volume_used_bytes: 40, + cache_volume_path: "/var/cache/syncode".to_owned(), + cached_images: Vec::new(), + cached_actions: Vec::new(), + }), + max_parallel: 1, + })), + }) + .await?; + + let welcome = next(&mut inbound).await?; + assert!(matches!( + welcome.body, + Some(control_message::Body::Welcome(_)) + )); + + let assignment = next(&mut inbound).await?; + let Some(control_message::Body::Assignment(assignment)) = assignment.body else { + panic!("expected the plan the push produced, got {assignment:?}"); + }; + // The node is handed a plan, not the workflow file the push carried. + let plan: VersionedPlan = serde_json::from_slice(&assignment.plan)?; + assert_eq!(plan.schema(), syncode_workflow::PlanSchemaVersion::CURRENT); + + // A plan says nothing about what it is being built from, so the assignment + // states it: without this a job cannot check anything out. + let origin = assignment.origin.ok_or("the assignment stated no origin")?; + assert_eq!(origin.number, 1); + assert_eq!(origin.repository, "syncode/fixture"); + assert_eq!(origin.commit, COMMIT); + assert_eq!(origin.reference, "refs/heads/main"); + assert_eq!(origin.event, "push"); + assert_eq!( + assignment.secrets, + vec![syncode_control_node::REPOSITORY_TOKEN_SECRET] + ); + assert!(!assignment.secret_capability.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn an_annotated_tag_push_uses_the_target_commit_and_tag_reference() -> TestResult { + let harness = start().await?; + let body = tag_body(); + + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, true).await?, + 202 + ); + let assignment = harness + .runs + .take_next(NodeId::fresh()) + .await? + .ok_or("the tag push produced no run")?; + + assert_eq!(assignment.origin().commit(), COMMIT); + assert_eq!(assignment.origin().reference(), "refs/tags/v0.5.0"); + assert_eq!(assignment.origin().event(), "push"); + Ok(()) +} + +#[tokio::test] +async fn a_delivery_nobody_signed_queues_nothing() -> TestResult { + let harness = start().await?; + + let body = push_body(); + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, false).await?, + 401 + ); + assert!( + harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .is_none() + ); + + Ok(()) +} + +#[tokio::test] +async fn a_retried_delivery_does_not_duplicate_its_run() -> TestResult { + let harness = start().await?; + let body = push_body(); + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, true).await?, + 202 + ); + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, true).await?, + 202 + ); + + let first = harness + .runs + .take_next(NodeId::fresh()) + .await? + .ok_or("the delivery produced no run")?; + assert!(harness.runs.take_next(NodeId::fresh()).await?.is_none()); + assert_eq!(first.origin().event(), "push"); + Ok(()) +} + +#[tokio::test] +async fn an_event_the_workflow_does_not_declare_is_accepted_and_ignored() -> TestResult { + let harness = start().await?; + + let body = push_body(); + assert_eq!(deliver(&harness, "issues", &body, true).await?, 202); + assert!( + harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .is_none() + ); + + Ok(()) +} + +#[tokio::test] +async fn a_legacy_forge_push_is_accepted_without_a_duplicate_run() -> TestResult { + let harness = start().await?; + let body = legacy_push_body(); + assert_eq!(deliver(&harness, "push", &body, true).await?, 202); + assert!(harness.runs.take_next(NodeId::fresh()).await?.is_none()); + Ok(()) +} + +fn pull_request_body(action: &str) -> String { + format!( + r#"{{"action":"{action}","number":7,"repository":{{"full_name":"syncode/demo"}},"pull_request":{{"head":{{"sha":"{COMMIT}"}},"base":{{"ref":"main"}}}}}}"# + ) +} + +#[tokio::test] +async fn a_pull_request_becomes_a_run_using_the_files_it_touches() -> TestResult { + let harness = start().await?; + + let body = pull_request_body("opened"); + assert_eq!(deliver(&harness, "pull_request", &body, true).await?, 202); + + // The workflow filters on `paths: src/**`. The webhook body carries no file + // list, so unless the control plane asks the forge for one, this run does + // not exist and nothing says why. + let assignment = harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .ok_or("the pull request produced no run")?; + let plan: VersionedPlan = serde_json::from_slice(assignment.plan())?; + assert_eq!(plan.schema(), syncode_workflow::PlanSchemaVersion::CURRENT); + + // A pull request head sits on no branch this control plane knows, but the + // forge publishes it under the pull request, which is what a checkout can + // fetch. + assert_eq!(assignment.origin().reference(), "refs/pull/7/head"); + assert_eq!(assignment.origin().event(), "pull_request"); + Ok(()) +} + +#[tokio::test] +async fn a_repository_plane_pull_request_reads_content_over_grpc() -> TestResult { + let harness = start().await?; + + let body = pull_request_body("opened"); + assert_eq!(deliver_native_pull_request(&harness, &body).await?, 202); + + let assignment = harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .ok_or("the native pull request produced no run")?; + assert_eq!(assignment.origin().repository(), "syncode/demo"); + assert_eq!(assignment.origin().reference(), "refs/pull/7/head"); + Ok(()) +} + +#[tokio::test] +async fn closing_a_pull_request_starts_nothing() -> TestResult { + let harness = start().await?; + + let body = pull_request_body("closed"); + assert_eq!(deliver(&harness, "pull_request", &body, true).await?, 202); + assert!( + harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .is_none() + ); + Ok(()) +} + +#[tokio::test] +async fn manual_and_scheduled_deliveries_become_runs() -> TestResult { + for event in ["workflow_dispatch", "schedule"] { + let harness = start().await?; + let body = format!( + r#"{{"ref":"refs/heads/main","after":"{COMMIT}","workflow":".gitea/workflows/ci.yml","repository":{{"full_name":"syncode/demo"}}}}"# + ); + assert_eq!(deliver(&harness, event, &body, true).await?, 202); + let assignment = harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .ok_or("event produced no run")?; + assert_eq!(assignment.origin().event(), event); + assert_eq!(assignment.origin().reference(), "refs/heads/main"); + assert!( + harness + .runs + .take_next(syncode_control_runs::NodeId::fresh()) + .await? + .is_none(), + "a requested event started a workflow other than the selected file" + ); + } + Ok(()) +} + +#[tokio::test] +async fn a_manual_delivery_can_select_a_tag() -> TestResult { + let harness = start().await?; + let body = format!( + r#"{{"ref":"refs/tags/v0.4.0","after":"{COMMIT}","workflow":".gitea/workflows/ci.yml","repository":{{"full_name":"syncode/demo"}}}}"# + ); + + assert_eq!( + deliver(&harness, "workflow_dispatch", &body, true).await?, + 202 + ); + let assignment = harness + .runs + .take_next(NodeId::fresh()) + .await? + .ok_or("manual tag event produced no run")?; + assert_eq!(assignment.origin().reference(), "refs/tags/v0.4.0"); + Ok(()) +} + +#[tokio::test] +async fn shadow_mode_persists_the_run_but_never_assigns_it() -> TestResult { + let harness = start_mode(true).await?; + let body = push_body(); + assert_eq!( + deliver(&harness, "repository.ref.updated", &body, true).await?, + 202 + ); + let (_sender, mut inbound) = open_node(&harness).await?; + + assert!( + tokio::time::timeout(Duration::from_millis(100), inbound.next()) + .await + .is_err(), + "shadow mode must not assign the run" + ); + assert!( + harness.runs.take_next(NodeId::fresh()).await?.is_some(), + "the shadow run must still exist in the aggregate" + ); + Ok(()) +} diff --git a/crates/control-node/proto/identity.proto b/crates/control-node/proto/identity.proto --- a/crates/control-node/proto/identity.proto +++ b/crates/control-node/proto/identity.proto @@ -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; +} diff --git a/crates/control-node/src/lib.rs b/crates/control-node/src/lib.rs --- a/crates/control-node/src/lib.rs +++ b/crates/control-node/src/lib.rs @@ -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"; diff --git a/crates/control-node/src/session/dispatch/assignment.rs b/crates/control-node/src/session/dispatch/assignment.rs --- a/crates/control-node/src/session/dispatch/assignment.rs +++ b/crates/control-node/src/session/dispatch/assignment.rs @@ -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 { - 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 { + 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 { + 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()); + } +} diff --git a/src/repository_credentials.rs b/src/repository_credentials.rs --- /dev/null +++ b/src/repository_credentials.rs @@ -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, + authorization: MetadataValue, +} + +#[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 { + 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 { + 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) + } +}