feat: Complete native Actions delivery #53

Manually merged
day01 merged 9 commits from feat/0.6-verify-dynamic-git into develop 2026-08-31 07:27:18 +00:00
12 changed files with 476 additions and 17 deletions
+17 -10
View File
@@ -1,225 +1,232 @@
use axum::extract::{Path, State};
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::routing::get;
use axum::{Json, Router};
use serde::Serialize;
use syncode_control_node::ActionsReadService;
use syncode_control_node::actions_wire::{
ActionJob, ActionRun, GetJobLogsRequest, GetRunRequest, ListRunsRequest,
};
use syncode_control_runs::RunLog;
use tonic::{Code, Request};
use tower_http::cors::{AllowOrigin, CorsLayer};

use crate::actions_read::{ActionsAuthorization, ActionsRead};

const SESSION_COOKIE: &str = "syncode_identity_session";

pub fn router<L, A>(read: ActionsRead<L, A>, cors_origins: Vec<String>) -> Router
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization + Clone,
{
let mut router = Router::new()
.route(
"/control/actions/repositories/{owner}/{repository}/runs",
get(list_runs::<L, A>),
)
.route("/control/actions/runs/{run}", get(get_run::<L, A>))
.route(
"/control/actions/runs/{run}/jobs/{job}/logs",
get(get_job_logs::<L, A>),
)
.with_state(read);
if !cors_origins.is_empty() {
router = router.layer(
CorsLayer::new()
.allow_credentials(true)
.allow_methods([Method::GET, Method::OPTIONS])
.allow_origin(AllowOrigin::predicate(move |origin, _| {
origin
.to_str()
.is_ok_and(|origin| cors_origins.iter().any(|value| value == origin))
})),
);
}
router
}

async fn list_runs<L, A>(
State(read): State<ActionsRead<L, A>>,
Path((owner, repository)): Path<(String, String)>,
headers: HeaderMap,
) -> Result<Json<Vec<WebRun>>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
let response = read
.list_runs(Request::new(ListRunsRequest {
session_token: session(&headers)?.to_owned(),
owner,
repository,
}))
.await
.map_err(WebError::from)?
.into_inner();
Ok(Json(response.runs.into_iter().map(WebRun::from).collect()))
}

async fn get_run<L, A>(
State(read): State<ActionsRead<L, A>>,
Path(run): Path<String>,
headers: HeaderMap,
) -> Result<Json<WebRun>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
read.get_run(Request::new(GetRunRequest {
session_token: session(&headers)?.to_owned(),
run_id: run,
}))
.await
.map(|response| Json(response.into_inner().into()))
.map_err(WebError::from)
}

async fn get_job_logs<L, A>(
State(read): State<ActionsRead<L, A>>,
Path((run, job)): Path<(String, String)>,
headers: HeaderMap,
) -> Result<Json<WebLogs>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
read.get_job_logs(Request::new(GetJobLogsRequest {
session_token: session(&headers)?.to_owned(),
run_id: run,
job_id: job,
}))
.await
.map(|response| {
Json(WebLogs {
lines: response.into_inner().lines,
})
})
.map_err(WebError::from)
}

fn session(headers: &HeaderMap) -> Result<&str, WebError> {
let cookie = headers
.get(header::COOKIE)
.ok_or_else(WebError::unauthenticated)?
.to_str()
.map_err(|_| WebError::unauthenticated())?;
cookie
.split(';')
.map(str::trim)
.find_map(|pair| pair.strip_prefix(&format!("{SESSION_COOKIE}=")))
.filter(|token| !token.is_empty())
.ok_or_else(WebError::unauthenticated)
}

#[derive(Serialize)]
pub struct WebRun {
id: String,
number: u64,
commit: String,
reference: String,
event: String,
workflow: String,
state: String,
conclusion: String,
jobs: Vec<WebJob>,
}

impl From<ActionRun> for WebRun {
fn from(run: ActionRun) -> Self {
Self {
id: run.id,
number: run.number,
commit: run.commit,
reference: run.reference,
event: run.event,
workflow: run.workflow,
state: run.state,
conclusion: run.conclusion,
jobs: run.jobs.into_iter().map(WebJob::from).collect(),
}
}
}

#[derive(Serialize)]
struct WebJob {
id: String,
key: String,
state: String,
conclusion: String,
}

impl From<ActionJob> for WebJob {
fn from(job: ActionJob) -> Self {
Self {
id: job.id,
key: job.key,
state: job.state,
conclusion: job.conclusion,
}
}
}

#[derive(Serialize)]
struct WebLogs {
lines: Vec<String>,
}

pub struct WebError {
status: StatusCode,
message: String,
}

impl WebError {
fn unauthenticated() -> Self {
Self {
status: StatusCode::UNAUTHORIZED,
message: "a SynCode identity session is required".to_owned(),
}
}
}

impl From<tonic::Status> for WebError {
fn from(error: tonic::Status) -> Self {
let status = match error.code() {
Code::InvalidArgument => StatusCode::BAD_REQUEST,
Code::Unauthenticated => StatusCode::UNAUTHORIZED,
Code::PermissionDenied => StatusCode::FORBIDDEN,
Code::NotFound => StatusCode::NOT_FOUND,
Code::Aborted => StatusCode::CONFLICT,
Code::Unavailable => StatusCode::SERVICE_UNAVAILABLE,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
Self {
status,
message: error.message().to_owned(),
}
}
}

impl axum::response::IntoResponse for WebError {
fn into_response(self) -> axum::response::Response {
(
self.status,
Json(ErrorBody {
error: self.message,
}),
)
.into_response()
}
}

#[derive(Serialize)]
struct ErrorBody {
error: String,
}
use axum::extract::{Path, State};
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::routing::get;
use axum::{Extension, Json, Router};
use serde::Serialize;
use std::sync::Arc;
use syncode_control_node::ActionsReadService;
use syncode_control_node::actions_wire::{
ActionJob, ActionRun, GetJobLogsRequest, GetRunRequest, ListRunsRequest,
};
use syncode_control_runs::RunLog;
use tonic::{Code, Request};
use tower_http::cors::{AllowOrigin, CorsLayer};

use crate::actions_read::{ActionsAuthorization, ActionsRead};

pub fn router<L, A>(
read: ActionsRead<L, A>,
cors_origins: Vec<String>,
session_cookie_name: String,
) -> Router
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization + Clone,
{
let mut router = Router::new()
.route(
"/control/actions/repositories/{owner}/{repository}/runs",
get(list_runs::<L, A>),
)
.route("/control/actions/runs/{run}", get(get_run::<L, A>))
.route(
"/control/actions/runs/{run}/jobs/{job}/logs",
get(get_job_logs::<L, A>),
)
.with_state(read)
.layer(Extension(Arc::<str>::from(session_cookie_name)));
if !cors_origins.is_empty() {
router = router.layer(
CorsLayer::new()
.allow_credentials(true)
.allow_methods([Method::GET, Method::OPTIONS])
.allow_origin(AllowOrigin::predicate(move |origin, _| {
origin
.to_str()
.is_ok_and(|origin| cors_origins.iter().any(|value| value == origin))
})),
);
}
router
}

async fn list_runs<L, A>(
State(read): State<ActionsRead<L, A>>,
Extension(session_cookie_name): Extension<Arc<str>>,
Path((owner, repository)): Path<(String, String)>,
headers: HeaderMap,
) -> Result<Json<Vec<WebRun>>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
let response = read
.list_runs(Request::new(ListRunsRequest {
session_token: session(&headers, &session_cookie_name)?.to_owned(),
owner,
repository,
}))
.await
.map_err(WebError::from)?
.into_inner();
Ok(Json(response.runs.into_iter().map(WebRun::from).collect()))
}

async fn get_run<L, A>(
State(read): State<ActionsRead<L, A>>,
Extension(session_cookie_name): Extension<Arc<str>>,
Path(run): Path<String>,
headers: HeaderMap,
) -> Result<Json<WebRun>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
read.get_run(Request::new(GetRunRequest {
session_token: session(&headers, &session_cookie_name)?.to_owned(),
run_id: run,
}))
.await
.map(|response| Json(response.into_inner().into()))
.map_err(WebError::from)
}

async fn get_job_logs<L, A>(
State(read): State<ActionsRead<L, A>>,
Extension(session_cookie_name): Extension<Arc<str>>,
Path((run, job)): Path<(String, String)>,
headers: HeaderMap,
) -> Result<Json<WebLogs>, WebError>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
read.get_job_logs(Request::new(GetJobLogsRequest {
session_token: session(&headers, &session_cookie_name)?.to_owned(),
run_id: run,
job_id: job,
}))
.await
.map(|response| {
Json(WebLogs {
lines: response.into_inner().lines,
})
})
.map_err(WebError::from)
}

fn session<'a>(headers: &'a HeaderMap, session_cookie_name: &str) -> Result<&'a str, WebError> {
let cookie = headers
.get(header::COOKIE)
.ok_or_else(WebError::unauthenticated)?
.to_str()
.map_err(|_| WebError::unauthenticated())?;
cookie
.split(';')
.map(str::trim)
.find_map(|pair| pair.strip_prefix(&format!("{session_cookie_name}=")))
.filter(|token| !token.is_empty())
.ok_or_else(WebError::unauthenticated)
}

#[derive(Serialize)]
pub struct WebRun {
id: String,
number: u64,
commit: String,
reference: String,
event: String,
workflow: String,
state: String,
conclusion: String,
jobs: Vec<WebJob>,
}

impl From<ActionRun> for WebRun {
fn from(run: ActionRun) -> Self {
Self {
id: run.id,
number: run.number,
commit: run.commit,
reference: run.reference,
event: run.event,
workflow: run.workflow,
state: run.state,
conclusion: run.conclusion,
jobs: run.jobs.into_iter().map(WebJob::from).collect(),
}
}
}

#[derive(Serialize)]
struct WebJob {
id: String,
key: String,
state: String,
conclusion: String,
}

impl From<ActionJob> for WebJob {
fn from(job: ActionJob) -> Self {
Self {
id: job.id,
key: job.key,
state: job.state,
conclusion: job.conclusion,
}
}
}

#[derive(Serialize)]
struct WebLogs {
lines: Vec<String>,
}

pub struct WebError {
status: StatusCode,
message: String,
}

impl WebError {
fn unauthenticated() -> Self {
Self {
status: StatusCode::UNAUTHORIZED,
message: "a SynCode identity session is required".to_owned(),
}
}
}

impl From<tonic::Status> for WebError {
fn from(error: tonic::Status) -> Self {
let status = match error.code() {
Code::InvalidArgument => StatusCode::BAD_REQUEST,
Code::Unauthenticated => StatusCode::UNAUTHORIZED,
Code::PermissionDenied => StatusCode::FORBIDDEN,
Code::NotFound => StatusCode::NOT_FOUND,
Code::Aborted => StatusCode::CONFLICT,
Code::Unavailable => StatusCode::SERVICE_UNAVAILABLE,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
Self {
status,
message: error.message().to_owned(),
}
}
}

impl axum::response::IntoResponse for WebError {
fn into_response(self) -> axum::response::Response {
(
self.status,
Json(ErrorBody {
error: self.message,
}),
)
.into_response()
}
}

#[derive(Serialize)]
struct ErrorBody {
error: String,
}
+35
View File
@@ -1,451 +1,486 @@
use serde::Deserialize;
use syncode_control_runs::Origin;
use syncode_workflow::{Event, EventKind, GitReference};
use thiserror::Error;

use crate::sources::ChangedFilesRequest;

/// One thing the repository event source says happened, in the shape deciding needs: which
/// repository, at which commit, and what the event was.
///
/// The event source speaks about branches and tags as full references. Everything
/// past this boundary keeps the reference kind while using its short name, because
/// that is what a workflow writes in its filters.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Delivery {
repository: String,
content_repository: Option<String>,
commit: String,
event: Event,
/// Which pull request this came from. It names the reference the commit
/// lives on, which no branch does for a pull request.
pull_request: Option<u64>,
changed_files: Option<ChangedFilesRequest>,
workflow: Option<String>,
delivery: Option<String>,
principal: Option<String>,
secrets_allowed: bool,
}

#[derive(Debug, Error)]
pub enum DeliveryError {
#[error("the repository event source sent a body that is not a known event: {0}")]
Unreadable(#[from] serde_json::Error),

#[error("the repository event source sent event {0:?}, which does not trigger runs")]
Unsupported(String),

#[error("the event names an unsupported repository reference {0:?}")]
NotARepositoryReference(String),

#[error("the event carries no commit to compile from")]
NoCommit,

#[error("the requested event carries no workflow to compile")]
NoWorkflow,

#[error(
"the repository event source sent a pull request {0:?} this control plane does not act on"
)]
UnactedPullRequest(String),

