feat: Complete native Actions delivery #53
+8
-1
@@ -1,222 +1,229 @@
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use serde::Deserialize;
|
||||
use thiserror::Error;
|
||||
use url::Url;
|
||||
|
||||
use crate::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};
|
||||
|
||||
/// The workflow directories supported by the repository service.
|
||||
const DIRECTORIES: [&str; 2] = [".gitea/workflows", ".github/workflows"];
|
||||
|
||||
pub struct RepositoryContents {
|
||||
base: Url,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
}
|
||||
|
||||
pub struct RepositorySecrets {
|
||||
base: Url,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RepositoryError {
|
||||
#[error("cannot reach the repository service: {0}")]
|
||||
Unreachable(#[from] reqwest::Error),
|
||||
|
||||
#[error("the repository service answered {status} for {path}")]
|
||||
Refused { status: u16, path: String },
|
||||
|
||||
#[error("the repository service returned a workflow file that is not valid base64: {0}")]
|
||||
Undecodable(#[from] base64::DecodeError),
|
||||
|
||||
#[error("cannot address {path} on the repository service")]
|
||||
Address { path: String },
|
||||
|
||||
#[error("the forge cannot compare native repository commits")]
|
||||
UnsupportedChangedFiles,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ChangedFile {
|
||||
filename: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct Entry {
|
||||
path: String,
|
||||
#[serde(rename = "type")]
|
||||
kind: String,
|
||||
content: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(untagged)]
|
||||
enum Contents {
|
||||
Many(Vec<Entry>),
|
||||
One(Entry),
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SecretResponse {
|
||||
value: String,
|
||||
}
|
||||
|
||||
impl RepositoryContents {
|
||||
#[must_use]
|
||||
pub fn new(base: Url, token: String) -> Self {
|
||||
Self {
|
||||
base,
|
||||
token,
|
||||
client: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn entries(&self, path: &str) -> Result<Vec<Entry>, RepositoryError> {
|
||||
let url = self
|
||||
.base
|
||||
.join(&format!("api/v1/repos/{path}"))
|
||||
.map_err(|_| RepositoryError::Address {
|
||||
path: path.to_owned(),
|
||||
})?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("Authorization", format!("token {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
// A repository without one of the directories is not a failure; it is a
|
||||
// repository that keeps its workflows in the other one, or in neither.
|
||||
if response.status() == reqwest::StatusCode::NOT_FOUND {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path: path.to_owned(),
|
||||
});
|
||||
}
|
||||
Ok(match response.json().await? {
|
||||
Contents::Many(entries) => entries,
|
||||
Contents::One(entry) => vec![entry],
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl RepositorySecrets {
|
||||
#[must_use]
|
||||
pub fn new(base: Url, token: String) -> Self {
|
||||
Self {
|
||||
base,
|
||||
token,
|
||||
client: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl crate::secrets::SecretSource for RepositorySecrets {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn resolve(
|
||||
&self,
|
||||
repository: &str,
|
||||
name: &str,
|
||||
) -> Result<Option<String>, RepositoryError> {
|
||||
let path = format!("api/internal/actions/syncode/secrets/{repository}/{name}");
|
||||
let url = self
|
||||
.base
|
||||
.join(&path)
|
||||
.map_err(|_| RepositoryError::Address { path: path.clone() })?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
if response.status() == reqwest::StatusCode::NOT_FOUND {
|
||||
return Ok(None);
|
||||
}
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path,
|
||||
});
|
||||
}
|
||||
Ok(Some(response.json::<SecretResponse>().await?.value))
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkflowSources for RepositoryContents {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, RepositoryError> {
|
||||
let mut files = Vec::new();
|
||||
for directory in DIRECTORIES {
|
||||
let listing = self
|
||||
.entries(&format!("{repository}/contents/{directory}?ref={commit}"))
|
||||
.await?;
|
||||
for entry in listing.into_iter().filter(|entry| entry.kind == "file") {
|
||||
let content = match entry.content {
|
||||
Some(content) => content,
|
||||
None => {
|
||||
let mut file = self
|
||||
.entries(&format!(
|
||||
"{repository}/contents/{}?ref={commit}",
|
||||
entry.path
|
||||
))
|
||||
.await?;
|
||||
match file.pop().and_then(|entry| entry.content) {
|
||||
Some(content) => content,
|
||||
None => continue,
|
||||
}
|
||||
}
|
||||
};
|
||||
let content = STANDARD.decode(content.replace(['\n', '\r'], ""))?;
|
||||
files.push(WorkflowFile::new(entry.path, content));
|
||||
}
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for RepositoryContents {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
repository: &str,
|
||||
request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, RepositoryError> {
|
||||
let ChangedFilesRequest::PullRequest(pull_request) = request else {
|
||||
return Err(RepositoryError::UnsupportedChangedFiles);
|
||||
};
|
||||
let url = self
|
||||
.base
|
||||
.join(&format!(
|
||||
"api/v1/repos/{repository}/pulls/{pull_request}/files"
|
||||
))
|
||||
.map_err(|_| RepositoryError::Address {
|
||||
path: repository.to_owned(),
|
||||
})?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("Authorization", format!("token {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path: format!("{repository}/pulls/{pull_request}/files"),
|
||||
});
|
||||
}
|
||||
let files: Vec<ChangedFile> = response.json().await?;
|
||||
Ok(files.into_iter().map(|file| file.filename).collect())
|
||||
}
|
||||
}
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use serde::Deserialize;
|
||||
use syncode_control_node::{RepositoryCoordinateRequestError, RepositoryCoordinates};
|
||||
use thiserror::Error;
|
||||
use url::Url;
|
||||
|
||||
use crate::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};
|
||||
|
||||
/// The workflow directories supported by the repository service.
|
||||
const DIRECTORIES: [&str; 2] = [".gitea/workflows", ".github/workflows"];
|
||||
|
||||
pub struct RepositoryContents {
|
||||
base: Url,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
}
|
||||
|
||||
pub struct RepositorySecrets {
|
||||
base: Url,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
coordinates: RepositoryCoordinates,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RepositoryError {
|
||||
#[error("cannot reach the repository service: {0}")]
|
||||
Unreachable(#[from] reqwest::Error),
|
||||
|
||||
#[error("the repository service answered {status} for {path}")]
|
||||
Refused { status: u16, path: String },
|
||||
|
||||
#[error("the repository service returned a workflow file that is not valid base64: {0}")]
|
||||
Undecodable(#[from] base64::DecodeError),
|
||||
|
||||
#[error("cannot address {path} on the repository service")]
|
||||
Address { path: String },
|
||||
|
||||
#[error("cannot resolve repository coordinates: {0}")]
|
||||
Coordinates(#[from] RepositoryCoordinateRequestError),
|
||||
|
||||
#[error("the forge cannot compare native repository commits")]
|
||||
UnsupportedChangedFiles,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ChangedFile {
|
||||
filename: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct Entry {
|
||||
path: String,
|
||||
#[serde(rename = "type")]
|
||||
kind: String,
|
||||
content: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(untagged)]
|
||||
enum Contents {
|
||||
Many(Vec<Entry>),
|
||||
One(Entry),
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SecretResponse {
|
||||
value: String,
|
||||
}
|
||||
|
||||
impl RepositoryContents {
|
||||
#[must_use]
|
||||
pub fn new(base: Url, token: String) -> Self {
|
||||
Self {
|
||||
base,
|
||||
token,
|
||||
client: reqwest::Client::new(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn entries(&self, path: &str) -> Result<Vec<Entry>, RepositoryError> {
|
||||
let url = self
|
||||
.base
|
||||
.join(&format!("api/v1/repos/{path}"))
|
||||
.map_err(|_| RepositoryError::Address {
|
||||
path: path.to_owned(),
|
||||
})?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("Authorization", format!("token {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
// A repository without one of the directories is not a failure; it is a
|
||||
// repository that keeps its workflows in the other one, or in neither.
|
||||
if response.status() == reqwest::StatusCode::NOT_FOUND {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path: path.to_owned(),
|
||||
});
|
||||
}
|
||||
Ok(match response.json().await? {
|
||||
Contents::Many(entries) => entries,
|
||||
Contents::One(entry) => vec![entry],
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl RepositorySecrets {
|
||||
#[must_use]
|
||||
pub fn new(base: Url, token: String, coordinates: RepositoryCoordinates) -> Self {
|
||||
Self {
|
||||
base,
|
||||
token,
|
||||
client: reqwest::Client::new(),
|
||||
coordinates,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl crate::secrets::SecretSource for RepositorySecrets {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn resolve(
|
||||
&self,
|
||||
repository: &str,
|
||||
name: &str,
|
||||
) -> Result<Option<String>, RepositoryError> {
|
||||
let repository = self.coordinates.resolve(repository).await?;
|
||||
let path = format!("api/internal/actions/syncode/secrets/{repository}/{name}");
|
||||
let url = self
|
||||
.base
|
||||
.join(&path)
|
||||
.map_err(|_| RepositoryError::Address { path: path.clone() })?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
if response.status() == reqwest::StatusCode::NOT_FOUND {
|
||||
return Ok(None);
|
||||
}
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path,
|
||||
});
|
||||
}
|
||||
Ok(Some(response.json::<SecretResponse>().await?.value))
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkflowSources for RepositoryContents {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn workflows(
|
||||
&self,
|
||||
repository: &str,
|
||||
commit: &str,
|
||||
) -> Result<Vec<WorkflowFile>, RepositoryError> {
|
||||
let mut files = Vec::new();
|
||||
for directory in DIRECTORIES {
|
||||
let listing = self
|
||||
.entries(&format!("{repository}/contents/{directory}?ref={commit}"))
|
||||
.await?;
|
||||
for entry in listing.into_iter().filter(|entry| entry.kind == "file") {
|
||||
let content = match entry.content {
|
||||
Some(content) => content,
|
||||
None => {
|
||||
let mut file = self
|
||||
.entries(&format!(
|
||||
"{repository}/contents/{}?ref={commit}",
|
||||
entry.path
|
||||
))
|
||||
.await?;
|
||||
match file.pop().and_then(|entry| entry.content) {
|
||||
Some(content) => content,
|
||||
None => continue,
|
||||
}
|
||||
}
|
||||
};
|
||||
let content = STANDARD.decode(content.replace(['\n', '\r'], ""))?;
|
||||
files.push(WorkflowFile::new(entry.path, content));
|
||||
}
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
}
|
||||
|
||||
impl ChangedFiles for RepositoryContents {
|
||||
type Error = RepositoryError;
|
||||
|
||||
async fn changed(
|
||||
&self,
|
||||
repository: &str,
|
||||
request: ChangedFilesRequest,
|
||||
) -> Result<Vec<String>, RepositoryError> {
|
||||
let ChangedFilesRequest::PullRequest(pull_request) = request else {
|
||||
return Err(RepositoryError::UnsupportedChangedFiles);
|
||||
};
|
||||
let url = self
|
||||
.base
|
||||
.join(&format!(
|
||||
"api/v1/repos/{repository}/pulls/{pull_request}/files"
|
||||
))
|
||||
.map_err(|_| RepositoryError::Address {
|
||||
path: repository.to_owned(),
|
||||
})?;
|
||||
let response = self
|
||||
.client
|
||||
.get(url)
|
||||
.header("Authorization", format!("token {}", self.token))
|
||||
.send()
|
||||
.await?;
|
||||
if !response.status().is_success() {
|
||||
return Err(RepositoryError::Refused {
|
||||
status: response.status().as_u16(),
|
||||
path: format!("{repository}/pulls/{pull_request}/files"),
|
||||
});
|
||||
}
|
||||
let files: Vec<ChangedFile> = response.json().await?;
|
||||
Ok(files.into_iter().map(|file| file.filename).collect())
|
||||
}
|
||||
}
|
||||
+5
-1
@@ -1,389 +1,393 @@
|
||||
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_http::router as actions_read_http_router;
|
||||
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::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, RepositoryCoordinates,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_CORS_ORIGINS", value_delimiter = ',')]
|
||||
control_cors_origins: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SESSION_COOKIE_NAME")]
|
||||
identity_session_cookie_name: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
|
||||
let repository_sources = RepositorySources::new(
|
||||
native_repository,
|
||||
RepositoryContents::new(
|
||||
arguments.repository_source,
|
||||
arguments.repository_source_token,
|
||||
),
|
||||
);
|
||||
let intake = Arc::new(Intake::new(
|
||||
repository_sources,
|
||||
runs.clone(),
|
||||
action_resolver,
|
||||
arguments.event_feed_secret,
|
||||
));
|
||||
let admin_token = arguments
|
||||
.admin_token
|
||||
.filter(|token| !token.is_empty())
|
||||
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
|
||||
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
|
||||
let actions_read = ActionsRead::new(
|
||||
runs.clone(),
|
||||
IdentityActionsAuthorization::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
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(),
|
||||
))
|
||||
.merge(actions_read_http_router(
|
||||
actions_read.clone(),
|
||||
arguments.control_cors_origins,
|
||||
arguments.identity_session_cookie_name,
|
||||
));
|
||||
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 repositories = RepositoryCoordinates::new(
|
||||
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(),
|
||||
);
|
||||
// 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,
|
||||
repositories.clone(),
|
||||
),
|
||||
Mode::Active => NodeSessionServer::new(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
repositories,
|
||||
),
|
||||
};
|
||||
tokio::select! {
|
||||
served = axum::serve(events, http).into_future() => 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_http::router as actions_read_http_router;
|
||||
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::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, RepositoryCoordinates,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_CORS_ORIGINS", value_delimiter = ',')]
|
||||
control_cors_origins: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SESSION_COOKIE_NAME")]
|
||||
identity_session_cookie_name: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
|
||||
let repository_sources = RepositorySources::new(
|
||||
native_repository,
|
||||
RepositoryContents::new(
|
||||
arguments.repository_source,
|
||||
arguments.repository_source_token,
|
||||
),
|
||||
);
|
||||
let intake = Arc::new(Intake::new(
|
||||
repository_sources,
|
||||
runs.clone(),
|
||||
action_resolver,
|
||||
arguments.event_feed_secret,
|
||||
));
|
||||
let admin_token = arguments
|
||||
.admin_token
|
||||
.filter(|token| !token.is_empty())
|
||||
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
|
||||
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
|
||||
let actions_read = ActionsRead::new(
|
||||
runs.clone(),
|
||||
IdentityActionsAuthorization::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
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(),
|
||||
))
|
||||
.merge(actions_read_http_router(
|
||||
actions_read.clone(),
|
||||
arguments.control_cors_origins,
|
||||
arguments.identity_session_cookie_name,
|
||||
));
|
||||
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 repositories = RepositoryCoordinates::new(
|
||||
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,
|
||||
repositories.clone(),
|
||||
),
|
||||
IdentityRepositoryCredentials::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
authority.clone(),
|
||||
);
|
||||
// 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,
|
||||
repositories.clone(),
|
||||
),
|
||||
Mode::Active => NodeSessionServer::new(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
repositories,
|
||||
),
|
||||
};
|
||||
tokio::select! {
|
||||
served = axum::serve(events, http).into_future() => 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;
|
||||
}
|
||||
}
|
||||
@@ -1,90 +1,141 @@
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control_node::RepositoryCoordinates;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
_request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
Err(Status::unimplemented("issue_workflow_repository_token"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_ids_resolve_to_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 client = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
|
||||
assert_eq!(
|
||||
client
|
||||
.resolve("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use axum::Json;
|
||||
use axum::http::{HeaderMap, StatusCode};
|
||||
use axum::routing::get;
|
||||
use serde_json::json;
|
||||
use syncode_control::repository::RepositorySecrets;
|
||||
use syncode_control::secrets::SecretSource;
|
||||
use syncode_control_node::RepositoryCoordinates;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
_request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
Err(Status::unimplemented("issue_workflow_repository_token"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_ids_resolve_to_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 client = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
|
||||
assert_eq!(
|
||||
client
|
||||
.resolve("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_secrets_use_resolved_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 secret_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let secret_endpoint = format!("http://{}/", secret_listener.local_addr()?);
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/secrets/syncode/pipelines-demo/REGISTRY_TOKEN",
|
||||
get(|headers: HeaderMap| async move {
|
||||
if headers
|
||||
.get("x-gitea-internal-auth")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer internal-token")
|
||||
{
|
||||
return Err(StatusCode::UNAUTHORIZED);
|
||||
}
|
||||
Ok(Json(json!({ "value": "registry-token" })))
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(secret_listener, app).await;
|
||||
});
|
||||
|
||||
let coordinates = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
|
||||
let source = RepositorySecrets::new(
|
||||
secret_endpoint.parse()?,
|
||||
"internal-token".to_owned(),
|
||||
coordinates,
|
||||
);
|
||||
assert_eq!(
|
||||
source
|
||||
.resolve("76128383-1df5-4979-9b13-c048a5287e9a", "REGISTRY_TOKEN")
|
||||
.await?,
|
||||
Some("registry-token".to_owned())
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user