1#![cfg(unix)]
31
32use std::{
33 path::{Path, PathBuf},
34 sync::{Arc, Mutex},
35};
36
37use grpc::{
38 DiscussionServiceServer, HookServiceServer, OperationLogQueryServiceServer,
39 SignalServiceServer, StateReviewServiceServer, TimelineServiceServer, TransactionServiceServer,
40};
41use objects::error::{HeddleError, Result};
42use repo::{Repository, operation_dedup::OperationDedupStore};
43use tokio::net::UnixListener;
44use tokio_stream::{StreamExt, wrappers::UnixListenerStream};
45use tonic::transport::Server;
46
47use crate::grpc_local_impl::{
48 GrpcLocalService, LocalDiscussionService, LocalHookService, LocalOperationLogQueryService,
49 LocalSignalService, LocalStateReviewService, LocalTimelineService, LocalTransactionService,
50};
51
52const PRIVATE_SOCKET_UMASK: libc::mode_t = 0o177;
53
54static SOCKET_BIND_UMASK_LOCK: Mutex<()> = Mutex::new(());
55
56pub fn default_socket_path(heddle_dir: &Path) -> PathBuf {
58 heddle_dir.join("sockets").join("grpc.sock")
59}
60
61pub fn default_pid_path(heddle_dir: &Path) -> PathBuf {
63 heddle_dir.join("sockets").join("grpc.pid")
64}
65
66pub struct LocalDaemonConfig {
69 pub socket_path: PathBuf,
70 pub pid_path: PathBuf,
71}
72
73impl LocalDaemonConfig {
74 pub fn from_repo(repo: &Repository) -> Self {
75 let heddle_dir = repo.heddle_dir();
76 Self {
77 socket_path: default_socket_path(heddle_dir),
78 pid_path: default_pid_path(heddle_dir),
79 }
80 }
81
82 pub fn with_socket(mut self, path: PathBuf) -> Self {
83 self.socket_path = path;
84 self
85 }
86}
87
88struct PidGuard {
91 pid_path: PathBuf,
92 socket_path: PathBuf,
93}
94
95pub const PIDFILE_MARKER: &str = "heddle-agent";
99
100#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct PidFileContents {
116 pub pid: i32,
117 pub started_at_secs: i64,
118}
119
120impl PidFileContents {
121 pub fn render(&self) -> String {
123 format!(
124 "{}\n{}\n{}\n",
125 self.pid, PIDFILE_MARKER, self.started_at_secs
126 )
127 }
128
129 pub fn parse(body: &str) -> Option<Self> {
133 let mut lines = body.lines();
134 let pid = lines.next()?.trim().parse::<i32>().ok()?;
135 let marker = lines.next()?.trim();
136 if marker != PIDFILE_MARKER {
137 return None;
138 }
139 let started_at_secs = lines.next()?.trim().parse::<i64>().ok()?;
140 Some(Self {
141 pid,
142 started_at_secs,
143 })
144 }
145}
146
147impl PidGuard {
148 fn install(pid_path: PathBuf, socket_path: PathBuf) -> Result<Self> {
149 if let Some(parent) = pid_path.parent() {
150 std::fs::create_dir_all(parent)?;
151 }
152 if pid_path.exists() {
159 let raw = std::fs::read_to_string(&pid_path).ok();
160 let parsed = raw.as_deref().and_then(PidFileContents::parse);
161 if let Some(existing) = parsed
162 && pid_alive(existing.pid)
163 && is_heddle_process(existing.pid)
164 {
165 return Err(HeddleError::Conflict(format!(
166 "heddle agent serve already running on this repo (pid {}); \
167 stop it first or remove {} if it's stale",
168 existing.pid,
169 pid_path.display()
170 )));
171 }
172 let _ = std::fs::remove_file(&pid_path);
174 if socket_path.exists() {
175 let _ = std::fs::remove_file(&socket_path);
176 }
177 }
178 let contents = PidFileContents {
180 pid: std::process::id() as i32,
181 started_at_secs: std::time::SystemTime::now()
182 .duration_since(std::time::UNIX_EPOCH)
183 .map(|d| d.as_secs() as i64)
184 .unwrap_or(0),
185 };
186 std::fs::write(&pid_path, contents.render())?;
187 Ok(Self {
188 pid_path,
189 socket_path,
190 })
191 }
192}
193
194impl Drop for PidGuard {
195 fn drop(&mut self) {
196 let _ = std::fs::remove_file(&self.pid_path);
197 let _ = std::fs::remove_file(&self.socket_path);
198 }
199}
200
201#[cfg(any(target_os = "linux", target_os = "macos"))]
202pub fn pid_alive(pid: i32) -> bool {
203 unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
207}
208
209#[cfg(not(any(target_os = "linux", target_os = "macos")))]
210pub fn pid_alive(_pid: i32) -> bool {
211 true
214}
215
216pub fn is_heddle_process(pid: i32) -> bool {
230 process_uid_matches_self(pid) && process_exe_matches_current(pid)
231}
232
233#[cfg(target_os = "linux")]
234fn process_uid_matches_self(pid: i32) -> bool {
235 use std::os::unix::fs::MetadataExt;
236
237 let path = PathBuf::from(format!("/proc/{pid}"));
238 let Ok(metadata) = std::fs::metadata(path) else {
239 return false;
240 };
241 metadata.uid() == unsafe { libc::geteuid() }
243}
244
245#[cfg(target_os = "macos")]
250fn process_uid_matches_self(pid: i32) -> bool {
251 let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() };
252 let size = std::mem::size_of::<libc::proc_bsdinfo>() as i32;
253 let ret = unsafe {
256 libc::proc_pidinfo(
257 pid,
258 libc::PROC_PIDTBSDINFO,
259 0,
260 &mut info as *mut _ as *mut libc::c_void,
261 size,
262 )
263 };
264 if ret <= 0 || ret < size {
265 return false;
266 }
267 info.pbi_uid == unsafe { libc::geteuid() }
269}
270
271#[cfg(not(any(target_os = "linux", target_os = "macos")))]
276fn process_uid_matches_self(_pid: i32) -> bool {
277 false
278}
279
280fn process_exe_matches_current(pid: i32) -> bool {
281 let Some(process_exe) = process_exe_path(pid) else {
282 return false;
283 };
284 let Ok(current_exe) = std::env::current_exe() else {
285 return false;
286 };
287 executable_identity_matches(&process_exe, ¤t_exe)
288}
289
290fn executable_identity_matches(process_exe: &Path, current_exe: &Path) -> bool {
291 let Ok(process_exe) = process_exe.canonicalize() else {
292 return false;
293 };
294 let Ok(current_exe) = current_exe.canonicalize() else {
295 return false;
296 };
297 process_exe == current_exe
298}
299
300#[cfg(target_os = "linux")]
301fn process_exe_path(pid: i32) -> Option<PathBuf> {
302 std::fs::read_link(format!("/proc/{pid}/exe")).ok()
303}
304
305#[cfg(target_os = "macos")]
306fn process_exe_path(pid: i32) -> Option<PathBuf> {
307 use std::{ffi::OsString, os::unix::ffi::OsStringExt};
308
309 let mut buf = vec![0u8; libc::PROC_PIDPATHINFO_MAXSIZE as usize];
310 let len = unsafe { libc::proc_pidpath(pid, buf.as_mut_ptr() as *mut _, buf.len() as u32) };
312 if len <= 0 {
313 return None;
314 }
315 Some(PathBuf::from(OsString::from_vec(
316 buf[..len as usize].to_vec(),
317 )))
318}
319
320#[cfg(not(any(target_os = "linux", target_os = "macos")))]
321fn process_exe_path(_pid: i32) -> Option<PathBuf> {
322 None
323}
324
325pub async fn serve(
328 repo: Repository,
329 config: LocalDaemonConfig,
330 shutdown: impl std::future::Future<Output = ()> + Send + 'static,
331) -> Result<()> {
332 create_private_socket_parent(&config.socket_path)?;
333 let _guard = PidGuard::install(config.pid_path.clone(), config.socket_path.clone())?;
335
336 if config.socket_path.exists() {
339 std::fs::remove_file(&config.socket_path)?;
340 }
341 let listener = bind_private_unix_listener(&config.socket_path)?;
342 set_socket_mode_0600(&config.socket_path)?;
347 listener.set_nonblocking(true).map_err(|e| {
348 HeddleError::Io(std::io::Error::new(
349 e.kind(),
350 format!(
351 "UnixListener::set_nonblocking({}): {e}",
352 config.socket_path.display()
353 ),
354 ))
355 })?;
356 let listener = UnixListener::from_std(listener).map_err(|e| {
357 HeddleError::Io(std::io::Error::new(
358 e.kind(),
359 format!(
360 "UnixListener::from_std({}): {e}",
361 config.socket_path.display()
362 ),
363 ))
364 })?;
365
366 let report = crate::transaction_replay::replay_active_transactions(&repo);
382 if report.has_hard_failures() {
383 tracing::error!(
384 recovered_txns = report.recovered_transaction_ids.len(),
385 orphan_tmps = report.orphan_temp_files_removed,
386 unparseable = report.unparseable_sentinels.len(),
387 failed_sentinel_writes = report.failed_sentinel_writes.len(),
388 failed_orphan_deletes = report.failed_orphan_deletes.len(),
389 failed_oplog_appends = report.failed_oplog_appends.len(),
390 unreadable_entries = report.unreadable_entries,
391 scan_error = report.scan_error.as_deref().unwrap_or(""),
392 "local-daemon: transaction replay hit hard failures; \
393 scan may not have run or audit-trail entries were lost"
394 );
395 } else if report.has_recoverable_failures() {
396 tracing::warn!(
397 recovered_txns = report.recovered_transaction_ids.len(),
398 orphan_tmps = report.orphan_temp_files_removed,
399 unparseable = report.unparseable_sentinels.len(),
400 failed_sentinel_writes = report.failed_sentinel_writes.len(),
401 failed_orphan_deletes = report.failed_orphan_deletes.len(),
402 unreadable_entries = report.unreadable_entries,
403 "local-daemon: transaction replay left recoverable failures on disk; \
404 next startup will retry, but operator inspection is recommended"
405 );
406 } else if !report.is_clean() {
407 tracing::info!(
408 recovered_txns = report.recovered_transaction_ids.len(),
409 orphan_tmps = report.orphan_temp_files_removed,
410 "local-daemon: transaction replay recovered prior in-flight state"
411 );
412 }
413
414 let dedup = Arc::new(OperationDedupStore::open(repo.heddle_dir())?);
415 let inner = GrpcLocalService::new(Arc::new(repo), dedup);
416
417 let state_review = StateReviewServiceServer::new(LocalStateReviewService::new(inner.clone()));
418 let discussion = DiscussionServiceServer::new(LocalDiscussionService::new(inner.clone()));
419 let signal = SignalServiceServer::new(LocalSignalService::new(inner.clone()));
420 let query =
421 OperationLogQueryServiceServer::new(LocalOperationLogQueryService::new(inner.clone()));
422 let timeline = TimelineServiceServer::new(LocalTimelineService::new(inner.clone()));
423 let transaction = TransactionServiceServer::new(LocalTransactionService::new(inner.clone()));
424 let hook = HookServiceServer::new(LocalHookService::new(inner));
425
426 let incoming = UnixListenerStream::new(listener).filter_map(guard_peer_connection);
432
433 Server::builder()
434 .add_service(state_review)
435 .add_service(discussion)
436 .add_service(signal)
437 .add_service(query)
438 .add_service(timeline)
439 .add_service(transaction)
440 .add_service(hook)
441 .serve_with_incoming_shutdown(incoming, shutdown)
442 .await
443 .map_err(|e| HeddleError::InvalidObject(format!("local daemon transport failed: {e}")))?;
444 Ok(())
445}
446
447fn create_private_socket_parent(socket_path: &Path) -> Result<()> {
448 if let Some(parent) = socket_path.parent() {
449 use std::os::unix::fs::DirBuilderExt;
450 let mut builder = std::fs::DirBuilder::new();
451 builder.recursive(true).mode(0o700);
452 builder.create(parent)?;
453 }
454 Ok(())
455}
456
457fn bind_private_unix_listener(socket_path: &Path) -> Result<std::os::unix::net::UnixListener> {
458 let _lock = SOCKET_BIND_UMASK_LOCK
459 .lock()
460 .map_err(|_| HeddleError::InvalidObject("daemon socket umask lock poisoned".to_string()))?;
461 let _umask = UmaskGuard::set(PRIVATE_SOCKET_UMASK);
469 std::os::unix::net::UnixListener::bind(socket_path).map_err(|e| {
470 HeddleError::Io(std::io::Error::new(
471 e.kind(),
472 format!("UnixListener::bind({}): {e}", socket_path.display()),
473 ))
474 })
475}
476
477struct UmaskGuard {
478 previous: libc::mode_t,
479}
480
481impl UmaskGuard {
482 fn set(mask: libc::mode_t) -> Self {
483 let previous = unsafe { libc::umask(mask) };
487 Self { previous }
488 }
489}
490
491impl Drop for UmaskGuard {
492 fn drop(&mut self) {
493 unsafe {
495 libc::umask(self.previous);
496 }
497 }
498}
499
500#[cfg(unix)]
501fn set_socket_mode_0600(path: &Path) -> Result<()> {
502 use std::os::unix::fs::PermissionsExt;
503 let permissions = std::fs::Permissions::from_mode(0o600);
504 std::fs::set_permissions(path, permissions)?;
505 Ok(())
506}
507
508pub fn check_peer_uid_matches_self(stream: &tokio::net::UnixStream) -> Result<()> {
516 let creds = stream
517 .peer_cred()
518 .map_err(|e| HeddleError::InvalidObject(format!("peer_cred failed: {e}")))?;
519 let our_uid = unsafe { libc::geteuid() };
521 enforce_peer_uid(creds.uid(), our_uid)
522}
523
524fn enforce_peer_uid(peer_uid: u32, our_uid: u32) -> Result<()> {
529 if peer_uid != our_uid {
530 return Err(HeddleError::Conflict(format!(
531 "peer uid {peer_uid} does not match daemon uid {our_uid}"
532 )));
533 }
534 Ok(())
535}
536
537fn guard_peer_connection(
543 conn: std::io::Result<tokio::net::UnixStream>,
544) -> Option<std::io::Result<tokio::net::UnixStream>> {
545 match conn {
546 Ok(stream) => match check_peer_uid_matches_self(&stream) {
547 Ok(()) => Some(Ok(stream)),
548 Err(e) => {
549 tracing::warn!(
550 error = %e,
551 "local-daemon: rejecting connection from peer with mismatched uid"
552 );
553 None
554 }
555 },
556 Err(e) => Some(Err(e)),
557 }
558}
559
560#[cfg(test)]
561mod tests {
562 use tempfile::TempDir;
563
564 use super::*;
565
566 #[test]
567 #[serial_test::serial(process_global)]
568 fn default_socket_path_lives_under_heddle_dir() {
569 let temp = TempDir::new().unwrap();
570 let heddle = temp.path().join(".heddle");
571 std::fs::create_dir_all(&heddle).unwrap();
572 let path = default_socket_path(&heddle);
573 assert!(path.starts_with(&heddle));
574 assert!(path.ends_with("grpc.sock"));
575 }
576
577 #[test]
578 #[serial_test::serial(process_global)]
579 fn create_private_socket_parent_creates_new_parent_0700() {
580 use std::os::unix::fs::PermissionsExt;
581
582 let temp = TempDir::new().unwrap();
583 let socket = temp
584 .path()
585 .join(".heddle")
586 .join("sockets")
587 .join("grpc.sock");
588 create_private_socket_parent(&socket).unwrap();
589
590 let mode = std::fs::metadata(socket.parent().unwrap())
591 .unwrap()
592 .permissions()
593 .mode()
594 & 0o777;
595 assert_eq!(mode, 0o700, "new socket parent must be private");
596 }
597
598 #[test]
599 #[serial_test::serial(process_global)]
600 fn bind_private_unix_listener_creates_socket_0600_before_chmod() {
601 use std::os::unix::fs::PermissionsExt;
602
603 let temp = TempDir::new().unwrap();
604 let socket = temp.path().join("grpc.sock");
605
606 let _listener = match bind_private_unix_listener(&socket) {
607 Ok(listener) => listener,
608 Err(HeddleError::Io(err)) if err.kind() == std::io::ErrorKind::PermissionDenied => {
609 eprintln!(
610 "skipping daemon socket mode test: local Unix listener bind denied: {err}"
611 );
612 return;
613 }
614 Err(err) => panic!("bind private Unix listener: {err}"),
615 };
616
617 let mode = std::fs::metadata(&socket).unwrap().permissions().mode() & 0o777;
618 assert_eq!(
619 mode, 0o600,
620 "socket must be born private before set_socket_mode_0600 runs"
621 );
622 }
623
624 #[test]
625 #[serial_test::serial(process_global)]
626 fn bind_private_unix_listener_restores_umask_after_bind_error() {
627 let temp = TempDir::new().unwrap();
628 let socket = temp.path().join("missing").join("grpc.sock");
629 let before = current_umask();
630
631 let result = bind_private_unix_listener(&socket);
632
633 let after = current_umask();
634 assert!(result.is_err(), "bind should fail for a missing parent");
635 assert_eq!(after, before, "bind errors must restore the prior umask");
636 }
637
638 fn current_umask() -> libc::mode_t {
639 unsafe {
642 let current = libc::umask(0);
643 libc::umask(current);
644 current
645 }
646 }
647
648 #[test]
649 #[serial_test::serial(process_global)]
650 fn pid_guard_writes_and_removes_pidfile() {
651 let temp = TempDir::new().unwrap();
652 let pid = temp.path().join("grpc.pid");
653 let sock = temp.path().join("grpc.sock");
654 let guard = PidGuard::install(pid.clone(), sock.clone()).unwrap();
655 assert!(pid.exists());
656 drop(guard);
657 assert!(!pid.exists());
658 assert!(!sock.exists());
659 }
660
661 #[test]
662 #[serial_test::serial(process_global)]
663 fn pid_guard_refuses_when_live_heddle_process_owns_pidfile() {
664 let temp = TempDir::new().unwrap();
665 let pid = temp.path().join("grpc.pid");
666 let sock = temp.path().join("grpc.sock");
667 let first = PidGuard::install(pid.clone(), sock.clone()).unwrap();
671 let result = PidGuard::install(pid.clone(), sock.clone());
672 assert!(result.is_err(), "expected refusal for live owner");
673 drop(first);
674 }
675
676 #[test]
677 #[serial_test::serial(process_global)]
678 fn pid_guard_sweeps_stale_pidfile_with_dead_pid() {
679 let temp = TempDir::new().unwrap();
680 let pid = temp.path().join("grpc.pid");
681 let sock = temp.path().join("grpc.sock");
682 let stale = PidFileContents {
684 pid: 2_147_483_646,
685 started_at_secs: 0,
686 };
687 std::fs::write(&pid, stale.render()).unwrap();
688 std::fs::write(&sock, "stale").unwrap();
689 let _guard = PidGuard::install(pid.clone(), sock.clone()).unwrap();
690 let raw = std::fs::read_to_string(&pid).unwrap();
692 let parsed = PidFileContents::parse(&raw).expect("guard wrote structured pidfile");
693 assert_eq!(parsed.pid, std::process::id() as i32);
694 assert!(parsed.started_at_secs > 0);
695 }
696
697 #[test]
698 #[serial_test::serial(process_global)]
699 fn pid_guard_sweeps_legacy_unstructured_pidfile() {
700 let temp = TempDir::new().unwrap();
704 let pid = temp.path().join("grpc.pid");
705 let sock = temp.path().join("grpc.sock");
706 std::fs::write(&pid, "12345").unwrap();
707 let _guard = PidGuard::install(pid.clone(), sock.clone()).unwrap();
708 let parsed = PidFileContents::parse(&std::fs::read_to_string(&pid).unwrap()).unwrap();
709 assert_eq!(parsed.pid, std::process::id() as i32);
710 }
711
712 #[test]
713 fn pidfile_contents_round_trip() {
714 let original = PidFileContents {
715 pid: 4321,
716 started_at_secs: 1_700_000_000,
717 };
718 let body = original.render();
719 let parsed = PidFileContents::parse(&body).expect("round-trip");
720 assert_eq!(parsed, original);
721 }
722
723 #[test]
724 fn pidfile_contents_rejects_missing_marker() {
725 let body = "1234\nnot-heddle-agent\n100\n";
728 assert!(PidFileContents::parse(body).is_none());
729 }
730
731 #[test]
732 fn pidfile_contents_rejects_bare_pid() {
733 assert!(PidFileContents::parse("12345").is_none());
736 }
737
738 #[test]
739 fn executable_identity_accepts_same_canonical_path() {
740 let current = std::env::current_exe().unwrap();
741 assert!(executable_identity_matches(¤t, ¤t));
742 }
743
744 #[test]
745 fn executable_identity_rejects_spoofed_heddle_path() {
746 let temp = TempDir::new().unwrap();
747 let spoofed = temp.path().join("contains-heddle").join("heddle-spoof");
748 std::fs::create_dir_all(spoofed.parent().unwrap()).unwrap();
749 std::fs::write(&spoofed, "not the current executable").unwrap();
750
751 let current = std::env::current_exe().unwrap();
752
753 assert!(
754 !executable_identity_matches(&spoofed, ¤t),
755 "a pathname containing heddle must not satisfy executable identity"
756 );
757 }
758
759 #[test]
760 fn is_heddle_process_accepts_self_pid() {
761 assert!(
762 is_heddle_process(std::process::id() as i32),
763 "the current process should resolve to the current executable"
764 );
765 }
766
767 #[test]
768 fn enforce_peer_uid_admits_matching_uid() {
769 assert!(enforce_peer_uid(1000, 1000).is_ok());
772 }
773
774 #[test]
775 fn enforce_peer_uid_rejects_mismatched_uid() {
776 let err = enforce_peer_uid(1001, 1000).unwrap_err();
779 assert!(
780 matches!(err, HeddleError::Conflict(_)),
781 "mismatched peer uid must be a Conflict, got {err:?}"
782 );
783 }
784
785 #[test]
786 fn guard_propagates_listener_io_errors() {
787 let io_err = std::io::Error::other("accept failed");
791 let out = guard_peer_connection(Err(io_err));
792 assert!(matches!(out, Some(Err(_))), "io errors must propagate");
793 }
794
795 #[tokio::test]
796 async fn guard_admits_same_process_peer() {
797 let (peer, _local) = tokio::net::UnixStream::pair().expect("socketpair");
801 let out = guard_peer_connection(Ok(peer));
802 assert!(
803 matches!(out, Some(Ok(_))),
804 "a same-uid peer must be admitted by the gate"
805 );
806 }
807
808 #[tokio::test]
809 async fn check_peer_uid_matches_self_admits_socketpair() {
810 let (peer, _local) = tokio::net::UnixStream::pair().expect("socketpair");
813 assert!(check_peer_uid_matches_self(&peer).is_ok());
814 }
815}