#[error("the repository event uses unsupported protocol version {0:?}")]
UnsupportedProtocol(String),

#[error("the repository event uses unsupported schema version {0}")]
UnsupportedSchema(u32),

#[error("the repository event body names message type {0:?}")]
UnexpectedMessageType(String),

#[error("the repository event carries invalid {0}")]
InvalidNativeField(&'static str),
}

/// The push payload, narrowed to the fields a decision needs. Everything else
/// the event source sends is ignored on purpose: fields we do not read cannot
/// break us when the source adds or renames them.
#[derive(Deserialize)]
struct Push {
#[serde(rename = "ref")]
reference: String,
after: String,
#[serde(default)]
head_commit: Option<HeadCommit>,
repository: Repository,
#[serde(default)]
commits: Vec<Commit>,
#[serde(default)]
sender: Option<Actor>,
}

#[derive(Deserialize)]
struct HeadCommit {
#[serde(default)]
id: String,
}

#[derive(Deserialize)]
struct Repository {
full_name: String,
}

/// The pull request payload, narrowed the same way the push one is.
#[derive(Deserialize)]
struct PullRequestEvent {
action: String,
number: u64,
pull_request: PullRequestBody,
repository: Repository,
#[serde(default)]
sender: Option<Actor>,
}

#[derive(Deserialize)]
struct Actor {
login: String,
}

#[derive(Deserialize)]
struct PullRequestBody {
head: Reference,
base: Reference,
}

#[derive(Deserialize)]
struct Reference {
#[serde(default)]
sha: String,
#[serde(rename = "ref", default)]
reference: String,
#[serde(default, rename = "repo")]
repository: Option<Repository>,
}

#[derive(Deserialize)]
struct Commit {
#[serde(default)]
added: Vec<String>,
#[serde(default)]
removed: Vec<String>,
#[serde(default)]
modified: Vec<String>,
}

#[derive(Deserialize)]
struct RequestedEvent {
#[serde(rename = "ref")]
reference: String,
after: String,
workflow: String,
repository: Repository,
#[serde(default)]
sender: Option<Actor>,
}

/// A commit that is all zeroes is how the event source says a reference was deleted.
/// There is nothing to read a workflow from, so there is nothing to run.
const DELETED: &str = "0000000000000000000000000000000000000000";

const BRANCH_PREFIX: &str = "refs/heads/";
const TAG_PREFIX: &str = "refs/tags/";

fn full_reference(reference: &str) -> Result<GitReference, DeliveryError> {
if let Some(branch) = reference.strip_prefix(BRANCH_PREFIX) {
return branch_or_error(branch, reference);
}
if let Some(tag) = reference.strip_prefix(TAG_PREFIX) {
return tag_or_error(tag, reference);
}
Err(DeliveryError::NotARepositoryReference(reference.to_owned()))
}

fn requested_reference(reference: &str) -> Result<GitReference, DeliveryError> {
if !reference.starts_with("refs/") {
return branch_or_error(reference, reference);
}
full_reference(reference)
}

fn branch_or_error(name: &str, original: &str) -> Result<GitReference, DeliveryError> {
if name.is_empty() {
return Err(DeliveryError::NotARepositoryReference(original.to_owned()));
}
Ok(GitReference::Branch(name.to_owned()))
}

fn tag_or_error(name: &str, original: &str) -> Result<GitReference, DeliveryError> {
if name.is_empty() {
return Err(DeliveryError::NotARepositoryReference(original.to_owned()));
}
Ok(GitReference::Tag(name.to_owned()))
}

impl Delivery {
pub fn read_all(event: &str, body: &[u8]) -> Result<Vec<Self>, DeliveryError> {
match event {
"repository.ref.updated" => crate::native_events::read(body).map(|updates| {
updates
.into_iter()
.map(|update| {
let changed_files = match &update.reference {
GitReference::Branch(_) => Some(ChangedFilesRequest::Commits {
before: update.before,
after: update.after.clone(),
}),
GitReference::Tag(_) => None,
};
Self {
repository: update.repository,
content_repository: None,
commit: update.after,
event: Event::new(EventKind::Push, update.reference, Vec::new()),
pull_request: None,
changed_files,
workflow: None,
delivery: None,
principal: update.principal,
secrets_allowed: true,
}
})
.collect()
}),
"push" => Ok(Vec::new()),
_ => Self::read(event, body).map(|delivery| vec![delivery]),
}
}

/// Read what the repository event source delivered. The event name comes
/// from the header rather than the body, because that is where the source states
/// it; a body alone does not say which kind of event it describes.
pub fn read(event: &str, body: &[u8]) -> Result<Self, DeliveryError> {
match event {
"push" => Self::push(body),
"pull_request" => Self::pull_request(body),
"workflow_dispatch" => Self::requested(body, EventKind::Manual),
"schedule" => Self::requested(body, EventKind::Schedule),
other => Err(DeliveryError::Unsupported(other.to_owned())),
}
}

fn requested(body: &[u8], kind: EventKind) -> Result<Self, DeliveryError> {
let event: RequestedEvent = serde_json::from_slice(body)?;
if event.after == DELETED || event.after.is_empty() {
return Err(DeliveryError::NoCommit);
}
if event.workflow.is_empty() {
return Err(DeliveryError::NoWorkflow);
}
let reference = requested_reference(&event.reference)?;
Ok(Self {
repository: event.repository.full_name,
content_repository: None,
commit: event.after,
event: Event::new(kind, reference, Vec::new()),
pull_request: None,
changed_files: None,
workflow: Some(event.workflow),
delivery: None,
principal: event.sender.map(|actor| actor.login),
secrets_allowed: true,
})
}

fn push(body: &[u8]) -> Result<Self, DeliveryError> {
let push: Push = serde_json::from_slice(body)?;
if push.after == DELETED || push.after.is_empty() {
return Err(DeliveryError::NoCommit);
}
let reference = full_reference(&push.reference)?;
let commit = match &reference {
GitReference::Branch(_) => push.after,
GitReference::Tag(_) => push
.head_commit
.map(|head| head.id)
.filter(|commit| !commit.is_empty() && commit != DELETED)
.ok_or(DeliveryError::NoCommit)?,
};

let mut changed: Vec<String> = push
.commits
.into_iter()
.flat_map(|commit| {
commit
.added
.into_iter()
.chain(commit.removed)
.chain(commit.modified)
})
.collect();
changed.sort_unstable();
changed.dedup();

Ok(Self {
repository: push.repository.full_name,
content_repository: None,
commit,
event: Event::new(EventKind::Push, reference, changed),
pull_request: None,
changed_files: None,
workflow: None,
delivery: None,
principal: push.sender.map(|actor| actor.login),
secrets_allowed: true,
})
}

fn pull_request(body: &[u8]) -> Result<Self, DeliveryError> {
let event: PullRequestEvent = serde_json::from_slice(body)?;
// Closing a pull request does not start work, and the dialect models
// reopening and synchronising as the same thing: there is new code on
// the branch, so build it.
if !matches!(
event.action.as_str(),
"opened" | "reopened" | "synchronized" | "synchronize" | "edited"
) {
return Err(DeliveryError::UnactedPullRequest(event.action));
}
if event.pull_request.head.sha.is_empty() {
return Err(DeliveryError::NoCommit);
}

// The dialect filters a pull request by the branch it is aimed at, not
// the one it comes from, and compiles the workflow at the head commit.
let secrets_allowed = event
.pull_request
.head
.repository
.as_ref()
.is_some_and(|head| head.full_name == event.repository.full_name);
Ok(Self {
repository: event.repository.full_name,
content_repository: None,
commit: event.pull_request.head.sha,
event: Event::new(
EventKind::PullRequest,
GitReference::Branch(event.pull_request.base.reference),
Vec::new(),
),
pull_request: Some(event.number),
changed_files: Some(ChangedFilesRequest::PullRequest(event.number)),
workflow: None,
delivery: None,
principal: event.sender.map(|actor| actor.login),
secrets_allowed,
})
}

/// Which pull request's file list is still missing, if any. The webhook body
/// does not carry it, and a path filter with nothing to admit does not fire
/// — so a workflow with `paths:` would silently never run on a pull request.
#[must_use]
pub fn awaiting_paths(&self) -> Option<ChangedFilesRequest> {
self.changed_files.clone()
}

#[must_use]
pub fn touching(mut self, paths: Vec<String>) -> Self {
self.event = Event::new(self.event.kind(), self.event.reference().clone(), paths);
self.changed_files = None;
self
}

#[must_use]
pub fn at_commit(mut self, commit: String) -> Self {
self.commit = commit;
self
}

#[must_use]
pub fn workflow(&self) -> Option<&str> {
self.workflow.as_deref()
}

/// Where a run compiled from `workflow` came from. The delivery is the same
/// for every workflow the commit carries, so the file is what tells two of
/// its runs apart.
pub fn origin(&self, workflow: &str) -> Origin {
Origin::new(
self.repository.clone(),
self.commit.clone(),
self.reference(),
self.event.kind().to_string(),
workflow.to_owned(),
)
.with_delivery(self.delivery.clone())
.with_principal(self.principal.clone())
.with_secret_trust(self.secrets_allowed)
}

#[must_use]
pub fn identified(mut self, delivery: String) -> Self {
self.delivery = Some(delivery);
self
}

#[must_use]
pub fn selecting(mut self, workflow: Option<String>) -> Self {
if workflow.is_some() {
self.workflow = workflow;
}
self
}

/// The reference the commit being built lives on. A pull request head is
/// not on a branch this control plane knows, but the forge does publish it
/// under the pull request, which is the reference a checkout can fetch.
fn reference(&self) -> String {
match self.pull_request {
Some(number) => format!("refs/pull/{number}/head"),
None => self.event.reference().full_name(),
}
}

#[must_use]
pub fn repository(&self) -> &str {
&self.repository
}

#[must_use]
pub fn content_repository(&self) -> &str {
self.content_repository
.as_deref()
.unwrap_or(&self.repository)
}

pub fn using_repository_content(
mut self,
repository: String,
before: String,
after: String,
) -> Result<Self, DeliveryError> {
if uuid::Uuid::parse_str(&repository).is_err()
|| !valid_commit_id(&before)
|| !valid_commit_id(&after)
{
return Err(DeliveryError::InvalidNativeField(
"repository content selector",
));
}
self.content_repository = Some(repository);
self.changed_files = Some(ChangedFilesRequest::Commits {
before: Some(before),
after,
});
Ok(self)
}

#[must_use]
pub fn commit(&self) -> &str {
&self.commit
}

#[must_use]
pub const fn event(&self) -> &Event {
&self.event
}
}

fn valid_commit_id(value: &str) -> bool {
matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}
use serde::Deserialize;
use syncode_control_runs::Origin;
use syncode_workflow::{Event, EventKind, GitReference};
use thiserror::Error;

use crate::sources::ChangedFilesRequest;

/// One thing the repository event source says happened, in the shape deciding needs: which
/// repository, at which commit, and what the event was.
///
/// The event source speaks about branches and tags as full references. Everything
/// past this boundary keeps the reference kind while using its short name, because
/// that is what a workflow writes in its filters.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Delivery {
repository: String,
content_repository: Option<String>,
commit: String,
event: Event,
/// Which pull request this came from. It names the reference the commit
/// lives on, which no branch does for a pull request.
pull_request: Option<u64>,
checkout_reference: Option<String>,
changed_files: Option<ChangedFilesRequest>,
workflow: Option<String>,
delivery: Option<String>,
principal: Option<String>,
secrets_allowed: bool,
}

#[derive(Debug, Error)]
pub enum DeliveryError {
#[error("the repository event source sent a body that is not a known event: {0}")]
Unreadable(#[from] serde_json::Error),

#[error("the repository event source sent event {0:?}, which does not trigger runs")]
Unsupported(String),

#[error("the event names an unsupported repository reference {0:?}")]
NotARepositoryReference(String),

#[error("the event carries no commit to compile from")]
NoCommit,

#[error("the requested event carries no workflow to compile")]
NoWorkflow,

#[error(
"the repository event source sent a pull request {0:?} this control plane does not act on"
)]
UnactedPullRequest(String),

#[error("the repository event uses unsupported protocol version {0:?}")]
UnsupportedProtocol(String),

#[error("the repository event uses unsupported schema version {0}")]
UnsupportedSchema(u32),

#[error("the repository event body names message type {0:?}")]
UnexpectedMessageType(String),

#[error("the repository event carries invalid {0}")]
InvalidNativeField(&'static str),
}

/// The push payload, narrowed to the fields a decision needs. Everything else
/// the event source sends is ignored on purpose: fields we do not read cannot
/// break us when the source adds or renames them.
#[derive(Deserialize)]
struct Push {
#[serde(rename = "ref")]
reference: String,
after: String,
#[serde(default)]
head_commit: Option<HeadCommit>,
repository: Repository,
#[serde(default)]
commits: Vec<Commit>,
#[serde(default)]
sender: Option<Actor>,
}

