1use std::sync::Arc;
43
44use gunnar_core::{Grant, HashKind, Level, PrincipalKind, RepoRef, Scope, UnixMillis, Visibility};
45use gunnar_grpc::seam::ControlCounts;
46use gunnar_grpc::MergedRef;
47use gunnar_grpc::{ControlError, ControlPlane, ControlResult, NewRepo as ProtoNewRepo};
48use gunnar_proto::{
49 BranchProtection, Commit, FileChange, GrantLevel, MirrorConfig, Namespace, Principal,
50 PublicKey, PullRequest, PullRequestState, ReadLfsObjectResponse, ReadObjectResponse,
51 RefNamePolicy, RefTarget, Repo, RepoPath, RepoStorage, Signature, StorageInfoResponse,
52};
53use gunnar_store::git_storage_trait::{GitServe, ObjType, RefRow};
54use gunnar_store::object;
55use gunnar_wire::RefSnapshot;
56
57use crate::storage::OpenRepo;
58use crate::stores::ServerStores;
59
60#[cfg(feature = "replication")]
75const ABDICATE_FLUSH_SECS: u64 = 30;
76
77#[cfg(feature = "replication")]
82fn abdicate_flush_secs() -> u64 {
83 resolve_flush_secs(std::env::var("GUNNAR_ABDICATE_FLUSH_SECS").ok().as_deref())
87}
88
89#[cfg(feature = "replication")]
92fn resolve_flush_secs(raw: Option<&str>) -> u64 {
93 raw.and_then(|v| v.trim().parse::<u64>().ok())
94 .filter(|&s| s > 0)
95 .unwrap_or(ABDICATE_FLUSH_SECS)
96}
97
98mod sso_plane;
100
101fn log_read_refusal(e: gunnar_core::txlog::Error) -> ControlError {
106 match e {
107 gunnar_core::txlog::Error::Released(_) => ControlError::Unavailable(format!(
108 "{} this box's data set is being grown (unmounted): its transaction log is back after the remount — ask again",
109 gunnar_grpc::PRIMARY_GROWING
110 )),
111 gunnar_core::txlog::Error::NotRetained { seq, first } => ControlError::FailedPrecondition(format!(
112 "seq {seq} is no longer retained; the oldest is {first} — a snapshot is needed, not a range"
113 )),
114 gunnar_core::txlog::Error::BeyondHead { seq, head } => ControlError::InvalidArgument(format!("seq {seq} is beyond the head {head}")),
115 other => ControlError::Internal(format!("transaction log: {other}")),
116 }
117}
118
119#[allow(clippy::type_complexity)] struct ChunkOut<'s> {
131 sink: &'s mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
132 open: Option<OutEntry>,
133}
134
135struct OutEntry {
137 seq: u64,
138 kind: u32,
139 meta: Option<Vec<u8>>,
141 blob_len: u64,
142 at: u64,
144 crc: Option<u32>,
146 ended: bool,
148}
149
150impl ChunkOut<'_> {
151 fn send(&mut self, blob: &[u8]) -> anyhow::Result<()> {
152 let Some(o) = self.open.as_mut() else {
153 anyhow::bail!("a piece outside an entry");
154 };
155 let offset = o.at;
156 o.at += blob.len() as u64;
157 anyhow::ensure!(
158 o.at <= o.blob_len,
159 "seq {}: more than its {} blob bytes",
160 o.seq,
161 o.blob_len
162 );
163 let last = o.at == o.blob_len;
164 o.ended |= last;
165 let chunk = gunnar_proto::LogChunk {
166 seq: o.seq,
167 kind: o.kind,
168 meta: o.meta.take().unwrap_or_default(),
169 blob_len: o.blob_len,
170 blob: blob.to_vec(),
171 last,
172 restore_head: 0,
173 restore_bytes: 0,
174 log_id: String::new(),
175 offset: Some(offset),
176 entry_crc: if last { o.crc } else { None },
177 snapshot_item: None,
178 snapshot_id: String::new(),
179 };
180 (self.sink)(chunk).map_err(anyhow::Error::from)
181 }
182}
183
184impl crate::twin::FrameSink for ChunkOut<'_> {
185 fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
186 self.begin_at(seq, kind, meta, blob_len, 0)
187 }
188 fn begin_at(
189 &mut self,
190 seq: u64,
191 kind: u16,
192 meta: &[u8],
193 blob_len: u64,
194 offset: u64,
195 ) -> anyhow::Result<()> {
196 anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
197 anyhow::ensure!(
198 offset <= blob_len,
199 "seq {seq}: from byte {offset} of a {blob_len}-byte blob"
200 );
201 self.open = Some(OutEntry {
202 seq,
203 kind: u32::from(kind),
204 meta: Some(meta.to_vec()),
205 blob_len,
206 at: offset,
207 crc: None,
208 ended: false,
209 });
210 Ok(())
211 }
212 fn checked(&mut self, payload_crc: u32) -> anyhow::Result<()> {
213 if let Some(o) = self.open.as_mut() {
214 o.crc = Some(payload_crc);
215 }
216 Ok(())
217 }
218 fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
219 self.send(bytes)
220 }
221 fn end(&mut self) -> anyhow::Result<()> {
222 if matches!(&self.open, Some(o) if !o.ended) {
225 self.send(&[])?;
226 }
227 let Some(o) = self.open.take() else {
228 anyhow::bail!("an entry ended that never began");
229 };
230 anyhow::ensure!(
231 o.at == o.blob_len && o.ended,
232 "seq {}: {} of {} blob bytes went out",
233 o.seq,
234 o.at,
235 o.blob_len
236 );
237 Ok(())
238 }
239}
240
241impl CoreControlPlane {
242 fn readable_log(&self) -> ControlResult<&gunnar_core::txlog::Log> {
248 if crate::twin::maintenance::is_held(self.stores.root()) {
249 return Err(ControlError::Unavailable(
250 "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(),
251 ));
252 }
253 if !self.stores.is_twin() {
254 if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
255 return Err(ControlError::FailedPrecondition(format!(
256 "{why}. Nothing is read from it"
257 )));
258 }
259 }
260 let log: &gunnar_core::txlog::Log = self.stores.txlog();
261 if log.is_released() {
262 return Err(log_read_refusal(gunnar_core::txlog::Error::Released(
263 log.dir().to_path_buf(),
264 )));
265 }
266 if self.stores.is_twin() {
269 if let Some(why) = crate::twin::snapshot::not_whole(self.stores.root()) {
270 return Err(ControlError::FailedPrecondition(format!(
271 "{why}. Nothing is read from it"
272 )));
273 }
274 }
275 Ok(log)
276 }
277}
278
279pub struct CoreControlPlane {
281 stores: Arc<ServerStores>,
282 grow: Option<Arc<crate::grow::GrowMachine>>,
286 twin_role: bool,
288 git_listener: Option<crate::readiness::GitListener>,
294 admission: Option<Arc<gunnar_ssh::Admission>>,
298 replication: crate::replication_view::ReplicationView,
311 #[cfg(feature = "replication")]
328 replication_slot: Option<Arc<crate::mirror::ReplicationSlot>>,
329 #[cfg(feature = "webhooks")]
333 webhook_slot: Option<Arc<crate::webhook::WebhookSlot>>,
334 #[cfg(feature = "metered")]
339 entitlement: Option<Arc<crate::entitlement::EntitlementSlot>>,
340 headroom: Option<Arc<gunnar_ssh::DiskHeadroomGate>>,
344 max_object_bytes: u64,
348 gc: crate::gc_queue::GcQueue,
357 purge: crate::purge_queue::PurgeQueue,
365 mirror: crate::mirror_queue::MirrorQueue,
373 sso_import: crate::sso_import_queue::SsoImportQueue,
377 bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
391 restore_keys: Vec<String>,
396}
397
398impl CoreControlPlane {
399 pub fn new(stores: Arc<ServerStores>) -> Self {
401 let for_gc = Arc::clone(&stores);
402 let bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>> = Arc::default();
403 CoreControlPlane {
404 stores,
405 grow: None,
406 twin_role: false,
407 git_listener: None,
408 admission: None,
409 replication: crate::replication_view::ReplicationView::new(),
410 #[cfg(feature = "replication")]
411 replication_slot: None,
412 #[cfg(feature = "webhooks")]
413 webhook_slot: None,
414 #[cfg(feature = "metered")]
415 entitlement: None,
416 headroom: None,
417 max_object_bytes: DEFAULT_MAX_OBJECT_BYTES,
418 gc: Self::gc_queue(Arc::clone(&for_gc)),
419 purge: Self::purge_queue(Arc::clone(&for_gc)),
420 mirror: Self::mirror_queue(Arc::clone(&for_gc), Arc::clone(&bus)),
421 sso_import: crate::sso_import_queue::SsoImportQueue::new(
422 crate::sso_import_queue::SsoImport,
423 ),
424 bus,
425 restore_keys: Vec::new(),
426 }
427 }
428
429 pub fn with_restore_keys(mut self, keys: Vec<String>) -> Self {
433 self.restore_keys = keys;
434 self
435 }
436
437 pub fn with_bus(self, bus: Arc<gunnar_grpc::EventBus>) -> Self {
445 let _ = self.bus.set(bus);
449 self
450 }
451
452 pub fn with_twin_role(mut self, twin_role: bool) -> Self {
457 self.twin_role = twin_role;
458 self
459 }
460
461 pub fn with_git_listener(mut self, listener: crate::readiness::GitListener) -> Self {
469 self.git_listener = Some(listener);
470 self
471 }
472
473 pub fn release_stale_purge_leases(stores: &ServerStores) -> u64 {
494 let gates = match stores.auth().gates() {
495 Ok(gates) => gates,
496 Err(err) => {
497 tracing::error!(
498 target: "gunnar::audit",
499 error = %err,
500 "the account gates could not be read at start-up; any purge lease left by a \
501 crash will stay in force until it expires"
502 );
503 return 0;
504 }
505 };
506 let mut released = 0;
507 for gate in gates {
508 if gate.reason != gunnar_core::GateReason::Purge {
509 continue;
510 }
511 match stores.auth().lift_gate_held_by(&gate.account, &gate.holder) {
512 Ok(Some(_)) => {
513 released += 1;
514 tracing::warn!(
515 target: "gunnar::audit",
516 event = "account.stale_lease_released",
517 account = %gate.account,
518 job = %gate.holder,
519 "an account was still locked by a purge that did not finish; the lockout \
520 has been lifted and the account may still hold bytes the purge did not \
521 reclaim — `repo list-orphans` shows them"
522 );
523 }
524 Ok(None) => {}
525 Err(err) => tracing::error!(
526 target: "gunnar::audit",
527 account = %gate.account,
528 error = %err,
529 "a stale purge lease could not be lifted at start-up"
530 ),
531 }
532 }
533 released
534 }
535
536 fn purge_queue(stores: Arc<ServerStores>) -> crate::purge_queue::PurgeQueue {
553 let for_release = Arc::clone(&stores);
554 crate::purge_queue::PurgeQueue::new(
555 Arc::new(
556 move |order: &crate::purge_queue::PurgeOrder,
557 progress: &crate::purge_queue::Progress| {
558 let request = crate::account_purge::PurgeRequest {
559 account: order.account.clone(),
560 principal_id: order.principal_id.clone(),
561 reason: order.reason.clone(),
562 remove_principal: order.remove_principal,
563 expect_empty: order.expect_empty,
564 };
565 let auth = stores.auth();
566 let account = order.account.clone();
567 let job_id = order.job_id.clone();
568 let mut step = |step: crate::account_purge::Step| match step {
569 crate::account_purge::Step::Found(stores_found) => {
570 progress.found(stores_found)
571 }
572 crate::account_purge::Step::Reclaimed { bytes, .. } => {
573 progress.reclaimed(bytes);
574 let until = gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS;
579 if let Err(err) = auth.renew_gate(&account, &job_id, until) {
580 tracing::warn!(
581 target: "gunnar::audit",
582 account = %account,
583 job = %job_id,
584 error = %err,
585 "a purge could not renew its lockout lease; the account is \
586 still shut and the walk continues"
587 );
588 }
589 }
590 crate::account_purge::Step::Failed(note) => progress.failed(note),
591 crate::account_purge::Step::PrincipalRemoved => {
592 progress.principal_removed()
593 }
594 };
595 let report = crate::account_purge::purge_account(&stores, request, &mut step)
596 .map_err(|err| format!("{err:#}"))?;
597 if report.failures.is_empty() {
602 Ok(())
603 } else {
604 Err(format!(
605 "{} of {} stores could not be reclaimed and are still on the disk; \
606 `repo list-orphans` shows them",
607 report.failures.len(),
608 report.stores_found
609 ))
610 }
611 },
612 ),
613 Arc::new(move |account: &str, job_id: &str| {
614 for_release
615 .auth()
616 .lift_gate_held_by(account, job_id)
617 .map(|lifted| lifted.is_some())
618 .map_err(|err| format!("{err}"))
619 }),
620 )
621 }
622
623 fn gc_queue(stores: Arc<ServerStores>) -> crate::gc_queue::GcQueue {
633 crate::gc_queue::GcQueue::new(Arc::new(move |path: &str| {
634 let repo_ref = gunnar_core::resolve_repo_path(path)
635 .map_err(|err| format!("`{path}` is not a valid <namespace>/<name>: {err:#}"))?;
636 let repo = stores
637 .catalog()
638 .get(&repo_ref)
639 .map_err(|err| format!("reading the catalog row for {path}: {err:#}"))?
640 .ok_or_else(|| format!("{path} is no longer a live repository"))?;
641 let report = crate::admin::gc_one(&stores, &repo)
642 .map_err(|err| format!("maintaining {path}: {err:#}"))?;
643 Ok(crate::admin::reclaimed_bytes(&report))
644 }))
645 }
646
647 fn mirror_queue(
662 stores: Arc<ServerStores>,
663 bus: Arc<std::sync::OnceLock<Arc<gunnar_grpc::EventBus>>>,
664 ) -> crate::mirror_queue::MirrorQueue {
665 crate::mirror_queue::MirrorQueue::new(Arc::new(
666 move |order: &crate::mirror_queue::MirrorOrder,
667 report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>| {
668 let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
669 format!(
670 "`{}` is not a valid <namespace>/<name>: {err:#}",
671 order.path
672 )
673 })?;
674 let first = first_sync_of(
681 stores
682 .mirrors()
683 .mark_started(&repo_ref, gunnar_grpc::bus::now_unix_ms()),
684 &order.path,
685 )?;
686 let outcome = read_credential(&stores, &repo_ref).and_then(|credential| {
687 crate::mirror_fetch::sync_once(&stores, order, &credential, report)
688 });
689 audit_mirror_outcome(
690 &stores,
691 &repo_ref,
692 order,
693 first,
694 outcome.as_ref().err().map(String::as_str),
695 report,
696 );
697
698 let error = outcome.as_ref().err().map(String::as_str);
704 if let Err(err) =
705 stores
706 .mirrors()
707 .record_sync(&repo_ref, gunnar_grpc::bus::now_unix_ms(), error)
708 {
709 tracing::warn!(
710 target: "gunnar::git",
711 repo = %order.path,
712 %err,
713 "a mirror sync finished and its registration could not be updated; the \
714 fetch itself is unaffected"
715 );
716 }
717
718 if outcome.is_ok() {
722 if let Some(bus) = bus.get() {
723 let job = report.snapshot();
724 let updates = job
725 .map(|job| job.report.updates)
726 .unwrap_or_default()
727 .into_iter()
728 .map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
729 ref_name,
730 old_oid,
731 new_oid,
732 })
733 .collect::<Vec<_>>();
734 bus.publish(gunnar_proto::event::Payload::MirrorSynced(
735 gunnar_proto::MirrorSynced {
736 repo: Some(RepoPath::new(repo_ref.namespace(), repo_ref.name())),
737 remote_url: order.upstream.clone(),
738 refs_changed: updates.len() as u32,
739 updates,
740 },
741 ));
742 }
743 }
744 outcome
745 },
746 ))
747 }
748
749 pub fn resume_interrupted_mirrors(&self) -> usize {
764 let settled = match self
765 .stores
766 .mirrors()
767 .settle_interrupted(gunnar_grpc::bus::now_unix_ms())
768 {
769 Ok(settled) => settled,
770 Err(err) => {
771 tracing::warn!(target: "gunnar::git", %err, "the mirror registrations could not be read to resume interrupted syncs");
772 return 0;
773 }
774 };
775 let mut resumed = 0;
776 for record in settled {
777 let Ok(repo_ref) = gunnar_core::resolve_repo_path(&record.repo) else {
778 continue;
779 };
780 let again = record.interrupted <= gunnar_core::MAX_RESUMES;
781 audit_line(
782 &self.stores,
783 &repo_ref,
784 if again {
785 "mirror.sync.interrupted"
786 } else {
787 "mirror.sync.failed"
788 },
789 &record.last_error,
790 true,
791 );
792 tracing::warn!(
793 target: "gunnar::git",
794 event = "mirror.sync.interrupted",
795 repo = %record.repo,
796 interrupted = record.interrupted,
797 resumed = again,
798 "a mirror sync was in flight when the server stopped"
799 );
800 if again {
801 self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
802 path: repo_ref.key(),
803 upstream: record.remote_url.clone(),
804 prune: record.prune,
805 ca: mirror_ca_of(&record),
806 });
807 resumed += 1;
808 }
809 }
810 resumed
811 }
812
813 fn gate_for_purge(
830 &self,
831 account: &str,
832 job_id: &str,
833 options: &gunnar_grpc::seam::PurgeAccountOptions,
834 ) -> ControlResult<()> {
835 if self
836 .stores
837 .auth()
838 .gate_covering(account)
839 .map_err(internal)?
840 .is_some()
841 {
842 return Ok(());
843 }
844 let lease = gunnar_core::AccountGate::purge_lease(
845 account,
846 job_id,
847 gunnar_core::now_ms() + gunnar_core::PURGE_LEASE_MS,
848 )
849 .by(&options.principal_id)
850 .noted(format!(
851 "this account is being permanently deleted: {}",
852 options.reason
853 ));
854 self.stores.auth().gate(lease).map_err(gate_error)?;
855 Ok(())
856 }
857
858 #[must_use]
862 pub fn with_headroom(mut self, gate: Arc<gunnar_ssh::DiskHeadroomGate>) -> Self {
863 self.headroom = Some(gate);
864 self
865 }
866
867 pub fn with_grow(mut self, grow: Arc<crate::grow::GrowMachine>) -> Self {
870 self.grow = Some(grow);
871 self
872 }
873
874 pub fn with_admission(mut self, admission: Arc<gunnar_ssh::Admission>) -> Self {
877 self.admission = Some(admission);
878 self
879 }
880
881 fn grow_machine(&self) -> ControlResult<&Arc<crate::grow::GrowMachine>> {
883 self.grow.as_ref().ok_or(ControlError::Unimplemented(
884 "this control plane was built with no runtime growth wired in, so it cannot grow the data set",
885 ))
886 }
887
888 fn verify_membership_event(
898 &self,
899 signed: gunnar_proto::SignedMembershipEvent,
900 group_id: &str,
901 member_id: &str,
902 action: gunnar_core::MembershipAction,
903 ) -> ControlResult<gunnar_core::SignedMembershipEvent> {
904 let signed = from_proto_membership(signed)?;
905 if signed.event.group != group_id || signed.event.member != member_id {
906 return Err(ControlError::InvalidArgument(format!(
907 "the signed event is about {:?}/{:?} but the change is {:?}/{:?}",
908 signed.event.group, signed.event.member, group_id, member_id
909 )));
910 }
911 if signed.event.action != action {
912 return Err(ControlError::InvalidArgument(
913 "the signed event's action does not match the change (add vs remove)".to_owned(),
914 ));
915 }
916 let key = self
917 .stores
918 .auth()
919 .signing_keys()
920 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
921 .into_iter()
922 .find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
923 .ok_or_else(|| {
924 ControlError::InvalidArgument(format!(
925 "{} is not registered on this server, so nobody reading this event later \
926 could check its signature",
927 signed.key_fingerprint
928 ))
929 })?;
930 if key.principal_id != signed.event.actor {
931 return Err(ControlError::InvalidArgument(format!(
932 "this membership change is attributed to {:?} and is signed by {}, which is \
933 {:?}'s key; a membership change is the actor's own word, so it must carry the \
934 actor's own signature",
935 signed.event.actor, signed.key_fingerprint, key.principal_id
936 )));
937 }
938 let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
939 .map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
940 gunnar_core::membership::verify_event(&signed, &public)
941 .map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
942 Ok(signed)
943 }
944
945 fn append_membership_event(
950 &self,
951 signed: &gunnar_core::SignedMembershipEvent,
952 ) -> ControlResult<u64> {
953 let log = self.stores.membership_log();
954 let mut line = serde_json::to_string(signed)
955 .map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
956 line.push('\n');
957 let _framed =
958 gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
959 .map_err(|err| {
960 ControlError::Unavailable(format!(
961 "framing the line in the transaction log: {err}"
962 ))
963 })?;
964 use std::io::Write as _;
965 let mut file = std::fs::OpenOptions::new()
966 .create(true)
967 .append(true)
968 .open(log.records_path())
969 .map_err(|err| {
970 ControlError::Unavailable(format!("opening the membership log: {err}"))
971 })?;
972 file.write_all(line.as_bytes()).map_err(|err| {
973 ControlError::Unavailable(format!("appending to the membership log: {err}"))
974 })?;
975 let count = log
976 .lines_of()
977 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
978 .len() as u64;
979 Ok(count.saturating_sub(1))
980 }
981
982 fn membership_log_events(&self) -> ControlResult<Vec<gunnar_core::SignedMembershipEvent>> {
984 self.stores
985 .membership_log()
986 .lines_of()
987 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
988 .iter()
989 .map(|line| {
990 serde_json::from_str(line).map_err(|err| {
991 ControlError::Internal(format!("decoding a membership event: {err}"))
992 })
993 })
994 .collect()
995 }
996
997 #[cfg(feature = "replication")]
1005 pub fn with_replication(mut self, slot: Arc<crate::mirror::ReplicationSlot>) -> Self {
1006 self.replication_slot = Some(slot);
1007 self
1008 }
1009
1010 #[cfg(feature = "replication")]
1016 fn replication_slot(&self) -> ControlResult<&Arc<crate::mirror::ReplicationSlot>> {
1017 self.replication_slot.as_ref().ok_or(ControlError::Unimplemented(
1018 "this control plane was built with no replication slot, so it cannot configure a standby",
1019 ))
1020 }
1021
1022 #[cfg(feature = "webhooks")]
1026 pub fn with_webhooks(mut self, slot: Arc<crate::webhook::WebhookSlot>) -> Self {
1027 self.webhook_slot = Some(slot);
1028 self
1029 }
1030
1031 #[cfg(feature = "metered")]
1033 pub fn with_entitlement(mut self, slot: Arc<crate::entitlement::EntitlementSlot>) -> Self {
1034 self.entitlement = Some(slot);
1035 self
1036 }
1037
1038 #[cfg(feature = "metered")]
1040 fn entitlement(&self) -> ControlResult<&Arc<crate::entitlement::EntitlementSlot>> {
1041 self.entitlement.as_ref().ok_or_else(|| {
1042 ControlError::FailedPrecondition(
1043 "this metered server was started without `--monetize-pubkey`, so it holds no \
1044 key to verify an entitlement fact against and refuses every one; every \
1045 namespace is Free"
1046 .into(),
1047 )
1048 })
1049 }
1050
1051 #[cfg(feature = "webhooks")]
1053 fn webhook_slot(&self) -> ControlResult<&Arc<crate::webhook::WebhookSlot>> {
1054 self.webhook_slot.as_ref().ok_or(ControlError::Unimplemented(
1055 "this control plane was built with no webhook slot, so it cannot configure an outbound endpoint",
1056 ))
1057 }
1058
1059 #[cfg(feature = "replication")]
1065 fn replication_snapshot(
1073 &self,
1074 include_refs: bool,
1075 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
1076 let slot = self.replication_slot()?;
1077 let config = slot
1078 .config()
1079 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
1080 let mut out = gunnar_proto::ReplicationStatusResponse {
1081 targets: config
1082 .targets
1083 .iter()
1084 .map(crate::replication_wire::target_to_proto)
1085 .collect(),
1086 role: config.role.to_proto() as i32,
1087 ..Default::default()
1088 };
1089 if let Some(replicator) = slot.replicator() {
1090 crate::replication_wire::running_to_proto(&replicator, include_refs, &mut out);
1091 }
1092 Ok(out)
1093 }
1094
1095 pub fn replication_view(&self) -> crate::replication_view::ReplicationView {
1103 self.replication.clone()
1104 }
1105
1106 fn forge_events(&self) -> ControlResult<Vec<gunnar_core::SignedEvent>> {
1112 self.stores
1113 .forge_log()
1114 .lines_of()
1115 .map_err(|err| ControlError::Unavailable(format!("reading the forge log: {err:#}")))?
1116 .iter()
1117 .map(|line| {
1118 serde_json::from_str(line).map_err(|err| {
1119 ControlError::Internal(format!("a forge event will not decode: {err}"))
1120 })
1121 })
1122 .collect()
1123 }
1124
1125 fn forge_subject_open(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<()> {
1128 let (namespace, name) = signed.event.repo.split_once('/').ok_or_else(|| {
1134 ControlError::InvalidArgument(format!(
1135 "`repo` must be `<namespace>/<repo>`, got {:?}",
1136 signed.event.repo
1137 ))
1138 })?;
1139 let path = RepoPath::new(namespace, name);
1140 match &signed.event.event {
1141 gunnar_core::Event::IssueOpened { .. } => {}
1144 gunnar_core::Event::IssueCommented { .. } | gunnar_core::Event::IssueClosed { .. } => {
1148 let repo_events: Vec<gunnar_core::ForgeEvent> = self
1149 .forge_events()?
1150 .into_iter()
1151 .filter(|e| e.event.repo == signed.event.repo)
1152 .map(|e| e.event)
1153 .collect();
1154 match gunnar_core::forge::fold_issue(signed.event.pull, &repo_events) {
1155 None => {
1156 return Err(ControlError::FailedPrecondition(format!(
1157 "no issue #{} in {}: nobody has opened one, so an event on it is \
1158 evidence about nothing",
1159 signed.event.pull, signed.event.repo
1160 )));
1161 }
1162 Some(issue) if !issue.open => {
1163 return Err(ControlError::FailedPrecondition(format!(
1164 "issue #{} in {} is closed; an event on it cannot change what has \
1165 already happened",
1166 signed.event.pull, signed.event.repo
1167 )));
1168 }
1169 Some(_) => {}
1170 }
1171 }
1172 gunnar_core::Event::Commented { .. }
1175 | gunnar_core::Event::Approved { .. }
1176 | gunnar_core::Event::Rejected { .. } => {
1177 let record = self.get_pull(&path, signed.event.pull)?;
1178 if record.state != gunnar_proto::PullRequestState::Open as i32 {
1179 return Err(ControlError::FailedPrecondition(format!(
1180 "pull request #{} in {} is not open; a review event cannot change what \
1181 has already happened",
1182 signed.event.pull, signed.event.repo
1183 )));
1184 }
1185 }
1186 }
1187 Ok(())
1188 }
1189
1190 fn forge_write(&self, signed: &gunnar_core::SignedEvent) -> ControlResult<u64> {
1192 let log = self.stores.forge_log();
1193 let mut line = serde_json::to_string(signed)
1194 .map_err(|err| ControlError::Internal(format!("encoding the event: {err}")))?;
1195 line.push('\n');
1196 let _framed =
1197 gunnar_core::txlog::sink::append_line(log.records_path(), line.trim_end_matches('\n'))
1198 .map_err(|err| {
1199 ControlError::Unavailable(format!(
1200 "framing the line in the transaction log: {err}"
1201 ))
1202 })?;
1203 use std::io::Write as _;
1204 let mut file = std::fs::OpenOptions::new()
1205 .create(true)
1206 .append(true)
1207 .open(log.records_path())
1208 .map_err(|err| ControlError::Unavailable(format!("opening the forge log: {err}")))?;
1209 file.write_all(line.as_bytes()).map_err(|err| {
1210 ControlError::Unavailable(format!("appending to the forge log: {err}"))
1211 })?;
1212 let count = self
1215 .stores
1216 .forge_log()
1217 .lines_of()
1218 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
1219 .len() as u64;
1220 Ok(count.saturating_sub(1))
1221 }
1222
1223 fn review_state(
1239 &self,
1240 repo: &str,
1241 state: &gunnar_core::forge::Review,
1242 all: &[gunnar_core::SignedEvent],
1243 ) -> ControlResult<gunnar_proto::ReviewState> {
1244 let keys: Vec<(String, String, gunnar_core::ssh_key::PublicKey)> = self
1245 .stores
1246 .auth()
1247 .signing_keys()
1248 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
1249 .iter()
1250 .filter_map(|row| {
1251 gunnar_core::ssh_key::PublicKey::from_openssh(&row.openssh)
1252 .ok()
1253 .map(|key| {
1254 (
1255 row.fingerprint_sha256.clone(),
1256 row.principal_id.clone(),
1257 key,
1258 )
1259 })
1260 })
1261 .collect();
1262 let attest_key = self
1265 .stores
1266 .forge_attest_public()
1267 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
1268
1269 let approvals = state
1270 .approvals
1271 .iter()
1272 .map(|(actor, commit)| {
1273 let signed = all.iter().rev().find(|e| {
1274 e.event.repo == repo
1275 && e.event.pull == state.pull
1276 && e.event.actor == *actor
1277 && matches!(&e.event.event, gunnar_core::Event::Approved { commit: c } if c == commit)
1278 });
1279 let (verified, verdict) = match signed {
1280 None => (false, "NO EVENT".to_owned()),
1281 Some(signed) if signed.attested.is_some() => match &attest_key {
1286 None => (false, "UNKNOWN KEY".to_owned()),
1287 Some(key) => match gunnar_core::forge::verify_event(signed, key) {
1288 Ok(()) => (
1289 true,
1290 signed.attested.as_ref().map(|a| a.words()).unwrap_or_default(),
1291 ),
1292 Err(problem) => (false, format!("BAD ({problem})")),
1293 },
1294 },
1295 Some(signed) => match keys
1296 .iter()
1297 .find(|(fingerprint, _, _)| *fingerprint == signed.key_fingerprint)
1298 {
1299 None => (false, "UNKNOWN KEY".to_owned()),
1300 Some((_, owner, _)) if *owner != *actor => {
1306 (false, format!("KEY OF {owner}, NOT {actor}"))
1307 }
1308 Some((_, _, key)) => match gunnar_core::forge::verify_event(signed, key) {
1309 Ok(()) => (true, "verified".to_owned()),
1310 Err(problem) => (false, format!("BAD ({problem})")),
1311 },
1312 },
1313 };
1314 gunnar_proto::Approval {
1315 actor: actor.clone(),
1316 commit: commit.clone(),
1317 verified,
1318 verdict,
1319 }
1320 })
1321 .collect();
1322
1323 Ok(gunnar_proto::ReviewState {
1324 repo: repo.to_owned(),
1325 pull: state.pull,
1326 approvals,
1327 comments: state.comments as u64,
1328 })
1329 }
1330
1331 pub fn with_max_object_bytes(mut self, bytes: u64) -> Self {
1346 self.max_object_bytes = bytes;
1347 self
1348 }
1349
1350 fn pulls(&self) -> ControlResult<gunnar_core::Pulls> {
1355 gunnar_core::Pulls::new(Arc::clone(self.stores.control_store()))
1356 .map_err(|err| ControlError::Internal(format!("opening the pulls table: {err:#}")))
1357 }
1358
1359 fn repo_ref(path: &RepoPath) -> ControlResult<RepoRef> {
1360 path.validate()
1361 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
1362 RepoRef::new(&path.namespace, &path.repo)
1363 .map_err(|err| ControlError::InvalidArgument(err.to_string()))
1364 }
1365
1366 fn tree_of(
1377 &self,
1378 git: &dyn GitServe,
1379 hash: gix_hash::Kind,
1380 path: &RepoPath,
1381 spec: &str,
1382 ) -> ControlResult<gix_hash::ObjectId> {
1383 let id = match gix_hash::ObjectId::from_hex(spec.as_bytes()) {
1384 Ok(id) => {
1385 if id.kind() != hash {
1386 return Err(ControlError::InvalidArgument(format!(
1387 "{spec} is a {} oid; this repository is {hash}",
1388 id.kind()
1389 )));
1390 }
1391 id
1392 }
1393 Err(_) => {
1394 gunnar_store::refname::validate(spec)
1395 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
1396 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
1397 let resolved = if spec == gunnar_store::HEAD {
1398 snapshot.head(git).map_err(wire_error)?
1399 } else {
1400 snapshot.resolve(spec).map_err(wire_error)?
1401 };
1402 resolved.ok_or_else(|| ControlError::NotFound(format!("{spec} in {path}")))?
1403 }
1404 };
1405 let id = peel(git, hash, id)?;
1406 let (kind, bytes) = git
1407 .read(id.as_bytes())
1408 .map_err(store_error)?
1409 .ok_or_else(|| ControlError::NotFound(format!("object {id} in {path}")))?;
1410 match kind {
1411 ObjType::Tree => Ok(id),
1412 ObjType::Commit => object::commit(&bytes, hash)
1413 .map(|commit| commit.tree)
1414 .map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}"))),
1415 other => Err(ControlError::InvalidArgument(format!(
1418 "{spec} names a {other:?}; a diff needs a commit, a tag or a tree"
1419 ))),
1420 }
1421 }
1422
1423 fn policy_of(&self, repo: &gunnar_core::Repo) -> ControlResult<gunnar_policy::PolicyConfig> {
1433 let stored = self.stores.policies().get(&repo.path()).map_err(internal)?;
1434 match stored {
1435 None => Ok(gunnar_policy::PolicyConfig::default()),
1436 Some(document) => gunnar_policy::PolicyConfig::from_json(&document).map_err(|err| {
1437 ControlError::Internal(format!(
1438 "the stored policy for {} does not parse: {err}",
1439 repo.key()
1440 ))
1441 }),
1442 }
1443 }
1444
1445 fn store_policy(
1452 &self,
1453 repo: &gunnar_core::Repo,
1454 config: &gunnar_policy::PolicyConfig,
1455 ) -> ControlResult<()> {
1456 let policies = self.stores.policies();
1457 if *config == gunnar_policy::PolicyConfig::default() {
1458 policies.remove(&repo.path()).map_err(internal)?;
1459 return Ok(());
1460 }
1461 let json = config
1462 .to_json()
1463 .map_err(|err| ControlError::Internal(format!("rendering the policy: {err}")))?;
1464 policies.set(&repo.path(), &json).map_err(internal)
1465 }
1466
1467 fn open(&self, path: &RepoPath) -> ControlResult<(gunnar_core::Repo, Arc<OpenRepo>)> {
1468 let repo_ref = Self::repo_ref(path)?;
1469 let repo = self
1470 .stores
1471 .catalog()
1472 .get(&repo_ref)
1473 .map_err(internal)?
1474 .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))?;
1475 let store = self.stores.repo_store(&repo).map_err(|err| {
1476 ControlError::Internal(format!("opening the repository's stores: {err:#}"))
1477 })?;
1478 Ok((repo, store))
1479 }
1480
1481 fn mirror_record(&self, path: &RepoPath) -> ControlResult<gunnar_core::MirrorRecord> {
1487 let repo_ref = Self::repo_ref(path)?;
1488 let record = self.stores.mirrors().get(&repo_ref).map_err(internal)?;
1489 match record {
1490 Some(record)
1491 if record.creates_repo
1492 && self
1493 .stores
1494 .catalog()
1495 .get(&repo_ref)
1496 .map_err(internal)?
1497 .is_none() =>
1498 {
1499 Ok(record)
1500 }
1501 record => {
1502 self.open(path)?;
1503 record.ok_or_else(|| {
1504 ControlError::NotFound(format!("no mirror registered for {path}"))
1505 })
1506 }
1507 }
1508 }
1509}
1510
1511fn gc_job_out(path: &RepoPath, job: crate::gc_queue::GcJob) -> gunnar_grpc::seam::GcJob {
1519 gunnar_grpc::seam::GcJob {
1520 job_id: job.job_id,
1521 path: path.clone(),
1522 state: match job.state {
1523 crate::gc_queue::GcState::Queued => gunnar_grpc::seam::GcState::Queued,
1524 crate::gc_queue::GcState::Running => gunnar_grpc::seam::GcState::Running,
1525 crate::gc_queue::GcState::Done => gunnar_grpc::seam::GcState::Done,
1526 crate::gc_queue::GcState::Failed => gunnar_grpc::seam::GcState::Failed,
1527 },
1528 queue_position: job.queue_position,
1529 bytes_reclaimed: job.report.bytes_reclaimed,
1533 error: job.error,
1534 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
1535 finished_at_unix_ms: job.finished_at_unix_ms,
1536 }
1537}
1538
1539fn gate_out(gate: gunnar_core::AccountGate) -> gunnar_grpc::seam::AccountGate {
1541 gunnar_grpc::seam::AccountGate {
1542 account: gate.account,
1543 reason: match gate.reason {
1544 gunnar_core::GateReason::Administrative => {
1545 gunnar_grpc::seam::GateReason::Administrative
1546 }
1547 gunnar_core::GateReason::Purge => gunnar_grpc::seam::GateReason::Purge,
1548 },
1549 holder: gate.holder,
1550 actor: gate.actor,
1551 note: gate.note,
1552 at_unix_ms: gate.at,
1553 expires_unix_ms: gate.expires.unwrap_or(0),
1557 }
1558}
1559
1560fn purge_job_out(job: crate::purge_queue::PurgeJob) -> gunnar_grpc::seam::PurgeJob {
1562 let report = job.report;
1566 gunnar_grpc::seam::PurgeJob {
1567 job_id: job.job_id,
1568 account: job.key,
1570 state: match job.state {
1571 crate::purge_queue::PurgeState::Queued => gunnar_grpc::seam::PurgeState::Queued,
1572 crate::purge_queue::PurgeState::Running => gunnar_grpc::seam::PurgeState::Running,
1573 crate::purge_queue::PurgeState::Done => gunnar_grpc::seam::PurgeState::Done,
1574 crate::purge_queue::PurgeState::Failed => gunnar_grpc::seam::PurgeState::Failed,
1575 },
1576 queue_position: job.queue_position,
1577 stores_found: report.stores_found,
1578 stores_purged: report.stores_purged,
1579 bytes_reclaimed: report.bytes_reclaimed,
1580 principal_removed: report.principal_removed,
1581 gate_released: report.gate_released,
1582 error: job.error,
1583 failures: report.failures,
1584 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
1585 started_at_unix_ms: job.started_at_unix_ms,
1586 finished_at_unix_ms: job.finished_at_unix_ms,
1587 }
1588}
1589
1590fn gate_error(err: gunnar_core::Error) -> ControlError {
1597 match err {
1598 gunnar_core::Error::ReservedPrincipalId(_) | gunnar_core::Error::InvalidPath { .. } => {
1599 ControlError::InvalidArgument(format!("{err}"))
1600 }
1601 other => internal(other),
1602 }
1603}
1604
1605fn day_activity(day: i64, row: &gunnar_core::FlowDay) -> gunnar_proto::DayActivity {
1613 gunnar_proto::DayActivity {
1614 day_unix_ms: gunnar_core::midnight_of(day),
1615 bytes_in: row.bytes_in,
1616 bytes_out: row.bytes_out,
1617 pushes: row.pushes,
1618 fetches: row.fetches,
1619 lfs_bytes_in: row.lfs_bytes_in,
1620 lfs_bytes_out: row.lfs_bytes_out,
1621 refused: row.refused(),
1622 refused_by_reason: gunnar_core::Refusal::ALL
1623 .into_iter()
1624 .filter(|reason| row.refused_for(*reason) > 0)
1625 .map(|reason| (reason.name().to_owned(), row.refused_for(reason)))
1626 .collect(),
1627 auth_failures: row.auth_failures,
1628 last_push_unix_ms: row.last_push_unix_ms,
1629 last_fetch_unix_ms: row.last_fetch_unix_ms,
1630 distinct_principals: row.distinct_principals(),
1631 distinct_principals_capped: row.principals_capped,
1632 }
1633}
1634
1635pub(crate) fn create_error(err: &anyhow::Error, path: &RepoPath) -> ControlError {
1640 match err.downcast_ref::<gunnar_core::Error>() {
1641 Some(gunnar_core::Error::RepoExists(_)) => {
1642 ControlError::AlreadyExists(format!("repository {path}"))
1643 }
1644 Some(gunnar_core::Error::NoSuchNamespace(ns)) => {
1645 ControlError::NotFound(format!("namespace {ns}"))
1646 }
1647 Some(e @ (gunnar_core::Error::LogFollows | gunnar_core::Error::OffStripe { .. })) => {
1648 ControlError::Unavailable(e.to_string())
1649 }
1650 Some(_) => ControlError::Internal(format!("{err:#}")),
1651 None => ControlError::FailedPrecondition(format!("{err:#}")),
1652 }
1653}
1654
1655pub(crate) fn first_sync_of<E: std::fmt::Display>(
1658 mark: Result<Option<gunnar_core::MirrorRecord>, E>,
1659 path: &str,
1660) -> Result<bool, String> {
1661 match mark {
1662 Ok(record) => {
1663 Ok(record.is_some_and(|record| record.creates_repo || record.sync_count == 0))
1664 }
1665 Err(err) => {
1666 tracing::error!(target: "gunnar::git", "ALERT mirror.mark-unwritten {path}: the in-flight mark was not written ({err}); the sync is not started");
1669 Err(format!("{path}: the mirror's in-flight mark was not written ({err}); nothing was fetched, ask again"))
1670 }
1671 }
1672}
1673
1674#[cfg(test)]
1675mod mirror_mark_tests {
1676 use super::*;
1677
1678 #[test]
1682 fn a_mark_that_is_not_written_fails_the_sync_by_name() {
1683 let why = first_sync_of::<String>(Err("the mirrors store is full".into()), "acme/linux")
1684 .expect_err("refused");
1685 assert!(
1686 why.contains("acme/linux")
1687 && why.contains("the mirrors store is full")
1688 && why.contains("in-flight mark"),
1689 "{why}"
1690 );
1691 assert_eq!(first_sync_of::<String>(Ok(None), "acme/linux"), Ok(false));
1692 }
1693}
1694
1695#[cfg(test)]
1696mod create_error_tests {
1697 use super::*;
1698
1699 #[test]
1702 fn a_create_on_a_restoring_box_is_unavailable() {
1703 let path = RepoPath::new("acme", "site");
1704 let restoring = anyhow::Error::new(gunnar_core::Error::LogFollows);
1705 assert!(
1706 matches!(create_error(&restoring, &path), ControlError::Unavailable(m) if m.contains("being restored"))
1707 );
1708 let exists = anyhow::Error::new(gunnar_core::Error::RepoExists(
1709 gunnar_core::RepoRef::new("acme", "site").unwrap(),
1710 ));
1711 assert!(matches!(
1712 create_error(&exists, &path),
1713 ControlError::AlreadyExists(_)
1714 ));
1715 assert!(matches!(
1716 create_error(&anyhow::anyhow!("a store already stands there"), &path),
1717 ControlError::FailedPrecondition(_)
1718 ));
1719 }
1720}
1721
1722fn internal(err: gunnar_core::Error) -> ControlError {
1751 use gunnar_core::Error as E;
1752 match err {
1753 e @ (E::InvalidPath { .. }
1755 | E::PathEscape { .. }
1756 | E::ReservedPrincipalId(_)
1757 | E::BadBranchName(_)
1758 | E::BadFingerprint(_)
1759 | E::Key(_)
1760 | E::InvalidInput { .. }) => ControlError::InvalidArgument(e.to_string()),
1761 e @ (E::NoSuchRepo(_)
1763 | E::NoSuchNamespace(_)
1764 | E::NoSuchPrincipal(_)
1765 | E::NoSuchCredential(_)) => ControlError::NotFound(e.to_string()),
1766 e @ (E::RepoExists(_)
1768 | E::NamespaceExists(_)
1769 | E::PrincipalExists(_)
1770 | E::NameExists { .. }
1771 | E::KeyExists { .. }
1772 | E::CredentialExists { .. }
1773 | E::InviteExists(_)
1774 | E::TokenExists { .. }
1775 | E::AlreadyBootstrapped { .. }
1776 | E::AlreadyProvisioned { .. }) => ControlError::AlreadyExists(e.to_string()),
1777 e @ (E::NamespaceNotEmpty { .. }
1781 | E::Contended { .. }
1782 | E::NotAPerson { .. }
1783 | E::NotAMachine { .. }) => ControlError::FailedPrecondition(e.to_string()),
1784 e @ (E::Denied { .. } | E::CounterRollback { .. } | E::InviteInvalid) => {
1785 ControlError::PermissionDenied(e.to_string())
1786 }
1787 e @ (E::Store(_) | E::Json(_)) => ControlError::Unavailable(e.to_string()),
1791 e @ E::OffStripe { .. } => ControlError::Unavailable(e.to_string()),
1796 e @ E::LogFollows => ControlError::Unavailable(e.to_string()),
1798 e => ControlError::Internal(e.to_string()),
1805 }
1806}
1807
1808#[cfg(feature = "engine-znippy")]
1825fn archive_error(err: anyhow::Error) -> ControlError {
1826 let io = err
1827 .chain()
1828 .any(|cause| cause.downcast_ref::<std::io::Error>().is_some());
1829 let message = format!("{err:#}");
1830 if io {
1831 ControlError::Unavailable(message)
1832 } else {
1833 ControlError::FailedPrecondition(message)
1834 }
1835}
1836
1837fn store_error(err: anyhow::Error) -> ControlError {
1859 ControlError::Internal(format!("{err:#}"))
1860}
1861
1862fn wire_error(err: gunnar_wire::Error) -> ControlError {
1870 ControlError::Internal(format!("{err:#}"))
1871}
1872
1873pub(crate) fn hash_kind(kind: HashKind) -> gunnar_proto::HashKind {
1879 match kind {
1880 HashKind::Sha1 => gunnar_proto::HashKind::Sha1,
1881 HashKind::Sha256 => gunnar_proto::HashKind::Sha256,
1882 }
1883}
1884
1885fn visibility_in(v: gunnar_proto::Visibility) -> ControlResult<Visibility> {
1899 match v {
1900 gunnar_proto::Visibility::Public => Ok(Visibility::Public),
1901 gunnar_proto::Visibility::Internal => Err(ControlError::InvalidArgument(
1902 "visibility \"internal\" has no implementation: since 2026-08-03, \"public\" means \
1903 readable with no key at all, so \"internal\" (any authenticated principal) \
1904 no longer aliases it and gunnar refuses to guess which you meant. Choose \
1905 \"private\" or \"public\""
1906 .into(),
1907 )),
1908 gunnar_proto::Visibility::Private | gunnar_proto::Visibility::Unspecified => {
1909 Ok(Visibility::Private)
1910 }
1911 }
1912}
1913
1914fn pull_out(path: &RepoPath, record: gunnar_core::PullRecord) -> PullRequest {
1916 PullRequest {
1917 repo: Some(path.clone()),
1918 number: record.number,
1919 title: record.title,
1920 description: record.description,
1921 source_ref: record.source_ref,
1922 target_ref: record.target_ref,
1923 state: match record.state {
1924 gunnar_core::PullState::Open => PullRequestState::Open,
1925 gunnar_core::PullState::Merged => PullRequestState::Merged,
1926 gunnar_core::PullState::Closed => PullRequestState::Closed,
1927 } as i32,
1928 author: record.author,
1929 created_at_unix_ms: record.created_ms,
1930 updated_at_unix_ms: record.updated_ms,
1931 source_oid: record.source_oid,
1932 merge_oid: record.merge_oid.unwrap_or_default(),
1933 }
1934}
1935
1936fn visibility_out(v: Visibility) -> gunnar_proto::Visibility {
1937 match v {
1938 Visibility::Private => gunnar_proto::Visibility::Private,
1939 Visibility::Public => gunnar_proto::Visibility::Public,
1940 }
1941}
1942
1943fn principal_kind_in(kind: gunnar_proto::PrincipalKind) -> PrincipalKind {
1944 match kind {
1945 gunnar_proto::PrincipalKind::Machine => PrincipalKind::Machine,
1946 gunnar_proto::PrincipalKind::Deploy => PrincipalKind::Deploy,
1947 gunnar_proto::PrincipalKind::Group => PrincipalKind::Group,
1948 gunnar_proto::PrincipalKind::User | gunnar_proto::PrincipalKind::Unspecified => {
1949 PrincipalKind::User
1950 }
1951 }
1952}
1953
1954impl CoreControlPlane {
1955 fn owner_kind(&self, name: &str) -> i32 {
1966 let auth = self.stores.auth();
1967 match auth.principal(name) {
1968 Ok(Some(p)) if p.kind == PrincipalKind::Group => {
1969 let members = auth.members_of(name).unwrap_or_default();
1970 let machines = !members.is_empty() && members.iter().all(|m| matches!(auth.principal(m), Ok(Some(q)) if q.kind == PrincipalKind::Machine));
1971 if machines {
1972 gunnar_proto::PrincipalKind::Machine as i32
1973 } else {
1974 principal_kind_out(p.kind) as i32
1975 }
1976 }
1977 Ok(Some(p)) => principal_kind_out(p.kind) as i32,
1978 _ => gunnar_proto::PrincipalKind::Unspecified as i32,
1979 }
1980 }
1981
1982 fn namespace_out(&self, namespace: gunnar_core::Namespace, employees: u32) -> Namespace {
1987 Namespace {
1988 owner_kind: self.owner_kind(&namespace.name),
1989 owner_server_grant: self
1992 .stores
1993 .auth()
1994 .holds_server_grant(&namespace.name)
1995 .unwrap_or(true),
1996 name: namespace.name,
1997 description: String::new(),
1998 created_at_unix_ms: 0,
1999 public: namespace.public,
2000 events_public: namespace.events_public,
2001 employees,
2002 }
2003 }
2004
2005 fn employee_count(&self, account: &str) -> ControlResult<u32> {
2007 Ok(self
2008 .stores
2009 .auth()
2010 .employees_of(account)
2011 .map_err(internal)?
2012 .len() as u32)
2013 }
2014}
2015
2016fn principal_kind_out(kind: PrincipalKind) -> gunnar_proto::PrincipalKind {
2017 match kind {
2018 PrincipalKind::User => gunnar_proto::PrincipalKind::User,
2019 PrincipalKind::Machine => gunnar_proto::PrincipalKind::Machine,
2020 PrincipalKind::Deploy => gunnar_proto::PrincipalKind::Deploy,
2021 PrincipalKind::Group => gunnar_proto::PrincipalKind::Group,
2022 PrincipalKind::Anonymous => gunnar_proto::PrincipalKind::Unspecified,
2031 }
2032}
2033
2034fn level_in(level: GrantLevel) -> ControlResult<Level> {
2035 match level {
2036 GrantLevel::Read => Ok(Level::Read),
2037 GrantLevel::Write => Ok(Level::Write),
2038 GrantLevel::Admin => Ok(Level::Admin),
2039 GrantLevel::Unspecified => Err(ControlError::InvalidArgument(
2040 "grant level unspecified".into(),
2041 )),
2042 }
2043}
2044
2045fn level_out(level: Level) -> GrantLevel {
2046 match level {
2047 Level::Read => GrantLevel::Read,
2048 Level::Write => GrantLevel::Write,
2049 Level::Admin => GrantLevel::Admin,
2050 }
2051}
2052
2053fn repo_out(repo: &gunnar_core::Repo) -> Repo {
2054 Repo {
2055 path: Some(RepoPath::new(&repo.namespace, &repo.name)),
2056 hash_kind: hash_kind(repo.hash_kind) as i32,
2057 default_branch: repo.default_branch.clone(),
2058 visibility: visibility_out(repo.visibility) as i32,
2059 created_at_unix_ms: repo.created,
2060 description: repo.description.clone(),
2061 encrypted: repo.encrypted,
2062 }
2063}
2064
2065fn web_credential_out(
2067 row: &gunnar_core::WebCredential,
2068) -> ControlResult<gunnar_proto::WebCredential> {
2069 let credential_id = gunnar_grpc::webauthn::from_b64url(&row.id).ok_or_else(|| {
2070 ControlError::Internal(format!(
2071 "a stored web credential has an id that is not base64url: {}",
2072 row.id
2073 ))
2074 })?;
2075 Ok(gunnar_proto::WebCredential {
2076 credential_id,
2077 principal_id: row.principal_id.clone(),
2078 label: row.label.clone(),
2079 created_at_unix_ms: row.created,
2080 last_used_unix_ms: row.last_used,
2081 sign_count: row.sign_count,
2082 })
2083}
2084
2085fn principal_out(principal: &gunnar_core::Principal) -> Principal {
2086 Principal {
2087 id: principal.id.clone(),
2088 name: principal.name.clone(),
2089 kind: principal_kind_out(principal.kind) as i32,
2090 created_at_unix_ms: principal.created,
2093 }
2094}
2095
2096mod clan_plane;
2098
2099fn key_out(key: &gunnar_core::PublicKey) -> PublicKey {
2100 PublicKey {
2101 fingerprint: key.fingerprint_sha256.clone(),
2102 algo: key.algo.clone(),
2103 principal_id: key.principal_id.clone(),
2104 comment: String::new(),
2107 added_at_unix_ms: key.added,
2108 last_used_unix_ms: key.last_used,
2109 expires_at_unix_ms: key.expires,
2110 }
2111}
2112
2113fn grant_out(grant: &Grant) -> gunnar_proto::Grant {
2118 let (namespace, repo) = match &grant.scope {
2119 Scope::Server => (String::new(), String::new()),
2120 Scope::Namespace(name) => (name.clone(), String::new()),
2121 Scope::Repo(path) => (path.namespace().to_owned(), path.name().to_owned()),
2122 };
2123 gunnar_proto::Grant {
2124 principal_id: grant.principal.clone(),
2125 namespace,
2126 repo,
2127 level: level_out(grant.level) as i32,
2128 granted_at_unix_ms: 0,
2129 }
2130}
2131
2132fn scope_in(namespace: &str, repo: &str) -> ControlResult<Scope> {
2133 match (namespace.is_empty(), repo.is_empty()) {
2134 (true, true) => Ok(Scope::Server),
2135 (false, true) => Ok(Scope::Namespace(namespace.to_owned())),
2136 (false, false) => RepoRef::new(namespace, repo)
2137 .map(Scope::Repo)
2138 .map_err(|err| ControlError::InvalidArgument(err.to_string())),
2139 (true, false) => Err(ControlError::InvalidArgument(
2140 "a repository-scoped grant must name a namespace (multi-tenancy has \
2141 no ungrouped repos)"
2142 .into(),
2143 )),
2144 }
2145}
2146
2147fn object_kind_out(kind: ObjType) -> ControlResult<gunnar_proto::ObjectKind> {
2157 match kind {
2158 ObjType::Blob => Ok(gunnar_proto::ObjectKind::Blob),
2159 ObjType::Tree => Ok(gunnar_proto::ObjectKind::Tree),
2160 ObjType::Commit => Ok(gunnar_proto::ObjectKind::Commit),
2161 ObjType::Tag => Ok(gunnar_proto::ObjectKind::Tag),
2162 other => Err(ControlError::Internal(format!(
2163 "the store answered a read with `{}`, which is a delta; GitServe::read is \
2164 documented to return the resolved type",
2165 other.as_str()
2166 ))),
2167 }
2168}
2169
2170fn oid_at_path(
2183 git: &dyn GitServe,
2184 hash_kind: gix_hash::Kind,
2185 tree: gix_hash::ObjectId,
2186 segments: &[&[u8]],
2187) -> ControlResult<Option<gix_hash::ObjectId>> {
2188 let mut current = tree;
2189 let last = segments.len().saturating_sub(1);
2190 for (index, segment) in segments.iter().enumerate() {
2191 let Some((kind, bytes)) = git.read(current.as_bytes()).map_err(store_error)? else {
2192 return Ok(None);
2193 };
2194 if kind != ObjType::Tree {
2195 return Ok(None);
2196 }
2197 let Some(entry) = object::tree_entry_named(&bytes, hash_kind, segment)
2198 .map_err(|err| ControlError::Internal(format!("decoding tree {current}: {err}")))?
2199 else {
2200 return Ok(None);
2201 };
2202 if index == last {
2203 return Ok(Some(entry.oid));
2204 }
2205 if entry.kind != object::EntryKind::Tree {
2206 return Ok(None);
2207 }
2208 current = entry.oid;
2209 }
2210 Ok(Some(current))
2211}
2212
2213#[derive(Clone, Copy)]
2219enum LogStart<'a> {
2220 One(&'a str),
2221 AllRefs,
2222}
2223
2224fn resolve_log_start(
2227 git: &dyn GitServe,
2228 hash: gix_hash::Kind,
2229 path: &RepoPath,
2230 start: &str,
2231) -> ControlResult<gix_hash::ObjectId> {
2232 let start = if start.is_empty() {
2233 gunnar_store::HEAD
2234 } else {
2235 start
2236 };
2237 let head = match gix_hash::ObjectId::from_hex(start.as_bytes()) {
2238 Ok(id) => id,
2239 Err(_) => {
2240 gunnar_store::refname::validate(start)
2241 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
2242 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
2243 let resolved = if start == gunnar_store::HEAD {
2247 snapshot.head(git).map_err(wire_error)?
2248 } else {
2249 snapshot.resolve(start).map_err(wire_error)?
2250 };
2251 resolved.ok_or_else(|| ControlError::NotFound(format!("{start} in {path}")))?
2252 }
2253 };
2254 if head.kind() != hash {
2255 return Err(ControlError::InvalidArgument(format!(
2256 "{start} is a {} oid; this repository is {hash}",
2257 head.kind()
2258 )));
2259 }
2260 Ok(head)
2261}
2262
2263impl CoreControlPlane {
2264 fn log_walk(
2265 &self,
2266 path: &RepoPath,
2267 start: LogStart<'_>,
2268 limit: u32,
2269 touching: &str,
2270 ) -> ControlResult<Vec<Commit>> {
2271 use std::collections::{BinaryHeap, HashMap, HashSet};
2272
2273 let (_, store) = self.open(path)?;
2274 let git = store.serve();
2275 let hash = store.hash_kind();
2276
2277 let filter: Option<Vec<&[u8]>> = if touching.is_empty() {
2281 None
2282 } else {
2283 let mut segments = Vec::new();
2284 for segment in touching.split('/').filter(|s| !s.is_empty()) {
2285 if segment == "." || segment == ".." {
2286 return Err(ControlError::InvalidArgument(format!(
2287 "the log path filter cannot contain {segment:?}"
2288 )));
2289 }
2290 segments.push(segment.as_bytes());
2291 }
2292 (!segments.is_empty()).then_some(segments)
2293 };
2294
2295 let heads: Vec<gix_hash::ObjectId> = match start {
2296 LogStart::One(start) => vec![resolve_log_start(git, hash, path, start)?],
2297 LogStart::AllRefs => {
2298 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
2299 let mut heads = Vec::new();
2300 if let Some(head) = snapshot.head(git).map_err(wire_error)? {
2301 heads.push(head);
2302 }
2303 for row in snapshot.rows().iter().filter(|r| {
2304 r.name.starts_with("refs/heads/") || r.name.starts_with("refs/tags/")
2305 }) {
2306 if let Some(id) = snapshot.resolve_row(row).map_err(wire_error)? {
2307 heads.push(id);
2308 }
2309 }
2310 heads
2311 }
2312 };
2313
2314 let limit = if limit == 0 { 50 } else { limit } as usize;
2320 let mut heap: BinaryHeap<(i64, gix_hash::ObjectId)> = BinaryHeap::new();
2321 let mut seen: HashSet<gix_hash::ObjectId> = HashSet::new();
2322 let mut queued: HashMap<gix_hash::ObjectId, object::Commit> = HashMap::new();
2323 let mut enqueue = |id: gix_hash::ObjectId,
2324 heap: &mut BinaryHeap<(i64, gix_hash::ObjectId)>,
2325 queued: &mut HashMap<gix_hash::ObjectId, object::Commit>|
2326 -> ControlResult<()> {
2327 if !seen.insert(id) {
2328 return Ok(());
2329 }
2330 let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
2333 return Ok(());
2334 };
2335 if kind != ObjType::Commit {
2336 return Ok(());
2337 }
2338 let commit = object::commit(&bytes, hash)
2339 .map_err(|err| ControlError::Internal(format!("decoding commit {id}: {err}")))?;
2340 heap.push((commit.committer.seconds, id));
2341 queued.insert(id, commit);
2342 Ok(())
2343 };
2344 for head in heads {
2345 let head = peel(git, hash, head)?;
2346 enqueue(head, &mut heap, &mut queued)?;
2347 }
2348 let mut out = Vec::new();
2349
2350 const MAX_FILTERED_SCAN: usize = 20_000;
2355 let mut scanned = 0usize;
2356 let mut tree_of: HashMap<gix_hash::ObjectId, gix_hash::ObjectId> = HashMap::new();
2359
2360 while let Some((_, id)) = heap.pop() {
2361 if out.len() >= limit {
2362 break;
2363 }
2364 let Some(commit) = queued.remove(&id) else {
2365 continue;
2366 };
2367 let object::Commit {
2368 tree,
2369 parents,
2370 author,
2371 committer,
2372 message,
2373 } = commit;
2374
2375 if let Some(segments) = &filter {
2382 scanned += 1;
2383 if scanned > MAX_FILTERED_SCAN {
2384 break;
2385 }
2386 tree_of.insert(id, tree);
2387 let current = oid_at_path(git, hash, tree, segments)?;
2388 let previous = match parents.first() {
2389 None => None,
2390 Some(parent) => {
2391 let parent_tree = match tree_of.get(parent) {
2392 Some(known) => Some(*known),
2393 None => match git.read(parent.as_bytes()).map_err(store_error)? {
2394 Some((ObjType::Commit, parent_bytes)) => {
2395 let decoded =
2396 object::commit(&parent_bytes, hash).map_err(|err| {
2397 ControlError::Internal(format!(
2398 "decoding commit {parent}: {err}"
2399 ))
2400 })?;
2401 tree_of.insert(*parent, decoded.tree);
2402 Some(decoded.tree)
2403 }
2404 _ => None,
2408 },
2409 };
2410 match parent_tree {
2411 Some(parent_tree) => oid_at_path(git, hash, parent_tree, segments)?,
2412 None => None,
2413 }
2414 }
2415 };
2416 if current == previous {
2417 for parent in parents {
2419 enqueue(parent, &mut heap, &mut queued)?;
2420 }
2421 continue;
2422 }
2423 }
2424
2425 out.push(Commit {
2426 oid: id.to_string(),
2427 parent_oids: parents.iter().map(|p| p.to_string()).collect(),
2428 tree_oid: tree.to_string(),
2429 author: Some(signature_out(
2430 &author.name,
2431 &author.email,
2432 author.seconds,
2433 author.offset_seconds,
2434 )),
2435 committer: Some(signature_out(
2436 &committer.name,
2437 &committer.email,
2438 committer.seconds,
2439 committer.offset_seconds,
2440 )),
2441 message: String::from_utf8_lossy(&message).into_owned(),
2442 });
2443 for parent in parents {
2444 enqueue(parent, &mut heap, &mut queued)?;
2445 }
2446 }
2447 Ok(out)
2448 }
2449}
2450
2451fn peel(
2452 git: &dyn GitServe,
2453 hash_kind: gix_hash::Kind,
2454 mut id: gix_hash::ObjectId,
2455) -> ControlResult<gix_hash::ObjectId> {
2456 for _ in 0..16 {
2457 let Some((kind, bytes)) = git.read(id.as_bytes()).map_err(store_error)? else {
2458 return Ok(id);
2459 };
2460 if kind != ObjType::Tag {
2461 return Ok(id);
2462 }
2463 id = object::tag_target(&bytes, hash_kind)
2464 .map_err(|err| ControlError::Internal(format!("decoding tag {id}: {err}")))?;
2465 }
2466 Err(ControlError::FailedPrecondition(format!(
2467 "the tag chain at {id} is longer than 16 links, or cyclic"
2468 )))
2469}
2470
2471fn ref_target_out(
2475 git: &dyn GitServe,
2476 hash_kind: gix_hash::Kind,
2477 snapshot: &RefSnapshot,
2478 row: &RefRow,
2479 want_peeled: bool,
2480) -> ControlResult<RefTarget> {
2481 let (oid, symbolic) = match (&row.oid, &row.symref_target) {
2482 (_, Some(inner)) => (snapshot.resolve(inner).map_err(wire_error)?, inner.clone()),
2486 (Some(raw), None) => (
2487 Some(gunnar_wire::refs::oid(raw).map_err(wire_error)?),
2488 String::new(),
2489 ),
2490 (None, None) => (None, String::new()),
2491 };
2492 let peeled = match (want_peeled, oid) {
2493 (true, Some(id)) => {
2498 let peeled = peel(git, hash_kind, id)?;
2499 if peeled == id {
2500 String::new()
2501 } else {
2502 peeled.to_string()
2503 }
2504 }
2505 _ => String::new(),
2506 };
2507 Ok(RefTarget {
2508 ref_name: row.name.clone(),
2509 oid: oid.map(|id| id.to_string()).unwrap_or_default(),
2510 symbolic_target: symbolic,
2511 peeled_oid: peeled,
2512 })
2513}
2514
2515fn signature_out(name: &[u8], email: &[u8], seconds: i64, tz_offset_seconds: i32) -> Signature {
2531 Signature {
2532 name: String::from_utf8_lossy(name).into_owned(),
2533 email: String::from_utf8_lossy(email).into_owned(),
2534 time_unix_seconds: seconds,
2535 tz_offset_minutes: tz_offset_seconds / 60,
2536 }
2537}
2538
2539impl gunnar_grpc::FirstBoot for CoreControlPlane {
2550 fn bootstrap_state(&self) -> ControlResult<gunnar_grpc::BootstrapState> {
2551 let auth = self.stores.auth();
2552 let window_open = auth.bootstrap_window_open().map_err(unavailable)?;
2555 let record = auth.bootstrap_record().map_err(unavailable)?;
2556 let (principals, keys, _grants) = auth.counts().map_err(unavailable)?;
2557 Ok(gunnar_grpc::BootstrapState {
2558 window_open,
2559 bootstrapped: record.is_some(),
2560 bootstrapped_at_unix_ms: record.as_ref().map(|r| r.at).unwrap_or_default(),
2561 bootstrapped_principal: record
2562 .as_ref()
2563 .map(|r| r.principal_id.clone())
2564 .unwrap_or_default(),
2565 bootstrapped_fingerprint: record
2566 .as_ref()
2567 .map(|r| r.fingerprint.clone())
2568 .unwrap_or_default(),
2569 principals: principals as u64,
2570 keys: keys as u64,
2571 serving_git: crate::readiness::is_serving_git(self.git_listener),
2581 readiness: crate::readiness::detail_with(self.git_listener),
2582 build: gunnar_common::BUILD_REF.to_owned(),
2583 role: if self.twin_role || self.stores.is_twin() {
2584 "twin"
2585 } else {
2586 "primary"
2587 }
2588 .to_owned(),
2589 recharge_from: awaited_twin(&self.stores),
2592 restoring: crate::twin::is_restoring(self.stores.root()),
2593 log_head_seq: self.stores.txlog().head_seq(),
2594 restore_partial: if crate::twin::is_restoring(self.stores.root()) {
2596 crate::twin::intake::partial(self.stores.root())
2597 } else {
2598 None
2599 },
2600 restore_snapshot_resumable: true,
2603 restore_snapshot: if crate::twin::is_restoring(self.stores.root()) {
2604 crate::twin::snapshot::held_snapshot(self.stores.root())
2605 .map(|h| (h.id, h.items, h.bytes, h.whole))
2606 } else {
2607 None
2608 },
2609 })
2610 }
2611
2612 fn enroll_first_identity(
2613 &self,
2614 name: &str,
2615 openssh_public_key: &str,
2616 ) -> ControlResult<gunnar_grpc::Enrolled> {
2617 #[cfg(feature = "replication")]
2620 if let Some(twin) = crate::repair::held(&self.stores) {
2621 return Err(ControlError::FailedPrecondition(
2622 crate::repair::held_refusal(&twin),
2623 ));
2624 }
2625 let enrolled = self
2626 .stores
2627 .auth()
2628 .enroll_first_user(name, openssh_public_key)
2629 .map_err(enrolment_error)?;
2630 if let Err(err) = self
2640 .stores
2641 .catalog()
2642 .create_namespace(&enrolled.record.account)
2643 {
2644 tracing::warn!(
2645 account = %enrolled.record.account,
2646 %err,
2647 "the first user was enrolled and the enrolment window is closed, but its account \
2648 namespace could not be created. The user is admin of it by rule and can create \
2649 it themselves with Repos.CreateNamespace."
2650 );
2651 }
2652 let (client_cert_pem, client_key_pem, issuer_cert_pem) = match self.stores.pki() {
2661 Ok(pki) => match pki.issuer() {
2662 Ok(Some(issuer)) => {
2663 match pki.issue_with_server_key(
2664 &enrolled.principal.id,
2665 gunnar_core::DEFAULT_LEAF_DAYS,
2666 ) {
2667 Ok(issued) => {
2668 if let Err(err) = self.stores.auth().register_certificate(
2669 &enrolled.principal.id,
2670 &issued.cert_der,
2671 None,
2672 ) {
2673 tracing::warn!(
2674 principal = %enrolled.principal.id,
2675 %err,
2676 "the bootstrap certificate was minted but could not be \
2677 registered; it is returned anyway and gunnar-admin can \
2678 register it with `gunnar key add`"
2679 );
2680 }
2681 (
2682 issued.cert_pem,
2683 issued.key_pem.unwrap_or_default(),
2684 issuer.cert_pem,
2685 )
2686 }
2687 Err(err) => {
2688 tracing::warn!(
2689 principal = %enrolled.principal.id,
2690 %err,
2691 "no bootstrap certificate: the issuer refused to mint"
2692 );
2693 (String::new(), String::new(), String::new())
2694 }
2695 }
2696 }
2697 Ok(None) => (String::new(), String::new(), String::new()),
2698 Err(err) => {
2699 tracing::warn!(%err, "no bootstrap certificate: the PKI document could not \
2700 be read");
2701 (String::new(), String::new(), String::new())
2702 }
2703 },
2704 Err(_) => (String::new(), String::new(), String::new()),
2705 };
2706 Ok(gunnar_grpc::Enrolled {
2707 principal: principal_out(&enrolled.principal),
2708 key: key_out(&enrolled.key),
2709 account: enrolled.record.account,
2710 bootstrapped_at_unix_ms: enrolled.record.at,
2711 client_cert_pem,
2712 client_key_pem,
2713 issuer_cert_pem,
2714 })
2715 }
2716}
2717
2718fn unavailable(err: gunnar_core::Error) -> ControlError {
2720 ControlError::Unavailable(format!("the identity table could not be read: {err}"))
2721}
2722
2723fn awaited_twin(stores: &ServerStores) -> String {
2729 #[cfg(feature = "wg-client")]
2730 if let Some(twin) = crate::pairing::recharge_awaited(stores) {
2731 return twin;
2732 }
2733 #[cfg(feature = "replication")]
2734 if let Some(twin) = crate::repair::held(stores) {
2735 return twin;
2736 }
2737 let _ = stores;
2738 String::new()
2739}
2740
2741fn enrolment_error(err: gunnar_core::Error) -> ControlError {
2742 match err {
2743 e @ (gunnar_core::Error::AlreadyBootstrapped { .. }
2746 | gunnar_core::Error::AlreadyProvisioned { .. }) => {
2747 ControlError::FailedPrecondition(e.to_string())
2748 }
2749 e @ (gunnar_core::Error::InvalidPath { .. }
2751 | gunnar_core::Error::ReservedPrincipalId(_)
2752 | gunnar_core::Error::Key(_)) => ControlError::InvalidArgument(e.to_string()),
2753 other => ControlError::Unavailable(other.to_string()),
2756 }
2757}
2758
2759impl CoreControlPlane {
2760 pub fn claim_operator_named(
2784 &self,
2785 fed: Option<&str>,
2786 asked: &str,
2787 ) -> ControlResult<gunnar_grpc::OperatorClaim> {
2788 let (name, named_by) = gunnar_grpc::seam::operator_name(fed, asked)?;
2789 let auth = self.stores.auth();
2790 let held = match auth.principal(name).map_err(internal)? {
2791 Some(p) => Some(p),
2792 None => auth.principal_by_name(name).map_err(internal)?,
2793 };
2794 if let Some(p) = held {
2795 if p.kind != PrincipalKind::User {
2796 return Err(ControlError::FailedPrecondition(format!(
2797 "the operator's name {name:?} is held by principal {:?}, which is not a person \
2798 ({:?}); an estate's machines are never its operator",
2799 p.id, p.kind
2800 )));
2801 }
2802 let adopted_namespace = false;
2803 if self
2804 .stores
2805 .catalog()
2806 .namespace(&p.id)
2807 .map_err(internal)?
2808 .is_none()
2809 {
2810 self.stores
2811 .catalog()
2812 .create_namespace(&p.id)
2813 .map_err(internal)?;
2814 }
2815 self.grant(&p.id, "", "", GrantLevel::Admin)?;
2816 return Ok(gunnar_grpc::OperatorClaim {
2817 principal: principal_out(&p),
2818 created: false,
2819 adopted_namespace,
2820 named_by,
2821 });
2822 }
2823 let principal =
2824 gunnar_core::Principal::new(name, name, PrincipalKind::User).map_err(internal)?;
2825 let created = auth.add_principal(principal).map_err(internal)?;
2826 let adopted_namespace = match self.stores.catalog().create_namespace(&created.id) {
2827 Ok(_) => false,
2828 Err(gunnar_core::Error::NamespaceExists(_)) => true,
2829 Err(err) => {
2830 let _ = auth.remove_principal(&created.id);
2831 return Err(internal(err));
2832 }
2833 };
2834 if let Err(err) = self.grant(&created.id, "", "", GrantLevel::Admin) {
2835 let _ = auth.remove_principal(&created.id);
2838 return Err(err);
2839 }
2840 self.stores.mirror_identity();
2841 tracing::warn!(
2842 target: "gunnar::audit",
2843 event = "operator.claimed",
2844 operator = %created.id,
2845 adopted_namespace,
2846 named_by,
2847 "the estate's operator was created: a person, keyless, gunnar-admin, with its account \
2848 namespace{}",
2849 if adopted_namespace {
2850 " — the namespace already existed with no owner (a first push conjured it) and is \
2851 now the operator's account"
2852 } else {
2853 ""
2854 }
2855 );
2856 Ok(gunnar_grpc::OperatorClaim {
2857 principal: principal_out(&created),
2858 created: true,
2859 adopted_namespace,
2860 named_by,
2861 })
2862 }
2863}
2864
2865mod enterprise_plane;
2867
2868impl ControlPlane for CoreControlPlane {
2869 fn enterprise(&self) -> Option<&dyn gunnar_grpc::enterprise::EnterprisePlane> {
2870 Some(self)
2871 }
2872
2873 fn create_namespace(&self, name: &str, _description: &str) -> ControlResult<Namespace> {
2876 if self
2880 .stores
2881 .auth()
2882 .employer_of(name)
2883 .map_err(internal)?
2884 .is_some()
2885 {
2886 return Err(ControlError::AlreadyExists(format!(
2887 "{name} is an employee's id, and an employee holds no account of its own"
2888 )));
2889 }
2890 let namespace = self
2891 .stores
2892 .catalog()
2893 .create_namespace(name)
2894 .map_err(internal)?;
2895 Ok(self.namespace_out(namespace, 0))
2896 }
2897
2898 fn list_namespaces(&self) -> ControlResult<Vec<Namespace>> {
2899 let mut employees: std::collections::BTreeMap<String, u32> = Default::default();
2900 for principal in self.stores.auth().principals().map_err(internal)? {
2901 if let Some(account) = self
2902 .stores
2903 .auth()
2904 .employer_of(&principal.id)
2905 .map_err(internal)?
2906 {
2907 *employees.entry(account).or_default() += 1;
2908 }
2909 }
2910 Ok(self
2911 .stores
2912 .catalog()
2913 .namespaces()
2914 .map_err(internal)?
2915 .into_iter()
2916 .map(|ns| {
2917 let count = employees.get(&ns.name).copied().unwrap_or(0);
2918 self.namespace_out(ns, count)
2919 })
2920 .collect())
2921 }
2922
2923 fn namespace(&self, name: &str) -> ControlResult<Option<Namespace>> {
2926 let Some(ns) = self.stores.catalog().namespace(name).map_err(internal)? else {
2927 return Ok(None);
2928 };
2929 let count = self.employee_count(name)?;
2930 Ok(Some(self.namespace_out(ns, count)))
2931 }
2932
2933 fn namespace_for_outsider(
2934 &self,
2935 principal_id: &str,
2936 name: &str,
2937 ) -> ControlResult<Option<Namespace>> {
2938 let auth = self.stores.auth();
2941 if auth
2942 .gate_covering(principal_id)
2943 .map_err(internal)?
2944 .is_some()
2945 || auth.gate_covering(name).map_err(internal)?.is_some()
2946 {
2947 return Ok(None);
2948 }
2949 self.namespace(name)
2950 }
2951
2952 fn set_namespace_settings(
2953 &self,
2954 name: &str,
2955 public: Option<bool>,
2956 events_public: Option<bool>,
2957 ) -> ControlResult<Namespace> {
2958 let catalog = self.stores.catalog();
2959 let mut ns = catalog
2960 .namespace(name)
2961 .map_err(internal)?
2962 .ok_or_else(|| ControlError::NotFound(format!("namespace {name}")))?;
2963 if let Some(public) = public {
2964 ns = catalog
2965 .set_namespace_public(name, public)
2966 .map_err(internal)?;
2967 }
2968 if let Some(events_public) = events_public {
2969 ns = catalog
2970 .set_namespace_events_public(name, events_public)
2971 .map_err(internal)?;
2972 }
2973 let count = self.employee_count(name)?;
2974 Ok(self.namespace_out(ns, count))
2975 }
2976
2977 fn delete_namespace(&self, name: &str, delete_repos: bool) -> ControlResult<u32> {
2978 if self.stores.auth().is_clan(name).map_err(internal)? {
2983 return Err(ControlError::FailedPrecondition(format!(
2984 "namespace {name} is clan {name}'s: it goes when the clan's admin dissolves the clan"
2985 )));
2986 }
2987 let gone = self
2988 .stores
2989 .catalog()
2990 .delete_namespace(name, delete_repos)
2991 .map_err(internal)?;
2992 if let Err(err) = self.stores.flow().forget(name) {
3003 tracing::warn!(
3004 target: "gunnar::store",
3005 event = "flow.forget_failed",
3006 namespace = %name,
3007 %err,
3008 "the namespace was deleted but its activity series could not be removed"
3009 );
3010 }
3011 Ok(gone.len() as u32)
3012 }
3013
3014 fn create_repo(&self, spec: ProtoNewRepo) -> ControlResult<Repo> {
3017 let path = Self::repo_ref(&spec.path)?;
3018 if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
3024 return Err(ControlError::ResourceExhausted(line));
3025 }
3026 let mut new = gunnar_core::NewRepo::new(path).visibility(visibility_in(spec.visibility)?);
3027 if spec.hash_kind != gunnar_proto::HashKind::Unspecified {
3028 new = new.hash_kind(match spec.hash_kind {
3029 gunnar_proto::HashKind::Sha1 => HashKind::Sha1,
3030 _ => HashKind::Sha256,
3031 });
3032 }
3033 if !spec.default_branch.is_empty() {
3034 new = new.default_branch(spec.default_branch.clone());
3035 }
3036 let repo = self
3045 .stores
3046 .create_repo(new, spec.adopt)
3047 .map_err(|err| create_error(&err, &spec.path))?;
3048 Ok(repo_out(&repo))
3049 }
3050
3051 fn write_first_commit(&self, path: &RepoPath, readme: &str) -> ControlResult<String> {
3052 let (repo, store) = self.open(path)?;
3053 let kind = match repo.hash_kind {
3054 HashKind::Sha1 => gix_hash::Kind::Sha1,
3055 _ => gix_hash::Kind::Sha256,
3056 };
3057 let when = std::time::SystemTime::now()
3058 .duration_since(std::time::UNIX_EPOCH)
3059 .map(|d| d.as_secs() as i64)
3060 .unwrap_or(0);
3061 let (pack, commit) = crate::first_commit::first_commit_pack(
3062 kind,
3063 "README.md",
3064 readme.as_bytes(),
3065 "Welcome",
3066 when,
3067 )
3068 .map_err(ControlError::InvalidArgument)?;
3069 store
3070 .serve()
3071 .put(
3072 &pack,
3073 &[gunnar_store::git_storage_trait::RefUpdate::set(
3074 repo.default_branch.clone(),
3075 commit.clone(),
3076 )],
3077 )
3078 .map_err(|err| {
3079 ControlError::Internal(format!("writing the first commit of {path}: {err:#}"))
3080 })?;
3081 Ok(commit)
3082 }
3083
3084 fn create_room(&self, namespace: &str, machine: bool) -> ControlResult<()> {
3089 use gunnar_ssh::{PlanAction, PlanVerdict};
3090 for gate in self
3091 .stores
3092 .plan_gates()
3093 .iter()
3094 .filter(|gate| !machine || gate.applies_to_machine())
3095 {
3096 if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
3097 return Err(ControlError::FailedPrecondition(line));
3098 }
3099 if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
3100 return Err(ControlError::ResourceExhausted(line));
3101 }
3102 }
3103 Ok(())
3104 }
3105
3106 fn get_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
3107 let repo_ref = Self::repo_ref(path)?;
3108 self.stores
3109 .catalog()
3110 .get(&repo_ref)
3111 .map_err(internal)?
3112 .as_ref()
3113 .map(repo_out)
3114 .ok_or_else(|| ControlError::NotFound(format!("repository {path}")))
3115 }
3116
3117 fn list_repos(&self, namespace: Option<&str>) -> ControlResult<Vec<Repo>> {
3118 Ok(self
3119 .stores
3120 .catalog()
3121 .list(namespace)
3122 .map_err(internal)?
3123 .iter()
3124 .map(repo_out)
3125 .collect())
3126 }
3127
3128 fn delete_repo(
3149 &self,
3150 path: &RepoPath,
3151 options: gunnar_grpc::seam::DeleteOptions,
3152 ) -> ControlResult<gunnar_grpc::seam::DeleteOutcome> {
3153 let repo_ref = Self::repo_ref(path)?;
3154 let was = self.stores.catalog().get(&repo_ref).map_err(internal)?;
3158 let orphan = was.is_none() && self.stores.is_orphan(&repo_ref).unwrap_or(false);
3167 if orphan && !options.purge {
3168 return Err(ControlError::FailedPrecondition(
3169 crate::stores::ServerStores::orphan_soft_delete_refusal(&repo_ref),
3170 ));
3171 }
3172 match self.stores.catalog().delete(&repo_ref) {
3176 Ok(_) => {}
3177 Err(gunnar_core::Error::NoSuchRepo(_))
3178 if orphan || self.stores.is_tombstoned(&repo_ref).unwrap_or(false) => {}
3179 Err(err) => return Err(internal(err)),
3180 }
3181 if options.purge {
3182 let (before, after) = self.stores.purge_repo(&repo_ref).map_err(|err| {
3183 ControlError::Internal(format!("removing the repository's directory: {err:#}"))
3184 })?;
3185 return Ok(gunnar_grpc::seam::DeleteOutcome::Purged {
3186 bytes_reclaimed: before.saturating_sub(after),
3187 });
3188 }
3189 let retention = std::time::Duration::from_secs(
3190 u64::from(
3191 options
3192 .retention_days
3193 .unwrap_or(crate::stores::DEFAULT_DELETE_RETENTION_DAYS),
3194 ) * 86_400,
3195 );
3196 let record = self
3197 .stores
3198 .soft_delete(&repo_ref, was.as_ref(), &options.principal_id, retention)
3199 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
3200 let bytes_kept = self
3201 .stores
3202 .repo_dir(&repo_ref)
3203 .map(|dir| crate::stores::dir_bytes(&dir))
3204 .unwrap_or(0);
3205 Ok(gunnar_grpc::seam::DeleteOutcome::Tombstoned {
3206 purge_after_unix_ms: record.purge_after_unix_ms.unwrap_or_default(),
3207 bytes_kept,
3208 })
3209 }
3210
3211 fn undelete_repo(&self, path: &RepoPath) -> ControlResult<Repo> {
3220 let repo_ref = Self::repo_ref(path)?;
3221 if self
3222 .stores
3223 .catalog()
3224 .get(&repo_ref)
3225 .map_err(internal)?
3226 .is_some()
3227 {
3228 return Err(ControlError::AlreadyExists(format!(
3229 "{path} is in the catalog already; it is not deleted"
3230 )));
3231 }
3232 let spec = self
3233 .stores
3234 .undelete_spec(&repo_ref)
3235 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
3236 let repo = self
3242 .stores
3243 .create_repo(spec, true)
3244 .map_err(|err| ControlError::Internal(format!("restoring {path}: {err:#}")))?;
3245 self.stores
3246 .clear_tombstone(&repo_ref)
3247 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
3248 Ok(repo_out(&repo))
3249 }
3250
3251 fn purge_repo(&self, path: &RepoPath) -> ControlResult<u64> {
3252 let repo_ref = Self::repo_ref(path)?;
3253 if self
3254 .stores
3255 .catalog()
3256 .get(&repo_ref)
3257 .map_err(internal)?
3258 .is_some()
3259 {
3260 return Err(ControlError::FailedPrecondition(format!(
3261 "{path} is a LIVE repository, not a deleted one. `repo delete {path} --purge` \
3262 deletes and reclaims it in one step; purge alone only ends a retention window."
3263 )));
3264 }
3265 let tombstone = self
3266 .stores
3267 .tombstone_of(&repo_ref)
3268 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
3269 if tombstone.is_none() {
3270 return Err(ControlError::NotFound(format!(
3271 "{path} is neither catalogued nor tombstoned; there is nothing to purge"
3272 )));
3273 }
3274 let (before, after) = self
3275 .stores
3276 .purge_repo(&repo_ref)
3277 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
3278 Ok(before.saturating_sub(after))
3279 }
3280
3281 fn gc_repo(&self, path: &RepoPath) -> ControlResult<gunnar_grpc::seam::GcJob> {
3293 let repo_ref = Self::repo_ref(path)?;
3294 if self
3301 .stores
3302 .catalog()
3303 .get(&repo_ref)
3304 .map_err(internal)?
3305 .is_none()
3306 {
3307 return Err(ControlError::NotFound(format!(
3308 "{path} is not a live repository; nothing to gc"
3309 )));
3310 }
3311 Ok(gc_job_out(path, self.gc.enqueue(&repo_ref.to_string())))
3312 }
3313
3314 fn gc_status(&self, path: &RepoPath) -> ControlResult<Option<gunnar_grpc::seam::GcJob>> {
3315 let repo_ref = Self::repo_ref(path)?;
3316 Ok(self
3317 .gc
3318 .status(&repo_ref.to_string())
3319 .map(|job| gc_job_out(path, job)))
3320 }
3321
3322 fn gate_account(
3325 &self,
3326 account: &str,
3327 actor: &str,
3328 note: &str,
3329 ) -> ControlResult<(gunnar_grpc::seam::AccountGate, bool)> {
3330 let before = self
3337 .stores
3338 .auth()
3339 .gate_covering(account)
3340 .map_err(internal)?
3341 .is_some();
3342 let gate = self
3343 .stores
3344 .auth()
3345 .gate(gunnar_core::AccountGate::administrative(
3346 account, actor, note,
3347 ))
3348 .map_err(gate_error)?;
3349 tracing::warn!(
3350 target: "gunnar::audit",
3351 event = "account.gated",
3352 account = %account,
3353 actor = %actor,
3354 note = %note,
3355 already = before,
3356 "an account was locked out: every level, both doors, until it is ungated"
3357 );
3358 Ok((gate_out(gate), before))
3359 }
3360
3361 fn ungate_account(
3362 &self,
3363 account: &str,
3364 ) -> ControlResult<Option<gunnar_grpc::seam::AccountGate>> {
3365 let lifted = self
3366 .stores
3367 .auth()
3368 .lift_gate(account)
3369 .map_err(internal)?
3370 .map(gate_out);
3371 if let Some(gate) = &lifted {
3372 tracing::warn!(
3373 target: "gunnar::audit",
3374 event = "account.ungated",
3375 account = %account,
3376 held_by = %gate.holder,
3377 "an account lockout was lifted"
3378 );
3379 }
3380 Ok(lifted)
3381 }
3382
3383 fn account_gates(&self) -> ControlResult<Vec<gunnar_grpc::seam::AccountGate>> {
3384 Ok(self
3385 .stores
3386 .auth()
3387 .gates()
3388 .map_err(internal)?
3389 .into_iter()
3390 .map(gate_out)
3391 .collect())
3392 }
3393
3394 fn purge_account(
3412 &self,
3413 account: &str,
3414 options: gunnar_grpc::seam::PurgeAccountOptions,
3415 ) -> ControlResult<(gunnar_grpc::seam::PurgeJob, gunnar_grpc::seam::AccountGate)> {
3416 let account = account.trim();
3417 if account.is_empty() {
3418 return Err(ControlError::InvalidArgument(
3419 "an account purge needs an account name".into(),
3420 ));
3421 }
3422 if let Some(p) = self.stores.auth().principal(account).map_err(internal)? {
3429 if p.kind != gunnar_core::PrincipalKind::User {
3430 return Err(ControlError::FailedPrecondition(format!(
3431 "{account} is a machine principal ({:?}), not a person: a machine's account is never \
3432 purged. Remove the principal itself if it is really to go.",
3433 p.kind
3434 )));
3435 }
3436 }
3437 let job_id = self.purge.next_job_id();
3438 self.gate_for_purge(account, &job_id, &options)?;
3443
3444 let order = crate::purge_queue::PurgeOrder::new(account, &job_id, &options.reason)
3445 .by(&options.principal_id)
3446 .removing_principal(options.remove_principal)
3447 .expecting_empty(options.expect_empty);
3448 let job = self.purge.enqueue(order);
3449
3450 let gate = self
3451 .stores
3452 .auth()
3453 .gate_covering(account)
3454 .map_err(internal)?
3455 .map(gate_out)
3456 .unwrap_or_else(|| gunnar_grpc::seam::AccountGate {
3461 account: account.to_owned(),
3462 reason: gunnar_grpc::seam::GateReason::Purge,
3463 holder: job_id.clone(),
3464 actor: options.principal_id.clone(),
3465 note: "the lockout was lifted between being taken and being read back".into(),
3466 at_unix_ms: gunnar_core::now_ms(),
3467 expires_unix_ms: 0,
3468 });
3469 Ok((purge_job_out(job), gate))
3470 }
3471
3472 fn purge_account_status(
3473 &self,
3474 account: &str,
3475 ) -> ControlResult<(
3476 Option<gunnar_grpc::seam::PurgeJob>,
3477 Option<gunnar_grpc::seam::AccountGate>,
3478 )> {
3479 let job = self.purge.status(account).map(purge_job_out);
3480 let gate = self
3481 .stores
3482 .auth()
3483 .gate_covering(account)
3484 .map_err(internal)?
3485 .map(gate_out);
3486 Ok((job, gate))
3487 }
3488
3489 fn default_delete_retention_days(&self) -> u32 {
3490 crate::stores::DEFAULT_DELETE_RETENTION_DAYS
3491 }
3492
3493 fn list_deleted(
3494 &self,
3495 namespace: Option<&str>,
3496 ) -> ControlResult<Vec<gunnar_grpc::seam::DeletedRepoRow>> {
3497 Ok(self
3498 .stores
3499 .tombstones()
3500 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
3501 .into_iter()
3502 .filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
3503 .map(|row| gunnar_grpc::seam::DeletedRepoRow {
3504 path: RepoPath {
3505 namespace: row.path.namespace().to_owned(),
3506 repo: row.path.name().to_owned(),
3507 },
3508 deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
3509 purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
3510 principal_id: row.tombstone.principal_id,
3511 bytes: row.bytes,
3512 })
3513 .collect())
3514 }
3515
3516 fn list_orphans(
3526 &self,
3527 namespace: Option<&str>,
3528 ) -> ControlResult<Vec<gunnar_grpc::seam::OrphanRow>> {
3529 Ok(self
3530 .stores
3531 .orphans()
3532 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
3533 .into_iter()
3534 .filter(|row| namespace.is_none_or(|want| row.path.namespace() == want))
3535 .map(|row| gunnar_grpc::seam::OrphanRow {
3536 path: RepoPath {
3537 namespace: row.path.namespace().to_owned(),
3538 repo: row.path.name().to_owned(),
3539 },
3540 dir: row.dir.display().to_string(),
3541 bytes: row.bytes,
3542 hash_marker: row.hash_marker.unwrap_or_default(),
3543 modified_unix_ms: row.modified_unix_ms,
3544 failed_reclaim: row.failed_reclaim.unwrap_or_default(),
3545 })
3546 .collect())
3547 }
3548
3549 fn reclaim_orphan(&self, path: &RepoPath) -> ControlResult<u64> {
3553 let repo_ref = Self::repo_ref(path)?;
3554 self.stores
3555 .reclaim_orphan(&repo_ref)
3556 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
3557 }
3558
3559 fn index_progress(&self) -> ControlResult<Vec<gunnar_grpc::seam::IndexProgressRow>> {
3560 Ok(self
3561 .stores
3562 .engine()
3563 .index_progress()
3564 .into_iter()
3565 .filter_map(|(dir, s)| {
3566 let (namespace, name) = self.stores.repo_of_dir(&dir)?;
3567 Some(gunnar_grpc::seam::IndexProgressRow {
3568 namespace,
3569 name,
3570 pack_id: s.pack_id,
3571 objects_done: s.objects_done,
3572 objects_total: s.objects_total,
3573 bytes_done: s.bytes_done,
3574 bytes_total: s.bytes_total,
3575 phase: if s.active {
3576 s.phase.to_string()
3577 } else {
3578 "idle".to_string()
3579 },
3580 started_at_unix_ms: s.started_at_ms as i64,
3581 resumed: s.resumed,
3582 reinflations: s.reinflations,
3583 pauses: s.pauses,
3584 })
3585 })
3586 .filter(|r| r.phase != "idle")
3587 .collect())
3588 }
3589
3590 fn set_default_branch(&self, path: &RepoPath, default_branch: &str) -> ControlResult<Repo> {
3591 let repo_ref = Self::repo_ref(path)?;
3592 let repo = self
3593 .stores
3594 .catalog()
3595 .set_default_branch(&repo_ref, default_branch)
3596 .map_err(internal)?;
3597 self.stores.repoint_head(&repo).map_err(|err| {
3605 ControlError::Internal(format!("pointing HEAD at the new default branch: {err:#}"))
3606 })?;
3607 Ok(repo_out(&repo))
3608 }
3609
3610 fn set_visibility(
3611 &self,
3612 path: &RepoPath,
3613 visibility: gunnar_proto::Visibility,
3614 ) -> ControlResult<Repo> {
3615 if visibility == gunnar_proto::Visibility::Unspecified {
3616 return Err(ControlError::InvalidArgument(
3617 "visibility unspecified".into(),
3618 ));
3619 }
3620 let repo_ref = Self::repo_ref(path)?;
3621 self.stores
3622 .catalog()
3623 .set_visibility(&repo_ref, visibility_in(visibility)?)
3624 .map(|repo| repo_out(&repo))
3625 .map_err(internal)
3626 }
3627
3628 fn recipient_keys(&self, principal_id: &str) -> ControlResult<Vec<gunnar_proto::RecipientKey>> {
3632 let public = self
3633 .stores
3634 .encryption_keys()
3635 .public_of(principal_id)
3636 .map_err(key_refusal)?;
3637 Ok(public
3638 .into_iter()
3639 .map(|public| gunnar_proto::RecipientKey {
3640 fingerprint: gunnar_core::auth::fingerprint_of_blob(public.as_bytes()).to_string(),
3641 algo: if public.starts_with("age1pq1") {
3642 gunnar_core::auth::PQ_RECIPIENT_ALGO.to_owned()
3643 } else {
3644 "X25519".to_owned()
3645 },
3646 openssh: public,
3647 })
3648 .collect())
3649 }
3650
3651 fn encryption_key(
3652 &self,
3653 principal_id: &str,
3654 ) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3655 Ok(encryption_key_out(
3656 self.stores
3657 .encryption_keys()
3658 .get(principal_id)
3659 .map_err(key_refusal)?,
3660 ))
3661 }
3662
3663 fn create_encryption_key(
3664 &self,
3665 principal_id: &str,
3666 public_key: &str,
3667 first: gunnar_proto::UnlockMethod,
3668 ) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3669 let first = method_in(first)?;
3670 let key = self
3671 .stores
3672 .encryption_keys()
3673 .create(
3674 principal_id,
3675 public_key,
3676 first,
3677 gunnar_core::now_ms().max(0) as u64,
3678 )
3679 .map_err(key_refusal)?;
3680 Ok(encryption_key_out(Some(key)))
3681 }
3682
3683 fn add_unlock_method(
3684 &self,
3685 principal_id: &str,
3686 method: gunnar_proto::UnlockMethod,
3687 ) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3688 let method = method_in(method)?;
3689 let key = self
3690 .stores
3691 .encryption_keys()
3692 .add_method(principal_id, method, gunnar_core::now_ms().max(0) as u64)
3693 .map_err(key_refusal)?;
3694 Ok(encryption_key_out(Some(key)))
3695 }
3696
3697 fn remove_unlock_method(
3698 &self,
3699 principal_id: &str,
3700 id: &str,
3701 ) -> ControlResult<gunnar_proto::GetEncryptionKeyResponse> {
3702 let key = self
3703 .stores
3704 .encryption_keys()
3705 .remove_method(principal_id, id)
3706 .map_err(key_refusal)?;
3707 Ok(encryption_key_out(Some(key)))
3708 }
3709
3710 fn set_encrypted(&self, path: &RepoPath, encrypted: bool) -> ControlResult<Repo> {
3711 let repo_ref = Self::repo_ref(path)?;
3712 self.stores
3713 .catalog()
3714 .set_encrypted(&repo_ref, encrypted)
3715 .map(|repo| repo_out(&repo))
3716 .map_err(internal)
3717 }
3718
3719 #[cfg(feature = "engine-znippy")]
3730 fn archive_out(
3731 &self,
3732 path: &RepoPath,
3733 from_generation: u64,
3734 sink: &mut dyn std::io::Write,
3735 ) -> ControlResult<gunnar_grpc::seam::ArchiveCut> {
3736 let repo_ref = Self::repo_ref(path)?;
3737 let dir = self
3738 .stores
3739 .repo_dir(&repo_ref)
3740 .map_err(|err| ControlError::Internal(format!("resolving {path}: {err:#}")))?;
3741 if !dir.is_dir() {
3742 return Err(ControlError::NotFound(format!(
3743 "{path} has nothing on disk at {}",
3744 dir.display()
3745 )));
3746 }
3747 let receipt =
3748 crate::storage::znippy_engine::archive_out_to(&dir, &mut { sink }, from_generation)
3749 .map_err(archive_error)?;
3750 self.stores.flow().record(
3763 &path.namespace,
3764 gunnar_core::now_ms(),
3765 &gunnar_core::Transfer {
3766 bytes_out: receipt.bytes_out,
3767 fetched: true,
3768 ..gunnar_core::Transfer::default()
3769 },
3770 );
3771 Ok(gunnar_grpc::seam::ArchiveCut {
3772 from_generation: receipt.from_generation,
3773 to_generation: receipt.to_generation,
3774 bytes_out: receipt.bytes_out,
3775 file_sha256: receipt.file_sha256,
3776 })
3777 }
3778
3779 fn list_refs(
3782 &self,
3783 path: &RepoPath,
3784 prefix: &str,
3785 peel_them: bool,
3786 ) -> ControlResult<Vec<RefTarget>> {
3787 let (_, store) = self.open(path)?;
3788 let git = store.serve();
3789 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3794 let mut out = Vec::new();
3795 for row in snapshot.with_prefix(Some(prefix)) {
3796 out.push(ref_target_out(
3797 git,
3798 store.hash_kind(),
3799 &snapshot,
3800 row,
3801 peel_them,
3802 )?);
3803 }
3804 Ok(out)
3805 }
3806
3807 fn get_ref(
3808 &self,
3809 path: &RepoPath,
3810 ref_name: &str,
3811 peel_them: bool,
3812 ) -> ControlResult<RefTarget> {
3813 let (_, store) = self.open(path)?;
3814 let git = store.serve();
3815 gunnar_store::refname::validate(ref_name)
3821 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
3822 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3823 let head_row;
3829 let row = if ref_name == gunnar_store::HEAD {
3830 head_row = git
3831 .head()
3832 .map_err(store_error)?
3833 .ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?;
3834 &head_row
3835 } else {
3836 snapshot
3837 .get(ref_name)
3838 .ok_or_else(|| ControlError::NotFound(format!("{ref_name} in {path}")))?
3839 };
3840 ref_target_out(git, store.hash_kind(), &snapshot, row, peel_them)
3841 }
3842
3843 fn log(
3844 &self,
3845 path: &RepoPath,
3846 start: &str,
3847 limit: u32,
3848 touching: &str,
3849 ) -> ControlResult<Vec<Commit>> {
3850 self.log_walk(path, LogStart::One(start), limit, touching)
3851 }
3852
3853 fn log_all_refs(
3857 &self,
3858 path: &RepoPath,
3859 limit: u32,
3860 touching: &str,
3861 ) -> ControlResult<Vec<Commit>> {
3862 self.log_walk(path, LogStart::AllRefs, limit, touching)
3863 }
3864
3865 fn create_pull(
3866 &self,
3867 path: &RepoPath,
3868 author: &str,
3869 title: &str,
3870 description: &str,
3871 source_ref: &str,
3872 target_ref: &str,
3873 ) -> ControlResult<PullRequest> {
3874 for (what, name) in [("source_ref", source_ref), ("target_ref", target_ref)] {
3875 if !name.starts_with("refs/heads/") {
3876 return Err(ControlError::InvalidArgument(format!(
3877 "{what} must be a full branch name under refs/heads/, got {name:?}: a pull \
3878 request merges branches, and a short name stored today is an ambiguity \
3879 resolved differently tomorrow"
3880 )));
3881 }
3882 }
3883 if source_ref == target_ref {
3884 return Err(ControlError::InvalidArgument(
3885 "a branch cannot be merged into itself".to_owned(),
3886 ));
3887 }
3888 if title.trim().is_empty() {
3889 return Err(ControlError::InvalidArgument(
3890 "a title is required".to_owned(),
3891 ));
3892 }
3893 let repo_ref = Self::repo_ref(path)?;
3894 let (_, store) = self.open(path)?;
3895 let git = store.serve();
3896 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
3897 let source = snapshot
3898 .resolve(source_ref)
3899 .map_err(wire_error)?
3900 .ok_or_else(|| ControlError::NotFound(format!("{source_ref} in {path}")))?;
3901 snapshot
3902 .resolve(target_ref)
3903 .map_err(wire_error)?
3904 .ok_or_else(|| ControlError::NotFound(format!("{target_ref} in {path}")))?;
3905 let record = self
3906 .pulls()?
3907 .create(
3908 &repo_ref.to_string(),
3909 author,
3910 title,
3911 description,
3912 source_ref,
3913 target_ref,
3914 &source.to_string(),
3915 )
3916 .map_err(internal)?;
3917 Ok(pull_out(path, record))
3918 }
3919
3920 fn list_pulls(&self, path: &RepoPath, state: i32) -> ControlResult<Vec<PullRequest>> {
3921 let repo_ref = Self::repo_ref(path)?;
3922 self.get_repo(path)?;
3925 let filter = match PullRequestState::try_from(state) {
3926 Ok(PullRequestState::Unspecified) | Err(_) => None,
3927 Ok(PullRequestState::Open) => Some(gunnar_core::PullState::Open),
3928 Ok(PullRequestState::Merged) => Some(gunnar_core::PullState::Merged),
3929 Ok(PullRequestState::Closed) => Some(gunnar_core::PullState::Closed),
3930 };
3931 Ok(self
3932 .pulls()?
3933 .list(&repo_ref.to_string(), filter)
3934 .map_err(internal)?
3935 .into_iter()
3936 .map(|record| pull_out(path, record))
3937 .collect())
3938 }
3939
3940 fn get_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
3941 let repo_ref = Self::repo_ref(path)?;
3942 self.get_repo(path)?;
3943 self.pulls()?
3944 .get(&repo_ref.to_string(), number)
3945 .map_err(internal)?
3946 .map(|record| pull_out(path, record))
3947 .ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))
3948 }
3949
3950 fn close_pull(&self, path: &RepoPath, number: u64) -> ControlResult<PullRequest> {
3951 let repo_ref = Self::repo_ref(path)?;
3952 self.get_repo(path)?;
3953 match self
3954 .pulls()?
3955 .close(&repo_ref.to_string(), number)
3956 .map_err(internal)?
3957 {
3958 gunnar_core::PullTransition::Done(record) => Ok(pull_out(path, record)),
3959 gunnar_core::PullTransition::NotFound => Err(ControlError::NotFound(format!(
3960 "pull request #{number} in {path}"
3961 ))),
3962 gunnar_core::PullTransition::NotOpen(state) => Err(ControlError::FailedPrecondition(
3963 format!("pull request #{number} is already {state:?}, not open"),
3964 )),
3965 }
3966 }
3967
3968 fn merge_pull(
3977 &self,
3978 path: &RepoPath,
3979 number: u64,
3980 principal_id: &str,
3981 options: &gunnar_grpc::MergeOptions,
3982 ) -> ControlResult<(PullRequest, MergedRef)> {
3983 self.merge_or_check(path, number, principal_id, options, false)
3984 }
3985
3986 fn merge_signing_key(&self) -> ControlResult<(String, String)> {
3988 let public = match self
3990 .stores
3991 .merge_signing_public()
3992 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
3993 {
3994 Some(public) => public,
3995 None => {
3996 if let Some(why) = self.frozen() {
3997 return Err(ControlError::Unavailable(format!(
3998 "{why}: no merge-signing key here yet; ask the primary"
3999 )));
4000 }
4001 self.stores
4002 .merge_signing_key()
4003 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
4004 .public_key()
4005 .clone()
4006 }
4007 };
4008 let mut public = public;
4009 public.set_comment("gunnar-merge-signing");
4010 let openssh = public
4011 .to_openssh()
4012 .map_err(|err| ControlError::Internal(format!("the merge-signing key: {err}")))?;
4013 Ok((openssh, gunnar_core::sshsig::fingerprint(&public)))
4014 }
4015
4016 fn check_merge(
4017 &self,
4018 path: &RepoPath,
4019 number: u64,
4020 principal_id: &str,
4021 options: &gunnar_grpc::MergeOptions,
4022 ) -> ControlResult<(PullRequest, MergedRef)> {
4023 self.merge_or_check(path, number, principal_id, options, true)
4024 }
4025
4026 fn read_object(
4027 &self,
4028 path: &RepoPath,
4029 oid: &str,
4030 max_bytes: u64,
4031 ) -> ControlResult<ReadObjectResponse> {
4032 let (_, store) = self.open(path)?;
4033 let git = store.serve();
4034 let id = gix_hash::ObjectId::from_hex(oid.as_bytes())
4035 .map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
4036
4037 if let Some((_, object_bytes)) = git.header(id.as_bytes()).map_err(store_error)? {
4045 if self.max_object_bytes != 0 && object_bytes > self.max_object_bytes {
4050 return Err(ControlError::ResourceExhausted(format!(
4051 "object {oid} is {object_bytes} bytes, over this server's {}-byte ceiling \
4052 for reading an object. Reading any part of it costs the whole object in \
4053 memory, because a prefix of a delta-encoded object needs its entire base \
4054 chain, so no max_bytes makes this request cheaper.",
4055 self.max_object_bytes
4056 )));
4057 }
4058 let response_bytes = if max_bytes == 0 {
4061 object_bytes
4062 } else {
4063 object_bytes.min(max_bytes)
4064 };
4065 if response_bytes > MAX_RESPONSE_BYTES {
4066 return Err(ControlError::ResourceExhausted(format!(
4067 "object {oid} is {object_bytes} bytes and the response would be \
4068 {response_bytes}, over the {MAX_RESPONSE_BYTES}-byte limit a gRPC client \
4069 can receive by default. Ask for a prefix with max_bytes."
4070 )));
4071 }
4072 }
4073
4074 let Some((kind, mut buf)) = git.read(id.as_bytes()).map_err(store_error)? else {
4084 return Err(ControlError::NotFound(format!("object {oid} in {path}")));
4085 };
4086 let size = buf.len() as u64;
4087
4088 let truncated = max_bytes != 0 && size > max_bytes;
4094 if truncated {
4095 buf.truncate(max_bytes as usize);
4096 }
4097 Ok(ReadObjectResponse {
4098 oid: id.to_string(),
4099 kind: object_kind_out(kind)? as i32,
4100 size,
4101 data: buf,
4102 truncated,
4103 })
4104 }
4105
4106 fn read_lfs_object(
4138 &self,
4139 path: &RepoPath,
4140 oid: &str,
4141 max_bytes: u64,
4142 range: Option<(u64, u64)>,
4143 ) -> ControlResult<ReadLfsObjectResponse> {
4144 let (_, store) = self.open(path)?;
4145 let id = gunnar_lfs::Oid::parse(oid)
4146 .map_err(|err| ControlError::InvalidArgument(format!("{oid}: {err}")))?;
4147 use gunnar_lfs::LfsStore as _;
4150
4151 let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir()).map_err(|err| {
4152 ControlError::Internal(format!("opening the repository's LFS store: {err}"))
4153 })?;
4154
4155 let ceiling = match max_bytes {
4156 0 => MAX_RESPONSE_BYTES,
4157 n => n.min(MAX_RESPONSE_BYTES),
4158 };
4159 let size = lfs
4160 .size_of(&id)
4161 .map_err(|err| ControlError::Internal(format!("reading the LFS store: {err}")))?
4162 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4163
4164 if let Some((offset, length)) = range {
4171 if offset > size || (offset == size && size > 0) {
4172 return Err(ControlError::InvalidArgument(format!(
4173 "offset {offset} is past the end of LFS object {oid}, which is {size} bytes"
4174 )));
4175 }
4176 if length > ceiling {
4177 return Err(ControlError::ResourceExhausted(format!(
4178 "a range of {length} bytes of LFS object {oid} is over the {ceiling}-byte ceiling \
4179 for this request; ask for {ceiling} bytes or fewer at a time"
4180 )));
4181 }
4182 let want = if length == 0 { ceiling } else { length }.min(size - offset);
4183 let object = lfs
4184 .open_at(&id, offset)
4185 .map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
4186 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4187 let mut data = Vec::with_capacity(want as usize);
4188 let mut reader = std::io::Read::take(object.reader, want);
4189 std::io::Read::read_to_end(&mut reader, &mut data).map_err(|err| {
4190 ControlError::Internal(format!("reading LFS object {oid}: {err}"))
4191 })?;
4192 if (data.len() as u64) < want {
4193 return Err(ControlError::Internal(format!(
4194 "LFS object {oid} ended {} bytes into a {want}-byte range at {offset}",
4195 data.len()
4196 )));
4197 }
4198 return Ok(ReadLfsObjectResponse {
4199 oid: oid.to_owned(),
4200 size: object.size,
4201 data,
4202 offset,
4203 });
4204 }
4205
4206 if size > ceiling {
4207 return Err(ControlError::ResourceExhausted(format!(
4208 "LFS object {oid} is {size} bytes, over the {ceiling}-byte ceiling for this \
4209 request. An LFS object is served whole or not at all — a prefix of one is not a \
4210 smaller file — so no max_bytes makes this request smaller. `git lfs` streams it, \
4211 and a ranged read (offset/length) reads it a range at a time."
4212 )));
4213 }
4214
4215 let object = lfs
4216 .open(&id)
4217 .map_err(|err| ControlError::Internal(format!("opening LFS object {oid}: {err}")))?
4218 .ok_or_else(|| ControlError::NotFound(format!("LFS object {oid} in {path}")))?;
4219 if object.size > ceiling {
4224 return Err(ControlError::ResourceExhausted(format!(
4225 "LFS object {oid} is {} bytes, over the {ceiling}-byte ceiling for this request",
4226 object.size
4227 )));
4228 }
4229 let mut data = Vec::with_capacity(object.size as usize);
4230 let mut reader = std::io::Read::take(object.reader, object.size);
4231 std::io::Read::read_to_end(&mut reader, &mut data)
4232 .map_err(|err| ControlError::Internal(format!("reading LFS object {oid}: {err}")))?;
4233
4234 Ok(ReadLfsObjectResponse {
4235 oid: oid.to_owned(),
4236 size: data.len() as u64,
4237 data,
4238 offset: 0,
4239 })
4240 }
4241
4242 fn diff(
4251 &self,
4252 path: &RepoPath,
4253 from: &str,
4254 to: &str,
4255 path_prefixes: &[String],
4256 ) -> ControlResult<Vec<FileChange>> {
4257 let (_repo, store) = self.open(path)?;
4260 let git = store.serve();
4261 let hash = store.hash_kind();
4262
4263 let resolve = |spec: &str| -> ControlResult<Option<gix_hash::ObjectId>> {
4267 if spec.is_empty() {
4268 return Ok(None);
4269 }
4270 Ok(Some(self.tree_of(git, hash, path, spec)?))
4271 };
4272 let old = resolve(from)?;
4273 let new = resolve(to)?;
4274
4275 crate::diff::compare(git, hash, old, new, path_prefixes).map_err(|err| match err {
4276 crate::diff::DiffError::Missing(oid) => {
4279 ControlError::NotFound(format!("object {oid} in {path}"))
4280 }
4281 crate::diff::DiffError::WrongKind { .. } => {
4282 ControlError::InvalidArgument(err.to_string())
4283 }
4284 crate::diff::DiffError::Corrupt { .. } | crate::diff::DiffError::Backend(_) => {
4285 ControlError::Internal(err.to_string())
4286 }
4287 })
4288 }
4289
4290 fn add_principal(
4293 &self,
4294 id: &str,
4295 name: &str,
4296 kind: gunnar_proto::PrincipalKind,
4297 ) -> ControlResult<Principal> {
4298 let principal =
4309 gunnar_core::Principal::new(id, name, principal_kind_in(kind)).map_err(internal)?;
4310 let created = self
4311 .stores
4312 .auth()
4313 .add_principal(principal)
4314 .map(|p| principal_out(&p))
4315 .map_err(internal)?;
4316 if let Err(err) = self.stores.catalog().create_namespace(id) {
4330 let _ = self.stores.auth().remove_principal(id);
4331 return Err(ControlError::AlreadyExists(format!(
4332 "cannot create the account namespace \"{id}\" for this user: {err}. A user's \
4333 account is the namespace named after their id and cannot be somebody else's; \
4334 the principal was not created."
4335 )));
4336 }
4337 self.stores.mirror_identity();
4341 Ok(created)
4342 }
4343
4344 fn enroll_employee(
4345 &self,
4346 account: &str,
4347 ask: &gunnar_grpc::EmployeeAsk,
4348 ) -> ControlResult<gunnar_grpc::EmployeeAnswer> {
4349 if self
4350 .stores
4351 .catalog()
4352 .namespace(account)
4353 .map_err(internal)?
4354 .is_none()
4355 {
4356 return Err(ControlError::NotFound(format!("account {account}")));
4357 }
4358 if self.stores.auth().is_clan(account).map_err(internal)? {
4361 return Err(ControlError::FailedPrecondition(format!(
4362 "{account} is a clan, not an account: employees are enrolled by an account's admin, \
4363 and a clan's members are accounts that join it"
4364 )));
4365 }
4366 let principal = if ask.seats == 0 {
4367 if self
4371 .stores
4372 .catalog()
4373 .namespace(&ask.id)
4374 .map_err(internal)?
4375 .is_some()
4376 {
4377 return Err(ControlError::AlreadyExists(format!(
4378 "namespace {} exists, so an employee may not be called that",
4379 ask.id
4380 )));
4381 }
4382 Some(
4383 gunnar_core::Principal::new(&ask.id, &ask.name, gunnar_core::PrincipalKind::User)
4384 .map_err(internal)?,
4385 )
4386 } else {
4387 None
4388 };
4389 let room = gunnar_core::RoomAsk {
4390 room_bytes: ask.room_bytes,
4391 seats: ask.seats,
4392 from_seat: ask.from_seat,
4393 cap: ask.cap,
4394 stored: ask.stored,
4395 order: ask.order.clone(),
4396 dry_run: ask.dry_run,
4397 };
4398 let answer = match self
4399 .stores
4400 .auth()
4401 .enroll_with_room(account, principal, &room)
4402 {
4403 Ok(answer) => answer,
4404 Err(gunnar_core::Error::RoomShort {
4407 need,
4408 unused,
4409 short,
4410 ..
4411 }) => {
4412 let (seats, seats_open) = self.stores.auth().seats_of(account).map_err(internal)?;
4413 return Ok(gunnar_grpc::EmployeeAnswer {
4414 principal: None,
4415 room_bytes: ask.room_bytes,
4416 seats,
4417 seats_open,
4418 need_bytes: need,
4419 unused_bytes: Some(unused),
4420 shortfall_bytes: short,
4421 unchanged: true,
4422 });
4423 }
4424 Err(gunnar_core::Error::PrincipalExists(id)) => {
4425 return Err(ControlError::AlreadyExists(format!(
4426 "principal {id} already exists, and an existing principal can never become an \
4427 employee"
4428 )))
4429 }
4430 Err(err @ gunnar_core::Error::NoOpenSeat(_)) => {
4431 return Err(ControlError::FailedPrecondition(err.to_string()))
4432 }
4433 Err(err @ gunnar_core::Error::NameExists { .. }) => {
4434 return Err(ControlError::AlreadyExists(err.to_string()))
4435 }
4436 Err(err @ gunnar_core::Error::InvalidInput { .. }) => {
4437 return Err(ControlError::InvalidArgument(err.to_string()))
4438 }
4439 Err(other) => return Err(internal(other)),
4440 };
4441 if !answer.unchanged {
4442 self.stores.mirror_identity();
4443 }
4444 Ok(gunnar_grpc::EmployeeAnswer {
4445 principal: answer.principal.as_ref().map(principal_out),
4446 room_bytes: answer.room_bytes,
4447 seats: answer.seats,
4448 seats_open: answer.seats_open,
4449 need_bytes: answer.need_bytes,
4450 unused_bytes: answer.unused_bytes,
4451 shortfall_bytes: answer.shortfall_bytes,
4452 unchanged: answer.unchanged,
4453 })
4454 }
4455
4456 fn seats(&self, account: &str) -> ControlResult<(u32, u32)> {
4457 self.stores.auth().seats_of(account).map_err(internal)
4458 }
4459
4460 fn clans(&self) -> Option<&dyn gunnar_grpc::clans::ClanPlane> {
4461 Some(self)
4462 }
4463
4464 fn has_employees(&self, account: &str) -> ControlResult<bool> {
4465 self.stores.auth().is_enterprise(account).map_err(internal)
4466 }
4467
4468 fn employees(&self, account: &str) -> ControlResult<Vec<Principal>> {
4469 Ok(self
4470 .stores
4471 .auth()
4472 .employees_of(account)
4473 .map_err(internal)?
4474 .iter()
4475 .map(principal_out)
4476 .collect())
4477 }
4478
4479 fn principal_by_name(&self, name: &str) -> ControlResult<Option<Principal>> {
4480 Ok(self
4481 .stores
4482 .auth()
4483 .principal_by_name(name)
4484 .map_err(internal)?
4485 .as_ref()
4486 .map(principal_out))
4487 }
4488
4489 fn operator_bootstrap_invite(
4500 &self,
4501 operator: &str,
4502 secret_sha256: &str,
4503 now_unix_ms: i64,
4504 ttl_ms: i64,
4505 ) -> ControlResult<gunnar_grpc::BootstrapInvite> {
4506 use gunnar_grpc::seam::OPERATOR_INVITE_ID;
4507 let hash = gunnar_grpc::seam::bootstrap_hash(secret_sha256)?;
4508 let ttl_ms = gunnar_grpc::seam::bootstrap_ttl(ttl_ms)?;
4509 let auth = self.stores.auth();
4510 if !auth
4511 .web_credentials(Some(operator))
4512 .map_err(internal)?
4513 .is_empty()
4514 {
4515 if auth
4516 .drop_first_invite(OPERATOR_INVITE_ID)
4517 .map_err(internal)?
4518 .is_some()
4519 {
4520 self.stores.mirror_identity();
4521 }
4522 return Ok(gunnar_grpc::BootstrapInvite::Spent);
4523 }
4524 let renewed = auth
4525 .first_invite(OPERATOR_INVITE_ID)
4526 .map_err(internal)?
4527 .is_some_and(|row| row.secret_hash == hash && row.principal_id == operator);
4528 auth.put_first_invite(gunnar_core::FirstInviteRow {
4529 id: OPERATOR_INVITE_ID.to_owned(),
4530 principal_id: operator.to_owned(),
4531 secret_hash: hash,
4532 created_by: "installer".to_owned(),
4533 created: now_unix_ms,
4534 expires: now_unix_ms + ttl_ms,
4535 note: "the operator's first passkey: the installer's first invite".to_owned(),
4536 server_level: None,
4537 })
4538 .map_err(internal)?;
4539 self.stores.mirror_identity();
4540 if !renewed {
4541 tracing::warn!(
4542 target: "gunnar::audit",
4543 event = "operator.bootstrap_invite",
4544 operator,
4545 ttl_days = ttl_ms / 86_400_000,
4546 "the installer's first invite (its hash) is recorded for the operator: his first passkey registers on his account, by gunnar's word"
4547 );
4548 }
4549 Ok(if renewed {
4550 gunnar_grpc::BootstrapInvite::Renewed
4551 } else {
4552 gunnar_grpc::BootstrapInvite::Recorded
4553 })
4554 }
4555
4556 fn claim_operator(&self, asked: &str) -> ControlResult<gunnar_grpc::OperatorClaim> {
4557 #[cfg(feature = "wg-client")]
4558 let fed = crate::fed_principals::fed_operator(self.stores.root());
4559 #[cfg(not(feature = "wg-client"))]
4560 let fed: Option<String> = None;
4561 self.claim_operator_named(fed.as_deref(), asked)
4562 }
4563
4564 fn rename_principal(&self, principal_id: &str, name: &str) -> ControlResult<Principal> {
4565 let renamed = self
4569 .stores
4570 .auth()
4571 .rename_principal(principal_id, name)
4572 .map(|p| principal_out(&p))
4573 .map_err(internal)?;
4574 self.stores.mirror_identity();
4575 Ok(renamed)
4576 }
4577
4578 fn list_principals(&self) -> ControlResult<Vec<Principal>> {
4579 Ok(self
4580 .stores
4581 .auth()
4582 .principals()
4583 .map_err(internal)?
4584 .iter()
4585 .map(principal_out)
4586 .collect())
4587 }
4588
4589 fn principal_activity(
4590 &self,
4591 ) -> ControlResult<std::collections::BTreeMap<String, gunnar_grpc::PrincipalSeen>> {
4592 Ok(self
4593 .stores
4594 .activity()
4595 .all()
4596 .into_iter()
4597 .map(|(id, seen)| {
4598 (
4599 id,
4600 gunnar_grpc::PrincipalSeen {
4601 last_sign_in_unix_ms: seen.signed_in,
4602 last_operation_unix_ms: seen.operated,
4603 },
4604 )
4605 })
4606 .collect())
4607 }
4608
4609 fn holds_server_grant(&self, principal_id: &str) -> ControlResult<bool> {
4610 self.stores
4611 .auth()
4612 .holds_server_grant(principal_id)
4613 .map_err(internal)
4614 }
4615
4616 fn add_credential(
4617 &self,
4618 principal_id: &str,
4619 credential_id: &[u8],
4620 public_key: &[u8],
4621 label: &str,
4622 sign_count: u32,
4623 ) -> ControlResult<gunnar_proto::WebCredential> {
4624 let stored = self
4625 .stores
4626 .auth()
4627 .add_web_credential(gunnar_core::WebCredential {
4628 id: gunnar_grpc::webauthn::b64url(credential_id),
4629 principal_id: principal_id.to_owned(),
4630 label: label.to_owned(),
4631 public_key_hex: hex::encode(public_key),
4632 sign_count,
4633 created: gunnar_core::now_ms(),
4634 last_used: None,
4635 sessions: Vec::new(),
4636 })
4637 .map_err(internal)?;
4638 web_credential_out(&stored)
4639 }
4640
4641 fn credentials(
4642 &self,
4643 principal_id: Option<&str>,
4644 ) -> ControlResult<Vec<gunnar_proto::WebCredential>> {
4645 self.stores
4646 .auth()
4647 .web_credentials(principal_id)
4648 .map_err(internal)?
4649 .iter()
4650 .map(web_credential_out)
4651 .collect()
4652 }
4653
4654 fn credential(
4655 &self,
4656 credential_id: &[u8],
4657 ) -> ControlResult<Option<gunnar_grpc::StoredCredential>> {
4658 let Some(row) = self
4659 .stores
4660 .auth()
4661 .web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
4662 .map_err(internal)?
4663 else {
4664 return Ok(None);
4665 };
4666 let public_key = hex::decode(&row.public_key_hex).map_err(|err| {
4667 ControlError::Internal(format!(
4668 "web credential {} holds a key that is not hex: {err}",
4669 row.id
4670 ))
4671 })?;
4672 Ok(Some(gunnar_grpc::StoredCredential {
4673 credential: web_credential_out(&row)?,
4674 public_key,
4675 }))
4676 }
4677
4678 fn record_passkey_use(
4679 &self,
4680 credential_id: &[u8],
4681 sign_count: u32,
4682 now_unix_ms: i64,
4683 ) -> ControlResult<gunnar_proto::WebCredential> {
4684 let row = self
4685 .stores
4686 .auth()
4687 .touch_web_credential(
4688 &gunnar_grpc::webauthn::b64url(credential_id),
4689 sign_count,
4690 now_unix_ms,
4691 )
4692 .map_err(internal)?;
4693 self.stores.mirror_identity();
4694 self.stores
4696 .activity()
4697 .signed_in(&row.principal_id, now_unix_ms);
4698 web_credential_out(&row)
4699 }
4700
4701 fn revoke_credential(
4702 &self,
4703 credential_id: &[u8],
4704 ) -> ControlResult<Option<gunnar_proto::WebCredential>> {
4705 let removed = self
4706 .stores
4707 .auth()
4708 .revoke_web_credential(&gunnar_grpc::webauthn::b64url(credential_id))
4709 .map_err(internal)?;
4710 self.stores.mirror_identity();
4712 removed.as_ref().map(web_credential_out).transpose()
4713 }
4714
4715 fn rename_credential(
4716 &self,
4717 credential_id: &[u8],
4718 label: &str,
4719 ) -> ControlResult<Option<gunnar_proto::WebCredential>> {
4720 self.stores
4721 .auth()
4722 .rename_web_credential(&gunnar_grpc::webauthn::b64url(credential_id), label)
4723 .map_err(internal)?
4724 .as_ref()
4725 .map(web_credential_out)
4726 .transpose()
4727 }
4728
4729 fn first_invite(&self) -> ControlResult<Option<gunnar_grpc::FirstInvite>> {
4730 Ok(self
4731 .stores
4732 .auth()
4733 .first_invite(gunnar_grpc::seam::OPERATOR_INVITE_ID)
4734 .map_err(internal)?
4735 .map(|row| gunnar_grpc::FirstInvite {
4736 principal_id: row.principal_id,
4737 secret_hash: row.secret_hash,
4738 created_at_unix_ms: row.created,
4739 expires_at_unix_ms: row.expires,
4740 }))
4741 }
4742
4743 fn drop_first_invite(&self) -> ControlResult<()> {
4744 if self
4745 .stores
4746 .auth()
4747 .drop_first_invite(gunnar_grpc::seam::OPERATOR_INVITE_ID)
4748 .map_err(internal)?
4749 .is_some()
4750 {
4751 self.stores.mirror_identity();
4752 }
4753 Ok(())
4754 }
4755
4756 fn employer_of(&self, principal_id: &str) -> ControlResult<Option<String>> {
4757 self.stores
4758 .auth()
4759 .employer_of(principal_id)
4760 .map_err(internal)
4761 }
4762
4763 fn sso(&self) -> Option<&dyn gunnar_grpc::sso::plane::SsoPlane> {
4764 Some(self)
4765 }
4766
4767 fn sso_import(
4768 &self,
4769 order: gunnar_grpc::sso::scim::ImportOrder,
4770 ) -> ControlResult<gunnar_grpc::sso::scim::ImportJob> {
4771 let account = order.account.clone();
4775 let id = self.sso_import.next_job_id();
4776 let job = self.sso_import.enqueue_as(order, id.clone());
4777 if job.job_id != id {
4778 return Err(ControlError::FailedPrecondition(format!(
4779 "an import for account {account} is already {} (job {}): wait for it to finish",
4780 if job.state == crate::job_queue::JobState::Queued {
4781 "queued"
4782 } else {
4783 "running"
4784 },
4785 job.job_id
4786 )));
4787 }
4788 Ok(crate::sso_import_queue::job_out(job))
4789 }
4790
4791 fn sso_import_status(
4792 &self,
4793 account: &str,
4794 ) -> ControlResult<Option<gunnar_grpc::sso::scim::ImportJob>> {
4795 Ok(self
4796 .sso_import
4797 .status(account)
4798 .map(crate::sso_import_queue::job_out))
4799 }
4800
4801 fn approve_machine_enrolment(
4802 &self,
4803 approval: gunnar_grpc::MachineApproval,
4804 ) -> ControlResult<gunnar_grpc::MachineApproved> {
4805 let step_up = approval
4806 .step_up
4807 .as_ref()
4808 .map(|(id, count)| (gunnar_grpc::webauthn::b64url(id), *count));
4809 let (principal, credential, created) = self
4810 .stores
4811 .auth()
4812 .approve_machine_credential(
4813 &approval.machine,
4814 level_in(approval.server_level)?,
4815 gunnar_core::WebCredential {
4816 id: gunnar_grpc::webauthn::b64url(&approval.credential_id),
4817 principal_id: String::new(),
4818 label: approval.label,
4819 public_key_hex: hex::encode(&approval.public_key),
4820 sign_count: approval.sign_count,
4821 created: approval.now_unix_ms,
4822 last_used: None,
4823 sessions: Vec::new(),
4824 },
4825 step_up.as_ref().map(|(id, count)| (id.as_str(), *count)),
4826 approval.now_unix_ms,
4827 )
4828 .map_err(internal)?;
4829 self.stores.mirror_identity();
4830 Ok(gunnar_grpc::MachineApproved {
4831 principal: principal_out(&principal),
4832 credential: web_credential_out(&credential)?,
4833 created,
4834 })
4835 }
4836
4837 fn remove_principal(&self, principal_id: &str) -> ControlResult<(u32, u32)> {
4838 let before_keys = self.stores.auth().keys().map_err(internal)?.len();
4839 let before_grants = self.stores.auth().grants(None).map_err(internal)?.len();
4840 self.stores
4841 .auth()
4842 .remove_principal(principal_id)
4843 .map_err(internal)?;
4844 let after_keys = self.stores.auth().keys().map_err(internal)?.len();
4845 let after_grants = self.stores.auth().grants(None).map_err(internal)?.len();
4846 self.stores.mirror_identity();
4847 Ok((
4848 before_keys.saturating_sub(after_keys) as u32,
4849 before_grants.saturating_sub(after_grants) as u32,
4850 ))
4851 }
4852
4853 fn add_key(
4854 &self,
4855 principal_id: &str,
4856 openssh_public_key: &str,
4857 _comment: &str,
4858 expires_at_unix_ms: Option<i64>,
4859 caller: &str,
4860 attestation: Option<gunnar_proto::SignedKeyAttestation>,
4861 ) -> ControlResult<PublicKey> {
4862 if openssh_public_key.trim().starts_with("age1") {
4869 return Err(ControlError::InvalidArgument(
4870 gunnar_core::auth::AGE_LINE_IS_NO_SSH_KEY.to_owned(),
4871 ));
4872 }
4873 let adding = gunnar_core::Fingerprint::of_key(
4874 &gunnar_core::auth::parse_public_key(openssh_public_key).map_err(internal)?,
4875 );
4876 let holders = self
4877 .stores
4878 .auth()
4879 .attesting_keys_of(principal_id)
4880 .map_err(internal)?;
4881 if holders.iter().any(|k| k.fingerprint == adding.as_str()) {
4885 return Err(ControlError::AlreadyExists(format!(
4886 "principal {principal_id} already holds key {}",
4887 adding.as_str()
4888 )));
4889 }
4890 let requirement =
4891 gunnar_core::key_attestation::Requirement::of(principal_id, caller, holders);
4892 let attestation = attestation.map(from_proto_attestation).transpose()?;
4893 gunnar_core::key_attestation::check(
4894 &requirement,
4895 principal_id,
4896 adding.as_str(),
4897 attestation.as_ref(),
4898 )
4899 .map_err(|problem| match problem {
4900 gunnar_core::AttestationProblem::Missing { .. } => {
4904 ControlError::FailedPrecondition(problem.to_string())
4905 }
4906 _ => ControlError::InvalidArgument(problem.to_string()),
4908 })?;
4909 let registered = self
4910 .stores
4911 .auth()
4912 .register_key(
4913 principal_id,
4914 openssh_public_key,
4915 expires_at_unix_ms.map(|ms| ms as UnixMillis),
4916 )
4917 .map(|key| key_out(&key))
4918 .map_err(internal)?;
4919 self.stores.mirror_identity();
4920 Ok(registered)
4921 }
4922
4923 fn revoke_key(&self, fingerprint: &str) -> ControlResult<()> {
4924 let parsed = gunnar_core::Fingerprint::parse(fingerprint)
4925 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
4926 let revoked = self.stores.auth().revoke_key(&parsed).map_err(internal)?;
4927 self.stores.mirror_identity();
4928 match revoked {
4929 Some(_) => Ok(()),
4930 None => Err(ControlError::NotFound(format!("key {fingerprint}"))),
4931 }
4932 }
4933
4934 fn list_keys(&self, principal_id: Option<&str>) -> ControlResult<Vec<PublicKey>> {
4935 let keys = match principal_id {
4936 Some(id) => self.stores.auth().keys_of(id).map_err(internal)?,
4937 None => self.stores.auth().keys().map_err(internal)?,
4938 };
4939 Ok(keys.iter().map(key_out).collect())
4940 }
4941
4942 fn grant(
4943 &self,
4944 principal_id: &str,
4945 namespace: &str,
4946 repo: &str,
4947 level: GrantLevel,
4948 ) -> ControlResult<gunnar_proto::Grant> {
4949 let scope = scope_in(namespace, repo)?;
4950 let grant = Grant {
4951 principal: principal_id.to_owned(),
4952 scope,
4953 level: level_in(level)?,
4954 };
4955 self.stores
4956 .auth()
4957 .grant(grant)
4958 .map(|g| grant_out(&g))
4959 .map_err(internal)
4960 }
4961
4962 fn revoke(&self, principal_id: &str, namespace: &str, repo: &str) -> ControlResult<()> {
4963 let scope = scope_in(namespace, repo)?;
4964 match self
4965 .stores
4966 .auth()
4967 .revoke_grant(principal_id, &scope)
4968 .map_err(internal)?
4969 {
4970 true => Ok(()),
4971 false => Err(ControlError::NotFound(format!(
4972 "grant for {principal_id} on {scope}"
4973 ))),
4974 }
4975 }
4976
4977 fn list_grants(
4978 &self,
4979 principal_id: Option<&str>,
4980 namespace: Option<&str>,
4981 repo: Option<&str>,
4982 ) -> ControlResult<Vec<gunnar_proto::Grant>> {
4983 Ok(self
4984 .stores
4985 .auth()
4986 .grants(principal_id)
4987 .map_err(internal)?
4988 .iter()
4989 .map(grant_out)
4990 .filter(|g| namespace.is_none_or(|want| want == g.namespace))
4991 .filter(|g| repo.is_none_or(|want| want == g.repo))
4992 .collect())
4993 }
4994
4995 fn add_group_member(
4998 &self,
4999 group_id: &str,
5000 member_id: &str,
5001 signed: gunnar_proto::SignedMembershipEvent,
5002 ) -> ControlResult<u64> {
5003 let signed = self.verify_membership_event(
5004 signed,
5005 group_id,
5006 member_id,
5007 gunnar_core::MembershipAction::Add,
5008 )?;
5009 self.stores
5013 .auth()
5014 .add_to_group(group_id, member_id)
5015 .map_err(internal)?;
5016 self.append_membership_event(&signed)
5017 }
5018
5019 fn remove_group_member(
5020 &self,
5021 group_id: &str,
5022 member_id: &str,
5023 signed: gunnar_proto::SignedMembershipEvent,
5024 ) -> ControlResult<(bool, u64)> {
5025 let signed = self.verify_membership_event(
5026 signed,
5027 group_id,
5028 member_id,
5029 gunnar_core::MembershipAction::Remove,
5030 )?;
5031 let was_member = self
5032 .stores
5033 .auth()
5034 .remove_from_group(group_id, member_id)
5035 .map_err(internal)?;
5036 let index = self.append_membership_event(&signed)?;
5040 Ok((was_member, index))
5041 }
5042
5043 fn membership_events(
5044 &self,
5045 group_id: Option<&str>,
5046 ) -> ControlResult<Vec<gunnar_proto::SignedMembershipEvent>> {
5047 Ok(self
5048 .membership_log_events()?
5049 .into_iter()
5050 .filter(|signed| group_id.is_none_or(|g| signed.event.group == g))
5051 .map(to_proto_membership)
5052 .collect())
5053 }
5054
5055 fn list_group_members(&self, group_id: &str) -> ControlResult<Vec<Principal>> {
5056 let auth = self.stores.auth();
5059 match auth.principal(group_id).map_err(internal)? {
5060 Some(p) if p.kind == PrincipalKind::Group => {}
5061 Some(_) => {
5062 return Err(ControlError::InvalidArgument(format!(
5063 "principal \"{group_id}\" is not a group"
5064 )))
5065 }
5066 None => return Err(ControlError::NotFound(format!("group {group_id}"))),
5067 }
5068 let mut members = Vec::new();
5071 for id in auth.members_of(group_id).map_err(internal)? {
5072 if let Some(p) = auth.principal(&id).map_err(internal)? {
5073 members.push(principal_out(&p));
5074 }
5075 }
5076 Ok(members)
5077 }
5078
5079 fn get_protection(
5082 &self,
5083 path: &RepoPath,
5084 ref_pattern: Option<&str>,
5085 ) -> ControlResult<Vec<BranchProtection>> {
5086 let (repo, _) = self.open(path)?;
5087 let config = self.policy_of(&repo)?;
5088 Ok(config
5089 .protection
5090 .iter()
5091 .filter(|rule| ref_pattern.is_none_or(|want| rule.pattern.as_str() == want))
5092 .map(protection_out)
5093 .collect())
5094 }
5095
5096 fn set_protection(
5116 &self,
5117 path: &RepoPath,
5118 protection: BranchProtection,
5119 ) -> ControlResult<BranchProtection> {
5120 let (repo, _) = self.open(path)?;
5121 if protection.max_object_bytes != 0 || protection.max_objects_per_push != 0 {
5122 return Err(ControlError::InvalidArgument(
5123 "max_object_bytes and max_objects_per_push are per-rule in the protocol and \
5124 repository-wide in this server's policy engine, so a value here would apply to \
5125 references this rule does not name. Leave them 0 and set the repository's \
5126 limits in its policy document."
5127 .to_owned(),
5128 ));
5129 }
5130 let pattern = gunnar_policy::RefPattern::new(protection.ref_pattern.clone())
5131 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
5132
5133 let mut config = self.policy_of(&repo)?;
5134 let mut rule = config
5137 .protection
5138 .iter()
5139 .find(|rule| rule.pattern == pattern)
5140 .cloned()
5141 .unwrap_or_else(|| {
5142 gunnar_policy::ProtectionRule::protect(pattern.as_str())
5143 .expect("the pattern was compiled one line above")
5144 });
5145 rule.pattern = pattern;
5146 rule.allow_force_push = protection.allow_force_push;
5147 rule.allow_delete = protection.allow_delete;
5148 rule.require_signature = protection.require_signature;
5149 rule.allowed_principals = if protection.allowed_principal_ids.is_empty() {
5153 None
5154 } else {
5155 Some(protection.allowed_principal_ids.clone())
5156 };
5157 rule.forbid_self_approval = protection.forbid_self_approval;
5158 rule.required_approvals = protection.required_approvals;
5159
5160 config
5161 .protection
5162 .retain(|other| other.pattern != rule.pattern);
5163 config.protection.push(rule.clone());
5164 self.store_policy(&repo, &config)?;
5165 Ok(protection_out(&rule))
5166 }
5167
5168 fn delete_protection(&self, path: &RepoPath, ref_pattern: &str) -> ControlResult<()> {
5169 let (repo, _) = self.open(path)?;
5170 let mut config = self.policy_of(&repo)?;
5171 let before = config.protection.len();
5172 config
5173 .protection
5174 .retain(|rule| rule.pattern.as_str() != ref_pattern);
5175 if config.protection.len() == before {
5176 return Err(ControlError::NotFound(format!(
5177 "no protection rule for {ref_pattern} in {path}"
5178 )));
5179 }
5180 self.store_policy(&repo, &config)
5181 }
5182
5183 fn get_ref_name_policy(&self, path: &RepoPath) -> ControlResult<RefNamePolicy> {
5184 let (repo, _) = self.open(path)?;
5185 let config = self.policy_of(&repo)?;
5186 Ok(ref_name_policy_out(&config.ref_names))
5187 }
5188
5189 fn live_gates(&self, path: &RepoPath) -> ControlResult<Option<Vec<String>>> {
5193 use gunnar_policy::Gate as _;
5194 let (repo, _) = self.open(path)?;
5195 let config = self.policy_of(&repo)?;
5196 let mut names: Vec<String> = config.gates().iter().map(|g| g.name().to_owned()).collect();
5197 if repo.encrypted {
5198 names.push("encrypted".to_owned());
5199 }
5200 Ok(Some(names))
5201 }
5202
5203 fn set_ref_name_policy(
5204 &self,
5205 path: &RepoPath,
5206 policy: RefNamePolicy,
5207 ) -> ControlResult<RefNamePolicy> {
5208 let (repo, _) = self.open(path)?;
5209 let compiled = gunnar_policy::RefNamePolicy::new(
5210 policy.allow_patterns.clone(),
5211 policy.deny_patterns.clone(),
5212 )
5213 .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
5214 let mut config = self.policy_of(&repo)?;
5215 config.ref_names = compiled.clone();
5216 self.store_policy(&repo, &config)?;
5217 Ok(ref_name_policy_out(&compiled))
5218 }
5219
5220 fn forge_append(&self, event: gunnar_proto::SignedForgeEvent) -> ControlResult<u64> {
5228 if event.attested.is_some() {
5229 return Err(ControlError::InvalidArgument(
5230 "this event carries an attestation, and an attestation is gunnar's own word: \
5231 only gunnar makes one, for a person's session that stepped up"
5232 .to_owned(),
5233 ));
5234 }
5235 let signed = from_proto_event(event)?;
5236
5237 let key = self
5244 .stores
5245 .auth()
5246 .signing_keys()
5247 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?
5248 .into_iter()
5249 .find(|row| row.fingerprint_sha256 == signed.key_fingerprint)
5250 .ok_or_else(|| {
5251 ControlError::InvalidArgument(format!(
5252 "{} is not registered on this server, so nobody reading this event later \
5253 could check its signature",
5254 signed.key_fingerprint
5255 ))
5256 })?;
5257 if key.principal_id != signed.event.actor {
5264 return Err(ControlError::InvalidArgument(format!(
5265 "this event is attributed to {:?} and is signed by {}, which is {:?}'s key; \
5266 an approval is the actor's own word, so it must carry the actor's own \
5267 signature",
5268 signed.event.actor, signed.key_fingerprint, key.principal_id
5269 )));
5270 }
5271 let public = gunnar_core::ssh_key::PublicKey::from_openssh(&key.openssh)
5272 .map_err(|err| ControlError::Internal(format!("stored key is unreadable: {err}")))?;
5273 gunnar_core::forge::verify_event(&signed, &public)
5274 .map_err(|problem| ControlError::InvalidArgument(format!("{problem}")))?;
5275
5276 self.forge_subject_open(&signed)?;
5277 self.forge_write(&signed)
5278 }
5279
5280 fn forge_append_attested(
5281 &self,
5282 event: gunnar_proto::SignedForgeEvent,
5283 session_stepped_up_at_unix_ms: i64,
5284 ) -> ControlResult<(u64, String)> {
5285 if !event.key_fingerprint.is_empty()
5290 || !event.signature.is_empty()
5291 || event.attested.is_some()
5292 {
5293 return Err(ControlError::InvalidArgument(
5294 "an event gunnar attests arrives unsigned: a signed one is checked against its \
5295 signer's key instead, and an attestation is gunnar's own to make"
5296 .to_owned(),
5297 ));
5298 }
5299 let event = from_proto_event(event)?.event;
5300 if event.actor.is_empty() {
5301 return Err(ControlError::InvalidArgument(
5302 "an attested event must name the person it is attested for".to_owned(),
5303 ));
5304 }
5305 let key = self
5306 .stores
5307 .forge_attest_key()
5308 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
5309 let signed = gunnar_core::forge::attest_event(
5310 &key,
5311 event,
5312 gunnar_core::Attested {
5313 session_stepped_up_at_unix_ms,
5314 },
5315 )
5316 .map_err(|err| ControlError::Internal(format!("signing the attested event: {err}")))?;
5317 self.forge_subject_open(&signed)?;
5318 let words = signed
5319 .attested
5320 .as_ref()
5321 .map(|a| a.words())
5322 .unwrap_or_default();
5323 Ok((self.forge_write(&signed)?, words))
5324 }
5325
5326 fn forge_review(
5327 &self,
5328 repo: &str,
5329 pull: u64,
5330 ) -> ControlResult<(
5331 gunnar_proto::ReviewState,
5332 Vec<gunnar_proto::SignedForgeEvent>,
5333 )> {
5334 let all = self.forge_events()?;
5335 let mine: Vec<gunnar_core::ForgeEvent> = all
5336 .iter()
5337 .filter(|signed| signed.event.repo == repo)
5338 .map(|signed| signed.event.clone())
5339 .collect();
5340 let state = gunnar_core::forge::fold(pull, &mine);
5341 let events = all
5342 .iter()
5343 .filter(|signed| signed.event.repo == repo && signed.event.pull == pull)
5344 .cloned()
5345 .map(to_proto_event)
5346 .collect();
5347 Ok((self.review_state(repo, &state, &all)?, events))
5348 }
5349
5350 fn forge_reviews(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::ReviewState>> {
5351 let all = self.forge_events()?;
5352 let mine: Vec<gunnar_core::ForgeEvent> = all
5353 .iter()
5354 .filter(|signed| signed.event.repo == repo)
5355 .map(|signed| signed.event.clone())
5356 .collect();
5357 let mut numbers: Vec<u64> = mine.iter().map(|event| event.pull).collect();
5358 numbers.sort_unstable();
5359 numbers.dedup();
5360 let mut out = Vec::new();
5361 for pull in numbers {
5362 let state = gunnar_core::forge::fold(pull, &mine);
5363 out.push(self.review_state(repo, &state, &all)?);
5364 }
5365 Ok(out)
5366 }
5367
5368 fn forge_issue(
5369 &self,
5370 repo: &str,
5371 number: u64,
5372 ) -> ControlResult<
5373 Option<(
5374 gunnar_proto::IssueState,
5375 Vec<gunnar_proto::SignedForgeEvent>,
5376 )>,
5377 > {
5378 let all = self.forge_events()?;
5379 let mine: Vec<gunnar_core::ForgeEvent> = all
5380 .iter()
5381 .filter(|signed| signed.event.repo == repo)
5382 .map(|signed| signed.event.clone())
5383 .collect();
5384 let Some(issue) = gunnar_core::forge::fold_issue(number, &mine) else {
5385 return Ok(None);
5386 };
5387 let mut opened_seen = 0u64;
5392 let events = all
5393 .iter()
5394 .filter(|signed| {
5395 if signed.event.repo != repo {
5396 return false;
5397 }
5398 match &signed.event.event {
5399 gunnar_core::Event::IssueOpened { .. } => {
5400 opened_seen += 1;
5401 opened_seen == number
5402 }
5403 gunnar_core::Event::IssueCommented { .. }
5404 | gunnar_core::Event::IssueClosed { .. } => signed.event.pull == number,
5405 _ => false,
5406 }
5407 })
5408 .cloned()
5409 .map(to_proto_event)
5410 .collect();
5411 Ok(Some((issue_state(repo, &issue), events)))
5412 }
5413
5414 fn forge_issues(&self, repo: &str) -> ControlResult<Vec<gunnar_proto::IssueState>> {
5415 let mine: Vec<gunnar_core::ForgeEvent> = self
5416 .forge_events()?
5417 .into_iter()
5418 .filter(|signed| signed.event.repo == repo)
5419 .map(|signed| signed.event)
5420 .collect();
5421 Ok(gunnar_core::forge::fold_issues(&mine)
5422 .iter()
5423 .map(|issue| issue_state(repo, issue))
5424 .collect())
5425 }
5426
5427 fn stats_extras(&self) -> ControlResult<gunnar_grpc::seam::StatsExtras> {
5437 let tombstones = self.stores.tombstones().unwrap_or_default();
5438 let report = self.replication.report();
5439 #[allow(unused_mut)]
5442 let mut tenants = crate::dashboard::TenantCounts::default();
5443 #[cfg(feature = "metered")]
5444 if let Some(slot) = &self.entitlement {
5445 tenants = slot.counts();
5446 }
5447 Ok(gunnar_grpc::seam::StatsExtras {
5448 tombstoned_repos: tombstones.len() as u64,
5449 tombstoned_bytes: tombstones.iter().map(|row| row.bytes).sum(),
5450 replication_configured: report.configured,
5451 replication_enqueued: report.enqueued,
5452 replication_synced: report.synced,
5453 replication_failed: report.failed,
5454 replication_bus_gaps: report.bus_gaps,
5455 replication_refs_total: report.refs.total,
5456 replication_refs_unmirrored: report.refs.unmirrored,
5457 standbys: report
5458 .standbys
5459 .into_iter()
5460 .map(|row| gunnar_proto::StandbyStatus {
5461 verdict: row.verdict().word().to_owned(),
5465 name: row.name,
5466 remote: row.remote,
5467 pending: row.pending,
5468 consecutive_failures: row.consecutive_failures,
5469 last_error: row.last_error,
5470 last_success_unix_ms: row.last_success_unix_ms,
5471 needs_full_resync: row.needs_full_resync,
5472 stale: row.stale,
5473 })
5474 .collect(),
5475 tenants_paid: tenants.paid,
5476 tenants_grace: tenants.grace,
5477 tenants_suspended: tenants.suspended,
5478 })
5479 }
5480
5481 fn counts(&self) -> ControlResult<ControlCounts> {
5482 let (namespaces, repos) = self.stores.catalog().counts().map_err(internal)?;
5483 let (principals, keys, grants) = self.stores.auth().counts().map_err(internal)?;
5484 Ok(ControlCounts {
5485 namespaces: namespaces as u64,
5486 repos: repos as u64,
5487 principals: principals as u64,
5488 keys: keys as u64,
5489 grants: grants as u64,
5490 })
5491 }
5492
5493 fn verify_chain(
5494 &self,
5495 request: &gunnar_proto::VerifyChainRequest,
5496 ) -> ControlResult<gunnar_proto::VerifyChainResponse> {
5497 let target = match request.target.as_str() {
5501 "" => None,
5502 spec => Some(spec),
5503 };
5504 let Some(repo) = &request.repo else {
5505 return Err(ControlError::InvalidArgument(
5506 "name a repository".to_owned(),
5507 ));
5508 };
5509 let rev = match request.rev.as_str() {
5510 "" => "HEAD",
5511 rev => rev,
5512 };
5513 let chain = crate::verify::verify(&self.stores, self, repo, rev, target)
5514 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5519 Ok(chain.to_proto())
5520 }
5521
5522 fn blame(
5523 &self,
5524 path: &RepoPath,
5525 rev: &str,
5526 file: &str,
5527 ) -> ControlResult<gunnar_proto::BlameResponse> {
5528 let rev = if rev.is_empty() { "HEAD" } else { rev };
5535 let rows = crate::blame::blame(&self.stores, self, path, rev, file)
5536 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5537 Ok(gunnar_proto::BlameResponse {
5538 lines: rows
5539 .into_iter()
5540 .map(|row| gunnar_proto::BlameLine {
5541 commit: row.commit.to_hex().to_string(),
5542 author: row.author,
5543 line: row.line,
5544 })
5545 .collect(),
5546 })
5547 }
5548
5549 fn search(
5550 &self,
5551 path: &RepoPath,
5552 rev: &str,
5553 pattern: &str,
5554 max_file_bytes: u64,
5555 max_matches: u32,
5556 ) -> ControlResult<gunnar_proto::SearchResponse> {
5557 let rev = if rev.is_empty() { "HEAD" } else { rev };
5561 let defaults = crate::search::Limits::default();
5562 let limits = crate::search::Limits {
5563 max_file_bytes: if max_file_bytes == 0 {
5564 defaults.max_file_bytes
5565 } else {
5566 max_file_bytes
5567 },
5568 max_matches: if max_matches == 0 {
5569 defaults.max_matches
5570 } else {
5571 max_matches as usize
5572 },
5573 };
5574 let hits = crate::search::search(&self.stores, self, path, rev, pattern, limits)
5575 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
5576 Ok(gunnar_proto::SearchResponse {
5577 matches: hits
5578 .into_iter()
5579 .map(|hit| gunnar_proto::SearchMatch {
5580 path: hit.path,
5581 line_number: hit.line_number as u32,
5582 line: hit.line,
5583 })
5584 .collect(),
5585 })
5586 }
5587
5588 fn issuer_info(&self) -> ControlResult<gunnar_proto::IssuerInfoResponse> {
5602 let pki = self.stores.pki_readonly().map_err(|err| {
5603 ControlError::Unavailable(format!("the PKI document could not be opened: {err:#}"))
5604 })?;
5605 crate::admin::issuer_document(&pki).map_err(|err| {
5609 ControlError::Unavailable(format!("the PKI document could not be read: {err:#}"))
5610 })
5611 }
5612
5613 fn setting(&self, name: &str) -> ControlResult<Option<Vec<u8>>> {
5614 self.stores.settings().get(name).map_err(internal)
5615 }
5616
5617 fn set_setting(&self, name: &str, value: &[u8]) -> ControlResult<Option<Vec<u8>>> {
5618 self.stores.settings().set(name, value).map_err(internal)
5619 }
5620
5621 fn audit_trail(&self, at_seq: u64) -> ControlResult<gunnar_proto::AuditTrailResponse> {
5637 let path = self.stores.audit_path();
5638 let replicated = !self.stores.is_twin()
5643 && !crate::twin::is_restoring(self.stores.root())
5644 && gunnar_core::txlog::sink::log_for(&path)
5645 .is_some_and(|(_, rel)| rel == gunnar_core::txlog::AUDIT_REL);
5646 let mut out = gunnar_proto::AuditTrailResponse {
5647 present: false,
5648 entries: 0,
5649 tip: String::new(),
5650 verified: false,
5651 replicated,
5652 hash_at_seq: String::new(),
5653 why: String::new(),
5654 tip_seq: 0,
5655 };
5656 let chain = match gunnar_core::AuditChain::open(&path) {
5657 Ok(chain) => chain,
5658 Err(err) => {
5659 out.why = format!("{err}");
5660 return Ok(out);
5661 }
5662 };
5663 let entries = match chain.entries() {
5664 Ok(entries) => entries,
5665 Err(err) => {
5666 out.why = format!("{err}");
5667 return Ok(out);
5668 }
5669 };
5670 out.present = true;
5671 out.entries = entries.len() as u64;
5672 if let Some(last) = entries.last() {
5673 out.tip.clone_from(&last.hash);
5674 out.tip_seq = last.seq;
5676 }
5677 match chain.verify() {
5678 Ok(_) => out.verified = true,
5679 Err(err) => out.why = format!("{err}"),
5680 }
5681 if at_seq != 0 {
5682 match entries.iter().find(|e| e.seq == at_seq) {
5687 Some(e) => out.hash_at_seq = e.hash.clone(),
5688 None => {
5689 let sep = if out.why.is_empty() { "" } else { "; " };
5690 out.why = format!(
5691 "{}{sep}this trail holds no entry at seq {at_seq} (it holds {} entries)",
5692 out.why,
5693 entries.len()
5694 );
5695 }
5696 }
5697 }
5698 Ok(out)
5699 }
5700
5701 fn storage_info(&self, path: Option<&RepoPath>) -> ControlResult<StorageInfoResponse> {
5702 let repos = match path {
5703 Some(path) => vec![self.open(path)?.0],
5704 None => self.stores.catalog().list(None).map_err(internal)?,
5705 };
5706 let mut total_hot = 0u64;
5707 let mut total_explode = 0u64;
5708 let mut total_budget = 0u64;
5709 let bands: std::collections::HashMap<std::path::PathBuf, crate::storage::ExplodeBand> =
5715 self.stores.engine().explode_use().into_iter().collect();
5716 let mut out = Vec::new();
5717 for repo in repos {
5718 let store = self.stores.repo_store(&repo).map_err(|err| {
5719 ControlError::Internal(format!("opening the repository's stores: {err:#}"))
5720 })?;
5721 let (loose, _objects_dir) = object_dir_stats(store.git_dir());
5722 let bytes = repo_bytes(store.git_dir()).hot();
5730 let refs = store.serve().refs().map_err(store_error)?.len() as u64;
5731 total_hot += bytes;
5732 let band = bands.get(store.git_dir()).copied().unwrap_or_default();
5736 total_explode += band.used();
5737 total_budget += band.budget;
5738 out.push(RepoStorage {
5739 repo: Some(RepoPath::new(&repo.namespace, &repo.name)),
5740 hash_kind: hash_kind(repo.hash_kind) as i32,
5741 loose_object_count: loose,
5742 hot_bytes: bytes,
5743 cold_bytes: 0,
5749 cold_object_count: 0,
5750 cold_measured: false,
5751 last_repack_at_unix_ms: 0,
5752 ref_count: refs,
5753 explode_bytes: band.used(),
5754 explode_budget_bytes: band.budget,
5755 explode_evicted_bytes: band.evicted_bytes,
5756 });
5757 }
5758 let engine = self.stores.engine();
5768 let (object_backend, ref_backend) = (engine.describe(), engine.describe());
5769 let tombstones = self.stores.tombstones().unwrap_or_default();
5774 let total_tombstoned_bytes = tombstones.iter().map(|row| row.bytes).sum();
5775 let tombstoned = tombstones
5776 .into_iter()
5777 .map(|row| gunnar_proto::TombstonedRepo {
5778 repo: Some(RepoPath {
5779 namespace: row.path.namespace().to_owned(),
5780 repo: row.path.name().to_owned(),
5781 }),
5782 bytes: row.bytes,
5783 deleted_at_unix_ms: row.tombstone.deleted_at_unix_ms,
5784 purge_after_unix_ms: row.tombstone.purge_after_unix_ms.unwrap_or_default(),
5785 })
5786 .collect();
5787 Ok(StorageInfoResponse {
5788 object_backend,
5789 ref_backend,
5790 engine: Some(engine.report()),
5792 total_hot_bytes: total_hot,
5793 total_cold_bytes: 0,
5795 cold_measured: false,
5796 repos: out,
5797 total_explode_bytes: total_explode,
5798 total_explode_budget_bytes: total_budget,
5799 total_tombstoned_bytes,
5800 tombstoned,
5801 reality: Some(self.stores.storage_reality()),
5806 })
5807 }
5808
5809 fn capacity(
5835 &self,
5836 request: &gunnar_proto::CapacityRequest,
5837 ) -> ControlResult<gunnar_proto::CapacityResponse> {
5838 let Some(gate) = self.headroom.as_ref() else {
5839 return Err(ControlError::Unimplemented(
5840 "Admin.Capacity (no disk-headroom gate is wired on this control plane, so the \
5841 --disk-floor in force is unknown and no quote can be made against it)",
5842 ));
5843 };
5844 let held_bytes = if request.namespace.is_empty() {
5849 0
5850 } else {
5851 self.usage(Some(&request.namespace))?
5852 .first()
5853 .map(|row| row.pack_bytes)
5854 .unwrap_or(0)
5855 };
5856 let root = gate.root().to_path_buf();
5857 let mut answer = crate::capacity::answer(&crate::capacity::Question {
5858 room: gate.headroom().map_err(|err| err.to_string()),
5859 floor_spec: gate.floor().to_string(),
5860 requested_bytes: request.requested_bytes,
5861 held_bytes,
5862 namespace: request.namespace.clone(),
5863 growth: crate::capacity::GrowthFacts::measure(&root),
5864 machine: crate::capacity::MachineFacts::measure(),
5865 root: root.clone(),
5866 });
5867 if request.disks {
5870 let slow = self.stores.slow();
5871 let serialdat = std::path::PathBuf::from(slow.serialdat());
5872 let mut disks = crate::capacity::measure_disks(&root, Some(&serialdat), gate.floor());
5873 let log = slow.path_of(crate::slow::LOG_DIR);
5877 let known: Vec<(&str, std::path::PathBuf)> = [
5878 ("metadata copy", slow.path_of(crate::slow::META_DIR)),
5879 ("crash record", slow.path_of(crate::slow::CRASH_DIR)),
5880 ("audit trail", log.join("audit-chain.jsonl")),
5881 ("logs", log),
5882 ("grow journal", serialdat.join("grow-journal")),
5883 ("repository mirror", serialdat.join(crate::slow::REPOS_DIR)),
5884 ]
5885 .into_iter()
5886 .filter(|(_, path)| path.starts_with(&serialdat))
5887 .collect();
5888 let used = disks
5889 .mounts
5890 .iter()
5891 .find(|m| m.labels.iter().any(|l| l == "serialdat") && m.measured)
5892 .map(|m| m.used_bytes);
5893 if disks
5894 .mounts
5895 .iter()
5896 .any(|m| m.labels.iter().any(|l| l == "serialdat"))
5897 {
5898 disks.serialdat = crate::capacity::serialdat_breakdown(&known, used);
5899 }
5900 answer.disks = Some(disks);
5901 }
5902 Ok(answer)
5903 }
5904
5905 fn grow_start(
5908 &self,
5909 request: &gunnar_proto::GrowStartRequest,
5910 ) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5911 self.grow_machine()?
5912 .start(request)
5913 .map_err(ControlError::FailedPrecondition)
5914 }
5915
5916 fn grow_status(&self) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5917 match &self.grow {
5918 Some(grow) => Ok(grow.status()),
5919 None => Ok(gunnar_proto::GrowStatusResponse {
5923 phase: "idle".to_owned(),
5924 engine_present: false,
5925 serving: true,
5926 server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
5927 detail: "this control plane has no runtime growth wired in".to_owned(),
5928 hotplug: crate::dashboard::probe::pci_hotplug().present,
5929 journal: gunnar_grpc::grow_journal::Summary::measure().map(|s| s.to_wire()),
5930 ..Default::default()
5931 }),
5932 }
5933 }
5934
5935 fn grow_resume(&self, give_up: bool) -> ControlResult<gunnar_proto::GrowStatusResponse> {
5936 self.grow_machine()?
5937 .resume(give_up)
5938 .map_err(ControlError::FailedPrecondition)
5939 }
5940
5941 fn usage(&self, namespace: Option<&str>) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
5957 self.usage_readable()?;
5962 if let Some(name) = namespace {
5970 let measured_at_unix_ms = gunnar_core::now_ms();
5971 let Some((held, _)) = self
5972 .stores
5973 .usage_ledger()
5974 .exact(&self.stores, name)
5975 .map_err(internal)?
5976 else {
5977 return Err(ControlError::NotFound(format!(
5978 "namespace {name} is not in the catalog"
5979 )));
5980 };
5981 let objects = self.open_objects_of(name);
5982 return Ok(vec![self.usage_row(
5983 name,
5984 &held,
5985 objects,
5986 measured_at_unix_ms,
5987 )]);
5988 }
5989 self.usage_all()
5990 }
5991
5992 fn repo_usage(&self, namespace: &str) -> ControlResult<Vec<gunnar_proto::RepoStored>> {
5995 let Some((_, repos)) = self
5996 .stores
5997 .usage_ledger()
5998 .exact(&self.stores, namespace)
5999 .map_err(internal)?
6000 else {
6001 return Err(ControlError::NotFound(format!(
6002 "namespace {namespace} is not in the catalog"
6003 )));
6004 };
6005 let open_objects: std::collections::HashMap<std::path::PathBuf, u64> = self
6006 .stores
6007 .engine()
6008 .open_object_counts()
6009 .into_iter()
6010 .collect();
6011 Ok(repos
6013 .into_iter()
6014 .map(|(name, bytes, dir)| gunnar_proto::RepoStored {
6015 repo: Some(gunnar_proto::RepoPath::new(namespace, &name)),
6016 stored_bytes: bytes.pack,
6017 cache_bytes: bytes.cache,
6018 lfs_bytes: bytes.lfs,
6019 objects: dir.and_then(|d| open_objects.get(&d).copied()).unwrap_or(0),
6020 })
6021 .collect())
6022 }
6023
6024 fn usage_page(
6029 &self,
6030 page_token: &str,
6031 page_size: usize,
6032 changed_since: &str,
6033 ) -> ControlResult<gunnar_proto::ReadManyUsageResponse> {
6034 self.usage_readable()?;
6035 let now = gunnar_core::now_ms();
6036 let ledger = self.stores.usage_ledger();
6037 let (cursor, after) = gunnar_grpc::seam::usage_page_token::decode(page_token);
6038 let cursor = if cursor.is_empty() {
6042 ledger.cursor_now(now)
6043 } else {
6044 cursor
6045 };
6046 let (rows, more, changed_only) = ledger
6047 .page(&self.stores, &after, page_size, changed_since, now)
6048 .map_err(internal)?;
6049 let next_page_token = match (more, rows.last()) {
6050 (true, Some((last, _))) => gunnar_grpc::seam::usage_page_token::encode(&cursor, last),
6051 _ => String::new(),
6052 };
6053 let usages = rows
6054 .iter()
6055 .map(|(name, held)| self.usage_row(name, held, held.objects, now))
6056 .collect();
6057 Ok(gunnar_proto::ReadManyUsageResponse {
6058 usages,
6059 next_page_token,
6060 cursor,
6061 changed_only,
6062 as_of_unix_ms: now,
6063 })
6064 }
6065
6066 fn activity(
6086 &self,
6087 namespace: &str,
6088 since_unix_ms: i64,
6089 until_unix_ms: i64,
6090 ) -> ControlResult<(Vec<gunnar_proto::DayActivity>, i64)> {
6091 if self
6092 .stores
6093 .catalog()
6094 .namespace(namespace)
6095 .map_err(internal)?
6096 .is_none()
6097 {
6098 return Err(ControlError::NotFound(format!(
6099 "namespace {namespace} is not in the catalog"
6100 )));
6101 }
6102 let now = gunnar_core::now_ms();
6103 let flow = self.stores.flow();
6104 if let Err(err) = flow.flush(now) {
6105 tracing::warn!(
6109 target: "gunnar::store",
6110 event = "flow.flush_failed",
6111 %err,
6112 "could not flush the flow ledger before an Activity read; the series is as of \
6113 the last successful flush"
6114 );
6115 }
6116 let today = gunnar_core::day_of(now);
6117 let from = if since_unix_ms == 0 {
6118 today - gunnar_core::RETENTION_DAYS
6119 } else {
6120 gunnar_core::day_of(since_unix_ms)
6121 }
6122 .max(today - gunnar_core::RETENTION_DAYS);
6123 let to = if until_unix_ms == 0 {
6124 today
6125 } else {
6126 gunnar_core::day_of(until_unix_ms)
6127 }
6128 .min(today);
6129 if from > to {
6130 return Ok((Vec::new(), gunnar_core::RETENTION_DAYS));
6131 }
6132 let days = flow
6133 .store()
6134 .series(namespace, from, to)
6135 .map_err(internal)?
6136 .into_iter()
6137 .map(|(day, row)| day_activity(day, &row))
6138 .collect();
6139 Ok((days, gunnar_core::RETENTION_DAYS))
6140 }
6141
6142 fn admit_plan(&self, namespace: &str) -> ControlResult<gunnar_proto::AdmitPlanResponse> {
6148 use gunnar_ssh::edda_admit::{Consumer, Work};
6149 use gunnar_ssh::governor::OpClass;
6150 if self
6151 .stores
6152 .catalog()
6153 .namespace(namespace)
6154 .map_err(internal)?
6155 .is_none()
6156 {
6157 return Err(ControlError::NotFound(format!(
6158 "namespace {namespace} is not in the catalog"
6159 )));
6160 }
6161 let Some(gate) = &self.admission else {
6162 return Err(ControlError::Unimplemented("Usage.AdmitPlan"));
6163 };
6164 let policy = gate.admit_policy();
6165 let usage = gate.admit_usage(&*policy, namespace);
6166 let limit = policy.limit();
6167 let who = Consumer::Namespace(namespace.to_owned());
6168 let estimates = policy.estimates();
6169 let kinds = [
6170 OpClass::Clone,
6171 OpClass::Fetch,
6172 OpClass::Push,
6173 OpClass::LsRemote,
6174 ]
6175 .into_iter()
6176 .map(|class| {
6177 let kind = gunnar_ssh::admission::work_kind(class);
6178 let plan = policy.plan(&Work::new(kind, None), &who, &usage);
6179 gunnar_proto::AdmitPlanKind {
6180 kind: kind.name().to_owned(),
6181 admit_class: plan.class.name().to_owned(),
6182 why: plan.why,
6183 estimate_bytes: plan.estimate.usage_bytes(),
6184 estimate_cpu_ms: plan.estimate.cpu_ms,
6185 samples: estimates
6186 .iter()
6187 .find(|row| row.kind == kind)
6188 .map_or(0, |row| row.samples),
6189 }
6190 })
6191 .collect();
6192 Ok(gunnar_proto::AdmitPlanResponse {
6193 policy: policy.policy().to_owned(),
6194 window_days: usage.window_days,
6195 used_bytes: usage.bytes,
6196 limit_bytes: limit.map(|l| l.bytes),
6197 budget_left_bytes: limit.map(|l| l.bytes.saturating_sub(usage.bytes)),
6198 kinds,
6199 })
6200 }
6201
6202 #[cfg(feature = "metered")]
6205 fn entitlement_get(
6206 &self,
6207 namespace: &str,
6208 ) -> ControlResult<Option<gunnar_proto::EntitlementFact>> {
6209 self.entitlement()?
6210 .get(namespace)
6211 .map_err(|err| ControlError::Internal(format!("{err:#}")))
6212 }
6213
6214 #[cfg(feature = "metered")]
6215 fn entitlement_list(&self) -> ControlResult<Vec<gunnar_proto::EntitlementFact>> {
6216 self.entitlement()?
6217 .list()
6218 .map_err(|err| ControlError::Internal(format!("{err:#}")))
6219 }
6220
6221 #[cfg(feature = "metered")]
6222 fn entitlement_refused_at_boot(&self) -> ControlResult<Option<gunnar_proto::RefusedAtBoot>> {
6223 Ok(self.entitlement()?.refusal())
6224 }
6225
6226 #[cfg(feature = "metered")]
6232 fn entitlement_set(
6233 &self,
6234 fact: gunnar_proto::EntitlementFact,
6235 ) -> ControlResult<(gunnar_proto::EntitlementState, u64)> {
6236 use crate::entitlement::SetRefused;
6237 let (state, seq) = self.entitlement()?.set(fact).map_err(|err| match err {
6238 SetRefused::Forged { .. } => ControlError::PermissionDenied(err.to_string()),
6239 SetRefused::Replayed { .. }
6248 | SetRefused::Unstamped { .. }
6249 | SetRefused::Behind { .. } => ControlError::PermissionDenied(err.to_string()),
6250 SetRefused::Malformed(_) => ControlError::InvalidArgument(err.to_string()),
6251 SetRefused::NotRecorded { .. } => ControlError::FailedPrecondition(err.to_string()),
6252 SetRefused::NoKey { .. } => ControlError::FailedPrecondition(err.to_string()),
6253 })?;
6254 Ok((crate::entitlement::state_to_proto(state), seq))
6255 }
6256
6257 #[cfg(feature = "metered")]
6269 fn actor_ticket_public_key_hex(&self) -> ControlResult<String> {
6270 Ok(self
6271 .stores
6272 .actor_ticket_key()
6273 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
6274 .public_key_hex()
6275 .to_owned())
6276 }
6277
6278 #[cfg(feature = "metered")]
6279 fn actor_ticket_sign(&self, message: &[u8]) -> ControlResult<[u8; 64]> {
6280 Ok(self
6281 .stores
6282 .actor_ticket_key()
6283 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
6284 .sign(message))
6285 }
6286
6287 #[cfg(feature = "metered")]
6291 fn sealed_get(&self, name: &str) -> ControlResult<Option<(String, Vec<u8>)>> {
6292 self.stores
6293 .sealed_slot_get(name)
6294 .map_err(|err| ControlError::Internal(format!("{err:#}")))
6295 }
6296
6297 #[cfg(feature = "metered")]
6298 fn sealed_put(&self, name: &str, owner: &str, value: &[u8]) -> ControlResult<bool> {
6299 self.stores
6300 .sealed_slot_put(name, owner, value)
6301 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))
6302 }
6303
6304 fn mirror_add(
6313 &self,
6314 path: &RepoPath,
6315 remote_url: &str,
6316 prune: bool,
6317 create_repo: bool,
6318 ) -> ControlResult<MirrorConfig> {
6319 if remote_url.trim().is_empty() {
6320 return Err(ControlError::InvalidArgument(
6321 "a mirror needs a remote URL".to_owned(),
6322 ));
6323 }
6324 crate::mirror_fetch::classify(remote_url).map_err(ControlError::InvalidArgument)?;
6333 crate::mirror_fetch::check_destination(remote_url, self.stores.root())
6334 .map_err(ControlError::InvalidArgument)?;
6335 let repo_ref = Self::repo_ref(path)?;
6336 let creates = create_repo && self.get_repo(path).is_err();
6346 if creates {
6347 if let Some(line) = self.headroom.as_ref().and_then(|gate| gate.refusal()) {
6350 return Err(ControlError::ResourceExhausted(line));
6351 }
6352 if self
6353 .stores
6354 .catalog()
6355 .namespace(path.namespace.as_str())
6356 .map_err(internal)?
6357 .is_none()
6358 {
6359 return Err(ControlError::NotFound(format!(
6360 "namespace {}",
6361 path.namespace
6362 )));
6363 }
6364 } else {
6365 self.open(path)?;
6369 }
6370 let mut record = gunnar_core::MirrorRecord::new(
6371 &repo_ref,
6372 remote_url,
6373 prune,
6374 gunnar_grpc::bus::now_unix_ms(),
6375 );
6376 record.creates_repo = creates;
6377 let stored = self
6378 .stores
6379 .mirrors()
6380 .set(&repo_ref, record)
6381 .map_err(internal)?;
6382 Ok(mirror_out(&stored))
6383 }
6384
6385 fn mirror_sync(&self, path: &RepoPath) -> ControlResult<MirrorConfig> {
6415 let record = self.mirror_record(path)?;
6416 let repo_ref = Self::repo_ref(path)?;
6417 crate::mirror_fetch::classify(&record.remote_url).map_err(ControlError::InvalidArgument)?;
6422 crate::mirror_fetch::check_destination(&record.remote_url, self.stores.root())
6425 .map_err(ControlError::InvalidArgument)?;
6426 let job = self.mirror.enqueue(crate::mirror_queue::MirrorOrder {
6427 path: repo_ref.key(),
6428 upstream: record.remote_url.clone(),
6429 prune: record.prune,
6430 ca: mirror_ca_of(&record),
6431 });
6432 let mut out = mirror_out(&record);
6433 out.job = Some(mirror_job_out(path, job));
6434 Ok(out)
6435 }
6436
6437 fn mirror_set_credential(
6450 &self,
6451 path: &RepoPath,
6452 kind: gunnar_proto::MirrorCredentialKind,
6453 secret: &str,
6454 username: &str,
6455 ) -> ControlResult<MirrorConfig> {
6456 gunnar_grpc::seam::credential_arguments(kind, secret)?;
6457 self.mirror_record(path)?;
6458 let repo_ref = Self::repo_ref(path)?;
6459
6460 let kind = credential_kind_in(kind);
6461 let secrets = self.stores.secrets().map_err(|err| {
6462 ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
6463 })?;
6464 let key = gunnar_core::credential_secret_key(&repo_ref);
6465 let stored = if kind.is_set() {
6466 gunnar_store::SecretStore::put(&secrets, &key, secret.as_bytes())
6467 } else {
6468 gunnar_store::SecretStore::delete(&secrets, &key)
6469 };
6470 stored.map_err(|err| {
6471 ControlError::Internal(format!(
6473 "the mirror credential for {path} could not be sealed: {err}"
6474 ))
6475 })?;
6476
6477 let record = self
6478 .stores
6479 .mirrors()
6480 .set_credential(&repo_ref, kind, username)
6481 .map_err(internal)?
6482 .ok_or_else(|| ControlError::NotFound(format!("no mirror registered for {path}")))?;
6483 Ok(mirror_out(&record))
6484 }
6485
6486 fn mirror_set_ca(&self, path: &RepoPath, ca_pem: &str) -> ControlResult<MirrorConfig> {
6487 self.mirror_record(path)?;
6488 let repo_ref = Self::repo_ref(path)?;
6489 let ca = if ca_pem.trim().is_empty() {
6490 None
6491 } else {
6492 let ders = gunnar_grpc::sso::idp_ca::parse_ca(
6496 ca_pem,
6497 gunnar_grpc::bus::now_unix_ms(),
6498 "the mirror's CA",
6499 "the upstream",
6500 )
6501 .map_err(ControlError::InvalidArgument)?;
6502 let (subject, fingerprint) = gunnar_grpc::sso::idp_ca::ca_shown(&ders[0]);
6503 Some((
6504 gunnar_grpc::sso::idp_ca::idp_ca_pem(&ders),
6505 subject,
6506 fingerprint,
6507 ))
6508 };
6509 let record = self
6510 .stores
6511 .mirrors()
6512 .set_ca(&repo_ref, ca)
6513 .map_err(internal)?
6514 .ok_or_else(|| ControlError::NotFound(format!("no mirror registered for {path}")))?;
6515 Ok(mirror_out(&record))
6516 }
6517
6518 fn mirror_status(
6519 &self,
6520 namespace: Option<&str>,
6521 path: Option<&RepoPath>,
6522 ) -> ControlResult<Vec<MirrorConfig>> {
6523 let all = self.stores.mirrors().all().map_err(internal)?;
6524 Ok(all
6525 .iter()
6526 .filter(|record| match path {
6527 Some(want) => record.repo == format!("{}/{}", want.namespace, want.repo),
6528 None => true,
6529 })
6530 .filter(|record| match namespace {
6531 Some(want) => record
6532 .repo
6533 .split_once('/')
6534 .is_some_and(|(ns, _)| ns == want),
6535 None => true,
6536 })
6537 .map(|record| {
6538 let mut out = mirror_out(record);
6539 out.job = out
6545 .repo
6546 .as_ref()
6547 .zip(self.mirror.status(&record.repo))
6548 .map(|(path, job)| mirror_job_out(path, job));
6549 if out.job.is_none() && !record.last_error.is_empty() {
6556 out.job = out.repo.clone().map(|path| gunnar_proto::MirrorJob {
6557 job_id: String::new(),
6558 path: Some(path),
6559 state: gunnar_proto::MirrorJobState::Failed as i32,
6560 error: record.last_error.clone(),
6561 ..Default::default()
6562 });
6563 }
6564 out
6565 })
6566 .collect())
6567 }
6568
6569 fn mirror_remove(&self, path: &RepoPath) -> ControlResult<()> {
6590 self.mirror_record(path)?;
6591 let repo_ref = Self::repo_ref(path)?;
6592 if !self.stores.mirrors().remove(&repo_ref).map_err(internal)? {
6593 return Err(ControlError::NotFound(format!(
6594 "no mirror registered for {path}"
6595 )));
6596 }
6597 let secrets = self.stores.secrets().map_err(|err| {
6598 ControlError::Internal(format!("the sealed secret store is not available: {err:#}"))
6599 })?;
6600 gunnar_store::SecretStore::delete(&secrets, &gunnar_core::credential_secret_key(&repo_ref))
6601 .map_err(|err| {
6602 ControlError::Internal(format!(
6606 "the registration for {path} was removed and its sealed credential could \
6607 not be deleted: {err}. The secret is still on this box; remove it before \
6608 trusting that it is gone."
6609 ))
6610 })?;
6611 Ok(())
6612 }
6613
6614 #[cfg(feature = "replication")]
6617 fn replication_set(
6618 &self,
6619 target: gunnar_proto::ReplicationTarget,
6620 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
6621 let slot = self.replication_slot()?;
6622 let mut target = crate::replication_wire::target_from_proto(target)?;
6623 let minted_public = if target.identity.is_none() {
6633 let (path, public) = crate::replication_wire::ensure_identity(self.stores.root())?;
6634 target.identity = Some(path);
6635 Some(public)
6636 } else {
6637 None
6638 };
6639 crate::replication_wire::require_public_only_identity(&target)?;
6645 let mut config = slot
6646 .config()
6647 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6648 config.set_target(target);
6652 slot.apply(config)
6653 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6654 let mut out = self.replication_snapshot(false)?;
6655 out.identity_public_key = minted_public.unwrap_or_default();
6656 Ok(out)
6657 }
6658
6659 #[cfg(feature = "replication")]
6660 fn replication_remove(
6661 &self,
6662 name: &str,
6663 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
6664 let slot = self.replication_slot()?;
6665 let mut config = slot
6666 .config()
6667 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6668 if !config.remove_target(name) {
6669 return Err(ControlError::NotFound(format!(
6674 "no replication target named {name:?}"
6675 )));
6676 }
6677 slot.apply(config)
6678 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
6679 self.replication_snapshot(false)
6680 }
6681
6682 fn is_group_member(&self, principal_id: &str, group: &str) -> ControlResult<bool> {
6685 self.stores
6686 .auth()
6687 .groups_of(principal_id)
6688 .map(|groups| groups.iter().any(|g| g == group))
6689 .map_err(|err| ControlError::Unavailable(format!("reading group membership: {err}")))
6690 }
6691
6692 fn become_twin(&self, primary: &str) -> ControlResult<String> {
6693 let key = crate::twin::become_twin(&self.stores, primary)
6694 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
6695 crate::twin::start_follower(Arc::clone(&self.stores));
6698 Ok(key)
6699 }
6700
6701 #[cfg(feature = "wg-client")]
6702 fn assign_tunnel_address(
6703 &self,
6704 address: &str,
6705 ) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
6706 let window_open = self.stores.auth().bootstrap_window_open().map_err(|err| {
6707 ControlError::Unavailable(format!("reading the bootstrap window: {err}"))
6708 })?;
6709 crate::appliance::tunnel_address_answer(self.stores.root(), address, window_open)
6710 }
6711
6712 #[cfg(feature = "wg-client")]
6713 fn set_tunnel_address(
6714 &self,
6715 address: &str,
6716 ) -> ControlResult<gunnar_proto::AssignTunnelAddressResponse> {
6717 crate::appliance::tunnel_address_answer(self.stores.root(), address, true)
6721 }
6722
6723 fn twin_csr(&self) -> ControlResult<String> {
6724 crate::twin_tls::request(self.stores.root())
6725 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6726 }
6727
6728 fn install_twin_certificate(
6729 &self,
6730 cert_pem: &str,
6731 ca_pem: &str,
6732 replication_port: u16,
6733 replication_host: &str,
6734 ) -> ControlResult<i64> {
6735 let host = (!replication_host.is_empty()).then_some(replication_host);
6736 crate::twin_tls::install_at(self.stores.root(), cert_pem, ca_pem, replication_port, host)
6737 .map(|expiry| expiry.days_left)
6738 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6739 }
6740
6741 fn issue_twin_certificate(
6742 &self,
6743 principal: &str,
6744 csr_pem: &str,
6745 ) -> ControlResult<(String, String, u16, String)> {
6746 crate::twin_tls::issue_for_twin(&self.stores, principal, csr_pem)
6747 .map(|(cert, ca, port)| {
6748 (
6749 cert,
6750 ca,
6751 port,
6752 crate::twin_tls::replication_host(self.stores.root()).unwrap_or_default(),
6753 )
6754 })
6755 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6756 }
6757
6758 fn twin_git_listener_needs_restart(&self) -> bool {
6759 crate::twin::git_listener_needs_restart(&self.stores)
6760 }
6761
6762 fn restore_begin(&self, owner: &str) -> ControlResult<u64> {
6763 crate::twin::restore_begin_by(&self.stores, owner).map_err(|err| {
6764 if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() {
6766 ControlError::PermissionDenied(format!("{err:#}"))
6767 } else {
6768 ControlError::FailedPrecondition(format!("{err:#}"))
6769 }
6770 })
6771 }
6772
6773 fn restore_keys(&self) -> ControlResult<Vec<String>> {
6777 #[allow(unused_mut)]
6778 let mut keys = self.restore_keys.clone();
6779 #[cfg(feature = "wg-client")]
6780 keys.extend(
6781 crate::fed_principals::restore_keys_at(self.stores.root())
6782 .map_err(ControlError::Unavailable)?,
6783 );
6784 keys.sort();
6785 keys.dedup();
6786 Ok(keys)
6787 }
6788
6789 fn restore_room(&self, announced_bytes: u64) -> ControlResult<()> {
6790 crate::twin::room_for_restore(self.stores.root(), announced_bytes)
6791 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))
6792 }
6793
6794 fn restore_intake<'a>(
6798 &'a self,
6799 owner: &str,
6800 ) -> ControlResult<Option<Box<dyn gunnar_grpc::RestoreIntake + 'a>>> {
6801 match crate::twin::intake::Intake::open(&self.stores, owner) {
6802 Ok(Some(intake)) => Ok(Some(Box::new(intake))),
6803 Ok(None) => Ok(None),
6804 Err(err) if err.downcast_ref::<crate::twin::RestoreNotYours>().is_some() => {
6806 Err(ControlError::PermissionDenied(format!("{err:#}")))
6807 }
6808 Err(err) => Err(ControlError::FailedPrecondition(format!("{err:#}"))),
6809 }
6810 }
6811
6812 fn restore_apply(&self, entry: gunnar_proto::LogEntry) -> ControlResult<bool> {
6813 let e = gunnar_core::txlog::Entry {
6814 seq: entry.seq,
6815 kind: entry.kind as u16,
6816 meta: entry.meta,
6817 blob: entry.blob,
6818 };
6819 match crate::twin::restore_apply(&self.stores, &e) {
6820 Ok(crate::twin::Applied::Lost(_)) => Ok(true),
6821 Ok(_) => Ok(false),
6822 Err(err) => Err(ControlError::Internal(format!(
6823 "restoring seq {}: {err:#}",
6824 e.seq
6825 ))),
6826 }
6827 }
6828
6829 fn restore_finish(&self, announced_head: u64, log_id_of_source: &str) -> ControlResult<u64> {
6830 let head =
6831 crate::twin::restore_finish_streamed(&self.stores, announced_head, log_id_of_source)
6832 .map_err(|err| ControlError::FailedPrecondition(format!("{err:#}")))?;
6833 crate::twin::restart_after_restore();
6835 Ok(head)
6836 }
6837
6838 fn snapshot_stream(
6839 &self,
6840 sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
6841 ) -> ControlResult<u64> {
6842 self.snapshot_stream_from(0, 0, "", sink)
6843 }
6844
6845 fn snapshot_stream_from(
6850 &self,
6851 from_item: u64,
6852 from_offset: u64,
6853 expect_id: &str,
6854 sink: &mut dyn FnMut(gunnar_proto::LogChunk) -> ControlResult<()>,
6855 ) -> ControlResult<u64> {
6856 let mut pieces = |p: crate::twin::snapshot::Piece<'_>,
6858 pos: Option<crate::twin::snapshot::PiecePos<'_>>|
6859 -> anyhow::Result<()> {
6860 sink(gunnar_proto::LogChunk {
6861 seq: 0,
6862 kind: u32::from(p.kind),
6863 meta: p.meta.to_vec(),
6864 blob_len: p.blob_len,
6865 blob: p.blob.to_vec(),
6866 last: p.last,
6867 restore_head: 0,
6868 restore_bytes: 0,
6869 log_id: String::new(),
6870 offset: pos.map(|q| q.offset),
6871 entry_crc: pos.and_then(|q| q.crc),
6872 snapshot_item: pos.map(|q| q.item),
6873 snapshot_id: pos.map(|q| q.id.to_owned()).unwrap_or_default(),
6874 })
6875 .map_err(|err| anyhow::anyhow!("{err}"))
6876 };
6877 crate::twin::snapshot::serve_from(
6878 &self.stores,
6879 from_item,
6880 from_offset,
6881 expect_id,
6882 &mut pieces,
6883 )
6884 .map_err(|err| {
6885 match err.downcast_ref::<crate::twin::snapshot::SnapshotChanged>() {
6886 Some(changed) => ControlError::FailedPrecondition(changed.to_string()),
6887 None => ControlError::Internal(format!("producing a snapshot: {err:#}")),
6888 }
6889 })
6890 }
6891
6892 fn log_retain(&self, from: u64) -> ControlResult<gunnar_proto::LogHeadResponse> {
6893 let log = self.stores.txlog();
6894 if log.is_follower() {
6895 return Err(ControlError::FailedPrecondition(
6896 "retention: a twin's log is the primary's numbers; retain on the primary".into(),
6897 ));
6898 }
6899 let from = if from == 0 { log.head_seq() + 1 } else { from };
6900 let first = log.retain_from(from).map_err(|err| {
6901 ControlError::Unavailable(format!("retaining the transaction log from {from}: {err}"))
6902 })?;
6903 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");
6904 self.log_head()
6905 }
6906
6907 fn log_layout(&self) -> ControlResult<Vec<(u64, u64, u64)>> {
6908 let log = self.stores.txlog();
6909 if log.is_follower() {
6910 return Err(ControlError::FailedPrecondition(
6911 "retention: a twin's log is the primary's numbers; retain on the primary".into(),
6912 ));
6913 }
6914 Ok(log.layout())
6915 }
6916
6917 fn data_volume(&self) -> Option<(u64, u64)> {
6918 if self.stores.txlog().is_released() {
6921 return None;
6922 }
6923 let vfs = rustix::fs::statvfs(self.stores.root()).ok()?;
6924 let total = vfs.f_blocks.saturating_mul(vfs.f_frsize);
6925 let free = vfs.f_bfree.saturating_mul(vfs.f_frsize);
6926 (total > 0).then(|| (total, total.saturating_sub(free)))
6927 }
6928
6929 fn twin_owed_read(&self) -> ControlResult<Vec<u8>> {
6930 match std::fs::read(self.stores.root().join(crate::twin::TWIN_OWED_FILE)) {
6931 Ok(b) => Ok(b),
6932 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
6933 Err(e) => Err(ControlError::Unavailable(format!(
6934 "reading {}: {e}",
6935 crate::twin::TWIN_OWED_FILE
6936 ))),
6937 }
6938 }
6939
6940 fn twin_owed_write(&self, bytes: &[u8]) -> ControlResult<()> {
6941 if self.stores.txlog().is_released() {
6942 return Err(ControlError::Unavailable("the data set is released for a growth; twin-owed.json is written after the remount".into()));
6943 }
6944 crate::durable::write_atomic_unframed(
6946 self.stores.root(),
6947 crate::twin::TWIN_OWED_FILE,
6948 bytes,
6949 )
6950 .map_err(|e| {
6951 ControlError::Unavailable(format!("writing {}: {e:#}", crate::twin::TWIN_OWED_FILE))
6952 })
6953 }
6954
6955 fn restore_snapshot_chunk(&self, chunk: &gunnar_proto::LogChunk) -> ControlResult<Option<u64>> {
6956 let kind = u16::try_from(chunk.kind).map_err(|_| {
6957 ControlError::InvalidArgument(format!("snapshot piece of kind {}", chunk.kind))
6958 })?;
6959 let piece = crate::twin::snapshot::Piece {
6960 kind,
6961 meta: &chunk.meta,
6962 blob_len: chunk.blob_len,
6963 blob: &chunk.blob,
6964 last: chunk.last,
6965 };
6966 crate::twin::snapshot::restore_piece(&self.stores, piece)
6967 .map_err(|err| ControlError::Internal(format!("installing the snapshot: {err:#}")))
6968 }
6969
6970 fn health_alarms(&self) -> Vec<gunnar_proto::HealthAlarm> {
6971 #[allow(unused_mut)]
6972 let mut alarms = Vec::new();
6973 if !self.stores.is_twin() && crate::twin::is_restoring(self.stores.root()) {
6975 alarms.push(gunnar_proto::HealthAlarm {
6976 event: "twin.restoring_at_open".into(),
6977 count: 1,
6978 since_unix_ms: 0,
6979 detail: "this box is MID-RESTORE: it originates no transactions and refuses every \
6980 control-plane write until a restore finishes — resend the twin's log"
6981 .into(),
6982 });
6983 }
6984 if !self.stores.is_twin() {
6987 if let Some(why) = crate::twin::fenced_why(self.stores.root()) {
6988 alarms.push(gunnar_proto::HealthAlarm {
6989 event: "twin.fenced_by_generation".into(),
6990 count: 1,
6991 since_unix_ms: 0,
6992 detail: why,
6993 });
6994 }
6995 }
6996 if self.stores.is_twin() {
6999 if let Some(d) = crate::twin::snapshot::dropped(self.stores.root()) {
7000 let too_small = crate::twin::snapshot::too_small_total(self.stores.root())
7001 .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))
7002 .unwrap_or_default();
7003 alarms.push(gunnar_proto::HealthAlarm {
7004 event: "twin.resync_without_copy".into(),
7005 count: 1,
7006 since_unix_ms: d.at_unix_ms,
7007 detail: format!("{}.{too_small}", d.sentence()),
7008 });
7009 }
7010 }
7011 if !self.stores.is_twin() {
7014 let (total, _) = crate::twin::volume_bytes(self.stores.root());
7015 match crate::twin::txlog_alarm_cap(
7016 std::env::var(crate::twin::TXLOG_ALARM_ENV).ok().as_deref(),
7017 total,
7018 ) {
7019 Ok(Some((cap, how))) => alarms.extend(crate::twin::txlog_size_alarm(
7020 self.stores.txlog().bytes(),
7021 cap,
7022 &how,
7023 )),
7024 Ok(None) => {}
7025 Err(why) => alarms.push(gunnar_proto::HealthAlarm {
7026 event: crate::twin::TXLOG_SIZE_ALARM.into(),
7027 count: 1,
7028 since_unix_ms: 0,
7029 detail: why,
7030 }),
7031 }
7032 }
7033 #[cfg(feature = "metered")]
7034 if let Some(a) = self.entitlement.as_ref().and_then(|slot| slot.alarm()) {
7035 alarms.push(a);
7036 }
7037 if let Some(a) = self.grow.as_ref().and_then(|g| g.alarm()) {
7040 alarms.push(a);
7041 }
7042 alarms.extend(crate::feed::alarms());
7045 alarms
7046 }
7047
7048 fn is_twin(&self) -> bool {
7049 self.stores.is_twin()
7050 }
7051
7052 fn frozen(&self) -> Option<String> {
7053 if self.stores.is_twin() {
7054 Some("this box is a twin".to_owned())
7055 } else if self
7056 .stores
7057 .browse()
7058 .load(std::sync::atomic::Ordering::SeqCst)
7059 {
7060 Some("this box is a browse replica".to_owned())
7061 } else {
7062 None
7063 }
7064 }
7065
7066 fn log_follows(&self) -> bool {
7067 self.stores.is_twin() || self.stores.txlog().is_follower()
7068 }
7069
7070 fn topology_facts(&self) -> gunnar_grpc::topology::Facts {
7071 crate::topology::facts(self.stores.root())
7072 }
7073
7074 fn maintenance_reasons(&self) -> Vec<String> {
7075 #[allow(unused_mut)]
7076 let mut why = Vec::new();
7077 #[cfg(feature = "metered")]
7078 if self
7079 .entitlement
7080 .as_ref()
7081 .is_some_and(|slot| slot.maintenance())
7082 {
7083 why.push(crate::entitlement::KEY_MISMATCH_ALARM.to_string());
7084 }
7085 why
7086 }
7087
7088 fn primary_heartbeat(&self, silent_after_ms: u64) -> Option<gunnar_proto::PrimaryHeartbeat> {
7089 let cfg = crate::twin::TwinConfig::load(self.stores.root())
7090 .ok()
7091 .flatten()?;
7092 let st = crate::twin::status(self.stores.root());
7096 let (last, primary_head, applied) = match &st {
7097 Some(s) => (s.last_answer_unix_ms, s.primary_head_seq, s.applied_seq),
7098 None => (0, 0, self.stores.txlog().head_seq()),
7099 };
7100 let own = st.as_ref().map(|s| (s.state.to_string(), s.why.clone()));
7101 Some(gunnar_grpc::heartbeat::primary_view(
7102 &cfg.primary,
7103 last,
7104 primary_head,
7105 applied,
7106 crate::twin::now_ms(),
7107 silent_after_ms,
7108 own.as_ref().map(|(a, b)| (a.as_str(), b.as_str())),
7109 ))
7110 }
7111
7112 fn log_head(&self) -> ControlResult<gunnar_proto::LogHeadResponse> {
7113 self.log_head_signed("")
7114 }
7115
7116 fn log_head_signed(&self, nonce: &str) -> ControlResult<gunnar_proto::LogHeadResponse> {
7117 if !nonce.trim().is_empty() {
7128 let root = self.stores.root();
7129 if crate::twin::is_restoring(root) {
7130 return Err(ControlError::Unavailable(
7131 "this box is being restored from a twin and is not a primary yet: it answers a twin's poll \
7132 once the replay has finished and the box has restarted"
7133 .to_owned(),
7134 ));
7135 }
7136 if crate::twin::reseal_pending(root) {
7137 return Err(ControlError::Unavailable(
7138 "this box finished a restore and must restart before it answers a twin: its host key is \
7139 sealed under the key-encrypting key the restart re-seals, so until then it cannot prove \
7140 which box it is"
7141 .to_owned(),
7142 ));
7143 }
7144 }
7145 let mut head = crate::twin::own_head(&self.stores);
7146 let root = self.stores.root();
7157 let generation = if crate::twin::is_twin_root(root) {
7158 crate::twin::pinned_generation(root)
7159 } else {
7160 crate::twin::log_generation(root)
7161 };
7162 head.generation = generation
7163 .map_err(|err| ControlError::Internal(format!("this box's log generation: {err:#}")))?;
7164 let (key, sig) = crate::twin::sign_head_answer(
7165 &self.stores,
7166 nonce,
7167 &head.log_id,
7168 head.first_seq,
7169 head.head_seq,
7170 head.generation,
7171 );
7172 head.answer_key = key;
7173 head.answer_sig = sig;
7174 Ok(head)
7175 }
7176
7177 fn fence_by_generation(&self, proof: &gunnar_proto::GenerationProof) -> ControlResult<()> {
7178 crate::twin::fence_by_proof(&self.stores, proof)
7179 .map(drop)
7180 .map_err(|err| {
7181 ControlError::Internal(format!(
7182 "fencing this box off as an old generation of its log: {err:#}"
7183 ))
7184 })
7185 }
7186
7187 fn log_fetch(
7188 &self,
7189 from: u64,
7190 to: u64,
7191 max_bytes: u64,
7192 ) -> ControlResult<Vec<gunnar_proto::LogEntry>> {
7193 let log = self.readable_log()?;
7194 let entries =
7195 crate::twin::fetch_range(log, from, to, max_bytes).map_err(log_read_refusal)?;
7196 Ok(entries
7197 .into_iter()
7198 .map(|e| gunnar_proto::LogEntry {
7199 seq: e.seq,
7200 kind: e.kind as u32,
7201 meta: e.meta,
7202 blob: e.blob,
7203 })
7204 .collect())
7205 }
7206
7207 fn log_fetch_stream(
7212 &self,
7213 from: u64,
7214 to: u64,
7215 max_bytes: u64,
7216 ) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
7217 self.log_fetch_stream_at(from, 0, to, max_bytes)
7218 }
7219
7220 fn log_fetch_stream_at(
7223 &self,
7224 from: u64,
7225 from_offset: u64,
7226 to: u64,
7227 max_bytes: u64,
7228 ) -> ControlResult<gunnar_grpc::seam::LogFetchProducer> {
7229 let log = self.readable_log()?;
7230 crate::twin::check_range(log, from, to).map_err(log_read_refusal)?;
7231 let stores = Arc::clone(&self.stores);
7232 Ok(Box::new(move |sink| {
7233 let mut out = ChunkOut { sink, open: None };
7234 crate::twin::stream_range_at(stores.txlog(), from, from_offset, to, max_bytes, &mut out)
7235 .map(drop)
7236 .map_err(|err| {
7237 let err = match err.downcast::<ControlError>() {
7238 Ok(said) => return said,
7239 Err(other) => other,
7240 };
7241 match err.downcast::<gunnar_core::txlog::Error>() {
7242 Ok(e) => log_read_refusal(e),
7243 Err(other) => ControlError::Internal(format!("transaction log: {other:#}")),
7244 }
7245 })
7246 }))
7247 }
7248
7249 #[cfg(feature = "replication")]
7250 fn replication_status(
7251 &self,
7252 include_refs: bool,
7253 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7254 self.replication_slot()?;
7255 self.replication_snapshot(include_refs)
7256 }
7257
7258 #[cfg(feature = "replication")]
7270 fn replication_flush(
7271 &self,
7272 request: &gunnar_proto::FlushReplicationRequest,
7273 ) -> ControlResult<gunnar_proto::FlushReplicationResponse> {
7274 let slot = self.replication_slot()?;
7275 let began = gunnar_grpc::bus::now_unix_ms();
7276 let started = std::time::Instant::now();
7277 let ceiling = if request.timeout_secs == 0 {
7278 abdicate_flush_secs()
7279 } else {
7280 u64::from(request.timeout_secs)
7281 };
7282 let Some(replicator) = slot.replicator() else {
7283 return Err(ControlError::FailedPrecondition(
7284 "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(),
7285 ));
7286 };
7287 if request.resync {
7288 replicator
7289 .resync_scoped(None, None, &[])
7290 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7291 }
7292 let deadline = started + std::time::Duration::from_secs(ceiling);
7293 loop {
7294 let drained = replicator.status().iter().all(|s| s.pending == 0);
7295 if drained || std::time::Instant::now() >= deadline {
7296 break;
7297 }
7298 std::thread::sleep(std::time::Duration::from_millis(100));
7299 }
7300 let status = replicator.status();
7301 let counters = replicator.counters();
7302 let pending: u64 = status.iter().map(|s| s.pending as u64).sum();
7303 let needs_full_resync = status.iter().any(|s| s.needs_full_resync);
7304 let last_success = status
7305 .iter()
7306 .map(|s| s.last_success_unix_ms.unwrap_or(0))
7307 .min()
7308 .unwrap_or(0);
7309 let in_flight = self.admission.as_ref().map(|a| a.in_flight()).unwrap_or(0);
7310 let catalog_repos = self
7311 .stores
7312 .catalog()
7313 .counts()
7314 .map(|(_, repos)| repos as u64)
7315 .unwrap_or(0);
7316 let standbys = self.replication_snapshot(false)?.standbys;
7317 let caught_up = in_flight == 0
7318 && !status.is_empty()
7319 && pending == 0
7320 && !needs_full_resync
7321 && status.iter().all(|s| {
7322 s.last_success_unix_ms.is_some_and(|t| t >= began)
7323 || s.pending == 0 && s.last_success_unix_ms.is_some()
7324 });
7325 let verdict = if status.is_empty() {
7326 "no standby is configured: nothing was flushed and nothing is proven".to_owned()
7327 } else if caught_up {
7328 format!(
7329 "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",
7330 status.len(),
7331 in_flight,
7332 counters.bus_gaps,
7333 if request.resync { " (every catalog repository was enqueued)" } else { "" }
7334 )
7335 } else {
7336 let names: Vec<String> = status
7337 .iter()
7338 .filter(|s| s.pending > 0 || s.needs_full_resync)
7339 .map(|s| {
7340 format!(
7341 "{} (pending {}{})",
7342 s.name,
7343 s.pending,
7344 if s.needs_full_resync {
7345 ", needs full resync"
7346 } else {
7347 ""
7348 }
7349 )
7350 })
7351 .collect();
7352 format!(
7353 "NOT caught up after {} s: {} transfer(s) in flight, {} repository(ies) pending [{}]{}; the growth must not unmount the set on this",
7354 started.elapsed().as_secs(),
7355 in_flight,
7356 pending,
7357 names.join(", "),
7358 if needs_full_resync { " — a standby has a hole retrying will not close: run `replication resync`" } else { "" }
7359 )
7360 };
7361 tracing::info!(
7362 target: "gunnar::replication",
7363 event = "replication.flush",
7364 caught_up,
7365 in_flight,
7366 pending,
7367 bus_gaps = counters.bus_gaps,
7368 needs_full_resync,
7369 waited_ms = started.elapsed().as_millis() as u64,
7370 "{verdict}"
7371 );
7372 Ok(gunnar_proto::FlushReplicationResponse {
7373 caught_up,
7374 in_flight,
7375 pending,
7376 bus_gaps: counters.bus_gaps,
7377 needs_full_resync,
7378 last_success_unix_ms: last_success,
7379 flushed_at_unix_ms: began,
7380 waited_ms: started.elapsed().as_millis() as u64,
7381 standbys,
7382 catalog_repos,
7383 verdict,
7384 })
7385 }
7386
7387 #[cfg(feature = "replication")]
7401 fn replication_demote(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7402 let slot = self.replication_slot()?;
7403 let mut config = slot
7404 .config()
7405 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7406 config.clear_targets();
7407 slot.apply(config)
7408 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7409 self.replication_snapshot(false)
7410 }
7411
7412 #[cfg(feature = "replication")]
7422 fn replication_set_role(
7423 &self,
7424 role: gunnar_proto::ReplicationRole,
7425 ) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7426 let slot = self.replication_slot()?;
7427 let mut config = slot
7428 .config()
7429 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7430 let wanted = crate::mirror::Role::from_proto(role);
7431 if config.role.is_browse() && wanted == crate::mirror::Role::Primary {
7439 return Err(ControlError::FailedPrecondition(
7440 "this box is a browse replica; it cannot be promoted to primary. It holds only \
7441 public repositories and serves anonymous reads — first `gunnar control \
7442 replication demote` it to a plain standby, then promote that"
7443 .into(),
7444 ));
7445 }
7446 config.role = wanted;
7447 slot.apply(config)
7448 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7449 self.replication_snapshot(false)
7450 }
7451
7452 #[cfg(feature = "replication")]
7474 fn replication_abdicate(&self) -> ControlResult<gunnar_proto::ReplicationStatusResponse> {
7475 let slot = self.replication_slot()?;
7476
7477 let mut config = slot
7480 .config()
7481 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7482 config.role = crate::mirror::Role::Standby;
7483 slot.apply(config)
7484 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7485
7486 if let Some(replicator) = slot.replicator() {
7491 replicator
7492 .resync_scoped(None, None, &[])
7493 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7494 let deadline =
7495 std::time::Instant::now() + std::time::Duration::from_secs(abdicate_flush_secs());
7496 loop {
7497 let drained = replicator.status().iter().all(|s| s.pending == 0);
7498 if drained || std::time::Instant::now() >= deadline {
7499 break;
7500 }
7501 std::thread::sleep(std::time::Duration::from_millis(100));
7502 }
7503 }
7504
7505 let mut config = slot
7509 .config()
7510 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7511 config.clear_targets();
7512 slot.apply(config)
7513 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7514 self.replication_snapshot(false)
7515 }
7516
7517 #[cfg(feature = "replication")]
7531 fn replication_resync(
7532 &self,
7533 name: &str,
7534 repo: Option<&gunnar_proto::RepoPath>,
7535 except: &[gunnar_proto::RepoPath],
7536 ) -> ControlResult<gunnar_proto::ResyncReplicationResponse> {
7537 let slot = self.replication_slot()?;
7538 let Some(replicator) = slot.replicator() else {
7539 return Err(ControlError::Unavailable(format!(
7540 "no replicator is running, so nothing would drain a backfill queue. \
7541 `replication.json` holds {} target(s); the server has not started them.",
7542 slot.config().map(|c| c.targets.len()).unwrap_or(0)
7543 )));
7544 };
7545 let only = match repo {
7546 None => None,
7547 Some(path) => Some(Self::repo_ref(path)?),
7548 };
7549 let except = except
7550 .iter()
7551 .map(Self::repo_ref)
7552 .collect::<ControlResult<Vec<_>>>()?;
7553 let reports = replicator
7554 .resync_scoped(Some(name), only.as_ref(), &except)
7555 .map_err(|err| ControlError::NotFound(format!("{err:#}")))?;
7556 Ok(gunnar_proto::ResyncReplicationResponse {
7557 standbys: reports
7558 .into_iter()
7559 .map(|r| gunnar_proto::ResyncStandby {
7560 standby: r.standby,
7561 in_scope: r.in_scope,
7562 queued: r.queued,
7563 already_pending: r.already_pending,
7564 refused: r.refused,
7565 sha1_repos: r.sha1_repos,
7566 excluded: r.excluded,
7567 })
7568 .collect(),
7569 status: Some(self.replication_snapshot(false)?),
7570 })
7571 }
7572
7573 #[cfg(feature = "webhooks")]
7576 fn webhook_add(
7577 &self,
7578 request: gunnar_proto::AddWebhookRequest,
7579 ) -> ControlResult<gunnar_proto::WebhookListResponse> {
7580 let slot = self.webhook_slot()?;
7581 let endpoint = crate::webhook::Endpoint {
7582 name: request.name,
7583 url: request.url,
7584 kinds: request.kinds,
7585 namespace: request.namespace,
7586 repo: request.repo,
7587 secret: Some(request.secret).filter(|s| !s.is_empty()),
7590 added_at_unix_ms: gunnar_grpc::bus::now_unix_ms(),
7591 };
7592 slot.add(endpoint)
7595 .map_err(|err| ControlError::InvalidArgument(format!("{err:#}")))
7596 }
7597
7598 #[cfg(feature = "webhooks")]
7599 fn webhook_list(&self) -> ControlResult<gunnar_proto::WebhookListResponse> {
7600 self.webhook_slot()?
7601 .list_response()
7602 .map_err(|err| ControlError::Internal(format!("{err:#}")))
7603 }
7604
7605 #[cfg(feature = "webhooks")]
7606 fn webhook_remove(&self, name: &str) -> ControlResult<gunnar_proto::WebhookListResponse> {
7607 let slot = self.webhook_slot()?;
7608 slot.remove(name)
7609 .map_err(|err| ControlError::Internal(format!("{err:#}")))?
7610 .ok_or_else(|| ControlError::NotFound(format!("no webhook endpoint named {name:?}")))
7614 }
7615}
7616
7617const MAX_RESPONSE_BYTES: u64 = 4 * 1024 * 1024;
7643
7644const READ_OBJECT_MEMORY_BUDGET: u64 = 512 * 1024 * 1024;
7653
7654const MAX_CONCURRENT_INFLATES: u64 = 8;
7673
7674const DEFAULT_MAX_OBJECT_BYTES: u64 = READ_OBJECT_MEMORY_BUDGET / MAX_CONCURRENT_INFLATES;
7700
7701impl CoreControlPlane {
7729 fn usage_readable(&self) -> ControlResult<()> {
7731 if self.stores.txlog().is_released()
7732 || crate::twin::maintenance::is_held(self.stores.root())
7733 {
7734 return Err(ControlError::Unavailable(format!(
7735 "{} this box's data set is being grown (unmounted): its usage is read again after the remount — ask again",
7736 gunnar_grpc::PRIMARY_GROWING
7737 )));
7738 }
7739 Ok(())
7740 }
7741
7742 fn open_objects_of(&self, namespace: &str) -> u64 {
7744 let prefix = self
7745 .stores
7746 .repos_root()
7747 .canonicalize()
7748 .unwrap_or_else(|_| self.stores.repos_root().to_path_buf())
7749 .join(namespace);
7750 self.stores
7751 .engine()
7752 .open_object_counts()
7753 .into_iter()
7754 .filter(|(dir, _)| dir.parent() == Some(prefix.as_path()))
7755 .map(|(_, n)| n)
7756 .sum()
7757 }
7758
7759 fn usage_row(
7762 &self,
7763 name: &str,
7764 held: &crate::usage_ledger::Held,
7765 objects: u64,
7766 measured_at_unix_ms: i64,
7767 ) -> gunnar_proto::NamespaceUsage {
7768 let mut row = gunnar_proto::NamespaceUsage {
7769 namespace: name.to_owned(),
7770 measured_at_unix_ms,
7771 pack_bytes: held.bytes.pack,
7772 cache_bytes: held.bytes.cache,
7773 lfs_bytes: held.bytes.lfs,
7774 tombstoned_bytes: held.tombstoned_bytes,
7775 repos: held.repos,
7776 objects,
7777 ..Default::default()
7778 };
7779 fill_flow(&mut row, self.stores.flow(), name, measured_at_unix_ms);
7780 row
7781 }
7782
7783 fn usage_all(&self) -> ControlResult<Vec<gunnar_proto::NamespaceUsage>> {
7787 let catalog = self.stores.catalog();
7788 let namespaces: Vec<String> = catalog
7789 .namespaces()
7790 .map_err(internal)?
7791 .into_iter()
7792 .map(|ns| ns.name)
7793 .collect();
7794 let repos = catalog.list(None).map_err(internal)?;
7795 let tombstones = self.stores.tombstones().unwrap_or_default();
7796 let open_objects: std::collections::HashMap<std::path::PathBuf, u64> = self
7797 .stores
7798 .engine()
7799 .open_object_counts()
7800 .into_iter()
7801 .collect();
7802 let flow = self.stores.flow();
7803 let measured_at_unix_ms = gunnar_core::now_ms();
7804 let mut by_ns: std::collections::HashMap<&str, Vec<&gunnar_core::Repo>> =
7805 std::collections::HashMap::new();
7806 for repo in &repos {
7807 by_ns.entry(repo.namespace.as_str()).or_default().push(repo);
7808 }
7809 let mut out = Vec::with_capacity(namespaces.len());
7810 for name in namespaces {
7811 let mut row = gunnar_proto::NamespaceUsage {
7812 namespace: name.clone(),
7813 measured_at_unix_ms,
7814 ..Default::default()
7815 };
7816 for repo in by_ns.get(name.as_str()).map(Vec::as_slice).unwrap_or(&[]) {
7817 let path = RepoRef::new(&repo.namespace, &repo.name).map_err(|err| {
7818 ControlError::Internal(format!(
7819 "a catalogued repository has an illegal path: {err}"
7820 ))
7821 })?;
7822 let dir = self
7823 .stores
7824 .repo_dir(&path)
7825 .map_err(|err| ControlError::Internal(format!("{err:#}")))?;
7826 let bytes = repo_bytes(&dir);
7827 row.repos += 1;
7828 row.pack_bytes += bytes.pack;
7829 row.cache_bytes += bytes.cache;
7830 row.lfs_bytes += bytes.lfs;
7831 row.objects += open_objects.get(&dir).copied().unwrap_or(0);
7832 }
7833 row.tombstoned_bytes = tombstones
7834 .iter()
7835 .filter(|t| t.path.namespace() == name)
7836 .map(|t| t.bytes)
7837 .sum();
7838 fill_flow(&mut row, flow, &name, measured_at_unix_ms);
7839 out.push(row);
7840 }
7841 Ok(out)
7842 }
7843}
7844
7845fn fill_flow(
7860 row: &mut gunnar_proto::NamespaceUsage,
7861 flow: &gunnar_core::FlowLedger,
7862 name: &str,
7863 now_unix_ms: i64,
7864) {
7865 let glance = flow.glance(name, now_unix_ms);
7866 row.bytes_in_today = glance.today.bytes_in;
7867 row.bytes_out_today = glance.today.bytes_out;
7868 row.pushes_today = glance.today.pushes;
7869 row.fetches_today = glance.today.fetches;
7870 row.refused_today = glance.today.refused();
7871 row.last_push_unix_ms = glance.last_push_unix_ms;
7872 row.last_fetch_unix_ms = glance.last_fetch_unix_ms;
7873}
7874
7875fn object_dir_stats(git_dir: &std::path::Path) -> (u64, u64) {
7881 fn walk(dir: &std::path::Path, loose: &mut u64, bytes: &mut u64, in_loose_shard: bool) {
7882 let Ok(entries) = std::fs::read_dir(dir) else {
7883 return;
7884 };
7885 for entry in entries.flatten() {
7886 let Ok(kind) = entry.file_type() else {
7887 continue;
7888 };
7889 let name = entry.file_name();
7890 let name = name.to_string_lossy();
7891 if kind.is_dir() {
7892 let shard = name.len() == 2 && name.chars().all(|c| c.is_ascii_hexdigit());
7894 walk(&entry.path(), loose, bytes, shard);
7895 } else if kind.is_file() {
7896 if in_loose_shard {
7897 *loose += 1;
7898 }
7899 if let Ok(meta) = entry.metadata() {
7900 *bytes += meta.len();
7901 }
7902 }
7903 }
7904 }
7905 let mut loose = 0;
7906 let mut bytes = 0;
7907 walk(&git_dir.join("objects"), &mut loose, &mut bytes, false);
7908 (loose, bytes)
7909}
7910
7911#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
7931pub struct RepoBytes {
7932 pub pack: u64,
7934 pub cache: u64,
7936 pub lfs: u64,
7938}
7939
7940impl RepoBytes {
7941 pub fn hot(&self) -> u64 {
7943 self.pack + self.cache
7944 }
7945}
7946
7947pub fn repo_bytes(git_dir: &std::path::Path) -> RepoBytes {
7949 fn walk(dir: &std::path::Path, acc: &mut RepoBytes) {
7950 let Ok(entries) = std::fs::read_dir(dir) else {
7951 return;
7952 };
7953 for entry in entries.flatten() {
7954 let Ok(meta) = entry.metadata() else {
7955 continue;
7956 };
7957 if meta.is_dir() {
7958 walk(&entry.path(), acc);
7959 } else if entry
7960 .file_name()
7961 .to_string_lossy()
7962 .starts_with("objects.exploded")
7963 {
7964 acc.cache += meta.len();
7965 } else {
7966 acc.pack += meta.len();
7967 }
7968 }
7969 }
7970 let mut acc = RepoBytes::default();
7971 let Ok(entries) = std::fs::read_dir(git_dir) else {
7972 return acc;
7973 };
7974 for entry in entries.flatten() {
7975 let Ok(meta) = entry.metadata() else {
7976 continue;
7977 };
7978 if meta.is_dir() && entry.file_name() == gunnar_lfs::fs_store::LFS_DIR {
7979 acc.lfs += crate::stores::dir_bytes(&entry.path());
7980 } else if meta.is_dir() {
7981 walk(&entry.path(), &mut acc);
7982 } else {
7983 acc.pack += meta.len();
7984 }
7985 }
7986 acc
7987}
7988
7989impl gunnar_grpc::Identity for CoreControlPlane {
8011 fn awaiting_enrolment(&self, fingerprint: &str) -> Option<String> {
8014 self.stores.awaiting_enrolment(fingerprint)
8015 }
8016
8017 fn registered_key(
8018 &self,
8019 fingerprint: &str,
8020 ) -> ControlResult<Option<gunnar_grpc::RegisteredKey>> {
8021 let Ok(parsed) = gunnar_core::Fingerprint::parse(fingerprint) else {
8025 return Ok(None);
8026 };
8027 let Some(principal) = self
8030 .stores
8031 .auth()
8032 .principal_by_fingerprint(&parsed)
8033 .map_err(internal)?
8034 else {
8035 return Ok(None);
8036 };
8037 let key = self
8038 .stores
8039 .auth()
8040 .keys_of(&principal.id)
8041 .map_err(internal)?
8042 .into_iter()
8043 .find(|key| key.fingerprint_sha256 == fingerprint);
8044 let machine = principal.kind == gunnar_core::PrincipalKind::Machine;
8045 Ok(key.map(|key| gunnar_grpc::RegisteredKey {
8046 principal_id: principal.id,
8047 machine,
8048 openssh: key.openssh,
8052 }))
8053 }
8054
8055 fn registered_certificate(
8070 &self,
8071 der: &[u8],
8072 ) -> ControlResult<Option<gunnar_grpc::CertificateIdentity>> {
8073 let fingerprint = gunnar_core::auth::fingerprint_of_certificate(der);
8077 let Some(principal) = self
8078 .stores
8079 .auth()
8080 .principal_by_fingerprint(&fingerprint)
8081 .map_err(internal)?
8082 else {
8083 return Ok(None);
8084 };
8085 let registered = self
8091 .stores
8092 .auth()
8093 .keys_of(&principal.id)
8094 .map_err(internal)?
8095 .into_iter()
8096 .any(|key| key.fingerprint_sha256 == fingerprint.to_string());
8097 if !registered {
8098 return Ok(None);
8099 }
8100 Ok(Some(gunnar_grpc::CertificateIdentity {
8101 principal_id: principal.id,
8102 fingerprint: fingerprint.to_string(),
8103 }))
8104 }
8105
8106 fn narrowed_methods(&self, principal_id: &str) -> ControlResult<Option<Vec<String>>> {
8109 if principal_id == gunnar_grpc::REPAIR_PRINCIPAL {
8112 return Ok(Some(
8113 gunnar_grpc::REPAIR_METHODS
8114 .iter()
8115 .map(|m| m.to_string())
8116 .collect(),
8117 ));
8118 }
8119 #[cfg(feature = "wg-client")]
8122 return crate::fed_principals::narrowed_of(self.stores.root(), principal_id)
8123 .map_err(ControlError::Internal);
8124 #[cfg(not(feature = "wg-client"))]
8125 {
8126 let _ = principal_id;
8127 Ok(None)
8128 }
8129 }
8130
8131 fn may(
8132 &self,
8133 principal_id: &str,
8134 access: &gunnar_grpc::Access,
8135 level: gunnar_proto::GrantLevel,
8136 ) -> ControlResult<bool> {
8137 let level = level_in(level)?;
8138 let auth = self.stores.auth();
8139 match access {
8140 gunnar_grpc::Access::Server => auth.may_server(principal_id, level).map_err(internal),
8141 gunnar_grpc::Access::Namespace(namespace) => auth
8142 .may_namespace(principal_id, namespace, level)
8143 .map_err(internal),
8144 gunnar_grpc::Access::Repo(path) => {
8145 let repo_ref = Self::repo_ref(path)?;
8146 let Some(repo) = self.stores.catalog().get(&repo_ref).map_err(internal)? else {
8150 return Ok(false);
8151 };
8152 auth.may(principal_id, &repo, level).map_err(internal)
8153 }
8154 }
8155 }
8156
8157 fn may_each_repo(
8160 &self,
8161 principal_id: &str,
8162 paths: &[RepoPath],
8163 level: gunnar_proto::GrantLevel,
8164 ) -> ControlResult<Vec<bool>> {
8165 let level = level_in(level)?;
8166 let refs = paths
8167 .iter()
8168 .map(Self::repo_ref)
8169 .collect::<ControlResult<Vec<_>>>()?;
8170 let rows = self.stores.catalog().get_many(&refs).map_err(internal)?;
8171 let known: Vec<gunnar_core::model::Repo> = rows.iter().flatten().cloned().collect();
8172 let mut answers = self
8173 .stores
8174 .auth()
8175 .may_each(principal_id, &known, level)
8176 .map_err(internal)?
8177 .into_iter();
8178 Ok(rows
8179 .iter()
8180 .map(|row| row.is_some() && answers.next().unwrap_or(false))
8181 .collect())
8182 }
8183}
8184
8185fn protection_out(rule: &gunnar_policy::ProtectionRule) -> BranchProtection {
8197 BranchProtection {
8198 ref_pattern: rule.pattern.as_str().to_owned(),
8199 allow_force_push: rule.allow_force_push,
8200 allow_delete: rule.allow_delete,
8201 require_signature: rule.require_signature,
8202 allowed_principal_ids: rule.allowed_principals.clone().unwrap_or_default(),
8205 max_object_bytes: 0,
8206 max_objects_per_push: 0,
8207 forbid_self_approval: rule.forbid_self_approval,
8208 required_approvals: rule.required_approvals,
8209 }
8210}
8211
8212fn ref_name_policy_out(policy: &gunnar_policy::RefNamePolicy) -> RefNamePolicy {
8213 RefNamePolicy {
8214 allow_patterns: policy.allow.iter().map(|p| p.as_str().to_owned()).collect(),
8215 deny_patterns: policy.deny.iter().map(|p| p.as_str().to_owned()).collect(),
8216 }
8217}
8218
8219fn mirror_out(record: &gunnar_core::MirrorRecord) -> MirrorConfig {
8228 let (namespace, repo) = record
8229 .repo
8230 .split_once('/')
8231 .unwrap_or((record.repo.as_str(), ""));
8232 MirrorConfig {
8233 repo: Some(RepoPath::new(namespace, repo)),
8234 remote_url: record.remote_url.clone(),
8235 prune: record.prune,
8236 state: if record.last_error.is_empty() {
8237 gunnar_proto::MirrorState::Idle as i32
8238 } else {
8239 gunnar_proto::MirrorState::Failed as i32
8240 },
8241 added_at_unix_ms: record.added_at_unix_ms,
8242 last_sync_at_unix_ms: record.last_sync_at_unix_ms,
8243 last_error: record.last_error.clone(),
8244 sync_count: record.sync_count,
8245 credential: credential_kind_out(record.credential) as i32,
8249 credential_username: record.credential_username.clone(),
8250 job: None,
8253 repo_pending: record.creates_repo,
8254 lfs_not_copied: record.lfs_detected,
8255 ca_subject: record.ca_subject.clone(),
8256 ca_fingerprint_sha256: record.ca_fingerprint_sha256.clone(),
8257 }
8258}
8259
8260fn mirror_ca_of(record: &gunnar_core::MirrorRecord) -> Vec<Vec<u8>> {
8263 record
8264 .ca_pem
8265 .as_deref()
8266 .map(gunnar_grpc::sso::idp_ca::idp_ca_blocks)
8267 .unwrap_or_default()
8268}
8269
8270fn credential_kind_in(
8272 kind: gunnar_proto::MirrorCredentialKind,
8273) -> gunnar_core::MirrorCredentialKind {
8274 match kind {
8275 gunnar_proto::MirrorCredentialKind::Unspecified => gunnar_core::MirrorCredentialKind::None,
8276 gunnar_proto::MirrorCredentialKind::SshKey => gunnar_core::MirrorCredentialKind::SshKey,
8277 gunnar_proto::MirrorCredentialKind::Token => gunnar_core::MirrorCredentialKind::Token,
8278 }
8279}
8280
8281fn credential_kind_out(
8283 kind: gunnar_core::MirrorCredentialKind,
8284) -> gunnar_proto::MirrorCredentialKind {
8285 match kind {
8286 gunnar_core::MirrorCredentialKind::None => gunnar_proto::MirrorCredentialKind::Unspecified,
8287 gunnar_core::MirrorCredentialKind::SshKey => gunnar_proto::MirrorCredentialKind::SshKey,
8288 gunnar_core::MirrorCredentialKind::Token => gunnar_proto::MirrorCredentialKind::Token,
8289 }
8290}
8291
8292fn read_credential(
8303 stores: &Arc<ServerStores>,
8304 repo: &gunnar_core::RepoRef,
8305) -> Result<crate::mirror_fetch::Credential, String> {
8306 let record = stores
8307 .mirrors()
8308 .get(repo)
8309 .map_err(|err| {
8310 format!(
8311 "reading the mirror registration for {}: {err:#}",
8312 repo.key()
8313 )
8314 })?
8315 .ok_or_else(|| format!("no mirror is registered for {} any more", repo.key()))?;
8316 if !record.credential.is_set() {
8317 return Ok(crate::mirror_fetch::Credential::None);
8318 }
8319 let secrets = stores
8320 .secrets()
8321 .map_err(|err| format!("the sealed secret store is not available: {err:#}"))?;
8322 let key = gunnar_core::credential_secret_key(repo);
8323 let sealed = gunnar_store::SecretStore::get(&secrets, &key)
8324 .map_err(|err| {
8328 format!(
8329 "the mirror credential for {} could not be read: {err}",
8330 repo.key()
8331 )
8332 })?
8333 .ok_or_else(|| {
8334 format!(
8335 "{} is registered as having {}, and the sealed value is not in the secret \
8336 store. Set it again with `Mirror.SetCredential`; dialling anonymously \
8337 instead would be refused by the upstream and read as a bad key.",
8338 repo.key(),
8339 record.credential.name()
8340 )
8341 })?;
8342 Ok(match record.credential {
8343 gunnar_core::MirrorCredentialKind::SshKey => {
8344 crate::mirror_fetch::Credential::SshKey(sealed)
8345 }
8346 gunnar_core::MirrorCredentialKind::Token => crate::mirror_fetch::Credential::Token {
8347 username: record.credential_username.clone(),
8348 secret: String::from_utf8(sealed).map_err(|_| {
8349 format!(
8350 "the token stored for {} is not utf-8, so it cannot be an HTTP header value",
8351 repo.key()
8352 )
8353 })?,
8354 },
8355 gunnar_core::MirrorCredentialKind::None => crate::mirror_fetch::Credential::None,
8356 })
8357}
8358
8359fn audit_line(
8363 stores: &ServerStores,
8364 repo: &gunnar_core::RepoRef,
8365 kind: &str,
8366 detail: &str,
8367 refused: bool,
8368) {
8369 tracing::info!(
8370 target: gunnar_grpc::AUDIT_TARGET,
8371 event = kind,
8372 repo = %repo.key(),
8373 detail,
8374 "mirror"
8375 );
8376 let Ok(Some(account)) = stores.auth().account_of_namespace(repo.namespace()) else {
8377 return;
8378 };
8379 let event = gunnar_core::account_audit::AccountEvent {
8380 kind,
8381 actor: "",
8382 actor_role: "system",
8383 target: &repo.key(),
8384 detail,
8385 outcome: if refused {
8386 gunnar_core::account_audit::Outcome::Refused
8387 } else {
8388 gunnar_core::account_audit::Outcome::Ok
8389 },
8390 };
8391 if let Err(err) = stores
8392 .account_audit()
8393 .append(&account, event, gunnar_core::now_ms())
8394 {
8395 tracing::warn!(target: "gunnar::git", repo = %repo.key(), %err, "a mirror sync's audit line was not written");
8396 }
8397}
8398
8399fn audit_mirror_outcome(
8403 stores: &ServerStores,
8404 repo: &gunnar_core::RepoRef,
8405 order: &crate::mirror_queue::MirrorOrder,
8406 first: bool,
8407 error: Option<&str>,
8408 report: &crate::job_queue::Reporter<crate::mirror_queue::Mirror>,
8409) {
8410 let upstream = crate::mirror_fetch::scrub(&order.upstream);
8411 match (first, error) {
8412 (true, None) => {
8413 let refs = report
8414 .snapshot()
8415 .map(|job| job.report.updates.len())
8416 .unwrap_or_default();
8417 audit_line(
8418 stores,
8419 repo,
8420 "mirror.import.done",
8421 &format!("imported {refs} reference(s) from {upstream}"),
8422 false,
8423 );
8424 }
8425 (true, Some(why)) => audit_line(
8426 stores,
8427 repo,
8428 "mirror.import.failed",
8429 &format!("from {upstream}: {why}"),
8430 true,
8431 ),
8432 (false, Some(why)) => audit_line(
8433 stores,
8434 repo,
8435 "mirror.sync.failed",
8436 &format!("from {upstream}: {why}"),
8437 true,
8438 ),
8439 (false, None) => {}
8440 }
8441}
8442
8443fn mirror_job_out(path: &RepoPath, job: crate::mirror_queue::MirrorJob) -> gunnar_proto::MirrorJob {
8453 let report = job.report;
8457 let report_refused = report.refused;
8458 let (progress_done, progress_total) = job.progress.unwrap_or((0, 0));
8459 gunnar_proto::MirrorJob {
8460 job_id: job.job_id,
8461 path: Some(path.clone()),
8462 state: match job.state {
8463 crate::mirror_queue::MirrorState::Queued => gunnar_proto::MirrorJobState::Queued as i32,
8464 crate::mirror_queue::MirrorState::Running => {
8465 gunnar_proto::MirrorJobState::Running as i32
8466 }
8467 crate::mirror_queue::MirrorState::Done => gunnar_proto::MirrorJobState::Done as i32,
8468 crate::mirror_queue::MirrorState::Failed => gunnar_proto::MirrorJobState::Failed as i32,
8469 },
8470 queue_position: job.queue_position,
8471 has_progress: job.progress.is_some(),
8475 progress_done,
8476 progress_total,
8477 objects: report.objects,
8478 bytes: report.bytes,
8479 updates: report
8480 .updates
8481 .into_iter()
8482 .map(|(ref_name, old_oid, new_oid)| gunnar_proto::RefUpdate {
8483 ref_name,
8484 old_oid,
8485 new_oid,
8486 })
8487 .collect(),
8488 error: job.error.unwrap_or_default(),
8489 enqueued_at_unix_ms: job.enqueued_at_unix_ms,
8490 started_at_unix_ms: job.started_at_unix_ms,
8491 finished_at_unix_ms: job.finished_at_unix_ms,
8492 refused_reason: match (job.state, report_refused) {
8495 (crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::Plan) => {
8496 gunnar_proto::MirrorRefusal::Plan as i32
8497 }
8498 (crate::mirror_queue::MirrorState::Failed, crate::mirror_queue::Refusal::PackCap) => {
8499 gunnar_proto::MirrorRefusal::PackCap as i32
8500 }
8501 _ => gunnar_proto::MirrorRefusal::Unspecified as i32,
8502 },
8503 }
8504}
8505
8506fn to_proto_membership(
8517 signed: gunnar_core::SignedMembershipEvent,
8518) -> gunnar_proto::SignedMembershipEvent {
8519 gunnar_proto::SignedMembershipEvent {
8520 event: Some(gunnar_proto::MembershipEvent {
8521 group: signed.event.group,
8522 member: signed.event.member,
8523 action: proto_action(signed.event.action) as i32,
8524 actor: signed.event.actor,
8525 at_unix_ms: signed.event.at_unix_ms,
8526 }),
8527 key_fingerprint: signed.key_fingerprint,
8528 signature: signed.signature,
8529 }
8530}
8531
8532fn from_proto_attestation(
8535 wire: gunnar_proto::SignedKeyAttestation,
8536) -> ControlResult<gunnar_core::SignedKeyAttestation> {
8537 let attestation = wire.attestation.ok_or_else(|| {
8538 ControlError::InvalidArgument("`attestation.attestation` is required".to_owned())
8539 })?;
8540 Ok(gunnar_core::SignedKeyAttestation {
8541 attestation: gunnar_core::KeyAttestation {
8542 principal_id: attestation.principal_id,
8543 fingerprint: attestation.fingerprint,
8544 at_unix_ms: attestation.at_unix_ms,
8545 },
8546 key_fingerprint: wire.key_fingerprint,
8547 signature: wire.signature,
8548 })
8549}
8550
8551fn from_proto_membership(
8557 wire: gunnar_proto::SignedMembershipEvent,
8558) -> ControlResult<gunnar_core::SignedMembershipEvent> {
8559 let event = wire
8560 .event
8561 .ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
8562 let action = match gunnar_proto::MembershipAction::try_from(event.action) {
8563 Ok(gunnar_proto::MembershipAction::Add) => gunnar_core::MembershipAction::Add,
8564 Ok(gunnar_proto::MembershipAction::Remove) => gunnar_core::MembershipAction::Remove,
8565 _ => {
8566 return Err(ControlError::InvalidArgument(
8567 "`event.action` must be ADD or REMOVE".to_owned(),
8568 ))
8569 }
8570 };
8571 Ok(gunnar_core::SignedMembershipEvent {
8572 event: gunnar_core::MembershipEvent {
8573 group: event.group,
8574 member: event.member,
8575 action,
8576 actor: event.actor,
8577 at_unix_ms: event.at_unix_ms,
8578 },
8579 key_fingerprint: wire.key_fingerprint,
8580 signature: wire.signature,
8581 })
8582}
8583
8584fn proto_action(action: gunnar_core::MembershipAction) -> gunnar_proto::MembershipAction {
8586 match action {
8587 gunnar_core::MembershipAction::Add => gunnar_proto::MembershipAction::Add,
8588 gunnar_core::MembershipAction::Remove => gunnar_proto::MembershipAction::Remove,
8589 }
8590}
8591
8592fn to_proto_event(signed: gunnar_core::SignedEvent) -> gunnar_proto::SignedForgeEvent {
8594 use gunnar_proto::forge_event::Kind;
8595
8596 let kind = match signed.event.event {
8597 gunnar_core::Event::Commented {
8598 body,
8599 path,
8600 line,
8601 blob_oid,
8602 } => Kind::Commented(gunnar_proto::Commented {
8603 body,
8604 path,
8605 line,
8606 blob_oid,
8607 }),
8608 gunnar_core::Event::Approved { commit } => {
8609 Kind::Approved(gunnar_proto::Approved { commit })
8610 }
8611 gunnar_core::Event::Rejected { commit, reason } => {
8612 Kind::Rejected(gunnar_proto::Rejected { commit, reason })
8613 }
8614 gunnar_core::Event::IssueOpened { title, body } => {
8615 Kind::IssueOpened(gunnar_proto::IssueOpened { title, body })
8616 }
8617 gunnar_core::Event::IssueCommented { body } => {
8618 Kind::IssueCommented(gunnar_proto::IssueCommented { body })
8619 }
8620 gunnar_core::Event::IssueClosed { reason } => {
8621 Kind::IssueClosed(gunnar_proto::IssueClosed { reason })
8622 }
8623 };
8624 gunnar_proto::SignedForgeEvent {
8625 event: Some(gunnar_proto::ForgeEvent {
8626 repo: signed.event.repo,
8627 pull: signed.event.pull,
8628 actor: signed.event.actor,
8629 at_unix_ms: signed.event.at_unix_ms,
8630 kind: Some(kind),
8631 }),
8632 key_fingerprint: signed.key_fingerprint,
8633 signature: signed.signature,
8634 attested: signed.attested.map(|a| gunnar_proto::ForgeAttestation {
8635 session_stepped_up_at_unix_ms: a.session_stepped_up_at_unix_ms,
8636 }),
8637 }
8638}
8639
8640fn issue_state(repo: &str, issue: &gunnar_core::forge::Issue) -> gunnar_proto::IssueState {
8646 gunnar_proto::IssueState {
8647 repo: repo.to_owned(),
8648 number: issue.number,
8649 title: issue.title.clone(),
8650 body: issue.body.clone(),
8651 author: issue.author.clone(),
8652 open: issue.open,
8653 comments: issue.comments as u64,
8654 opened_at_unix_ms: issue.opened_at_unix_ms,
8655 }
8656}
8657
8658fn from_proto_event(
8663 wire: gunnar_proto::SignedForgeEvent,
8664) -> ControlResult<gunnar_core::SignedEvent> {
8665 use gunnar_proto::forge_event::Kind;
8666
8667 let event = wire
8668 .event
8669 .ok_or_else(|| ControlError::InvalidArgument("`event` is required".to_owned()))?;
8670 let kind = event
8671 .kind
8672 .ok_or_else(|| ControlError::InvalidArgument("`event.kind` is required".to_owned()))?;
8673 let inner = match kind {
8674 Kind::Commented(it) => gunnar_core::Event::Commented {
8675 body: it.body,
8676 path: it.path,
8677 line: it.line,
8678 blob_oid: it.blob_oid,
8679 },
8680 Kind::Approved(it) => gunnar_core::Event::Approved { commit: it.commit },
8681 Kind::Rejected(it) => gunnar_core::Event::Rejected {
8682 commit: it.commit,
8683 reason: it.reason,
8684 },
8685 Kind::IssueOpened(it) => gunnar_core::Event::IssueOpened {
8686 title: it.title,
8687 body: it.body,
8688 },
8689 Kind::IssueCommented(it) => gunnar_core::Event::IssueCommented { body: it.body },
8690 Kind::IssueClosed(it) => gunnar_core::Event::IssueClosed { reason: it.reason },
8691 };
8692 Ok(gunnar_core::SignedEvent {
8693 event: gunnar_core::ForgeEvent {
8694 repo: event.repo,
8695 pull: event.pull,
8696 actor: event.actor,
8697 at_unix_ms: event.at_unix_ms,
8698 event: inner,
8699 },
8700 key_fingerprint: wire.key_fingerprint,
8701 signature: wire.signature,
8702 attested: None,
8706 })
8707}
8708#[cfg(test)]
8709mod tests {
8710 use super::*;
8711
8712 #[test]
8718 fn a_write_on_a_restoring_box_is_unavailable_not_internal() {
8719 let td = tempfile::tempdir().unwrap();
8720 let stores = crate::stores::ServerStores::open(td.path()).unwrap();
8721 stores.txlog().set_follower(true);
8722 let p = gunnar_core::Principal::new("alice", "alice", gunnar_core::PrincipalKind::User)
8723 .unwrap();
8724 let err = stores
8725 .auth()
8726 .add_principal(p)
8727 .map_err(internal)
8728 .expect_err("a follower log takes no write");
8729 assert!(matches!(err, ControlError::Unavailable(_)), "{err:?}");
8730 assert!(err.to_string().contains("restor"), "{err}");
8731 }
8732
8733 #[test]
8743 fn internal_is_refused_by_name_and_the_two_real_visibilities_pass() {
8744 let err = visibility_in(gunnar_proto::Visibility::Internal)
8745 .expect_err("INTERNAL has no implementation and must be refused");
8746 let text = err.to_string();
8747 assert!(
8748 text.contains("internal") && text.contains("private"),
8749 "the refusal must name the value and the real choices: {text}"
8750 );
8751 assert_eq!(
8752 visibility_in(gunnar_proto::Visibility::Public).expect("public is real"),
8753 Visibility::Public
8754 );
8755 assert_eq!(
8756 visibility_in(gunnar_proto::Visibility::Private).expect("private is real"),
8757 Visibility::Private
8758 );
8759 assert_eq!(
8761 visibility_in(gunnar_proto::Visibility::Unspecified).expect("unspecified defaults"),
8762 Visibility::Private
8763 );
8764 }
8765
8766 #[cfg(feature = "replication")]
8778 #[test]
8779 fn abdicate_flush_ceiling_honours_a_sane_override_and_rejects_the_rest() {
8780 assert_eq!(
8782 resolve_flush_secs(Some("120")),
8783 120,
8784 "a positive override must win"
8785 );
8786 assert_eq!(
8787 resolve_flush_secs(Some(" 90 ")),
8788 90,
8789 "surrounding whitespace is trimmed"
8790 );
8791 assert_eq!(resolve_flush_secs(None), ABDICATE_FLUSH_SECS);
8794 assert_ne!(
8795 ABDICATE_FLUSH_SECS, 120,
8796 "default must differ from the override arm"
8797 );
8798 assert_eq!(
8800 resolve_flush_secs(Some("")),
8801 ABDICATE_FLUSH_SECS,
8802 "blank falls back"
8803 );
8804 assert_eq!(
8805 resolve_flush_secs(Some("nope")),
8806 ABDICATE_FLUSH_SECS,
8807 "non-numeric falls back"
8808 );
8809 assert_eq!(
8810 resolve_flush_secs(Some("0")),
8811 ABDICATE_FLUSH_SECS,
8812 "zero would truncate — refused"
8813 );
8814 assert_eq!(
8815 resolve_flush_secs(Some("-5")),
8816 ABDICATE_FLUSH_SECS,
8817 "negative does not parse as u64"
8818 );
8819 }
8820}
8821
8822fn key_refusal(r: gunnar_core::encryption_key::KeyRefusal) -> ControlError {
8824 use gunnar_core::encryption_key::KeyRefusal as R;
8825 match r {
8826 R::AlreadyHasKey(s) | R::MethodHeld(s) => ControlError::AlreadyExists(s),
8827 R::NoSuchMethod(s) => ControlError::NotFound(s),
8828 R::NoKey(s) => ControlError::FailedPrecondition(s),
8829 R::LastMethod(s) => ControlError::FailedPrecondition(s),
8830 R::Invalid(s) => ControlError::InvalidArgument(s),
8831 R::Store(e) => internal(e),
8832 }
8833}
8834
8835fn method_in(
8836 m: gunnar_proto::UnlockMethod,
8837) -> ControlResult<gunnar_core::encryption_key::UnlockMethod> {
8838 Ok(gunnar_core::encryption_key::UnlockMethod {
8839 id: String::new(),
8840 kind: gunnar_core::encryption_key::MethodKind::parse(&m.kind).map_err(key_refusal)?,
8841 label: m.label.trim().to_owned(),
8842 method_ref: m.method_ref.trim().to_owned(),
8843 rp_id: m.rp_id.trim().to_owned(),
8844 wrap: m.wrap,
8845 added_ms: 0,
8846 })
8847}
8848
8849fn encryption_key_out(
8850 key: Option<gunnar_core::encryption_key::EncryptionKey>,
8851) -> gunnar_proto::GetEncryptionKeyResponse {
8852 match key {
8853 None => gunnar_proto::GetEncryptionKeyResponse::default(),
8854 Some(k) => gunnar_proto::GetEncryptionKeyResponse {
8855 exists: true,
8856 public_key: k.public_key,
8857 methods: k
8858 .methods
8859 .into_iter()
8860 .map(|m| gunnar_proto::UnlockMethod {
8861 id: m.id,
8862 kind: m.kind.word().to_owned(),
8863 label: m.label,
8864 method_ref: m.method_ref,
8865 rp_id: m.rp_id,
8866 wrap: m.wrap,
8867 added_ms: m.added_ms,
8868 })
8869 .collect(),
8870 },
8871 }
8872}
8873
8874const MERGE_RECHECKS: usize = 3;
8877
8878fn gunnar_ident() -> crate::merge_engine::Ident {
8880 crate::merge_engine::Ident {
8881 name: "Gunnar".to_owned(),
8882 email: "noreply@gunnar.invalid".to_owned(),
8883 }
8884}
8885
8886fn principal_ident(principal: &gunnar_core::Principal) -> crate::merge_engine::Ident {
8890 crate::merge_engine::Ident {
8891 name: principal.name.clone(),
8892 email: format!("{}@gunnar.invalid", principal.id),
8893 }
8894}
8895
8896fn short_ref(name: &str) -> &str {
8898 name.strip_prefix("refs/heads/").unwrap_or(name)
8899}
8900
8901enum MergeAttempt {
8903 Done(Box<(PullRequest, MergedRef)>),
8904 TargetMoved,
8905}
8906
8907fn choose_strategy(
8910 asked: gunnar_proto::MergeStrategy,
8911 fast_forward: bool,
8912 allowed: &gunnar_policy::MergePolicy,
8913 record: &gunnar_core::PullRecord,
8914) -> ControlResult<gunnar_proto::MergeStrategy> {
8915 use gunnar_proto::MergeStrategy as S;
8916 let behind = || {
8917 format!(
8918 "not a fast-forward: {} does not descend from {}",
8919 record.source_ref, record.target_ref
8920 )
8921 };
8922 let refuse = |line: String| Err(ControlError::FailedPrecondition(line));
8923 match asked {
8924 S::FastForward if !allowed.allow_fast_forward => refuse(
8925 "fast-forward merges are turned off for this repository (policy merge.allow_fast_forward); merge without asking for one"
8926 .to_owned(),
8927 ),
8928 S::FastForward if !fast_forward => refuse(format!(
8929 "{}. Merge without asking for a fast-forward (gunnar writes a merge commit), or rebase or merge locally, push the result, and merge again",
8930 behind()
8931 )),
8932 S::FastForward => Ok(S::FastForward),
8933 S::MergeCommit if !allowed.allow_merge_commit => refuse(
8934 "merge commits are turned off for this repository (policy merge.allow_merge_commit)"
8935 .to_owned(),
8936 ),
8937 S::MergeCommit => Ok(S::MergeCommit),
8938 S::Squash if !allowed.allow_squash => refuse(
8939 "squash merges are not turned on for this repository (policy merge.allow_squash)"
8940 .to_owned(),
8941 ),
8942 S::Squash => Ok(S::Squash),
8943 S::Unspecified => {
8944 if fast_forward && allowed.allow_fast_forward {
8945 Ok(S::FastForward)
8946 } else if allowed.allow_merge_commit {
8947 Ok(S::MergeCommit)
8948 } else if allowed.allow_squash {
8949 Ok(S::Squash)
8950 } else if fast_forward {
8951 refuse(
8952 "this repository's policy allows no merge strategy (merge.allow_fast_forward, merge.allow_merge_commit and merge.allow_squash are all off)"
8953 .to_owned(),
8954 )
8955 } else {
8956 refuse(format!(
8957 "{}, and this repository allows only fast-forward merges (policy merge.allow_merge_commit is off): rebase or merge locally, push the result, and merge again",
8958 behind()
8959 ))
8960 }
8961 }
8962 }
8963}
8964
8965impl CoreControlPlane {
8966 fn merge_or_check(
8971 &self,
8972 path: &RepoPath,
8973 number: u64,
8974 principal_id: &str,
8975 options: &gunnar_grpc::MergeOptions,
8976 dry_run: bool,
8977 ) -> ControlResult<(PullRequest, MergedRef)> {
8978 let mut rechecked = false;
8979 for _ in 0..=MERGE_RECHECKS {
8980 match self.merge_attempt(path, number, principal_id, options, dry_run, rechecked)? {
8981 MergeAttempt::Done(out) => return Ok(*out),
8982 MergeAttempt::TargetMoved => rechecked = true,
8983 }
8984 }
8985 Err(ControlError::FailedPrecondition(format!(
8986 "the target of pull request #{number} kept moving while gunnar merged it ({} checks against {} tips); no ref moved — merge again",
8987 MERGE_RECHECKS + 1,
8988 MERGE_RECHECKS + 1
8989 )))
8990 }
8991
8992 #[allow(clippy::too_many_lines)]
8995 fn merge_attempt(
8996 &self,
8997 path: &RepoPath,
8998 number: u64,
8999 principal_id: &str,
9000 options: &gunnar_grpc::MergeOptions,
9001 dry_run: bool,
9002 mut rechecked: bool,
9003 ) -> ControlResult<MergeAttempt> {
9004 use gunnar_proto::MergeStrategy as S;
9005 let repo_ref = Self::repo_ref(path)?;
9006 let key = repo_ref.to_string();
9007 let pulls = self.pulls()?;
9008 let record = pulls
9009 .get(&key, number)
9010 .map_err(internal)?
9011 .ok_or_else(|| ControlError::NotFound(format!("pull request #{number} in {path}")))?;
9012 if record.state != gunnar_core::PullState::Open {
9013 return Err(ControlError::FailedPrecondition(format!(
9014 "pull request #{number} is already {:?}, not open",
9015 record.state
9016 )));
9017 }
9018
9019 let (repo_record, store) = self.open(path)?;
9020 let git = store.serve();
9021 let snapshot = RefSnapshot::read(git).map_err(wire_error)?;
9022 let source = snapshot
9023 .resolve(&record.source_ref)
9024 .map_err(wire_error)?
9025 .ok_or_else(|| {
9026 ControlError::FailedPrecondition(format!(
9027 "the source branch {} is gone; nothing to merge",
9028 record.source_ref
9029 ))
9030 })?;
9031 let target = snapshot
9032 .resolve(&record.target_ref)
9033 .map_err(wire_error)?
9034 .ok_or_else(|| {
9035 ControlError::FailedPrecondition(format!(
9036 "the target branch {} is gone",
9037 record.target_ref
9038 ))
9039 })?;
9040 if let Some(expected) = &options.expected_target_oid {
9043 if !expected.eq_ignore_ascii_case(&target.to_string()) {
9044 rechecked = true;
9045 }
9046 }
9047
9048 let budget = gunnar_policy::ancestry::DEFAULT_ANCESTRY_BUDGET;
9064 let contained = source == target
9065 || gunnar_policy::ancestry::is_descendant_of(git, &target, &source, budget).map_err(
9066 |err| {
9067 ControlError::FailedPrecondition(format!(
9068 "the ancestry walk could not be completed ({err}); refusing rather than \
9069 guessing"
9070 ))
9071 },
9072 )?;
9073 if contained {
9074 let approvers =
9075 self.approvals_enough(&key, number, &record, &repo_record, &source.to_string())?;
9076 let done = MergedRef {
9077 ref_name: record.target_ref.clone(),
9078 old_oid: target.to_string(),
9079 new_oid: target.to_string(),
9080 head_oid: source.to_string(),
9081 rechecked,
9082 contained_in: target.to_string(),
9083 approvers: approvers.clone().unwrap_or_default(),
9084 closed: if approvers.is_some() {
9085 String::new()
9086 } else {
9087 gunnar_grpc::CONTENT_ALREADY_ON_TARGET.to_owned()
9088 },
9089 ..MergedRef::default()
9090 };
9091 if dry_run {
9092 return Ok(MergeAttempt::Done(Box::new((
9093 pull_out(path, record.clone()),
9094 done,
9095 ))));
9096 }
9097 if approvers.is_none() {
9098 let closed = match pulls
9099 .close_because(&key, number, gunnar_grpc::CONTENT_ALREADY_ON_TARGET)
9100 .map_err(internal)?
9101 {
9102 gunnar_core::PullTransition::Done(record) => record,
9103 gunnar_core::PullTransition::NotFound => {
9104 return Err(ControlError::NotFound(format!(
9105 "pull request #{number} in {path}"
9106 )))
9107 }
9108 gunnar_core::PullTransition::NotOpen(state) => {
9109 return Err(ControlError::FailedPrecondition(format!(
9110 "pull request #{number} is already {state:?}, not open"
9111 )))
9112 }
9113 };
9114 return Ok(MergeAttempt::Done(Box::new((pull_out(path, closed), done))));
9115 }
9116 let merged = self.mark_merged(&pulls, &key, number, &target.to_string())?;
9117 return Ok(MergeAttempt::Done(Box::new((pull_out(path, merged), done))));
9118 }
9119
9120 let head = source.to_string();
9136 let signed = self.forge_events()?;
9139 let events: Vec<gunnar_core::ForgeEvent> = signed
9140 .iter()
9141 .filter(|s| s.event.repo == key)
9142 .map(|s| s.event.clone())
9143 .collect();
9144 let review = gunnar_core::forge::fold(number, &events);
9145 let approved = self.review_state(&key, &review, &signed)?;
9146 let ok = approved
9147 .approvals
9148 .iter()
9149 .any(|a| a.verified && a.commit == head);
9150 if !ok {
9151 let named: Vec<String> = approved
9152 .approvals
9153 .iter()
9154 .map(|a| {
9155 format!(
9156 "{} → {} [{}]",
9157 a.actor,
9158 &a.commit[..12.min(a.commit.len())],
9159 a.verdict
9160 )
9161 })
9162 .collect();
9163 return Err(ControlError::FailedPrecondition(format!(
9164 "no verified approval names {}, the commit {} is at. gunnar does not merge on \
9165 an approval of something else: an approval names the commit it approved, so a \
9166 branch that moved after the review has not been reviewed.{}",
9167 &head[..12.min(head.len())],
9168 record.source_ref,
9169 if named.is_empty() {
9170 " Nobody has approved this pull request.".to_owned()
9171 } else {
9172 format!(" What is approved: {}", named.join(", "))
9173 }
9174 )));
9175 }
9176
9177 let policy = self.policy_of(&repo_record)?;
9190 let rules: Vec<&gunnar_policy::ProtectionRule> = policy
9191 .protection
9192 .iter()
9193 .filter(|rule| rule.pattern.matches_ref(&record.target_ref))
9194 .collect();
9195 let forbid_self = rules.iter().any(|rule| rule.forbid_self_approval);
9196 let required = rules
9197 .iter()
9198 .map(|rule| rule.required_approvals)
9199 .max()
9200 .unwrap_or(0)
9201 .max(1) as usize;
9202 let approvers: Vec<&str> = approved
9203 .approvals
9204 .iter()
9205 .filter(|a| a.verified && a.commit == head)
9206 .map(|a| a.actor.as_str())
9207 .collect();
9208 let counted: Vec<&str> = approvers
9209 .iter()
9210 .copied()
9211 .filter(|actor| !(forbid_self && *actor == record.author))
9212 .collect();
9213 if forbid_self && counted.is_empty() && !approvers.is_empty() {
9214 return Err(ControlError::FailedPrecondition(format!(
9215 "the protection on {} forbids self-approval, and the only verified approval \
9216 of {} is the author's own ({}). An author's sign-off on their own work is \
9217 not a review; somebody else has to look at it.",
9218 record.target_ref,
9219 &head[..12.min(head.len())],
9220 record.author
9221 )));
9222 }
9223 if counted.len() < required {
9224 return Err(ControlError::FailedPrecondition(format!(
9225 "the protection on {} requires {required} distinct verified approvals of {} \
9226 and there {} ({}). Approvals by the same principal count once{}.",
9227 record.target_ref,
9228 &head[..12.min(head.len())],
9229 if counted.len() == 1 {
9230 "is 1".to_owned()
9231 } else {
9232 format!("are {}", counted.len())
9233 },
9234 if counted.is_empty() {
9235 "nobody".to_owned()
9236 } else {
9237 counted.join(", ")
9238 },
9239 if forbid_self {
9240 ", and the author's own does not count here"
9241 } else {
9242 ""
9243 }
9244 )));
9245 }
9246
9247 let approvers: Vec<String> = counted.iter().map(|id| (*id).to_owned()).collect();
9258 let hash = source.kind();
9259 let plan = MergedRef {
9260 ref_name: record.target_ref.clone(),
9261 old_oid: target.to_string(),
9262 new_oid: String::new(),
9263 strategy: S::Unspecified,
9264 head_oid: source.to_string(),
9265 conflicts: Vec::new(),
9266 approvers: approvers.clone(),
9267 rechecked,
9268 contained_in: String::new(),
9269 closed: String::new(),
9270 };
9271 let inconclusive = |err: gunnar_policy::Error| {
9272 ControlError::FailedPrecondition(format!(
9275 "the ancestry walk could not be completed ({err}); refusing rather than guessing"
9276 ))
9277 };
9278
9279 let fast_forward = gunnar_policy::ancestry::is_descendant_of(git, &source, &target, budget)
9280 .map_err(inconclusive)?;
9281 let strategy = choose_strategy(options.strategy, fast_forward, &policy.merge, &record)?;
9282
9283 let principal = self
9284 .stores
9285 .auth()
9286 .principal(principal_id)
9287 .map_err(internal)?
9288 .ok_or_else(|| {
9289 ControlError::FailedPrecondition(format!(
9294 "the principal {principal_id} is no longer in the identity table; \
9295 refusing to move a protected reference for an identity that \
9296 cannot be resolved"
9297 ))
9298 })?;
9299
9300 let mut made = crate::merge_engine::NewObjects::default();
9306 let engine = |err: crate::merge_engine::EngineError| {
9307 ControlError::FailedPrecondition(format!(
9308 "gunnar could not merge {} into {} ({err}); nothing was written",
9309 record.source_ref, record.target_ref
9310 ))
9311 };
9312 let result = match strategy {
9313 S::FastForward => source,
9314 S::MergeCommit | S::Squash => {
9315 let base = crate::merge_engine::merge_base(git, hash, &target, &source, budget)
9316 .map_err(engine)?;
9317 let tree = |commit| crate::merge_engine::commit_tree(git, hash, commit);
9318 let (base_tree, ours, theirs) = (
9319 tree(&base).map_err(engine)?,
9320 tree(&target).map_err(engine)?,
9321 tree(&source).map_err(engine)?,
9322 );
9323 let mut merger = crate::merge_engine::Merger::new(git, hash, repo_record.encrypted);
9324 let merged_tree = match merger
9325 .merge_trees(Some(&base_tree), &ours, &theirs)
9326 .map_err(engine)?
9327 {
9328 crate::merge_engine::TreeMerge::Clean(tree) => tree,
9329 crate::merge_engine::TreeMerge::Conflicts(conflicts) => {
9330 let listed: Vec<String> =
9331 conflicts.iter().map(ToString::to_string).collect();
9332 if dry_run {
9333 return Ok(MergeAttempt::Done(Box::new((
9334 pull_out(path, record.clone()),
9335 MergedRef {
9336 strategy,
9337 conflicts: listed,
9338 ..plan
9339 },
9340 ))));
9341 }
9342 return Err(ControlError::FailedPrecondition(format!(
9343 "merge conflict: {} ({}) and {} ({}) both changed {} {}, and gunnar \
9344 does not guess: {}. Nothing was written. Merge {} into {} locally, \
9345 resolve them, push, and have the new head reviewed.",
9346 record.source_ref,
9347 &plan.head_oid[..12],
9348 record.target_ref,
9349 &plan.old_oid[..12],
9350 listed.len(),
9351 if listed.len() == 1 { "path" } else { "paths" },
9352 listed.join("; "),
9353 short_ref(&record.target_ref),
9354 short_ref(&record.source_ref),
9355 )));
9356 }
9357 };
9358 made = merger.new;
9359 let parents = if strategy == S::MergeCommit {
9360 vec![target, source]
9361 } else {
9362 vec![target]
9363 };
9364 let message = self
9365 .merge_message(strategy, &record, git, &source, &target, &approvers, budget);
9366 let when = std::time::SystemTime::now()
9367 .duration_since(std::time::UNIX_EPOCH)
9368 .map(|d| d.as_secs() as i64)
9369 .unwrap_or(0);
9370 let bytes = crate::merge_engine::commit_bytes(
9371 &merged_tree,
9372 &parents,
9373 &principal_ident(&principal),
9374 &gunnar_ident(),
9375 when,
9376 &message,
9377 );
9378 let key = self
9382 .stores
9383 .merge_signing_key()
9384 .map_err(|err| ControlError::Unavailable(format!("{err:#}")))?;
9385 let bytes = crate::merge_engine::sign_commit(&bytes, hash, &key).map_err(engine)?;
9386 made.add_commit(hash, bytes).map_err(engine)?
9387 }
9388 S::Unspecified => {
9389 return Err(ControlError::Internal(
9390 "the merge strategy was not decided".to_owned(),
9391 ))
9392 }
9393 };
9394 let overlay = crate::merge_engine::Overlay::new(git, &made);
9395
9396 let policy_principal = gunnar_policy::Principal {
9409 id: principal.id.clone(),
9410 name: principal.name.clone(),
9411 kind: crate::transport::policy_principal_kind(principal.kind),
9412 };
9413 let repo_id = gunnar_policy::RepoId::new(&path.namespace, &path.repo);
9414 let introduced = gunnar_policy::ancestry::commits_between(
9418 &overlay, &result, &target, budget,
9419 )
9420 .map_err(|err| {
9421 ControlError::FailedPrecondition(format!(
9422 "the commits this merge would add could not be enumerated ({err}); \
9423 refusing rather than running branch protection over an empty set"
9424 ))
9425 })?;
9426 let updates = [gunnar_policy::RefUpdate::from_wire(
9427 record.target_ref.clone(),
9428 target,
9429 result,
9430 )];
9431 let keys = crate::transport::key_directory(&self.stores);
9432 let ctx = gunnar_policy::PushContext::new(
9433 &policy_principal,
9434 &repo_id,
9435 &updates,
9436 &introduced,
9437 &overlay,
9438 &keys,
9439 );
9440 let mut gates: Vec<(&str, Box<dyn gunnar_policy::Gate>)> = vec![(
9441 "branch protection",
9442 Box::new(gunnar_policy::BranchProtectionGate::new(
9443 policy.protection.clone(),
9444 )),
9445 )];
9446 if policy.signatures.required {
9447 gates.push((
9448 "the signed-commits policy",
9449 Box::new(gunnar_policy::SignedCommitsGate::new(
9450 policy.signatures.clone(),
9451 )),
9452 ));
9453 }
9454 if policy.ai_trailers.refuse {
9455 gates.push((
9456 "the attribution-trailer policy",
9457 Box::new(gunnar_policy::AiTrailerGate::new(
9458 policy.ai_trailers.clone(),
9459 )),
9460 ));
9461 }
9462 for (label, gate) in &gates {
9463 match gate.check(&ctx) {
9464 Ok(gunnar_policy::Verdict::Allow) => {}
9465 Ok(verdict) => {
9466 return Err(ControlError::FailedPrecondition(format!(
9467 "{label} refuses this merge: {}",
9468 verdict.reason().unwrap_or_default()
9469 )));
9470 }
9471 Err(err) => {
9474 return Err(ControlError::FailedPrecondition(format!(
9475 "{label} could not decide this merge ({err}); refusing rather \
9476 than guessing"
9477 )));
9478 }
9479 }
9480 }
9481
9482 if dry_run {
9485 let new_oid = if strategy == S::FastForward {
9486 result.to_string()
9487 } else {
9488 String::new()
9489 };
9490 return Ok(MergeAttempt::Done(Box::new((
9491 pull_out(path, record.clone()),
9492 MergedRef {
9493 strategy,
9494 new_oid,
9495 ..plan
9496 },
9497 ))));
9498 }
9499
9500 if let Some(pack) = made.pack(hash, git).map_err(engine)? {
9502 git.put_pack(&pack).map_err(|err| {
9503 ControlError::Internal(format!(
9504 "storing the merge of pull request #{number}: {err:#}; no ref moved"
9505 ))
9506 })?;
9507 }
9508 if let Err(err) = git.update_ref(
9509 &record.target_ref,
9510 Some(target.as_bytes()),
9511 Some(result.as_bytes()),
9512 ) {
9513 let now = RefSnapshot::read(git)
9516 .map_err(wire_error)?
9517 .resolve(&record.target_ref)
9518 .map_err(wire_error)?;
9519 if now != Some(target) {
9520 return Ok(MergeAttempt::TargetMoved);
9521 }
9522 return Err(ControlError::FailedPrecondition(format!(
9523 "the target ref could not be moved ({err}); no ref moved — retry"
9524 )));
9525 }
9526 let merged = self.mark_merged(&pulls, &key, number, &result.to_string())?;
9527 Ok(MergeAttempt::Done(Box::new((
9528 pull_out(path, merged),
9529 MergedRef {
9530 strategy,
9531 new_oid: result.to_string(),
9532 ..plan
9533 },
9534 ))))
9535 }
9536
9537 fn approvals_enough(
9542 &self,
9543 key: &str,
9544 number: u64,
9545 record: &gunnar_core::PullRecord,
9546 repo_record: &gunnar_core::Repo,
9547 head: &str,
9548 ) -> ControlResult<Option<Vec<String>>> {
9549 let signed = self.forge_events()?;
9550 let events: Vec<gunnar_core::ForgeEvent> = signed
9551 .iter()
9552 .filter(|s| s.event.repo == key)
9553 .map(|s| s.event.clone())
9554 .collect();
9555 let review = gunnar_core::forge::fold(number, &events);
9556 let approved = self.review_state(key, &review, &signed)?;
9557 let policy = self.policy_of(repo_record)?;
9558 let rules: Vec<&gunnar_policy::ProtectionRule> = policy
9559 .protection
9560 .iter()
9561 .filter(|rule| rule.pattern.matches_ref(&record.target_ref))
9562 .collect();
9563 let forbid_self = rules.iter().any(|rule| rule.forbid_self_approval);
9564 let required = rules
9565 .iter()
9566 .map(|rule| rule.required_approvals)
9567 .max()
9568 .unwrap_or(0)
9569 .max(1) as usize;
9570 let counted: Vec<String> = approved
9571 .approvals
9572 .iter()
9573 .filter(|a| a.verified && a.commit == head)
9574 .filter(|a| !(forbid_self && a.actor == record.author))
9575 .map(|a| a.actor.clone())
9576 .collect();
9577 Ok((counted.len() >= required).then_some(counted))
9578 }
9579
9580 fn mark_merged(
9582 &self,
9583 pulls: &gunnar_core::Pulls,
9584 key: &str,
9585 number: u64,
9586 oid: &str,
9587 ) -> ControlResult<gunnar_core::PullRecord> {
9588 match pulls.mark_merged(key, number, oid).map_err(internal)? {
9589 gunnar_core::PullTransition::Done(record) => Ok(record),
9590 other => Err(ControlError::Internal(format!(
9593 "the ref moved to {oid} but the record could not be marked merged \
9594 ({other:?}); the repository is correct, the record is not"
9595 ))),
9596 }
9597 }
9598
9599 #[allow(clippy::too_many_arguments)]
9604 fn merge_message(
9605 &self,
9606 strategy: gunnar_proto::MergeStrategy,
9607 record: &gunnar_core::PullRecord,
9608 git: &dyn GitServe,
9609 source: &gix_hash::ObjectId,
9610 target: &gix_hash::ObjectId,
9611 approvers: &[String],
9612 budget: usize,
9613 ) -> String {
9614 let title = record.title.lines().next().unwrap_or("").trim();
9615 let mut message = if strategy == gunnar_proto::MergeStrategy::Squash {
9616 let mut out = format!(
9617 "{} (#{})\n\nSquashed pull request #{} from {} into {}",
9618 if title.is_empty() {
9619 "Squashed changes"
9620 } else {
9621 title
9622 },
9623 record.number,
9624 record.number,
9625 short_ref(&record.source_ref),
9626 short_ref(&record.target_ref),
9627 );
9628 let mut commits = gunnar_policy::ancestry::commits_between(git, source, target, budget)
9629 .unwrap_or_default();
9630 commits.reverse();
9631 let subjects = crate::merge_engine::subjects(git, source.kind(), &commits);
9632 if !subjects.is_empty() {
9633 out.push_str(":\n");
9634 for subject in subjects.iter().take(100) {
9635 out.push_str(&format!("\n* {subject}"));
9636 }
9637 }
9638 out
9639 } else {
9640 let mut out = format!(
9641 "Merge pull request #{} from {} into {}",
9642 record.number,
9643 short_ref(&record.source_ref),
9644 short_ref(&record.target_ref),
9645 );
9646 if !title.is_empty() {
9647 out.push_str(&format!("\n\n{title}"));
9648 }
9649 out
9650 };
9651 message.push_str(&format!(
9652 "\n\nPull-request: #{}\nHead: {source}\nTarget: {target}",
9653 record.number
9654 ));
9655 for id in approvers {
9656 let ident = match self.stores.auth().principal(id) {
9657 Ok(Some(p)) => principal_ident(&p),
9658 _ => crate::merge_engine::Ident {
9659 name: id.clone(),
9660 email: format!("{id}@gunnar.invalid"),
9661 },
9662 };
9663 message.push_str(&format!("\nApproved-by: {}", ident.line()));
9664 }
9665 message.push('\n');
9666 message
9667 }
9668}