rtimed 0.14.0

High-performance NTP/PTP time synchronization daemon (NTPv4, NTS, PTP). Installs as the `rtime` binary.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use std::sync::Arc;

use axum::extract::{Request, State};
use axum::http::StatusCode;
use axum::middleware::{self, Next};
use axum::response::Response;
use axum::routing::get;
use axum::{Json, Router};
use serde::Serialize;
use subtle::ConstantTimeEq;
use tokio::sync::RwLock;

#[derive(Serialize)]
pub struct StatusResponse {
    pub version: String,
    pub uptime_seconds: f64,
    pub clock: ClockStatus,
    pub sources: Vec<SourceStatus>,
}

#[derive(Clone, Serialize)]
pub struct ClockStatus {
    pub offset_ms: f64,
    pub jitter_ms: f64,
    pub frequency_ppm: f64,
    pub stratum: u8,
    pub synchronized: bool,
}

impl Default for ClockStatus {
    fn default() -> Self {
        Self {
            offset_ms: 0.0,
            jitter_ms: 0.0,
            frequency_ppm: 0.0,
            stratum: 16,
            synchronized: false,
        }
    }
}

#[derive(Clone, Serialize)]
pub struct SourceStatus {
    pub id: String,
    pub offset_ms: f64,
    pub delay_ms: f64,
    pub jitter_ms: f64,
    pub stratum: u8,
    pub reachable: bool,
    pub selected: bool,
}

/// Shared daemon state accessible by the management API.
pub struct DaemonStatus {
    pub start_time: std::time::Instant,
    pub clock: ClockStatus,
    pub sources: Vec<SourceStatus>,
}

impl DaemonStatus {
    pub fn new() -> Self {
        Self {
            start_time: std::time::Instant::now(),
            clock: ClockStatus::default(),
            sources: Vec::new(),
        }
    }
}

pub fn management_router(status: Arc<RwLock<DaemonStatus>>, api_key: Option<String>) -> Router {
    let router = Router::new()
        .route("/api/v1/status", get(get_status))
        .route("/api/v1/sources", get(get_sources))
        .with_state(status);

    if let Some(key) = api_key {
        let key = Arc::new(key);
        router.layer(middleware::from_fn(move |req, next| {
            let key = Arc::clone(&key);
            auth_middleware(req, next, key)
        }))
    } else {
        router
    }
}

async fn auth_middleware(
    req: Request,
    next: Next,
    api_key: Arc<String>,
) -> Result<Response, StatusCode> {
    let auth_header = req
        .headers()
        .get("authorization")
        .and_then(|v| v.to_str().ok());

    match auth_header {
        Some(header) if header.starts_with("Bearer ") => {
            let token = &header.as_bytes()[7..];
            let expected = api_key.as_bytes();
            // Constant-time comparison to prevent timing side-channel attacks.
            // We first check length equality (which leaks length but not content),
            // then compare bytes in constant time.
            if token.len() == expected.len() && token.ct_eq(expected).into() {
                Ok(next.run(req).await)
            } else {
                Err(StatusCode::UNAUTHORIZED)
            }
        }
        _ => Err(StatusCode::UNAUTHORIZED),
    }
}

async fn get_status(State(status): State<Arc<RwLock<DaemonStatus>>>) -> Json<StatusResponse> {
    let state = status.read().await;
    let uptime = state.start_time.elapsed().as_secs_f64();

    Json(StatusResponse {
        version: env!("CARGO_PKG_VERSION").to_string(),
        uptime_seconds: uptime,
        clock: state.clock.clone(),
        sources: state.sources.clone(),
    })
}

async fn get_sources(State(status): State<Arc<RwLock<DaemonStatus>>>) -> Json<Vec<SourceStatus>> {
    let state = status.read().await;
    Json(state.sources.clone())
}