#[derive(Deserialize)]
struct HeadCommit {
#[serde(default)]
id: String,
}

#[derive(Deserialize)]
struct Repository {
full_name: String,
}

/// The pull request payload, narrowed the same way the push one is.
#[derive(Deserialize)]
struct PullRequestEvent {
action: String,
number: u64,
pull_request: PullRequestBody,
repository: Repository,
#[serde(default)]
sender: Option<Actor>,
}

#[derive(Deserialize)]
struct Actor {
login: String,
}

#[derive(Deserialize)]
struct PullRequestBody {
head: Reference,
base: Reference,
}

#[derive(Deserialize)]
struct Reference {
#[serde(default)]
sha: String,
#[serde(rename = "ref", default)]
reference: String,
#[serde(default, rename = "repo")]
repository: Option<Repository>,
}

#[derive(Deserialize)]
struct Commit {
#[serde(default)]
added: Vec<String>,
#[serde(default)]
removed: Vec<String>,
#[serde(default)]
modified: Vec<String>,
}

#[derive(Deserialize)]
struct RequestedEvent {
#[serde(rename = "ref")]
reference: String,
after: String,
workflow: String,
repository: Repository,
#[serde(default)]
sender: Option<Actor>,
}

/// A commit that is all zeroes is how the event source says a reference was deleted.
/// There is nothing to read a workflow from, so there is nothing to run.
const DELETED: &str = "0000000000000000000000000000000000000000";

const BRANCH_PREFIX: &str = "refs/heads/";
const TAG_PREFIX: &str = "refs/tags/";

fn full_reference(reference: &str) -> Result<GitReference, DeliveryError> {
if let Some(branch) = reference.strip_prefix(BRANCH_PREFIX) {
return branch_or_error(branch, reference);
}
if let Some(tag) = reference.strip_prefix(TAG_PREFIX) {
return tag_or_error(tag, reference);
}
Err(DeliveryError::NotARepositoryReference(reference.to_owned()))
}

fn requested_reference(reference: &str) -> Result<GitReference, DeliveryError> {
if !reference.starts_with("refs/") {
return branch_or_error(reference, reference);
}
full_reference(reference)
}

fn branch_or_error(name: &str, original: &str) -> Result<GitReference, DeliveryError> {
if name.is_empty() {
return Err(DeliveryError::NotARepositoryReference(original.to_owned()));
}
Ok(GitReference::Branch(name.to_owned()))
}

fn tag_or_error(name: &str, original: &str) -> Result<GitReference, DeliveryError> {
if name.is_empty() {
return Err(DeliveryError::NotARepositoryReference(original.to_owned()));
}
Ok(GitReference::Tag(name.to_owned()))
}

impl Delivery {
pub fn read_all(event: &str, body: &[u8]) -> Result<Vec<Self>, DeliveryError> {
match event {
"repository.ref.updated" => crate::native_events::read(body).map(|updates| {
updates
.into_iter()
.map(|update| {
let changed_files = match &update.reference {
GitReference::Branch(_) => Some(ChangedFilesRequest::Commits {
before: update.before,
after: update.after.clone(),
}),
GitReference::Tag(_) => None,
};
Self {
repository: update.repository,
content_repository: None,
commit: update.after,
event: Event::new(EventKind::Push, update.reference, Vec::new()),
pull_request: None,
checkout_reference: None,
changed_files,
workflow: None,
delivery: None,
principal: update.principal,
secrets_allowed: true,
}
})
.collect()
}),
"collaboration.pull_request.created"
| "collaboration.pull_request.synchronized"
| "collaboration.pull_request.state_changed" => crate::collaboration_events::read(body)
.map(|update| {
vec![Self {
repository: update.repository,
content_repository: Some(update.content_repository),
commit: update.commit.clone(),
event: Event::new(
EventKind::PullRequest,
GitReference::Branch(update.base_branch),
Vec::new(),
),
pull_request: Some(update.number),
checkout_reference: Some(
GitReference::Branch(update.head_branch).full_name(),
),
changed_files: Some(ChangedFilesRequest::Commits {
before: Some(update.base_commit),
after: update.commit,
}),
workflow: None,
delivery: None,
principal: Some(update.principal),
secrets_allowed: update.trusted,
}]
}),
"push" => Ok(Vec::new()),
_ => Self::read(event, body).map(|delivery| vec![delivery]),
}
}

/// Read what the repository event source delivered. The event name comes
/// from the header rather than the body, because that is where the source states
/// it; a body alone does not say which kind of event it describes.
pub fn read(event: &str, body: &[u8]) -> Result<Self, DeliveryError> {
match event {
"push" => Self::push(body),
"pull_request" => Self::pull_request(body),
"workflow_dispatch" => Self::requested(body, EventKind::Manual),
"schedule" => Self::requested(body, EventKind::Schedule),
other => Err(DeliveryError::Unsupported(other.to_owned())),
}
}

fn requested(body: &[u8], kind: EventKind) -> Result<Self, DeliveryError> {
let event: RequestedEvent = serde_json::from_slice(body)?;
if event.after == DELETED || event.after.is_empty() {
return Err(DeliveryError::NoCommit);
}
if event.workflow.is_empty() {
return Err(DeliveryError::NoWorkflow);
}
let reference = requested_reference(&event.reference)?;
Ok(Self {
repository: event.repository.full_name,
content_repository: None,
commit: event.after,
event: Event::new(kind, reference, Vec::new()),
pull_request: None,
checkout_reference: None,
changed_files: None,
workflow: Some(event.workflow),
delivery: None,
principal: event.sender.map(|actor| actor.login),
secrets_allowed: true,
})
}

fn push(body: &[u8]) -> Result<Self, DeliveryError> {
let push: Push = serde_json::from_slice(body)?;
if push.after == DELETED || push.after.is_empty() {
return Err(DeliveryError::NoCommit);
}
let reference = full_reference(&push.reference)?;
let commit = match &reference {
GitReference::Branch(_) => push.after,
GitReference::Tag(_) => push
.head_commit
.map(|head| head.id)
.filter(|commit| !commit.is_empty() && commit != DELETED)
.ok_or(DeliveryError::NoCommit)?,
};

let mut changed: Vec<String> = push
.commits
.into_iter()
.flat_map(|commit| {
commit
.added
.into_iter()
.chain(commit.removed)
.chain(commit.modified)
})
.collect();
changed.sort_unstable();
changed.dedup();

Ok(Self {
repository: push.repository.full_name,
content_repository: None,
commit,
event: Event::new(EventKind::Push, reference, changed),
pull_request: None,
checkout_reference: None,
changed_files: None,
workflow: None,
delivery: None,
principal: push.sender.map(|actor| actor.login),
secrets_allowed: true,
})
}

fn pull_request(body: &[u8]) -> Result<Self, DeliveryError> {
let event: PullRequestEvent = serde_json::from_slice(body)?;
// Closing a pull request does not start work, and the dialect models
// reopening and synchronising as the same thing: there is new code on
// the branch, so build it.
if !matches!(
event.action.as_str(),
"opened" | "reopened" | "synchronized" | "synchronize" | "edited"
) {
return Err(DeliveryError::UnactedPullRequest(event.action));
}
if event.pull_request.head.sha.is_empty() {
return Err(DeliveryError::NoCommit);
}

// The dialect filters a pull request by the branch it is aimed at, not
// the one it comes from, and compiles the workflow at the head commit.
let secrets_allowed = event
.pull_request
.head
.repository
.as_ref()
.is_some_and(|head| head.full_name == event.repository.full_name);
Ok(Self {
repository: event.repository.full_name,
content_repository: None,
commit: event.pull_request.head.sha,
event: Event::new(
EventKind::PullRequest,
GitReference::Branch(event.pull_request.base.reference),
Vec::new(),
),
pull_request: Some(event.number),
checkout_reference: None,
changed_files: Some(ChangedFilesRequest::PullRequest(event.number)),
workflow: None,
delivery: None,
principal: event.sender.map(|actor| actor.login),
secrets_allowed,
})
}

/// Which pull request's file list is still missing, if any. The webhook body
/// does not carry it, and a path filter with nothing to admit does not fire
/// — so a workflow with `paths:` would silently never run on a pull request.
#[must_use]
pub fn awaiting_paths(&self) -> Option<ChangedFilesRequest> {
self.changed_files.clone()
}

#[must_use]
pub fn touching(mut self, paths: Vec<String>) -> Self {
self.event = Event::new(self.event.kind(), self.event.reference().clone(), paths);
self.changed_files = None;
self
}

#[must_use]
pub fn at_commit(mut self, commit: String) -> Self {
self.commit = commit;
self
}

#[must_use]
pub fn workflow(&self) -> Option<&str> {
self.workflow.as_deref()
}

/// Where a run compiled from `workflow` came from. The delivery is the same
/// for every workflow the commit carries, so the file is what tells two of
/// its runs apart.
pub fn origin(&self, workflow: &str) -> Origin {
Origin::new(
self.repository.clone(),
self.commit.clone(),
self.reference(),
self.event.kind().to_string(),
workflow.to_owned(),
)
.with_delivery(self.delivery.clone())
.with_principal(self.principal.clone())
.with_secret_trust(self.secrets_allowed)
}

#[must_use]
pub fn identified(mut self, delivery: String) -> Self {
self.delivery = Some(delivery);
self
}

#[must_use]
pub fn selecting(mut self, workflow: Option<String>) -> Self {
if workflow.is_some() {
self.workflow = workflow;
}
self
}

/// The reference the commit being built lives on. A pull request head is
/// not on a branch this control plane knows, but the forge does publish it
/// under the pull request, which is the reference a checkout can fetch.
fn reference(&self) -> String {
if let Some(reference) = &self.checkout_reference {
return reference.clone();
}
match self.pull_request {
Some(number) => format!("refs/pull/{number}/head"),
None => self.event.reference().full_name(),
}
}

#[must_use]
pub fn repository(&self) -> &str {
&self.repository
}

#[must_use]
pub fn content_repository(&self) -> &str {
self.content_repository
.as_deref()
.unwrap_or(&self.repository)
}

pub fn using_repository_content(
mut self,
repository: String,
before: String,
after: String,
) -> Result<Self, DeliveryError> {
if uuid::Uuid::parse_str(&repository).is_err()
|| !valid_commit_id(&before)
|| !valid_commit_id(&after)
{
return Err(DeliveryError::InvalidNativeField(
"repository content selector",
));
}
self.content_repository = Some(repository);
self.changed_files = Some(ChangedFilesRequest::Commits {
before: Some(before),
after,
});
Ok(self)
}

#[must_use]
pub fn commit(&self) -> &str {
&self.commit
}

#[must_use]
pub const fn event(&self) -> &Event {
&self.event
}
}

fn valid_commit_id(value: &str) -> bool {
matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}
+8 -1
View File
@@ -1,222 +1,229 @@
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use serde::Deserialize;
use thiserror::Error;
use url::Url;

use crate::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};

/// The workflow directories supported by the repository service.
const DIRECTORIES: [&str; 2] = [".gitea/workflows", ".github/workflows"];

pub struct RepositoryContents {
base: Url,
token: String,
client: reqwest::Client,
}

pub struct RepositorySecrets {
base: Url,
token: String,
client: reqwest::Client,
}

#[derive(Debug, Error)]
pub enum RepositoryError {
#[error("cannot reach the repository service: {0}")]
Unreachable(#[from] reqwest::Error),

#[error("the repository service answered {status} for {path}")]
Refused { status: u16, path: String },

#[error("the repository service returned a workflow file that is not valid base64: {0}")]
Undecodable(#[from] base64::DecodeError),

#[error("cannot address {path} on the repository service")]
Address { path: String },

#[error("the forge cannot compare native repository commits")]
UnsupportedChangedFiles,
}

#[derive(Deserialize)]
struct ChangedFile {
filename: String,
}

#[derive(Deserialize)]
struct Entry {
path: String,
#[serde(rename = "type")]
kind: String,
content: Option<String>,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum Contents {
Many(Vec<Entry>),
One(Entry),
}

#[derive(Deserialize)]
struct SecretResponse {
value: String,
}

impl RepositoryContents {
#[must_use]
pub fn new(base: Url, token: String) -> Self {
Self {
base,
token,
client: reqwest::Client::new(),
}
}

async fn entries(&self, path: &str) -> Result<Vec<Entry>, RepositoryError> {
let url = self
.base
.join(&format!("api/v1/repos/{path}"))
.map_err(|_| RepositoryError::Address {
path: path.to_owned(),
})?;
let response = self
.client
.get(url)
.header("Authorization", format!("token {}", self.token))
.send()
.await?;
// A repository without one of the directories is not a failure; it is a
// repository that keeps its workflows in the other one, or in neither.
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(Vec::new());
}
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path: path.to_owned(),
});
}
Ok(match response.json().await? {
Contents::Many(entries) => entries,
Contents::One(entry) => vec![entry],
})
}
}

