fix: read action sources from repository plane #44
+133
-89
@@ -1,125 +1,169 @@
|
||||
use std::io::Read;
|
||||
|
||||
use flate2::read::GzDecoder;
|
||||
use serde::Deserialize;
|
||||
use syncode_workflow::{GitCommit, RepositoryUrl};
|
||||
use url::Url;
|
||||
|
||||
use crate::actions::{ActionRepositoryPort, ActionResolutionError, RepositorySnapshot};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ForgeActionRepository {
|
||||
client: reqwest::Client,
|
||||
token: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct CommitResponse {
|
||||
sha: String,
|
||||
}
|
||||
|
||||
impl ForgeActionRepository {
|
||||
#[must_use]
|
||||
pub fn new(token: String) -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
token,
|
||||
}
|
||||
}
|
||||
|
||||
async fn get(&self, url: Url) -> Result<reqwest::Response, ActionResolutionError> {
|
||||
let response = self
|
||||
.client
|
||||
.get(url.clone())
|
||||
.header("Authorization", format!("token {}", self.token))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?;
|
||||
if !response.status().is_success() {
|
||||
return Err(ActionResolutionError::Repository(format!(
|
||||
"{} returned {}",
|
||||
url,
|
||||
response.status()
|
||||
)));
|
||||
}
|
||||
Ok(response)
|
||||
}
|
||||
}
|
||||
|
||||
impl ActionRepositoryPort for ForgeActionRepository {
|
||||
async fn fetch(
|
||||
&self,
|
||||
repository: RepositoryUrl,
|
||||
revision: String,
|
||||
) -> Result<RepositorySnapshot, ActionResolutionError> {
|
||||
let (mut api, owner, name) = repository_coordinates(&repository)?;
|
||||
{
|
||||
let mut segments = api
|
||||
.path_segments_mut()
|
||||
.map_err(|_| ActionResolutionError::Repository(repository.to_string()))?;
|
||||
segments.extend([
|
||||
"api", "v1", "repos", &owner, &name, "git", "commits", &revision,
|
||||
]);
|
||||
}
|
||||
api.query_pairs_mut()
|
||||
.append_pair("stat", "false")
|
||||
.append_pair("verification", "false")
|
||||
.append_pair("files", "false");
|
||||
let commit = self
|
||||
.get(api)
|
||||
.await?
|
||||
.json::<CommitResponse>()
|
||||
.await
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?
|
||||
.sha
|
||||
.parse::<GitCommit>()
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?;
|
||||
|
||||
let (mut archive_url, owner, name) = repository_coordinates(&repository)?;
|
||||
{
|
||||
let archive = format!("{commit}.tar.gz");
|
||||
let mut segments = archive_url
|
||||
.path_segments_mut()
|
||||
.map_err(|_| ActionResolutionError::Repository(repository.to_string()))?;
|
||||
segments.extend(["api", "v1", "repos", &owner, &name, "archive", &archive]);
|
||||
}
|
||||
let compressed = self
|
||||
.get(archive_url)
|
||||
.await?
|
||||
.bytes()
|
||||
.await
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?;
|
||||
let mut archive = Vec::new();
|
||||
GzDecoder::new(compressed.as_ref())
|
||||
.read_to_end(&mut archive)
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?;
|
||||
Ok(RepositorySnapshot { commit, archive })
|
||||
}
|
||||
}
|
||||
|
||||
fn repository_coordinates(
|
||||
repository: &RepositoryUrl,
|
||||
) -> Result<(Url, String, String), ActionResolutionError> {
|
||||
let url = Url::parse(repository.as_ref())
|
||||
.map_err(|error| ActionResolutionError::Repository(error.to_string()))?;
|
||||
let segments = url
|
||||
.path_segments()
|
||||
.ok_or_else(|| ActionResolutionError::Repository(repository.to_string()))?
|
||||
.filter(|segment| !segment.is_empty())
|
||||
.map(str::to_owned)
|
||||
.collect::<Vec<_>>();
|
||||
let [owner, name] = segments.as_slice() else {
|
||||
return Err(ActionResolutionError::Repository(format!(
|
||||
"{repository} must identify exactly one owner and repository"
|
||||
)));
|
||||
};
|
||||
let mut base = url;
|
||||
base.set_path("/");
|
||||
base.set_query(None);
|
||||
base.set_fragment(None);
|
||||
Ok((
|
||||
base,
|
||||
owner.to_owned(),
|
||||
name.trim_end_matches(".git").to_owned(),
|
||||
))
|
||||
}
|
||||
use syncode_control_node::identity_wire::ResolveRepositoryRequest;
|
||||
use syncode_control_node::identity_wire::identity_client::IdentityClient;
|
||||
use syncode_repository_api_grpc::proto::repository_read_client::RepositoryReadClient;
|
||||
use syncode_repository_api_grpc::proto::{
|
||||
ArchiveFormat, ArchiveRequest, GetCommitRequest, ObjectId, ResolveRevisionRequest,
|
||||
};
|
||||
use syncode_workflow::{GitCommit, RepositoryUrl};
|
||||
use tonic::Request;
|
||||
use tonic::metadata::{Ascii, MetadataValue};
|
||||
use tonic::transport::Channel;
|
||||
use url::Url;
|
||||
|
||||
use crate::actions::{ActionRepositoryPort, ActionResolutionError, RepositorySnapshot};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct NativeActionRepository {
|
||||
repositories: RepositoryReadClient<Channel>,
|
||||
identity: IdentityClient<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
impl NativeActionRepository {
|
||||
pub async fn connect(
|
||||
repository_endpoint: String,
|
||||
identity_endpoint: String,
|
||||
identity_shared_secret: String,
|
||||
) -> Result<Self, ActionResolutionError> {
|
||||
let repositories = RepositoryReadClient::connect(repository_endpoint)
|
||||
.await
|
||||
.map_err(repository_error)?;
|
||||
let identity = IdentityClient::connect(identity_endpoint)
|
||||
.await
|
||||
.map_err(repository_error)?;
|
||||
let authorization = format!("Bearer {identity_shared_secret}")
|
||||
.parse()
|
||||
.map_err(repository_error)?;
|
||||
Ok(Self {
|
||||
repositories,
|
||||
identity,
|
||||
authorization,
|
||||
})
|
||||
}
|
||||
|
||||
async fn repository_id(
|
||||
&self,
|
||||
repository: &RepositoryUrl,
|
||||
) -> Result<String, ActionResolutionError> {
|
||||
match repository_address(repository)? {
|
||||
RepositoryAddress::Id(id) => Ok(id),
|
||||
RepositoryAddress::Coordinates { owner, name } => {
|
||||
let mut request = Request::new(ResolveRepositoryRequest { owner, name });
|
||||
request
|
||||
.metadata_mut()
|
||||
.insert("authorization", self.authorization.clone());
|
||||
Ok(self
|
||||
.identity
|
||||
.clone()
|
||||
.resolve_repository(request)
|
||||
.await
|
||||
.map_err(repository_error)?
|
||||
.into_inner()
|
||||
.repository_id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn archive(
|
||||
&self,
|
||||
repository_id: String,
|
||||
tree: ObjectId,
|
||||
) -> Result<Vec<u8>, ActionResolutionError> {
|
||||
let mut stream = self
|
||||
.repositories
|
||||
.clone()
|
||||
.archive(ArchiveRequest {
|
||||
repository_id,
|
||||
tree: Some(tree),
|
||||
format: ArchiveFormat::Tar.into(),
|
||||
prefix: "source/".to_owned(),
|
||||
})
|
||||
.await
|
||||
.map_err(repository_error)?
|
||||
.into_inner();
|
||||
let mut archive = Vec::new();
|
||||
while let Some(chunk) = stream.message().await.map_err(repository_error)? {
|
||||
archive.extend(chunk.data);
|
||||
}
|
||||
Ok(archive)
|
||||
}
|
||||
}
|
||||
|
||||
impl ActionRepositoryPort for NativeActionRepository {
|
||||
async fn fetch(
|
||||
&self,
|
||||
repository: RepositoryUrl,
|
||||
revision: String,
|
||||
) -> Result<RepositorySnapshot, ActionResolutionError> {
|
||||
let repository_id = self.repository_id(&repository).await?;
|
||||
let resolved = self
|
||||
.repositories
|
||||
.clone()
|
||||
.resolve_revision(ResolveRevisionRequest {
|
||||
repository_id: repository_id.clone(),
|
||||
revision,
|
||||
})
|
||||
.await
|
||||
.map_err(repository_error)?
|
||||
.into_inner();
|
||||
let object = resolved.peeled.ok_or_else(|| {
|
||||
ActionResolutionError::Repository(
|
||||
"repository service omitted the peeled revision".to_owned(),
|
||||
)
|
||||
})?;
|
||||
let commit = self
|
||||
.repositories
|
||||
.clone()
|
||||
.get_commit(GetCommitRequest {
|
||||
repository_id: repository_id.clone(),
|
||||
object_id: Some(object),
|
||||
})
|
||||
.await
|
||||
.map_err(repository_error)?
|
||||
.into_inner();
|
||||
let commit_id = commit.id.ok_or_else(|| {
|
||||
ActionResolutionError::Repository("repository service omitted the commit id".to_owned())
|
||||
})?;
|
||||
let tree = commit.tree.ok_or_else(|| {
|
||||
ActionResolutionError::Repository(
|
||||
"repository service omitted the commit tree".to_owned(),
|
||||
)
|
||||
})?;
|
||||
let commit = commit_id
|
||||
.hex
|
||||
.parse::<GitCommit>()
|
||||
.map_err(repository_error)?;
|
||||
let archive = self.archive(repository_id, tree).await?;
|
||||
Ok(RepositorySnapshot { commit, archive })
|
||||
}
|
||||
}
|
||||
|
||||
enum RepositoryAddress {
|
||||
Id(String),
|
||||
Coordinates { owner: String, name: String },
|
||||
}
|
||||
|
||||
fn repository_address(
|
||||
repository: &RepositoryUrl,
|
||||
) -> Result<RepositoryAddress, ActionResolutionError> {
|
||||
let url = Url::parse(repository.as_ref()).map_err(repository_error)?;
|
||||
let segments = url
|
||||
.path_segments()
|
||||
.ok_or_else(|| ActionResolutionError::Repository(repository.to_string()))?
|
||||
.filter(|segment| !segment.is_empty())
|
||||
.collect::<Vec<_>>();
|
||||
match segments.as_slice() {
|
||||
[id] if uuid::Uuid::parse_str(id).is_ok() => Ok(RepositoryAddress::Id((*id).to_owned())),
|
||||
[owner, name] => Ok(RepositoryAddress::Coordinates {
|
||||
owner: (*owner).to_owned(),
|
||||
name: name.trim_end_matches(".git").to_owned(),
|
||||
}),
|
||||
_ => Err(ActionResolutionError::Repository(format!(
|
||||
"{repository} must identify one repository id or owner and repository"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
fn repository_error(error: impl std::fmt::Display) -> ActionResolutionError {
|
||||
ActionResolutionError::Repository(error.to_string())
|
||||
}
|
||||
+10
-3
@@ -1,374 +1,381 @@
|
||||
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::ForgeActionRepository;
|
||||
use syncode_control::action_store::FileActionStore;
|
||||
use syncode_control::actions::ActionResolver;
|
||||
use syncode_control::actions_read::ActionsRead;
|
||||
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
|
||||
use syncode_control::admin::{Admin, router as admin_router};
|
||||
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
|
||||
use syncode_control::checks::Checks;
|
||||
use syncode_control::maintenance;
|
||||
use syncode_control::projection;
|
||||
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
|
||||
use syncode_control::repository_grpc::NativeRepositoryContents;
|
||||
use syncode_control::repository_sources::RepositorySources;
|
||||
use syncode_control::secrets::RuntimeSecrets;
|
||||
use syncode_control::token::EnrolmentScope;
|
||||
use syncode_control::webhook::{Intake, router};
|
||||
use syncode_control_node::{
|
||||
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
|
||||
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_URL")]
|
||||
projection_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
|
||||
projection_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
|
||||
projection_min_run_number: u64,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
projection_min_run_number: arguments.projection_min_run_number,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_resolver = ActionResolver::new(
|
||||
ForgeActionRepository::new(arguments.repository_source_token.clone()),
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).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)?;
|
||||
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_grpc::NativeRepositoryContents;
|
||||
use syncode_control::repository_sources::RepositorySources;
|
||||
use syncode_control::secrets::RuntimeSecrets;
|
||||
use syncode_control::token::EnrolmentScope;
|
||||
use syncode_control::webhook::{Intake, router};
|
||||
use syncode_control_node::{
|
||||
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
|
||||
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient,
|
||||
};
|
||||
use syncode_control_nodes::{Nodes, Scope};
|
||||
use syncode_control_runs::{
|
||||
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
|
||||
};
|
||||
use syncode_control_store::Postgres;
|
||||
use tokio::net::TcpListener;
|
||||
use tonic::transport::Server;
|
||||
use url::Url;
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
#[command(name = "syncode-control", version, about)]
|
||||
struct Arguments {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Where nodes open their session.
|
||||
#[arg(long, default_value = "127.0.0.1:8090")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Where the native repository event feed delivers events.
|
||||
#[arg(long, default_value = "127.0.0.1:8091")]
|
||||
listen_events: SocketAddr,
|
||||
|
||||
/// The repository service this control plane reads workflows from.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
|
||||
repository_source: Url,
|
||||
|
||||
/// A token allowed to read repository contents.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
|
||||
repository_source_token: String,
|
||||
|
||||
/// Where native repositories are read over gRPC.
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
|
||||
repository_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
|
||||
identity_grpc: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
|
||||
identity_shared_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
|
||||
action_mirror_url: Url,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_ALLOWLIST",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_allowlist: Vec<Url>,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ACTION_OCI_REGISTRIES",
|
||||
value_delimiter = ',',
|
||||
num_args = 1..
|
||||
)]
|
||||
action_oci_registries: Vec<String>,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_STORE")]
|
||||
action_store: PathBuf,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
|
||||
action_artifact_public_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
|
||||
secret_source: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
|
||||
secret_source_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
|
||||
capability_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
|
||||
artifact_signing_key: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_URL")]
|
||||
projection_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
|
||||
projection_token: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
|
||||
projection_min_run_number: u64,
|
||||
|
||||
/// The secret used to sign the native event feed.
|
||||
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
|
||||
event_feed_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
|
||||
collaboration_event_url: Url,
|
||||
|
||||
#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
|
||||
collaboration_event_secret: String,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
|
||||
source_node_id: uuid::Uuid,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
|
||||
default_value_t = 1
|
||||
)]
|
||||
check_event_interval_seconds: u64,
|
||||
|
||||
#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
|
||||
check_event_batch: i64,
|
||||
|
||||
/// Bearer token protecting the operational API.
|
||||
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
|
||||
admin_token: Option<String>,
|
||||
|
||||
/// Where the run log is kept.
|
||||
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
|
||||
database: String,
|
||||
|
||||
#[arg(long, default_value_t = 8)]
|
||||
database_connections: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
|
||||
organization_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
|
||||
repository_concurrency: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
|
||||
principal_concurrency: u32,
|
||||
|
||||
#[arg(
|
||||
long,
|
||||
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
|
||||
default_value_t = 10_000
|
||||
)]
|
||||
organization_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
|
||||
principal_queue_quota: u32,
|
||||
|
||||
#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
|
||||
audit_retention_days: u32,
|
||||
|
||||
/// Whether compiled shadow runs may be assigned to nodes.
|
||||
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
|
||||
mode: Mode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
|
||||
enum Mode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
enum Command {
|
||||
/// Issue an enrolment token a node can spend for an identity.
|
||||
///
|
||||
/// The token is stored before it is printed, so one that reaches an
|
||||
/// operator is one the control plane will honour.
|
||||
IssueToken {
|
||||
/// How far the token reaches: `instance`, `organisation:<name>` or
|
||||
/// `repository:<owner>/<name>`.
|
||||
#[arg(long, default_value = "instance")]
|
||||
scope: String,
|
||||
},
|
||||
/// Revoke a node identity immediately.
|
||||
RevokeNode {
|
||||
/// The node UUID printed at enrolment.
|
||||
node: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn Error>> {
|
||||
let arguments = Arguments::parse();
|
||||
// The log is read before anything is served: a node that reconnects has to
|
||||
// meet the run it was holding, not a control plane that has forgotten it.
|
||||
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
|
||||
let policy = SchedulerPolicy::new(
|
||||
arguments.organization_concurrency,
|
||||
arguments.repository_concurrency,
|
||||
arguments.principal_concurrency,
|
||||
arguments.organization_queue_quota,
|
||||
arguments.principal_queue_quota,
|
||||
);
|
||||
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
|
||||
let nodes = Nodes::restored(store.clone()).await?;
|
||||
|
||||
if let Some(command) = arguments.command {
|
||||
match command {
|
||||
Command::IssueToken { scope } => {
|
||||
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
|
||||
let token = nodes.issue_token(scope).await?;
|
||||
println!("{}", token.secret().expose());
|
||||
}
|
||||
Command::RevokeNode { node } => {
|
||||
nodes.revoke(node.parse::<NodeId>()?).await?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store
|
||||
.record_audit(AuditEvent::new("control.configuration", "applied").after(
|
||||
AuditPayload::Configuration(AuditConfiguration {
|
||||
mode: match arguments.mode {
|
||||
Mode::Shadow => AuditControlMode::Shadow,
|
||||
Mode::Active => AuditControlMode::Active,
|
||||
},
|
||||
organization_concurrency: arguments.organization_concurrency,
|
||||
repository_concurrency: arguments.repository_concurrency,
|
||||
principal_concurrency: arguments.principal_concurrency,
|
||||
organization_queue_quota: arguments.organization_queue_quota,
|
||||
principal_queue_quota: arguments.principal_queue_quota,
|
||||
audit_retention_days: arguments.audit_retention_days,
|
||||
projection_min_run_number: arguments.projection_min_run_number,
|
||||
}),
|
||||
))
|
||||
.await?;
|
||||
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
|
||||
tokio::spawn(maintenance::retain_audit(
|
||||
store.clone(),
|
||||
arguments.audit_retention_days,
|
||||
));
|
||||
|
||||
let action_mirror = arguments
|
||||
.action_mirror_url
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()?;
|
||||
let action_allowlist = arguments
|
||||
.action_allowlist
|
||||
.into_iter()
|
||||
.map(|repository| {
|
||||
repository
|
||||
.to_string()
|
||||
.parse::<syncode_workflow::RepositoryUrl>()
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
let action_store = FileActionStore::new(arguments.action_store);
|
||||
let action_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
|
||||
let repository_sources = RepositorySources::new(
|
||||
native_repository,
|
||||
RepositoryContents::new(
|
||||
arguments.repository_source,
|
||||
arguments.repository_source_token,
|
||||
),
|
||||
);
|
||||
let intake = Arc::new(Intake::new(
|
||||
repository_sources,
|
||||
runs.clone(),
|
||||
action_resolver,
|
||||
arguments.event_feed_secret,
|
||||
));
|
||||
let admin_token = arguments
|
||||
.admin_token
|
||||
.filter(|token| !token.is_empty())
|
||||
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
|
||||
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
|
||||
let http = router(intake)
|
||||
.merge(admin_router(Arc::new(
|
||||
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
|
||||
)))
|
||||
.merge(action_delivery_router(
|
||||
action_store,
|
||||
artifact_authority.clone(),
|
||||
runs.clone(),
|
||||
));
|
||||
let events = TcpListener::bind(arguments.listen_events).await?;
|
||||
let check_events = CheckEventPublisher::new(
|
||||
store.clone(),
|
||||
CheckEventPublisherConfig {
|
||||
endpoint: arguments.collaboration_event_url.as_str(),
|
||||
secret: arguments.collaboration_event_secret,
|
||||
source_node_id: arguments.source_node_id,
|
||||
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
|
||||
batch: arguments.check_event_batch,
|
||||
},
|
||||
)?;
|
||||
|
||||
let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
|
||||
let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?;
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -1,215 +1,240 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use syncode_repository_api_grpc::proto::repository_read_server::{
|
||||
RepositoryRead, RepositoryReadServer,
|
||||
};
|
||||
use syncode_repository_api_grpc::proto::*;
|
||||
use tokio_stream::Stream;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
pub const REPOSITORY: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89210";
|
||||
pub const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c";
|
||||
pub const TAG_OBJECT: &str = "8888888888888888888888888888888888888888";
|
||||
const ROOT: &str = "1111111111111111111111111111111111111111";
|
||||
const GITEA: &str = "2222222222222222222222222222222222222222";
|
||||
const WORKFLOWS: &str = "3333333333333333333333333333333333333333";
|
||||
const CI: &str = "4444444444444444444444444444444444444444";
|
||||
const OTHER: &str = "5555555555555555555555555555555555555555";
|
||||
const SRC: &str = "6666666666666666666666666666666666666666";
|
||||
const MAIN: &str = "7777777777777777777777777777777777777777";
|
||||
|
||||
pub const WORKFLOW: &str = r#"
|
||||
name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
tags: ["v*"]
|
||||
pull_request:
|
||||
branches: [main]
|
||||
paths: ["src/**"]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: [self-hosted, linux]
|
||||
steps:
|
||||
- run: echo native
|
||||
"#;
|
||||
|
||||
const OTHER_WORKFLOW: &str = r#"
|
||||
name: Other
|
||||
on: [workflow_dispatch]
|
||||
jobs:
|
||||
other:
|
||||
runs-on: [self-hosted, linux]
|
||||
steps:
|
||||
- run: echo other
|
||||
"#;
|
||||
|
||||
#[derive(Default)]
|
||||
struct FixtureRepository;
|
||||
|
||||
pub fn service() -> RepositoryReadServer<impl RepositoryRead> {
|
||||
RepositoryReadServer::new(FixtureRepository)
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl RepositoryRead for FixtureRepository {
|
||||
async fn get_commit(
|
||||
&self,
|
||||
_request: Request<GetCommitRequest>,
|
||||
) -> Result<Response<Commit>, Status> {
|
||||
Ok(Response::new(Commit {
|
||||
id: Some(object(COMMIT)),
|
||||
tree: Some(object(ROOT)),
|
||||
parents: Vec::new(),
|
||||
message: Vec::new(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_tree(&self, request: Request<ListTreeRequest>) -> Result<Response<Tree>, Status> {
|
||||
let tree = request
|
||||
.into_inner()
|
||||
.object_id
|
||||
.ok_or_else(|| Status::invalid_argument("missing tree"))?;
|
||||
let entries = match tree.hex.as_str() {
|
||||
ROOT => vec![
|
||||
entry(b".gitea", GITEA, TreeEntryKind::Tree),
|
||||
entry(b"src", SRC, TreeEntryKind::Tree),
|
||||
],
|
||||
GITEA => vec![entry(b"workflows", WORKFLOWS, TreeEntryKind::Tree)],
|
||||
WORKFLOWS => vec![
|
||||
entry(b"ci.yml", CI, TreeEntryKind::Blob),
|
||||
entry(b"other.yml", OTHER, TreeEntryKind::Blob),
|
||||
],
|
||||
SRC => vec![entry(b"main.rs", MAIN, TreeEntryKind::Blob)],
|
||||
_ => return Err(Status::not_found("tree")),
|
||||
};
|
||||
Ok(Response::new(Tree { entries }))
|
||||
}
|
||||
|
||||
type GetBlobStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
|
||||
|
||||
async fn get_blob(
|
||||
&self,
|
||||
request: Request<GetBlobRequest>,
|
||||
) -> Result<Response<Self::GetBlobStream>, Status> {
|
||||
let object = request
|
||||
.into_inner()
|
||||
.object_id
|
||||
.ok_or_else(|| Status::invalid_argument("missing blob"))?;
|
||||
let content = match object.hex.as_str() {
|
||||
CI => WORKFLOW,
|
||||
OTHER => OTHER_WORKFLOW,
|
||||
_ => return Err(Status::not_found("blob")),
|
||||
};
|
||||
Ok(Response::new(Box::pin(tokio_stream::iter([Ok(
|
||||
BlobChunk {
|
||||
data: content.as_bytes().to_vec(),
|
||||
},
|
||||
)]))))
|
||||
}
|
||||
|
||||
async fn diff(&self, _request: Request<DiffRequest>) -> Result<Response<DiffResult>, Status> {
|
||||
Ok(Response::new(DiffResult {
|
||||
files: vec![DiffFile {
|
||||
old_path: b"src/main.rs".to_vec(),
|
||||
new_path: b"src/main.rs".to_vec(),
|
||||
kind: DiffChangeKind::Modified.into(),
|
||||
..Default::default()
|
||||
}],
|
||||
}))
|
||||
}
|
||||
|
||||
type ArchiveStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
|
||||
|
||||
async fn resolve_revision(
|
||||
&self,
|
||||
request: Request<ResolveRevisionRequest>,
|
||||
) -> Result<Response<ResolvedRevision>, Status> {
|
||||
let revision = request.into_inner().revision;
|
||||
let target = object(&revision);
|
||||
Ok(Response::new(ResolvedRevision {
|
||||
target: Some(target),
|
||||
peeled: Some(object(COMMIT)),
|
||||
kind: ObjectKind::Commit.into(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_commits(
|
||||
&self,
|
||||
_request: Request<ListCommitsRequest>,
|
||||
) -> Result<Response<CommitList>, Status> {
|
||||
Err(Status::unimplemented("list_commits"))
|
||||
}
|
||||
|
||||
async fn merge_base(
|
||||
&self,
|
||||
_request: Request<MergeBaseRequest>,
|
||||
) -> Result<Response<MergeBaseResult>, Status> {
|
||||
Err(Status::unimplemented("merge_base"))
|
||||
}
|
||||
|
||||
async fn blame(
|
||||
&self,
|
||||
_request: Request<BlameRequest>,
|
||||
) -> Result<Response<BlameResult>, Status> {
|
||||
Err(Status::unimplemented("blame"))
|
||||
}
|
||||
|
||||
async fn language_statistics(
|
||||
&self,
|
||||
_request: Request<LanguageStatisticsRequest>,
|
||||
) -> Result<Response<LanguageStatisticsResult>, Status> {
|
||||
Err(Status::unimplemented("language_statistics"))
|
||||
}
|
||||
|
||||
async fn activity(
|
||||
&self,
|
||||
_request: Request<ActivityRequest>,
|
||||
) -> Result<Response<ActivityResult>, Status> {
|
||||
Err(Status::unimplemented("activity"))
|
||||
}
|
||||
|
||||
async fn verify_commit_signature(
|
||||
&self,
|
||||
_request: Request<VerifyCommitSignatureRequest>,
|
||||
) -> Result<Response<CommitSignature>, Status> {
|
||||
Err(Status::unimplemented("verify_commit_signature"))
|
||||
}
|
||||
|
||||
async fn list_lfs_pointers(
|
||||
&self,
|
||||
_request: Request<ListLfsPointersRequest>,
|
||||
) -> Result<Response<LfsPointerList>, Status> {
|
||||
Err(Status::unimplemented("list_lfs_pointers"))
|
||||
}
|
||||
|
||||
async fn search_code(
|
||||
&self,
|
||||
_request: Request<SearchCodeRequest>,
|
||||
) -> Result<Response<CodeSearchResult>, Status> {
|
||||
Err(Status::unimplemented("search_code"))
|
||||
}
|
||||
|
||||
async fn archive(
|
||||
&self,
|
||||
_request: Request<ArchiveRequest>,
|
||||
) -> Result<Response<Self::ArchiveStream>, Status> {
|
||||
Err(Status::unimplemented("archive"))
|
||||
}
|
||||
}
|
||||
|
||||
fn object(hex: &str) -> ObjectId {
|
||||
ObjectId {
|
||||
format: ObjectFormat::Sha1.into(),
|
||||
hex: hex.to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn entry(name: &[u8], id: &str, kind: TreeEntryKind) -> TreeEntry {
|
||||
TreeEntry {
|
||||
name: name.to_vec(),
|
||||
id: Some(object(id)),
|
||||
kind: kind.into(),
|
||||
mode: 0,
|
||||
}
|
||||
}
|
||||
use std::pin::Pin;
|
||||
|
||||
use syncode_repository_api_grpc::proto::repository_read_server::{
|
||||
RepositoryRead, RepositoryReadServer,
|
||||
};
|
||||
use syncode_repository_api_grpc::proto::*;
|
||||
use tokio_stream::Stream;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
pub const REPOSITORY: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89210";
|
||||
pub const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c";
|
||||
pub const TAG_OBJECT: &str = "8888888888888888888888888888888888888888";
|
||||
pub const ROOT: &str = "1111111111111111111111111111111111111111";
|
||||
const GITEA: &str = "2222222222222222222222222222222222222222";
|
||||
const WORKFLOWS: &str = "3333333333333333333333333333333333333333";
|
||||
const CI: &str = "4444444444444444444444444444444444444444";
|
||||
const OTHER: &str = "5555555555555555555555555555555555555555";
|
||||
const SRC: &str = "6666666666666666666666666666666666666666";
|
||||
const MAIN: &str = "7777777777777777777777777777777777777777";
|
||||
|
||||
pub const WORKFLOW: &str = r#"
|
||||
name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
tags: ["v*"]
|
||||
pull_request:
|
||||
branches: [main]
|
||||
paths: ["src/**"]
|
||||
jobs:
|
||||
build:
|
||||
runs-on: [self-hosted, linux]
|
||||
steps:
|
||||
- run: echo native
|
||||
"#;
|
||||
|
||||
const OTHER_WORKFLOW: &str = r#"
|
||||
name: Other
|
||||
on: [workflow_dispatch]
|
||||
jobs:
|
||||
other:
|
||||
runs-on: [self-hosted, linux]
|
||||
steps:
|
||||
- run: echo other
|
||||
"#;
|
||||
|
||||
#[derive(Default)]
|
||||
struct FixtureRepository;
|
||||
|
||||
pub fn service() -> RepositoryReadServer<impl RepositoryRead> {
|
||||
RepositoryReadServer::new(FixtureRepository)
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl RepositoryRead for FixtureRepository {
|
||||
async fn get_commit(
|
||||
&self,
|
||||
_request: Request<GetCommitRequest>,
|
||||
) -> Result<Response<Commit>, Status> {
|
||||
Ok(Response::new(Commit {
|
||||
id: Some(object(COMMIT)),
|
||||
tree: Some(object(ROOT)),
|
||||
parents: Vec::new(),
|
||||
message: Vec::new(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_tree(&self, request: Request<ListTreeRequest>) -> Result<Response<Tree>, Status> {
|
||||
let tree = request
|
||||
.into_inner()
|
||||
.object_id
|
||||
.ok_or_else(|| Status::invalid_argument("missing tree"))?;
|
||||
let entries = match tree.hex.as_str() {
|
||||
ROOT => vec![
|
||||
entry(b".gitea", GITEA, TreeEntryKind::Tree),
|
||||
entry(b"src", SRC, TreeEntryKind::Tree),
|
||||
],
|
||||
GITEA => vec![entry(b"workflows", WORKFLOWS, TreeEntryKind::Tree)],
|
||||
WORKFLOWS => vec![
|
||||
entry(b"ci.yml", CI, TreeEntryKind::Blob),
|
||||
entry(b"other.yml", OTHER, TreeEntryKind::Blob),
|
||||
],
|
||||
SRC => vec![entry(b"main.rs", MAIN, TreeEntryKind::Blob)],
|
||||
_ => return Err(Status::not_found("tree")),
|
||||
};
|
||||
Ok(Response::new(Tree { entries }))
|
||||
}
|
||||
|
||||
type GetBlobStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
|
||||
|
||||
async fn get_blob(
|
||||
&self,
|
||||
request: Request<GetBlobRequest>,
|
||||
) -> Result<Response<Self::GetBlobStream>, Status> {
|
||||
let object = request
|
||||
.into_inner()
|
||||
.object_id
|
||||
.ok_or_else(|| Status::invalid_argument("missing blob"))?;
|
||||
let content = match object.hex.as_str() {
|
||||
CI => WORKFLOW,
|
||||
OTHER => OTHER_WORKFLOW,
|
||||
_ => return Err(Status::not_found("blob")),
|
||||
};
|
||||
Ok(Response::new(Box::pin(tokio_stream::iter([Ok(
|
||||
BlobChunk {
|
||||
data: content.as_bytes().to_vec(),
|
||||
},
|
||||
)]))))
|
||||
}
|
||||
|
||||
async fn diff(&self, _request: Request<DiffRequest>) -> Result<Response<DiffResult>, Status> {
|
||||
Ok(Response::new(DiffResult {
|
||||
files: vec![DiffFile {
|
||||
old_path: b"src/main.rs".to_vec(),
|
||||
new_path: b"src/main.rs".to_vec(),
|
||||
kind: DiffChangeKind::Modified.into(),
|
||||
..Default::default()
|
||||
}],
|
||||
}))
|
||||
}
|
||||
|
||||
type ArchiveStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
|
||||
|
||||
async fn resolve_revision(
|
||||
&self,
|
||||
request: Request<ResolveRevisionRequest>,
|
||||
) -> Result<Response<ResolvedRevision>, Status> {
|
||||
let revision = request.into_inner().revision;
|
||||
let target = object(&revision);
|
||||
Ok(Response::new(ResolvedRevision {
|
||||
target: Some(target),
|
||||
peeled: Some(object(COMMIT)),
|
||||
kind: ObjectKind::Commit.into(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_commits(
|
||||
&self,
|
||||
_request: Request<ListCommitsRequest>,
|
||||
) -> Result<Response<CommitList>, Status> {
|
||||
Err(Status::unimplemented("list_commits"))
|
||||
}
|
||||
|
||||
async fn merge_base(
|
||||
&self,
|
||||
_request: Request<MergeBaseRequest>,
|
||||
) -> Result<Response<MergeBaseResult>, Status> {
|
||||
Err(Status::unimplemented("merge_base"))
|
||||
}
|
||||
|
||||
async fn blame(
|
||||
&self,
|
||||
_request: Request<BlameRequest>,
|
||||
) -> Result<Response<BlameResult>, Status> {
|
||||
Err(Status::unimplemented("blame"))
|
||||
}
|
||||
|
||||
async fn language_statistics(
|
||||
&self,
|
||||
_request: Request<LanguageStatisticsRequest>,
|
||||
) -> Result<Response<LanguageStatisticsResult>, Status> {
|
||||
Err(Status::unimplemented("language_statistics"))
|
||||
}
|
||||
|
||||
async fn activity(
|
||||
&self,
|
||||
_request: Request<ActivityRequest>,
|
||||
) -> Result<Response<ActivityResult>, Status> {
|
||||
Err(Status::unimplemented("activity"))
|
||||
}
|
||||
|
||||
async fn verify_commit_signature(
|
||||
&self,
|
||||
_request: Request<VerifyCommitSignatureRequest>,
|
||||
) -> Result<Response<CommitSignature>, Status> {
|
||||
Err(Status::unimplemented("verify_commit_signature"))
|
||||
}
|
||||
|
||||
async fn list_lfs_pointers(
|
||||
&self,
|
||||
_request: Request<ListLfsPointersRequest>,
|
||||
) -> Result<Response<LfsPointerList>, Status> {
|
||||
Err(Status::unimplemented("list_lfs_pointers"))
|
||||
}
|
||||
|
||||
async fn search_code(
|
||||
&self,
|
||||
_request: Request<SearchCodeRequest>,
|
||||
) -> Result<Response<CodeSearchResult>, Status> {
|
||||
Err(Status::unimplemented("search_code"))
|
||||
}
|
||||
|
||||
async fn archive(
|
||||
&self,
|
||||
request: Request<ArchiveRequest>,
|
||||
) -> Result<Response<Self::ArchiveStream>, Status> {
|
||||
let request = request.into_inner();
|
||||
if request.repository_id != REPOSITORY
|
||||
|| request.tree.as_ref().map(|tree| tree.hex.as_str()) != Some(ROOT)
|
||||
|| request.format != i32::from(ArchiveFormat::Tar)
|
||||
|| request.prefix != "source/"
|
||||
{
|
||||
return Err(Status::invalid_argument("unexpected archive request"));
|
||||
}
|
||||
let content = b"name: fixture\nruns:\n using: node20\n main: index.js\n";
|
||||
let mut archive = tar::Builder::new(Vec::new());
|
||||
let mut header = tar::Header::new_gnu();
|
||||
header
|
||||
.set_path("source/action.yml")
|
||||
.map_err(|error| Status::internal(error.to_string()))?;
|
||||
header.set_size(content.len() as u64);
|
||||
header.set_mode(0o644);
|
||||
header.set_cksum();
|
||||
archive
|
||||
.append(&header, content.as_slice())
|
||||
.map_err(|error| Status::internal(error.to_string()))?;
|
||||
let bytes = archive
|
||||
.into_inner()
|
||||
.map_err(|error| Status::internal(error.to_string()))?;
|
||||
Ok(Response::new(Box::pin(tokio_stream::iter([Ok(
|
||||
BlobChunk { data: bytes },
|
||||
)]))))
|
||||
}
|
||||
}
|
||||
|
||||
fn object(hex: &str) -> ObjectId {
|
||||
ObjectId {
|
||||
format: ObjectFormat::Sha1.into(),
|
||||
hex: hex.to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn entry(name: &[u8], id: &str, kind: TreeEntryKind) -> TreeEntry {
|
||||
TreeEntry {
|
||||
name: name.to_vec(),
|
||||
id: Some(object(id)),
|
||||
kind: kind.into(),
|
||||
mode: 0,
|
||||
}
|
||||
}
|
||||
@@ -1,0 +1,121 @@
|
||||
#![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, ResolveRepositoryRequest,
|
||||
ResolveRepositoryResponse, ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
#[derive(Default)]
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
let request = request.into_inner();
|
||||
if request.owner != "actions" || request.name != "checkout" {
|
||||
return Err(Status::not_found("repository"));
|
||||
}
|
||||
Ok(Response::new(ResolveRepositoryResponse {
|
||||
repository_id: repository::REPOSITORY.to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
async fn source() -> TestResult<NativeActionRepository> {
|
||||
let repository_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let repository_endpoint = format!("http://{}", repository_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(repository::service())
|
||||
.serve_with_incoming(TcpListenerStream::new(repository_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
Ok(NativeActionRepository::connect(
|
||||
repository_endpoint,
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)
|
||||
.await?)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resolves_coordinates_and_fetches_an_immutable_native_archive() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
"https://dev.syncode.sh/actions/checkout".parse()?,
|
||||
"v4".to_owned(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
let paths = tar::Archive::new(snapshot.archive.as_slice())
|
||||
.entries()?
|
||||
.map(|entry| entry.and_then(|entry| entry.path().map(|path| path.into_owned())))
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
assert_eq!(paths, [std::path::PathBuf::from("source/action.yml")]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn accepts_a_native_repository_id_without_coordinate_resolution() -> TestResult {
|
||||
let source = source().await?;
|
||||
let snapshot = source
|
||||
.fetch(
|
||||
format!("https://dev.syncode.sh/{}", repository::REPOSITORY).parse()?,
|
||||
repository::COMMIT.to_owned(),
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(snapshot.commit.as_ref(), repository::COMMIT);
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user