use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::{Arc, Mutex as StdMutex, OnceLock};
use serde_json::{json, Value};
use tokio::sync::Mutex as AsyncMutex;
use khive_runtime::{NamespaceToken, RuntimeError};
use khive_storage::event::Event;
use khive_types::{EventKind, EventOutcome, SubstrateKind};
use crate::write_argv::{
build_add_argv, build_branch_argv, build_commit_argv, build_push_argv, reject_force,
validate_repo_path, GitArgError,
};
use crate::write_policy::{GitWritePolicy, GitWritePolicyError};
use crate::GitPack;
fn to_invalid_input(e: GitArgError) -> RuntimeError {
RuntimeError::InvalidInput(e.to_string())
}
fn to_policy_denied(e: GitWritePolicyError) -> RuntimeError {
RuntimeError::InvalidInput(e.to_string())
}
static REPO_LOCKS: OnceLock<StdMutex<HashMap<PathBuf, Arc<AsyncMutex<()>>>>> = OnceLock::new();
fn repo_write_lock(repo: &Path) -> Arc<AsyncMutex<()>> {
let key = std::fs::canonicalize(repo).unwrap_or_else(|_| repo.to_path_buf());
let registry = REPO_LOCKS.get_or_init(|| StdMutex::new(HashMap::new()));
let mut guard = registry.lock().unwrap_or_else(|e| e.into_inner());
guard
.entry(key)
.or_insert_with(|| Arc::new(AsyncMutex::new(())))
.clone()
}
fn parse_repo_param(params: &Value) -> Result<PathBuf, RuntimeError> {
match params.get("repo") {
Some(Value::String(raw)) => Ok(PathBuf::from(raw)),
None | Some(Value::Null) => Err(RuntimeError::InvalidInput("repo is required".into())),
Some(other) => Err(RuntimeError::InvalidInput(format!(
"repo must be a string, got {other:?}"
))),
}
}
fn parse_optional_string<'a>(
params: &'a Value,
name: &str,
) -> Result<Option<&'a str>, RuntimeError> {
match params.get(name) {
None | Some(Value::Null) => Ok(None),
Some(Value::String(value)) => Ok(Some(value)),
Some(other) => Err(RuntimeError::InvalidInput(format!(
"{name} must be a string, got {other:?}"
))),
}
}
fn parse_paths_param(params: &Value) -> Result<Vec<String>, RuntimeError> {
match params.get("paths") {
None | Some(Value::Null) => Ok(Vec::new()),
Some(Value::Array(arr)) => arr
.iter()
.map(|v| {
v.as_str().map(str::to_string).ok_or_else(|| {
RuntimeError::InvalidInput("paths entries must be strings".into())
})
})
.collect(),
Some(other) => Err(RuntimeError::InvalidInput(format!(
"paths must be an array of strings, got {other:?}"
))),
}
}
fn parse_force_param(params: &Value) -> Result<Option<bool>, RuntimeError> {
match params.get("force") {
None | Some(Value::Null) => Ok(None),
Some(Value::Bool(b)) => Ok(Some(*b)),
Some(other) => Err(RuntimeError::InvalidInput(format!(
"force must be a boolean, got {other:?}; force-push is never permitted through this verb"
))),
}
}
fn run_git(repo: &Path, argv: &[String]) -> Result<String, RuntimeError> {
let mut command = Command::new("git");
command
.arg("-c")
.arg("core.hooksPath=/dev/null")
.arg("-C")
.arg(repo)
.args(argv);
#[cfg(test)]
command
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null");
let output = command
.output()
.map_err(|e| RuntimeError::InvalidInput(format!("spawning git {argv:?}: {e}")))?;
if !output.status.success() {
return Err(RuntimeError::InvalidInput(format!(
"git {argv:?} failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
)));
}
Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}
fn current_branch(repo: &Path) -> Result<String, RuntimeError> {
let out = run_git(
repo,
&[
"symbolic-ref".to_string(),
"--short".to_string(),
"HEAD".to_string(),
],
)?;
Ok(out.trim().to_string())
}
struct WritePreflightError {
error: RuntimeError,
branch: Option<String>,
outcome: EventOutcome,
}
impl WritePreflightError {
fn denied(error: RuntimeError, branch: Option<&str>) -> Self {
Self {
error,
branch: branch.map(str::to_string),
outcome: EventOutcome::Denied,
}
}
fn runtime(error: RuntimeError) -> Self {
Self {
error,
branch: None,
outcome: EventOutcome::Error,
}
}
}
struct CommitPreflight {
branch: String,
add_argv: Option<Vec<String>>,
commit_argv: Vec<String>,
}
fn prepare_commit(repo: &Path, params: &Value) -> Result<CommitPreflight, WritePreflightError> {
validate_repo_path(repo)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, None))?;
let branch = current_branch(repo).map_err(WritePreflightError::runtime)?;
let message = params
.get("message")
.and_then(Value::as_str)
.ok_or_else(|| RuntimeError::InvalidInput("git.commit requires message".into()))
.map_err(|e| WritePreflightError::denied(e, Some(&branch)))?;
let paths =
parse_paths_param(params).map_err(|e| WritePreflightError::denied(e, Some(&branch)))?;
let author = parse_optional_string(params, "author")
.map_err(|e| WritePreflightError::denied(e, Some(&branch)))?;
let add_argv = if paths.is_empty() {
None
} else {
Some(
build_add_argv(&paths)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, Some(&branch)))?,
)
};
let commit_argv = build_commit_argv(message, &paths, author)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, Some(&branch)))?;
Ok(CommitPreflight {
branch,
add_argv,
commit_argv,
})
}
struct BranchPreflight {
name: String,
from: Option<String>,
argv: Vec<String>,
}
fn prepare_branch(repo: &Path, params: &Value) -> Result<BranchPreflight, WritePreflightError> {
validate_repo_path(repo)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, None))?;
let name = params
.get("name")
.and_then(Value::as_str)
.ok_or_else(|| RuntimeError::InvalidInput("git.branch requires name".into()))
.map_err(|e| WritePreflightError::denied(e, None))?;
let from = parse_optional_string(params, "from")
.map_err(|e| WritePreflightError::denied(e, Some(name)))?;
let argv = build_branch_argv(name, from)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, Some(name)))?;
Ok(BranchPreflight {
name: name.to_string(),
from: from.map(str::to_string),
argv,
})
}
struct PushPreflight {
branch: String,
remote: String,
argv: Vec<String>,
}
fn prepare_push(repo: &Path, params: &Value) -> Result<PushPreflight, WritePreflightError> {
validate_repo_path(repo)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, None))?;
let branch = params
.get("branch")
.and_then(Value::as_str)
.ok_or_else(|| RuntimeError::InvalidInput("git.push requires branch".into()))
.map_err(|e| WritePreflightError::denied(e, None))?;
let remote = parse_optional_string(params, "remote")
.map_err(|e| WritePreflightError::denied(e, Some(branch)))?
.unwrap_or("origin");
let force =
parse_force_param(params).map_err(|e| WritePreflightError::denied(e, Some(branch)))?;
reject_force(force)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, Some(branch)))?;
let argv = build_push_argv(remote, branch)
.map_err(to_invalid_input)
.map_err(|e| WritePreflightError::denied(e, Some(branch)))?;
Ok(PushPreflight {
branch: branch.to_string(),
remote: remote.to_string(),
argv,
})
}
impl GitPack {
fn enforce_write_policy(&self, repo: &Path, branch: &str) -> Result<PathBuf, RuntimeError> {
let policy = GitWritePolicy::from_config(&self.runtime().config().git_write);
policy.check(repo, branch).map_err(to_policy_denied)
}
async fn audit_early_failure(
&self,
token: &NamespaceToken,
verb: &str,
repo: &Path,
branch: Option<&str>,
outcome: EventOutcome,
error: RuntimeError,
) -> RuntimeError {
self.emit_write_audit(token, verb, repo, branch, "deny", outcome, None)
.await;
error
}
async fn parse_audited_repo(
&self,
token: &NamespaceToken,
verb: &str,
params: &Value,
) -> Result<PathBuf, RuntimeError> {
match parse_repo_param(params) {
Ok(repo) => Ok(repo),
Err(error) => Err(self
.audit_early_failure(
token,
verb,
Path::new("<invalid-repo>"),
None,
EventOutcome::Denied,
error,
)
.await),
}
}
pub(crate) async fn handle_commit(
&self,
token: &NamespaceToken,
params: Value,
) -> Result<Value, RuntimeError> {
let repo = self
.parse_audited_repo(token, "git.commit", ¶ms)
.await?;
let lock = repo_write_lock(&repo);
let _guard = lock.lock().await;
let CommitPreflight {
branch,
add_argv,
commit_argv,
} = match prepare_commit(&repo, ¶ms) {
Ok(preflight) => preflight,
Err(failure) => {
return Err(self
.audit_early_failure(
token,
"git.commit",
&repo,
failure.branch.as_deref(),
failure.outcome,
failure.error,
)
.await)
}
};
let canonical_repo = match self.enforce_write_policy(&repo, &branch) {
Ok(p) => p,
Err(e) => {
self.emit_write_audit(
token,
"git.commit",
&repo,
Some(&branch),
"deny",
EventOutcome::Denied,
None,
)
.await;
return Err(e);
}
};
let exec: Result<String, RuntimeError> = (|| {
if let Some(add_argv) = &add_argv {
run_git(&canonical_repo, add_argv)?;
}
run_git(&canonical_repo, &commit_argv)?;
let sha = run_git(
&canonical_repo,
&["rev-parse".to_string(), "HEAD".to_string()],
)?
.trim()
.to_string();
Ok(sha)
})();
match exec {
Ok(sha) => {
self.emit_write_audit(
token,
"git.commit",
&canonical_repo,
Some(&branch),
"allow",
EventOutcome::Success,
Some(&sha),
)
.await;
Ok(json!({
"repo": canonical_repo.display().to_string(),
"sha": sha,
}))
}
Err(e) => {
self.emit_write_audit(
token,
"git.commit",
&canonical_repo,
Some(&branch),
"allow",
EventOutcome::Error,
None,
)
.await;
Err(e)
}
}
}
pub(crate) async fn handle_branch(
&self,
token: &NamespaceToken,
params: Value,
) -> Result<Value, RuntimeError> {
let repo = self
.parse_audited_repo(token, "git.branch", ¶ms)
.await?;
let lock = repo_write_lock(&repo);
let _guard = lock.lock().await;
let BranchPreflight { name, from, argv } = match prepare_branch(&repo, ¶ms) {
Ok(preflight) => preflight,
Err(failure) => {
return Err(self
.audit_early_failure(
token,
"git.branch",
&repo,
failure.branch.as_deref(),
failure.outcome,
failure.error,
)
.await)
}
};
let canonical_repo = match self.enforce_write_policy(&repo, &name) {
Ok(p) => p,
Err(e) => {
self.emit_write_audit(
token,
"git.branch",
&repo,
Some(&name),
"deny",
EventOutcome::Denied,
None,
)
.await;
return Err(e);
}
};
match run_git(&canonical_repo, &argv) {
Ok(_) => {
self.emit_write_audit(
token,
"git.branch",
&canonical_repo,
Some(&name),
"allow",
EventOutcome::Success,
None,
)
.await;
Ok(json!({
"repo": canonical_repo.display().to_string(),
"name": name,
"from": from,
}))
}
Err(e) => {
self.emit_write_audit(
token,
"git.branch",
&canonical_repo,
Some(&name),
"allow",
EventOutcome::Error,
None,
)
.await;
Err(e)
}
}
}
pub(crate) async fn handle_push(
&self,
token: &NamespaceToken,
params: Value,
) -> Result<Value, RuntimeError> {
let repo = self.parse_audited_repo(token, "git.push", ¶ms).await?;
let lock = repo_write_lock(&repo);
let _guard = lock.lock().await;
let PushPreflight {
branch,
remote,
argv,
} = match prepare_push(&repo, ¶ms) {
Ok(preflight) => preflight,
Err(failure) => {
return Err(self
.audit_early_failure(
token,
"git.push",
&repo,
failure.branch.as_deref(),
failure.outcome,
failure.error,
)
.await)
}
};
let canonical_repo = match self.enforce_write_policy(&repo, &branch) {
Ok(p) => p,
Err(e) => {
self.emit_write_audit(
token,
"git.push",
&repo,
Some(&branch),
"deny",
EventOutcome::Denied,
None,
)
.await;
return Err(e);
}
};
match run_git(&canonical_repo, &argv) {
Ok(_) => {
self.emit_write_audit(
token,
"git.push",
&canonical_repo,
Some(&branch),
"allow",
EventOutcome::Success,
None,
)
.await;
Ok(json!({
"repo": canonical_repo.display().to_string(),
"remote": remote,
"branch": branch,
}))
}
Err(e) => {
self.emit_write_audit(
token,
"git.push",
&canonical_repo,
Some(&branch),
"allow",
EventOutcome::Error,
None,
)
.await;
Err(e)
}
}
}
#[allow(clippy::too_many_arguments)]
async fn emit_write_audit(
&self,
token: &NamespaceToken,
verb: &str,
repo: &Path,
branch: Option<&str>,
decision: &str,
outcome: EventOutcome,
sha: Option<&str>,
) {
let Ok(store) = self.runtime().events(token) else {
return;
};
let mut payload = json!({
"repo": repo.display().to_string(),
"decision": decision,
});
if let Some(b) = branch {
payload["branch"] = json!(b);
}
if let Some(s) = sha {
payload["sha"] = json!(s);
}
let event = Event::new(
token.namespace().as_str(),
verb,
EventKind::Audit,
SubstrateKind::Event,
token.actor().id.clone(),
)
.with_outcome(outcome)
.with_payload(payload);
if let Err(e) = store.append_event(event).await {
tracing::warn!(
verb,
error = %e,
"git write audit event store write failed (non-fatal)"
);
}
}
}