impl RepositorySecrets {
#[must_use]
pub fn new(base: Url, token: String) -> Self {
Self {
base,
token,
client: reqwest::Client::new(),
}
}
}

impl crate::secrets::SecretSource for RepositorySecrets {
type Error = RepositoryError;

async fn resolve(
&self,
repository: &str,
name: &str,
) -> Result<Option<String>, RepositoryError> {
let path = format!("api/internal/actions/syncode/secrets/{repository}/{name}");
let url = self
.base
.join(&path)
.map_err(|_| RepositoryError::Address { path: path.clone() })?;
let response = self
.client
.get(url)
.header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token))
.send()
.await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(None);
}
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path,
});
}
Ok(Some(response.json::<SecretResponse>().await?.value))
}
}

impl WorkflowSources for RepositoryContents {
type Error = RepositoryError;

async fn workflows(
&self,
repository: &str,
commit: &str,
) -> Result<Vec<WorkflowFile>, RepositoryError> {
let mut files = Vec::new();
for directory in DIRECTORIES {
let listing = self
.entries(&format!("{repository}/contents/{directory}?ref={commit}"))
.await?;
for entry in listing.into_iter().filter(|entry| entry.kind == "file") {
let content = match entry.content {
Some(content) => content,
None => {
let mut file = self
.entries(&format!(
"{repository}/contents/{}?ref={commit}",
entry.path
))
.await?;
match file.pop().and_then(|entry| entry.content) {
Some(content) => content,
None => continue,
}
}
};
let content = STANDARD.decode(content.replace(['\n', '\r'], ""))?;
files.push(WorkflowFile::new(entry.path, content));
}
}
Ok(files)
}
}

impl ChangedFiles for RepositoryContents {
type Error = RepositoryError;

async fn changed(
&self,
repository: &str,
request: ChangedFilesRequest,
) -> Result<Vec<String>, RepositoryError> {
let ChangedFilesRequest::PullRequest(pull_request) = request else {
return Err(RepositoryError::UnsupportedChangedFiles);
};
let url = self
.base
.join(&format!(
"api/v1/repos/{repository}/pulls/{pull_request}/files"
))
.map_err(|_| RepositoryError::Address {
path: repository.to_owned(),
})?;
let response = self
.client
.get(url)
.header("Authorization", format!("token {}", self.token))
.send()
.await?;
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path: format!("{repository}/pulls/{pull_request}/files"),
});
}
let files: Vec<ChangedFile> = response.json().await?;
Ok(files.into_iter().map(|file| file.filename).collect())
}
}
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use serde::Deserialize;
use syncode_control_node::{RepositoryCoordinateRequestError, RepositoryCoordinates};
use thiserror::Error;
use url::Url;

use crate::sources::{ChangedFiles, ChangedFilesRequest, WorkflowFile, WorkflowSources};

/// The workflow directories supported by the repository service.
const DIRECTORIES: [&str; 2] = [".gitea/workflows", ".github/workflows"];

pub struct RepositoryContents {
base: Url,
token: String,
client: reqwest::Client,
}

pub struct RepositorySecrets {
base: Url,
token: String,
client: reqwest::Client,
coordinates: RepositoryCoordinates,
}

#[derive(Debug, Error)]
pub enum RepositoryError {
#[error("cannot reach the repository service: {0}")]
Unreachable(#[from] reqwest::Error),

#[error("the repository service answered {status} for {path}")]
Refused { status: u16, path: String },

#[error("the repository service returned a workflow file that is not valid base64: {0}")]
Undecodable(#[from] base64::DecodeError),

#[error("cannot address {path} on the repository service")]
Address { path: String },

#[error("cannot resolve repository coordinates: {0}")]
Coordinates(#[from] RepositoryCoordinateRequestError),

#[error("the forge cannot compare native repository commits")]
UnsupportedChangedFiles,
}

#[derive(Deserialize)]
struct ChangedFile {
filename: String,
}

#[derive(Deserialize)]
struct Entry {
path: String,
#[serde(rename = "type")]
kind: String,
content: Option<String>,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum Contents {
Many(Vec<Entry>),
One(Entry),
}

#[derive(Deserialize)]
struct SecretResponse {
value: String,
}

impl RepositoryContents {
#[must_use]
pub fn new(base: Url, token: String) -> Self {
Self {
base,
token,
client: reqwest::Client::new(),
}
}

async fn entries(&self, path: &str) -> Result<Vec<Entry>, RepositoryError> {
let url = self
.base
.join(&format!("api/v1/repos/{path}"))
.map_err(|_| RepositoryError::Address {
path: path.to_owned(),
})?;
let response = self
.client
.get(url)
.header("Authorization", format!("token {}", self.token))
.send()
.await?;
// A repository without one of the directories is not a failure; it is a
// repository that keeps its workflows in the other one, or in neither.
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(Vec::new());
}
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path: path.to_owned(),
});
}
Ok(match response.json().await? {
Contents::Many(entries) => entries,
Contents::One(entry) => vec![entry],
})
}
}

impl RepositorySecrets {
#[must_use]
pub fn new(base: Url, token: String, coordinates: RepositoryCoordinates) -> Self {
Self {
base,
token,
client: reqwest::Client::new(),
coordinates,
}
}
}

impl crate::secrets::SecretSource for RepositorySecrets {
type Error = RepositoryError;

async fn resolve(
&self,
repository: &str,
name: &str,
) -> Result<Option<String>, RepositoryError> {
let repository = self.coordinates.resolve(repository).await?;
let path = format!("api/internal/actions/syncode/secrets/{repository}/{name}");
let url = self
.base
.join(&path)
.map_err(|_| RepositoryError::Address { path: path.clone() })?;
let response = self
.client
.get(url)
.header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token))
.send()
.await?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(None);
}
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path,
});
}
Ok(Some(response.json::<SecretResponse>().await?.value))
}
}

impl WorkflowSources for RepositoryContents {
type Error = RepositoryError;

async fn workflows(
&self,
repository: &str,
commit: &str,
) -> Result<Vec<WorkflowFile>, RepositoryError> {
let mut files = Vec::new();
for directory in DIRECTORIES {
let listing = self
.entries(&format!("{repository}/contents/{directory}?ref={commit}"))
.await?;
for entry in listing.into_iter().filter(|entry| entry.kind == "file") {
let content = match entry.content {
Some(content) => content,
None => {
let mut file = self
.entries(&format!(
"{repository}/contents/{}?ref={commit}",
entry.path
))
.await?;
match file.pop().and_then(|entry| entry.content) {
Some(content) => content,
None => continue,
}
}
};
let content = STANDARD.decode(content.replace(['\n', '\r'], ""))?;
files.push(WorkflowFile::new(entry.path, content));
}
}
Ok(files)
}
}

impl ChangedFiles for RepositoryContents {
type Error = RepositoryError;

async fn changed(
&self,
repository: &str,
request: ChangedFilesRequest,
) -> Result<Vec<String>, RepositoryError> {
let ChangedFilesRequest::PullRequest(pull_request) = request else {
return Err(RepositoryError::UnsupportedChangedFiles);
};
let url = self
.base
.join(&format!(
"api/v1/repos/{repository}/pulls/{pull_request}/files"
))
.map_err(|_| RepositoryError::Address {
path: repository.to_owned(),
})?;
let response = self
.client
.get(url)
.header("Authorization", format!("token {}", self.token))
.send()
.await?;
if !response.status().is_success() {
return Err(RepositoryError::Refused {
status: response.status().as_u16(),
path: format!("{repository}/pulls/{pull_request}/files"),
});
}
let files: Vec<ChangedFile> = response.json().await?;
Ok(files.into_iter().map(|file| file.filename).collect())
}
}
+1
View File
@@ -1,28 +1,29 @@
pub mod action_delivery;
pub mod action_oci;
pub mod action_repository;
pub mod action_store;
pub mod actions;
pub mod actions_read;
pub mod actions_read_http;
pub mod actions_read_identity;
pub mod admin;
pub mod check_events;
pub mod checks;
pub mod events;
pub mod intake;
pub mod maintenance;
mod native_events;
#[path = "forge.rs"]
pub mod repository;
pub mod repository_credentials;
pub mod repository_grpc;
pub mod repository_sources;
pub mod reusable;
mod secret_reference_syntax;
pub mod secret_references;
pub mod secrets;
pub mod sources;
pub mod token;
pub mod trigger;
pub mod webhook;
pub mod action_delivery;
pub mod action_oci;
pub mod action_repository;
pub mod action_store;
pub mod actions;
pub mod actions_read;
pub mod actions_read_http;
pub mod actions_read_identity;
pub mod admin;
pub mod check_events;
pub mod checks;
mod collaboration_events;
pub mod events;
pub mod intake;
pub mod maintenance;
mod native_events;
#[path = "forge.rs"]
pub mod repository;
pub mod repository_credentials;
pub mod repository_grpc;
pub mod repository_sources;
pub mod reusable;
mod secret_reference_syntax;
pub mod secret_references;
pub mod secrets;
pub mod sources;
pub mod token;
pub mod trigger;
pub mod webhook;
+24 -2
View File
@@ -1,385 +1,407 @@
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use clap::{Parser, Subcommand, ValueEnum};
use syncode_control::action_delivery::router as action_delivery_router;
use syncode_control::action_oci::PinnedOciResolver;
use syncode_control::action_repository::NativeActionRepository;
use syncode_control::action_store::FileActionStore;
use syncode_control::actions::ActionResolver;
use syncode_control::actions_read::ActionsRead;
use syncode_control::actions_read_http::router as actions_read_http_router;
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
use syncode_control::admin::{Admin, router as admin_router};
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
use syncode_control::checks::Checks;
use syncode_control::maintenance;
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
use syncode_control::repository_grpc::NativeRepositoryContents;
use syncode_control::repository_sources::RepositorySources;
use syncode_control::secrets::RuntimeSecrets;
use syncode_control::token::EnrolmentScope;
use syncode_control::webhook::{Intake, router};
use syncode_control_node::{
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, RepositoryCoordinates,
};
use syncode_control_nodes::{Nodes, Scope};
use syncode_control_runs::{
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
};
use syncode_control_store::Postgres;
use tokio::net::TcpListener;
use tonic::transport::Server;
use url::Url;

#[derive(Debug, Parser)]
#[command(name = "syncode-control", version, about)]
struct Arguments {
#[command(subcommand)]
command: Option<Command>,

/// Where nodes open their session.
#[arg(long, default_value = "127.0.0.1:8090")]
listen: SocketAddr,

/// Where the native repository event feed delivers events.
#[arg(long, default_value = "127.0.0.1:8091")]
listen_events: SocketAddr,

/// The repository service this control plane reads workflows from.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
repository_source: Url,

/// A token allowed to read repository contents.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
repository_source_token: String,

/// Where native repositories are read over gRPC.
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
repository_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
identity_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
identity_shared_secret: String,

#[arg(long, env = "SYNCODE_CONTROL_CORS_ORIGINS", value_delimiter = ',')]
control_cors_origins: Vec<String>,

#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
action_mirror_url: Url,

#[arg(
long,
env = "SYNCODE_ACTION_ALLOWLIST",
value_delimiter = ',',
num_args = 1..
)]
action_allowlist: Vec<Url>,

#[arg(
long,
env = "SYNCODE_ACTION_OCI_REGISTRIES",
value_delimiter = ',',
num_args = 1..
)]
action_oci_registries: Vec<String>,

#[arg(long, env = "SYNCODE_ACTION_STORE")]
action_store: PathBuf,

#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
action_artifact_public_url: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
secret_source: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
secret_source_token: String,

#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
capability_signing_key: String,

#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
artifact_signing_key: String,

/// The secret used to sign the native event feed.
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
event_feed_secret: String,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
collaboration_event_url: Url,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
collaboration_event_secret: String,

#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
source_node_id: uuid::Uuid,

#[arg(
long,
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
default_value_t = 1
)]
check_event_interval_seconds: u64,

#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
check_event_batch: i64,

/// Bearer token protecting the operational API.
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
admin_token: Option<String>,

/// Where the run log is kept.
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
database: String,

#[arg(long, default_value_t = 8)]
database_connections: u32,

#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
organization_concurrency: u32,

#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
repository_concurrency: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
principal_concurrency: u32,

#[arg(
long,
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
default_value_t = 10_000
)]
organization_queue_quota: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
principal_queue_quota: u32,

#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
audit_retention_days: u32,

/// Whether compiled shadow runs may be assigned to nodes.
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
mode: Mode,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
enum Mode {
Shadow,
Active,
}

