fix: mark native action projections #46
@@ -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<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let projection_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?;
|
||||
let (sent, mut received) = mpsc::unbounded_channel();
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/projection",
|
||||
post(move |Json(body): Json<serde_json::Value>| {
|
||||
let sent = sent.clone();
|
||||
async move {
|
||||
let _ = sent.send(body);
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(projection_listener, app).await;
|
||||
});
|
||||
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = runs
|
||||
.queue(
|
||||
JobId::fresh(),
|
||||
Origin::new(
|
||||
"76128383-1df5-4979-9b13-c048a5287e9a".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned())),
|
||||
b"plan".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
let projection = runs.projection(run).await?.expect("projection");
|
||||
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<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let projection_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?;
|
||||
let (sent, mut received) = mpsc::unbounded_channel();
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/projection",
|
||||
post(move |Json(body): Json<serde_json::Value>| {
|
||||
let sent = sent.clone();
|
||||
async move {
|
||||
let _ = sent.send(body);
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(projection_listener, app).await;
|
||||
});
|
||||
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = runs
|
||||
.queue(
|
||||
JobId::fresh(),
|
||||
Origin::new(
|
||||
"76128383-1df5-4979-9b13-c048a5287e9a".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned())),
|
||||
b"plan".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
let projection = runs.projection(run).await?.expect("projection");
|
||||
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(())
|
||||
}
|
||||
@@ -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<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[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<Self, ProjectionClientError> {
|
||||
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<ProjectedRun, ProjectionRequestError> {
|
||||
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<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[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<Self, ProjectionClientError> {
|
||||
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<LegacyProjection<'a>, 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,
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user