diff --git a/tests/projection_client.rs b/tests/projection_client.rs --- a/tests/projection_client.rs +++ b/tests/projection_client.rs @@ -1,124 +1,125 @@ -#![allow(clippy::expect_used)] - -use std::error::Error; - -use axum::Json; -use axum::http::StatusCode; -use axum::routing::post; -use syncode_control_node::ProjectionClient; -use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer}; -use syncode_control_node::identity_wire::{ - CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest, - GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse, - ValidateSessionRequest, ValidateSessionResponse, -}; -use syncode_control_runs::{Forgotten, JobId, Origin, Runs}; -use tokio::net::TcpListener; -use tokio::sync::mpsc; -use tokio_stream::wrappers::TcpListenerStream; -use tonic::transport::Server; -use tonic::{Request, Response, Status}; - -type TestResult = Result>; - -struct FixtureIdentity; - -#[tonic::async_trait] -impl Identity for FixtureIdentity { - async fn validate_session( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("validate_session")) - } - - async fn check_capability( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("check_capability")) - } - - async fn resolve_repository( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("resolve_repository")) - } - - async fn get_repository_coordinates( - &self, - request: Request, - ) -> Result, Status> { - if request - .metadata() - .get("authorization") - .and_then(|value| value.to_str().ok()) - != Some("Bearer shared-secret") - { - return Err(Status::unauthenticated("missing authorization")); - } - Ok(Response::new(GetRepositoryCoordinatesResponse { - owner: "syncode".to_owned(), - name: "pipelines-demo".to_owned(), - })) - } -} - -#[tokio::test] -async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult { - let identity_listener = TcpListener::bind("127.0.0.1:0").await?; - let identity_endpoint = format!("http://{}", identity_listener.local_addr()?); - tokio::spawn(async move { - let _ = Server::builder() - .add_service(IdentityServer::new(FixtureIdentity)) - .serve_with_incoming(TcpListenerStream::new(identity_listener)) - .await; - }); - - let projection_listener = TcpListener::bind("127.0.0.1:0").await?; - let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?; - let (sent, mut received) = mpsc::unbounded_channel(); - let app = axum::Router::new().route( - "/api/internal/actions/syncode/projection", - post(move |Json(body): Json| { - let sent = sent.clone(); - async move { - let _ = sent.send(body); - StatusCode::NO_CONTENT - } - }), - ); - tokio::spawn(async move { - let _ = axum::serve(projection_listener, app).await; - }); - - let runs = Runs::restored(Forgotten::default()).await?; - let run = runs - .queue( - JobId::fresh(), - Origin::new( - "76128383-1df5-4979-9b13-c048a5287e9a".to_owned(), - "commit".to_owned(), - "refs/heads/main".to_owned(), - "push".to_owned(), - ".gitea/workflows/ci.yml".to_owned(), - ) - .with_delivery(Some("delivery".to_owned())), - b"plan".to_vec(), - ) - .await?; - let projection = runs.projection(run).await?.expect("projection"); - ProjectionClient::new( - projection_base, - String::new(), - identity_endpoint, - "shared-secret".to_owned(), - )? - .send(&projection) - .await?; - - let body = received.recv().await.expect("projection body"); - assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo"); - Ok(()) -} +#![allow(clippy::expect_used)] + +use std::error::Error; + +use axum::Json; +use axum::http::StatusCode; +use axum::routing::post; +use syncode_control_node::ProjectionClient; +use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer}; +use syncode_control_node::identity_wire::{ + CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest, + GetRepositoryCoordinatesResponse, ResolveRepositoryRequest, ResolveRepositoryResponse, + ValidateSessionRequest, ValidateSessionResponse, +}; +use syncode_control_runs::{Forgotten, JobId, Origin, Runs}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; + +type TestResult = Result>; + +struct FixtureIdentity; + +#[tonic::async_trait] +impl Identity for FixtureIdentity { + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("resolve_repository")) + } + + async fn get_repository_coordinates( + &self, + request: Request, + ) -> Result, Status> { + if request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()) + != Some("Bearer shared-secret") + { + return Err(Status::unauthenticated("missing authorization")); + } + Ok(Response::new(GetRepositoryCoordinatesResponse { + owner: "syncode".to_owned(), + name: "pipelines-demo".to_owned(), + })) + } +} + +#[tokio::test] +async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult { + let identity_listener = TcpListener::bind("127.0.0.1:0").await?; + let identity_endpoint = format!("http://{}", identity_listener.local_addr()?); + tokio::spawn(async move { + let _ = Server::builder() + .add_service(IdentityServer::new(FixtureIdentity)) + .serve_with_incoming(TcpListenerStream::new(identity_listener)) + .await; + }); + + let projection_listener = TcpListener::bind("127.0.0.1:0").await?; + let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?; + let (sent, mut received) = mpsc::unbounded_channel(); + let app = axum::Router::new().route( + "/api/internal/actions/syncode/projection", + post(move |Json(body): Json| { + let sent = sent.clone(); + async move { + let _ = sent.send(body); + StatusCode::NO_CONTENT + } + }), + ); + tokio::spawn(async move { + let _ = axum::serve(projection_listener, app).await; + }); + + let runs = Runs::restored(Forgotten::default()).await?; + let run = runs + .queue( + JobId::fresh(), + Origin::new( + "76128383-1df5-4979-9b13-c048a5287e9a".to_owned(), + "commit".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ) + .with_delivery(Some("delivery".to_owned())), + b"plan".to_vec(), + ) + .await?; + let projection = runs.projection(run).await?.expect("projection"); + ProjectionClient::new( + projection_base, + String::new(), + identity_endpoint, + "shared-secret".to_owned(), + )? + .send(&projection) + .await?; + + let body = received.recv().await.expect("projection body"); + assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo"); + assert_eq!(body["origin"]["native"], true); + Ok(()) +} diff --git a/crates/control-node/src/projection.rs b/crates/control-node/src/projection.rs --- a/crates/control-node/src/projection.rs +++ b/crates/control-node/src/projection.rs @@ -1,139 +1,177 @@ -use syncode_control_runs::ProjectedRun; -use thiserror::Error; -use tonic::Request; -use tonic::metadata::{Ascii, MetadataValue}; -use tonic::transport::{Channel, Endpoint}; -use url::Url; -use uuid::Uuid; - -use crate::identity_wire::GetRepositoryCoordinatesRequest; -use crate::identity_wire::identity_client::IdentityClient; - -/// Pushes a run's projected state to the forge. Dispatch uses it to close the -/// gap between handing a node a token and the forge knowing what that token is -/// for; the periodic sweep in the binary uses it for everything after. -#[derive(Clone)] -pub struct ProjectionClient { - endpoint: Url, - token: String, - client: reqwest::Client, - identity: IdentityClient, - authorization: MetadataValue, -} - -#[derive(Debug, Error)] -pub enum ProjectionClientError { - #[error("cannot address the projection endpoint")] - Address, - #[error("cannot address the identity endpoint")] - IdentityAddress, - #[error("cannot authorize identity requests")] - IdentityAuthorization, -} - -#[derive(Debug, Error)] -pub enum ProjectionRequestError { - #[error(transparent)] - Transport(#[from] reqwest::Error), - #[error("HTTP {status}: {body}")] - Rejected { - status: reqwest::StatusCode, - body: String, - }, - #[error("identity rejected the repository projection: {0}")] - Identity(#[from] tonic::Status), - #[error("invalid repository projection origin {0:?}")] - InvalidRepository(String), -} - -impl ProjectionClient { - pub fn new( - base: Url, - token: String, - identity_endpoint: String, - identity_shared_secret: String, - ) -> Result { - let endpoint = base - .join("api/internal/actions/syncode/projection") - .map_err(|_| ProjectionClientError::Address)?; - let identity = IdentityClient::new( - Endpoint::from_shared(identity_endpoint) - .map_err(|_| ProjectionClientError::IdentityAddress)? - .connect_lazy(), - ); - let authorization = format!("Bearer {identity_shared_secret}") - .parse() - .map_err(|_| ProjectionClientError::IdentityAuthorization)?; - Ok(Self { - endpoint, - token, - client: reqwest::Client::new(), - identity, - authorization, - }) - } - - pub async fn send(&self, projection: &ProjectedRun) -> Result<(), ProjectionRequestError> { - let projection = self.for_legacy_projection(projection).await?; - let response = self - .client - .post(self.endpoint.clone()) - .header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token)) - .json(&projection) - .send() - .await?; - let status = response.status(); - if status.is_success() { - return Ok(()); - } - let body = response.text().await?; - Err(ProjectionRequestError::Rejected { status, body }) - } - - async fn for_legacy_projection( - &self, - projection: &ProjectedRun, - ) -> Result { - let repository = projection.origin.repository(); - if Uuid::parse_str(repository).is_ok() { - let mut request = Request::new(GetRepositoryCoordinatesRequest { - repository_id: repository.to_owned(), - }); - request - .metadata_mut() - .insert("authorization", self.authorization.clone()); - let coordinates = self - .identity - .clone() - .get_repository_coordinates(request) - .await? - .into_inner(); - if coordinates.owner.is_empty() - || coordinates.name.is_empty() - || coordinates.owner.contains('/') - || coordinates.name.contains('/') - { - return Err(ProjectionRequestError::InvalidRepository(format!( - "{}/{}", - coordinates.owner, coordinates.name - ))); - } - let mut projected = projection.clone(); - projected.origin = projected - .origin - .with_repository(format!("{}/{}", coordinates.owner, coordinates.name)); - return Ok(projected); - } - let Some((owner, name)) = repository.split_once('/') else { - return Err(ProjectionRequestError::InvalidRepository( - repository.to_owned(), - )); - }; - if owner.is_empty() || name.is_empty() || name.contains('/') { - return Err(ProjectionRequestError::InvalidRepository( - repository.to_owned(), - )); - } - Ok(projection.clone()) - } -} +use serde::Serialize; +use syncode_control_runs::{ProjectedJob, ProjectedRun, RunId, RunNumber}; +use thiserror::Error; +use tonic::Request; +use tonic::metadata::{Ascii, MetadataValue}; +use tonic::transport::{Channel, Endpoint}; +use url::Url; +use uuid::Uuid; + +use crate::identity_wire::GetRepositoryCoordinatesRequest; +use crate::identity_wire::identity_client::IdentityClient; + +/// Pushes a run's projected state to the forge. Dispatch uses it to close the +/// gap between handing a node a token and the forge knowing what that token is +/// for; the periodic sweep in the binary uses it for everything after. +#[derive(Clone)] +pub struct ProjectionClient { + endpoint: Url, + token: String, + client: reqwest::Client, + identity: IdentityClient, + authorization: MetadataValue, +} + +#[derive(Debug, Error)] +pub enum ProjectionClientError { + #[error("cannot address the projection endpoint")] + Address, + #[error("cannot address the identity endpoint")] + IdentityAddress, + #[error("cannot authorize identity requests")] + IdentityAuthorization, +} + +#[derive(Debug, Error)] +pub enum ProjectionRequestError { + #[error(transparent)] + Transport(#[from] reqwest::Error), + #[error("HTTP {status}: {body}")] + Rejected { + status: reqwest::StatusCode, + body: String, + }, + #[error("identity rejected the repository projection: {0}")] + Identity(#[from] tonic::Status), + #[error("invalid repository projection origin {0:?}")] + InvalidRepository(String), +} + +#[derive(Serialize)] +struct LegacyProjection<'a> { + run: &'a RunId, + number: &'a RunNumber, + origin: LegacyOrigin<'a>, + sequence: u64, + jobs: &'a [ProjectedJob], +} + +#[derive(Serialize)] +struct LegacyOrigin<'a> { + repository: String, + commit: &'a str, + reference: &'a str, + event: &'a str, + workflow: &'a str, + delivery: Option<&'a str>, + principal: Option<&'a str>, + secrets_allowed: bool, + native: bool, +} + +impl ProjectionClient { + pub fn new( + base: Url, + token: String, + identity_endpoint: String, + identity_shared_secret: String, + ) -> Result { + let endpoint = base + .join("api/internal/actions/syncode/projection") + .map_err(|_| ProjectionClientError::Address)?; + let identity = IdentityClient::new( + Endpoint::from_shared(identity_endpoint) + .map_err(|_| ProjectionClientError::IdentityAddress)? + .connect_lazy(), + ); + let authorization = format!("Bearer {identity_shared_secret}") + .parse() + .map_err(|_| ProjectionClientError::IdentityAuthorization)?; + Ok(Self { + endpoint, + token, + client: reqwest::Client::new(), + identity, + authorization, + }) + } + + pub async fn send(&self, projection: &ProjectedRun) -> Result<(), ProjectionRequestError> { + let projection = self.for_legacy_projection(projection).await?; + let response = self + .client + .post(self.endpoint.clone()) + .header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token)) + .json(&projection) + .send() + .await?; + let status = response.status(); + if status.is_success() { + return Ok(()); + } + let body = response.text().await?; + Err(ProjectionRequestError::Rejected { status, body }) + } + + async fn for_legacy_projection<'a>( + &self, + projection: &'a ProjectedRun, + ) -> Result, ProjectionRequestError> { + let repository = projection.origin.repository(); + let native = Uuid::parse_str(repository).is_ok(); + let repository = if native { + let mut request = Request::new(GetRepositoryCoordinatesRequest { + repository_id: repository.to_owned(), + }); + request + .metadata_mut() + .insert("authorization", self.authorization.clone()); + let coordinates = self + .identity + .clone() + .get_repository_coordinates(request) + .await? + .into_inner(); + if coordinates.owner.is_empty() + || coordinates.name.is_empty() + || coordinates.owner.contains('/') + || coordinates.name.contains('/') + { + return Err(ProjectionRequestError::InvalidRepository(format!( + "{}/{}", + coordinates.owner, coordinates.name + ))); + } + format!("{}/{}", coordinates.owner, coordinates.name) + } else { + let Some((owner, name)) = repository.split_once('/') else { + return Err(ProjectionRequestError::InvalidRepository( + repository.to_owned(), + )); + }; + if owner.is_empty() || name.is_empty() || name.contains('/') { + return Err(ProjectionRequestError::InvalidRepository( + repository.to_owned(), + )); + } + repository.to_owned() + }; + Ok(LegacyProjection { + run: &projection.run, + number: &projection.number, + origin: LegacyOrigin { + repository, + commit: projection.origin.commit(), + reference: projection.origin.reference(), + event: projection.origin.event(), + workflow: projection.origin.workflow(), + delivery: projection.origin.delivery(), + principal: projection.origin.principal(), + secrets_allowed: projection.origin.secrets_allowed(), + native, + }, + sequence: projection.sequence, + jobs: &projection.jobs, + }) + } +}