#[derive(Debug, Subcommand)]
enum Command {
/// Issue an enrolment token a node can spend for an identity.
///
/// The token is stored before it is printed, so one that reaches an
/// operator is one the control plane will honour.
IssueToken {
/// How far the token reaches: `instance`, `organisation:<name>` or
/// `repository:<owner>/<name>`.
#[arg(long, default_value = "instance")]
scope: String,
},
/// Revoke a node identity immediately.
RevokeNode {
/// The node UUID printed at enrolment.
node: String,
},
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let arguments = Arguments::parse();
// The log is read before anything is served: a node that reconnects has to
// meet the run it was holding, not a control plane that has forgotten it.
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
let policy = SchedulerPolicy::new(
arguments.organization_concurrency,
arguments.repository_concurrency,
arguments.principal_concurrency,
arguments.organization_queue_quota,
arguments.principal_queue_quota,
);
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
let nodes = Nodes::restored(store.clone()).await?;

if let Some(command) = arguments.command {
match command {
Command::IssueToken { scope } => {
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
let token = nodes.issue_token(scope).await?;
println!("{}", token.secret().expose());
}
Command::RevokeNode { node } => {
nodes.revoke(node.parse::<NodeId>()?).await?;
}
}
return Ok(());
}

store
.record_audit(AuditEvent::new("control.configuration", "applied").after(
AuditPayload::Configuration(AuditConfiguration {
mode: match arguments.mode {
Mode::Shadow => AuditControlMode::Shadow,
Mode::Active => AuditControlMode::Active,
},
organization_concurrency: arguments.organization_concurrency,
repository_concurrency: arguments.repository_concurrency,
principal_concurrency: arguments.principal_concurrency,
organization_queue_quota: arguments.organization_queue_quota,
principal_queue_quota: arguments.principal_queue_quota,
audit_retention_days: arguments.audit_retention_days,
}),
))
.await?;
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
tokio::spawn(maintenance::retain_audit(
store.clone(),
arguments.audit_retention_days,
));

let action_mirror = arguments
.action_mirror_url
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()?;
let action_allowlist = arguments
.action_allowlist
.into_iter()
.map(|repository| {
repository
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()
})
.collect::<Result<Vec<_>, _>>()?;
let action_store = FileActionStore::new(arguments.action_store);
let action_repository = NativeActionRepository::connect(
arguments.repository_grpc.clone(),
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?;
let action_resolver = ActionResolver::new(
action_repository,
action_store.clone(),
PinnedOciResolver::new(arguments.action_oci_registries)?,
action_mirror,
action_allowlist,
);
let native_repository =
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
let repository_sources = RepositorySources::new(
native_repository,
RepositoryContents::new(
arguments.repository_source,
arguments.repository_source_token,
),
);
let intake = Arc::new(Intake::new(
repository_sources,
runs.clone(),
action_resolver,
arguments.event_feed_secret,
));
let admin_token = arguments
.admin_token
.filter(|token| !token.is_empty())
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
let actions_read = ActionsRead::new(
runs.clone(),
IdentityActionsAuthorization::connect(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?,
);
let http = router(intake)
.merge(admin_router(Arc::new(
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
)))
.merge(action_delivery_router(
action_store,
artifact_authority.clone(),
runs.clone(),
))
.merge(actions_read_http_router(
actions_read.clone(),
arguments.control_cors_origins,
));
let events = TcpListener::bind(arguments.listen_events).await?;
let check_events = CheckEventPublisher::new(
store.clone(),
CheckEventPublisherConfig {
endpoint: arguments.collaboration_event_url.as_str(),
secret: arguments.collaboration_event_secret,
source_node_id: arguments.source_node_id,
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
batch: arguments.check_event_batch,
},
)?;

let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
let repositories = RepositoryCoordinates::new(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)?;
let secret_service = RuntimeSecrets::new(
runs.clone(),
nodes.clone(),
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
IdentityRepositoryCredentials::connect(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?,
authority.clone(),
);
// Both ends of the control plane run for as long as the other does: without
// events there is nothing to assign, and without sessions there is nobody to
// assign it to. Whichever stops first takes the process down with it.
let node_service = match arguments.mode {
Mode::Shadow => NodeSessionServer::shadow(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
repositories.clone(),
),
Mode::Active => NodeSessionServer::new(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
repositories,
),
};
tokio::select! {
served = axum::serve(events, http).into_future() => served?,
published = check_events.run() => published?,
served = Server::builder()
.add_service(GeneratedServer::new(node_service))
.add_service(GeneratedSecretsServer::new(secret_service))
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
.add_service(GeneratedActionsReadServer::new(actions_read))
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
}

Ok(())
}

async fn shutdown() {
if let Err(error) = tokio::signal::ctrl_c().await {
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
std::future::pending::<()>().await;
}
}
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use clap::{Parser, Subcommand, ValueEnum};
use syncode_control::action_delivery::router as action_delivery_router;
use syncode_control::action_oci::PinnedOciResolver;
use syncode_control::action_repository::NativeActionRepository;
use syncode_control::action_store::FileActionStore;
use syncode_control::actions::ActionResolver;
use syncode_control::actions_read::ActionsRead;
use syncode_control::actions_read_http::router as actions_read_http_router;
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
use syncode_control::admin::{Admin, router as admin_router};
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
use syncode_control::checks::Checks;
use syncode_control::maintenance;
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
use syncode_control::repository_grpc::NativeRepositoryContents;
use syncode_control::repository_sources::RepositorySources;
use syncode_control::secrets::{RuntimeGitConfig, RuntimeSecrets};
use syncode_control::token::EnrolmentScope;
use syncode_control::webhook::{Intake, router};
use syncode_control_node::{
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, RepositoryCoordinates,
};
use syncode_control_nodes::{Nodes, Scope};
use syncode_control_runs::{
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
};
use syncode_control_store::Postgres;
use tokio::net::TcpListener;
use tonic::transport::Server;
use url::Url;

#[derive(Debug, Parser)]
#[command(name = "syncode-control", version, about)]
struct Arguments {
#[command(subcommand)]
command: Option<Command>,

/// Where nodes open their session.
#[arg(long, default_value = "127.0.0.1:8090")]
listen: SocketAddr,

/// Where the native repository event feed delivers events.
#[arg(long, default_value = "127.0.0.1:8091")]
listen_events: SocketAddr,

/// The repository service this control plane reads workflows from.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
repository_source: Url,

/// A token allowed to read repository contents.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
repository_source_token: String,

/// Where native repositories are read over gRPC.
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
repository_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
identity_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
identity_shared_secret: String,

#[arg(long, env = "SYNCODE_CONTROL_CORS_ORIGINS", value_delimiter = ',')]
control_cors_origins: Vec<String>,

#[arg(long, env = "SYNCODE_IDENTITY_SESSION_COOKIE_NAME")]
identity_session_cookie_name: String,

#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
action_mirror_url: Url,

#[arg(
long,
env = "SYNCODE_ACTION_ALLOWLIST",
value_delimiter = ',',
num_args = 1..
)]
action_allowlist: Vec<Url>,

#[arg(
long,
env = "SYNCODE_ACTION_OCI_REGISTRIES",
value_delimiter = ',',
num_args = 1..
)]
action_oci_registries: Vec<String>,

#[arg(long, env = "SYNCODE_ACTION_STORE")]
action_store: PathBuf,

#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
action_artifact_public_url: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
secret_source: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
secret_source_token: String,

#[arg(long, env = "SYNCODE_GIT_CANONICAL_URL")]
git_canonical_url: Url,

#[arg(long, env = "SYNCODE_GIT_TARGET_URL")]
git_target_url: Url,

#[arg(long, env = "SYNCODE_GIT_DEPENDENCIES", value_delimiter = ',')]
git_dependencies: Vec<String>,

#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
capability_signing_key: String,

#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
artifact_signing_key: String,

/// The secret used to sign the native event feed.
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
event_feed_secret: String,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
collaboration_event_url: Url,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
collaboration_event_secret: String,

#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
source_node_id: uuid::Uuid,

#[arg(
long,
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
default_value_t = 1
)]
check_event_interval_seconds: u64,

#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
check_event_batch: i64,

/// Bearer token protecting the operational API.
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
admin_token: Option<String>,

/// Where the run log is kept.
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
database: String,

#[arg(long, default_value_t = 8)]
database_connections: u32,

#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
organization_concurrency: u32,

#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
repository_concurrency: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
principal_concurrency: u32,

#[arg(
long,
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
default_value_t = 10_000
)]
organization_queue_quota: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
principal_queue_quota: u32,

#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
audit_retention_days: u32,

/// Whether compiled shadow runs may be assigned to nodes.
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
mode: Mode,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
enum Mode {
Shadow,
Active,
}

