Lines
0 %
Functions
use std::{os::fd::OwnedFd, time::Duration};
use anyhow::Context;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use tokio::time::timeout;
use zlink::{ReplyError, service::MethodReply};
use crate::{
proto::{WhodStatusUpdate, WhodUserEntry, finger_protocol::FingerResponseUserEntry},
server::{
fingerd::{self, FingerRequestInfo, FingerRequestNetworking, finger_utmp_users},
ignore_list::IgnoreList,
rwhod::RwhodStatusStore,
},
};
// Types for 'no.ntnu.pvv.roowho2.rwhod'
#[zlink::proxy("no.ntnu.pvv.roowho2.rwhod")]
pub trait VarlinkRwhodClientProxy {
async fn rwho(
&mut self,
all: bool,
) -> zlink::Result<Result<VarlinkRwhoResponse, VarlinkRwhodClientError>>;
async fn ruptime(
) -> zlink::Result<Result<VarlinkRuptimeResponse, VarlinkRwhodClientError>>;
}
#[derive(Debug, Deserialize)]
#[serde(tag = "method", content = "parameters")]
pub enum VarlinkRwhodClientRequest {
#[serde(rename = "no.ntnu.pvv.roowho2.rwhod.Rwho")]
Rwho {
/// Retrieve all users, even those that have been idle for a long time.
#[serde(rename = "no.ntnu.pvv.roowho2.rwhod.Ruptime")]
Ruptime {
/// Count all users, even those that have been idle for a long time.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(untagged)]
pub enum VarlinkRwhodClientResponse {
Rwho(VarlinkRwhoResponse),
Ruptime(VarlinkRuptimeResponse),
pub type VarlinkRwhoResponse = Vec<(String, WhodUserEntry)>;
pub type VarlinkRuptimeResponse = Vec<WhodStatusUpdate>;
#[derive(Debug, Clone, PartialEq, ReplyError)]
#[zlink(interface = "no.ntnu.pvv.roowho2.rwhod")]
pub enum VarlinkRwhodClientError {
InvalidRequest,
TimedOut,
Disabled,
// Types for 'no.ntnu.pvv.roowho2.finger'
#[zlink::proxy("no.ntnu.pvv.roowho2.finger")]
pub trait VarlinkFingerClientProxy {
async fn finger(
user_queries: Option<Vec<String>>,
match_fullnames: bool,
request_info: FingerRequestInfo,
request_networking: FingerRequestNetworking,
disable_user_account_db: bool,
raw_remote_output: bool,
) -> zlink::Result<Result<VarlinkFingerResponse, VarlinkFingerClientError>>;
pub enum VarlinkFingerClientRequest {
#[serde(rename = "no.ntnu.pvv.roowho2.finger.Finger")]
Finger {
#[derive(Debug, Serialize)]
pub enum VarlinkFingerClientResponse {
Finger(VarlinkFingerResponse),
pub type VarlinkFingerResponse = Vec<FingerResponseUserEntry>;
#[zlink(interface = "no.ntnu.pvv.roowho2.finger")]
pub enum VarlinkFingerClientError {
// --------------------
#[allow(unused)]
pub enum VarlinkMethod {
Rwhod(VarlinkRwhodClientRequest),
Finger(VarlinkFingerClientRequest),
pub enum VarlinkReply {
Rwhod(VarlinkRwhodClientResponse),
Finger(VarlinkFingerClientResponse),
pub enum VarlinkReplyError {
Rwhod(VarlinkRwhodClientError),
Finger(VarlinkFingerClientError),
#[derive(Debug, Clone)]
pub struct VarlinkRoowhoo2ClientServer {
whod_status_store: RwhodStatusStore,
rwhod_enabled: bool,
fingerd_enabled: bool,
finger_ignore_list: Option<IgnoreList>,
impl VarlinkRoowhoo2ClientServer {
pub fn new(
) -> Self {
Self {
whod_status_store,
rwhod_enabled,
fingerd_enabled,
finger_ignore_list,
async fn handle_rwho_request(&self, all: bool) -> VarlinkRwhoResponse {
tracing::debug!(all, "Handling Rwho request");
let store = self.whod_status_store.read().await;
let mut all_user_entries = Vec::with_capacity(store.len());
for status_update in store.values() {
all_user_entries.extend(
status_update
.users
.iter()
.filter(|user| all || user.idle_time < chrono::Duration::hours(1))
.cloned()
.map(|user| (status_update.hostname.clone(), user)),
);
all_user_entries
async fn handle_ruptime_request(&self, all: bool) -> VarlinkRuptimeResponse {
tracing::debug!(all, "Handling Ruptime request");
store
.values()
.map(|mut status_update| {
if !all {
.retain(|user| user.idle_time < chrono::Duration::hours(1));
})
.collect()
async fn handle_finger_request(
&self,
_request_networking: FingerRequestNetworking,
_disable_user_account_db: bool,
_raw_remote_output: bool,
) -> VarlinkFingerResponse {
tracing::debug!(
user_queries = ?user_queries,
match_fullnames = match_fullnames,
request_info = ?request_info,
"Handling Finger request",
match user_queries {
Some(usernames) => usernames
.into_iter()
.flat_map::<Vec<_>, _>(|username| {
fingerd::search_for_user(
&username,
match_fullnames,
&request_info,
self.finger_ignore_list.as_ref(),
)
.map(|res| (username.clone(), res))
.dedup_by(|a, b| match (&a.1, &b.1) {
(Ok(user_a), Ok(user_b)) => user_a.username == user_b.username,
_ => false,
.filter_map(|(username, user)| match user {
Ok(user_info) => Some(user_info),
Err(err) => {
tracing::error!(
"Error retrieving local user information for '{}': {}",
username,
err
None
.map(Box::new)
.map(FingerResponseUserEntry::Structured)
.collect(),
None => finger_utmp_users(&request_info, self.finger_ignore_list.as_ref())
.filter_map(|res| match res {
tracing::error!("Error retrieving local user information: {}", err);
impl zlink::Service<zlink::unix::Stream> for VarlinkRoowhoo2ClientServer {
type MethodCall<'de> = VarlinkMethod;
type ReplyParams<'se> = VarlinkReply;
type ReplyStreamParams = ();
type ReplyStream = futures_util::stream::Empty<(zlink::Reply<()>, Vec<OwnedFd>)>;
type ReplyError<'se> = VarlinkReplyError;
async fn handle<'service>(
&'service mut self,
call: &'service zlink::Call<Self::MethodCall<'_>>,
_conn: &mut zlink::Connection<zlink::unix::Stream>,
_fds: Vec<std::os::fd::OwnedFd>,
) -> zlink::service::HandleResult<
Self::ReplyParams<'service>,
Self::ReplyStream,
Self::ReplyError<'service>,
> {
match call.method() {
VarlinkMethod::Rwhod(VarlinkRwhodClientRequest::Rwho { all }) => {
if !self.rwhod_enabled {
return (
MethodReply::Error(VarlinkReplyError::Rwhod(
VarlinkRwhodClientError::Disabled,
)),
Default::default(),
let result =
match timeout(Duration::from_secs(2), self.handle_rwho_request(*all)).await {
Ok(response) => response,
Err(_) => {
tracing::error!("Rwho request timed out after 2 seconds");
VarlinkRwhodClientError::TimedOut,
(
MethodReply::Single(Some(VarlinkReply::Rwhod(
VarlinkRwhodClientResponse::Rwho(result),
))),
VarlinkMethod::Rwhod(VarlinkRwhodClientRequest::Ruptime { all }) => {
let result = match timeout(
Duration::from_secs(2),
self.handle_ruptime_request(*all),
.await
{
tracing::error!("Ruptime request timed out after 2 seconds");
VarlinkRwhodClientResponse::Ruptime(result),
VarlinkMethod::Finger(VarlinkFingerClientRequest::Finger {
user_queries,
request_info,
request_networking,
disable_user_account_db,
raw_remote_output,
}) => {
if !self.fingerd_enabled {
MethodReply::Error(VarlinkReplyError::Finger(
VarlinkFingerClientError::Disabled,
self.handle_finger_request(
user_queries.clone(),
*match_fullnames,
request_info.clone(),
request_networking.clone(),
*disable_user_account_db,
*raw_remote_output,
),
tracing::error!("Finger request timed out after 2 seconds");
VarlinkFingerClientError::TimedOut,
MethodReply::Single(Some(VarlinkReply::Finger(
VarlinkFingerClientResponse::Finger(result),
pub async fn varlink_client_server_task(
socket: zlink::unix::Listener,
) -> anyhow::Result<()> {
let service = VarlinkRoowhoo2ClientServer::new(
let server = zlink::Server::new(socket, service);
tracing::info!("Starting Rwhod client API server");
server
.run()
.context("Rwhod client API server failed")?;
Ok(())