use std::str::FromStr;
use std::sync::Arc;
use alkcall::channels::operations::{
ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
};
use alkcall::core::auth::AuthContext;
use alkcall::registry::registration::OperationRegistry;
use alkcall::registry::spec::{
AccessControl, ChannelOpenSpec, ErrorDefinition, OperationSpec, OperationType, Visibility,
};
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::sync::Semaphore;
use alkcall::core::Connection;
use crate::adapter::SessionHandles;
use crate::authorize::AuthorizedRepo;
use crate::backend::{GitPackGen, GitPackIngest, GitRefs, GitRegistry};
use crate::duplex::{serve_duplex, serve_duplex_resolved};
use crate::limits::Limits;
use crate::registry::{validate_repo_id, AccessAction};
use crate::substrate::Service;
pub const OP_GIT_OPEN: &str = "channels/git/sub";
pub const GIT_ALPN: &str = "alk/git";
struct GitChannelPlan {
marker: AuthorizedRepo,
service: Service,
}
#[allow(clippy::too_many_arguments)]
pub fn register_openable(
core: &ChannelCore,
registry: Arc<dyn GitRegistry>,
refs: Arc<dyn GitRefs>,
packgen: Arc<dyn GitPackGen>,
ingest: Arc<dyn GitPackIngest>,
push_options_negotiated: bool,
limits: Limits,
pipeline: Arc<Semaphore>,
op_registry: &mut OperationRegistry,
auth: AuthContext,
) -> Result<(), String> {
let spec = git_open_spec();
let establisher = make_git_establisher(Arc::clone(®istry));
let open_handler = make_git_open_handler(
registry,
refs,
packgen,
ingest,
push_options_negotiated,
limits,
pipeline,
);
core.register_openable_with_establisher(
spec,
Some(establisher),
open_handler,
op_registry,
auth,
None,
)
}
pub fn git_open_spec() -> OperationSpec {
let open_failed_schema = json!({
"type": "object",
"properties": {
"reason": {
"type": "string",
"enum": ["dial_failed", "unknown_resource", "handler_error", "timeout"]
},
"message": { "type": "string" }
},
"required": ["reason", "message"]
});
OperationSpec::new(
OP_GIT_OPEN,
OperationType::Sub,
Visibility::External,
json!({
"type": "object",
"properties": {
"repo": { "type": "string" },
"service": {
"type": "string",
"enum": ["git-upload-pack", "git-receive-pack"]
}
},
"required": ["repo", "service"],
"additionalProperties": false
}),
json!({
"type": "object",
"properties": {
"channel_id": { "type": "integer", "minimum": 0 }
}
}),
vec![ErrorDefinition {
code: "channel:open_failed".to_string(),
description: "Establishment failed after channel allocation: the establisher \
rejected the params (malformed params, an unknown service — \
git-upload-archive is not in the enum — or a repo-id grammar \
violation) or the resolve→authorize gate denied the \
service-selected action. Details: { reason, message }. \
No channel_id is returned."
.to_string(),
schema: open_failed_schema,
http_status: None,
}],
AccessControl::default(),
None,
)
.with_description(
"Open a git session (alk/git). The open-op params {repo, service} select the \
repository and the git service; the establisher authorizes the service-selected \
action (read for fetch, write for push) before any channel data flows. The data \
plane is the pkt-line protocol session on the channel's duplex stream.",
)
.with_channel_open(ChannelOpenSpec::new(GIT_ALPN))
}
#[doc(hidden)]
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub struct OpenParams {
pub repo: String,
pub service: String,
}
#[doc(hidden)]
pub fn make_git_establisher(registry: Arc<dyn GitRegistry>) -> OpenEstablisher {
Arc::new(move |input: Value, auth: AuthContext| {
let registry = Arc::clone(®istry);
let identity = auth.identity.clone();
Box::pin(async move {
let params: OpenParams = match serde_json::from_value(input) {
Ok(params) => params,
Err(e) => {
return Err(EstablishmentError::HandlerError {
message: format!("malformed params: {e}"),
})
}
};
if let Err(cause) = validate_repo_id(¶ms.repo) {
return Err(EstablishmentError::HandlerError {
message: format!("invalid repo id: {cause}"),
});
}
let service = match Service::from_str(¶ms.service) {
Ok(service) => service,
Err(e) => {
return Err(EstablishmentError::HandlerError {
message: format!("malformed params: {e}"),
})
}
};
let action = if service.is_write() {
AccessAction::Write
} else {
AccessAction::Read
};
let marker = match registry.resolve(¶ms.repo).await {
Ok(record) => AuthorizedRepo::authorize(record, identity.as_ref(), action),
Err(_) => None,
};
let Some(marker) = marker else {
return Err(EstablishmentError::HandlerError {
message: format!("access denied: {}", params.repo),
});
};
Ok(Establishment::new(
Arc::new(GitChannelPlan { marker, service }) as ChannelPlan,
))
})
})
}
fn make_git_open_handler(
registry: Arc<dyn GitRegistry>,
refs: Arc<dyn GitRefs>,
packgen: Arc<dyn GitPackGen>,
ingest: Arc<dyn GitPackIngest>,
push_options_negotiated: bool,
limits: Limits,
pipeline: Arc<Semaphore>,
) -> OpenHandler {
Arc::new(
move |input: Value,
plan: Option<ChannelPlan>,
channel_conn: Connection,
auth: AuthContext| {
let handles = SessionHandles {
identity: auth.identity.clone(),
registry: Arc::clone(®istry),
refs: Arc::clone(&refs),
packgen: Arc::clone(&packgen),
ingest: Arc::clone(&ingest),
push_options_negotiated,
limits,
pipeline: Arc::clone(&pipeline),
};
tokio::spawn(async move {
let stream = match channel_conn.accept_bi().await {
Ok(stream) => stream,
Err(_) => return,
};
let established = plan
.as_ref()
.and_then(|plan| plan.downcast_ref::<GitChannelPlan>())
.map(|plan| (plan.marker.clone(), plan.service));
match established {
Some((marker, service)) => {
let _ = serve_duplex(
&marker,
stream,
handles.limits,
service,
Arc::clone(&handles.refs),
Arc::clone(&handles.packgen),
Arc::clone(&handles.ingest),
handles.push_options_negotiated,
Arc::clone(&handles.pipeline),
)
.await;
}
None => {
let registry = Arc::clone(&handles.registry);
let params = serde_json::from_value::<OpenParams>(input);
if let Ok(params) = params {
if let Ok(service) = Service::from_str(¶ms.service) {
let _ = serve_duplex_resolved(
¶ms.repo,
handles.identity.as_ref(),
service,
stream,
handles.limits,
registry,
Arc::clone(&handles.refs),
Arc::clone(&handles.packgen),
Arc::clone(&handles.ingest),
handles.push_options_negotiated,
Arc::clone(&handles.pipeline),
)
.await;
}
}
}
}
})
},
)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::authorize::authorize;
use crate::error::RegistryError;
use crate::registry::{RepoRecord, Visibility as RepoVisibility};
use alkcall::core::Identity;
use async_trait::async_trait;
use std::collections::{BTreeMap, HashMap};
#[derive(Default)]
struct TestRegistry {
rows: parking_lot::Mutex<BTreeMap<String, RepoRecord>>,
}
impl TestRegistry {
async fn put(&self, record: RepoRecord) {
self.rows.lock().insert(record.repo_id.clone(), record);
}
}
#[async_trait]
impl GitRegistry for TestRegistry {
async fn resolve(&self, repo_id: &str) -> Result<RepoRecord, RegistryError> {
match self.rows.lock().get(repo_id) {
Some(record) => Ok(record.clone()),
None => Err(RegistryError::NotFound),
}
}
}
fn private_record(id: &str) -> RepoRecord {
RepoRecord {
repo_id: id.to_string(),
visibility: RepoVisibility::Private,
grants: BTreeMap::new(),
}
}
fn identity(id: &str) -> Identity {
Identity {
id: id.to_string(),
scopes: Vec::new(),
resources: HashMap::new(),
}
}
fn auth_with(identity: Option<Identity>) -> AuthContext {
let mut auth = AuthContext::anonymous(b"alk/git");
auth.identity = identity;
auth
}
async fn establish(
registry: Arc<TestRegistry>,
input: Value,
identity: Option<Identity>,
) -> Result<Establishment, EstablishmentError> {
let establisher = make_git_establisher(registry);
establisher(input, auth_with(identity)).await
}
async fn granted_registry(grant: AccessAction) -> Arc<TestRegistry> {
let mut record = private_record("owner/doubled");
record.grants.insert("u-7731".to_owned(), vec![grant]);
let registry = Arc::new(TestRegistry::default());
registry.put(record).await;
registry
}
#[test]
fn git_open_spec_pins_the_channels_git_sub_operation() {
let spec = git_open_spec();
assert_eq!(spec.name, OP_GIT_OPEN);
assert_eq!(OP_GIT_OPEN, "channels/git/sub");
assert_eq!(spec.op_type, OperationType::Sub);
assert_eq!(spec.visibility, Visibility::External);
let marker = spec.channel_open.expect("channel_open marker set");
assert_eq!(marker.alpn, GIT_ALPN);
assert_eq!(GIT_ALPN, "alk/git");
}
#[test]
fn git_open_spec_pins_the_closed_repo_service_params_schema() {
let spec = git_open_spec();
let schema = &spec.input_schema;
assert_eq!(
schema["required"],
json!(["repo", "service"]),
"both params required"
);
assert_eq!(
schema["properties"]["service"]["enum"],
json!(["git-upload-pack", "git-receive-pack"]),
"the ADR-016 service enum"
);
assert_eq!(
schema["additionalProperties"],
json!(false),
"fail-closed extensions"
);
}
#[test]
fn git_open_spec_carries_an_empty_access_control() {
let spec = git_open_spec();
assert!(
!spec.access_control.has_restrictions(),
"no static scope gate: the establisher is the open-time ACL point"
);
}
#[test]
fn git_open_spec_declares_the_open_failed_error() {
let spec = git_open_spec();
let def = spec
.error_schemas
.iter()
.find(|e| e.code == "channel:open_failed")
.expect("channel:open_failed ErrorDefinition declared");
let reasons: Vec<&str> = def.schema["properties"]["reason"]["enum"]
.as_array()
.expect("reason enum")
.iter()
.filter_map(|v| v.as_str())
.collect();
assert_eq!(
reasons,
vec![
"dial_failed",
"unknown_resource",
"handler_error",
"timeout"
]
);
}
#[tokio::test]
async fn the_establisher_authorizes_the_service_selected_action() {
let registry = granted_registry(AccessAction::Read).await;
let fetch = establish(
registry.clone(),
json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
Some(identity("u-7731")),
)
.await;
assert!(fetch.is_ok(), "read grant satisfies fetch");
let push = establish(
registry,
json!({ "repo": "owner/doubled", "service": "git-receive-pack" }),
Some(identity("u-7731")),
)
.await;
assert!(
matches!(&push, Err(EstablishmentError::HandlerError { message })
if message == "access denied: owner/doubled"),
"a read grant does not satisfy push (write is identity + write-or-manage)"
);
let mut write_record = private_record("owner/doubled");
write_record
.grants
.insert("u-7731".to_owned(), vec![AccessAction::Write]);
let registry = Arc::new(TestRegistry::default());
registry.put(write_record).await;
for service in ["git-receive-pack", "git-upload-pack"] {
let open = establish(
registry.clone(),
json!({ "repo": "owner/doubled", "service": service }),
Some(identity("u-7731")),
)
.await;
assert!(open.is_ok(), "a write grant (⊇ read) satisfies {service}");
}
}
#[tokio::test]
async fn the_establisher_rejects_git_upload_archive_at_the_gate() {
let registry = Arc::new(TestRegistry::default());
let rejected = establish(
registry,
json!({ "repo": "owner/doubled", "service": "git-upload-archive" }),
Some(identity("u-7731")),
)
.await;
match &rejected {
Err(EstablishmentError::HandlerError { message }) => {
assert!(message.contains("git-upload-archive"), "{message}");
}
_ => panic!("git-upload-archive is not in the enum"),
}
}
#[tokio::test]
async fn the_establisher_rejects_malformed_params_and_grammar_violations() {
for (input, why) in [
(json!({ "service": "git-upload-pack" }), "missing repo"),
(json!({ "repo": "owner/doubled" }), "missing service"),
(
json!({ "repo": "owner/doubled", "service": "git-upload-pack", "cwd": 42 }),
"an extra field fails closed",
),
] {
let rejected = establish(Arc::new(TestRegistry::default()), input, None).await;
match &rejected {
Err(EstablishmentError::HandlerError { message }) => {
assert!(message.contains("malformed params"), "{why}: {message}");
}
_ => panic!("{why}: expected handler_error"),
}
}
let rejected = establish(
Arc::new(TestRegistry::default()),
json!({ "repo": "UPPER/x", "service": "git-upload-pack" }),
None,
)
.await;
match &rejected {
Err(EstablishmentError::HandlerError { message }) => {
assert!(message.contains("invalid repo id"), "{message}");
}
_ => panic!("the correction is the message"),
}
}
#[tokio::test]
async fn the_establisher_collapses_unknown_repo_and_unauthorized_into_one_denial() {
let registry = Arc::new(TestRegistry::default());
registry.put(private_record("owner/doubled")).await;
let holder = identity("grantless");
let unknown = match establish(
Arc::new(TestRegistry::default()),
json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
Some(holder.clone()),
)
.await
{
Err(e) => e,
Ok(_) => panic!("an unknown repo denies"),
};
let unauthorized = match establish(
registry,
json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
Some(holder),
)
.await
{
Err(e) => e,
Ok(_) => panic!("an unauthorized identity on a private repo denies"),
};
match (&unknown, &unauthorized) {
(EstablishmentError::HandlerError { .. }, EstablishmentError::HandlerError { .. }) => {
assert_eq!(unknown.reason(), unauthorized.reason(), "one reason");
assert_eq!(
unknown.message(),
unauthorized.message(),
"one generic message"
);
}
_other => panic!("both deny as the collapsed reason"),
}
assert!(unknown.message().contains("access denied"));
}
#[tokio::test]
async fn the_established_plan_crosses_marker_and_service() {
let registry = granted_registry(AccessAction::Write).await;
let identity = identity("u-7731");
for service in ["git-upload-pack", "git-receive-pack"] {
let establishment = establish(
registry.clone(),
json!({ "repo": "owner/doubled", "service": service }),
Some(identity.clone()),
)
.await
.expect("the granted identity's open establishes");
let plan = establishment
.plan
.as_ref()
.and_then(|plan| plan.downcast_ref::<GitChannelPlan>())
.expect("the plan is the GitChannelPlan slot");
assert_eq!(plan.service.as_str(), service);
assert_eq!(
plan.marker.action(),
if service == "git-receive-pack" {
AccessAction::Write
} else {
AccessAction::Read
}
);
assert_eq!(plan.marker.record().repo_id, "owner/doubled");
}
}
#[test]
fn the_checked_marker_agrees_with_the_policy_function() {
let mut record = private_record("owner/doubled");
record
.grants
.insert("u-7731".to_owned(), vec![AccessAction::Read]);
let holder = identity("u-7731");
assert!(authorize(&record, Some(&holder), AccessAction::Read));
let marker = AuthorizedRepo::authorize(record, Some(&holder), AccessAction::Read)
.expect("the passing check is a marker");
assert_eq!(marker.identity_id(), Some("u-7731"));
assert_eq!(marker.action(), AccessAction::Read);
}
}