#[derive(Debug, Subcommand)]
enum Command {
/// Issue an enrolment token a node can spend for an identity.
///
/// The token is stored before it is printed, so one that reaches an
/// operator is one the control plane will honour.
IssueToken {
/// How far the token reaches: `instance`, `organisation:<name>` or
/// `repository:<owner>/<name>`.
#[arg(long, default_value = "instance")]
scope: String,
},
/// Revoke a node identity immediately.
RevokeNode {
/// The node UUID printed at enrolment.
node: String,
},
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let arguments = Arguments::parse();
// The log is read before anything is served: a node that reconnects has to
// meet the run it was holding, not a control plane that has forgotten it.
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
let policy = SchedulerPolicy::new(
arguments.organization_concurrency,
arguments.repository_concurrency,
arguments.principal_concurrency,
arguments.organization_queue_quota,
arguments.principal_queue_quota,
);
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
let nodes = Nodes::restored(store.clone()).await?;

if let Some(command) = arguments.command {
match command {
Command::IssueToken { scope } => {
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
let token = nodes.issue_token(scope).await?;
println!("{}", token.secret().expose());
}
Command::RevokeNode { node } => {
nodes.revoke(node.parse::<NodeId>()?).await?;
}
}
return Ok(());
}

store
.record_audit(AuditEvent::new("control.configuration", "applied").after(
AuditPayload::Configuration(AuditConfiguration {
mode: match arguments.mode {
Mode::Shadow => AuditControlMode::Shadow,
Mode::Active => AuditControlMode::Active,
},
organization_concurrency: arguments.organization_concurrency,
repository_concurrency: arguments.repository_concurrency,
principal_concurrency: arguments.principal_concurrency,
organization_queue_quota: arguments.organization_queue_quota,
principal_queue_quota: arguments.principal_queue_quota,
audit_retention_days: arguments.audit_retention_days,
}),
))
.await?;
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
tokio::spawn(maintenance::retain_audit(
store.clone(),
arguments.audit_retention_days,
));

let action_mirror = arguments
.action_mirror_url
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()?;
let action_allowlist = arguments
.action_allowlist
.into_iter()
.map(|repository| {
repository
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()
})
.collect::<Result<Vec<_>, _>>()?;
let action_store = FileActionStore::new(arguments.action_store);
let action_repository = NativeActionRepository::connect(
arguments.repository_grpc.clone(),
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?;
let action_resolver = ActionResolver::new(
action_repository,
action_store.clone(),
PinnedOciResolver::new(arguments.action_oci_registries)?,
action_mirror,
action_allowlist,
);
let native_repository =
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).await?;
let repository_sources = RepositorySources::new(
native_repository,
RepositoryContents::new(
arguments.repository_source,
arguments.repository_source_token,
),
);
let intake = Arc::new(Intake::new(
repository_sources,
runs.clone(),
action_resolver,
arguments.event_feed_secret,
));
let admin_token = arguments
.admin_token
.filter(|token| !token.is_empty())
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
let actions_read = ActionsRead::new(
runs.clone(),
IdentityActionsAuthorization::connect(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?,
);
let http = router(intake)
.merge(admin_router(Arc::new(
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
)))
.merge(action_delivery_router(
action_store,
artifact_authority.clone(),
runs.clone(),
))
.merge(actions_read_http_router(
actions_read.clone(),
arguments.control_cors_origins,
arguments.identity_session_cookie_name,
));
let events = TcpListener::bind(arguments.listen_events).await?;
let check_events = CheckEventPublisher::new(
store.clone(),
CheckEventPublisherConfig {
endpoint: arguments.collaboration_event_url.as_str(),
secret: arguments.collaboration_event_secret,
source_node_id: arguments.source_node_id,
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
batch: arguments.check_event_batch,
},
)?;

let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
let repositories = RepositoryCoordinates::new(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)?;
let secret_service = RuntimeSecrets::new(
runs.clone(),
nodes.clone(),
RepositorySecrets::new(
arguments.secret_source,
arguments.secret_source_token,
repositories.clone(),
),
IdentityRepositoryCredentials::connect(
arguments.identity_grpc.clone(),
arguments.identity_shared_secret.clone(),
)
.await?,
RuntimeGitConfig::new(
arguments.git_canonical_url,
arguments.git_target_url,
arguments.git_dependencies,
),
authority.clone(),
);
// Both ends of the control plane run for as long as the other does: without
// events there is nothing to assign, and without sessions there is nobody to
// assign it to. Whichever stops first takes the process down with it.
let node_service = match arguments.mode {
Mode::Shadow => NodeSessionServer::shadow(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
repositories.clone(),
),
Mode::Active => NodeSessionServer::new(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
repositories,
),
};
tokio::select! {
served = axum::serve(events, http).into_future() => served?,
published = check_events.run() => published?,
served = Server::builder()
.add_service(GeneratedServer::new(node_service))
.add_service(GeneratedSecretsServer::new(secret_service))
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
.add_service(GeneratedActionsReadServer::new(actions_read))
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
}

Ok(())
}

async fn shutdown() {
if let Err(error) = tokio::signal::ctrl_c().await {
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
std::future::pending::<()>().await;
}
}
+48 -2
View File
@@ -1,67 +1,113 @@
use syncode_control_node::identity_wire::IssueWorkflowRepositoryTokenRequest;
use syncode_control_node::identity_wire::identity_client::IdentityClient;
use thiserror::Error;
use tonic::Request;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::transport::Channel;

use crate::secrets::RepositoryCredentialSource;

#[derive(Clone)]
pub struct IdentityRepositoryCredentials {
client: IdentityClient<Channel>,
authorization: MetadataValue<Ascii>,
}

#[derive(Debug, Error)]
pub enum IdentityRepositoryCredentialsError {
#[error("cannot authorize identity requests")]
Authorization,
#[error("cannot connect to identity: {0}")]
Connect(#[from] tonic::transport::Error),
#[error("identity rejected the workflow repository credential: {0}")]
Request(#[from] tonic::Status),
#[error("identity returned an empty workflow repository credential")]
Empty,
}

impl IdentityRepositoryCredentials {
pub async fn connect(
endpoint: String,
shared_secret: String,
) -> Result<Self, IdentityRepositoryCredentialsError> {
let client = IdentityClient::connect(endpoint).await?;
let authorization = format!("Bearer {shared_secret}")
.parse()
.map_err(|_| IdentityRepositoryCredentialsError::Authorization)?;
Ok(Self {
client,
authorization,
})
}
}

impl RepositoryCredentialSource for IdentityRepositoryCredentials {
type Error = IdentityRepositoryCredentialsError;

async fn issue(&self, user: &str, repository: &str) -> Result<String, Self::Error> {
let mut request = Request::new(IssueWorkflowRepositoryTokenRequest {
user_id: user.to_owned(),
repository_id: repository.to_owned(),
});
request
.metadata_mut()
.insert("authorization", self.authorization.clone());
let token = self
.client
.clone()
.issue_workflow_repository_token(request)
.await?
.into_inner()
.token;
if token.is_empty() {
return Err(IdentityRepositoryCredentialsError::Empty);
}
Ok(token)
}
}
use syncode_control_node::identity_wire::identity_client::IdentityClient;
use syncode_control_node::identity_wire::{
IssueWorkflowRepositoryTokenRequest, ResolveRepositoryRequest,
};
use thiserror::Error;
use tonic::Request;
use tonic::metadata::{Ascii, MetadataValue};
use tonic::transport::Channel;

use crate::secrets::RepositoryCredentialSource;

#[derive(Clone)]
pub struct IdentityRepositoryCredentials {
client: IdentityClient<Channel>,
authorization: MetadataValue<Ascii>,
}

#[derive(Debug, Error)]
pub enum IdentityRepositoryCredentialsError {
#[error("cannot authorize identity requests")]
Authorization,
#[error("cannot connect to identity: {0}")]
Connect(#[from] tonic::transport::Error),
#[error("identity rejected the workflow repository credential: {0}")]
Request(#[from] tonic::Status),
#[error("identity returned an empty workflow repository credential")]
Empty,
#[error("invalid repository coordinates {0:?}")]
InvalidRepository(String),
}

impl IdentityRepositoryCredentials {
pub async fn connect(
endpoint: String,
shared_secret: String,
) -> Result<Self, IdentityRepositoryCredentialsError> {
let client = IdentityClient::connect(endpoint).await?;
let authorization = format!("Bearer {shared_secret}")
.parse()
.map_err(|_| IdentityRepositoryCredentialsError::Authorization)?;
Ok(Self {
client,
authorization,
})
}
}

impl RepositoryCredentialSource for IdentityRepositoryCredentials {
type Error = IdentityRepositoryCredentialsError;

async fn issue(&self, user: &str, repository: &str) -> Result<String, Self::Error> {
let repository = self.repository_id(repository).await?;
let mut request = Request::new(IssueWorkflowRepositoryTokenRequest {
user_id: user.to_owned(),
repository_id: repository,
});
request
.metadata_mut()
.insert("authorization", self.authorization.clone());
let token = self
.client
.clone()
.issue_workflow_repository_token(request)
.await?
.into_inner()
.token;
if token.is_empty() {
return Err(IdentityRepositoryCredentialsError::Empty);
}
Ok(token)
}
}

impl IdentityRepositoryCredentials {
async fn repository_id(
&self,
repository: &str,
) -> Result<String, IdentityRepositoryCredentialsError> {
if uuid::Uuid::parse_str(repository).is_ok() {
return Ok(repository.to_owned());
}
let Some((owner, name)) = repository.split_once('/') else {
return Err(IdentityRepositoryCredentialsError::InvalidRepository(
repository.to_owned(),
));
};
if owner.is_empty() || name.is_empty() || name.contains('/') {
return Err(IdentityRepositoryCredentialsError::InvalidRepository(
repository.to_owned(),
));
}
let mut request = Request::new(ResolveRepositoryRequest {
owner: owner.to_owned(),
name: name.to_owned(),
});
request
.metadata_mut()
.insert("authorization", self.authorization.clone());
let repository_id = self
.client
.clone()
.resolve_repository(request)
.await?
.into_inner()
.repository_id;
if uuid::Uuid::parse_str(&repository_id).is_err() {
return Err(IdentityRepositoryCredentialsError::InvalidRepository(
repository_id,
));
}
Ok(repository_id)
}
}
+117
View File
@@ -1,182 +1,299 @@
use std::error::Error;
use std::future::Future;
use std::time::SystemTime;

use syncode_control_node::wire::{SecretRequest, SecretResponse};
use syncode_control_node::{CapabilityAuthority, REPOSITORY_TOKEN_SECRET, RuntimeSecretsService};
use syncode_control_nodes::{Lifecycle, NodeStore, Nodes};
use syncode_control_runs::{RunLog, Runs, SecretReadOutcome};
use syncode_workflow::SecretName;
use tonic::{Request, Response, Status};

pub trait SecretSource: Send + Sync + 'static {
type Error: Error + Send + Sync + 'static;

fn resolve(
&self,
repository: &str,
name: &str,
) -> impl Future<Output = Result<Option<String>, Self::Error>> + Send;
}

pub trait RepositoryCredentialSource: Send + Sync + 'static {
type Error: Error + Send + Sync + 'static;

fn issue(
&self,
user: &str,
repository: &str,
) -> impl Future<Output = Result<String, Self::Error>> + Send;
}

pub struct RuntimeSecrets<L, S, F, R> {
runs: Runs<L>,
nodes: Nodes<S>,
source: F,
repository_credentials: R,
authority: CapabilityAuthority,
}

impl<L, S, F, R> RuntimeSecrets<L, S, F, R> {
pub const fn new(
runs: Runs<L>,
nodes: Nodes<S>,
source: F,
repository_credentials: R,
authority: CapabilityAuthority,
) -> Self {
Self {
runs,
nodes,
source,
repository_credentials,
authority,
}
}
}

#[tonic::async_trait]
impl<L, S, F, R> RuntimeSecretsService for RuntimeSecrets<L, S, F, R>
where
L: RunLog + 'static,
S: NodeStore + 'static,
F: SecretSource,
R: RepositoryCredentialSource,
{
async fn resolve(
&self,
request: Request<SecretRequest>,
) -> Result<Response<SecretResponse>, Status> {
let request = request.into_inner();
let claims = self
.authority
.verify(&request.capability, SystemTime::now())
.map_err(|error| Status::unauthenticated(error.to_string()))?;
let name = request.name.to_ascii_uppercase();
if name.parse::<SecretName>().is_err() {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(Status::invalid_argument("invalid secret name"));
}
let origin = match self.authorize(claims, &name).await {
Ok(origin) => origin,
Err(error) => {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(error);
}
};
let value = if name == REPOSITORY_TOKEN_SECRET {
let user = match origin.principal() {
Some(user) => user,
None => {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(Status::permission_denied(
"workflow repository credentials require an originating user",
));
}
};
match self
.repository_credentials
.issue(user, origin.repository())
.await
{
Ok(value) => value,
Err(error) => {
self.audit(claims, name, SecretReadOutcome::SourceFailure)
.await?;
return Err(Status::unavailable(format!(
"repository credential source unavailable: {error}"
)));
}
}
} else {
match self.source.resolve(origin.repository(), &name).await {
Ok(Some(value)) => value,
Ok(None) => {
self.audit(claims, name, SecretReadOutcome::Missing).await?;
return Err(Status::not_found("secret not found"));
}
Err(error) => {
self.audit(claims, name, SecretReadOutcome::SourceFailure)
.await?;
return Err(Status::unavailable(format!(
"secret source unavailable: {error}"
)));
}
}
};
if let Err(error) = self.authorize(claims, &name).await {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(error);
}
self.audit(claims, name, SecretReadOutcome::Granted).await?;
Ok(Response::new(SecretResponse { value }))
}
}

impl<L: RunLog, S: NodeStore, F, R> RuntimeSecrets<L, S, F, R> {
async fn authorize(
&self,
claims: syncode_control_node::CapabilityClaims,
name: &str,
) -> Result<syncode_control_runs::Origin, Status> {
let origin = self
.runs
.authorize_secret(
claims.run(),
claims.job(),
claims.node(),
claims.fence(),
name,
)
.await
.map_err(|error| Status::permission_denied(error.to_string()))?;
if !origin.secrets_allowed() {
return Err(Status::permission_denied(
"secrets are denied for an untrusted pull request",
));
}
let lifecycle = self
.nodes
.lifecycle(claims.node())
.await
.map_err(|error| Status::permission_denied(error.to_string()))?;
if !matches!(lifecycle, Lifecycle::Active | Lifecycle::Draining) {
return Err(Status::permission_denied(format!(
"node is not allowed to read secrets while {lifecycle:?}"
)));
}
Ok(origin)
}

async fn audit(
&self,
claims: syncode_control_node::CapabilityClaims,
name: String,
outcome: SecretReadOutcome,
) -> Result<(), Status> {
self.runs
.audit_secret(claims.run(), claims.job(), claims.node(), name, outcome)
.await
.map_err(|error| Status::internal(format!("cannot audit secret read: {error}")))
}
}
use std::error::Error;
use std::future::Future;
use std::time::SystemTime;

use syncode_control_node::wire::{SecretRequest, SecretResponse};
use syncode_control_node::{CapabilityAuthority, REPOSITORY_TOKEN_SECRET, RuntimeSecretsService};
use syncode_control_nodes::{Lifecycle, NodeStore, Nodes};
use syncode_control_runs::{RunLog, Runs, SecretReadOutcome};
use syncode_workflow::SecretName;
use tonic::{Request, Response, Status};
use url::Url;

const GIT_CONFIG_SECRET: &str = "SYNCODE_GIT_CONFIG";

pub struct RuntimeGitConfig {
canonical_url: Url,
target_url: Url,
dependencies: Vec<String>,
}

impl RuntimeGitConfig {
pub const fn new(canonical_url: Url, target_url: Url, dependencies: Vec<String>) -> Self {
Self {
canonical_url,
target_url,
dependencies,
}
}
}

pub trait SecretSource: Send + Sync + 'static {
type Error: Error + Send + Sync + 'static;

fn resolve(
&self,
repository: &str,
name: &str,
) -> impl Future<Output = Result<Option<String>, Self::Error>> + Send;
}

pub trait RepositoryCredentialSource: Send + Sync + 'static {
type Error: Error + Send + Sync + 'static;

fn issue(
&self,
user: &str,
repository: &str,
) -> impl Future<Output = Result<String, Self::Error>> + Send;
}

pub struct RuntimeSecrets<L, S, F, R> {
runs: Runs<L>,
nodes: Nodes<S>,
source: F,
repository_credentials: R,
git: RuntimeGitConfig,
authority: CapabilityAuthority,
}

impl<L, S, F, R> RuntimeSecrets<L, S, F, R> {
pub const fn new(
runs: Runs<L>,
nodes: Nodes<S>,
source: F,
repository_credentials: R,
git: RuntimeGitConfig,
authority: CapabilityAuthority,
) -> Self {
Self {
runs,
nodes,
source,
repository_credentials,
git,
authority,
}
}
}

#[tonic::async_trait]
impl<L, S, F, R> RuntimeSecretsService for RuntimeSecrets<L, S, F, R>
where
L: RunLog + 'static,
S: NodeStore + 'static,
F: SecretSource,
R: RepositoryCredentialSource,
{
async fn resolve(
&self,
request: Request<SecretRequest>,
) -> Result<Response<SecretResponse>, Status> {
let request = request.into_inner();
let claims = self
.authority
.verify(&request.capability, SystemTime::now())
.map_err(|error| Status::unauthenticated(error.to_string()))?;
let name = request.name.to_ascii_uppercase();
if name.parse::<SecretName>().is_err() {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(Status::invalid_argument("invalid secret name"));
}
let origin = match self.authorize(claims, &name).await {
Ok(origin) => origin,
Err(error) => {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(error);
}
};
let value = if name == REPOSITORY_TOKEN_SECRET {
let user = match origin.principal() {
Some(user) => user,
None => {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(Status::permission_denied(
"workflow repository credentials require an originating user",
));
}
};
match self
.repository_credentials
.issue(user, origin.repository())
.await
{
Ok(value) => value,
Err(error) => {
self.audit(claims, name, SecretReadOutcome::SourceFailure)
.await?;
return Err(Status::unavailable(format!(
"repository credential source unavailable: {error}"
)));
}
}
} else if name == GIT_CONFIG_SECRET {
let user = match origin.principal() {
Some(user) => user,
None => {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(Status::permission_denied(
"workflow Git credentials require an originating user",
));
}
};
match self.git_config(user).await {
Ok(value) => value,
Err(error) => {
self.audit(claims, name, SecretReadOutcome::SourceFailure)
.await?;
return Err(error);
}
}
} else {
match self.source.resolve(origin.repository(), &name).await {
Ok(Some(value)) => value,
Ok(None) => {
self.audit(claims, name, SecretReadOutcome::Missing).await?;
return Err(Status::not_found("secret not found"));
}
Err(error) => {
self.audit(claims, name, SecretReadOutcome::SourceFailure)
.await?;
return Err(Status::unavailable(format!(
"secret source unavailable: {error}"
)));
}
}
};
if let Err(error) = self.authorize(claims, &name).await {
self.audit(claims, name, SecretReadOutcome::Denied).await?;
return Err(error);
}
self.audit(claims, name, SecretReadOutcome::Granted).await?;
Ok(Response::new(SecretResponse { value }))
}
}

impl<L, S, F, R> RuntimeSecrets<L, S, F, R>
where
R: RepositoryCredentialSource,
{
async fn git_config(&self, user: &str) -> Result<String, Status> {
let mut config = String::new();
for repository in &self.git.dependencies {
let token = self
.repository_credentials
.issue(user, repository)
.await
.map_err(|error| {
Status::unavailable(format!(
"repository credential source unavailable: {error}"
))
})?;
config.push_str(&git_config_entry(
&self.git.canonical_url,
&self.git.target_url,
repository,
&token,
)?);
}
if config.is_empty() {
return Err(Status::failed_precondition(
"workflow Git dependencies are not configured",
));
}
Ok(config)
}
}

fn git_config_entry(
canonical_base: &Url,
target_base: &Url,
repository: &str,
token: &str,
) -> Result<String, Status> {
let path = format!("{repository}.git");
let canonical = canonical_base
.join(&path)
.map_err(|_| Status::invalid_argument("invalid canonical Git dependency URL"))?;
let mut target = target_base
.join(&path)
.map_err(|_| Status::invalid_argument("invalid target Git dependency URL"))?;
target
.set_username("syn")
.map_err(|()| Status::invalid_argument("invalid target Git dependency URL"))?;
target
.set_password(Some(token))
.map_err(|()| Status::invalid_argument("invalid workflow repository credential"))?;
Ok(format!(
"[url \"\"{target}\"\"]\n\tinsteadOf = {canonical}\n"
))
}

impl<L: RunLog, S: NodeStore, F, R> RuntimeSecrets<L, S, F, R> {
async fn authorize(
&self,
claims: syncode_control_node::CapabilityClaims,
name: &str,
) -> Result<syncode_control_runs::Origin, Status> {
let origin = self
.runs
.authorize_secret(
claims.run(),
claims.job(),
claims.node(),
claims.fence(),
name,
)
.await
.map_err(|error| Status::permission_denied(error.to_string()))?;
if !origin.secrets_allowed() {
return Err(Status::permission_denied(
"secrets are denied for an untrusted pull request",
));
}
let lifecycle = self
.nodes
.lifecycle(claims.node())
.await
.map_err(|error| Status::permission_denied(error.to_string()))?;
if !matches!(lifecycle, Lifecycle::Active | Lifecycle::Draining) {
return Err(Status::permission_denied(format!(
"node is not allowed to read secrets while {lifecycle:?}"
)));
}
Ok(origin)
}

async fn audit(
&self,
claims: syncode_control_node::CapabilityClaims,
name: String,
outcome: SecretReadOutcome,
) -> Result<(), Status> {
self.runs
.audit_secret(claims.run(), claims.job(), claims.node(), name, outcome)
.await
.map_err(|error| Status::internal(format!("cannot audit secret read: {error}")))
}
}

#[cfg(test)]
mod tests {
use super::git_config_entry;

#[test]
fn git_config_rewrites_canonical_dependencies_to_the_current_instance()
-> Result<(), Box<dyn std::error::Error>> {
let actual = git_config_entry(
&"https://syncode.sh/".parse()?,
&"https://dev.syncode.sh/".parse()?,
"syncode/repo",
"syn_rat_token",
)?;
assert_eq!(
actual,
"[url \"\"https://syn:syn_rat_token@dev.syncode.sh/syncode/repo.git\"\"]\n\tinsteadOf = https://syncode.sh/syncode/repo.git\n"
);
Ok(())
}
}
+2
View File
@@ -1,173 +1,175 @@
#![allow(clippy::expect_used)]

#[path = "support/repository.rs"]
#[allow(dead_code)]
mod repository;

use std::error::Error;

use syncode_control::action_repository::NativeActionRepository;
use syncode_control::actions::ActionRepositoryPort;
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
use syncode_control::secrets::RepositoryCredentialSource;
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
use syncode_control_node::identity_wire::{
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
ValidateSessionRequest, ValidateSessionResponse,
};
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

#[derive(Default)]
struct FixtureIdentity;

#[tonic::async_trait]
impl Identity for FixtureIdentity {
async fn issue_workflow_repository_token(
&self,
request: Request<IssueWorkflowRepositoryTokenRequest>,
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
if request
.metadata()
.get("authorization")
.and_then(|value| value.to_str().ok())
!= Some("Bearer shared-secret")
{
return Err(Status::unauthenticated("missing authorization"));
}
let request = request.into_inner();
if request.user_id != "user-id" || request.repository_id != repository::REPOSITORY {
return Err(Status::invalid_argument("unexpected token scope"));
}
Ok(Response::new(IssueWorkflowRepositoryTokenResponse {
token: "workflow-repository-token".to_owned(),
expires_at_unix: 1,
}))
}

async fn get_repository_coordinates(
&self,
_request: Request<GetRepositoryCoordinatesRequest>,
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
Err(Status::unimplemented("get_repository_coordinates"))
}

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(),
}))
}
}

