use std::sync::Arc;
use gunnar_core::{Grant, HashKind, Level, PrincipalKind, RepoRef, Scope, UnixMillis, Visibility};
use gunnar_grpc::seam::ControlCounts;
use gunnar_grpc::MergedRef;
use gunnar_grpc::{ControlError, ControlPlane, ControlResult, NewRepo as ProtoNewRepo};
use gunnar_proto::{
BranchProtection, Commit, FileChange, GrantLevel, MirrorConfig, Namespace, Principal,
PublicKey, PullRequest, PullRequestState, ReadLfsObjectResponse, ReadObjectResponse,
RefNamePolicy, RefTarget, Repo, RepoPath, RepoStorage, Signature, StorageInfoResponse,
};
use gunnar_store::git_storage_trait::{GitServe, ObjType, RefRow};
use gunnar_store::object;
use gunnar_wire::RefSnapshot;
use crate::storage::OpenRepo;
use crate::stores::ServerStores;
#[cfg(feature = "replication")]
const ABDICATE_FLUSH_SECS: u64 = 30;
#[cfg(feature = "replication")]
fn abdicate_flush_secs() -> u64 {
resolve_flush_secs(std::env::var("GUNNAR_ABDICATE_FLUSH_SECS").ok().as_deref())
}
#[cfg(feature = "replication")]
fn resolve_flush_secs(raw: Option<&str>) -> u64 {
raw.and_then(|v| v.trim().parse::<u64>().ok())
.filter(|&s| s > 0)
.unwrap_or(ABDICATE_FLUSH_SECS)
}
mod sso_plane;
fn log_read_refusal(e: gunnar_core::txlog::Error) -> ControlError {
match e {
gunnar_core::txlog::Error::Released(_) => ControlError::Unavailable(format!(
"{} this box's data set is being grown (unmounted): its transaction log is back after the remount — ask again",
gunnar_grpc::PRIMARY_GROWING
)),
gunnar_core::txlog::Error::NotRetained { seq, first } => ControlError::FailedPrecondition(format!(
"seq {seq} is no longer retained; the oldest is {first} — a snapshot is needed, not a range"
)),
gunnar_core::txlog::Error::BeyondHead { seq, head } => ControlError::InvalidArgument(format!("seq {seq} is beyond the head {head}")),
other => ControlError::Internal(format!("transaction log: {other}")),
}
}
#[allow(clippy::type_complexity)] struct ChunkOut<'s> {
sink: &'s mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
open: Option<OutEntry>,
}
struct OutEntry {
seq: u64,
kind: u32,
meta: Option<Vec<u8>>,
blob_len: u64,
at: u64,
crc: Option<u32>,
ended: bool,
}
impl ChunkOut<'_> {
fn send(&mut self, blob: &[u8]) -> anyhow::Result<()> {
let Some(o) = self.open.as_mut() else {
anyhow::bail!("a piece outside an entry");
};
let offset = o.at;
o.at += blob.len() as u64;
anyhow::ensure!(
o.at <= o.blob_len,
"seq {}: more than its {} blob bytes",
o.seq,
o.blob_len
);
let last = o.at == o.blob_len;
o.ended |= last;
let chunk = gunnar_proto::LogChunk {
seq: o.seq,
kind: o.kind,
meta: o.meta.take().unwrap_or_default(),
blob_len: o.blob_len,
blob: blob.to_vec(),
last,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: Some(offset),
entry_crc: if last { o.crc } else { None },
snapshot_item: None,
snapshot_id: String::new(),
};
(self.sink)(chunk).map_err(anyhow::Error::from)
}
}
impl crate::twin::FrameSink for ChunkOut<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
self.begin_at(seq, kind, meta, blob_len, 0)
}
fn begin_at(
&mut self,
seq: u64,
kind: u16,
meta: &[u8],
blob_len: u64,
offset: u64,
) -> anyhow::Result<()> {
anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
anyhow::ensure!(
offset <= blob_len,
"seq {seq}: from byte {offset} of a {blob_len}-byte blob"
);
self.open = Some(OutEntry {
seq,
kind: u32::from(kind),
meta: Some(meta.to_vec()),
blob_len,
at: offset,
crc: None,
ended: false,
});
Ok(())
}
fn checked(&mut self, payload_crc: u32) -> anyhow::Result<()> {
if let Some(o) = self.open.as_mut() {
o.crc = Some(payload_crc);
}
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
self.send(bytes)
}
fn end(&mut self) -> anyhow::Result<()> {
if matches!(&self.open, Some(o) if !o.ended) {
self.send(&[])?;
}
let Some(o) = self.open.take() else {
anyhow::bail!("an entry ended that never began");
};
anyhow::ensure!(
o.at == o.blob_len && o.ended,
"seq {}: {} of {} blob bytes went out",
o.seq,
o.at,
o.blob_len
);
Ok(())
}
}
impl CoreControlPlane {
fn readable_log(&self) -> ControlResult<&gunnar_core::txlog::Log> {
if crate::twin::maintenance::is_held(self.stores.root()) {
return Err(ControlError::Unavailable(
"this box's data set is held for its own growth (unmounted): its transaction log is not readable until the growth mounts it back".into(),
));
}
if !self.stores.is_twin() {
if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
return Err(ControlError::FailedPrecondition(format!(
"{why}. Nothing is read from it"
)));
}
}
let log: &gunnar_core::txlog::Log = self.stores.txlog();
if log.is_released() {
return Err(log_read_refusal(gunnar_core::txlog::Error::Released(
log.dir().to_path_buf(),
)));
}
if self.stores.is_twin() {
if let Some(why) = crate::twin::snapshot::not_whole(self.stores.root()) {
return Err(ControlError::FailedPrecondition(format!(
"{why}. Nothing is read from it"
)));
}
}
Ok(log)
}
}
pub struct CoreControlPlane {
stores: Arc<ServerStores>,
grow: Option<Arc<crate::grow::GrowMachine>>,
twin_role: bool,
git_listener: Option<crate::readiness::GitListener>,
admission: Option<Arc<gunnar_ssh::Admission>>,
replication: crate::replication_view::ReplicationView,
#[cfg(feature = "replication")]
replication_slot: Option<Arc<crate::mirror::ReplicationSlot>>,
#[cfg(feature = "webhooks")]
webhook_slot: Option<Arc<crate::webhook::WebhookSlot>>,
#[cfg(feature = "metered")]
entitlement: Option<Arc<crate::entitlement::EntitlementSlot>>,
headroom: Option<Arc<gunnar_ssh::DiskHeadroomGate>>,
max_object_bytes: u64,
gc: crate::gc_queue::GcQueue,
purge: crate::purge_queue::PurgeQueue,
mirror: crate::mirror_queue::MirrorQueue,
sso_import: crate::sso_import_queue::SsoImportQueue,
bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
restore_keys: Vec<String>,
}
impl CoreControlPlane {
pub fn new(stores: Arc<ServerStores>) -> Self {
let for_gc = Arc::clone(&stores);
let bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>> = Arc::default();
CoreControlPlane {
stores,
grow: None,
twin_role: false,
git_listener: None,
admission: None,
replication: crate::replication_view::ReplicationView::new(),
#[cfg(feature = "replication")]
replication_slot: None,
#[cfg(feature = "webhooks")]
webhook_slot: None,
#[cfg(feature = "metered")]
entitlement: None,
headroom: None,
max_object_bytes: DEFAULT_MAX_OBJECT_BYTES,
gc: Self::gc_queue(Arc::clone(&for_gc)),
purge: Self::purge_queue(Arc::clone(&for_gc)),
mirror: Self::mirror_queue(Arc::clone(&for_gc), Arc::clone(&bus)),
sso_import: crate::sso_import_queue::SsoImportQueue::new(
crate::sso_import_queue::SsoImport,
),
bus,
restore_keys: Vec::new(),
}
}
pub fn with_restore_keys(mut self, keys: Vec<String>) -> Self {
self.restore_keys = keys;
self
}
pub fn with_bus(self, bus: Arc<gunnar_grpc::EventBus>) -> Self {
let _ = self.bus.set(bus);
self
}
pub fn with_twin_role(mut self, twin_role: bool) -> Self {
self.twin_role = twin_role;
self
}
pub fn with_git_listener(mut self, listener: crate::readiness::GitListener) -> Self {
self.git_listener = Some(listener);
self
}
pub fn release_stale_purge_leases(stores: &ServerStores) -> u64 {
let gates = match stores.auth().gates() {
Ok(gates) => gates,
Err(err) => {
tracing::error!(
target: "gunnar::audit",
error = %err,
"the account gates could not be read at start-up; any purge lease left by a \
crash will stay in force until it expires"
);
return 0;
}
};
let mut released = 0;
for gate in gates {
if gate.reason != gunnar_core::GateReason::Purge {
continue;
}
match stores.auth().lift_gate_held_by(&gate.account, &gate.holder) {
Ok(Some(_)) => {
released += 1;
tracing::warn!(
target: "gunnar::audit",
event = "account.stale_lease_released",
account = %gate.account,
job = %gate.holder,
"an account was still locked by a purge that did not finish; the lockout \
has been lifted and the account may still hold bytes the purge did not \
reclaim — `repo list-orphans` shows them"
);
}
Ok(None) => {}
Err(err) => tracing::error!(
target: "gunnar::audit",
account = %gate.account,
error = %err,
"a stale purge lease could not be lifted at start-up"
),
}
}
released
}
fn purge_queue(stores: Arc<ServerStores>) -> crate::purge_queue::PurgeQueue {
let for_release = Arc::clone(&stores);
crate::purge_queue::PurgeQueue::new(
Arc::new(
move |order: &crate::purge_queue::PurgeOrder,
progress: &crate::purge_queue::Progress| {
let request = crate::account_purge::PurgeRequest {
account: order.account.clone(),
principal_id: order.principal_id.clone(),
reason: order.reason.clone(),
remove_principal: order.remove_principal,
expect_empty: order.expect_empty,
};
let auth = stores.auth();
let account = order.account.clone();
let job_id = order.job_id.clone();
let mut step = |step: crate::account_purge::Step| match step {
crate::account_purge::Step::Found(stores_found) => {
progress.found(stores_found)
}
crate::account_purge::Step::Reclaimed { bytes, .. } => {
progress.reclaimed(bytes);
let until = gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS;
if let Err(err) = auth.renew_gate(&account, &job_id, until) {
tracing::warn!(
target: "gunnar::audit",
account = %account,
job = %job_id,
error = %err,
"a purge could not renew its lockout lease; the account is \
still shut and the walk continues"
);
}
}
crate::account_purge::Step::Failed(note) => progress.failed(note),
crate::account_purge::Step::PrincipalRemoved => {
progress.principal_removed()
}
};
let report = crate::account_purge::purge_account(&stores, request, &mut step)
.map_err(|err| format!("{err:#}"))?;
if report.failures.is_empty() {
Ok(())
} else {
Err(format!(
"{} of {} stores could not be reclaimed and are still on the disk; \
`repo list-orphans` shows them",
report.failures.len(),
report.stores_found
))
}
},
),
Arc::new(move |account: &str, job_id: &str| {
for_release
.auth()
.lift_gate_held_by(account, job_id)
.map(|lifted| lifted.is_some())
.map_err(|err| format!("{err}"))
}),
)
}
fn gc_queue(stores: Arc<ServerStores>) -> crate::gc_queue::GcQueue {
crate::gc_queue::GcQueue::new(Arc::new(move |path: &str| {
let repo_ref = gunnar_core::resolve_repo_path(path)
.map_err(|err| format!("`{path}` is not a valid <namespace>/<name>: {err:#}"))?;
let repo = stores
.catalog()
.get(&repo_ref)
.map_err(|err| format!("reading the catalog row for {path}: {err:#}"))?
.ok_or_else(|| format!("{path} is no longer a live repository"))?;
let report = crate::admin::gc_one(&stores, &repo)
.map_err(|err| format!("maintaining {path}: {err:#}"))?;
Ok(crate::admin::reclaimed_bytes(&report))
}))
}
fn mirror_queue(
stores: Arc<ServerStores>,
bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
) -> crate::mirror_queue::MirrorQueue {
crate::mirror_queue::MirrorQueue::new(Arc::new(
move |order: &crate::mirror_queue::MirrorOrder,
report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>| {
let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
format!(
"`{}` is not a valid <namespace>/<name>: {err:#}",
order.path
)
})?;
let first = first_sync_of(
stores
.mirrors()
.mark_started(&repo_ref, gunnar_grpc::bus::now_unix_ms()),
&order.path,
)?;
let outcome = read_credential(&stores, &repo_ref).and_then(|credential| {
crate::mirror_fetch::sync_once(&stores, order, &credential, report)
});
audit_mirror_outcome(
&stores,
&repo_ref,
order,
first,
outcome.as_ref().err().map(String::as_str),
report,
);
let error = outcome.as_ref().err().map(String::as_str);
if let Err(err) =
stores
.mirrors()
.record_sync(&repo_ref, gunnar_grpc::bus::now_unix_ms(), error)
{
tracing::warn!(
target: "gunnar::git",
repo = %order.path,
%err,
"a mirror sync finished and its registration could not be updated; the \
fetch itself is unaffected"
);
}
if outcome.is_ok() {
if let Some(bus) = bus.get() {
let job = report.snapshot();
let updates = job
.map(|job| job.report.updates)
.unwrap_or_default()
.into_iter()
.map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
ref_name,
old_oid,
new_oid,
})
.collect::<Vec<_>>();
bus.publish(gunnar_proto::event::Payload::MirrorSynced(
gunnar_proto::MirrorSynced {
repo: Some(RepoPath::new(repo_ref.namespace(), repo_ref.name())),
remote_url: order.upstream.clone(),
refs_changed: updates.len() as u32,
updates,
},
));
}
}
outcome
},
))
}
pub fn resume_interrupted_mirrors(&self) -> usize {
let settled = match self
.stores
.mirrors()
.settle_interrupted(gunnar_grpc::bus::now_unix_ms())
{
Ok(settled) => settled,
Err(err) => {
tracing::warn!(target: "gunnar::git", %err, "the mirror registrations could not be read to resume interrupted syncs");
return 0;
}
};
let mut resumed = 0;
for record in settled {
let Ok(repo_ref) = gunnar_core::resolve_repo_path(&record.repo) else {
continue;
};
let again = record.interrupted <= gunnar_core::MAX_RESUMES;
audit_line(
&self.stores,
&repo_ref,
if again {
"mirror.sync.interrupted"
} else {
"mirror.sync.failed"
},
&record.last_error,
true,
);
tracing::warn!(
target: "gunnar::git",
event = "mirror.sync.interrupted",
repo = %record.repo,
interrupted = record.interrupted,
resumed = again,
"a mirror sync was in flight when the server stopped"
);
if again {
self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
path: repo_ref.key(),
upstream: record.remote_url.clone(),
prune: record.prune,
});
resumed += 1;
}
}
resumed
}
fn gate_for_purge(
&self,
account: &str,
job_id: &str,
options: &gunnar_grpc::seam::PurgeAccountOptions,
) -> ControlResult<()> {
if self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.is_some()
{
return Ok(());
}
let lease = gunnar_core::AccountGate::purge_lease(
account,
job_id,
gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS,
)
.by(&options.principal_id)
.noted(format!(
"this account is being permanently deleted: {}",
options.reason
));
self.stores.auth().gate(lease).map_err(gate_error)?;
Ok(())
}
#[must_use]
pub fn with_headroom(mut self, gate: Arc<gunnar_ssh::DiskHeadroomGate>) -> Self {
self.headroom = Some(gate);
self
}
pub fn with_grow(mut self, grow: Arc<crate::grow::GrowMachine>) -> Self {
self.grow = Some(grow);
self
}
pub fn with_admission(mut self, admission: Arc<gunnar_ssh::Admission>) -> Self {
self.admission = Some(admission);
self
}
fn grow_machine(&self) -> ControlResult<&Arc<crate::grow::GrowMachine>> {
self.grow.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no runtime growth wired in, so it cannot grow the data set",
))
}
fn verify_membership_event(
&self,
signed: gunnar_proto::SignedMembershipEvent,
group_id: &str,
member_id: &str,
action: gunnar_core::MembershipAction,
) -> ControlResult<gunnar_core::SignedMembershipEvent> {
let signed = from_proto_membership(signed)?;
if signed.event.group != group_id || signed.event.member != member_id {
return Err(ControlError::InvalidArgument(format!(
"the signed event is about {:?}/{:?} but the change is {:?}/{:?}",
signed.event.group, signed.event.member, group_id, member_id
)));
}
if signed.event.action != action {
return Err(ControlError::InvalidArgument(
"the signed event's action does not match the change (add vs remove)".to_owned(),
));
}
let key = self
.stores
.auth()
.signing_keys()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.into_iter()
.find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
.ok_or_else(|| {
ControlError::InvalidArgument(format!(
"{} is not registered on this server, so nobody reading this event later \
could check its signature",
signed.key_fingerprint
))
})?;
if key.principal_id != signed.event.actor {
return Err(ControlError::InvalidArgument(format!(
"this membership change is attributed to {:?} and is signed by {}, which is \
{:?}'s key; a membership change is the actor's own word, so it must carry the \
actor's own signature",
signed.event.actor, signed.key_fingerprint, key.principal_id
)));
}
let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
.map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
gunnar_core::membership::verify_event(&signed, &public)
.map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
Ok(signed)
}
fn append_membership_event(
&self,
signed: &gunnar_core::SignedMembershipEvent,
) -> ControlResult<u64> {
let log = self.stores.membership_log();
let mut line = serde_json::to_string(signed)
.map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
line.push('\n');
let _framed =
gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
.map_err(|err| {
ControlError::Unavailable(format!(
"framing the line in the transaction log: {err}"
))
})?;
use std::io::Write as _;
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(log.records_path())
.map_err(|err| {
ControlError::Unavailable(format!("opening the membership log: {err}"))
})?;
file.write_all(line.as_bytes()).map_err(|err| {
ControlError::Unavailable(format!("appending to the membership log: {err}"))
})?;
let count = log
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.len() as u64;
Ok(count.saturating_sub(1))
}
fn membership_log_events(&self) -> ControlResult<Vec<gunnar_core::SignedMembershipEvent>> {
self.stores
.membership_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.iter()
.map(|line| {
serde_json::from_str(line).map_err(|err| {
ControlError::Internal(format!("decoding a membership event: {err}"))
})
})
.collect()
}
#[cfg(feature = "replication")]
pub fn with_replication(mut self, slot: Arc<crate::mirror::ReplicationSlot>) -> Self {
self.replication_slot = Some(slot);
self
}
#[cfg(feature = "replication")]
fn replication_slot(&self) -> ControlResult<&Arc<crate::mirror::ReplicationSlot>> {
self.replication_slot.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no replication slot, so it cannot configure a standby",
))
}
#[cfg(feature = "webhooks")]
pub fn with_webhooks(mut self, slot: Arc<crate::webhook::WebhookSlot>) -> Self {
self.webhook_slot = Some(slot);
self
}
#[cfg(feature = "metered")]
pub fn with_entitlement(mut self, slot: Arc<crate::entitlement::EntitlementSlot>) -> Self {
self.entitlement = Some(slot);
self
}
#[cfg(feature = "metered")]
fn entitlement(&self) -> ControlResult<&Arc<crate::entitlement::EntitlementSlot>> {
self.entitlement.as_ref().ok_or_else(|| {
ControlError::FailedPrecondition(
"this metered server was started without `--monetize-pubkey`, so it holds no \
key to verify an entitlement fact against and refuses every one; every \
namespace is Free"
.into(),
)
})
}
#[cfg(feature = "webhooks")]
fn webhook_slot(&self) -> ControlResult<&Arc<crate::webhook::WebhookSlot>> {
self.webhook_slot.as_ref().ok_or(ControlError::Unimplemented(
"this control plane was built with no webhook slot, so it cannot configure an outbound endpoint",
))
}
#[cfg(feature = "replication")]
fn replication_snapshot(
&self,
include_refs: bool,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let mut out = gunnar_proto::ReplicationStatusResponse {
targets: config
.targets
.iter()
.map(crate::replication_wire::target_to_proto)
.collect(),
role: config.role.to_proto() as i32,
..Default::default()
};
if let Some(replicator) = slot.replicator() {
crate::replication_wire::running_to_proto(&replicator, include_refs, &mut out);
}
Ok(out)
}
pub fn replication_view(&self) -> crate::replication_view::ReplicationView {
self.replication.clone()
}
fn forge_events(&self) -> ControlResult<Vec<gunnar_core::SignedEvent>> {
self.stores
.forge_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("reading the forge log: {err:#}")))?
.iter()
.map(|line| {
serde_json::from_str(line).map_err(|err| {
ControlError::Internal(format!("a forge event will not decode: {err}"))
})
})
.collect()
}
fn forge_subject_open(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<()> {
let (namespace, name) = signed.event.repo.split_once('/').ok_or_else(|| {
ControlError::InvalidArgument(format!(
"`repo` must be `<namespace>/<repo>`, got {:?}",
signed.event.repo
))
})?;
let path = RepoPath::new(namespace, name);
match &signed.event.event {
gunnar_core::Event::IssueOpened { .. } => {}
gunnar_core::Event::IssueCommented { .. } | gunnar_core::Event::IssueClosed { .. } => {
let repo_events: Vec<gunnar_core::ForgeEvent> = self
.forge_events()?
.into_iter()
.filter(|e| e.event.repo == signed.event.repo)
.map(|e| e.event)
.collect();
match gunnar_core::forge::fold_issue(signed.event.pull, &repo_events) {
None => {
return Err(ControlError::FailedPrecondition(format!(
"no issue #{} in {}: nobody has opened one, so an event on it is \
evidence about nothing",
signed.event.pull, signed.event.repo
)));
}
Some(issue) if !issue.open => {
return Err(ControlError::FailedPrecondition(format!(
"issue #{} in {} is closed; an event on it cannot change what has \
already happened",
signed.event.pull, signed.event.repo
)));
}
Some(_) => {}
}
}
gunnar_core::Event::Commented { .. }
| gunnar_core::Event::Approved { .. }
| gunnar_core::Event::Rejected { .. } => {
let record = self.get_pull(&path, signed.event.pull)?;
if record.state != gunnar_proto::PullRequestState::Open as i32 {
return Err(ControlError::FailedPrecondition(format!(
"pull request #{} in {} is not open; a review event cannot change what \
has already happened",
signed.event.pull, signed.event.repo
)));
}
}
}
Ok(())
}
fn forge_write(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<u64> {
let log = self.stores.forge_log();
let mut line = serde_json::to_string(signed)
.map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
line.push('\n');
let _framed =
gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
.map_err(|err| {
ControlError::Unavailable(format!(
"framing the line in the transaction log: {err}"
))
})?;
use std::io::Write as _;
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(log.records_path())
.map_err(|err| ControlError::Unavailable(format!("opening the forge log: {err}")))?;
file.write_all(line.as_bytes()).map_err(|err| {
ControlError::Unavailable(format!("appending to the forge log: {err}"))
})?;
let count = self
.stores
.forge_log()
.lines_of()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.len() as u64;
Ok(count.saturating_sub(1))
}
fn review_state(
&self,
repo: &str,
state: &gunnar_core::forge::Review,
all: &[gunnar_core::SignedEvent],
) -> ControlResult<gunnar_proto::ReviewState> {
let keys: Vec<(String, String, gunnar_core::ssh_key::PublicKey)> = self
.stores
.auth()
.signing_keys()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.iter()
.filter_map(|row| {
gunnar_core::ssh_key::PublicKey::from_openssh(&row.openssh)
.ok()
.map(|key| {
(
row.fingerprint_sha256.clone(),
row.principal_id.clone(),
key,
)
})
})
.collect();
let attest_key = self
.stores
.forge_attest_public()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
let approvals = state
.approvals
.iter()
.map(|(actor, commit)| {
let signed = all.iter().rev().find(|e| {
e.event.repo == repo
&& e.event.pull == state.pull
&& e.event.actor == *actor
&& matches!(&e.event.event, gunnar_core::Event::Approved { commit: c } if c == commit)
});
let (verified, verdict) = match signed {
None => (false, "NO EVENT".to_owned()),
Some(signed) if signed.attested.is_some() => match &attest_key {
None => (false, "UNKNOWN KEY".to_owned()),
Some(key) => match gunnar_core::forge::verify_event(signed, key) {
Ok(()) => (
true,
signed.attested.as_ref().map(|a| a.words()).unwrap_or_default(),
),
Err(problem) => (false, format!("BAD ({problem})")),
},
},
Some(signed) => match keys
.iter()
.find(|(fingerprint, _, _)| *fingerprint == signed.key_fingerprint)
{
None => (false, "UNKNOWN KEY".to_owned()),
Some((_, owner, _)) if *owner != *actor => {
(false, format!("KEY OF {owner}, NOT {actor}"))
}
Some((_, _, key)) => match gunnar_core::forge::verify_event(signed, key) {
Ok(()) => (true, "verified".to_owned()),
Err(problem) => (false, format!("BAD ({problem})")),
},
},
};
gunnar_proto::Approval {
actor: actor.clone(),
commit: commit.clone(),
verified,
verdict,
}
})
.collect();
Ok(gunnar_proto::ReviewState {
repo: repo.to_owned(),
pull: state.pull,
approvals,
comments: state.comments as u64,
})
}
pub fn with_max_object_bytes(mut self, bytes: u64) -> Self {
self.max_object_bytes = bytes;
self
}
fn pulls(&self) -> ControlResult<gunnar_core::Pulls> {
gunnar_core::Pulls::new(Arc::clone(self.stores.control_store()))
.map_err(|err| ControlError::Internal(format!("opening the pulls table: {err:#}")))
}
fn repo_ref(path: &RepoPath) -> ControlResult<RepoRef> {
path.validate()
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
RepoRef::new(&path.namespace, &path.repo)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))
}
fn tree_of(
&self,
git: &dyn GitServe,
hash: gix_hash::Kind,
path: &RepoPath,
spec: &str,
) -> ControlResult<gix_hash::ObjectId> {
let id = match gix_hash::ObjectId::from_hex(spec.as_bytes()) {
Ok(id) => {
if id.kind() != hash {
return Err(ControlError::InvalidArgument(format!(
"{spec} is a {} oid; this repository is {hash}",
id.kind()
)));
}
id
}
Err(_) => {
gunnar_store::refname::validate(spec)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let resolved = if spec == gunnar_store::HEAD {
snapshot.head(git).map_err(wire_error)?
} else {
snapshot.resolve(spec).map_err(wire_error)?
};
resolved.ok_or_else(|| ControlError::NotFound(format!("{spec} in {path}")))?
}
};
let id = peel(git, hash, id)?;
let (kind, bytes) = git
.read(id.as_bytes())
.map_err(store_error)?
.ok_or_else(|| ControlError::NotFound(format!("object {id} in {path}")))?;
match kind {
ObjType::Tree => Ok(id),
ObjType::Commit => object::commit(&bytes, hash)
.map(|commit| commit.tree)
.map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}"))),
other => Err(ControlError::InvalidArgument(format!(
"{spec} names a {other:?}; a diff needs a commit, a tag or a tree"
))),
}
}
fn policy_of(&self, repo: &gunnar_core::Repo) -> ControlResult<gunnar_policy::PolicyConfig> {
let stored = self.stores.policies().get(&repo.path()).map_err(internal)?;
match stored {
None => Ok(gunnar_policy::PolicyConfig::default()),
Some(document) => gunnar_policy::PolicyConfig::from_json(&document).map_err(|err| {
ControlError::Internal(format!(
"the stored policy for {} does not parse: {err}",
repo.key()
))
}),
}
}
fn store_policy(
&self,
repo: &gunnar_core::Repo,
config: &gunnar_policy::PolicyConfig,
) -> ControlResult<()> {
let policies = self.stores.policies();
if *config == gunnar_policy::PolicyConfig::default() {
policies.remove(&repo.path()).map_err(internal)?;
return Ok(());
}
let json = config
.to_json()
.map_err(|err| ControlError::Internal(format!("rendering the policy: {err}")))?;
policies.set(&repo.path(), &json).map_err(internal)
}
fn open(&self, path: &RepoPath) -> ControlResult<(gunnar_core::Repo, Arc<OpenRepo>)> {
let repo_ref = Self::repo_ref(path)?;
let repo = self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
let store = self.stores.repo_store(&repo).map_err(|err| {
ControlError::Internal(format!("opening the repository's stores: {err:#}"))
})?;
Ok((repo, store))
}
fn mirror_record(&self, path: &RepoPath) -> ControlResult<gunnar_core::MirrorRecord> {
let repo_ref = Self::repo_ref(path)?;
let record = self.stores.mirrors().get(&repo_ref).map_err(internal)?;
match record {
Some(record)
if record.creates_repo
&& self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_none() =>
{
Ok(record)
}
record => {
self.open(path)?;
record.ok_or_else(|| {
ControlError::NotFound(format!("no mirror registered for {path}"))
})
}
}
}
}
fn gc_job_out(path: &RepoPath, job: crate::gc_queue::GcJob) -> gunnar_grpc::seam::GcJob {
gunnar_grpc::seam::GcJob {
job_id: job.job_id,
path: path.clone(),
state: match job.state {
crate::gc_queue::GcState::Queued => gunnar_grpc::seam::GcState::Queued,
crate::gc_queue::GcState::Running => gunnar_grpc::seam::GcState::Running,
crate::gc_queue::GcState::Done => gunnar_grpc::seam::GcState::Done,
crate::gc_queue::GcState::Failed => gunnar_grpc::seam::GcState::Failed,
},
queue_position: job.queue_position,
bytes_reclaimed: job.report.bytes_reclaimed,
error: job.error,
enqueued_at_unix_ms: job.enqueued_at_unix_ms,
finished_at_unix_ms: job.finished_at_unix_ms,
}
}
fn gate_out(gate: gunnar_core::AccountGate) -> gunnar_grpc::seam::AccountGate {
gunnar_grpc::seam::AccountGate {
account: gate.account,
reason: match gate.reason {
gunnar_core::GateReason::Administrative => {
gunnar_grpc::seam::GateReason::Administrative
}
gunnar_core::GateReason::Purge => gunnar_grpc::seam::GateReason::Purge,
},
holder: gate.holder,
actor: gate.actor,
note: gate.note,
at_unix_ms: gate.at,
expires_unix_ms: gate.expires.unwrap_or(0),
}
}
fn purge_job_out(job: crate::purge_queue::PurgeJob) -> gunnar_grpc::seam::PurgeJob {
let report = job.report;
gunnar_grpc::seam::PurgeJob {
job_id: job.job_id,
account: job.key,
state: match job.state {
crate::purge_queue::PurgeState::Queued => gunnar_grpc::seam::PurgeState::Queued,
crate::purge_queue::PurgeState::Running => gunnar_grpc::seam::PurgeState::Running,
crate::purge_queue::PurgeState::Done => gunnar_grpc::seam::PurgeState::Done,
crate::purge_queue::PurgeState::Failed => gunnar_grpc::seam::PurgeState::Failed,
},
queue_position: job.queue_position,
stores_found: report.stores_found,
stores_purged: report.stores_purged,
bytes_reclaimed: report.bytes_reclaimed,
principal_removed: report.principal_removed,
gate_released: report.gate_released,
error: job.error,
failures: report.failures,
enqueued_at_unix_ms: job.enqueued_at_unix_ms,
started_at_unix_ms: job.started_at_unix_ms,
finished_at_unix_ms: job.finished_at_unix_ms,
}
}
fn gate_error(err: gunnar_core::Error) -> ControlError {
match err {
gunnar_core::Error::ReservedPrincipalId(_) | gunnar_core::Error::InvalidPath { .. } => {
ControlError::InvalidArgument(format!("{err}"))
}
other => internal(other),
}
}
fn day_activity(day: i64, row: &gunnar_core::FlowDay) -> gunnar_proto::DayActivity {
gunnar_proto::DayActivity {
day_unix_ms: gunnar_core::midnight_of(day),
bytes_in: row.bytes_in,
bytes_out: row.bytes_out,
pushes: row.pushes,
fetches: row.fetches,
lfs_bytes_in: row.lfs_bytes_in,
lfs_bytes_out: row.lfs_bytes_out,
refused: row.refused(),
refused_by_reason: gunnar_core::Refusal::ALL
.into_iter()
.filter(|reason| row.refused_for(*reason) > 0)
.map(|reason| (reason.name().to_owned(), row.refused_for(reason)))
.collect(),
auth_failures: row.auth_failures,
last_push_unix_ms: row.last_push_unix_ms,
last_fetch_unix_ms: row.last_fetch_unix_ms,
distinct_principals: row.distinct_principals(),
distinct_principals_capped: row.principals_capped,
}
}
pub(crate) fn create_error(err: &anyhow::Error, path: &RepoPath) -> ControlError {
match err.downcast_ref::<gunnar_core::Error>() {
Some(gunnar_core::Error::RepoExists(_)) => {
ControlError::AlreadyExists(format!("repository {path}"))
}
Some(gunnar_core::Error::NoSuchNamespace(ns)) => {
ControlError::NotFound(format!("namespace {ns}"))
}
Some(e @ (gunnar_core::Error::LogFollows | gunnar_core::Error::OffStripe { .. })) => {
ControlError::Unavailable(e.to_string())
}
Some(_) => ControlError::Internal(format!("{err:#}")),
None => ControlError::FailedPrecondition(format!("{err:#}")),
}
}
pub(crate) fn first_sync_of<E: std::fmt::Display>(
mark: Result<Option<gunnar_core::MirrorRecord>, E>,
path: &str,
) -> Result<bool, String> {
match mark {
Ok(record) => {
Ok(record.is_some_and(|record| record.creates_repo || record.sync_count == 0))
}
Err(err) => {
tracing::error!(target: "gunnar::git", "ALERT mirror.mark-unwritten {path}: the in-flight mark was not written ({err}); the sync is not started");
Err(format!("{path}: the mirror's in-flight mark was not written ({err}); nothing was fetched, ask again"))
}
}
}
#[cfg(test)]
mod mirror_mark_tests {
use super::*;
#[test]
fn a_mark_that_is_not_written_fails_the_sync_by_name() {
let why = first_sync_of::<String>(Err("the mirrors store is full".into()), "acme/linux")
.expect_err("refused");
assert!(
why.contains("acme/linux")
&& why.contains("the mirrors store is full")
&& why.contains("in-flight mark"),
"{why}"
);
assert_eq!(first_sync_of::<String>(Ok(None), "acme/linux"), Ok(false));
}
}
#[cfg(test)]
mod create_error_tests {
use super::*;
#[test]
fn a_create_on_a_restoring_box_is_unavailable() {
let path = RepoPath::new("acme", "site");
let restoring = anyhow::Error::new(gunnar_core::Error::LogFollows);
assert!(
matches!(create_error(&restoring, &path), ControlError::Unavailable(m) if m.contains("being restored"))
);
let exists = anyhow::Error::new(gunnar_core::Error::RepoExists(
gunnar_core::RepoRef::new("acme", "site").unwrap(),
));
assert!(matches!(
create_error(&exists, &path),
ControlError::AlreadyExists(_)
));
assert!(matches!(
create_error(&anyhow::anyhow!("a store already stands there"), &path),
ControlError::FailedPrecondition(_)
));
}
}
fn internal(err: gunnar_core::Error) -> ControlError {
use gunnar_core::Error as E;
match err {
e @ (E::InvalidPath { .. }
| E::PathEscape { .. }
| E::ReservedPrincipalId(_)
| E::BadBranchName(_)
| E::BadFingerprint(_)
| E::Key(_)
| E::InvalidInput { .. }) => ControlError::InvalidArgument(e.to_string()),
e @ (E::NoSuchRepo(_)
| E::NoSuchNamespace(_)
| E::NoSuchPrincipal(_)
| E::NoSuchCredential(_)) => ControlError::NotFound(e.to_string()),
e @ (E::RepoExists(_)
| E::NamespaceExists(_)
| E::PrincipalExists(_)
| E::NameExists { .. }
| E::KeyExists { .. }
| E::CredentialExists { .. }
| E::InviteExists(_)
| E::TokenExists { .. }
| E::AlreadyBootstrapped { .. }
| E::AlreadyProvisioned { .. }) => ControlError::AlreadyExists(e.to_string()),
e @ (E::NamespaceNotEmpty { .. }
| E::Contended { .. }
| E::NotAPerson { .. }
| E::NotAMachine { .. }) => ControlError::FailedPrecondition(e.to_string()),
e @ (E::Denied { .. } | E::CounterRollback { .. } | E::InviteInvalid) => {
ControlError::PermissionDenied(e.to_string())
}
e @ (E::Store(_) | E::Json(_)) => ControlError::Unavailable(e.to_string()),
e @ E::OffStripe { .. } => ControlError::Unavailable(e.to_string()),
e @ E::LogFollows => ControlError::Unavailable(e.to_string()),
e => ControlError::Internal(e.to_string()),
}
}
#[cfg(feature = "engine-znippy")]
fn archive_error(err: anyhow::Error) -> ControlError {
let io = err
.chain()
.any(|cause| cause.downcast_ref::<std::io::Error>().is_some());
let message = format!("{err:#}");
if io {
ControlError::Unavailable(message)
} else {
ControlError::FailedPrecondition(message)
}
}
fn store_error(err: anyhow::Error) -> ControlError {
ControlError::Internal(format!("{err:#}"))
}
fn wire_error(err: gunnar_wire::Error) -> ControlError {
ControlError::Internal(format!("{err:#}"))
}
pub(crate) fn hash_kind(kind: HashKind) -> gunnar_proto::HashKind {
match kind {
HashKind::Sha1 => gunnar_proto::HashKind::Sha1,
HashKind::Sha256 => gunnar_proto::HashKind::Sha256,
}
}
fn visibility_in(v: gunnar_proto::Visibility) -> ControlResult<Visibility> {
match v {
gunnar_proto::Visibility::Public => Ok(Visibility::Public),
gunnar_proto::Visibility::Internal => Err(ControlError::InvalidArgument(
"visibility \"internal\" has no implementation: since 2026-08-03, \"public\" means \
readable with no key at all, so \"internal\" (any authenticated principal) \
no longer aliases it and gunnar refuses to guess which you meant. Choose \
\"private\" or \"public\""
.into(),
)),
gunnar_proto::Visibility::Private | gunnar_proto::Visibility::Unspecified => {
Ok(Visibility::Private)
}
}
}
fn pull_out(path: &RepoPath, record: gunnar_core::PullRecord) -> PullRequest {
PullRequest {
repo: Some(path.clone()),
number: record.number,
title: record.title,
description: record.description,
source_ref: record.source_ref,
target_ref: record.target_ref,
state: match record.state {
gunnar_core::PullState::Open => PullRequestState::Open,
gunnar_core::PullState::Merged => PullRequestState::Merged,
gunnar_core::PullState::Closed => PullRequestState::Closed,
} as i32,
author: record.author,
created_at_unix_ms: record.created_ms,
updated_at_unix_ms: record.updated_ms,
source_oid: record.source_oid,
merge_oid: record.merge_oid.unwrap_or_default(),
}
}
fn visibility_out(v: Visibility) -> gunnar_proto::Visibility {
match v {
Visibility::Private => gunnar_proto::Visibility::Private,
Visibility::Public => gunnar_proto::Visibility::Public,
}
}
fn principal_kind_in(kind: gunnar_proto::PrincipalKind) -> PrincipalKind {
match kind {
gunnar_proto::PrincipalKind::Machine => PrincipalKind::Machine,
gunnar_proto::PrincipalKind::Deploy => PrincipalKind::Deploy,
gunnar_proto::PrincipalKind::Group => PrincipalKind::Group,
gunnar_proto::PrincipalKind::User | gunnar_proto::PrincipalKind::Unspecified => {
PrincipalKind::User
}
}
}
impl CoreControlPlane {
fn owner_kind(&self, name: &str) -> i32 {
let auth = self.stores.auth();
match auth.principal(name) {
Ok(Some(p)) if p.kind == PrincipalKind::Group => {
let members = auth.members_of(name).unwrap_or_default();
let machines = !members.is_empty() && members.iter().all(|m| matches!(auth.principal(m), Ok(Some(q)) if q.kind == PrincipalKind::Machine));
if machines {
gunnar_proto::PrincipalKind::Machine as i32
} else {
principal_kind_out(p.kind) as i32
}
}
Ok(Some(p)) => principal_kind_out(p.kind) as i32,
_ => gunnar_proto::PrincipalKind::Unspecified as i32,
}
}
fn namespace_out(&self, namespace: gunnar_core::Namespace, employees: u32) -> Namespace {
Namespace {
owner_kind: self.owner_kind(&namespace.name),
owner_server_grant: self
.stores
.auth()
.holds_server_grant(&namespace.name)
.unwrap_or(true),
name: namespace.name,
description: String::new(),
created_at_unix_ms: 0,
public: namespace.public,
events_public: namespace.events_public,
employees,
}
}
fn employee_count(&self, account: &str) -> ControlResult<u32> {
Ok(self
.stores
.auth()
.employees_of(account)
.map_err(internal)?
.len() as u32)
}
}
fn principal_kind_out(kind: PrincipalKind) -> gunnar_proto::PrincipalKind {
match kind {
PrincipalKind::User => gunnar_proto::PrincipalKind::User,
PrincipalKind::Machine => gunnar_proto::PrincipalKind::Machine,
PrincipalKind::Deploy => gunnar_proto::PrincipalKind::Deploy,
PrincipalKind::Group => gunnar_proto::PrincipalKind::Group,
PrincipalKind::Anonymous => gunnar_proto::PrincipalKind::Unspecified,
}
}
fn level_in(level: GrantLevel) -> ControlResult<Level> {
match level {
GrantLevel::Read => Ok(Level::Read),
GrantLevel::Write => Ok(Level::Write),
GrantLevel::Admin => Ok(Level::Admin),
GrantLevel::Unspecified => Err(ControlError::InvalidArgument(
"grant level unspecified".into(),
)),
}
}
fn level_out(level: Level) -> GrantLevel {
match level {
Level::Read => GrantLevel::Read,
Level::Write => GrantLevel::Write,
Level::Admin => GrantLevel::Admin,
}
}
fn repo_out(repo: &gunnar_core::Repo) -> Repo {
Repo {
path: Some(RepoPath::new(&repo.namespace, &repo.name)),
hash_kind: hash_kind(repo.hash_kind) as i32,
default_branch: repo.default_branch.clone(),
visibility: visibility_out(repo.visibility) as i32,
created_at_unix_ms: repo.created,
description: repo.description.clone(),
encrypted: repo.encrypted,
}
}
fn web_credential_out(
row: &gunnar_core::WebCredential,
) -> ControlResult<gunnar_proto::WebCredential> {
let credential_id = gunnar_grpc::webauthn::from_b64url(&row.id).ok_or_else(|| {
ControlError::Internal(format!(
"a stored web credential has an id that is not base64url: {}",
row.id
))
})?;
Ok(gunnar_proto::WebCredential {
credential_id,
principal_id: row.principal_id.clone(),
label: row.label.clone(),
created_at_unix_ms: row.created,
last_used_unix_ms: row.last_used,
sign_count: row.sign_count,
})
}
fn principal_out(principal: &gunnar_core::Principal) -> Principal {
Principal {
id: principal.id.clone(),
name: principal.name.clone(),
kind: principal_kind_out(principal.kind) as i32,
created_at_unix_ms: principal.created,
}
}
mod clan_plane;
fn key_out(key: &gunnar_core::PublicKey) -> PublicKey {
PublicKey {
fingerprint: key.fingerprint_sha256.clone(),
algo: key.algo.clone(),
principal_id: key.principal_id.clone(),
comment: String::new(),
added_at_unix_ms: key.added,
last_used_unix_ms: key.last_used,
expires_at_unix_ms: key.expires,
}
}
fn grant_out(grant: &Grant) -> gunnar_proto::Grant {
let (namespace, repo) = match &grant.scope {
Scope::Server => (String::new(), String::new()),
Scope::Namespace(name) => (name.clone(), String::new()),
Scope::Repo(path) => (path.namespace().to_owned(), path.name().to_owned()),
};
gunnar_proto::Grant {
principal_id: grant.principal.clone(),
namespace,
repo,
level: level_out(grant.level) as i32,
granted_at_unix_ms: 0,
}
}
fn scope_in(namespace: &str, repo: &str) -> ControlResult<Scope> {
match (namespace.is_empty(), repo.is_empty()) {
(true, true) => Ok(Scope::Server),
(false, true) => Ok(Scope::Namespace(namespace.to_owned())),
(false, false) => RepoRef::new(namespace, repo)
.map(Scope::Repo)
.map_err(|err| ControlError::InvalidArgument(err.to_string())),
(true, false) => Err(ControlError::InvalidArgument(
"a repository-scoped grant must name a namespace (multi-tenancy has \
no ungrouped repos)"
.into(),
)),
}
}
fn object_kind_out(kind: ObjType) -> ControlResult<gunnar_proto::ObjectKind> {
match kind {
ObjType::Blob => Ok(gunnar_proto::ObjectKind::Blob),
ObjType::Tree => Ok(gunnar_proto::ObjectKind::Tree),
ObjType::Commit => Ok(gunnar_proto::ObjectKind::Commit),
ObjType::Tag => Ok(gunnar_proto::ObjectKind::Tag),
other => Err(ControlError::Internal(format!(
"the store answered a read with `{}`, which is a delta; GitServe::read is \
documented to return the resolved type",
other.as_str()
))),
}
}
fn oid_at_path(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
tree: gix_hash::ObjectId,
segments: &[&[u8]],
) -> ControlResult<Option<gix_hash::ObjectId>> {
let mut current = tree;
let last = segments.len().saturating_sub(1);
for (index, segment) in segments.iter().enumerate() {
let Some((kind, bytes)) = git.read(current.as_bytes()).map_err(store_error)? else {
return Ok(None);
};
if kind != ObjType::Tree {
return Ok(None);
}
let Some(entry) = object::tree_entry_named(&bytes, hash_kind, segment)
.map_err(|err| ControlError::Internal(format!("decoding tree {current}: {err}")))?
else {
return Ok(None);
};
if index == last {
return Ok(Some(entry.oid));
}
if entry.kind != object::EntryKind::Tree {
return Ok(None);
}
current = entry.oid;
}
Ok(Some(current))
}
#[derive(Clone, Copy)]
enum LogStart<'a> {
One(&'a str),
AllRefs,
}
fn resolve_log_start(
git: &dyn GitServe,
hash: gix_hash::Kind,
path: &RepoPath,
start: &str,
) -> ControlResult<gix_hash::ObjectId> {
let start = if start.is_empty() {
gunnar_store::HEAD
} else {
start
};
let head = match gix_hash::ObjectId::from_hex(start.as_bytes()) {
Ok(id) => id,
Err(_) => {
gunnar_store::refname::validate(start)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let resolved = if start == gunnar_store::HEAD {
snapshot.head(git).map_err(wire_error)?
} else {
snapshot.resolve(start).map_err(wire_error)?
};
resolved.ok_or_else(|| ControlError::NotFound(format!("{start} in {path}")))?
}
};
if head.kind() != hash {
return Err(ControlError::InvalidArgument(format!(
"{start} is a {} oid; this repository is {hash}",
head.kind()
)));
}
Ok(head)
}
impl CoreControlPlane {
fn log_walk(
&self,
path: &RepoPath,
start: LogStart<'_>,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
use std::collections::{BinaryHeap, HashMap, HashSet};
let (_, store) = self.open(path)?;
let git = store.serve();
let hash = store.hash_kind();
let filter: Option<Vec<&[u8]>> = if touching.is_empty() {
None
} else {
let mut segments = Vec::new();
for segment in touching.split('/').filter(|s| !s.is_empty()) {
if segment == "." || segment == ".." {
return Err(ControlError::InvalidArgument(format!(
"the log path filter cannot contain {segment:?}"
)));
}
segments.push(segment.as_bytes());
}
(!segments.is_empty()).then_some(segments)
};
let heads: Vec<gix_hash::ObjectId> = match start {
LogStart::One(start) => vec![resolve_log_start(git, hash, path, start)?],
LogStart::AllRefs => {
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let mut heads = Vec::new();
if let Some(head) = snapshot.head(git).map_err(wire_error)? {
heads.push(head);
}
for row in snapshot.rows().iter().filter(|r| {
r.name.starts_with("refs/heads/") || r.name.starts_with("refs/tags/")
}) {
if let Some(id) = snapshot.resolve_row(row).map_err(wire_error)? {
heads.push(id);
}
}
heads
}
};
let limit = if limit == 0 { 50 } else { limit } as usize;
let mut heap: BinaryHeap<(i64, gix_hash::ObjectId)> = BinaryHeap::new();
let mut seen: HashSet<gix_hash::ObjectId> = HashSet::new();
let mut queued: HashMap<gix_hash::ObjectId, object::Commit> = HashMap::new();
let mut enqueue = |id: gix_hash::ObjectId,
heap: &mut BinaryHeap<(i64, gix_hash::ObjectId)>,
queued: &mut HashMap<gix_hash::ObjectId, object::Commit>|
-> ControlResult<()> {
if !seen.insert(id) {
return Ok(());
}
let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
return Ok(());
};
if kind != ObjType::Commit {
return Ok(());
}
let commit = object::commit(&bytes, hash)
.map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}")))?;
heap.push((commit.committer.seconds, id));
queued.insert(id, commit);
Ok(())
};
for head in heads {
let head = peel(git, hash, head)?;
enqueue(head, &mut heap, &mut queued)?;
}
let mut out = Vec::new();
const MAX_FILTERED_SCAN: usize = 20_000;
let mut scanned = 0usize;
let mut tree_of: HashMap<gix_hash::ObjectId, gix_hash::ObjectId> = HashMap::new();
while let Some((_, id)) = heap.pop() {
if out.len() >= limit {
break;
}
let Some(commit) = queued.remove(&id) else {
continue;
};
let object::Commit {
tree,
parents,
author,
committer,
message,
} = commit;
if let Some(segments) = &filter {
scanned += 1;
if scanned > MAX_FILTERED_SCAN {
break;
}
tree_of.insert(id, tree);
let current = oid_at_path(git, hash, tree, segments)?;
let previous = match parents.first() {
None => None,
Some(parent) => {
let parent_tree = match tree_of.get(parent) {
Some(known) => Some(*known),
None => match git.read(parent.as_bytes()).map_err(store_error)? {
Some((ObjType::Commit, parent_bytes)) => {
let decoded =
object::commit(&parent_bytes, hash).map_err(|err| {
ControlError::Internal(format!(
"decoding commit {parent}: {err}"
))
})?;
tree_of.insert(*parent, decoded.tree);
Some(decoded.tree)
}
_ => None,
},
};
match parent_tree {
Some(parent_tree) => oid_at_path(git, hash, parent_tree, segments)?,
None => None,
}
}
};
if current == previous {
for parent in parents {
enqueue(parent, &mut heap, &mut queued)?;
}
continue;
}
}
out.push(Commit {
oid: id.to_string(),
parent_oids: parents.iter().map(|p| p.to_string()).collect(),
tree_oid: tree.to_string(),
author: Some(signature_out(
&author.name,
&author.email,
author.seconds,
author.offset_seconds,
)),
committer: Some(signature_out(
&committer.name,
&committer.email,
committer.seconds,
committer.offset_seconds,
)),
message: String::from_utf8_lossy(&message).into_owned(),
});
for parent in parents {
enqueue(parent, &mut heap, &mut queued)?;
}
}
Ok(out)
}
}
fn peel(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
mut id: gix_hash::ObjectId,
) -> ControlResult<gix_hash::ObjectId> {
for _ in 0..16 {
let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
return Ok(id);
};
if kind != ObjType::Tag {
return Ok(id);
}
id = object::tag_target(&bytes, hash_kind)
.map_err(|err| ControlError::Internal(format!("decoding tag {id}: {err}")))?;
}
Err(ControlError::FailedPrecondition(format!(
"the tag chain at {id} is longer than 16 links, or cyclic"
)))
}
fn ref_target_out(
git: &dyn GitServe,
hash_kind: gix_hash::Kind,
snapshot: &RefSnapshot,
row: &RefRow,
want_peeled: bool,
) -> ControlResult<RefTarget> {
let (oid, symbolic) = match (&row.oid, &row.symref_target) {
(_, Some(inner)) => (snapshot.resolve(inner).map_err(wire_error)?, inner.clone()),
(Some(raw), None) => (
Some(gunnar_wire::refs::oid(raw).map_err(wire_error)?),
String::new(),
),
(None, None) => (None, String::new()),
};
let peeled = match (want_peeled, oid) {
(true, Some(id)) => {
let peeled = peel(git, hash_kind, id)?;
if peeled == id {
String::new()
} else {
peeled.to_string()
}
}
_ => String::new(),
};
Ok(RefTarget {
ref_name: row.name.clone(),
oid: oid.map(|id| id.to_string()).unwrap_or_default(),
symbolic_target: symbolic,
peeled_oid: peeled,
})
}
fn signature_out(name: &[u8], email: &[u8], seconds: i64, tz_offset_seconds: i32) -> Signature {
Signature {
name: String::from_utf8_lossy(name).into_owned(),
email: String::from_utf8_lossy(email).into_owned(),
time_unix_seconds: seconds,
tz_offset_minutes: tz_offset_seconds / 60,
}
}
impl gunnar_grpc::FirstBoot for CoreControlPlane {
fn bootstrap_state(&self) -> ControlResult<gunnar_grpc::BootstrapState> {
let auth = self.stores.auth();
let window_open = auth.bootstrap_window_open().map_err(unavailable)?;
let record = auth.bootstrap_record().map_err(unavailable)?;
let (principals, keys, _grants) = auth.counts().map_err(unavailable)?;
Ok(gunnar_grpc::BootstrapState {
window_open,
bootstrapped: record.is_some(),
bootstrapped_at_unix_ms: record.as_ref().map(|r| r.at).unwrap_or_default(),
bootstrapped_principal: record
.as_ref()
.map(|r| r.principal_id.clone())
.unwrap_or_default(),
bootstrapped_fingerprint: record
.as_ref()
.map(|r| r.fingerprint.clone())
.unwrap_or_default(),
principals: principals as u64,
keys: keys as u64,
serving_git: crate::readiness::is_serving_git(self.git_listener),
readiness: crate::readiness::detail_with(self.git_listener),
build: gunnar_common::BUILD_REF.to_owned(),
role: if self.twin_role || self.stores.is_twin() {
"twin"
} else {
"primary"
}
.to_owned(),
recharge_from: awaited_twin(&self.stores),
restoring: crate::twin::is_restoring(self.stores.root()),
log_head_seq: self.stores.txlog().head_seq(),
restore_partial: if crate::twin::is_restoring(self.stores.root()) {
crate::twin::intake::partial(self.stores.root())
} else {
None
},
restore_snapshot_resumable: true,
restore_snapshot: if crate::twin::is_restoring(self.stores.root()) {
crate::twin::snapshot::held_snapshot(self.stores.root())
.map(|h| (h.id, h.items, h.bytes, h.whole))
} else {
None
},
})
}
fn enroll_first_identity(
&self,
name: &str,
openssh_public_key: &str,
) -> ControlResult<gunnar_grpc::Enrolled> {
#[cfg(feature = "replication")]
if let Some(twin) = crate::repair::held(&self.stores) {
return Err(ControlError::FailedPrecondition(
crate::repair::held_refusal(&twin),
));
}
let enrolled = self
.stores
.auth()
.enroll_first_user(name, openssh_public_key)
.map_err(enrolment_error)?;
if let Err(err) = self
.stores
.catalog()
.create_namespace(&enrolled.record.account)
{
tracing::warn!(
account = %enrolled.record.account,
%err,
"the first user was enrolled and the enrolment window is closed, but its account \
namespace could not be created. The user is admin of it by rule and can create \
it themselves with Repos.CreateNamespace."
);
}
let (client_cert_pem, client_key_pem, issuer_cert_pem) = match self.stores.pki() {
Ok(pki) => match pki.issuer() {
Ok(Some(issuer)) => {
match pki.issue_with_server_key(
&enrolled.principal.id,
gunnar_core::DEFAULT_LEAF_DAYS,
) {
Ok(issued) => {
if let Err(err) = self.stores.auth().register_certificate(
&enrolled.principal.id,
&issued.cert_der,
None,
) {
tracing::warn!(
principal = %enrolled.principal.id,
%err,
"the bootstrap certificate was minted but could not be \
registered; it is returned anyway and gunnar-admin can \
register it with `gunnar key add`"
);
}
(
issued.cert_pem,
issued.key_pem.unwrap_or_default(),
issuer.cert_pem,
)
}
Err(err) => {
tracing::warn!(
principal = %enrolled.principal.id,
%err,
"no bootstrap certificate: the issuer refused to mint"
);
(String::new(), String::new(), String::new())
}
}
}
Ok(None) => (String::new(), String::new(), String::new()),
Err(err) => {
tracing::warn!(%err, "no bootstrap certificate: the PKI document could not \
be read");
(String::new(), String::new(), String::new())
}
},
Err(_) => (String::new(), String::new(), String::new()),
};
Ok(gunnar_grpc::Enrolled {
principal: principal_out(&enrolled.principal),
key: key_out(&enrolled.key),
account: enrolled.record.account,
bootstrapped_at_unix_ms: enrolled.record.at,
client_cert_pem,
client_key_pem,
issuer_cert_pem,
})
}
}
fn unavailable(err: gunnar_core::Error) -> ControlError {
ControlError::Unavailable(format!("the identity table could not be read: {err}"))
}
fn awaited_twin(stores: &ServerStores) -> String {
#[cfg(feature = "wg-client")]
if let Some(twin) = crate::pairing::recharge_awaited(stores) {
return twin;
}
#[cfg(feature = "replication")]
if let Some(twin) = crate::repair::held(stores) {
return twin;
}
let _ = stores;
String::new()
}
fn enrolment_error(err: gunnar_core::Error) -> ControlError {
match err {
e @ (gunnar_core::Error::AlreadyBootstrapped { .. }
| gunnar_core::Error::AlreadyProvisioned { .. }) => {
ControlError::FailedPrecondition(e.to_string())
}
e @ (gunnar_core::Error::InvalidPath { .. }
| gunnar_core::Error::ReservedPrincipalId(_)
| gunnar_core::Error::Key(_)) => ControlError::InvalidArgument(e.to_string()),
other => ControlError::Unavailable(other.to_string()),
}
}
impl CoreControlPlane {
pub fn claim_operator_named(
&self,
fed: Option<&str>,
asked: &str,
) -> ControlResult<gunnar_grpc::OperatorClaim> {
let (name, named_by) = gunnar_grpc::seam::operator_name(fed, asked)?;
let auth = self.stores.auth();
let held = match auth.principal(name).map_err(internal)? {
Some(p) => Some(p),
None => auth.principal_by_name(name).map_err(internal)?,
};
if let Some(p) = held {
if p.kind != PrincipalKind::User {
return Err(ControlError::FailedPrecondition(format!(
"the operator's name {name:?} is held by principal {:?}, which is not a person \
({:?}); an estate's machines are never its operator",
p.id, p.kind
)));
}
let adopted_namespace = false;
if self
.stores
.catalog()
.namespace(&p.id)
.map_err(internal)?
.is_none()
{
self.stores
.catalog()
.create_namespace(&p.id)
.map_err(internal)?;
}
self.grant(&p.id, "", "", GrantLevel::Admin)?;
return Ok(gunnar_grpc::OperatorClaim {
principal: principal_out(&p),
created: false,
adopted_namespace,
named_by,
});
}
let principal =
gunnar_core::Principal::new(name, name, PrincipalKind::User).map_err(internal)?;
let created = auth.add_principal(principal).map_err(internal)?;
let adopted_namespace = match self.stores.catalog().create_namespace(&created.id) {
Ok(_) => false,
Err(gunnar_core::Error::NamespaceExists(_)) => true,
Err(err) => {
let _ = auth.remove_principal(&created.id);
return Err(internal(err));
}
};
if let Err(err) = self.grant(&created.id, "", "", GrantLevel::Admin) {
let _ = auth.remove_principal(&created.id);
return Err(err);
}
self.stores.mirror_identity();
tracing::warn!(
target: "gunnar::audit",
event = "operator.claimed",
operator = %created.id,
adopted_namespace,
named_by,
"the estate's operator was created: a person, keyless, gunnar-admin, with its account \
namespace{}",
if adopted_namespace {
" — the namespace already existed with no owner (a first push conjured it) and is \
now the operator's account"
} else {
""
}
);
Ok(gunnar_grpc::OperatorClaim {
principal: principal_out(&created),
created: true,
adopted_namespace,
named_by,
})
}
}
mod enterprise_plane;
impl ControlPlane for CoreControlPlane {
fn enterprise(&self) -> Option<&dyn gunnar_grpc::enterprise::EnterprisePlane> {
Some(self)
}
fn create_namespace(&self, name: &str, _description: &str) -> ControlResult<Namespace> {
if self
.stores
.auth()
.employer_of(name)
.map_err(internal)?
.is_some()
{
return Err(ControlError::AlreadyExists(format!(
"{name} is an employee's id, and an employee holds no account of its own"
)));
}
let namespace = self
.stores
.catalog()
.create_namespace(name)
.map_err(internal)?;
Ok(self.namespace_out(namespace, 0))
}
fn list_namespaces(&self) -> ControlResult<Vec<Namespace>> {
let mut employees: std::collections::BTreeMap<String, u32> = Default::default();
for principal in self.stores.auth().principals().map_err(internal)? {
if let Some(account) = self
.stores
.auth()
.employer_of(&principal.id)
.map_err(internal)?
{
*employees.entry(account).or_default() += 1;
}
}
Ok(self
.stores
.catalog()
.namespaces()
.map_err(internal)?
.into_iter()
.map(|ns| {
let count = employees.get(&ns.name).copied().unwrap_or(0);
self.namespace_out(ns, count)
})
.collect())
}
fn namespace(&self, name: &str) -> ControlResult<Option<Namespace>> {
let Some(ns) = self.stores.catalog().namespace(name).map_err(internal)? else {
return Ok(None);
};
let count = self.employee_count(name)?;
Ok(Some(self.namespace_out(ns, count)))
}
fn namespace_for_outsider(
&self,
principal_id: &str,
name: &str,
) -> ControlResult<Option<Namespace>> {
let auth = self.stores.auth();
if auth
.gate_covering(principal_id)
.map_err(internal)?
.is_some()
|| auth.gate_covering(name).map_err(internal)?.is_some()
{
return Ok(None);
}
self.namespace(name)
}
fn set_namespace_settings(
&self,
name: &str,
public: Option<bool>,
events_public: Option<bool>,
) -> ControlResult<Namespace> {
let catalog = self.stores.catalog();
let mut ns = catalog
.namespace(name)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("namespace {name}")))?;
if let Some(public) = public {
ns = catalog
.set_namespace_public(name, public)
.map_err(internal)?;
}
if let Some(events_public) = events_public {
ns = catalog
.set_namespace_events_public(name, events_public)
.map_err(internal)?;
}
let count = self.employee_count(name)?;
Ok(self.namespace_out(ns, count))
}
fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32> {
if self.stores.auth().is_clan(name).map_err(internal)? {
return Err(ControlError::FailedPrecondition(format!(
"namespace {name} is clan {name}'s: it goes when the clan's admin dissolves the clan"
)));
}
let gone = self
.stores
.catalog()
.delete_namespace(name, delete_repos)
.map_err(internal)?;
if let Err(err) = self.stores.flow().forget(name) {
tracing::warn!(
target: "gunnar::store",
event = "flow.forget_failed",
namespace = %name,
%err,
"the namespace was deleted but its activity series could not be removed"
);
}
Ok(gone.len() as u32)
}
fn create_repo(&self, spec: ProtoNewRepo) -> ControlResult<Repo> {
let path = Self::repo_ref(&spec.path)?;
if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
return Err(ControlError::ResourceExhausted(line));
}
let mut new = gunnar_core::NewRepo::new(path).visibility(visibility_in(spec.visibility)?);
if spec.hash_kind != gunnar_proto::HashKind::Unspecified {
new = new.hash_kind(match spec.hash_kind {
gunnar_proto::HashKind::Sha1 => HashKind::Sha1,
_ => HashKind::Sha256,
});
}
if !spec.default_branch.is_empty() {
new = new.default_branch(spec.default_branch.clone());
}
let repo = self
.stores
.create_repo(new, spec.adopt)
.map_err(|err| create_error(&err, &spec.path))?;
Ok(repo_out(&repo))
}
fn write_first_commit(&self, path: &RepoPath, readme: &str) -> ControlResult<String> {
let (repo, store) = self.open(path)?;
let kind = match repo.hash_kind {
HashKind::Sha1 => gix_hash::Kind::Sha1,
_ => gix_hash::Kind::Sha256,
};
let when = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0);
let (pack, commit) = crate::first_commit::first_commit_pack(
kind,
"README.md",
readme.as_bytes(),
"Welcome",
when,
)
.map_err(ControlError::InvalidArgument)?;
store
.serve()
.put(
&pack,
&[gunnar_store::git_storage_trait::RefUpdate::set(
repo.default_branch.clone(),
commit.clone(),
)],
)
.map_err(|err| {
ControlError::Internal(format!("writing the first commit of {path}: {err:#}"))
})?;
Ok(commit)
}
fn create_room(&self, namespace: &str, machine: bool) -> ControlResult<()> {
use gunnar_ssh::{PlanAction, PlanVerdict};
for gate in self
.stores
.plan_gates()
.iter()
.filter(|gate| !machine || gate.applies_to_machine())
{
if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
return Err(ControlError::FailedPrecondition(line));
}
if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
return Err(ControlError::ResourceExhausted(line));
}
}
Ok(())
}
fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.as_ref()
.map(repo_out)
.ok_or_else(|| ControlError::NotFound(format!("repository {path}")))
}
fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>> {
Ok(self
.stores
.catalog()
.list(namespace)
.map_err(internal)?
.iter()
.map(repo_out)
.collect())
}
fn delete_repo(
&self,
path: &RepoPath,
options: gunnar_grpc::seam::DeleteOptions,
) -> ControlResult<gunnar_grpc::seam::DeleteOutcome> {
let repo_ref = Self::repo_ref(path)?;
let was = self.stores.catalog().get(&repo_ref).map_err(internal)?;
let orphan = was.is_none() && self.stores.is_orphan(&repo_ref).unwrap_or(false);
if orphan && !options.purge {
return Err(ControlError::FailedPrecondition(
crate::stores::ServerStores::orphan_soft_delete_refusal(&repo_ref),
));
}
match self.stores.catalog().delete(&repo_ref) {
Ok(_) => {}
Err(gunnar_core::Error::NoSuchRepo(_))
if orphan || self.stores.is_tombstoned(&repo_ref).unwrap_or(false) => {}
Err(err) => return Err(internal(err)),
}
if options.purge {
let (before, after) = self.stores.purge_repo(&repo_ref).map_err(|err| {
ControlError::Internal(format!("removing the repository's directory: {err:#}"))
})?;
return Ok(gunnar_grpc::seam::DeleteOutcome::Purged {
bytes_reclaimed: before.saturating_sub(after),
});
}
let retention = std::time::Duration::from_secs(
u64::from(
options
.retention_days
.unwrap_or(crate::stores::DEFAULT_DELETE_RETENTION_DAYS),
) * 86_400,
);
let record = self
.stores
.soft_delete(&repo_ref, was.as_ref(), &options.principal_id, retention)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
let bytes_kept = self
.stores
.repo_dir(&repo_ref)
.map(|dir| crate::stores::dir_bytes(&dir))
.unwrap_or(0);
Ok(gunnar_grpc::seam::DeleteOutcome::Tombstoned {
purge_after_unix_ms: record.purge_after_unix_ms.unwrap_or_default(),
bytes_kept,
})
}
fn undelete_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_some()
{
return Err(ControlError::AlreadyExists(format!(
"{path} is in the catalog already; it is not deleted"
)));
}
let spec = self
.stores
.undelete_spec(&repo_ref)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
let repo = self
.stores
.create_repo(spec, true)
.map_err(|err| ControlError::Internal(format!("restoring {path}: {err:#}")))?;
self.stores
.clear_tombstone(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
Ok(repo_out(&repo))
}
fn purge_repo(&self, path: &RepoPath) -> ControlResult<u64> {
let repo_ref = Self::repo_ref(path)?;
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_some()
{
return Err(ControlError::FailedPrecondition(format!(
"{path} is a LIVE repository, not a deleted one. `repo delete {path} --purge` \
deletes and reclaims it in one step; purge alone only ends a retention window."
)));
}
let tombstone = self
.stores
.tombstone_of(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
if tombstone.is_none() {
return Err(ControlError::NotFound(format!(
"{path} is neither catalogued nor tombstoned; there is nothing to purge"
)));
}
let (before, after) = self
.stores
.purge_repo(&repo_ref)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
Ok(before.saturating_sub(after))
}
fn gc_repo(&self, path: &RepoPath) -> ControlResult<gunnar_grpc::seam::GcJob> {
let repo_ref = Self::repo_ref(path)?;
if self
.stores
.catalog()
.get(&repo_ref)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!(
"{path} is not a live repository; nothing to gc"
)));
}
Ok(gc_job_out(path, self.gc.enqueue(&repo_ref.to_string())))
}
fn gc_status(&self, path: &RepoPath) -> ControlResult<Option<gunnar_grpc::seam::GcJob>> {
let repo_ref = Self::repo_ref(path)?;
Ok(self
.gc
.status(&repo_ref.to_string())
.map(|job| gc_job_out(path, job)))
}
fn gate_account(
&self,
account: &str,
actor: &str,
note: &str,
) -> ControlResult<(gunnar_grpc::seam::AccountGate, bool)> {
let before = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.is_some();
let gate = self
.stores
.auth()
.gate(gunnar_core::AccountGate::administrative(
account, actor, note,
))
.map_err(gate_error)?;
tracing::warn!(
target: "gunnar::audit",
event = "account.gated",
account = %account,
actor = %actor,
note = %note,
already = before,
"an account was locked out: every level, both doors, until it is ungated"
);
Ok((gate_out(gate), before))
}
fn ungate_account(
&self,
account: &str,
) -> ControlResult<Option<gunnar_grpc::seam::AccountGate>> {
let lifted = self
.stores
.auth()
.lift_gate(account)
.map_err(internal)?
.map(gate_out);
if let Some(gate) = &lifted {
tracing::warn!(
target: "gunnar::audit",
event = "account.ungated",
account = %account,
held_by = %gate.holder,
"an account lockout was lifted"
);
}
Ok(lifted)
}
fn account_gates(&self) -> ControlResult<Vec<gunnar_grpc::seam::AccountGate>> {
Ok(self
.stores
.auth()
.gates()
.map_err(internal)?
.into_iter()
.map(gate_out)
.collect())
}
fn purge_account(
&self,
account: &str,
options: gunnar_grpc::seam::PurgeAccountOptions,
) -> ControlResult<(gunnar_grpc::seam::PurgeJob, gunnar_grpc::seam::AccountGate)> {
let account = account.trim();
if account.is_empty() {
return Err(ControlError::InvalidArgument(
"an account purge needs an account name".into(),
));
}
if let Some(p) = self.stores.auth().principal(account).map_err(internal)? {
if p.kind != gunnar_core::PrincipalKind::User {
return Err(ControlError::FailedPrecondition(format!(
"{account} is a machine principal ({:?}), not a person: a machine's account is never \
purged. Remove the principal itself if it is really to go.",
p.kind
)));
}
}
let job_id = self.purge.next_job_id();
self.gate_for_purge(account, &job_id, &options)?;
let order = crate::purge_queue::PurgeOrder::new(account, &job_id, &options.reason)
.by(&options.principal_id)
.removing_principal(options.remove_principal)
.expecting_empty(options.expect_empty);
let job = self.purge.enqueue(order);
let gate = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.map(gate_out)
.unwrap_or_else(|| gunnar_grpc::seam::AccountGate {
account: account.to_owned(),
reason: gunnar_grpc::seam::GateReason::Purge,
holder: job_id.clone(),
actor: options.principal_id.clone(),
note: "the lockout was lifted between being taken and being read back".into(),
at_unix_ms: gunnar_core::now_ms(),
expires_unix_ms: 0,
});
Ok((purge_job_out(job), gate))
}
fn purge_account_status(
&self,
account: &str,
) -> ControlResult<(
Option<gunnar_grpc::seam::PurgeJob>,
Option<gunnar_grpc::seam::AccountGate>,
)> {
let job = self.purge.status(account).map(purge_job_out);
let gate = self
.stores
.auth()
.gate_covering(account)
.map_err(internal)?
.map(gate_out);
Ok((job, gate))
}
fn default_delete_retention_days(&self) -> u32 {
crate::stores::DEFAULT_DELETE_RETENTION_DAYS
}
fn list_deleted(
&self,
namespace: Option<&str>,
) -> ControlResult<Vec<gunnar_grpc::seam::DeletedRepoRow>> {
Ok(self
.stores
.tombstones()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.into_iter()
.filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
.map(|row| gunnar_grpc::seam::DeletedRepoRow {
path: RepoPath {
namespace: row.path.namespace().to_owned(),
repo: row.path.name().to_owned(),
},
deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
principal_id: row.tombstone.principal_id,
bytes: row.bytes,
})
.collect())
}
fn list_orphans(
&self,
namespace: Option<&str>,
) -> ControlResult<Vec<gunnar_grpc::seam::OrphanRow>> {
Ok(self
.stores
.orphans()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.into_iter()
.filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
.map(|row| gunnar_grpc::seam::OrphanRow {
path: RepoPath {
namespace: row.path.namespace().to_owned(),
repo: row.path.name().to_owned(),
},
dir: row.dir.display().to_string(),
bytes: row.bytes,
hash_marker: row.hash_marker.unwrap_or_default(),
modified_unix_ms: row.modified_unix_ms,
failed_reclaim: row.failed_reclaim.unwrap_or_default(),
})
.collect())
}
fn reclaim_orphan(&self, path: &RepoPath) -> ControlResult<u64> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.reclaim_orphan(&repo_ref)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn index_progress(&self) -> ControlResult<Vec<gunnar_grpc::seam::IndexProgressRow>> {
Ok(self
.stores
.engine()
.index_progress()
.into_iter()
.filter_map(|(dir, s)| {
let (namespace, name) = self.stores.repo_of_dir(&dir)?;
Some(gunnar_grpc::seam::IndexProgressRow {
namespace,
name,
pack_id: s.pack_id,
objects_done: s.objects_done,
objects_total: s.objects_total,
bytes_done: s.bytes_done,
bytes_total: s.bytes_total,
phase: if s.active {
s.phase.to_string()
} else {
"idle".to_string()
},
started_at_unix_ms: s.started_at_ms as i64,
resumed: s.resumed,
reinflations: s.reinflations,
pauses: s.pauses,
})
})
.filter(|r| r.phase != "idle")
.collect())
}
fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
let repo = self
.stores
.catalog()
.set_default_branch(&repo_ref, default_branch)
.map_err(internal)?;
self.stores.repoint_head(&repo).map_err(|err| {
ControlError::Internal(format!("pointing HEAD at the new default branch: {err:#}"))
})?;
Ok(repo_out(&repo))
}
fn set_visibility(
&self,
path: &RepoPath,
visibility: gunnar_proto::Visibility,
) -> ControlResult<Repo> {
if visibility == gunnar_proto::Visibility::Unspecified {
return Err(ControlError::InvalidArgument(
"visibility unspecified".into(),
));
}
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.set_visibility(&repo_ref, visibility_in(visibility)?)
.map(|repo| repo_out(&repo))
.map_err(internal)
}
fn recipient_keys(&self, principal_id: &str) -> ControlResult<Vec<gunnar_proto::RecipientKey>> {
let public = self
.stores
.encryption_keys()
.public_of(principal_id)
.map_err(key_refusal)?;
Ok(public
.into_iter()
.map(|public| gunnar_proto::RecipientKey {
fingerprint: gunnar_core::auth::fingerprint_of_blob(public.as_bytes()).to_string(),
algo: if public.starts_with("age1pq1") {
gunnar_core::auth::PQ_RECIPIENT_ALGO.to_owned()
} else {
"X25519".to_owned()
},
openssh: public,
})
.collect())
}
fn encryption_key(
&self,
principal_id: &str,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
Ok(encryption_key_out(
self.stores
.encryption_keys()
.get(principal_id)
.map_err(key_refusal)?,
))
}
fn create_encryption_key(
&self,
principal_id: &str,
public_key: &str,
first: gunnar_proto::UnlockMethod,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let first = method_in(first)?;
let key = self
.stores
.encryption_keys()
.create(
principal_id,
public_key,
first,
gunnar_core::now_ms().max(0) as u64,
)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn add_unlock_method(
&self,
principal_id: &str,
method: gunnar_proto::UnlockMethod,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let method = method_in(method)?;
let key = self
.stores
.encryption_keys()
.add_method(principal_id, method, gunnar_core::now_ms().max(0) as u64)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn remove_unlock_method(
&self,
principal_id: &str,
id: &str,
) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
let key = self
.stores
.encryption_keys()
.remove_method(principal_id, id)
.map_err(key_refusal)?;
Ok(encryption_key_out(Some(key)))
}
fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
let repo_ref = Self::repo_ref(path)?;
self.stores
.catalog()
.set_encrypted(&repo_ref, encrypted)
.map(|repo| repo_out(&repo))
.map_err(internal)
}
#[cfg(feature = "engine-znippy")]
fn archive_out(
&self,
path: &RepoPath,
from_generation: u64,
sink: &mut dyn std::io::Write,
) -> ControlResult<gunnar_grpc::seam::ArchiveCut> {
let repo_ref = Self::repo_ref(path)?;
let dir = self
.stores
.repo_dir(&repo_ref)
.map_err(|err| ControlError::Internal(format!("resolving {path}: {err:#}")))?;
if !dir.is_dir() {
return Err(ControlError::NotFound(format!(
"{path} has nothing on disk at {}",
dir.display()
)));
}
let receipt =
crate::storage::znippy_engine::archive_out_to(&dir, &mut { sink }, from_generation)
.map_err(archive_error)?;
self.stores.flow().record(
&path.namespace,
gunnar_core::now_ms(),
&gunnar_core::Transfer {
bytes_out: receipt.bytes_out,
fetched: true,
..gunnar_core::Transfer::default()
},
);
Ok(gunnar_grpc::seam::ArchiveCut {
from_generation: receipt.from_generation,
to_generation: receipt.to_generation,
bytes_out: receipt.bytes_out,
file_sha256: receipt.file_sha256,
})
}
fn list_refs(
&self,
path: &RepoPath,
prefix: &str,
peel_them: bool,
) -> ControlResult<Vec<RefTarget>> {
let (_, store) = self.open(path)?;
let git = store.serve();
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let mut out = Vec::new();
for row in snapshot.with_prefix(Some(prefix)) {
out.push(ref_target_out(
git,
store.hash_kind(),
&snapshot,
row,
peel_them,
)?);
}
Ok(out)
}
fn get_ref(
&self,
path: &RepoPath,
ref_name: &str,
peel_them: bool,
) -> ControlResult<RefTarget> {
let (_, store) = self.open(path)?;
let git = store.serve();
gunnar_store::refname::validate(ref_name)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let head_row;
let row = if ref_name == gunnar_store::HEAD {
head_row = git
.head()
.map_err(store_error)?
.ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?;
&head_row
} else {
snapshot
.get(ref_name)
.ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?
};
ref_target_out(git, store.hash_kind(), &snapshot, row, peel_them)
}
fn log(
&self,
path: &RepoPath,
start: &str,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
self.log_walk(path, LogStart::One(start), limit, touching)
}
fn log_all_refs(
&self,
path: &RepoPath,
limit: u32,
touching: &str,
) -> ControlResult<Vec<Commit>> {
self.log_walk(path, LogStart::AllRefs, limit, touching)
}
fn create_pull(
&self,
path: &RepoPath,
author: &str,
title: &str,
description: &str,
source_ref: &str,
target_ref: &str,
) -> ControlResult<PullRequest> {
for (what, name) in [("source_ref", source_ref), ("target_ref", target_ref)] {
if !name.starts_with("refs/heads/") {
return Err(ControlError::InvalidArgument(format!(
"{what} must be a full branch name under refs/heads/, got {name:?}: a pull \
request merges branches, and a short name stored today is an ambiguity \
resolved differently tomorrow"
)));
}
}
if source_ref == target_ref {
return Err(ControlError::InvalidArgument(
"a branch cannot be merged into itself".to_owned(),
));
}
if title.trim().is_empty() {
return Err(ControlError::InvalidArgument(
"a title is required".to_owned(),
));
}
let repo_ref = Self::repo_ref(path)?;
let (_, store) = self.open(path)?;
let git = store.serve();
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let source = snapshot
.resolve(source_ref)
.map_err(wire_error)?
.ok_or_else(|| ControlError::NotFound(format!("{source_ref} in {path}")))?;
snapshot
.resolve(target_ref)
.map_err(wire_error)?
.ok_or_else(|| ControlError::NotFound(format!("{target_ref} in {path}")))?;
let record = self
.pulls()?
.create(
&repo_ref.to_string(),
author,
title,
description,
source_ref,
target_ref,
&source.to_string(),
)
.map_err(internal)?;
Ok(pull_out(path, record))
}
fn list_pulls(&self, path: &RepoPath, state: i32) -> ControlResult<Vec<PullRequest>> {
let repo_ref = Self::repo_ref(path)?;
self.get_repo(path)?;
let filter = match PullRequestState::try_from(state) {
Ok(PullRequestState::Unspecified) | Err(_) => None,
Ok(PullRequestState::Open) => Some(gunnar_core::PullState::Open),
Ok(PullRequestState::Merged) => Some(gunnar_core::PullState::Merged),
Ok(PullRequestState::Closed) => Some(gunnar_core::PullState::Closed),
};
Ok(self
.pulls()?
.list(&repo_ref.to_string(), filter)
.map_err(internal)?
.into_iter()
.map(|record| pull_out(path, record))
.collect())
}
fn get_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
let repo_ref = Self::repo_ref(path)?;
self.get_repo(path)?;
self.pulls()?
.get(&repo_ref.to_string(), number)
.map_err(internal)?
.map(|record| pull_out(path, record))
.ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))
}
fn close_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
let repo_ref = Self::repo_ref(path)?;
self.get_repo(path)?;
match self
.pulls()?
.close(&repo_ref.to_string(), number)
.map_err(internal)?
{
gunnar_core::PullTransition::Done(record) => Ok(pull_out(path, record)),
gunnar_core::PullTransition::NotFound => Err(ControlError::NotFound(format!(
"pull request #{number} in {path}"
))),
gunnar_core::PullTransition::NotOpen(state) => Err(ControlError::FailedPrecondition(
format!("pull request #{number} is already {state:?}, not open"),
)),
}
}
fn merge_pull(
&self,
path: &RepoPath,
number: u64,
principal_id: &str,
) -> ControlResult<(PullRequest, MergedRef)> {
let repo_ref = Self::repo_ref(path)?;
let key = repo_ref.to_string();
let pulls = self.pulls()?;
let record = pulls
.get(&key, number)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))?;
if record.state != gunnar_core::PullState::Open {
return Err(ControlError::FailedPrecondition(format!(
"pull request #{number} is already {:?}, not open",
record.state
)));
}
let (repo_record, store) = self.open(path)?;
let git = store.serve();
let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
let source = snapshot
.resolve(&record.source_ref)
.map_err(wire_error)?
.ok_or_else(|| {
ControlError::FailedPrecondition(format!(
"the source branch {} is gone; nothing to merge",
record.source_ref
))
})?;
let target = snapshot
.resolve(&record.target_ref)
.map_err(wire_error)?
.ok_or_else(|| {
ControlError::FailedPrecondition(format!(
"the target branch {} is gone",
record.target_ref
))
})?;
let head = source.to_string();
let signed = self.forge_events()?;
let events: Vec<gunnar_core::ForgeEvent> = signed
.iter()
.filter(|s| s.event.repo == key)
.map(|s| s.event.clone())
.collect();
let review = gunnar_core::forge::fold(number, &events);
let approved = self.review_state(&key, &review, &signed)?;
let ok = approved
.approvals
.iter()
.any(|a| a.verified && a.commit == head);
if !ok {
let named: Vec<String> = approved
.approvals
.iter()
.map(|a| {
format!(
"{} → {} [{}]",
a.actor,
&a.commit[..12.min(a.commit.len())],
a.verdict
)
})
.collect();
return Err(ControlError::FailedPrecondition(format!(
"no verified approval names {}, the commit {} is at. gunnar does not merge on \
an approval of something else: an approval names the commit it approved, so a \
branch that moved after the review has not been reviewed.{}",
&head[..12.min(head.len())],
record.source_ref,
if named.is_empty() {
" Nobody has approved this pull request.".to_owned()
} else {
format!(" What is approved: {}", named.join(", "))
}
)));
}
let policy = self.policy_of(&repo_record)?;
let rules: Vec<&gunnar_policy::ProtectionRule> = policy
.protection
.iter()
.filter(|rule| rule.pattern.matches_ref(&record.target_ref))
.collect();
let forbid_self = rules.iter().any(|rule| rule.forbid_self_approval);
let required = rules
.iter()
.map(|rule| rule.required_approvals)
.max()
.unwrap_or(0)
.max(1) as usize;
let approvers: Vec<&str> = approved
.approvals
.iter()
.filter(|a| a.verified && a.commit == head)
.map(|a| a.actor.as_str())
.collect();
let counted: Vec<&str> = approvers
.iter()
.copied()
.filter(|actor| !(forbid_self && *actor == record.author))
.collect();
if forbid_self && counted.is_empty() && !approvers.is_empty() {
return Err(ControlError::FailedPrecondition(format!(
"the protection on {} forbids self-approval, and the only verified approval \
of {} is the author's own ({}). An author's sign-off on their own work is \
not a review; somebody else has to look at it.",
record.target_ref,
&head[..12.min(head.len())],
record.author
)));
}
if counted.len() < required {
return Err(ControlError::FailedPrecondition(format!(
"the protection on {} requires {required} distinct verified approvals of {} \
and there {} ({}). Approvals by the same principal count once{}.",
record.target_ref,
&head[..12.min(head.len())],
if counted.len() == 1 {
"is 1".to_owned()
} else {
format!("are {}", counted.len())
},
if counted.is_empty() {
"nobody".to_owned()
} else {
counted.join(", ")
},
if forbid_self {
", and the author's own does not count here"
} else {
""
}
)));
}
if source != target {
let principal = self
.stores
.auth()
.principal(principal_id)
.map_err(internal)?
.ok_or_else(|| {
ControlError::FailedPrecondition(format!(
"the principal {principal_id} is no longer in the identity table; \
refusing to move a protected reference for an identity that \
cannot be resolved"
))
})?;
let gate = gunnar_policy::BranchProtectionGate::new(policy.protection.clone());
let repo_id = gunnar_policy::RepoId::new(&path.namespace, &path.repo);
let policy_principal = gunnar_policy::Principal {
id: principal.id.clone(),
name: principal.name.clone(),
kind: crate::transport::policy_principal_kind(principal.kind),
};
let introduced = gunnar_policy::ancestry::commits_between(
git,
&source,
&target,
gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
)
.map_err(|err| {
ControlError::FailedPrecondition(format!(
"the commits this merge would add could not be enumerated ({err}); \
refusing rather than running branch protection over an empty set"
))
})?;
let updates = [gunnar_policy::RefUpdate::from_wire(
record.target_ref.clone(),
target,
source,
)];
let keys = crate::transport::key_directory(&self.stores);
let ctx = gunnar_policy::PushContext::new(
&policy_principal,
&repo_id,
&updates,
&introduced,
git,
&keys,
);
match gunnar_policy::Gate::check(&gate, &ctx) {
Ok(gunnar_policy::Verdict::Allow) => {}
Ok(verdict) => {
return Err(ControlError::FailedPrecondition(format!(
"branch protection refuses this merge: {}",
verdict.reason().unwrap_or_default()
)));
}
Err(err) => {
return Err(ControlError::FailedPrecondition(format!(
"branch protection could not decide this merge ({err}); refusing rather \
than guessing"
)));
}
}
}
if source != target {
match gunnar_policy::ancestry::is_descendant_of(
git,
&source,
&target,
gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET,
) {
Ok(true) => {}
Ok(false) => {
return Err(ControlError::FailedPrecondition(format!(
"not a fast-forward: {} does not descend from {}. gunnar has no merge \
engine yet — rebase or merge locally, push the result, and merge again",
record.source_ref, record.target_ref
)));
}
Err(err) => {
return Err(ControlError::FailedPrecondition(format!(
"the ancestry walk could not be completed ({err}); refusing rather than \
guessing"
)));
}
}
git.update_ref(
&record.target_ref,
Some(target.as_bytes()),
Some(source.as_bytes()),
)
.map_err(|err| {
ControlError::FailedPrecondition(format!(
"the target ref moved while merging ({err}); nothing was written — retry \
against the new tip"
))
})?;
}
let merged = match pulls
.mark_merged(&key, number, &source.to_string())
.map_err(internal)?
{
gunnar_core::PullTransition::Done(record) => record,
other => {
return Err(ControlError::Internal(format!(
"the ref moved to {source} but the record could not be marked merged \
({other:?}); the repository is correct, the record is not"
)));
}
};
Ok((
pull_out(path, merged),
MergedRef {
ref_name: record.target_ref.clone(),
old_oid: target.to_string(),
new_oid: source.to_string(),
},
))
}
fn read_object(
&self,
path: &RepoPath,
oid: &str,
max_bytes: u64,
) -> ControlResult<ReadObjectResponse> {
let (_, store) = self.open(path)?;
let git = store.serve();
let id = gix_hash::ObjectId::from_hex(oid.as_bytes())
.map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
if let Some((_, object_bytes)) = git.header(id.as_bytes()).map_err(store_error)? {
if self.max_object_bytes != 0 && object_bytes > self.max_object_bytes {
return Err(ControlError::ResourceExhausted(format!(
"object {oid} is {object_bytes} bytes, over this server's {}-byte ceiling \
for reading an object. Reading any part of it costs the whole object in \
memory, because a prefix of a delta-encoded object needs its entire base \
chain, so no max_bytes makes this request cheaper.",
self.max_object_bytes
)));
}
let response_bytes = if max_bytes == 0 {
object_bytes
} else {
object_bytes.min(max_bytes)
};
if response_bytes > MAX_RESPONSE_BYTES {
return Err(ControlError::ResourceExhausted(format!(
"object {oid} is {object_bytes} bytes and the response would be \
{response_bytes}, over the {MAX_RESPONSE_BYTES}-byte limit a gRPC client \
can receive by default. Ask for a prefix with max_bytes."
)));
}
}
let Some((kind, mut buf)) = git.read(id.as_bytes()).map_err(store_error)? else {
return Err(ControlError::NotFound(format!("object {oid} in {path}")));
};
let size = buf.len() as u64;
let truncated = max_bytes != 0 && size > max_bytes;
if truncated {
buf.truncate(max_bytes as usize);
}
Ok(ReadObjectResponse {
oid: id.to_string(),
kind: object_kind_out(kind)? as i32,
size,
data: buf,
truncated,
})
}
fn read_lfs_object(
&self,
path: &RepoPath,
oid: &str,
max_bytes: u64,
range: Option<(u64, u64)>,
) -> ControlResult<ReadLfsObjectResponse> {
let (_, store) = self.open(path)?;
let id = gunnar_lfs::Oid::parse(oid)
.map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
use gunnar_lfs::LfsStore as _;
let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir()).map_err(|err| {
ControlError::Internal(format!("opening the repository's LFS store: {err}"))
})?;
let ceiling = match max_bytes {
0 => MAX_RESPONSE_BYTES,
n => n.min(MAX_RESPONSE_BYTES),
};
let size = lfs
.size_of(&id)
.map_err(|err| ControlError::Internal(format!("reading the LFS store: {err}")))?
.ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
if let Some((offset, length)) = range {
if offset > size || (offset == size && size > 0) {
return Err(ControlError::InvalidArgument(format!(
"offset {offset} is past the end of LFS object {oid}, which is {size} bytes"
)));
}
if length > ceiling {
return Err(ControlError::ResourceExhausted(format!(
"a range of {length} bytes of LFS object {oid} is over the {ceiling}-byte ceiling \
for this request; ask for {ceiling} bytes or fewer at a time"
)));
}
let want = if length == 0 { ceiling } else { length }.min(size - offset);
let object = lfs
.open_at(&id, offset)
.map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
.ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
let mut data = Vec::with_capacity(want as usize);
let mut reader = std::io::Read::take(object.reader, want);
std::io::Read::read_to_end(&mut reader, &mut data).map_err(|err| {
ControlError::Internal(format!("reading LFS object {oid}: {err}"))
})?;
if (data.len() as u64) < want {
return Err(ControlError::Internal(format!(
"LFS object {oid} ended {} bytes into a {want}-byte range at {offset}",
data.len()
)));
}
return Ok(ReadLfsObjectResponse {
oid: oid.to_owned(),
size: object.size,
data,
offset,
});
}
if size > ceiling {
return Err(ControlError::ResourceExhausted(format!(
"LFS object {oid} is {size} bytes, over the {ceiling}-byte ceiling for this \
request. An LFS object is served whole or not at all — a prefix of one is not a \
smaller file — so no max_bytes makes this request smaller. `git lfs` streams it, \
and a ranged read (offset/length) reads it a range at a time."
)));
}
let object = lfs
.open(&id)
.map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
.ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
if object.size > ceiling {
return Err(ControlError::ResourceExhausted(format!(
"LFS object {oid} is {} bytes, over the {ceiling}-byte ceiling for this request",
object.size
)));
}
let mut data = Vec::with_capacity(object.size as usize);
let mut reader = std::io::Read::take(object.reader, object.size);
std::io::Read::read_to_end(&mut reader, &mut data)
.map_err(|err| ControlError::Internal(format!("reading LFS object {oid}: {err}")))?;
Ok(ReadLfsObjectResponse {
oid: oid.to_owned(),
size: data.len() as u64,
data,
offset: 0,
})
}
fn diff(
&self,
path: &RepoPath,
from: &str,
to: &str,
path_prefixes: &[String],
) -> ControlResult<Vec<FileChange>> {
let (_repo, store) = self.open(path)?;
let git = store.serve();
let hash = store.hash_kind();
let resolve = |spec: &str| -> ControlResult<Option<gix_hash::ObjectId>> {
if spec.is_empty() {
return Ok(None);
}
Ok(Some(self.tree_of(git, hash, path, spec)?))
};
let old = resolve(from)?;
let new = resolve(to)?;
crate::diff::compare(git, hash, old, new, path_prefixes).map_err(|err| match err {
crate::diff::DiffError::Missing(oid) => {
ControlError::NotFound(format!("object {oid} in {path}"))
}
crate::diff::DiffError::WrongKind { .. } => {
ControlError::InvalidArgument(err.to_string())
}
crate::diff::DiffError::Corrupt { .. } | crate::diff::DiffError::Backend(_) => {
ControlError::Internal(err.to_string())
}
})
}
fn add_principal(
&self,
id: &str,
name: &str,
kind: gunnar_proto::PrincipalKind,
) -> ControlResult<Principal> {
let principal =
gunnar_core::Principal::new(id, name, principal_kind_in(kind)).map_err(internal)?;
let created = self
.stores
.auth()
.add_principal(principal)
.map(|p| principal_out(&p))
.map_err(internal)?;
if let Err(err) = self.stores.catalog().create_namespace(id) {
let _ = self.stores.auth().remove_principal(id);
return Err(ControlError::AlreadyExists(format!(
"cannot create the account namespace \"{id}\" for this user: {err}. A user's \
account is the namespace named after their id and cannot be somebody else's; \
the principal was not created."
)));
}
self.stores.mirror_identity();
Ok(created)
}
fn enroll_employee(
&self,
account: &str,
ask: &gunnar_grpc::EmployeeAsk,
) -> ControlResult<gunnar_grpc::EmployeeAnswer> {
if self
.stores
.catalog()
.namespace(account)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!("account {account}")));
}
if self.stores.auth().is_clan(account).map_err(internal)? {
return Err(ControlError::FailedPrecondition(format!(
"{account} is a clan, not an account: employees are enrolled by an account's admin, \
and a clan's members are accounts that join it"
)));
}
let principal = if ask.seats == 0 {
if self
.stores
.catalog()
.namespace(&ask.id)
.map_err(internal)?
.is_some()
{
return Err(ControlError::AlreadyExists(format!(
"namespace {} exists, so an employee may not be called that",
ask.id
)));
}
Some(
gunnar_core::Principal::new(&ask.id, &ask.name, gunnar_core::PrincipalKind::User)
.map_err(internal)?,
)
} else {
None
};
let room = gunnar_core::RoomAsk {
room_bytes: ask.room_bytes,
seats: ask.seats,
from_seat: ask.from_seat,
cap: ask.cap,
stored: ask.stored,
order: ask.order.clone(),
dry_run: ask.dry_run,
};
let answer = match self
.stores
.auth()
.enroll_with_room(account, principal, &room)
{
Ok(answer) => answer,
Err(gunnar_core::Error::RoomShort {
need,
unused,
short,
..
}) => {
let (seats, seats_open) = self.stores.auth().seats_of(account).map_err(internal)?;
return Ok(gunnar_grpc::EmployeeAnswer {
principal: None,
room_bytes: ask.room_bytes,
seats,
seats_open,
need_bytes: need,
unused_bytes: Some(unused),
shortfall_bytes: short,
unchanged: true,
});
}
Err(gunnar_core::Error::PrincipalExists(id)) => {
return Err(ControlError::AlreadyExists(format!(
"principal {id} already exists, and an existing principal can never become an \
employee"
)))
}
Err(err @ gunnar_core::Error::NoOpenSeat(_)) => {
return Err(ControlError::FailedPrecondition(err.to_string()))
}
Err(err @ gunnar_core::Error::NameExists { .. }) => {
return Err(ControlError::AlreadyExists(err.to_string()))
}
Err(err @ gunnar_core::Error::InvalidInput { .. }) => {
return Err(ControlError::InvalidArgument(err.to_string()))
}
Err(other) => return Err(internal(other)),
};
if !answer.unchanged {
self.stores.mirror_identity();
}
Ok(gunnar_grpc::EmployeeAnswer {
principal: answer.principal.as_ref().map(principal_out),
room_bytes: answer.room_bytes,
seats: answer.seats,
seats_open: answer.seats_open,
need_bytes: answer.need_bytes,
unused_bytes: answer.unused_bytes,
shortfall_bytes: answer.shortfall_bytes,
unchanged: answer.unchanged,
})
}
fn seats(&self, account: &str) -> ControlResult<(u32, u32)> {
self.stores.auth().seats_of(account).map_err(internal)
}
fn clans(&self) -> Option<&dyn gunnar_grpc::clans::ClanPlane> {
Some(self)
}
fn has_employees(&self, account: &str) -> ControlResult<bool> {
self.stores.auth().is_enterprise(account).map_err(internal)
}
fn employees(&self, account: &str) -> ControlResult<Vec<Principal>> {
Ok(self
.stores
.auth()
.employees_of(account)
.map_err(internal)?
.iter()
.map(principal_out)
.collect())
}
fn principal_by_name(&self, name: &str) -> ControlResult<Option<Principal>> {
Ok(self
.stores
.auth()
.principal_by_name(name)
.map_err(internal)?
.as_ref()
.map(principal_out))
}
fn operator_bootstrap_invite(
&self,
operator: &str,
secret_sha256: &str,
now_unix_ms: i64,
ttl_ms: i64,
) -> ControlResult<gunnar_grpc::BootstrapInvite> {
use gunnar_grpc::seam::OPERATOR_INVITE_ID;
let hash = gunnar_grpc::seam::bootstrap_hash(secret_sha256)?;
let ttl_ms = gunnar_grpc::seam::bootstrap_ttl(ttl_ms)?;
let auth = self.stores.auth();
if !auth
.web_credentials(Some(operator))
.map_err(internal)?
.is_empty()
{
if auth
.drop_first_invite(OPERATOR_INVITE_ID)
.map_err(internal)?
.is_some()
{
self.stores.mirror_identity();
}
return Ok(gunnar_grpc::BootstrapInvite::Spent);
}
let renewed = auth
.first_invite(OPERATOR_INVITE_ID)
.map_err(internal)?
.is_some_and(|row| row.secret_hash == hash && row.principal_id == operator);
auth.put_first_invite(gunnar_core::FirstInviteRow {
id: OPERATOR_INVITE_ID.to_owned(),
principal_id: operator.to_owned(),
secret_hash: hash,
created_by: "installer".to_owned(),
created: now_unix_ms,
expires: now_unix_ms + ttl_ms,
note: "the operator's first passkey: the installer's first invite".to_owned(),
server_level: None,
})
.map_err(internal)?;
self.stores.mirror_identity();
if !renewed {
tracing::warn!(
target: "gunnar::audit",
event = "operator.bootstrap_invite",
operator,
ttl_days = ttl_ms / 86_400_000,
"the installer's first invite (its hash) is recorded for the operator: his first passkey registers on his account, by gunnar's word"
);
}
Ok(if renewed {
gunnar_grpc::BootstrapInvite::Renewed
} else {
gunnar_grpc::BootstrapInvite::Recorded
})
}
fn claim_operator(&self, asked: &str) -> ControlResult<gunnar_grpc::OperatorClaim> {
#[cfg(feature = "wg-client")]
let fed = crate::fed_principals::fed_operator(self.stores.root());
#[cfg(not(feature = "wg-client"))]
let fed: Option<String> = None;
self.claim_operator_named(fed.as_deref(), asked)
}
fn rename_principal(&self, principal_id: &str, name: &str) -> ControlResult<Principal> {
let renamed = self
.stores
.auth()
.rename_principal(principal_id, name)
.map(|p| principal_out(&p))
.map_err(internal)?;
self.stores.mirror_identity();
Ok(renamed)
}
fn list_principals(&self) -> ControlResult<Vec<Principal>> {
Ok(self
.stores
.auth()
.principals()
.map_err(internal)?
.iter()
.map(principal_out)
.collect())
}
fn principal_activity(
&self,
) -> ControlResult<std::collections::BTreeMap<String, gunnar_grpc::PrincipalSeen>> {
Ok(self
.stores
.activity()
.all()
.into_iter()
.map(|(id, seen)| {
(
id,
gunnar_grpc::PrincipalSeen {
last_sign_in_unix_ms: seen.signed_in,
last_operation_unix_ms: seen.operated,
},
)
})
.collect())
}
fn holds_server_grant(&self, principal_id: &str) -> ControlResult<bool> {
self.stores
.auth()
.holds_server_grant(principal_id)
.map_err(internal)
}
fn add_credential(
&self,
principal_id: &str,
credential_id: &[u8],
public_key: &[u8],
label: &str,
sign_count: u32,
) -> ControlResult<gunnar_proto::WebCredential> {
let stored = self
.stores
.auth()
.add_web_credential(gunnar_core::WebCredential {
id: gunnar_grpc::webauthn::b64url(credential_id),
principal_id: principal_id.to_owned(),
label: label.to_owned(),
public_key_hex: hex::encode(public_key),
sign_count,
created: gunnar_core::now_ms(),
last_used: None,
sessions: Vec::new(),
})
.map_err(internal)?;
web_credential_out(&stored)
}
fn credentials(
&self,
principal_id: Option<&str>,
) -> ControlResult<Vec<gunnar_proto::WebCredential>> {
self.stores
.auth()
.web_credentials(principal_id)
.map_err(internal)?
.iter()
.map(web_credential_out)
.collect()
}
fn credential(
&self,
credential_id: &[u8],
) -> ControlResult<Option<gunnar_grpc::StoredCredential>> {
let Some(row) = self
.stores
.auth()
.web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
.map_err(internal)?
else {
return Ok(None);
};
let public_key = hex::decode(&row.public_key_hex).map_err(|err| {
ControlError::Internal(format!(
"web credential {} holds a key that is not hex: {err}",
row.id
))
})?;
Ok(Some(gunnar_grpc::StoredCredential {
credential: web_credential_out(&row)?,
public_key,
}))
}
fn record_passkey_use(
&self,
credential_id: &[u8],
sign_count: u32,
now_unix_ms: i64,
) -> ControlResult<gunnar_proto::WebCredential> {
let row = self
.stores
.auth()
.touch_web_credential(
&gunnar_grpc::webauthn::b64url(credential_id),
sign_count,
now_unix_ms,
)
.map_err(internal)?;
self.stores.mirror_identity();
self.stores
.activity()
.signed_in(&row.principal_id, now_unix_ms);
web_credential_out(&row)
}
fn revoke_credential(
&self,
credential_id: &[u8],
) -> ControlResult<Option<gunnar_proto::WebCredential>> {
let removed = self
.stores
.auth()
.revoke_web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
.map_err(internal)?;
self.stores.mirror_identity();
removed.as_ref().map(web_credential_out).transpose()
}
fn rename_credential(
&self,
credential_id: &[u8],
label: &str,
) -> ControlResult<Option<gunnar_proto::WebCredential>> {
self.stores
.auth()
.rename_web_credential(&gunnar_grpc::webauthn::b64url(credential_id), label)
.map_err(internal)?
.as_ref()
.map(web_credential_out)
.transpose()
}
fn first_invite(&self) -> ControlResult<Option<gunnar_grpc::FirstInvite>> {
Ok(self
.stores
.auth()
.first_invite(gunnar_grpc::seam::OPERATOR_INVITE_ID)
.map_err(internal)?
.map(|row| gunnar_grpc::FirstInvite {
principal_id: row.principal_id,
secret_hash: row.secret_hash,
created_at_unix_ms: row.created,
expires_at_unix_ms: row.expires,
}))
}
fn drop_first_invite(&self) -> ControlResult<()> {
if self
.stores
.auth()
.drop_first_invite(gunnar_grpc::seam::OPERATOR_INVITE_ID)
.map_err(internal)?
.is_some()
{
self.stores.mirror_identity();
}
Ok(())
}
fn employer_of(&self, principal_id: &str) -> ControlResult<Option<String>> {
self.stores
.auth()
.employer_of(principal_id)
.map_err(internal)
}
fn sso(&self) -> Option<&dyn gunnar_grpc::sso::plane::SsoPlane> {
Some(self)
}
fn sso_import(
&self,
order: gunnar_grpc::sso::scim::ImportOrder,
) -> ControlResult<gunnar_grpc::sso::scim::ImportJob> {
let account = order.account.clone();
let id = self.sso_import.next_job_id();
let job = self.sso_import.enqueue_as(order, id.clone());
if job.job_id != id {
return Err(ControlError::FailedPrecondition(format!(
"an import for account {account} is already {} (job {}): wait for it to finish",
if job.state == crate::job_queue::JobState::Queued {
"queued"
} else {
"running"
},
job.job_id
)));
}
Ok(crate::sso_import_queue::job_out(job))
}
fn sso_import_status(
&self,
account: &str,
) -> ControlResult<Option<gunnar_grpc::sso::scim::ImportJob>> {
Ok(self
.sso_import
.status(account)
.map(crate::sso_import_queue::job_out))
}
fn approve_machine_enrolment(
&self,
approval: gunnar_grpc::MachineApproval,
) -> ControlResult<gunnar_grpc::MachineApproved> {
let step_up = approval
.step_up
.as_ref()
.map(|(id, count)| (gunnar_grpc::webauthn::b64url(id), *count));
let (principal, credential, created) = self
.stores
.auth()
.approve_machine_credential(
&approval.machine,
level_in(approval.server_level)?,
gunnar_core::WebCredential {
id: gunnar_grpc::webauthn::b64url(&approval.credential_id),
principal_id: String::new(),
label: approval.label,
public_key_hex: hex::encode(&approval.public_key),
sign_count: approval.sign_count,
created: approval.now_unix_ms,
last_used: None,
sessions: Vec::new(),
},
step_up.as_ref().map(|(id, count)| (id.as_str(), *count)),
approval.now_unix_ms,
)
.map_err(internal)?;
self.stores.mirror_identity();
Ok(gunnar_grpc::MachineApproved {
principal: principal_out(&principal),
credential: web_credential_out(&credential)?,
created,
})
}
fn remove_principal(&self, principal_id: &str) -> ControlResult<(u32, u32)> {
let before_keys = self.stores.auth().keys().map_err(internal)?.len();
let before_grants = self.stores.auth().grants(None).map_err(internal)?.len();
self.stores
.auth()
.remove_principal(principal_id)
.map_err(internal)?;
let after_keys = self.stores.auth().keys().map_err(internal)?.len();
let after_grants = self.stores.auth().grants(None).map_err(internal)?.len();
self.stores.mirror_identity();
Ok((
before_keys.saturating_sub(after_keys) as u32,
before_grants.saturating_sub(after_grants) as u32,
))
}
fn add_key(
&self,
principal_id: &str,
openssh_public_key: &str,
_comment: &str,
expires_at_unix_ms: Option<i64>,
caller: &str,
attestation: Option<gunnar_proto::SignedKeyAttestation>,
) -> ControlResult<PublicKey> {
let adding = match gunnar_core::auth::pq_recipient_fingerprint(openssh_public_key) {
Some(fingerprint) => fingerprint,
None => gunnar_core::Fingerprint::of_key(
&gunnar_core::auth::parse_public_key(openssh_public_key).map_err(internal)?,
),
};
let holders = self
.stores
.auth()
.attesting_keys_of(principal_id)
.map_err(internal)?;
if holders.iter().any(|k| k.fingerprint == adding.as_str()) {
return Err(ControlError::AlreadyExists(format!(
"principal {principal_id} already holds key {}",
adding.as_str()
)));
}
let requirement =
gunnar_core::key_attestation::Requirement::of(principal_id, caller, holders);
let attestation = attestation.map(from_proto_attestation).transpose()?;
gunnar_core::key_attestation::check(
&requirement,
principal_id,
adding.as_str(),
attestation.as_ref(),
)
.map_err(|problem| match problem {
gunnar_core::AttestationProblem::Missing { .. } => {
ControlError::FailedPrecondition(problem.to_string())
}
_ => ControlError::InvalidArgument(problem.to_string()),
})?;
let registered = self
.stores
.auth()
.register_key(
principal_id,
openssh_public_key,
expires_at_unix_ms.map(|ms| ms as UnixMillis),
)
.map(|key| key_out(&key))
.map_err(internal)?;
self.stores.mirror_identity();
Ok(registered)
}
fn revoke_key(&self, fingerprint: &str) -> ControlResult<()> {
let parsed = gunnar_core::Fingerprint::parse(fingerprint)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let revoked = self.stores.auth().revoke_key(&parsed).map_err(internal)?;
self.stores.mirror_identity();
match revoked {
Some(_) => Ok(()),
None => Err(ControlError::NotFound(format!("key {fingerprint}"))),
}
}
fn list_keys(&self, principal_id: Option<&str>) -> ControlResult<Vec<PublicKey>> {
let keys = match principal_id {
Some(id) => self.stores.auth().keys_of(id).map_err(internal)?,
None => self.stores.auth().keys().map_err(internal)?,
};
Ok(keys.iter().map(key_out).collect())
}
fn grant(
&self,
principal_id: &str,
namespace: &str,
repo: &str,
level: GrantLevel,
) -> ControlResult<gunnar_proto::Grant> {
let scope = scope_in(namespace, repo)?;
let grant = Grant {
principal: principal_id.to_owned(),
scope,
level: level_in(level)?,
};
self.stores
.auth()
.grant(grant)
.map(|g| grant_out(&g))
.map_err(internal)
}
fn revoke(&self, principal_id: &str, namespace: &str, repo: &str) -> ControlResult<()> {
let scope = scope_in(namespace, repo)?;
match self
.stores
.auth()
.revoke_grant(principal_id, &scope)
.map_err(internal)?
{
true => Ok(()),
false => Err(ControlError::NotFound(format!(
"grant for {principal_id} on {scope}"
))),
}
}
fn list_grants(
&self,
principal_id: Option<&str>,
namespace: Option<&str>,
repo: Option<&str>,
) -> ControlResult<Vec<gunnar_proto::Grant>> {
Ok(self
.stores
.auth()
.grants(principal_id)
.map_err(internal)?
.iter()
.map(grant_out)
.filter(|g| namespace.is_none_or(|want| want == g.namespace))
.filter(|g| repo.is_none_or(|want| want == g.repo))
.collect())
}
fn add_group_member(
&self,
group_id: &str,
member_id: &str,
signed: gunnar_proto::SignedMembershipEvent,
) -> ControlResult<u64> {
let signed = self.verify_membership_event(
signed,
group_id,
member_id,
gunnar_core::MembershipAction::Add,
)?;
self.stores
.auth()
.add_to_group(group_id, member_id)
.map_err(internal)?;
self.append_membership_event(&signed)
}
fn remove_group_member(
&self,
group_id: &str,
member_id: &str,
signed: gunnar_proto::SignedMembershipEvent,
) -> ControlResult<(bool, u64)> {
let signed = self.verify_membership_event(
signed,
group_id,
member_id,
gunnar_core::MembershipAction::Remove,
)?;
let was_member = self
.stores
.auth()
.remove_from_group(group_id, member_id)
.map_err(internal)?;
let index = self.append_membership_event(&signed)?;
Ok((was_member, index))
}
fn membership_events(
&self,
group_id: Option<&str>,
) -> ControlResult<Vec<gunnar_proto::SignedMembershipEvent>> {
Ok(self
.membership_log_events()?
.into_iter()
.filter(|signed| group_id.is_none_or(|g| signed.event.group == g))
.map(to_proto_membership)
.collect())
}
fn list_group_members(&self, group_id: &str) -> ControlResult<Vec<Principal>> {
let auth = self.stores.auth();
match auth.principal(group_id).map_err(internal)? {
Some(p) if p.kind == PrincipalKind::Group => {}
Some(_) => {
return Err(ControlError::InvalidArgument(format!(
"principal \"{group_id}\" is not a group"
)))
}
None => return Err(ControlError::NotFound(format!("group {group_id}"))),
}
let mut members = Vec::new();
for id in auth.members_of(group_id).map_err(internal)? {
if let Some(p) = auth.principal(&id).map_err(internal)? {
members.push(principal_out(&p));
}
}
Ok(members)
}
fn get_protection(
&self,
path: &RepoPath,
ref_pattern: Option<&str>,
) -> ControlResult<Vec<BranchProtection>> {
let (repo, _) = self.open(path)?;
let config = self.policy_of(&repo)?;
Ok(config
.protection
.iter()
.filter(|rule| ref_pattern.is_none_or(|want| rule.pattern.as_str() == want))
.map(protection_out)
.collect())
}
fn set_protection(
&self,
path: &RepoPath,
protection: BranchProtection,
) -> ControlResult<BranchProtection> {
let (repo, _) = self.open(path)?;
if protection.max_object_bytes != 0 || protection.max_objects_per_push != 0 {
return Err(ControlError::InvalidArgument(
"max_object_bytes and max_objects_per_push are per-rule in the protocol and \
repository-wide in this server's policy engine, so a value here would apply to \
references this rule does not name. Leave them 0 and set the repository's \
limits in its policy document."
.to_owned(),
));
}
let pattern = gunnar_policy::RefPattern::new(protection.ref_pattern.clone())
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let mut config = self.policy_of(&repo)?;
let mut rule = config
.protection
.iter()
.find(|rule| rule.pattern == pattern)
.cloned()
.unwrap_or_else(|| {
gunnar_policy::ProtectionRule::protect(pattern.as_str())
.expect("the pattern was compiled one line above")
});
rule.pattern = pattern;
rule.allow_force_push = protection.allow_force_push;
rule.allow_delete = protection.allow_delete;
rule.require_signature = protection.require_signature;
rule.allowed_principals = if protection.allowed_principal_ids.is_empty() {
None
} else {
Some(protection.allowed_principal_ids.clone())
};
rule.forbid_self_approval = protection.forbid_self_approval;
rule.required_approvals = protection.required_approvals;
config
.protection
.retain(|other| other.pattern != rule.pattern);
config.protection.push(rule.clone());
self.store_policy(&repo, &config)?;
Ok(protection_out(&rule))
}
fn delete_protection(&self, path: &RepoPath, ref_pattern: &str) -> ControlResult<()> {
let (repo, _) = self.open(path)?;
let mut config = self.policy_of(&repo)?;
let before = config.protection.len();
config
.protection
.retain(|rule| rule.pattern.as_str() != ref_pattern);
if config.protection.len() == before {
return Err(ControlError::NotFound(format!(
"no protection rule for {ref_pattern} in {path}"
)));
}
self.store_policy(&repo, &config)
}
fn get_ref_name_policy(&self, path: &RepoPath) -> ControlResult<RefNamePolicy> {
let (repo, _) = self.open(path)?;
let config = self.policy_of(&repo)?;
Ok(ref_name_policy_out(&config.ref_names))
}
fn set_ref_name_policy(
&self,
path: &RepoPath,
policy: RefNamePolicy,
) -> ControlResult<RefNamePolicy> {
let (repo, _) = self.open(path)?;
let compiled = gunnar_policy::RefNamePolicy::new(
policy.allow_patterns.clone(),
policy.deny_patterns.clone(),
)
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
let mut config = self.policy_of(&repo)?;
config.ref_names = compiled.clone();
self.store_policy(&repo, &config)?;
Ok(ref_name_policy_out(&compiled))
}
fn forge_append(&self, event: gunnar_proto::SignedForgeEvent) -> ControlResult<u64> {
if event.attested.is_some() {
return Err(ControlError::InvalidArgument(
"this event carries an attestation, and an attestation is gunnar's own word: \
only gunnar makes one, for a person's session that stepped up"
.to_owned(),
));
}
let signed = from_proto_event(event)?;
let key = self
.stores
.auth()
.signing_keys()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
.into_iter()
.find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
.ok_or_else(|| {
ControlError::InvalidArgument(format!(
"{} is not registered on this server, so nobody reading this event later \
could check its signature",
signed.key_fingerprint
))
})?;
if key.principal_id != signed.event.actor {
return Err(ControlError::InvalidArgument(format!(
"this event is attributed to {:?} and is signed by {}, which is {:?}'s key; \
an approval is the actor's own word, so it must carry the actor's own \
signature",
signed.event.actor, signed.key_fingerprint, key.principal_id
)));
}
let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
.map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
gunnar_core::forge::verify_event(&signed, &public)
.map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
self.forge_subject_open(&signed)?;
self.forge_write(&signed)
}
fn forge_append_attested(
&self,
event: gunnar_proto::SignedForgeEvent,
session_stepped_up_at_unix_ms: i64,
) -> ControlResult<(u64, String)> {
if !event.key_fingerprint.is_empty()
|| !event.signature.is_empty()
|| event.attested.is_some()
{
return Err(ControlError::InvalidArgument(
"an event gunnar attests arrives unsigned: a signed one is checked against its \
signer's key instead, and an attestation is gunnar's own to make"
.to_owned(),
));
}
let event = from_proto_event(event)?.event;
if event.actor.is_empty() {
return Err(ControlError::InvalidArgument(
"an attested event must name the person it is attested for".to_owned(),
));
}
let key = self
.stores
.forge_attest_key()
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
let signed = gunnar_core::forge::attest_event(
&key,
event,
gunnar_core::Attested {
session_stepped_up_at_unix_ms,
},
)
.map_err(|err| ControlError::Internal(format!("signing the attested event: {err}")))?;
self.forge_subject_open(&signed)?;
let words = signed
.attested
.as_ref()
.map(|a| a.words())
.unwrap_or_default();
Ok((self.forge_write(&signed)?, words))
}
fn forge_review(
&self,
repo: &str,
pull: u64,
) -> ControlResult<(
gunnar_proto::ReviewState,
Vec<gunnar_proto::SignedForgeEvent>,
)> {
let all = self.forge_events()?;
let mine: Vec<gunnar_core::ForgeEvent> = all
.iter()
.filter(|signed| signed.event.repo == repo)
.map(|signed| signed.event.clone())
.collect();
let state = gunnar_core::forge::fold(pull, &mine);
let events = all
.iter()
.filter(|signed| signed.event.repo == repo && signed.event.pull == pull)
.cloned()
.map(to_proto_event)
.collect();
Ok((self.review_state(repo, &state, &all)?, events))
}
fn forge_reviews(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::ReviewState>> {
let all = self.forge_events()?;
let mine: Vec<gunnar_core::ForgeEvent> = all
.iter()
.filter(|signed| signed.event.repo == repo)
.map(|signed| signed.event.clone())
.collect();
let mut numbers: Vec<u64> = mine.iter().map(|event| event.pull).collect();
numbers.sort_unstable();
numbers.dedup();
let mut out = Vec::new();
for pull in numbers {
let state = gunnar_core::forge::fold(pull, &mine);
out.push(self.review_state(repo, &state, &all)?);
}
Ok(out)
}
fn forge_issue(
&self,
repo: &str,
number: u64,
) -> ControlResult<
Option<(
gunnar_proto::IssueState,
Vec<gunnar_proto::SignedForgeEvent>,
)>,
> {
let all = self.forge_events()?;
let mine: Vec<gunnar_core::ForgeEvent> = all
.iter()
.filter(|signed| signed.event.repo == repo)
.map(|signed| signed.event.clone())
.collect();
let Some(issue) = gunnar_core::forge::fold_issue(number, &mine) else {
return Ok(None);
};
let mut opened_seen = 0u64;
let events = all
.iter()
.filter(|signed| {
if signed.event.repo != repo {
return false;
}
match &signed.event.event {
gunnar_core::Event::IssueOpened { .. } => {
opened_seen += 1;
opened_seen == number
}
gunnar_core::Event::IssueCommented { .. }
| gunnar_core::Event::IssueClosed { .. } => signed.event.pull == number,
_ => false,
}
})
.cloned()
.map(to_proto_event)
.collect();
Ok(Some((issue_state(repo, &issue), events)))
}
fn forge_issues(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::IssueState>> {
let mine: Vec<gunnar_core::ForgeEvent> = self
.forge_events()?
.into_iter()
.filter(|signed| signed.event.repo == repo)
.map(|signed| signed.event)
.collect();
Ok(gunnar_core::forge::fold_issues(&mine)
.iter()
.map(|issue| issue_state(repo, issue))
.collect())
}
fn stats_extras(&self) -> ControlResult<gunnar_grpc::seam::StatsExtras> {
let tombstones = self.stores.tombstones().unwrap_or_default();
let report = self.replication.report();
#[allow(unused_mut)]
let mut tenants = crate::dashboard::TenantCounts::default();
#[cfg(feature = "metered")]
if let Some(slot) = &self.entitlement {
tenants = slot.counts();
}
Ok(gunnar_grpc::seam::StatsExtras {
tombstoned_repos: tombstones.len() as u64,
tombstoned_bytes: tombstones.iter().map(|row| row.bytes).sum(),
replication_configured: report.configured,
replication_enqueued: report.enqueued,
replication_synced: report.synced,
replication_failed: report.failed,
replication_bus_gaps: report.bus_gaps,
replication_refs_total: report.refs.total,
replication_refs_unmirrored: report.refs.unmirrored,
standbys: report
.standbys
.into_iter()
.map(|row| gunnar_proto::StandbyStatus {
verdict: row.verdict().word().to_owned(),
name: row.name,
remote: row.remote,
pending: row.pending,
consecutive_failures: row.consecutive_failures,
last_error: row.last_error,
last_success_unix_ms: row.last_success_unix_ms,
needs_full_resync: row.needs_full_resync,
stale: row.stale,
})
.collect(),
tenants_paid: tenants.paid,
tenants_grace: tenants.grace,
tenants_suspended: tenants.suspended,
})
}
fn counts(&self) -> ControlResult<ControlCounts> {
let (namespaces, repos) = self.stores.catalog().counts().map_err(internal)?;
let (principals, keys, grants) = self.stores.auth().counts().map_err(internal)?;
Ok(ControlCounts {
namespaces: namespaces as u64,
repos: repos as u64,
principals: principals as u64,
keys: keys as u64,
grants: grants as u64,
})
}
fn verify_chain(
&self,
request: &gunnar_proto::VerifyChainRequest,
) -> ControlResult<gunnar_proto::VerifyChainResponse> {
let target = match request.target.as_str() {
"" => None,
spec => Some(spec),
};
let chain = if !request.artifact_sha256.is_empty() {
crate::verify::verify_artifact(&self.stores, self, &request.artifact_sha256, target)
} else {
let Some(repo) = &request.repo else {
return Err(ControlError::InvalidArgument(
"name a repository, or an artifact_sha256 to start from".to_owned(),
));
};
let rev = match request.rev.as_str() {
"" => "HEAD",
rev => rev,
};
crate::verify::verify(&self.stores, self, repo, rev, target)
}
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
Ok(chain.to_proto())
}
fn blame(
&self,
path: &RepoPath,
rev: &str,
file: &str,
) -> ControlResult<gunnar_proto::BlameResponse> {
let rev = if rev.is_empty() { "HEAD" } else { rev };
let rows = crate::blame::blame(&self.stores, self, path, rev, file)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
Ok(gunnar_proto::BlameResponse {
lines: rows
.into_iter()
.map(|row| gunnar_proto::BlameLine {
commit: row.commit.to_hex().to_string(),
author: row.author,
line: row.line,
})
.collect(),
})
}
fn search(
&self,
path: &RepoPath,
rev: &str,
pattern: &str,
max_file_bytes: u64,
max_matches: u32,
) -> ControlResult<gunnar_proto::SearchResponse> {
let rev = if rev.is_empty() { "HEAD" } else { rev };
let defaults = crate::search::Limits::default();
let limits = crate::search::Limits {
max_file_bytes: if max_file_bytes == 0 {
defaults.max_file_bytes
} else {
max_file_bytes
},
max_matches: if max_matches == 0 {
defaults.max_matches
} else {
max_matches as usize
},
};
let hits = crate::search::search(&self.stores, self, path, rev, pattern, limits)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
Ok(gunnar_proto::SearchResponse {
matches: hits
.into_iter()
.map(|hit| gunnar_proto::SearchMatch {
path: hit.path,
line_number: hit.line_number as u32,
line: hit.line,
})
.collect(),
})
}
fn issuer_info(&self) -> ControlResult<gunnar_proto::IssuerInfoResponse> {
let pki = self.stores.pki_readonly().map_err(|err| {
ControlError::Unavailable(format!("the PKI document could not be opened: {err:#}"))
})?;
crate::admin::issuer_document(&pki).map_err(|err| {
ControlError::Unavailable(format!("the PKI document could not be read: {err:#}"))
})
}
fn setting(&self, name: &str) -> ControlResult<Option<Vec<u8>>> {
self.stores.settings().get(name).map_err(internal)
}
fn set_setting(&self, name: &str, value: &[u8]) -> ControlResult<Option<Vec<u8>>> {
self.stores.settings().set(name, value).map_err(internal)
}
fn audit_trail(&self, at_seq: u64) -> ControlResult<gunnar_proto::AuditTrailResponse> {
let path = self.stores.audit_path();
let replicated = !self.stores.is_twin()
&& !crate::twin::is_restoring(self.stores.root())
&& gunnar_core::txlog::sink::log_for(&path)
.is_some_and(|(_, rel)| rel == gunnar_core::txlog::AUDIT_REL);
let mut out = gunnar_proto::AuditTrailResponse {
present: false,
entries: 0,
tip: String::new(),
verified: false,
replicated,
hash_at_seq: String::new(),
why: String::new(),
tip_seq: 0,
};
let chain = match gunnar_core::AuditChain::open(&path) {
Ok(chain) => chain,
Err(err) => {
out.why = format!("{err}");
return Ok(out);
}
};
let entries = match chain.entries() {
Ok(entries) => entries,
Err(err) => {
out.why = format!("{err}");
return Ok(out);
}
};
out.present = true;
out.entries = entries.len() as u64;
if let Some(last) = entries.last() {
out.tip.clone_from(&last.hash);
out.tip_seq = last.seq;
}
match chain.verify() {
Ok(_) => out.verified = true,
Err(err) => out.why = format!("{err}"),
}
if at_seq != 0 {
match entries.iter().find(|e| e.seq == at_seq) {
Some(e) => out.hash_at_seq = e.hash.clone(),
None => {
let sep = if out.why.is_empty() { "" } else { "; " };
out.why = format!(
"{}{sep}this trail holds no entry at seq {at_seq} (it holds {} entries)",
out.why,
entries.len()
);
}
}
}
Ok(out)
}
fn storage_info(&self, path: Option<&RepoPath>) -> ControlResult<StorageInfoResponse> {
let repos = match path {
Some(path) => vec![self.open(path)?.0],
None => self.stores.catalog().list(None).map_err(internal)?,
};
let mut total_hot = 0u64;
let mut total_explode = 0u64;
let mut total_budget = 0u64;
let bands: std::collections::HashMap<std::path::PathBuf, crate::storage::ExplodeBand> =
self.stores.engine().explode_use().into_iter().collect();
let mut out = Vec::new();
for repo in repos {
let store = self.stores.repo_store(&repo).map_err(|err| {
ControlError::Internal(format!("opening the repository's stores: {err:#}"))
})?;
let (loose, _objects_dir) = object_dir_stats(store.git_dir());
let bytes = repo_bytes(store.git_dir()).hot();
let refs = store.serve().refs().map_err(store_error)?.len() as u64;
total_hot += bytes;
let band = bands.get(store.git_dir()).copied().unwrap_or_default();
total_explode += band.used();
total_budget += band.budget;
out.push(RepoStorage {
repo: Some(RepoPath::new(&repo.namespace, &repo.name)),
hash_kind: hash_kind(repo.hash_kind) as i32,
loose_object_count: loose,
hot_bytes: bytes,
cold_bytes: 0,
cold_object_count: 0,
cold_measured: false,
last_repack_at_unix_ms: 0,
ref_count: refs,
explode_bytes: band.used(),
explode_budget_bytes: band.budget,
explode_evicted_bytes: band.evicted_bytes,
});
}
let engine = self.stores.engine();
let (object_backend, ref_backend) = (engine.describe(), engine.describe());
let tombstones = self.stores.tombstones().unwrap_or_default();
let total_tombstoned_bytes = tombstones.iter().map(|row| row.bytes).sum();
let tombstoned = tombstones
.into_iter()
.map(|row| gunnar_proto::TombstonedRepo {
repo: Some(RepoPath {
namespace: row.path.namespace().to_owned(),
repo: row.path.name().to_owned(),
}),
bytes: row.bytes,
deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
})
.collect();
Ok(StorageInfoResponse {
object_backend,
ref_backend,
engine: Some(engine.report()),
total_hot_bytes: total_hot,
total_cold_bytes: 0,
cold_measured: false,
repos: out,
total_explode_bytes: total_explode,
total_explode_budget_bytes: total_budget,
total_tombstoned_bytes,
tombstoned,
reality: Some(self.stores.storage_reality()),
})
}
fn capacity(
&self,
request: &gunnar_proto::CapacityRequest,
) -> ControlResult<gunnar_proto::CapacityResponse> {
let Some(gate) = self.headroom.as_ref() else {
return Err(ControlError::Unimplemented(
"Admin.Capacity (no disk-headroom gate is wired on this control plane, so the \
--disk-floor in force is unknown and no quote can be made against it)",
));
};
let held_bytes = if request.namespace.is_empty() {
0
} else {
self.usage(Some(&request.namespace))?
.first()
.map(|row| row.pack_bytes)
.unwrap_or(0)
};
let root = gate.root().to_path_buf();
let mut answer = crate::capacity::answer(&crate::capacity::Question {
room: gate.headroom().map_err(|err| err.to_string()),
floor_spec: gate.floor().to_string(),
requested_bytes: request.requested_bytes,
held_bytes,
namespace: request.namespace.clone(),
growth: crate::capacity::GrowthFacts::measure(&root),
machine: crate::capacity::MachineFacts::measure(),
root: root.clone(),
});
if request.disks {
let slow = self.stores.slow();
let serialdat = std::path::PathBuf::from(slow.serialdat());
let mut disks = crate::capacity::measure_disks(&root, Some(&serialdat), gate.floor());
let log = slow.path_of(crate::slow::LOG_DIR);
let known: Vec<(&str, std::path::PathBuf)> = [
("metadata copy", slow.path_of(crate::slow::META_DIR)),
("crash record", slow.path_of(crate::slow::CRASH_DIR)),
("audit trail", log.join("audit-chain.jsonl")),
("logs", log),
("grow journal", serialdat.join("grow-journal")),
("repository mirror", serialdat.join(crate::slow::REPOS_DIR)),
]
.into_iter()
.filter(|(_, path)| path.starts_with(&serialdat))
.collect();
let used = disks
.mounts
.iter()
.find(|m| m.labels.iter().any(|l| l == "serialdat") && m.measured)
.map(|m| m.used_bytes);
if disks
.mounts
.iter()
.any(|m| m.labels.iter().any(|l| l == "serialdat"))
{
disks.serialdat = crate::capacity::serialdat_breakdown(&known, used);
}
answer.disks = Some(disks);
}
Ok(answer)
}
fn grow_start(
&self,
request: &gunnar_proto::GrowStartRequest,
) -> ControlResult<gunnar_proto::GrowStatusResponse> {
self.grow_machine()?
.start(request)
.map_err(ControlError::FailedPrecondition)
}
fn grow_status(&self) -> ControlResult<gunnar_proto::GrowStatusResponse> {
match &self.grow {
Some(grow) => Ok(grow.status()),
None => Ok(gunnar_proto::GrowStatusResponse {
phase: "idle".to_owned(),
engine_present: false,
serving: true,
server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
detail: "this control plane has no runtime growth wired in".to_owned(),
hotplug: crate::dashboard::probe::pci_hotplug().present,
journal: gunnar_grpc::grow_journal::Summary::measure().map(|s| s.to_wire()),
..Default::default()
}),
}
}
fn grow_resume(&self, give_up: bool) -> ControlResult<gunnar_proto::GrowStatusResponse> {
self.grow_machine()?
.resume(give_up)
.map_err(ControlError::FailedPrecondition)
}
fn usage(&self, namespace: Option<&str>) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
self.usage_readable()?;
if let Some(name) = namespace {
let measured_at_unix_ms = gunnar_core::now_ms();
let Some((held, _)) = self
.stores
.usage_ledger()
.exact(&self.stores, name)
.map_err(internal)?
else {
return Err(ControlError::NotFound(format!(
"namespace {name} is not in the catalog"
)));
};
let objects = self.open_objects_of(name);
return Ok(vec![self.usage_row(
name,
&held,
objects,
measured_at_unix_ms,
)]);
}
self.usage_all()
}
fn repo_usage(&self, namespace: &str) -> ControlResult<Vec<gunnar_proto::RepoStored>> {
let Some((_, repos)) = self
.stores
.usage_ledger()
.exact(&self.stores, namespace)
.map_err(internal)?
else {
return Err(ControlError::NotFound(format!(
"namespace {namespace} is not in the catalog"
)));
};
let open_objects: std::collections::HashMap<std::path::PathBuf, u64> = self
.stores
.engine()
.open_object_counts()
.into_iter()
.collect();
Ok(repos
.into_iter()
.map(|(name, bytes, dir)| gunnar_proto::RepoStored {
repo: Some(gunnar_proto::RepoPath::new(namespace, &name)),
stored_bytes: bytes.pack,
cache_bytes: bytes.cache,
lfs_bytes: bytes.lfs,
objects: dir.and_then(|d| open_objects.get(&d).copied()).unwrap_or(0),
})
.collect())
}
fn usage_page(
&self,
page_token: &str,
page_size: usize,
changed_since: &str,
) -> ControlResult<gunnar_proto::ReadManyUsageResponse> {
self.usage_readable()?;
let now = gunnar_core::now_ms();
let ledger = self.stores.usage_ledger();
let (cursor, after) = gunnar_grpc::seam::usage_page_token::decode(page_token);
let cursor = if cursor.is_empty() {
ledger.cursor_now(now)
} else {
cursor
};
let (rows, more, changed_only) = ledger
.page(&self.stores, &after, page_size, changed_since, now)
.map_err(internal)?;
let next_page_token = match (more, rows.last()) {
(true, Some((last, _))) => gunnar_grpc::seam::usage_page_token::encode(&cursor, last),
_ => String::new(),
};
let usages = rows
.iter()
.map(|(name, held)| self.usage_row(name, held, held.objects, now))
.collect();
Ok(gunnar_proto::ReadManyUsageResponse {
usages,
next_page_token,
cursor,
changed_only,
as_of_unix_ms: now,
})
}
fn activity(
&self,
namespace: &str,
since_unix_ms: i64,
until_unix_ms: i64,
) -> ControlResult<(Vec<gunnar_proto::DayActivity>, i64)> {
if self
.stores
.catalog()
.namespace(namespace)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!(
"namespace {namespace} is not in the catalog"
)));
}
let now = gunnar_core::now_ms();
let flow = self.stores.flow();
if let Err(err) = flow.flush(now) {
tracing::warn!(
target: "gunnar::store",
event = "flow.flush_failed",
%err,
"could not flush the flow ledger before an Activity read; the series is as of \
the last successful flush"
);
}
let today = gunnar_core::day_of(now);
let from = if since_unix_ms == 0 {
today - gunnar_core::RETENTION_DAYS
} else {
gunnar_core::day_of(since_unix_ms)
}
.max(today - gunnar_core::RETENTION_DAYS);
let to = if until_unix_ms == 0 {
today
} else {
gunnar_core::day_of(until_unix_ms)
}
.min(today);
if from > to {
return Ok((Vec::new(), gunnar_core::RETENTION_DAYS));
}
let days = flow
.store()
.series(namespace, from, to)
.map_err(internal)?
.into_iter()
.map(|(day, row)| day_activity(day, &row))
.collect();
Ok((days, gunnar_core::RETENTION_DAYS))
}
fn admit_plan(&self, namespace: &str) -> ControlResult<gunnar_proto::AdmitPlanResponse> {
use gunnar_ssh::edda_admit::{Consumer, Work};
use gunnar_ssh::governor::OpClass;
if self
.stores
.catalog()
.namespace(namespace)
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!(
"namespace {namespace} is not in the catalog"
)));
}
let Some(gate) = &self.admission else {
return Err(ControlError::Unimplemented("Usage.AdmitPlan"));
};
let policy = gate.admit_policy();
let usage = gate.admit_usage(&*policy, namespace);
let limit = policy.limit();
let who = Consumer::Namespace(namespace.to_owned());
let estimates = policy.estimates();
let kinds = [
OpClass::Clone,
OpClass::Fetch,
OpClass::Push,
OpClass::LsRemote,
]
.into_iter()
.map(|class| {
let kind = gunnar_ssh::admission::work_kind(class);
let plan = policy.plan(&Work::new(kind, None), &who, &usage);
gunnar_proto::AdmitPlanKind {
kind: kind.name().to_owned(),
admit_class: plan.class.name().to_owned(),
why: plan.why,
estimate_bytes: plan.estimate.usage_bytes(),
estimate_cpu_ms: plan.estimate.cpu_ms,
samples: estimates
.iter()
.find(|row| row.kind == kind)
.map_or(0, |row| row.samples),
}
})
.collect();
Ok(gunnar_proto::AdmitPlanResponse {
policy: policy.policy().to_owned(),
window_days: usage.window_days,
used_bytes: usage.bytes,
limit_bytes: limit.map(|l| l.bytes),
budget_left_bytes: limit.map(|l| l.bytes.saturating_sub(usage.bytes)),
kinds,
})
}
#[cfg(feature = "metered")]
fn entitlement_get(
&self,
namespace: &str,
) -> ControlResult<Option<gunnar_proto::EntitlementFact>> {
self.entitlement()?
.get(namespace)
.map_err(|err| ControlError::Internal(format!("{err:#}")))
}
#[cfg(feature = "metered")]
fn entitlement_list(&self) -> ControlResult<Vec<gunnar_proto::EntitlementFact>> {
self.entitlement()?
.list()
.map_err(|err| ControlError::Internal(format!("{err:#}")))
}
#[cfg(feature = "metered")]
fn entitlement_refused_at_boot(&self) -> ControlResult<Option<gunnar_proto::RefusedAtBoot>> {
Ok(self.entitlement()?.refusal())
}
#[cfg(feature = "metered")]
fn entitlement_set(
&self,
fact: gunnar_proto::EntitlementFact,
) -> ControlResult<(gunnar_proto::EntitlementState, u64)> {
use crate::entitlement::SetRefused;
let (state, seq) = self.entitlement()?.set(fact).map_err(|err| match err {
SetRefused::Forged { .. } => ControlError::PermissionDenied(err.to_string()),
SetRefused::Replayed { .. }
| SetRefused::Unstamped { .. }
| SetRefused::Behind { .. } => ControlError::PermissionDenied(err.to_string()),
SetRefused::Malformed(_) => ControlError::InvalidArgument(err.to_string()),
SetRefused::NotRecorded { .. } => ControlError::FailedPrecondition(err.to_string()),
SetRefused::NoKey { .. } => ControlError::FailedPrecondition(err.to_string()),
})?;
Ok((crate::entitlement::state_to_proto(state), seq))
}
#[cfg(feature = "metered")]
fn actor_ticket_public_key_hex(&self) -> ControlResult<String> {
Ok(self
.stores
.actor_ticket_key()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.public_key_hex()
.to_owned())
}
#[cfg(feature = "metered")]
fn actor_ticket_sign(&self, message: &[u8]) -> ControlResult<[u8; 64]> {
Ok(self
.stores
.actor_ticket_key()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.sign(message))
}
#[cfg(feature = "metered")]
fn sealed_get(&self, name: &str) -> ControlResult<Option<(String, Vec<u8>)>> {
self.stores
.sealed_slot_get(name)
.map_err(|err| ControlError::Internal(format!("{err:#}")))
}
#[cfg(feature = "metered")]
fn sealed_put(&self, name: &str, owner: &str, value: &[u8]) -> ControlResult<bool> {
self.stores
.sealed_slot_put(name, owner, value)
.map_err(|err| ControlError::Unavailable(format!("{err:#}")))
}
fn mirror_add(
&self,
path: &RepoPath,
remote_url: &str,
prune: bool,
create_repo: bool,
) -> ControlResult<MirrorConfig> {
if remote_url.trim().is_empty() {
return Err(ControlError::InvalidArgument(
"a mirror needs a remote URL".to_owned(),
));
}
crate::mirror_fetch::classify(remote_url).map_err(ControlError::InvalidArgument)?;
crate::mirror_fetch::check_destination(remote_url, self.stores.root())
.map_err(ControlError::InvalidArgument)?;
let repo_ref = Self::repo_ref(path)?;
let creates = create_repo && self.get_repo(path).is_err();
if creates {
if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
return Err(ControlError::ResourceExhausted(line));
}
if self
.stores
.catalog()
.namespace(path.namespace.as_str())
.map_err(internal)?
.is_none()
{
return Err(ControlError::NotFound(format!(
"namespace {}",
path.namespace
)));
}
} else {
self.open(path)?;
}
let mut record = gunnar_core::MirrorRecord::new(
&repo_ref,
remote_url,
prune,
gunnar_grpc::bus::now_unix_ms(),
);
record.creates_repo = creates;
let stored = self
.stores
.mirrors()
.set(&repo_ref, record)
.map_err(internal)?;
Ok(mirror_out(&stored))
}
fn mirror_sync(&self, path: &RepoPath) -> ControlResult<MirrorConfig> {
let record = self.mirror_record(path)?;
let repo_ref = Self::repo_ref(path)?;
crate::mirror_fetch::classify(&record.remote_url).map_err(ControlError::InvalidArgument)?;
crate::mirror_fetch::check_destination(&record.remote_url, self.stores.root())
.map_err(ControlError::InvalidArgument)?;
let job = self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
path: repo_ref.key(),
upstream: record.remote_url.clone(),
prune: record.prune,
});
let mut out = mirror_out(&record);
out.job = Some(mirror_job_out(path, job));
Ok(out)
}
fn mirror_set_credential(
&self,
path: &RepoPath,
kind: gunnar_proto::MirrorCredentialKind,
secret: &str,
username: &str,
) -> ControlResult<MirrorConfig> {
gunnar_grpc::seam::credential_arguments(kind, secret)?;
self.mirror_record(path)?;
let repo_ref = Self::repo_ref(path)?;
let kind = credential_kind_in(kind);
let secrets = self.stores.secrets().map_err(|err| {
ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
})?;
let key = gunnar_core::credential_secret_key(&repo_ref);
let stored = if kind.is_set() {
gunnar_store::SecretStore::put(&secrets, &key, secret.as_bytes())
} else {
gunnar_store::SecretStore::delete(&secrets, &key)
};
stored.map_err(|err| {
ControlError::Internal(format!(
"the mirror credential for {path} could not be sealed: {err}"
))
})?;
let record = self
.stores
.mirrors()
.set_credential(&repo_ref, kind, username)
.map_err(internal)?
.ok_or_else(|| ControlError::NotFound(format!("no mirror registered for {path}")))?;
Ok(mirror_out(&record))
}
fn mirror_status(
&self,
namespace: Option<&str>,
path: Option<&RepoPath>,
) -> ControlResult<Vec<MirrorConfig>> {
let all = self.stores.mirrors().all().map_err(internal)?;
Ok(all
.iter()
.filter(|record| match path {
Some(want) => record.repo == format!("{}/{}", want.namespace, want.repo),
None => true,
})
.filter(|record| match namespace {
Some(want) => record
.repo
.split_once('/')
.is_some_and(|(ns, _)| ns == want),
None => true,
})
.map(|record| {
let mut out = mirror_out(record);
out.job = out
.repo
.as_ref()
.zip(self.mirror.status(&record.repo))
.map(|(path, job)| mirror_job_out(path, job));
if out.job.is_none() && !record.last_error.is_empty() {
out.job = out.repo.clone().map(|path| gunnar_proto::MirrorJob {
job_id: String::new(),
path: Some(path),
state: gunnar_proto::MirrorJobState::Failed as i32,
error: record.last_error.clone(),
..Default::default()
});
}
out
})
.collect())
}
fn mirror_remove(&self, path: &RepoPath) -> ControlResult<()> {
self.mirror_record(path)?;
let repo_ref = Self::repo_ref(path)?;
if !self.stores.mirrors().remove(&repo_ref).map_err(internal)? {
return Err(ControlError::NotFound(format!(
"no mirror registered for {path}"
)));
}
let secrets = self.stores.secrets().map_err(|err| {
ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
})?;
gunnar_store::SecretStore::delete(&secrets, &gunnar_core::credential_secret_key(&repo_ref))
.map_err(|err| {
ControlError::Internal(format!(
"the registration for {path} was removed and its sealed credential could \
not be deleted: {err}. The secret is still on this box; remove it before \
trusting that it is gone."
))
})?;
Ok(())
}
#[cfg(feature = "replication")]
fn replication_set(
&self,
target: gunnar_proto::ReplicationTarget,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let mut target = crate::replication_wire::target_from_proto(target)?;
let minted_public = if target.identity.is_none() {
let (path, public) = crate::replication_wire::ensure_identity(self.stores.root())?;
target.identity = Some(path);
Some(public)
} else {
None
};
crate::replication_wire::require_public_only_identity(&target)?;
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
config.set_target(target);
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let mut out = self.replication_snapshot(false)?;
out.identity_public_key = minted_public.unwrap_or_default();
Ok(out)
}
#[cfg(feature = "replication")]
fn replication_remove(
&self,
name: &str,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
if !config.remove_target(name) {
return Err(ControlError::NotFound(format!(
"no replication target named {name:?}"
)));
}
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
self.replication_snapshot(false)
}
fn is_group_member(&self, principal_id: &str, group: &str) -> ControlResult<bool> {
self.stores
.auth()
.groups_of(principal_id)
.map(|groups| groups.iter().any(|g| g == group))
.map_err(|err| ControlError::Unavailable(format!("reading group membership: {err}")))
}
fn become_twin(&self, primary: &str) -> ControlResult<String> {
let key = crate::twin::become_twin(&self.stores, primary)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
crate::twin::start_follower(Arc::clone(&self.stores));
Ok(key)
}
#[cfg(feature = "wg-client")]
fn assign_tunnel_address(
&self,
address: &str,
) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
let window_open = self.stores.auth().bootstrap_window_open().map_err(|err| {
ControlError::Unavailable(format!("reading the bootstrap window: {err}"))
})?;
crate::appliance::tunnel_address_answer(self.stores.root(), address, window_open)
}
#[cfg(feature = "wg-client")]
fn set_tunnel_address(
&self,
address: &str,
) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
crate::appliance::tunnel_address_answer(self.stores.root(), address, true)
}
fn twin_csr(&self) -> ControlResult<String> {
crate::twin_tls::request(self.stores.root())
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn install_twin_certificate(
&self,
cert_pem: &str,
ca_pem: &str,
replication_port: u16,
replication_host: &str,
) -> ControlResult<i64> {
let host = (!replication_host.is_empty()).then_some(replication_host);
crate::twin_tls::install_at(self.stores.root(), cert_pem, ca_pem, replication_port, host)
.map(|expiry| expiry.days_left)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn issue_twin_certificate(
&self,
principal: &str,
csr_pem: &str,
) -> ControlResult<(String, String, u16, String)> {
crate::twin_tls::issue_for_twin(&self.stores, principal, csr_pem)
.map(|(cert, ca, port)| {
(
cert,
ca,
port,
crate::twin_tls::replication_host(self.stores.root()).unwrap_or_default(),
)
})
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn twin_git_listener_needs_restart(&self) -> bool {
crate::twin::git_listener_needs_restart(&self.stores)
}
fn restore_begin(&self, owner: &str) -> ControlResult<u64> {
crate::twin::restore_begin_by(&self.stores, owner).map_err(|err| {
if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() {
ControlError::PermissionDenied(format!("{err:#}"))
} else {
ControlError::FailedPrecondition(format!("{err:#}"))
}
})
}
fn restore_keys(&self) -> ControlResult<Vec<String>> {
#[allow(unused_mut)]
let mut keys = self.restore_keys.clone();
#[cfg(feature = "wg-client")]
keys.extend(
crate::fed_principals::restore_keys_at(self.stores.root())
.map_err(ControlError::Unavailable)?,
);
keys.sort();
keys.dedup();
Ok(keys)
}
fn restore_room(&self, announced_bytes: u64) -> ControlResult<()> {
crate::twin::room_for_restore(self.stores.root(), announced_bytes)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
}
fn restore_intake<'a>(
&'a self,
owner: &str,
) -> ControlResult<Option<Box<dyn gunnar_grpc::RestoreIntake + 'a>>> {
match crate::twin::intake::Intake::open(&self.stores, owner) {
Ok(Some(intake)) => Ok(Some(Box::new(intake))),
Ok(None) => Ok(None),
Err(err) if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() => {
Err(ControlError::PermissionDenied(format!("{err:#}")))
}
Err(err) => Err(ControlError::FailedPrecondition(format!("{err:#}"))),
}
}
fn restore_apply(&self, entry: gunnar_proto::LogEntry) -> ControlResult<bool> {
let e = gunnar_core::txlog::Entry {
seq: entry.seq,
kind: entry.kind as u16,
meta: entry.meta,
blob: entry.blob,
};
match crate::twin::restore_apply(&self.stores, &e) {
Ok(crate::twin::Applied::Lost(_)) => Ok(true),
Ok(_) => Ok(false),
Err(err) => Err(ControlError::Internal(format!(
"restoring seq {}: {err:#}",
e.seq
))),
}
}
fn restore_finish(&self, announced_head: u64, log_id_of_source: &str) -> ControlResult<u64> {
let head =
crate::twin::restore_finish_streamed(&self.stores, announced_head, log_id_of_source)
.map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
crate::twin::restart_after_restore();
Ok(head)
}
fn snapshot_stream(
&self,
sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
) -> ControlResult<u64> {
self.snapshot_stream_from(0, 0, "", sink)
}
fn snapshot_stream_from(
&self,
from_item: u64,
from_offset: u64,
expect_id: &str,
sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
) -> ControlResult<u64> {
let mut pieces = |p: crate::twin::snapshot::Piece<'_>,
pos: Option<crate::twin::snapshot::PiecePos<'_>>|
-> anyhow::Result<()> {
sink(gunnar_proto::LogChunk {
seq: 0,
kind: u32::from(p.kind),
meta: p.meta.to_vec(),
blob_len: p.blob_len,
blob: p.blob.to_vec(),
last: p.last,
restore_head: 0,
restore_bytes: 0,
log_id: String::new(),
offset: pos.map(|q| q.offset),
entry_crc: pos.and_then(|q| q.crc),
snapshot_item: pos.map(|q| q.item),
snapshot_id: pos.map(|q| q.id.to_owned()).unwrap_or_default(),
})
.map_err(|err| anyhow::anyhow!("{err}"))
};
crate::twin::snapshot::serve_from(
&self.stores,
from_item,
from_offset,
expect_id,
&mut pieces,
)
.map_err(|err| {
match err.downcast_ref::<crate::twin::snapshot::SnapshotChanged>() {
Some(changed) => ControlError::FailedPrecondition(changed.to_string()),
None => ControlError::Internal(format!("producing a snapshot: {err:#}")),
}
})
}
fn log_retain(&self, from: u64) -> ControlResult<gunnar_proto::LogHeadResponse> {
let log = self.stores.txlog();
if log.is_follower() {
return Err(ControlError::FailedPrecondition(
"retention: a twin's log is the primary's numbers; retain on the primary".into(),
));
}
let from = if from == 0 { log.head_seq() + 1 } else { from };
let first = log.retain_from(from).map_err(|err| {
ControlError::Unavailable(format!("retaining the transaction log from {from}: {err}"))
})?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, from, first, head = log.head_seq(), "the transaction log now retains from {first}; a twin behind it takes a snapshot");
self.log_head()
}
fn log_layout(&self) -> ControlResult<Vec<(u64, u64, u64)>> {
let log = self.stores.txlog();
if log.is_follower() {
return Err(ControlError::FailedPrecondition(
"retention: a twin's log is the primary's numbers; retain on the primary".into(),
));
}
Ok(log.layout())
}
fn data_volume(&self) -> Option<(u64, u64)> {
if self.stores.txlog().is_released() {
return None;
}
let vfs = rustix::fs::statvfs(self.stores.root()).ok()?;
let total = vfs.f_blocks.saturating_mul(vfs.f_frsize);
let free = vfs.f_bfree.saturating_mul(vfs.f_frsize);
(total > 0).then(|| (total, total.saturating_sub(free)))
}
fn twin_owed_read(&self) -> ControlResult<Vec<u8>> {
match std::fs::read(self.stores.root().join(crate::twin::TWIN_OWED_FILE)) {
Ok(b) => Ok(b),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(ControlError::Unavailable(format!(
"reading {}: {e}",
crate::twin::TWIN_OWED_FILE
))),
}
}
fn twin_owed_write(&self, bytes: &[u8]) -> ControlResult<()> {
if self.stores.txlog().is_released() {
return Err(ControlError::Unavailable("the data set is released for a growth; twin-owed.json is written after the remount".into()));
}
crate::durable::write_atomic_unframed(
self.stores.root(),
crate::twin::TWIN_OWED_FILE,
bytes,
)
.map_err(|e| {
ControlError::Unavailable(format!("writing {}: {e:#}", crate::twin::TWIN_OWED_FILE))
})
}
fn restore_snapshot_chunk(&self, chunk: &gunnar_proto::LogChunk) -> ControlResult<Option<u64>> {
let kind = u16::try_from(chunk.kind).map_err(|_| {
ControlError::InvalidArgument(format!("snapshot piece of kind {}", chunk.kind))
})?;
let piece = crate::twin::snapshot::Piece {
kind,
meta: &chunk.meta,
blob_len: chunk.blob_len,
blob: &chunk.blob,
last: chunk.last,
};
crate::twin::snapshot::restore_piece(&self.stores, piece)
.map_err(|err| ControlError::Internal(format!("installing the snapshot: {err:#}")))
}
fn health_alarms(&self) -> Vec<gunnar_proto::HealthAlarm> {
#[allow(unused_mut)]
let mut alarms = Vec::new();
if !self.stores.is_twin() && crate::twin::is_restoring(self.stores.root()) {
alarms.push(gunnar_proto::HealthAlarm {
event: "twin.restoring_at_open".into(),
count: 1,
since_unix_ms: 0,
detail: "this box is MID-RESTORE: it originates no transactions and refuses every \
control-plane write until a restore finishes — resend the twin's log"
.into(),
});
}
if !self.stores.is_twin() {
if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
alarms.push(gunnar_proto::HealthAlarm {
event: "twin.fenced_by_generation".into(),
count: 1,
since_unix_ms: 0,
detail: why,
});
}
}
if self.stores.is_twin() {
if let Some(d) = crate::twin::snapshot::dropped(self.stores.root()) {
let too_small = crate::twin::snapshot::too_small_total(self.stores.root())
.map(|t| format!(" The resync did not fit even then: this twin's volume ({t} bytes) must be grown to at least main's data set before it is tried again ({}).", gunnar_grpc::TWIN_TOO_SMALL))
.unwrap_or_default();
alarms.push(gunnar_proto::HealthAlarm {
event: "twin.resync_without_copy".into(),
count: 1,
since_unix_ms: d.at_unix_ms,
detail: format!("{}.{too_small}", d.sentence()),
});
}
}
if !self.stores.is_twin() {
let (total, _) = crate::twin::volume_bytes(self.stores.root());
match crate::twin::txlog_alarm_cap(
std::env::var(crate::twin::TXLOG_ALARM_ENV).ok().as_deref(),
total,
) {
Ok(Some((cap, how))) => alarms.extend(crate::twin::txlog_size_alarm(
self.stores.txlog().bytes(),
cap,
&how,
)),
Ok(None) => {}
Err(why) => alarms.push(gunnar_proto::HealthAlarm {
event: crate::twin::TXLOG_SIZE_ALARM.into(),
count: 1,
since_unix_ms: 0,
detail: why,
}),
}
}
#[cfg(feature = "metered")]
if let Some(a) = self.entitlement.as_ref().and_then(|slot| slot.alarm()) {
alarms.push(a);
}
if let Some(a) = self.grow.as_ref().and_then(|g| g.alarm()) {
alarms.push(a);
}
alarms.extend(crate::feed::alarms());
alarms
}
fn is_twin(&self) -> bool {
self.stores.is_twin()
}
fn frozen(&self) -> Option<String> {
if self.stores.is_twin() {
Some("this box is a twin".to_owned())
} else if self
.stores
.browse()
.load(std::sync::atomic::Ordering::SeqCst)
{
Some("this box is a browse replica".to_owned())
} else {
None
}
}
fn log_follows(&self) -> bool {
self.stores.is_twin() || self.stores.txlog().is_follower()
}
fn topology_facts(&self) -> gunnar_grpc::topology::Facts {
crate::topology::facts(self.stores.root())
}
fn maintenance_reasons(&self) -> Vec<String> {
#[allow(unused_mut)]
let mut why = Vec::new();
#[cfg(feature = "metered")]
if self
.entitlement
.as_ref()
.is_some_and(|slot| slot.maintenance())
{
why.push(crate::entitlement::KEY_MISMATCH_ALARM.to_string());
}
why
}
fn primary_heartbeat(&self, silent_after_ms: u64) -> Option<gunnar_proto::PrimaryHeartbeat> {
let cfg = crate::twin::TwinConfig::load(self.stores.root())
.ok()
.flatten()?;
let st = crate::twin::status(self.stores.root());
let (last, primary_head, applied) = match &st {
Some(s) => (s.last_answer_unix_ms, s.primary_head_seq, s.applied_seq),
None => (0, 0, self.stores.txlog().head_seq()),
};
let own = st.as_ref().map(|s| (s.state.to_string(), s.why.clone()));
Some(gunnar_grpc::heartbeat::primary_view(
&cfg.primary,
last,
primary_head,
applied,
crate::twin::now_ms(),
silent_after_ms,
own.as_ref().map(|(a, b)| (a.as_str(), b.as_str())),
))
}
fn log_head(&self) -> ControlResult<gunnar_proto::LogHeadResponse> {
self.log_head_signed("")
}
fn log_head_signed(&self, nonce: &str) -> ControlResult<gunnar_proto::LogHeadResponse> {
if !nonce.trim().is_empty() {
let root = self.stores.root();
if crate::twin::is_restoring(root) {
return Err(ControlError::Unavailable(
"this box is being restored from a twin and is not a primary yet: it answers a twin's poll \
once the replay has finished and the box has restarted"
.to_owned(),
));
}
if crate::twin::reseal_pending(root) {
return Err(ControlError::Unavailable(
"this box finished a restore and must restart before it answers a twin: its host key is \
sealed under the key-encrypting key the restart re-seals, so until then it cannot prove \
which box it is"
.to_owned(),
));
}
}
let mut head = crate::twin::own_head(&self.stores);
let root = self.stores.root();
let generation = if crate::twin::is_twin_root(root) {
crate::twin::pinned_generation(root)
} else {
crate::twin::log_generation(root)
};
head.generation = generation
.map_err(|err| ControlError::Internal(format!("this box's log generation: {err:#}")))?;
let (key, sig) = crate::twin::sign_head_answer(
&self.stores,
nonce,
&head.log_id,
head.first_seq,
head.head_seq,
head.generation,
);
head.answer_key = key;
head.answer_sig = sig;
Ok(head)
}
fn fence_by_generation(&self, proof: &gunnar_proto::GenerationProof) -> ControlResult<()> {
crate::twin::fence_by_proof(&self.stores, proof)
.map(drop)
.map_err(|err| {
ControlError::Internal(format!(
"fencing this box off as an old generation of its log: {err:#}"
))
})
}
fn log_fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> ControlResult<Vec<gunnar_proto::LogEntry>> {
let log = self.readable_log()?;
let entries =
crate::twin::fetch_range(log, from, to, max_bytes).map_err(log_read_refusal)?;
Ok(entries
.into_iter()
.map(|e| gunnar_proto::LogEntry {
seq: e.seq,
kind: e.kind as u32,
meta: e.meta,
blob: e.blob,
})
.collect())
}
fn log_fetch_stream(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
self.log_fetch_stream_at(from, 0, to, max_bytes)
}
fn log_fetch_stream_at(
&self,
from: u64,
from_offset: u64,
to: u64,
max_bytes: u64,
) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
let log = self.readable_log()?;
crate::twin::check_range(log, from, to).map_err(log_read_refusal)?;
let stores = Arc::clone(&self.stores);
Ok(Box::new(move |sink| {
let mut out = ChunkOut { sink, open: None };
crate::twin::stream_range_at(stores.txlog(), from, from_offset, to, max_bytes, &mut out)
.map(drop)
.map_err(|err| {
let err = match err.downcast::<ControlError>() {
Ok(said) => return said,
Err(other) => other,
};
match err.downcast::<gunnar_core::txlog::Error>() {
Ok(e) => log_read_refusal(e),
Err(other) => ControlError::Internal(format!("transaction log: {other:#}")),
}
})
}))
}
#[cfg(feature = "replication")]
fn replication_status(
&self,
include_refs: bool,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
self.replication_slot()?;
self.replication_snapshot(include_refs)
}
#[cfg(feature = "replication")]
fn replication_flush(
&self,
request: &gunnar_proto::FlushReplicationRequest,
) -> ControlResult<gunnar_proto::FlushReplicationResponse> {
let slot = self.replication_slot()?;
let began = gunnar_grpc::bus::now_unix_ms();
let started = std::time::Instant::now();
let ceiling = if request.timeout_secs == 0 {
abdicate_flush_secs()
} else {
u64::from(request.timeout_secs)
};
let Some(replicator) = slot.replicator() else {
return Err(ControlError::FailedPrecondition(
"no replicator is running on this box (no standby configured, or this box is not a primary); there is nothing to flush and nothing to prove".to_owned(),
));
};
if request.resync {
replicator
.resync_scoped(None, None, &[])
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
}
let deadline = started + std::time::Duration::from_secs(ceiling);
loop {
let drained = replicator.status().iter().all(|s| s.pending == 0);
if drained || std::time::Instant::now() >= deadline {
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
let status = replicator.status();
let counters = replicator.counters();
let pending: u64 = status.iter().map(|s| s.pending as u64).sum();
let needs_full_resync = status.iter().any(|s| s.needs_full_resync);
let last_success = status
.iter()
.map(|s| s.last_success_unix_ms.unwrap_or(0))
.min()
.unwrap_or(0);
let in_flight = self.admission.as_ref().map(|a| a.in_flight()).unwrap_or(0);
let catalog_repos = self
.stores
.catalog()
.counts()
.map(|(_, repos)| repos as u64)
.unwrap_or(0);
let standbys = self.replication_snapshot(false)?.standbys;
let caught_up = in_flight == 0
&& !status.is_empty()
&& pending == 0
&& !needs_full_resync
&& status.iter().all(|s| {
s.last_success_unix_ms.is_some_and(|t| t >= began)
|| s.pending == 0 && s.last_success_unix_ms.is_some()
});
let verdict = if status.is_empty() {
"no standby is configured: nothing was flushed and nothing is proven".to_owned()
} else if caught_up {
format!(
"CAUGHT UP as far as replication can prove: {} standby(s), nothing pending, no full-resync hole, {} transfer(s) in flight, {} bus gap(s) since boot{}. What this does NOT prove (DATA-SET-GROWTH-FLOW.md §2.1): refs outside heads/tags, LFS objects, the control database, visibility, tombstones, and — unless `resync` ran once since the twin existed — repositories nobody pushed since",
status.len(),
in_flight,
counters.bus_gaps,
if request.resync { " (every catalog repository was enqueued)" } else { "" }
)
} else {
let names: Vec<String> = status
.iter()
.filter(|s| s.pending > 0 || s.needs_full_resync)
.map(|s| {
format!(
"{} (pending {}{})",
s.name,
s.pending,
if s.needs_full_resync {
", needs full resync"
} else {
""
}
)
})
.collect();
format!(
"NOT caught up after {} s: {} transfer(s) in flight, {} repository(ies) pending [{}]{}; the growth must not unmount the set on this",
started.elapsed().as_secs(),
in_flight,
pending,
names.join(", "),
if needs_full_resync { " — a standby has a hole retrying will not close: run `replication resync`" } else { "" }
)
};
tracing::info!(
target: "gunnar::replication",
event = "replication.flush",
caught_up,
in_flight,
pending,
bus_gaps = counters.bus_gaps,
needs_full_resync,
waited_ms = started.elapsed().as_millis() as u64,
"{verdict}"
);
Ok(gunnar_proto::FlushReplicationResponse {
caught_up,
in_flight,
pending,
bus_gaps: counters.bus_gaps,
needs_full_resync,
last_success_unix_ms: last_success,
flushed_at_unix_ms: began,
waited_ms: started.elapsed().as_millis() as u64,
standbys,
catalog_repos,
verdict,
})
}
#[cfg(feature = "replication")]
fn replication_demote(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
config.clear_targets();
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
self.replication_snapshot(false)
}
#[cfg(feature = "replication")]
fn replication_set_role(
&self,
role: gunnar_proto::ReplicationRole,
) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let wanted = crate::mirror::Role::from_proto(role);
if config.role.is_browse() && wanted == crate::mirror::Role::Primary {
return Err(ControlError::FailedPrecondition(
"this box is a browse replica; it cannot be promoted to primary. It holds only \
public repositories and serves anonymous reads — first `gunnar control \
replication demote` it to a plain standby, then promote that"
.into(),
));
}
config.role = wanted;
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
self.replication_snapshot(false)
}
#[cfg(feature = "replication")]
fn replication_abdicate(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
let slot = self.replication_slot()?;
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
config.role = crate::mirror::Role::Standby;
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
if let Some(replicator) = slot.replicator() {
replicator
.resync_scoped(None, None, &[])
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let deadline =
std::time::Instant::now() + std::time::Duration::from_secs(abdicate_flush_secs());
loop {
let drained = replicator.status().iter().all(|s| s.pending == 0);
if drained || std::time::Instant::now() >= deadline {
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
}
let mut config = slot
.config()
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
config.clear_targets();
slot.apply(config)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
self.replication_snapshot(false)
}
#[cfg(feature = "replication")]
fn replication_resync(
&self,
name: &str,
repo: Option<&gunnar_proto::RepoPath>,
except: &[gunnar_proto::RepoPath],
) -> ControlResult<gunnar_proto::ResyncReplicationResponse> {
let slot = self.replication_slot()?;
let Some(replicator) = slot.replicator() else {
return Err(ControlError::Unavailable(format!(
"no replicator is running, so nothing would drain a backfill queue. \
`replication.json` holds {} target(s); the server has not started them.",
slot.config().map(|c| c.targets.len()).unwrap_or(0)
)));
};
let only = match repo {
None => None,
Some(path) => Some(Self::repo_ref(path)?),
};
let except = except
.iter()
.map(Self::repo_ref)
.collect::<ControlResult<Vec<_>>>()?;
let reports = replicator
.resync_scoped(Some(name), only.as_ref(), &except)
.map_err(|err| ControlError::NotFound(format!("{err:#}")))?;
Ok(gunnar_proto::ResyncReplicationResponse {
standbys: reports
.into_iter()
.map(|r| gunnar_proto::ResyncStandby {
standby: r.standby,
in_scope: r.in_scope,
queued: r.queued,
already_pending: r.already_pending,
refused: r.refused,
sha1_repos: r.sha1_repos,
excluded: r.excluded,
})
.collect(),
status: Some(self.replication_snapshot(false)?),
})
}
#[cfg(feature = "webhooks")]
fn webhook_add(
&self,
request: gunnar_proto::AddWebhookRequest,
) -> ControlResult<gunnar_proto::WebhookListResponse> {
let slot = self.webhook_slot()?;
let endpoint = crate::webhook::Endpoint {
name: request.name,
url: request.url,
kinds: request.kinds,
namespace: request.namespace,
repo: request.repo,
secret: Some(request.secret).filter(|s| !s.is_empty()),
added_at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
};
slot.add(endpoint)
.map_err(|err| ControlError::InvalidArgument(format!("{err:#}")))
}
#[cfg(feature = "webhooks")]
fn webhook_list(&self) -> ControlResult<gunnar_proto::WebhookListResponse> {
self.webhook_slot()?
.list_response()
.map_err(|err| ControlError::Internal(format!("{err:#}")))
}
#[cfg(feature = "webhooks")]
fn webhook_remove(&self, name: &str) -> ControlResult<gunnar_proto::WebhookListResponse> {
let slot = self.webhook_slot()?;
slot.remove(name)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?
.ok_or_else(|| ControlError::NotFound(format!("no webhook endpoint named {name:?}")))
}
}
const MAX_RESPONSE_BYTES: u64 = 4 * 1024 * 1024;
const READ_OBJECT_MEMORY_BUDGET: u64 = 512 * 1024 * 1024;
const MAX_CONCURRENT_INFLATES: u64 = 8;
const DEFAULT_MAX_OBJECT_BYTES: u64 = READ_OBJECT_MEMORY_BUDGET / MAX_CONCURRENT_INFLATES;
impl CoreControlPlane {
fn usage_readable(&self) -> ControlResult<()> {
if self.stores.txlog().is_released()
|| crate::twin::maintenance::is_held(self.stores.root())
{
return Err(ControlError::Unavailable(format!(
"{} this box's data set is being grown (unmounted): its usage is read again after the remount — ask again",
gunnar_grpc::PRIMARY_GROWING
)));
}
Ok(())
}
fn open_objects_of(&self, namespace: &str) -> u64 {
let prefix = self
.stores
.repos_root()
.canonicalize()
.unwrap_or_else(|_| self.stores.repos_root().to_path_buf())
.join(namespace);
self.stores
.engine()
.open_object_counts()
.into_iter()
.filter(|(dir, _)| dir.parent() == Some(prefix.as_path()))
.map(|(_, n)| n)
.sum()
}
fn usage_row(
&self,
name: &str,
held: &crate::usage_ledger::Held,
objects: u64,
measured_at_unix_ms: i64,
) -> gunnar_proto::NamespaceUsage {
let mut row = gunnar_proto::NamespaceUsage {
namespace: name.to_owned(),
measured_at_unix_ms,
pack_bytes: held.bytes.pack,
cache_bytes: held.bytes.cache,
lfs_bytes: held.bytes.lfs,
tombstoned_bytes: held.tombstoned_bytes,
repos: held.repos,
objects,
..Default::default()
};
fill_flow(&mut row, self.stores.flow(), name, measured_at_unix_ms);
row
}
fn usage_all(&self) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
let catalog = self.stores.catalog();
let namespaces: Vec<String> = catalog
.namespaces()
.map_err(internal)?
.into_iter()
.map(|ns| ns.name)
.collect();
let repos = catalog.list(None).map_err(internal)?;
let tombstones = self.stores.tombstones().unwrap_or_default();
let open_objects: std::collections::HashMap<std::path::PathBuf, u64> = self
.stores
.engine()
.open_object_counts()
.into_iter()
.collect();
let flow = self.stores.flow();
let measured_at_unix_ms = gunnar_core::now_ms();
let mut by_ns: std::collections::HashMap<&str, Vec<&gunnar_core::Repo>> =
std::collections::HashMap::new();
for repo in &repos {
by_ns.entry(repo.namespace.as_str()).or_default().push(repo);
}
let mut out = Vec::with_capacity(namespaces.len());
for name in namespaces {
let mut row = gunnar_proto::NamespaceUsage {
namespace: name.clone(),
measured_at_unix_ms,
..Default::default()
};
for repo in by_ns.get(name.as_str()).map(Vec::as_slice).unwrap_or(&[]) {
let path = RepoRef::new(&repo.namespace, &repo.name).map_err(|err| {
ControlError::Internal(format!(
"a catalogued repository has an illegal path: {err}"
))
})?;
let dir = self
.stores
.repo_dir(&path)
.map_err(|err| ControlError::Internal(format!("{err:#}")))?;
let bytes = repo_bytes(&dir);
row.repos += 1;
row.pack_bytes += bytes.pack;
row.cache_bytes += bytes.cache;
row.lfs_bytes += bytes.lfs;
row.objects += open_objects.get(&dir).copied().unwrap_or(0);
}
row.tombstoned_bytes = tombstones
.iter()
.filter(|t| t.path.namespace() == name)
.map(|t| t.bytes)
.sum();
fill_flow(&mut row, flow, &name, measured_at_unix_ms);
out.push(row);
}
Ok(out)
}
}
fn fill_flow(
row: &mut gunnar_proto::NamespaceUsage,
flow: &gunnar_core::FlowLedger,
name: &str,
now_unix_ms: i64,
) {
let glance = flow.glance(name, now_unix_ms);
row.bytes_in_today = glance.today.bytes_in;
row.bytes_out_today = glance.today.bytes_out;
row.pushes_today = glance.today.pushes;
row.fetches_today = glance.today.fetches;
row.refused_today = glance.today.refused();
row.last_push_unix_ms = glance.last_push_unix_ms;
row.last_fetch_unix_ms = glance.last_fetch_unix_ms;
}
fn object_dir_stats(git_dir: &std::path::Path) -> (u64, u64) {
fn walk(dir: &std::path::Path, loose: &mut u64, bytes: &mut u64, in_loose_shard: bool) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let Ok(kind) = entry.file_type() else {
continue;
};
let name = entry.file_name();
let name = name.to_string_lossy();
if kind.is_dir() {
let shard = name.len() == 2 && name.chars().all(|c| c.is_ascii_hexdigit());
walk(&entry.path(), loose, bytes, shard);
} else if kind.is_file() {
if in_loose_shard {
*loose += 1;
}
if let Ok(meta) = entry.metadata() {
*bytes += meta.len();
}
}
}
}
let mut loose = 0;
let mut bytes = 0;
walk(&git_dir.join("objects"), &mut loose, &mut bytes, false);
(loose, bytes)
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RepoBytes {
pub pack: u64,
pub cache: u64,
pub lfs: u64,
}
impl RepoBytes {
pub fn hot(&self) -> u64 {
self.pack + self.cache
}
}
pub fn repo_bytes(git_dir: &std::path::Path) -> RepoBytes {
fn walk(dir: &std::path::Path, acc: &mut RepoBytes) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let Ok(meta) = entry.metadata() else {
continue;
};
if meta.is_dir() {
walk(&entry.path(), acc);
} else if entry
.file_name()
.to_string_lossy()
.starts_with("objects.exploded")
{
acc.cache += meta.len();
} else {
acc.pack += meta.len();
}
}
}
let mut acc = RepoBytes::default();
let Ok(entries) = std::fs::read_dir(git_dir) else {
return acc;
};
for entry in entries.flatten() {
let Ok(meta) = entry.metadata() else {
continue;
};
if meta.is_dir() && entry.file_name() == gunnar_lfs::fs_store::LFS_DIR {
acc.lfs += crate::stores::dir_bytes(&entry.path());
} else if meta.is_dir() {
walk(&entry.path(), &mut acc);
} else {
acc.pack += meta.len();
}
}
acc
}
impl gunnar_grpc::Identity for CoreControlPlane {
fn awaiting_enrolment(&self, fingerprint: &str) -> Option<String> {
self.stores.awaiting_enrolment(fingerprint)
}
fn registered_key(
&self,
fingerprint: &str,
) -> ControlResult<Option<gunnar_grpc::RegisteredKey>> {
let Ok(parsed) = gunnar_core::Fingerprint::parse(fingerprint) else {
return Ok(None);
};
let Some(principal) = self
.stores
.auth()
.principal_by_fingerprint(&parsed)
.map_err(internal)?
else {
return Ok(None);
};
let key = self
.stores
.auth()
.keys_of(&principal.id)
.map_err(internal)?
.into_iter()
.find(|key| key.fingerprint_sha256 == fingerprint);
let machine = principal.kind == gunnar_core::PrincipalKind::Machine;
Ok(key.map(|key| gunnar_grpc::RegisteredKey {
principal_id: principal.id,
machine,
openssh: key.openssh,
}))
}
fn registered_certificate(
&self,
der: &[u8],
) -> ControlResult<Option<gunnar_grpc::CertificateIdentity>> {
let fingerprint = gunnar_core::auth::fingerprint_of_certificate(der);
let Some(principal) = self
.stores
.auth()
.principal_by_fingerprint(&fingerprint)
.map_err(internal)?
else {
return Ok(None);
};
let registered = self
.stores
.auth()
.keys_of(&principal.id)
.map_err(internal)?
.into_iter()
.any(|key| key.fingerprint_sha256 == fingerprint.to_string());
if !registered {
return Ok(None);
}
Ok(Some(gunnar_grpc::CertificateIdentity {
principal_id: principal.id,
fingerprint: fingerprint.to_string(),
}))
}
fn narrowed_methods(&self, principal_id: &str) -> ControlResult<Option<Vec<String>>> {
if principal_id == gunnar_grpc::REPAIR_PRINCIPAL {
return Ok(Some(
gunnar_grpc::REPAIR_METHODS
.iter()
.map(|m| m.to_string())
.collect(),
));
}
#[cfg(feature = "wg-client")]
return crate::fed_principals::narrowed_of(self.stores.root(), principal_id)
.map_err(ControlError::Internal);
#[cfg(not(feature = "wg-client"))]
{
let _ = principal_id;
Ok(None)
}
}
fn may(
&self,
principal_id: &str,
access: &gunnar_grpc::Access,
level: gunnar_proto::GrantLevel,
) -> ControlResult<bool> {
let level = level_in(level)?;
let auth = self.stores.auth();
match access {
gunnar_grpc::Access::Server => auth.may_server(principal_id, level).map_err(internal),
gunnar_grpc::Access::Namespace(namespace) => auth
.may_namespace(principal_id, namespace, level)
.map_err(internal),
gunnar_grpc::Access::Repo(path) => {
let repo_ref = Self::repo_ref(path)?;
let Some(repo) = self.stores.catalog().get(&repo_ref).map_err(internal)? else {
return Ok(false);
};
auth.may(principal_id, &repo, level).map_err(internal)
}
}
}
fn may_each_repo(
&self,
principal_id: &str,
paths: &[RepoPath],
level: gunnar_proto::GrantLevel,
) -> ControlResult<Vec<bool>> {
let level = level_in(level)?;
let refs = paths
.iter()
.map(Self::repo_ref)
.collect::<ControlResult<Vec<_>>>()?;
let rows = self.stores.catalog().get_many(&refs).map_err(internal)?;
let known: Vec<gunnar_core::model::Repo> = rows.iter().flatten().cloned().collect();
let mut answers = self
.stores
.auth()
.may_each(principal_id, &known, level)
.map_err(internal)?
.into_iter();
Ok(rows
.iter()
.map(|row| row.is_some() && answers.next().unwrap_or(false))
.collect())
}
}
fn protection_out(rule: &gunnar_policy::ProtectionRule) -> BranchProtection {
BranchProtection {
ref_pattern: rule.pattern.as_str().to_owned(),
allow_force_push: rule.allow_force_push,
allow_delete: rule.allow_delete,
require_signature: rule.require_signature,
allowed_principal_ids: rule.allowed_principals.clone().unwrap_or_default(),
max_object_bytes: 0,
max_objects_per_push: 0,
forbid_self_approval: rule.forbid_self_approval,
required_approvals: rule.required_approvals,
}
}
fn ref_name_policy_out(policy: &gunnar_policy::RefNamePolicy) -> RefNamePolicy {
RefNamePolicy {
allow_patterns: policy.allow.iter().map(|p| p.as_str().to_owned()).collect(),
deny_patterns: policy.deny.iter().map(|p| p.as_str().to_owned()).collect(),
}
}
fn mirror_out(record: &gunnar_core::MirrorRecord) -> MirrorConfig {
let (namespace, repo) = record
.repo
.split_once('/')
.unwrap_or((record.repo.as_str(), ""));
MirrorConfig {
repo: Some(RepoPath::new(namespace, repo)),
remote_url: record.remote_url.clone(),
prune: record.prune,
state: if record.last_error.is_empty() {
gunnar_proto::MirrorState::Idle as i32
} else {
gunnar_proto::MirrorState::Failed as i32
},
added_at_unix_ms: record.added_at_unix_ms,
last_sync_at_unix_ms: record.last_sync_at_unix_ms,
last_error: record.last_error.clone(),
sync_count: record.sync_count,
credential: credential_kind_out(record.credential) as i32,
credential_username: record.credential_username.clone(),
job: None,
repo_pending: record.creates_repo,
lfs_not_copied: record.lfs_detected,
}
}
fn credential_kind_in(
kind: gunnar_proto::MirrorCredentialKind,
) -> gunnar_core::MirrorCredentialKind {
match kind {
gunnar_proto::MirrorCredentialKind::Unspecified => gunnar_core::MirrorCredentialKind::None,
gunnar_proto::MirrorCredentialKind::SshKey => gunnar_core::MirrorCredentialKind::SshKey,
gunnar_proto::MirrorCredentialKind::Token => gunnar_core::MirrorCredentialKind::Token,
}
}
fn credential_kind_out(
kind: gunnar_core::MirrorCredentialKind,
) -> gunnar_proto::MirrorCredentialKind {
match kind {
gunnar_core::MirrorCredentialKind::None => gunnar_proto::MirrorCredentialKind::Unspecified,
gunnar_core::MirrorCredentialKind::SshKey => gunnar_proto::MirrorCredentialKind::SshKey,
gunnar_core::MirrorCredentialKind::Token => gunnar_proto::MirrorCredentialKind::Token,
}
}
fn read_credential(
stores: &Arc<ServerStores>,
repo: &gunnar_core::RepoRef,
) -> Result<crate::mirror_fetch::Credential, String> {
let record = stores
.mirrors()
.get(repo)
.map_err(|err| {
format!(
"reading the mirror registration for {}: {err:#}",
repo.key()
)
})?
.ok_or_else(|| format!("no mirror is registered for {} any more", repo.key()))?;
if !record.credential.is_set() {
return Ok(crate::mirror_fetch::Credential::None);
}
let secrets = stores
.secrets()
.map_err(|err| format!("the sealed secret store is not available: {err:#}"))?;
let key = gunnar_core::credential_secret_key(repo);
let sealed = gunnar_store::SecretStore::get(&secrets, &key)
.map_err(|err| {
format!(
"the mirror credential for {} could not be read: {err}",
repo.key()
)
})?
.ok_or_else(|| {
format!(
"{} is registered as having {}, and the sealed value is not in the secret \
store. Set it again with `Mirror.SetCredential`; dialling anonymously \
instead would be refused by the upstream and read as a bad key.",
repo.key(),
record.credential.name()
)
})?;
Ok(match record.credential {
gunnar_core::MirrorCredentialKind::SshKey => {
crate::mirror_fetch::Credential::SshKey(sealed)
}
gunnar_core::MirrorCredentialKind::Token => crate::mirror_fetch::Credential::Token {
username: record.credential_username.clone(),
secret: String::from_utf8(sealed).map_err(|_| {
format!(
"the token stored for {} is not utf-8, so it cannot be an HTTP header value",
repo.key()
)
})?,
},
gunnar_core::MirrorCredentialKind::None => crate::mirror_fetch::Credential::None,
})
}
fn audit_line(
stores: &ServerStores,
repo: &gunnar_core::RepoRef,
kind: &str,
detail: &str,
refused: bool,
) {
tracing::info!(
target: gunnar_grpc::AUDIT_TARGET,
event = kind,
repo = %repo.key(),
detail,
"mirror"
);
let Ok(Some(account)) = stores.auth().account_of_namespace(repo.namespace()) else {
return;
};
let event = gunnar_core::account_audit::AccountEvent {
kind,
actor: "",
actor_role: "system",
target: &repo.key(),
detail,
outcome: if refused {
gunnar_core::account_audit::Outcome::Refused
} else {
gunnar_core::account_audit::Outcome::Ok
},
};
if let Err(err) = stores
.account_audit()
.append(&account, event, gunnar_core::now_ms())
{
tracing::warn!(target: "gunnar::git", repo = %repo.key(), %err, "a mirror sync's audit line was not written");
}
}
fn audit_mirror_outcome(
stores: &ServerStores,
repo: &gunnar_core::RepoRef,
order: &crate::mirror_queue::MirrorOrder,
first: bool,
error: Option<&str>,
report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>,
) {
let upstream = crate::mirror_fetch::scrub(&order.upstream);
match (first, error) {
(true, None) => {
let refs = report
.snapshot()
.map(|job| job.report.updates.len())
.unwrap_or_default();
audit_line(
stores,
repo,
"mirror.import.done",
&format!("imported {refs} reference(s) from {upstream}"),
false,
);
}
(true, Some(why)) => audit_line(
stores,
repo,
"mirror.import.failed",
&format!("from {upstream}: {why}"),
true,
),
(false, Some(why)) => audit_line(
stores,
repo,
"mirror.sync.failed",
&format!("from {upstream}: {why}"),
true,
),
(false, None) => {}
}
}
fn mirror_job_out(path: &RepoPath, job: crate::mirror_queue::MirrorJob) -> gunnar_proto::MirrorJob {
let report = job.report;
let report_refused = report.refused;
let (progress_done, progress_total) = job.progress.unwrap_or((0, 0));
gunnar_proto::MirrorJob {
job_id: job.job_id,
path: Some(path.clone()),
state: match job.state {
crate::mirror_queue::MirrorState::Queued => gunnar_proto::MirrorJobState::Queued as i32,
crate::mirror_queue::MirrorState::Running => {
gunnar_proto::MirrorJobState::Running as i32
}
crate::mirror_queue::MirrorState::Done => gunnar_proto::MirrorJobState::Done as i32,
crate::mirror_queue::MirrorState::Failed => gunnar_proto::MirrorJobState::Failed as i32,
},
queue_position: job.queue_position,
has_progress: job.progress.is_some(),
progress_done,
progress_total,
objects: report.objects,
bytes: report.bytes,
updates: report
.updates
.into_iter()
.map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
ref_name,
old_oid,
new_oid,
})
.collect(),
error: job.error.unwrap_or_default(),
enqueued_at_unix_ms: job.enqueued_at_unix_ms,
started_at_unix_ms: job.started_at_unix_ms,
finished_at_unix_ms: job.finished_at_unix_ms,
refused_reason: match (job.state, report_refused) {
(crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::Plan) => {
gunnar_proto::MirrorRefusal::Plan as i32
}
(crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::PackCap) => {
gunnar_proto::MirrorRefusal::PackCap as i32
}
_ => gunnar_proto::MirrorRefusal::Unspecified as i32,
},
}
}
fn to_proto_membership(
signed: gunnar_core::SignedMembershipEvent,
) -> gunnar_proto::SignedMembershipEvent {
gunnar_proto::SignedMembershipEvent {
event: Some(gunnar_proto::MembershipEvent {
group: signed.event.group,
member: signed.event.member,
action: proto_action(signed.event.action) as i32,
actor: signed.event.actor,
at_unix_ms: signed.event.at_unix_ms,
}),
key_fingerprint: signed.key_fingerprint,
signature: signed.signature,
}
}
fn from_proto_attestation(
wire: gunnar_proto::SignedKeyAttestation,
) -> ControlResult<gunnar_core::SignedKeyAttestation> {
let attestation = wire.attestation.ok_or_else(|| {
ControlError::InvalidArgument("`attestation.attestation` is required".to_owned())
})?;
Ok(gunnar_core::SignedKeyAttestation {
attestation: gunnar_core::KeyAttestation {
principal_id: attestation.principal_id,
fingerprint: attestation.fingerprint,
at_unix_ms: attestation.at_unix_ms,
},
key_fingerprint: wire.key_fingerprint,
signature: wire.signature,
})
}
fn from_proto_membership(
wire: gunnar_proto::SignedMembershipEvent,
) -> ControlResult<gunnar_core::SignedMembershipEvent> {
let event = wire
.event
.ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
let action = match gunnar_proto::MembershipAction::try_from(event.action) {
Ok(gunnar_proto::MembershipAction::Add) => gunnar_core::MembershipAction::Add,
Ok(gunnar_proto::MembershipAction::Remove) => gunnar_core::MembershipAction::Remove,
_ => {
return Err(ControlError::InvalidArgument(
"`event.action` must be ADD or REMOVE".to_owned(),
))
}
};
Ok(gunnar_core::SignedMembershipEvent {
event: gunnar_core::MembershipEvent {
group: event.group,
member: event.member,
action,
actor: event.actor,
at_unix_ms: event.at_unix_ms,
},
key_fingerprint: wire.key_fingerprint,
signature: wire.signature,
})
}
fn proto_action(action: gunnar_core::MembershipAction) -> gunnar_proto::MembershipAction {
match action {
gunnar_core::MembershipAction::Add => gunnar_proto::MembershipAction::Add,
gunnar_core::MembershipAction::Remove => gunnar_proto::MembershipAction::Remove,
}
}
fn to_proto_event(signed: gunnar_core::SignedEvent) -> gunnar_proto::SignedForgeEvent {
use gunnar_proto::forge_event::Kind;
let kind = match signed.event.event {
gunnar_core::Event::Commented {
body,
path,
line,
blob_oid,
} => Kind::Commented(gunnar_proto::Commented {
body,
path,
line,
blob_oid,
}),
gunnar_core::Event::Approved { commit } => {
Kind::Approved(gunnar_proto::Approved { commit })
}
gunnar_core::Event::Rejected { commit, reason } => {
Kind::Rejected(gunnar_proto::Rejected { commit, reason })
}
gunnar_core::Event::IssueOpened { title, body } => {
Kind::IssueOpened(gunnar_proto::IssueOpened { title, body })
}
gunnar_core::Event::IssueCommented { body } => {
Kind::IssueCommented(gunnar_proto::IssueCommented { body })
}
gunnar_core::Event::IssueClosed { reason } => {
Kind::IssueClosed(gunnar_proto::IssueClosed { reason })
}
};
gunnar_proto::SignedForgeEvent {
event: Some(gunnar_proto::ForgeEvent {
repo: signed.event.repo,
pull: signed.event.pull,
actor: signed.event.actor,
at_unix_ms: signed.event.at_unix_ms,
kind: Some(kind),
}),
key_fingerprint: signed.key_fingerprint,
signature: signed.signature,
attested: signed.attested.map(|a| gunnar_proto::ForgeAttestation {
session_stepped_up_at_unix_ms: a.session_stepped_up_at_unix_ms,
}),
}
}
fn issue_state(repo: &str, issue: &gunnar_core::forge::Issue) -> gunnar_proto::IssueState {
gunnar_proto::IssueState {
repo: repo.to_owned(),
number: issue.number,
title: issue.title.clone(),
body: issue.body.clone(),
author: issue.author.clone(),
open: issue.open,
comments: issue.comments as u64,
opened_at_unix_ms: issue.opened_at_unix_ms,
}
}
fn from_proto_event(
wire: gunnar_proto::SignedForgeEvent,
) -> ControlResult<gunnar_core::SignedEvent> {
use gunnar_proto::forge_event::Kind;
let event = wire
.event
.ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
let kind = event
.kind
.ok_or_else(|| ControlError::InvalidArgument("`event.kind` is required".to_owned()))?;
let inner = match kind {
Kind::Commented(it) => gunnar_core::Event::Commented {
body: it.body,
path: it.path,
line: it.line,
blob_oid: it.blob_oid,
},
Kind::Approved(it) => gunnar_core::Event::Approved { commit: it.commit },
Kind::Rejected(it) => gunnar_core::Event::Rejected {
commit: it.commit,
reason: it.reason,
},
Kind::IssueOpened(it) => gunnar_core::Event::IssueOpened {
title: it.title,
body: it.body,
},
Kind::IssueCommented(it) => gunnar_core::Event::IssueCommented { body: it.body },
Kind::IssueClosed(it) => gunnar_core::Event::IssueClosed { reason: it.reason },
};
Ok(gunnar_core::SignedEvent {
event: gunnar_core::ForgeEvent {
repo: event.repo,
pull: event.pull,
actor: event.actor,
at_unix_ms: event.at_unix_ms,
event: inner,
},
key_fingerprint: wire.key_fingerprint,
signature: wire.signature,
attested: None,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_write_on_a_restoring_box_is_unavailable_not_internal() {
let td = tempfile::tempdir().unwrap();
let stores = crate::stores::ServerStores::open(td.path()).unwrap();
stores.txlog().set_follower(true);
let p = gunnar_core::Principal::new("alice", "alice", gunnar_core::PrincipalKind::User)
.unwrap();
let err = stores
.auth()
.add_principal(p)
.map_err(internal)
.expect_err("a follower log takes no write");
assert!(matches!(err, ControlError::Unavailable(_)), "{err:?}");
assert!(err.to_string().contains("restor"), "{err}");
}
#[test]
fn internal_is_refused_by_name_and_the_two_real_visibilities_pass() {
let err = visibility_in(gunnar_proto::Visibility::Internal)
.expect_err("INTERNAL has no implementation and must be refused");
let text = err.to_string();
assert!(
text.contains("internal") && text.contains("private"),
"the refusal must name the value and the real choices: {text}"
);
assert_eq!(
visibility_in(gunnar_proto::Visibility::Public).expect("public is real"),
Visibility::Public
);
assert_eq!(
visibility_in(gunnar_proto::Visibility::Private).expect("private is real"),
Visibility::Private
);
assert_eq!(
visibility_in(gunnar_proto::Visibility::Unspecified).expect("unspecified defaults"),
Visibility::Private
);
}
#[cfg(feature = "replication")]
#[test]
fn abdicate_flush_ceiling_honours_a_sane_override_and_rejects_the_rest() {
assert_eq!(
resolve_flush_secs(Some("120")),
120,
"a positive override must win"
);
assert_eq!(
resolve_flush_secs(Some(" 90 ")),
90,
"surrounding whitespace is trimmed"
);
assert_eq!(resolve_flush_secs(None), ABDICATE_FLUSH_SECS);
assert_ne!(
ABDICATE_FLUSH_SECS, 120,
"default must differ from the override arm"
);
assert_eq!(
resolve_flush_secs(Some("")),
ABDICATE_FLUSH_SECS,
"blank falls back"
);
assert_eq!(
resolve_flush_secs(Some("nope")),
ABDICATE_FLUSH_SECS,
"non-numeric falls back"
);
assert_eq!(
resolve_flush_secs(Some("0")),
ABDICATE_FLUSH_SECS,
"zero would truncate — refused"
);
assert_eq!(
resolve_flush_secs(Some("-5")),
ABDICATE_FLUSH_SECS,
"negative does not parse as u64"
);
}
}
fn key_refusal(r: gunnar_core::encryption_key::KeyRefusal) -> ControlError {
use gunnar_core::encryption_key::KeyRefusal as R;
match r {
R::AlreadyHasKey(s) | R::MethodHeld(s) => ControlError::AlreadyExists(s),
R::NoSuchMethod(s) => ControlError::NotFound(s),
R::NoKey(s) => ControlError::FailedPrecondition(s),
R::LastMethod(s) => ControlError::FailedPrecondition(s),
R::Invalid(s) => ControlError::InvalidArgument(s),
R::Store(e) => internal(e),
}
}
fn method_in(
m: gunnar_proto::UnlockMethod,
) -> ControlResult<gunnar_core::encryption_key::UnlockMethod> {
Ok(gunnar_core::encryption_key::UnlockMethod {
id: String::new(),
kind: gunnar_core::encryption_key::MethodKind::parse(&m.kind).map_err(key_refusal)?,
label: m.label.trim().to_owned(),
method_ref: m.method_ref.trim().to_owned(),
rp_id: m.rp_id.trim().to_owned(),
wrap: m.wrap,
added_ms: 0,
})
}
fn encryption_key_out(
key: Option<gunnar_core::encryption_key::EncryptionKey>,
) -> gunnar_proto::GetEncryptionKeyResponse {
match key {
None => gunnar_proto::GetEncryptionKeyResponse::default(),
Some(k) => gunnar_proto::GetEncryptionKeyResponse {
exists: true,
public_key: k.public_key,
methods: k
.methods
.into_iter()
.map(|m| gunnar_proto::UnlockMethod {
id: m.id,
kind: m.kind.word().to_owned(),
label: m.label,
method_ref: m.method_ref,
rp_id: m.rp_id,
wrap: m.wrap,
added_ms: m.added_ms,
})
.collect(),
},
}
}