use std::{
ffi::{CStr, CString},
fs, io,
os::{
fd::AsRawFd,
unix::fs::{FileTypeExt, MetadataExt, PermissionsExt},
},
path::{Path, PathBuf},
sync::Arc,
};
use mercurio_server::server::{ListenerConfig, Transport};
use tokio::net::{unix::UCred, UnixListener};
use tracing::{info, warn};
use crate::config::AdminConfig;
const MAX_LOOKUP_BUFFER: usize = 1 << 20;
pub struct AdminSocket {
path: PathBuf,
device: u64,
inode: u64,
}
impl Drop for AdminSocket {
fn drop(&mut self) {
let ours = fs::symlink_metadata(&self.path)
.is_ok_and(|meta| meta.dev() == self.device && meta.ino() == self.inode);
if ours {
let _ = fs::remove_file(&self.path);
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct Group {
gid: u32,
members: Vec<u32>,
}
struct Admitted {
own_uid: u32,
group: Option<Group>,
}
impl Admitted {
fn admits(&self, uid: u32, gid: u32) -> bool {
uid == 0
|| uid == self.own_uid
|| self
.group
.as_ref()
.is_some_and(|group| group.gid == gid || group.members.contains(&uid))
}
}
pub fn open(config: &AdminConfig) -> Option<(ListenerConfig, AdminSocket)> {
if config.socket.is_empty() {
return None;
}
let path = Path::new(&config.socket);
match bind(path, &config.group) {
Ok(opened) => {
info!("Serving the admin socket on {}", path.display());
Some(opened)
}
Err(e) => {
warn!("Running without the admin socket {}: {}", path.display(), e);
None
}
}
}
fn bind(path: &Path, group_name: &str) -> Result<(ListenerConfig, AdminSocket), String> {
let own_uid = unsafe { libc::geteuid() };
check_parent(path, own_uid)?;
match fs::symlink_metadata(path) {
Ok(meta) if meta.file_type().is_socket() => {
fs::remove_file(path).map_err(|e| format!("can't remove the socket left there: {e}"))?
}
Ok(_) => return Err("something other than a socket is there".to_string()),
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e.to_string()),
}
let group = admin_group(group_name);
let listener = bind_owner_only(path).map_err(|e| e.to_string())?;
let meta = fs::symlink_metadata(path).map_err(|e| e.to_string())?;
let socket = AdminSocket {
path: path.to_path_buf(),
device: meta.dev(),
inode: meta.ino(),
};
if meta.mode() & 0o077 != 0 {
return Err("the socket was not created owner-only".to_string());
}
if let Some(group) = &group {
match std::os::unix::fs::lchown(path, None, Some(group.gid)) {
Ok(()) => fs::set_permissions(path, fs::Permissions::from_mode(0o660))
.map_err(|e| e.to_string())?,
Err(e) => warn!(
"The admin socket stays owner-only, so `{}` can't open it: {}",
group_name, e
),
}
}
let admitted = Admitted { own_uid, group };
let mut config = ListenerConfig::new(listener, Transport::Mqtt);
config.trusted = true;
config.peer_filter = Some(Arc::new(move |peer: &UCred| {
admitted.admits(peer.uid(), peer.gid())
}));
Ok((config, socket))
}
fn bind_owner_only(path: &Path) -> io::Result<UnixListener> {
let socket = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None)?;
if unsafe { libc::fchmod(socket.as_raw_fd(), 0o600) } != 0 {
return Err(io::Error::last_os_error());
}
socket.bind(&socket2::SockAddr::unix(path)?)?;
socket.listen(1024)?;
socket.set_nonblocking(true)?;
UnixListener::from_std(socket.into())
}
fn check_parent(path: &Path, own_uid: u32) -> Result<(), String> {
let parent = match path.parent() {
Some(parent) if !parent.as_os_str().is_empty() => parent,
_ => Path::new("."),
};
let meta = fs::metadata(parent).map_err(|e| format!("{}: {e}", parent.display()))?;
if meta.uid() != 0 && meta.uid() != own_uid {
return Err(format!("{} belongs to another user", parent.display()));
}
if meta.mode() & 0o022 != 0 {
return Err(format!("{} is writable by other users", parent.display()));
}
Ok(())
}
fn admin_group(name: &str) -> Option<Group> {
if name.is_empty() {
return None;
}
match lookup_group(name) {
Ok(Some(group)) => Some(group),
Ok(None) => {
warn!(
"Group `{}` doesn't exist; only root and the broker's user may use the admin socket",
name
);
None
}
Err(e) => {
warn!(
"Can't read group `{}`, so only root and the broker's user may use the admin socket: {}",
name, e
);
None
}
}
}
fn lookup_group(name: &str) -> io::Result<Option<Group>> {
let name = CString::new(name).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?;
let mut buffer: Vec<libc::c_char> = vec![0; 1024];
loop {
let mut entry: libc::group = unsafe { std::mem::zeroed() };
let mut found: *mut libc::group = std::ptr::null_mut();
let code = unsafe {
libc::getgrnam_r(
name.as_ptr(),
&mut entry,
buffer.as_mut_ptr(),
buffer.len(),
&mut found,
)
};
if code == libc::ERANGE && buffer.len() < MAX_LOOKUP_BUFFER {
buffer.resize(buffer.len() * 2, 0);
continue;
}
if code != 0 {
return Err(io::Error::from_raw_os_error(code));
}
if found.is_null() {
return Ok(None);
}
let mut names = Vec::new();
let mut member = entry.gr_mem;
while !member.is_null() && unsafe { !(*member).is_null() } {
names.push(unsafe { CStr::from_ptr(*member) }.to_owned());
member = unsafe { member.add(1) };
}
let members = names
.iter()
.filter_map(|name| lookup_user(name).ok().flatten())
.collect();
return Ok(Some(Group {
gid: entry.gr_gid,
members,
}));
}
}
fn lookup_user(name: &CStr) -> io::Result<Option<u32>> {
let mut buffer: Vec<libc::c_char> = vec![0; 1024];
loop {
let mut entry: libc::passwd = unsafe { std::mem::zeroed() };
let mut found: *mut libc::passwd = std::ptr::null_mut();
let code = unsafe {
libc::getpwnam_r(
name.as_ptr(),
&mut entry,
buffer.as_mut_ptr(),
buffer.len(),
&mut found,
)
};
if code == libc::ERANGE && buffer.len() < MAX_LOOKUP_BUFFER {
buffer.resize(buffer.len() * 2, 0);
continue;
}
if code != 0 {
return Err(io::Error::from_raw_os_error(code));
}
return Ok((!found.is_null()).then_some(entry.pw_uid));
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::net::UnixStream;
fn private_dir(name: &str) -> PathBuf {
let dir =
std::env::temp_dir().join(format!("mercuriod-admin-{}-{name}", std::process::id()));
let _ = fs::remove_dir_all(&dir);
fs::create_dir(&dir).unwrap();
fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).unwrap();
dir
}
#[test]
fn root_the_brokers_user_and_the_group_are_admitted() {
let admitted = Admitted {
own_uid: 1000,
group: Some(Group {
gid: 500,
members: vec![1001],
}),
};
assert!(admitted.admits(0, 0));
assert!(admitted.admits(1000, 1000));
assert!(admitted.admits(1001, 1001), "a listed member");
assert!(admitted.admits(1002, 500), "the group is its primary group");
assert!(!admitted.admits(1002, 1002));
let alone = Admitted {
own_uid: 1000,
group: None,
};
assert!(alone.admits(0, 0));
assert!(!alone.admits(1001, 500));
}
#[test]
fn groups_and_users_are_looked_up_by_name() {
assert_eq!(lookup_group("root").unwrap().map(|g| g.gid), Some(0));
assert_eq!(lookup_group("mercurio-no-such-group").unwrap(), None);
assert_eq!(lookup_user(c"root").unwrap(), Some(0));
assert_eq!(lookup_user(c"mercurio-no-such-user").unwrap(), None);
}
#[tokio::test]
async fn the_socket_is_private_and_admits_the_brokers_user() {
let dir = private_dir("mode");
let path = dir.join("admin.sock");
let (config, socket) = bind(&path, "").unwrap();
let meta = fs::symlink_metadata(&path).unwrap();
assert!(meta.file_type().is_socket());
assert_eq!(meta.mode() & 0o777, 0o600);
assert!(config.trusted);
let client = UnixStream::connect(&path).await.unwrap();
let peer = client.peer_cred().unwrap();
let filter = config.peer_filter.as_ref().expect("a peer filter");
assert!(filter(&peer));
drop(socket);
assert!(!path.exists(), "removed when the broker stops");
fs::remove_dir_all(&dir).unwrap();
}
fn own_group_name() -> Option<String> {
let mut buffer: Vec<libc::c_char> = vec![0; 4096];
let mut entry: libc::group = unsafe { std::mem::zeroed() };
let mut found: *mut libc::group = std::ptr::null_mut();
let code = unsafe {
libc::getgrgid_r(
libc::getgid(),
&mut entry,
buffer.as_mut_ptr(),
buffer.len(),
&mut found,
)
};
if code != 0 || found.is_null() {
return None;
}
Some(
unsafe { CStr::from_ptr(entry.gr_name) }
.to_string_lossy()
.into_owned(),
)
}
#[tokio::test]
async fn the_socket_opens_to_the_group_once_it_belongs_to_it() {
let Some(name) = own_group_name() else {
return;
};
let dir = private_dir("group");
let path = dir.join("admin.sock");
let (_config, _socket) = bind(&path, &name).unwrap();
let meta = fs::symlink_metadata(&path).unwrap();
assert_eq!(meta.mode() & 0o777, 0o660);
assert_eq!(meta.gid(), unsafe { libc::getgid() });
fs::remove_dir_all(&dir).unwrap();
}
#[tokio::test]
async fn the_socket_is_born_owner_only() {
let dir = private_dir("born");
let path = dir.join("admin.sock");
let _listener = bind_owner_only(&path).unwrap();
let meta = fs::symlink_metadata(&path).unwrap();
assert!(meta.file_type().is_socket());
assert_eq!(meta.mode() & 0o077, 0);
fs::remove_dir_all(&dir).unwrap();
}
#[tokio::test]
async fn a_stale_socket_is_replaced_and_a_newer_one_kept() {
let dir = private_dir("stale");
let path = dir.join("admin.sock");
drop(std::os::unix::net::UnixListener::bind(&path).unwrap());
let (_config, first) = bind(&path, "").unwrap();
fs::remove_file(&path).unwrap();
let _newer = std::os::unix::net::UnixListener::bind(&path).unwrap();
drop(first);
assert!(path.exists());
fs::remove_dir_all(&dir).unwrap();
}
#[tokio::test]
async fn unsafe_or_missing_places_are_refused() {
let dir = private_dir("refused");
let file = dir.join("file");
fs::write(&file, b"keep").unwrap();
assert!(bind(&file, "").is_err());
assert_eq!(fs::read(&file).unwrap(), b"keep");
let shared = dir.join("shared");
fs::create_dir(&shared).unwrap();
fs::set_permissions(&shared, fs::Permissions::from_mode(0o777)).unwrap();
assert!(bind(&shared.join("admin.sock"), "").is_err());
assert!(!shared.join("admin.sock").exists());
assert!(bind(&dir.join("missing/admin.sock"), "").is_err());
let off = AdminConfig {
socket: String::new(),
group: String::new(),
};
assert!(open(&off).is_none());
fs::remove_dir_all(&dir).unwrap();
}
}