#[tokio::test]
async fn issues_a_scoped_repository_credential_over_identity_grpc() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let endpoint = format!("http://{}", listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(IdentityServer::new(FixtureIdentity))
.serve_with_incoming(TcpListenerStream::new(listener))
.await;
});
let credentials =
IdentityRepositoryCredentials::connect(endpoint, "shared-secret".to_owned()).await?;

let token = credentials.issue("user-id", repository::REPOSITORY).await?;

assert_eq!(token, "workflow-repository-token");
Ok(())
}

async fn source() -> TestResult<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(())
}
#![allow(clippy::expect_used)]

#[path = "support/repository.rs"]
#[allow(dead_code)]
mod repository;

use std::error::Error;

use syncode_control::action_repository::NativeActionRepository;
use syncode_control::actions::ActionRepositoryPort;
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
use syncode_control::secrets::RepositoryCredentialSource;
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
use syncode_control_node::identity_wire::{
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
ValidateSessionRequest, ValidateSessionResponse,
};
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

#[derive(Default)]
struct FixtureIdentity;

#[tonic::async_trait]
impl Identity for FixtureIdentity {
async fn issue_workflow_repository_token(
&self,
request: Request<IssueWorkflowRepositoryTokenRequest>,
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
if request
.metadata()
.get("authorization")
.and_then(|value| value.to_str().ok())
!= Some("Bearer shared-secret")
{
return Err(Status::unauthenticated("missing authorization"));
}
let request = request.into_inner();
if request.user_id != "user-id" || request.repository_id != repository::REPOSITORY {
return Err(Status::invalid_argument("unexpected token scope"));
}
Ok(Response::new(IssueWorkflowRepositoryTokenResponse {
token: "workflow-repository-token".to_owned(),
expires_at_unix: 1,
}))
}

async fn get_repository_coordinates(
&self,
_request: Request<GetRepositoryCoordinatesRequest>,
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
Err(Status::unimplemented("get_repository_coordinates"))
}

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(),
}))
}
}

#[tokio::test]
async fn issues_a_scoped_repository_credential_over_identity_grpc() -> TestResult {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let endpoint = format!("http://{}", listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(IdentityServer::new(FixtureIdentity))
.serve_with_incoming(TcpListenerStream::new(listener))
.await;
});
let credentials =
IdentityRepositoryCredentials::connect(endpoint, "shared-secret".to_owned()).await?;

let token = credentials.issue("user-id", repository::REPOSITORY).await?;

assert_eq!(token, "workflow-repository-token");
let token = credentials.issue("user-id", "actions/checkout").await?;
assert_eq!(token, "workflow-repository-token");
Ok(())
}

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(())
}
+10 -2
View File
@@ -1,108 +1,116 @@
#![allow(clippy::expect_used)]

use std::error::Error;

use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode, header};
use syncode_control::actions_read::{ActionsAuthorization, ActionsRead, AuthorizationError};
use syncode_control::actions_read_http;
use syncode_control_runs::{Forgotten, JobId, Origin, Runs};
use tower::ServiceExt;

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

const REPOSITORY_ID: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89214";

#[derive(Clone)]
struct Authorization;

#[tonic::async_trait]
impl ActionsAuthorization for Authorization {
async fn repository(
&self,
session_token: &str,
owner: &str,
name: &str,
) -> Result<String, AuthorizationError> {
if session_token != "valid-session" {
return Err(AuthorizationError::Unauthenticated);
}
if owner == "syncode" && name == "control" {
Ok(REPOSITORY_ID.to_owned())
} else {
Err(AuthorizationError::NotFound)
}
}

async fn authorize(
&self,
session_token: &str,
repository_id: &str,
) -> Result<(), AuthorizationError> {
if session_token == "valid-session" && repository_id == REPOSITORY_ID {
Ok(())
} else {
Err(AuthorizationError::Denied)
}
}
}

#[tokio::test]
async fn lists_runs_for_a_browser_identity_session() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let run = runs
.queue(
JobId::fresh(),
Origin::new(
REPOSITORY_ID.to_owned(),
"abc123".to_owned(),
"refs/heads/main".to_owned(),
"push".to_owned(),
".syncode/workflows/ci.yml".to_owned(),
),
b"plan".to_vec(),
)
.await?;
let router = actions_read_http::router(
ActionsRead::new(runs, Authorization),
vec!["https://new.dev.syncode.sh".to_owned()],
);

let response = router
.oneshot(
Request::builder()
.uri("/control/actions/repositories/syncode/control/runs")
.header(header::COOKIE, "syncode_identity_session=valid-session")
.header(header::ORIGIN, "https://new.dev.syncode.sh")
.body(Body::empty())?,
)
.await?;

assert_eq!(StatusCode::OK, response.status());
assert_eq!(
"https://new.dev.syncode.sh",
response.headers()[header::ACCESS_CONTROL_ALLOW_ORIGIN]
);
let body: serde_json::Value =
serde_json::from_slice(&to_bytes(response.into_body(), 4096).await?)?;
assert_eq!(run.to_string(), body[0]["id"]);
assert_eq!("abc123", body[0]["commit"]);
Ok(())
}

#[tokio::test]
async fn rejects_a_request_without_the_identity_cookie() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let router = actions_read_http::router(ActionsRead::new(runs, Authorization), Vec::new());

let response = router
.oneshot(
Request::builder()
.uri("/control/actions/repositories/syncode/control/runs")
.body(Body::empty())?,
)
.await?;

assert_eq!(StatusCode::UNAUTHORIZED, response.status());
Ok(())
}
#![allow(clippy::expect_used)]

