lda/wakey
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 18 Wiki Activity

more megafile breakdown

Browse Source
This commit is contained in:
lda
2026-04-28 17:18:59 +07:00 Verified
parent 23656cd6d7
commit 10baa53e4c
9 changed files with 1267 additions and 1196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ui/src/pages/ObservationsPage.tsx
+1 -1
View File
@@ -47,7 +47,7 @@ export function ObservationsPage() {
[], [],
); );
const [devices, setDevices] = useState<KnownDevice[]>([]); const [devices, setDevices] = useState<KnownDevice[]>([]);
const [filter, setFilter] = useState<Filter>("unknown"); const [filter, setFilter] = useState<Filter>("all");
const [query, setQuery] = useState(""); const [query, setQuery] = useState("");
const [selectedDevices, setSelectedDevices] = useState< const [selectedDevices, setSelectedDevices] = useState<
Record<string, string> Record<string, string>
wakey-control-plane/src/api/control.rs
+14 -815
View File
@@ -1,819 +1,18 @@
use axum::Json; mod devices;
use axum::extract::{Path as AxumPath, Query, State}; mod enroll;
use axum::http::StatusCode; mod observations;
use axum::response::IntoResponse; mod stats;
use serde::{Deserialize, Serialize};
use tracing::{info, warn};
use crate::api::json_error; pub use devices::{
use crate::runtime::{AppState, SessionEvent}; attach_device_identifier, attach_observation_identifier, create_known_device,
use crate::state::{ forget_known_device, list_known_devices,
AgentDeviceObservationEvent, AgentDeviceObservationInput, AgentDeviceObservationView,
AuditEventInput, DeviceIdentifierInput, KnownDeviceInput,
}; };
pub use enroll::{
#[derive(Debug, Deserialize)] EnrollTokenStatus, IssueEnrollTokenResponse, RevokeAgentResponse, RevokeEnrollTokenResponse,
pub struct EnrollRequest { enroll, healthz, issue_enroll_token, list_enroll_tokens, revoke_agent, revoke_enroll_token,
pub enroll_token: String, set_agent_nickname,
}
#[derive(Debug, Serialize)]
pub struct EnrollResponse {
pub agent_id: String,
pub agent_token: String,
pub server_url: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct IssueEnrollTokenResponse {
pub enroll_token: String,
pub expires_at_unix: u64,
}
#[derive(Debug, Deserialize)]
pub struct IssueEnrollTokenQuery {
pub ttl_seconds: Option<u64>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct EnrollTokenStatus {
pub enroll_token: String,
pub expires_at_unix: u64,
pub expired: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct RevokeEnrollTokenResponse {
pub token: String,
pub revoked: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct RevokeAgentResponse {
pub agent_id: String,
pub revoked: bool,
}
#[derive(Debug, Deserialize)]
pub struct SetAgentNicknameRequest {
pub nickname: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct SetAgentNicknameResponse {
pub agent_id: String,
pub nickname: Option<String>,
pub updated: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct StateStatsResponse {
pub db_path: String,
pub schema_version: u32,
pub agent_count: usize,
pub enroll_token_count: usize,
pub expired_enroll_token_count: usize,
}
#[derive(Debug, Deserialize)]
pub struct CreateKnownDeviceRequest {
pub display_name: String,
#[serde(default)]
pub pinned: bool,
pub notes: Option<String>,
#[serde(default)]
pub identifiers: Vec<DeviceIdentifierRequest>,
}
#[derive(Debug, Deserialize)]
pub struct DeviceIdentifierRequest {
pub kind: String,
pub value: String,
}
#[derive(Debug, Deserialize)]
pub struct AttachObservationIdentifierRequest {
pub observation_key: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct KnownDeviceResponse {
pub device_id: String,
pub display_name: String,
pub pinned: bool,
pub created_at_unix: u64,
pub updated_at_unix: u64,
pub notes: Option<String>,
pub identifiers: Vec<DeviceIdentifierResponse>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct DeviceIdentifierResponse {
pub identifier_key: String,
pub device_id: String,
pub kind: String,
pub value: String,
pub created_at_unix: u64,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ForgetKnownDeviceResponse {
pub device_id: String,
pub forgotten: bool,
}
#[derive(Debug, Deserialize)]
pub struct UploadAgentObservationsRequest {
pub agent_id: String,
pub agent_token: String,
#[serde(default)]
pub observations: Vec<AgentObservationRequest>,
}
#[derive(Debug, Deserialize)]
pub struct AgentObservationRequest {
pub kind: String,
pub action: String,
pub mac: Option<String>,
pub ip: Option<String>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct UploadAgentObservationsResponse {
pub accepted: usize,
}
#[derive(Debug, Deserialize)]
pub struct ListObservationsQuery {
pub agent_id: Option<String>,
pub limit: Option<usize>,
}
#[derive(Debug, Deserialize)]
pub struct ListObservationHistoryQuery {
pub agent_id: Option<String>,
pub kind: Option<String>,
pub mac: Option<String>,
pub ip: Option<String>,
pub observation_key: Option<String>,
pub limit: Option<usize>,
}
pub async fn healthz() -> &'static str {
"ok"
}
pub async fn enroll(
State(state): State<AppState>,
Json(req): Json<EnrollRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.enroll(&req.enroll_token).await {
Ok(issued) => {
info!(agent_id = %issued.agent_id, "agent enrollment accepted");
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "agent".into(),
actor_id: Some(issued.agent_id.clone()),
agent_id: Some(issued.agent_id.clone()),
request_id: None,
event_type: "agent_enroll".into(),
outcome: "ok".into(),
latency_ms: None,
message: "agent enrollment accepted".into(),
metadata: serde_json::json!({}),
})
.await
{
warn!(error = %err, "failed to append audit event for enroll success");
}
Ok((
StatusCode::OK,
Json(EnrollResponse {
agent_id: issued.agent_id,
agent_token: issued.agent_token,
server_url: state.public_url,
}),
))
}
Err(err) => {
warn!(error = %err, "agent enrollment rejected");
if let Err(audit_err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "agent".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "agent_enroll".into(),
outcome: "rejected".into(),
latency_ms: None,
message: err.to_string(),
metadata: serde_json::json!({}),
})
.await
{
warn!(error = %audit_err, "failed to append audit event for enroll rejection");
}
Err(json_error(
StatusCode::UNAUTHORIZED,
"enrollment_rejected",
&err.to_string(),
))
}
}
}
pub async fn issue_enroll_token(
State(state): State<AppState>,
Query(query): Query<IssueEnrollTokenQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let ttl = std::time::Duration::from_secs(
query
.ttl_seconds
.unwrap_or(state.enroll_token_ttl.as_secs())
.max(1),
);
match state.store.issue_enroll_token(ttl).await {
Ok(issued) => {
info!(
expires_at_unix = issued.expires_at_unix,
"issued enroll token"
);
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_issue".into(),
outcome: "ok".into(),
latency_ms: None,
message: "issued enroll token".into(),
metadata: serde_json::json!({
"ttl_seconds": ttl.as_secs(),
"expires_at_unix": issued.expires_at_unix,
}),
})
.await
{
warn!(error = %err, "failed to append audit event for token issuance");
}
Ok((
StatusCode::OK,
Json(IssueEnrollTokenResponse {
enroll_token: issued.enroll_token,
expires_at_unix: issued.expires_at_unix,
}),
))
}
Err(err) => {
warn!(error = %err, "failed to issue enroll token");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"issue_enroll_token_failed",
&err.to_string(),
))
}
}
}
pub async fn list_enroll_tokens(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.list_enroll_tokens().await {
Ok(tokens) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_list".into(),
outcome: "ok".into(),
latency_ms: None,
message: "listed enroll tokens".into(),
metadata: serde_json::json!({
"count": tokens.len(),
}),
})
.await
{
warn!(error = %err, "failed to append audit event for token listing");
}
let body = tokens
.into_iter()
.map(|t| EnrollTokenStatus {
enroll_token: t.enroll_token,
expires_at_unix: t.expires_at_unix,
expired: t.expired,
})
.collect::<Vec<_>>();
Ok((StatusCode::OK, Json(body)))
}
Err(err) => {
warn!(error = %err, "failed to list enroll tokens");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_enroll_tokens_failed",
&err.to_string(),
))
}
}
}
pub async fn revoke_enroll_token(
State(state): State<AppState>,
AxumPath(token): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.revoke_enroll_token(&token).await {
Ok(revoked) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_revoke".into(),
outcome: if revoked {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if revoked {
"revoked enroll token".into()
} else {
"enroll token not found".into()
},
metadata: serde_json::json!({ "token": token }),
})
.await
{
warn!(error = %err, "failed to append audit event for token revoke");
}
Ok((
StatusCode::OK,
Json(RevokeEnrollTokenResponse { token, revoked }),
))
}
Err(err) => {
warn!(error = %err, "failed to revoke enroll token");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"revoke_enroll_token_failed",
&err.to_string(),
))
}
}
}
pub async fn revoke_agent(
State(state): State<AppState>,
AxumPath(agent_id): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.revoke_agent(&agent_id).await {
Ok(revoked) => {
if revoked {
// Request a graceful websocket close, then remove session from active map.
if let Some(session) = state.sessions.write().await.remove(&agent_id) {
let _ = session.tx.send(SessionEvent::Close);
}
}
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: Some(agent_id.clone()),
request_id: None,
event_type: "agent_revoke".into(),
outcome: if revoked {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if revoked {
"revoked agent credentials".into()
} else {
"agent credentials not found".into()
},
metadata: serde_json::json!({ "agent_id": agent_id }),
})
.await
{
warn!(error = %err, "failed to append audit event for agent revoke");
}
Ok((
StatusCode::OK,
Json(RevokeAgentResponse { agent_id, revoked }),
))
}
Err(err) => {
warn!(error = %err, "failed to revoke agent credentials");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"revoke_agent_failed",
&err.to_string(),
))
}
}
}
pub async fn set_agent_nickname(
State(state): State<AppState>,
AxumPath(agent_id): AxumPath<String>,
Json(req): Json<SetAgentNicknameRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let normalized = req
.nickname
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
.map(ToOwned::to_owned);
match state
.store
.set_agent_nickname(&agent_id, normalized.as_deref())
.await
{
Ok(updated) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: Some(agent_id.clone()),
request_id: None,
event_type: "agent_nickname_set".into(),
outcome: if updated {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if updated {
"updated agent nickname".into()
} else {
"agent not found for nickname update".into()
},
metadata: serde_json::json!({
"agent_id": agent_id,
"nickname": normalized,
}),
})
.await
{
warn!(error = %err, "failed to append audit event for nickname set");
}
Ok((
StatusCode::OK,
Json(SetAgentNicknameResponse {
agent_id,
nickname: normalized,
updated,
}),
))
}
Err(err) => {
warn!(error = %err, "failed to update agent nickname");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"set_agent_nickname_failed",
&err.to_string(),
))
}
}
}
pub async fn create_known_device(
State(state): State<AppState>,
Json(req): Json<CreateKnownDeviceRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let input = KnownDeviceInput {
display_name: req.display_name,
pinned: req.pinned,
notes: req.notes,
identifiers: req
.identifiers
.into_iter()
.map(|identifier| DeviceIdentifierInput {
kind: identifier.kind,
value: identifier.value,
})
.collect(),
}; };
pub use observations::{
match state.store.create_known_device(input).await { list_agent_observation_history, list_agent_observations, upload_agent_observations,
Ok(device) => Ok((StatusCode::CREATED, Json(known_device_response(device)))),
Err(err) => {
warn!(error = %err, "failed to create known device");
Err(json_error(
StatusCode::BAD_REQUEST,
"create_known_device_failed",
&err.to_string(),
))
}
}
}
pub async fn list_known_devices(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.list_known_devices().await {
Ok(devices) => Ok((
StatusCode::OK,
Json(
devices
.into_iter()
.map(known_device_response)
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list known devices");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_known_devices_failed",
&err.to_string(),
))
}
}
}
pub async fn forget_known_device(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.forget_known_device(&device_id).await {
Ok(forgotten) => Ok((
StatusCode::OK,
Json(ForgetKnownDeviceResponse {
device_id,
forgotten,
}),
)),
Err(err) => {
warn!(error = %err, "failed to forget known device");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"forget_known_device_failed",
&err.to_string(),
))
}
}
}
pub async fn attach_device_identifier(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
Json(req): Json<DeviceIdentifierRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let input = DeviceIdentifierInput {
kind: req.kind,
value: req.value,
}; };
pub use stats::{StateStatsResponse, state_stats};
match state
.store
.attach_device_identifier(&device_id, input)
.await
{
Ok(Some(device)) => Ok((StatusCode::OK, Json(known_device_response(device)))),
Ok(None) => Err(json_error(
StatusCode::NOT_FOUND,
"known_device_not_found",
"known device not found",
)),
Err(err) => {
warn!(error = %err, "failed to attach device identifier");
Err(json_error(
StatusCode::BAD_REQUEST,
"attach_device_identifier_failed",
&err.to_string(),
))
}
}
}
pub async fn attach_observation_identifier(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
Json(req): Json<AttachObservationIdentifierRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state
.store
.attach_observation_identifier(&device_id, &req.observation_key)
.await
{
Ok(Some(device)) => Ok((StatusCode::OK, Json(known_device_response(device)))),
Ok(None) => Err(json_error(
StatusCode::NOT_FOUND,
"known_device_not_found",
"known device not found",
)),
Err(err) => {
warn!(error = %err, "failed to attach observation identifier");
Err(json_error(
StatusCode::BAD_REQUEST,
"attach_observation_identifier_failed",
&err.to_string(),
))
}
}
}
pub async fn upload_agent_observations(
State(state): State<AppState>,
Json(req): Json<UploadAgentObservationsRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
if !state
.store
.verify_agent_token(&req.agent_id, &req.agent_token)
.await
{
return Err(json_error(
StatusCode::UNAUTHORIZED,
"agent_auth_rejected",
"agent credentials rejected",
));
}
let observations = req
.observations
.into_iter()
.map(|observation| AgentDeviceObservationInput {
kind: observation.kind,
action: observation.action,
mac: observation.mac,
ip: observation.ip,
hostname: observation.hostname,
first_seen_unix: observation.first_seen_unix,
last_seen_unix: observation.last_seen_unix,
})
.collect();
match state
.store
.upsert_agent_observations(&req.agent_id, observations)
.await
{
Ok(accepted) => Ok((
StatusCode::OK,
Json(UploadAgentObservationsResponse { accepted }),
)),
Err(err) => {
warn!(error = %err, agent_id = %req.agent_id, "failed to upload agent observations");
Err(json_error(
StatusCode::BAD_REQUEST,
"upload_observations_failed",
&err.to_string(),
))
}
}
}
pub async fn list_agent_observations(
State(state): State<AppState>,
Query(query): Query<ListObservationsQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state
.store
.list_agent_observation_views(query.agent_id.as_deref(), query.limit.unwrap_or(500))
.await
{
Ok(observations) => Ok((
StatusCode::OK,
Json(
observations
.into_iter()
.map(agent_observation_response) // no-op premium
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list agent observations");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_observations_failed",
&err.to_string(),
))
}
}
}
pub async fn list_agent_observation_history(
State(state): State<AppState>,
Query(query): Query<ListObservationHistoryQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let kind = normalized_query_value(query.kind).map(|value| value.to_ascii_lowercase());
let mac = normalized_query_value(query.mac).map(|value| value.to_ascii_lowercase());
let ip = normalized_query_value(query.ip);
let observation_key = normalized_query_value(query.observation_key);
match state
.store
.list_agent_observation_events(
query.agent_id.as_deref(),
kind.as_deref(),
mac.as_deref(),
ip.as_deref(),
observation_key.as_deref(),
query.limit.unwrap_or(500),
)
.await
{
Ok(events) => Ok((
StatusCode::OK,
Json(
events
.into_iter()
.map(agent_observation_event_response) // no-op premium
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list agent observation history");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_observation_history_failed",
&err.to_string(),
))
}
}
}
pub async fn state_stats(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.stats().await {
Ok(stats) => Ok((
StatusCode::OK,
Json(StateStatsResponse {
db_path: stats.db_path.display().to_string(),
schema_version: stats.schema_version,
agent_count: stats.agent_count,
enroll_token_count: stats.enroll_token_count,
expired_enroll_token_count: stats.expired_enroll_token_count,
}),
)),
Err(err) => {
warn!(error = %err, "failed to read state stats");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"state_stats_failed",
&err.to_string(),
))
}
}
}
fn agent_observation_response(
observation: AgentDeviceObservationView,
) -> AgentDeviceObservationView {
observation
}
fn agent_observation_event_response(
event: AgentDeviceObservationEvent,
) -> AgentDeviceObservationEvent {
event
}
fn normalized_query_value(value: Option<String>) -> Option<String> {
value
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn known_device_response(device: crate::state::KnownDevice) -> KnownDeviceResponse {
KnownDeviceResponse {
device_id: device.device_id,
display_name: device.display_name,
pinned: device.pinned,
created_at_unix: device.created_at_unix,
updated_at_unix: device.updated_at_unix,
notes: device.notes,
identifiers: device
.identifiers
.into_iter()
.map(|identifier| DeviceIdentifierResponse {
identifier_key: identifier.identifier_key,
device_id: identifier.device_id,
kind: identifier.kind,
value: identifier.value,
created_at_unix: identifier.created_at_unix,
})
.collect(),
}
}
wakey-control-plane/src/api/control/devices.rs
+216
View File
@@ -0,0 +1,216 @@
use axum::Json;
use axum::extract::{Path as AxumPath, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use tracing::warn;
use crate::api::json_error;
use crate::runtime::AppState;
use crate::state::{DeviceIdentifierInput, KnownDeviceInput};
#[derive(Debug, Deserialize)]
pub struct CreateKnownDeviceRequest {
pub display_name: String,
#[serde(default)]
pub pinned: bool,
pub notes: Option<String>,
#[serde(default)]
pub identifiers: Vec<DeviceIdentifierRequest>,
}
#[derive(Debug, Deserialize)]
pub struct DeviceIdentifierRequest {
pub kind: String,
pub value: String,
}
#[derive(Debug, Deserialize)]
pub struct AttachObservationIdentifierRequest {
pub observation_key: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct KnownDeviceResponse {
pub device_id: String,
pub display_name: String,
pub pinned: bool,
pub created_at_unix: u64,
pub updated_at_unix: u64,
pub notes: Option<String>,
pub identifiers: Vec<DeviceIdentifierResponse>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct DeviceIdentifierResponse {
pub identifier_key: String,
pub device_id: String,
pub kind: String,
pub value: String,
pub created_at_unix: u64,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ForgetKnownDeviceResponse {
pub device_id: String,
pub forgotten: bool,
}
pub async fn create_known_device(
State(state): State<AppState>,
Json(req): Json<CreateKnownDeviceRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let input = KnownDeviceInput {
display_name: req.display_name,
pinned: req.pinned,
notes: req.notes,
identifiers: req
.identifiers
.into_iter()
.map(|identifier| DeviceIdentifierInput {
kind: identifier.kind,
value: identifier.value,
})
.collect(),
};
match state.store.create_known_device(input).await {
Ok(device) => Ok((StatusCode::CREATED, Json(known_device_response(device)))),
Err(err) => {
warn!(error = %err, "failed to create known device");
Err(json_error(
StatusCode::BAD_REQUEST,
"create_known_device_failed",
&err.to_string(),
))
}
}
}
pub async fn list_known_devices(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.list_known_devices().await {
Ok(devices) => Ok((
StatusCode::OK,
Json(
devices
.into_iter()
.map(known_device_response)
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list known devices");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_known_devices_failed",
&err.to_string(),
))
}
}
}
pub async fn forget_known_device(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.forget_known_device(&device_id).await {
Ok(forgotten) => Ok((
StatusCode::OK,
Json(ForgetKnownDeviceResponse {
device_id,
forgotten,
}),
)),
Err(err) => {
warn!(error = %err, "failed to forget known device");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"forget_known_device_failed",
&err.to_string(),
))
}
}
}
pub async fn attach_device_identifier(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
Json(req): Json<DeviceIdentifierRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let input = DeviceIdentifierInput {
kind: req.kind,
value: req.value,
};
match state
.store
.attach_device_identifier(&device_id, input)
.await
{
Ok(Some(device)) => Ok((StatusCode::OK, Json(known_device_response(device)))),
Ok(None) => Err(json_error(
StatusCode::NOT_FOUND,
"known_device_not_found",
"known device not found",
)),
Err(err) => {
warn!(error = %err, "failed to attach device identifier");
Err(json_error(
StatusCode::BAD_REQUEST,
"attach_device_identifier_failed",
&err.to_string(),
))
}
}
}
pub async fn attach_observation_identifier(
State(state): State<AppState>,
AxumPath(device_id): AxumPath<String>,
Json(req): Json<AttachObservationIdentifierRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state
.store
.attach_observation_identifier(&device_id, &req.observation_key)
.await
{
Ok(Some(device)) => Ok((StatusCode::OK, Json(known_device_response(device)))),
Ok(None) => Err(json_error(
StatusCode::NOT_FOUND,
"known_device_not_found",
"known device not found",
)),
Err(err) => {
warn!(error = %err, "failed to attach observation identifier");
Err(json_error(
StatusCode::BAD_REQUEST,
"attach_observation_identifier_failed",
&err.to_string(),
))
}
}
}
fn known_device_response(device: crate::state::KnownDevice) -> KnownDeviceResponse {
KnownDeviceResponse {
device_id: device.device_id,
display_name: device.display_name,
pinned: device.pinned,
created_at_unix: device.created_at_unix,
updated_at_unix: device.updated_at_unix,
notes: device.notes,
identifiers: device
.identifiers
.into_iter()
.map(|identifier| DeviceIdentifierResponse {
identifier_key: identifier.identifier_key,
device_id: identifier.device_id,
kind: identifier.kind,
value: identifier.value,
created_at_unix: identifier.created_at_unix,
})
.collect(),
}
}
wakey-control-plane/src/api/control/enroll.rs
+399
View File
@@ -0,0 +1,399 @@
use axum::Json;
use axum::extract::{Path as AxumPath, Query, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use tracing::{info, warn};
use crate::api::json_error;
use crate::runtime::{AppState, SessionEvent};
use crate::state::AuditEventInput;
#[derive(Debug, Deserialize)]
pub struct EnrollRequest {
pub enroll_token: String,
}
#[derive(Debug, Serialize)]
pub struct EnrollResponse {
pub agent_id: String,
pub agent_token: String,
pub server_url: String,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct IssueEnrollTokenResponse {
pub enroll_token: String,
pub expires_at_unix: u64,
}
#[derive(Debug, Deserialize)]
pub struct IssueEnrollTokenQuery {
pub ttl_seconds: Option<u64>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct EnrollTokenStatus {
pub enroll_token: String,
pub expires_at_unix: u64,
pub expired: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct RevokeEnrollTokenResponse {
pub token: String,
pub revoked: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct RevokeAgentResponse {
pub agent_id: String,
pub revoked: bool,
}
#[derive(Debug, Deserialize)]
pub struct SetAgentNicknameRequest {
pub nickname: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct SetAgentNicknameResponse {
pub agent_id: String,
pub nickname: Option<String>,
pub updated: bool,
}
pub async fn healthz() -> &'static str {
"ok"
}
pub async fn enroll(
State(state): State<AppState>,
Json(req): Json<EnrollRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.enroll(&req.enroll_token).await {
Ok(issued) => {
info!(agent_id = %issued.agent_id, "agent enrollment accepted");
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "agent".into(),
actor_id: Some(issued.agent_id.clone()),
agent_id: Some(issued.agent_id.clone()),
request_id: None,
event_type: "agent_enroll".into(),
outcome: "ok".into(),
latency_ms: None,
message: "agent enrollment accepted".into(),
metadata: serde_json::json!({}),
})
.await
{
warn!(error = %err, "failed to append audit event for enroll success");
}
Ok((
StatusCode::OK,
Json(EnrollResponse {
agent_id: issued.agent_id,
agent_token: issued.agent_token,
server_url: state.public_url,
}),
))
}
Err(err) => {
warn!(error = %err, "agent enrollment rejected");
if let Err(audit_err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "agent".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "agent_enroll".into(),
outcome: "rejected".into(),
latency_ms: None,
message: err.to_string(),
metadata: serde_json::json!({}),
})
.await
{
warn!(error = %audit_err, "failed to append audit event for enroll rejection");
}
Err(json_error(
StatusCode::UNAUTHORIZED,
"enrollment_rejected",
&err.to_string(),
))
}
}
}
pub async fn issue_enroll_token(
State(state): State<AppState>,
Query(query): Query<IssueEnrollTokenQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let ttl = std::time::Duration::from_secs(
query
.ttl_seconds
.unwrap_or(state.enroll_token_ttl.as_secs())
.max(1),
);
match state.store.issue_enroll_token(ttl).await {
Ok(issued) => {
info!(
expires_at_unix = issued.expires_at_unix,
"issued enroll token"
);
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_issue".into(),
outcome: "ok".into(),
latency_ms: None,
message: "issued enroll token".into(),
metadata: serde_json::json!({
"ttl_seconds": ttl.as_secs(),
"expires_at_unix": issued.expires_at_unix,
}),
})
.await
{
warn!(error = %err, "failed to append audit event for token issuance");
}
Ok((
StatusCode::OK,
Json(IssueEnrollTokenResponse {
enroll_token: issued.enroll_token,
expires_at_unix: issued.expires_at_unix,
}),
))
}
Err(err) => {
warn!(error = %err, "failed to issue enroll token");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"issue_enroll_token_failed",
&err.to_string(),
))
}
}
}
pub async fn list_enroll_tokens(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.list_enroll_tokens().await {
Ok(tokens) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_list".into(),
outcome: "ok".into(),
latency_ms: None,
message: "listed enroll tokens".into(),
metadata: serde_json::json!({
"count": tokens.len(),
}),
})
.await
{
warn!(error = %err, "failed to append audit event for token listing");
}
let body = tokens
.into_iter()
.map(|t| EnrollTokenStatus {
enroll_token: t.enroll_token,
expires_at_unix: t.expires_at_unix,
expired: t.expired,
})
.collect::<Vec<_>>();
Ok((StatusCode::OK, Json(body)))
}
Err(err) => {
warn!(error = %err, "failed to list enroll tokens");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_enroll_tokens_failed",
&err.to_string(),
))
}
}
}
pub async fn revoke_enroll_token(
State(state): State<AppState>,
AxumPath(token): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.revoke_enroll_token(&token).await {
Ok(revoked) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: None,
request_id: None,
event_type: "enroll_token_revoke".into(),
outcome: if revoked {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if revoked {
"revoked enroll token".into()
} else {
"enroll token not found".into()
},
metadata: serde_json::json!({ "token": token }),
})
.await
{
warn!(error = %err, "failed to append audit event for token revoke");
}
Ok((
StatusCode::OK,
Json(RevokeEnrollTokenResponse { token, revoked }),
))
}
Err(err) => {
warn!(error = %err, "failed to revoke enroll token");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"revoke_enroll_token_failed",
&err.to_string(),
))
}
}
}
pub async fn revoke_agent(
State(state): State<AppState>,
AxumPath(agent_id): AxumPath<String>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.revoke_agent(&agent_id).await {
Ok(revoked) => {
if revoked {
// Request a graceful websocket close, then remove session from active map.
if let Some(session) = state.sessions.write().await.remove(&agent_id) {
let _ = session.tx.send(SessionEvent::Close);
}
}
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: Some(agent_id.clone()),
request_id: None,
event_type: "agent_revoke".into(),
outcome: if revoked {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if revoked {
"revoked agent credentials".into()
} else {
"agent credentials not found".into()
},
metadata: serde_json::json!({ "agent_id": agent_id }),
})
.await
{
warn!(error = %err, "failed to append audit event for agent revoke");
}
Ok((
StatusCode::OK,
Json(RevokeAgentResponse { agent_id, revoked }),
))
}
Err(err) => {
warn!(error = %err, "failed to revoke agent credentials");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"revoke_agent_failed",
&err.to_string(),
))
}
}
}
pub async fn set_agent_nickname(
State(state): State<AppState>,
AxumPath(agent_id): AxumPath<String>,
Json(req): Json<SetAgentNicknameRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let normalized = req
.nickname
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
.map(ToOwned::to_owned);
match state
.store
.set_agent_nickname(&agent_id, normalized.as_deref())
.await
{
Ok(updated) => {
if let Err(err) = state
.store
.append_audit_event(AuditEventInput {
actor_type: "admin_api".into(),
actor_id: None,
agent_id: Some(agent_id.clone()),
request_id: None,
event_type: "agent_nickname_set".into(),
outcome: if updated {
"ok".into()
} else {
"not_found".into()
},
latency_ms: None,
message: if updated {
"updated agent nickname".into()
} else {
"agent not found for nickname update".into()
},
metadata: serde_json::json!({
"agent_id": agent_id,
"nickname": normalized,
}),
})
.await
{
warn!(error = %err, "failed to append audit event for nickname set");
}
Ok((
StatusCode::OK,
Json(SetAgentNicknameResponse {
agent_id,
nickname: normalized,
updated,
}),
))
}
Err(err) => {
warn!(error = %err, "failed to update agent nickname");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"set_agent_nickname_failed",
&err.to_string(),
))
}
}
}
wakey-control-plane/src/api/control/observations.rs
+189
View File
@@ -0,0 +1,189 @@
use axum::Json;
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::Deserialize;
use tracing::warn;
use crate::api::json_error;
use crate::runtime::AppState;
use crate::state::{
AgentDeviceObservationEvent, AgentDeviceObservationInput, AgentDeviceObservationView,
};
#[derive(Debug, Deserialize)]
pub struct UploadAgentObservationsRequest {
pub agent_id: String,
pub agent_token: String,
#[serde(default)]
pub observations: Vec<AgentObservationRequest>,
}
#[derive(Debug, Deserialize)]
pub struct AgentObservationRequest {
pub kind: String,
pub action: String,
pub mac: Option<String>,
pub ip: Option<String>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
}
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct UploadAgentObservationsResponse {
pub accepted: usize,
}
#[derive(Debug, Deserialize)]
pub struct ListObservationsQuery {
pub agent_id: Option<String>,
pub limit: Option<usize>,
}
#[derive(Debug, Deserialize)]
pub struct ListObservationHistoryQuery {
pub agent_id: Option<String>,
pub kind: Option<String>,
pub mac: Option<String>,
pub ip: Option<String>,
pub observation_key: Option<String>,
pub limit: Option<usize>,
}
pub async fn upload_agent_observations(
State(state): State<AppState>,
Json(req): Json<UploadAgentObservationsRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
if !state
.store
.verify_agent_token(&req.agent_id, &req.agent_token)
.await
{
return Err(json_error(
StatusCode::UNAUTHORIZED,
"agent_auth_rejected",
"agent credentials rejected",
));
}
let observations = req
.observations
.into_iter()
.map(|observation| AgentDeviceObservationInput {
kind: observation.kind,
action: observation.action,
mac: observation.mac,
ip: observation.ip,
hostname: observation.hostname,
first_seen_unix: observation.first_seen_unix,
last_seen_unix: observation.last_seen_unix,
})
.collect();
match state
.store
.upsert_agent_observations(&req.agent_id, observations)
.await
{
Ok(accepted) => Ok((
StatusCode::OK,
Json(UploadAgentObservationsResponse { accepted }),
)),
Err(err) => {
warn!(error = %err, agent_id = %req.agent_id, "failed to upload agent observations");
Err(json_error(
StatusCode::BAD_REQUEST,
"upload_observations_failed",
&err.to_string(),
))
}
}
}
pub async fn list_agent_observations(
State(state): State<AppState>,
Query(query): Query<ListObservationsQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state
.store
.list_agent_observation_views(query.agent_id.as_deref(), query.limit.unwrap_or(500))
.await
{
Ok(observations) => Ok((
StatusCode::OK,
Json(
observations
.into_iter()
.map(agent_observation_response)
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list agent observations");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_observations_failed",
&err.to_string(),
))
}
}
}
pub async fn list_agent_observation_history(
State(state): State<AppState>,
Query(query): Query<ListObservationHistoryQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let kind = normalized_query_value(query.kind).map(|value| value.to_ascii_lowercase());
let mac = normalized_query_value(query.mac).map(|value| value.to_ascii_lowercase());
let ip = normalized_query_value(query.ip);
let observation_key = normalized_query_value(query.observation_key);
match state
.store
.list_agent_observation_events(
query.agent_id.as_deref(),
kind.as_deref(),
mac.as_deref(),
ip.as_deref(),
observation_key.as_deref(),
query.limit.unwrap_or(500),
)
.await
{
Ok(events) => Ok((
StatusCode::OK,
Json(
events
.into_iter()
.map(agent_observation_event_response)
.collect::<Vec<_>>(),
),
)),
Err(err) => {
warn!(error = %err, "failed to list agent observation history");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"list_observation_history_failed",
&err.to_string(),
))
}
}
}
fn agent_observation_response(
observation: AgentDeviceObservationView,
) -> AgentDeviceObservationView {
observation
}
fn agent_observation_event_response(
event: AgentDeviceObservationEvent,
) -> AgentDeviceObservationEvent {
event
}
fn normalized_query_value(value: Option<String>) -> Option<String> {
value
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
wakey-control-plane/src/api/control/stats.rs
+43
View File
@@ -0,0 +1,43 @@
use axum::Json;
use axum::extract::State;
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use tracing::warn;
use crate::api::json_error;
use crate::runtime::AppState;
#[derive(Debug, Serialize, Deserialize)]
pub struct StateStatsResponse {
pub db_path: String,
pub schema_version: u32,
pub agent_count: usize,
pub enroll_token_count: usize,
pub expired_enroll_token_count: usize,
}
pub async fn state_stats(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.stats().await {
Ok(stats) => Ok((
StatusCode::OK,
Json(StateStatsResponse {
db_path: stats.db_path.display().to_string(),
schema_version: stats.schema_version,
agent_count: stats.agent_count,
enroll_token_count: stats.enroll_token_count,
expired_enroll_token_count: stats.expired_enroll_token_count,
}),
)),
Err(err) => {
warn!(error = %err, "failed to read state stats");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"state_stats_failed",
&err.to_string(),
))
}
}
}
wakey-linux/src/dhcp.rs
+17 -378
View File
@@ -1,11 +1,19 @@
use std::io::{self, ErrorKind};
use std::net::IpAddr;
use std::path::PathBuf; use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
use macaddr::MacAddr; mod leases;
use serde::{Deserialize, Serialize}; mod observations;
use wakey_core::{DhcpLease, DhcpLeaseWithState};
pub use leases::{
enrich_leases_with_nud_state, parse_dhcp_lease_line, read_dhcp_leases,
read_dhcp_leases_from_path, read_dhcp_leases_with_names, read_dhcp_leases_with_names_from_paths,
};
pub use observations::{
LocalDeviceObservation, LocalObservationStore, ObservedDhcpClient, ObservedNeighbor,
list_local_observations, list_local_observations_from_path, load_mac_name_cache,
load_mac_name_cache_from_path, load_observation_store, load_observation_store_from_path,
observe_dhcp_event, observe_neighbor_event,
};
const DEFAULT_DHCP_LEASES: &str = "/tmp/dhcp.leases"; const DEFAULT_DHCP_LEASES: &str = "/tmp/dhcp.leases";
const DEFAULT_MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json"; const DEFAULT_MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json";
@@ -14,363 +22,23 @@ const DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
const MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE"; const MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE";
const OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE"; const OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LocalObservationStore {
#[serde(default)]
pub dhcp_clients: std::collections::BTreeMap<String, ObservedDhcpClient>,
#[serde(default)]
pub neighbors: std::collections::BTreeMap<String, ObservedNeighbor>,
}
#[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) fn now_unix() -> u64 {
pub struct ObservedDhcpClient {
pub mac: String,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
pub last_action: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObservedNeighbor {
pub key: String,
pub mac: Option<String>,
pub ip: Option<IpAddr>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
pub last_action: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalDeviceObservation {
pub kind: String,
pub action: String,
pub mac: Option<String>,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
}
/// Load the MAC-to-name cache used to preserve useful names across lease churn.
pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<String, String>> {
load_mac_name_cache_from_path(mac_name_cache_path()).await
}
pub async fn load_mac_name_cache_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<std::collections::BTreeMap<String, String>> {
match tokio::fs::read_to_string(path).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
/// Persist the MAC-to-name cache back to disk.
async fn save_mac_name_cache(map: &std::collections::BTreeMap<String, String>) -> io::Result<()> {
save_mac_name_cache_to_path(mac_name_cache_path(), map).await
}
async fn save_mac_name_cache_to_path(
path: impl AsRef<std::path::Path>,
map: &std::collections::BTreeMap<String, String>,
) -> io::Result<()> {
let s = serde_json::to_string(map).map_err(io::Error::other)?;
let _ = tokio::fs::write(path, s).await;
Ok(())
}
pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
match tokio::fs::read_to_string(observation_store_path()).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
pub async fn load_observation_store_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<LocalObservationStore> {
match tokio::fs::read_to_string(path).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
async fn save_observation_store(store: &LocalObservationStore) -> io::Result<()> {
let s = serde_json::to_string(store).map_err(io::Error::other)?;
tokio::fs::write(observation_store_path(), s).await
}
pub async fn list_local_observations() -> io::Result<Vec<LocalDeviceObservation>> {
let store = load_observation_store().await?;
list_local_observations_from_store(store)
}
pub async fn list_local_observations_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<LocalDeviceObservation>> {
let store = load_observation_store_from_path(path).await?;
list_local_observations_from_store(store)
}
fn list_local_observations_from_store(
store: LocalObservationStore,
) -> io::Result<Vec<LocalDeviceObservation>> {
let mut out = Vec::with_capacity(store.dhcp_clients.len() + store.neighbors.len());
out.extend(
store
.dhcp_clients
.into_values()
.map(|row| LocalDeviceObservation {
kind: "dhcp".into(),
action: row.last_action,
mac: Some(row.mac),
ip: row.ip,
hostname: row.hostname,
first_seen_unix: row.first_seen_unix,
last_seen_unix: row.last_seen_unix,
}),
);
out.extend(
store
.neighbors
.into_values()
.map(|row| LocalDeviceObservation {
kind: "neigh".into(),
action: row.last_action,
mac: row.mac,
ip: row.ip,
hostname: None,
first_seen_unix: row.first_seen_unix,
last_seen_unix: row.last_seen_unix,
}),
);
out.sort_by(|a, b| {
b.last_seen_unix
.cmp(&a.last_seen_unix)
.then(a.kind.cmp(&b.kind))
.then(a.mac.cmp(&b.mac))
.then(a.ip.cmp(&b.ip))
});
Ok(out)
}
/// Observe a DHCP hotplug event and update the local MAC-to-name cache.
pub async fn observe_dhcp_event(
action: &str,
mac: MacAddr,
ip: Option<IpAddr>,
hostname: Option<&str>,
) -> io::Result<bool> {
if !matches!(action, "add" | "update" | "old" | "remove") {
// old not emitted by hotplug
return Ok(false);
}
let hostname = hostname
.map(str::trim)
.filter(|v| !v.is_empty() && *v != "*")
.map(ToOwned::to_owned);
let now = now_unix();
let mac_s = mac.to_string().to_ascii_lowercase();
let mut store = load_observation_store().await.unwrap_or_default();
let mut changed = false;
store
.dhcp_clients
.entry(mac_s.clone())
.and_modify(|row| {
if row.ip != ip
|| row.hostname != hostname
|| row.last_action != action
|| row.last_seen_unix != now
{
row.ip = ip;
row.hostname = hostname.clone();
row.last_action = action.to_string();
row.last_seen_unix = now;
changed = true;
}
})
.or_insert_with(|| {
changed = true;
ObservedDhcpClient {
mac: mac_s.clone(),
ip,
hostname: hostname.clone(),
first_seen_unix: now,
last_seen_unix: now,
last_action: action.to_string(),
}
});
if changed {
save_observation_store(&store).await?;
}
if let Some(hostname) = hostname {
let mut cache = load_mac_name_cache().await.unwrap_or_default();
if cache.get(&mac_s).map(|v| v != &hostname).unwrap_or(true) {
cache.insert(mac_s, hostname);
save_mac_name_cache(&cache).await?;
changed = true;
}
}
Ok(changed)
}
pub async fn observe_neighbor_event(
action: &str,
mac: Option<MacAddr>,
ip: Option<IpAddr>,
) -> io::Result<bool> {
if !matches!(action, "add" | "update" | "old" | "remove") {
// update and remove not emitted by hotplug
return Ok(false);
}
let Some(key) = mac
.map(|value| format!("mac:{}", value.to_string().to_ascii_lowercase()))
.or_else(|| ip.map(|value| format!("ip:{}", value)))
else {
return Ok(false);
};
let now = now_unix();
let mac = mac.map(|value| value.to_string().to_ascii_lowercase());
let mut store = load_observation_store().await.unwrap_or_default();
let mut changed = false;
store
.neighbors
.entry(key.clone())
.and_modify(|row| {
if row.mac != mac
|| row.ip != ip
|| row.last_action != action
|| row.last_seen_unix != now
{
row.mac = mac.clone();
row.ip = ip;
row.last_action = action.to_string();
row.last_seen_unix = now;
changed = true;
}
})
.or_insert_with(|| {
changed = true;
ObservedNeighbor {
key,
mac,
ip,
first_seen_unix: now,
last_seen_unix: now,
last_action: action.to_string(),
}
});
if changed {
save_observation_store(&store).await?;
}
Ok(changed)
}
/// Parse one `dnsmasq`-style DHCP lease line.
pub fn parse_dhcp_lease_line(line: &str) -> Option<DhcpLease> {
let mut c = line.split_whitespace();
let expires_epoch: u64 = c.next()?.parse().ok()?;
let mac = c.next()?.parse().ok()?;
let ip = c.next()?.parse().ok()?;
let name = c.next().filter(|c| *c != "*").map(str::to_string);
Some(DhcpLease {
expires_epoch,
ip,
mac,
name,
})
}
/// Read raw DHCP leases from the configured dnsmasq lease file.
pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
read_dhcp_leases_from_path(dhcp_leases_path()).await
}
pub async fn read_dhcp_leases_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<DhcpLease>> {
match tokio::fs::read_to_string(path).await {
Ok(file) => Ok(file.lines().filter_map(parse_dhcp_lease_line).collect()),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// Read DHCP leases and fill missing names from the MAC-name cache.
pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
read_dhcp_leases_with_names_from_paths(
dhcp_leases_path(),
observation_store_path(),
mac_name_cache_path(),
)
.await
}
pub async fn read_dhcp_leases_with_names_from_paths(
leases_path: impl AsRef<std::path::Path>,
observation_store_path: impl AsRef<std::path::Path>,
mac_name_cache_path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<DhcpLease>> {
let leases = read_dhcp_leases_from_path(leases_path).await?;
let observations = load_observation_store_from_path(observation_store_path)
.await
.unwrap_or_default();
let mac_name_cache_path = mac_name_cache_path.as_ref();
let mut cache = load_mac_name_cache_from_path(mac_name_cache_path)
.await
.unwrap_or_default();
let mut changed = false;
let mut leases_with_names = Vec::with_capacity(leases.len());
for mut l in leases {
let mac_s = l.mac.to_string();
if let Some(ref name) = l.name {
if cache.get(&mac_s).map(|v| v != name).unwrap_or(true) {
cache.insert(mac_s, name.clone());
changed = true;
}
} else if let Some(prev) = observations
.dhcp_clients
.get(&mac_s)
.and_then(|row| row.hostname.as_ref())
{
l.name = Some(prev.clone());
} else if let Some(prev) = cache.get(&mac_s) {
l.name = Some(prev.clone());
}
leases_with_names.push(l);
}
if changed {
let _ = save_mac_name_cache_to_path(mac_name_cache_path, &cache).await;
}
Ok(leases_with_names)
}
fn now_unix() -> u64 {
SystemTime::now() SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs()) .map(|duration| duration.as_secs())
.unwrap_or(0) .unwrap_or(0)
} }
fn dhcp_leases_path() -> PathBuf { pub(crate) fn dhcp_leases_path() -> PathBuf {
configured_path(DHCP_LEASES_ENV, DEFAULT_DHCP_LEASES) configured_path(DHCP_LEASES_ENV, DEFAULT_DHCP_LEASES)
} }
fn mac_name_cache_path() -> PathBuf { pub(crate) fn mac_name_cache_path() -> PathBuf {
configured_path(MAC_NAME_CACHE_ENV, DEFAULT_MAC_NAME_CACHE) configured_path(MAC_NAME_CACHE_ENV, DEFAULT_MAC_NAME_CACHE)
} }
fn observation_store_path() -> PathBuf { pub(crate) fn observation_store_path() -> PathBuf {
configured_path(OBSERVATION_STORE_ENV, DEFAULT_OBSERVATION_STORE) configured_path(OBSERVATION_STORE_ENV, DEFAULT_OBSERVATION_STORE)
} }
@@ -381,35 +49,6 @@ fn configured_path(env_key: &str, default: &str) -> PathBuf {
.unwrap_or_else(|| PathBuf::from(default)) .unwrap_or_else(|| PathBuf::from(default))
} }
/// Enrich DHCP leases with the best currently known neighbor state per IP.
pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLeaseWithState> {
let ips: Vec<IpAddr> = leases.iter().map(|l| l.ip).collect();
let mut map: std::collections::HashMap<IpAddr, wakey_core::NeighborState> =
std::collections::HashMap::new();
if let Ok(rows) =
crate::devices::get_neighbors(&[] as &[&str], &ips, &[] as &[&str], &[], &[]).await
{
for row in rows {
let state = row.state;
let r = state.rank();
map.entry(row.ip)
.and_modify(|e| {
if r > e.rank() {
*e = state
}
})
.or_insert(state);
}
}
leases
.into_iter()
.map(|lease_line| DhcpLeaseWithState {
nud_state: map.get(&lease_line.ip).copied(),
lease_line,
})
.collect()
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
wakey-linux/src/dhcp/leases.rs
+117
View File
@@ -0,0 +1,117 @@
use std::io::{self, ErrorKind};
use std::net::IpAddr;
use wakey_core::{DhcpLease, DhcpLeaseWithState};
use super::{dhcp_leases_path, mac_name_cache_path, observation_store_path};
use crate::dhcp::observations::{
load_mac_name_cache_from_path, load_observation_store_from_path, save_mac_name_cache_to_path,
};
/// Parse one `dnsmasq`-style DHCP lease line.
pub fn parse_dhcp_lease_line(line: &str) -> Option<DhcpLease> {
let mut c = line.split_whitespace();
let expires_epoch: u64 = c.next()?.parse().ok()?;
let mac = c.next()?.parse().ok()?;
let ip = c.next()?.parse().ok()?;
let name = c.next().filter(|c| *c != "*").map(str::to_string);
Some(DhcpLease {
expires_epoch,
ip,
mac,
name,
})
}
/// Read raw DHCP leases from the configured dnsmasq lease file.
pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
read_dhcp_leases_from_path(dhcp_leases_path()).await
}
pub async fn read_dhcp_leases_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<DhcpLease>> {
match tokio::fs::read_to_string(path).await {
Ok(file) => Ok(file.lines().filter_map(parse_dhcp_lease_line).collect()),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e),
}
}
/// Read DHCP leases and fill missing names from the MAC-name cache.
pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
read_dhcp_leases_with_names_from_paths(
dhcp_leases_path(),
observation_store_path(),
mac_name_cache_path(),
)
.await
}
pub async fn read_dhcp_leases_with_names_from_paths(
leases_path: impl AsRef<std::path::Path>,
observation_store_path: impl AsRef<std::path::Path>,
mac_name_cache_path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<DhcpLease>> {
let leases = read_dhcp_leases_from_path(leases_path).await?;
let observations = load_observation_store_from_path(observation_store_path)
.await
.unwrap_or_default();
let mac_name_cache_path = mac_name_cache_path.as_ref();
let mut cache = load_mac_name_cache_from_path(mac_name_cache_path)
.await
.unwrap_or_default();
let mut changed = false;
let mut leases_with_names = Vec::with_capacity(leases.len());
for mut l in leases {
let mac_s = l.mac.to_string();
if let Some(ref name) = l.name {
if cache.get(&mac_s).map(|v| v != name).unwrap_or(true) {
cache.insert(mac_s, name.clone());
changed = true;
}
} else if let Some(prev) = observations
.dhcp_clients
.get(&mac_s)
.and_then(|row| row.hostname.as_ref())
{
l.name = Some(prev.clone());
} else if let Some(prev) = cache.get(&mac_s) {
l.name = Some(prev.clone());
}
leases_with_names.push(l);
}
if changed {
let _ = save_mac_name_cache_to_path(mac_name_cache_path, &cache).await;
}
Ok(leases_with_names)
}
/// Enrich DHCP leases with the best currently known neighbor state per IP.
pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLeaseWithState> {
let ips: Vec<IpAddr> = leases.iter().map(|l| l.ip).collect();
let mut map: std::collections::HashMap<IpAddr, wakey_core::NeighborState> =
std::collections::HashMap::new();
if let Ok(rows) =
crate::devices::get_neighbors(&[] as &[&str], &ips, &[] as &[&str], &[], &[]).await
{
for row in rows {
let state = row.state;
let r = state.rank();
map.entry(row.ip)
.and_modify(|e| {
if r > e.rank() {
*e = state
}
})
.or_insert(state);
}
}
leases
.into_iter()
.map(|lease_line| DhcpLeaseWithState {
nud_state: map.get(&lease_line.ip).copied(),
lease_line,
})
.collect()
}
wakey-linux/src/dhcp/observations.rs
+269
View File
@@ -0,0 +1,269 @@
use std::io::{self, ErrorKind};
use std::net::IpAddr;
use macaddr::MacAddr;
use serde::{Deserialize, Serialize};
use super::{mac_name_cache_path, now_unix, observation_store_path};
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LocalObservationStore {
#[serde(default)]
pub dhcp_clients: std::collections::BTreeMap<String, ObservedDhcpClient>,
#[serde(default)]
pub neighbors: std::collections::BTreeMap<String, ObservedNeighbor>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObservedDhcpClient {
pub mac: String,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
pub last_action: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObservedNeighbor {
pub key: String,
pub mac: Option<String>,
pub ip: Option<IpAddr>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
pub last_action: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalDeviceObservation {
pub kind: String,
pub action: String,
pub mac: Option<String>,
pub ip: Option<IpAddr>,
pub hostname: Option<String>,
pub first_seen_unix: u64,
pub last_seen_unix: u64,
}
/// Load the MAC-to-name cache used to preserve useful names across lease churn.
pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<String, String>> {
load_mac_name_cache_from_path(mac_name_cache_path()).await
}
pub async fn load_mac_name_cache_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<std::collections::BTreeMap<String, String>> {
match tokio::fs::read_to_string(path).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
/// Persist the MAC-to-name cache back to disk.
async fn save_mac_name_cache(map: &std::collections::BTreeMap<String, String>) -> io::Result<()> {
save_mac_name_cache_to_path(mac_name_cache_path(), map).await
}
pub(super) async fn save_mac_name_cache_to_path(
path: impl AsRef<std::path::Path>,
map: &std::collections::BTreeMap<String, String>,
) -> io::Result<()> {
let s = serde_json::to_string(map).map_err(io::Error::other)?;
let _ = tokio::fs::write(path, s).await;
Ok(())
}
pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
match tokio::fs::read_to_string(observation_store_path()).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
pub async fn load_observation_store_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<LocalObservationStore> {
match tokio::fs::read_to_string(path).await {
Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e),
}
}
async fn save_observation_store(store: &LocalObservationStore) -> io::Result<()> {
let s = serde_json::to_string(store).map_err(io::Error::other)?;
tokio::fs::write(observation_store_path(), s).await
}
pub async fn list_local_observations() -> io::Result<Vec<LocalDeviceObservation>> {
let store = load_observation_store().await?;
list_local_observations_from_store(store)
}
pub async fn list_local_observations_from_path(
path: impl AsRef<std::path::Path>,
) -> io::Result<Vec<LocalDeviceObservation>> {
let store = load_observation_store_from_path(path).await?;
list_local_observations_from_store(store)
}
fn list_local_observations_from_store(
store: LocalObservationStore,
) -> io::Result<Vec<LocalDeviceObservation>> {
let mut out = Vec::with_capacity(store.dhcp_clients.len() + store.neighbors.len());
out.extend(
store
.dhcp_clients
.into_values()
.map(|row| LocalDeviceObservation {
kind: "dhcp".into(),
action: row.last_action,
mac: Some(row.mac),
ip: row.ip,
hostname: row.hostname,
first_seen_unix: row.first_seen_unix,
last_seen_unix: row.last_seen_unix,
}),
);
out.extend(
store
.neighbors
.into_values()
.map(|row| LocalDeviceObservation {
kind: "neigh".into(),
action: row.last_action,
mac: row.mac,
ip: row.ip,
hostname: None,
first_seen_unix: row.first_seen_unix,
last_seen_unix: row.last_seen_unix,
}),
);
out.sort_by(|a, b| {
b.last_seen_unix
.cmp(&a.last_seen_unix)
.then(a.kind.cmp(&b.kind))
.then(a.mac.cmp(&b.mac))
.then(a.ip.cmp(&b.ip))
});
Ok(out)
}
/// Observe a DHCP hotplug event and update the local MAC-to-name cache.
pub async fn observe_dhcp_event(
action: &str,
mac: MacAddr,
ip: Option<IpAddr>,
hostname: Option<&str>,
) -> io::Result<bool> {
if !matches!(action, "add" | "update" | "old" | "remove") {
// old not emitted by hotplug
return Ok(false);
}
let hostname = hostname
.map(str::trim)
.filter(|v| !v.is_empty() && *v != "*")
.map(ToOwned::to_owned);
let now = now_unix();
let mac_s = mac.to_string().to_ascii_lowercase();
let mut store = load_observation_store().await.unwrap_or_default();
let mut changed = false;
store
.dhcp_clients
.entry(mac_s.clone())
.and_modify(|row| {
if row.ip != ip
|| row.hostname != hostname
|| row.last_action != action
|| row.last_seen_unix != now
{
row.ip = ip;
row.hostname = hostname.clone();
row.last_action = action.to_string();
row.last_seen_unix = now;
changed = true;
}
})
.or_insert_with(|| {
changed = true;
ObservedDhcpClient {
mac: mac_s.clone(),
ip,
hostname: hostname.clone(),
first_seen_unix: now,
last_seen_unix: now,
last_action: action.to_string(),
}
});
if changed {
save_observation_store(&store).await?;
}
if let Some(hostname) = hostname {
let mut cache = load_mac_name_cache().await.unwrap_or_default();
if cache.get(&mac_s).map(|v| v != &hostname).unwrap_or(true) {
cache.insert(mac_s, hostname);
save_mac_name_cache(&cache).await?;
changed = true;
}
}
Ok(changed)
}
pub async fn observe_neighbor_event(
action: &str,
mac: Option<MacAddr>,
ip: Option<IpAddr>,
) -> io::Result<bool> {
if !matches!(action, "add" | "update" | "old" | "remove") {
// update and remove not emitted by hotplug
return Ok(false);
}
let Some(key) = mac
.map(|value| format!("mac:{}", value.to_string().to_ascii_lowercase()))
.or_else(|| ip.map(|value| format!("ip:{}", value)))
else {
return Ok(false);
};
let now = now_unix();
let mac = mac.map(|value| value.to_string().to_ascii_lowercase());
let mut store = load_observation_store().await.unwrap_or_default();
let mut changed = false;
store
.neighbors
.entry(key.clone())
.and_modify(|row| {
if row.mac != mac
|| row.ip != ip
|| row.last_action != action
|| row.last_seen_unix != now
{
row.mac = mac.clone();
row.ip = ip;
row.last_action = action.to_string();
row.last_seen_unix = now;
changed = true;
}
})
.or_insert_with(|| {
changed = true;
ObservedNeighbor {
key,
mac,
ip,
first_seen_unix: now,
last_seen_unix: now,
last_action: action.to_string(),
}
});
if changed {
save_observation_store(&store).await?;
}
Ok(changed)
}