use std::error::Error;

use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode, header};
use syncode_control::actions_read::{ActionsAuthorization, ActionsRead, AuthorizationError};
use syncode_control::actions_read_http;
use syncode_control_runs::{Forgotten, JobId, Origin, Runs};
use tower::ServiceExt;

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

const REPOSITORY_ID: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89214";

#[derive(Clone)]
struct Authorization;

#[tonic::async_trait]
impl ActionsAuthorization for Authorization {
async fn repository(
&self,
session_token: &str,
owner: &str,
name: &str,
) -> Result<String, AuthorizationError> {
if session_token != "valid-session" {
return Err(AuthorizationError::Unauthenticated);
}
if owner == "syncode" && name == "control" {
Ok(REPOSITORY_ID.to_owned())
} else {
Err(AuthorizationError::NotFound)
}
}

async fn authorize(
&self,
session_token: &str,
repository_id: &str,
) -> Result<(), AuthorizationError> {
if session_token == "valid-session" && repository_id == REPOSITORY_ID {
Ok(())
} else {
Err(AuthorizationError::Denied)
}
}
}

#[tokio::test]
async fn lists_runs_for_a_browser_identity_session() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let run = runs
.queue(
JobId::fresh(),
Origin::new(
REPOSITORY_ID.to_owned(),
"abc123".to_owned(),
"refs/heads/main".to_owned(),
"push".to_owned(),
".syncode/workflows/ci.yml".to_owned(),
),
b"plan".to_vec(),
)
.await?;
let router = actions_read_http::router(
ActionsRead::new(runs, Authorization),
vec!["https://new.dev.syncode.sh".to_owned()],
"syncode_dev_identity_session".to_owned(),
);

let response = router
.oneshot(
Request::builder()
.uri("/control/actions/repositories/syncode/control/runs")
.header(
header::COOKIE,
"syncode_identity_session=prod-session; syncode_dev_identity_session=valid-session",
)
.header(header::ORIGIN, "https://new.dev.syncode.sh")
.body(Body::empty())?,
)
.await?;

assert_eq!(StatusCode::OK, response.status());
assert_eq!(
"https://new.dev.syncode.sh",
response.headers()[header::ACCESS_CONTROL_ALLOW_ORIGIN]
);
let body: serde_json::Value =
serde_json::from_slice(&to_bytes(response.into_body(), 4096).await?)?;
assert_eq!(run.to_string(), body[0]["id"]);
assert_eq!("abc123", body[0]["commit"]);
Ok(())
}

#[tokio::test]
async fn rejects_a_request_without_the_identity_cookie() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let router = actions_read_http::router(
ActionsRead::new(runs, Authorization),
Vec::new(),
"syncode_dev_identity_session".to_owned(),
);

let response = router
.oneshot(
Request::builder()
.uri("/control/actions/repositories/syncode/control/runs")
.body(Body::empty())?,
)
.await?;

assert_eq!(StatusCode::UNAUTHORIZED, response.status());
Ok(())
}
+51
View File
@@ -1,90 +1,141 @@
#![allow(clippy::expect_used)]

use std::error::Error;

use syncode_control_node::RepositoryCoordinates;
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
use syncode_control_node::identity_wire::{
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
ValidateSessionRequest, ValidateSessionResponse,
};
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

struct FixtureIdentity;

#[tonic::async_trait]
impl Identity for FixtureIdentity {
async fn issue_workflow_repository_token(
&self,
_request: Request<IssueWorkflowRepositoryTokenRequest>,
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
Err(Status::unimplemented("issue_workflow_repository_token"))
}

async fn validate_session(
&self,
_request: Request<ValidateSessionRequest>,
) -> Result<Response<ValidateSessionResponse>, Status> {
Err(Status::unimplemented("validate_session"))
}

async fn check_capability(
&self,
_request: Request<CheckCapabilityRequest>,
) -> Result<Response<CheckCapabilityResponse>, Status> {
Err(Status::unimplemented("check_capability"))
}

async fn resolve_repository(
&self,
_request: Request<ResolveRepositoryRequest>,
) -> Result<Response<ResolveRepositoryResponse>, Status> {
Err(Status::unimplemented("resolve_repository"))
}

async fn get_repository_coordinates(
&self,
request: Request<GetRepositoryCoordinatesRequest>,
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
if request
.metadata()
.get("authorization")
.and_then(|value| value.to_str().ok())
!= Some("Bearer shared-secret")
{
return Err(Status::unauthenticated("missing authorization"));
}
Ok(Response::new(GetRepositoryCoordinatesResponse {
owner: "syncode".to_owned(),
name: "pipelines-demo".to_owned(),
}))
}
}

#[tokio::test]
async fn native_repository_ids_resolve_to_coordinates() -> TestResult {
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(IdentityServer::new(FixtureIdentity))
.serve_with_incoming(TcpListenerStream::new(identity_listener))
.await;
});

let client = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
assert_eq!(
client
.resolve("76128383-1df5-4979-9b13-c048a5287e9a")
.await?,
"syncode/pipelines-demo"
);
Ok(())
}
#![allow(clippy::expect_used)]

use std::error::Error;

use axum::Json;
use axum::http::{HeaderMap, StatusCode};
use axum::routing::get;
use serde_json::json;
use syncode_control::repository::RepositorySecrets;
use syncode_control::secrets::SecretSource;
use syncode_control_node::RepositoryCoordinates;
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
use syncode_control_node::identity_wire::{
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
ValidateSessionRequest, ValidateSessionResponse,
};
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

struct FixtureIdentity;

#[tonic::async_trait]
impl Identity for FixtureIdentity {
async fn issue_workflow_repository_token(
&self,
_request: Request<IssueWorkflowRepositoryTokenRequest>,
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
Err(Status::unimplemented("issue_workflow_repository_token"))
}

async fn validate_session(
&self,
_request: Request<ValidateSessionRequest>,
) -> Result<Response<ValidateSessionResponse>, Status> {
Err(Status::unimplemented("validate_session"))
}

async fn check_capability(
&self,
_request: Request<CheckCapabilityRequest>,
) -> Result<Response<CheckCapabilityResponse>, Status> {
Err(Status::unimplemented("check_capability"))
}

async fn resolve_repository(
&self,
_request: Request<ResolveRepositoryRequest>,
) -> Result<Response<ResolveRepositoryResponse>, Status> {
Err(Status::unimplemented("resolve_repository"))
}

async fn get_repository_coordinates(
&self,
request: Request<GetRepositoryCoordinatesRequest>,
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
if request
.metadata()
.get("authorization")
.and_then(|value| value.to_str().ok())
!= Some("Bearer shared-secret")
{
return Err(Status::unauthenticated("missing authorization"));
}
Ok(Response::new(GetRepositoryCoordinatesResponse {
owner: "syncode".to_owned(),
name: "pipelines-demo".to_owned(),
}))
}
}

#[tokio::test]
async fn native_repository_ids_resolve_to_coordinates() -> TestResult {
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(IdentityServer::new(FixtureIdentity))
.serve_with_incoming(TcpListenerStream::new(identity_listener))
.await;
});

let client = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
assert_eq!(
client
.resolve("76128383-1df5-4979-9b13-c048a5287e9a")
.await?,
"syncode/pipelines-demo"
);
Ok(())
}

#[tokio::test]
async fn native_repository_secrets_use_resolved_coordinates() -> TestResult {
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(IdentityServer::new(FixtureIdentity))
.serve_with_incoming(TcpListenerStream::new(identity_listener))
.await;
});

let secret_listener = TcpListener::bind("127.0.0.1:0").await?;
let secret_endpoint = format!("http://{}/", secret_listener.local_addr()?);
let app = axum::Router::new().route(
"/api/internal/actions/syncode/secrets/syncode/pipelines-demo/REGISTRY_TOKEN",
get(|headers: HeaderMap| async move {
if headers
.get("x-gitea-internal-auth")
.and_then(|value| value.to_str().ok())
!= Some("Bearer internal-token")
{
return Err(StatusCode::UNAUTHORIZED);
}
Ok(Json(json!({ "value": "registry-token" })))
}),
);
tokio::spawn(async move {
let _ = axum::serve(secret_listener, app).await;
});

let coordinates = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
let source = RepositorySecrets::new(
secret_endpoint.parse()?,
"internal-token".to_owned(),
coordinates,
);
assert_eq!(
source
.resolve("76128383-1df5-4979-9b13-c048a5287e9a", "REGISTRY_TOKEN")
.await?,
Some("registry-token".to_owned())
);
Ok(())
}
+30
View File
File diff suppressed because it is too large Load Diff
+133
View File
@@ -1,0 +1,133 @@
use serde::Deserialize;

use crate::events::DeliveryError;

const PROTOCOL_VERSION: &str = "1.0";
const SCHEMA_VERSION: u32 = 1;

pub(crate) struct PullRequestUpdate {
pub repository: String,
pub content_repository: String,
pub commit: String,
pub base_commit: String,
pub base_branch: String,
pub head_branch: String,
pub number: u64,
pub principal: String,
pub trusted: bool,
}

#[derive(Deserialize)]
struct Envelope {
message_id: String,
protocol_version: String,
schema_version: u32,
source_node_id: String,
repository_id: String,
message_type: String,
sequence: i64,
term: i64,
payload: Payload,
}

#[derive(Deserialize)]
struct Payload {
repository_id: String,
head_repository_id: String,
number: u64,
base_branch: String,
base_oid: String,
head_branch: String,
head_oid: String,
actor_id: String,
state: String,
}

pub(crate) fn read(body: &[u8]) -> Result<PullRequestUpdate, DeliveryError> {
let envelope: Envelope = serde_json::from_slice(body)?;
validate(&envelope)?;
if envelope.payload.state != "open" {
return Err(DeliveryError::UnactedPullRequest(envelope.payload.state));
}
if !valid_uuid(&envelope.payload.repository_id)
|| !valid_uuid(&envelope.payload.head_repository_id)
|| !valid_uuid(&envelope.payload.actor_id)
|| envelope.payload.number == 0
|| envelope.payload.base_branch.is_empty()
|| envelope.payload.head_branch.is_empty()
|| !valid_object_id(&envelope.payload.base_oid)
|| !valid_object_id(&envelope.payload.head_oid)
{
return Err(DeliveryError::InvalidNativeField(
"collaboration pull request",
));
}
Ok(PullRequestUpdate {
repository: envelope.payload.repository_id.clone(),
content_repository: envelope.payload.head_repository_id.clone(),
commit: envelope.payload.head_oid.clone(),
base_commit: envelope.payload.base_oid,
base_branch: envelope.payload.base_branch,
head_branch: envelope.payload.head_branch,
number: envelope.payload.number,
principal: envelope.payload.actor_id,
trusted: envelope.payload.repository_id == envelope.payload.head_repository_id,
})
}

fn validate(envelope: &Envelope) -> Result<(), DeliveryError> {
if envelope.protocol_version != PROTOCOL_VERSION {
return Err(DeliveryError::UnsupportedProtocol(
envelope.protocol_version.clone(),
));
}
if envelope.schema_version != SCHEMA_VERSION {
return Err(DeliveryError::UnsupportedSchema(envelope.schema_version));
}
if !matches!(
envelope.message_type.as_str(),
"collaboration.pull_request.created"
| "collaboration.pull_request.synchronized"
| "collaboration.pull_request.state_changed"
) {
return Err(DeliveryError::UnexpectedMessageType(
envelope.message_type.clone(),
));
}
if !valid_uuid(&envelope.message_id)
|| !valid_uuid(&envelope.source_node_id)
|| !valid_uuid(&envelope.repository_id)
|| envelope.sequence <= 0
|| envelope.term <= 0
{
return Err(DeliveryError::InvalidNativeField(
"collaboration event envelope",
));
}
Ok(())
}

fn valid_uuid(value: &str) -> bool {
uuid::Uuid::parse_str(value).is_ok()
}

fn valid_object_id(value: &str) -> bool {
matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}

#[cfg(test)]
mod tests {
use super::read;

#[test]
fn reads_an_open_native_pull_request() -> Result<(), crate::events::DeliveryError> {
let body = br#"{"message_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89211","protocol_version":"1.0","schema_version":1,"source_node_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89212","repository_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89214","message_type":"collaboration.pull_request.created","sequence":1,"term":1,"payload":{"repository_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89214","head_repository_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89214","number":3,"base_branch":"develop","base_oid":"1111111111111111111111111111111111111111","head_branch":"feature","head_oid":"2222222222222222222222222222222222222222","actor_id":"018f47e2-b2c4-7f19-8a6d-13ef76c89213","state":"open"}}"#;

let event = read(body)?;

assert_eq!(event.number, 3);
assert_eq!(event.head_branch, "feature");
assert!(event.trusted);
Ok(())
}
}