isb_apps/monitor/
mod.rs

1//! Uptime monitors: what users see of an org's apps, checked from outside
2//! the app every interval, with history, incidents and notifications.
3//!
4//! A monitor is an HTTP(S) request (status range, keyword present or
5//! absent, headers from secrets, redirects, a certificate expiry warning),
6//! a TCP connect, or an **app** monitor that follows an app by reference:
7//! its served domain's public URL, or with no domain the app's own
8//! endpoint. When the domain sits behind Cloudflare Access without a
9//! service token, it is checked hop by hop (`chain`): Cloudflare's edge,
10//! the org's tunnel, and the ingress. A **service** monitor does the same for one service of a
11//! compose stack. Every app with a served domain gets an app monitor of its
12//! own (`app-<name>`, `auto`), and every compose stack service with one a
13//! service monitor (`stack-<stack>-<service>`, `auto`, see [`auto`]), unless
14//! the org opts out or excludes it.
15//!
16//! Checks run in the daemon that runs the org's apps ([`service`]); state,
17//! thresholds and flap damping are in [`state`], the probes in [`probe`],
18//! history in [`store`]. Channels hear `monitor.down`, `monitor.up` and
19//! `monitor.cert_expiring` through the notifier like every other event.
20//!
21//! Definitions are kept in `<state>/orgs/<org>/monitors/monitors.json`, the
22//! org's settings beside them in `settings.json`.
23
24pub mod auto;
25mod chain;
26mod health_path;
27pub mod heartbeat;
28pub mod probe;
29pub mod service;
30pub mod state;
31pub mod store;
32mod target;
33mod view;
34
35use std::path::{Path, PathBuf};
36
37use serde::{Deserialize, Serialize};
38
39use crate::error::{Error, Result};
40use crate::org::OrgId;
41pub use service::Monitors;
42
43/// Monitors per org at most.
44pub const MAX_PER_ORG: usize = 200;
45/// The shortest interval, seconds.
46pub const MIN_INTERVAL: u64 = 30;
47/// The prefix of an app's own monitor.
48pub const AUTO_PREFIX: &str = "app-";
49/// The prefix of a compose stack service's own monitor.
50pub const STACK_PREFIX: &str = "stack-";
51/// The org secrets an app monitor presents to Cloudflare Access, when both
52/// exist and the monitor sets no Access headers of its own.
53pub const ACCESS_ID_SECRET: &str = "CF_ACCESS_CLIENT_ID";
54pub const ACCESS_SECRET_SECRET: &str = "CF_ACCESS_CLIENT_SECRET";
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
57#[serde(rename_all = "lowercase")]
58pub enum Kind {
59    Http,
60    Tcp,
61    App,
62    Service,
63}
64
65impl Kind {
66    pub fn as_str(self) -> &'static str {
67        match self {
68            Kind::Http => "http",
69            Kind::Tcp => "tcp",
70            Kind::App => "app",
71            Kind::Service => "service",
72        }
73    }
74}
75
76/// A request header: a plain `value`, or the value of org secret `secret`
77/// (read at check time, never shown).
78#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
79#[serde(deny_unknown_fields)]
80pub struct Header {
81    pub name: String,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub value: Option<String>,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub secret: Option<String>,
86}
87
88/// One monitor. Fields that do not apply to its type are refused.
89#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
90#[serde(deny_unknown_fields)]
91pub struct Monitor {
92    pub name: String,
93    #[serde(rename = "type")]
94    pub kind: Kind,
95    /// http: the URL.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub url: Option<String>,
98    /// tcp: host and port.
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub host: Option<String>,
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub port: Option<u16>,
103    /// app: the app, by name.
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub app: Option<String>,
106    /// service: the compose stack, by name.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub stack: Option<String>,
109    /// service: the stack's service, by name.
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub service: Option<String>,
112    /// app, service: which of its domains (default: the first one served).
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub domain: Option<String>,
115    /// app, service: the path to request (default: the path the service's healthcheck requests over HTTP, else the domain's path).
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub path: Option<String>,
118    /// `GET` or `HEAD`.
119    #[serde(default = "get")]
120    pub method: String,
121    /// Status codes that count as up: `200-399`, `200,204`, `200-299,301`.
122    #[serde(default = "expected")]
123    pub expected_status: String,
124    /// The body must contain this.
125    #[serde(default, skip_serializing_if = "Option::is_none")]
126    pub keyword: Option<String>,
127    /// The body must not contain this.
128    #[serde(default, skip_serializing_if = "Option::is_none")]
129    pub keyword_absent: Option<String>,
130    #[serde(default)]
131    pub follow_redirects: bool,
132    #[serde(default, skip_serializing_if = "Vec::is_empty")]
133    pub headers: Vec<Header>,
134    /// Seconds between checks (at least 30).
135    #[serde(default = "interval")]
136    pub interval: u64,
137    /// Seconds one check may take.
138    #[serde(default = "timeout")]
139    pub timeout: u64,
140    /// Failed checks in a row that make it down.
141    #[serde(default = "two")]
142    pub failure_threshold: u32,
143    /// Successful checks in a row that make it up again.
144    #[serde(default = "two")]
145    pub recovery_threshold: u32,
146    /// Warn this many days before an HTTPS certificate expires (0: never).
147    #[serde(default = "cert_days")]
148    pub cert_expiry_days: u32,
149    #[serde(default)]
150    pub paused: bool,
151    /// Made for an app (`app-<name>`) or a compose stack service
152    /// (`stack-<stack>-<service>`) with a served domain.
153    #[serde(default)]
154    pub auto: bool,
155    #[serde(default)]
156    pub created_at: u64,
157    #[serde(default)]
158    pub updated_at: u64,
159}
160
161fn get() -> String {
162    "GET".into()
163}
164fn expected() -> String {
165    "200-399".into()
166}
167fn interval() -> u64 {
168    60
169}
170fn timeout() -> u64 {
171    10
172}
173fn two() -> u32 {
174    2
175}
176fn cert_days() -> u32 {
177    14
178}
179
180/// An org's monitoring settings.
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(deny_unknown_fields)]
183pub struct Settings {
184    /// Give every app with a served domain a monitor of its own.
185    #[serde(default = "yes")]
186    pub auto_monitors: bool,
187    /// Apps that get no monitor of their own.
188    #[serde(default, skip_serializing_if = "Vec::is_empty")]
189    pub exclude_apps: Vec<String>,
190    /// Compose stack services that get no monitor of their own, as
191    /// `<stack>/<service>`.
192    #[serde(default, skip_serializing_if = "Vec::is_empty")]
193    pub exclude_services: Vec<String>,
194}
195
196fn yes() -> bool {
197    true
198}
199
200impl Default for Settings {
201    fn default() -> Settings {
202        Settings {
203            auto_monitors: true,
204            exclude_apps: Vec::new(),
205            exclude_services: Vec::new(),
206        }
207    }
208}
209
210/// A monitor name: [a-z0-9-], a letter first, at most 63.
211pub fn validate_name(n: &str) -> Result<()> {
212    let ok = !n.is_empty()
213        && n.len() <= 63
214        && n.starts_with(|c: char| c.is_ascii_lowercase())
215        && n.chars()
216            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
217    if ok {
218        Ok(())
219    } else {
220        Err(Error::invalid(format!(
221            "monitor name {n:?}: [a-z0-9-], starting with a letter, at most 63 characters"
222        )))
223    }
224}
225
226/// A compose service's name, as a service monitor refers to it.
227pub fn validate_service_name(s: &str) -> Result<()> {
228    let ok = !s.is_empty()
229        && s.len() <= 63
230        && s.chars()
231            .all(|c| c.is_ascii_alphanumeric() || "._-".contains(c));
232    if ok {
233        Ok(())
234    } else {
235        Err(Error::invalid(format!(
236            "service {s:?}: a compose service's name ([A-Za-z0-9._-], at most 63)"
237        )))
238    }
239}
240
241/// Parse `200-399`, `200,204,300-399` into inclusive ranges.
242pub fn parse_status(s: &str) -> Result<Vec<(u16, u16)>> {
243    let bad = || {
244        Error::invalid(format!(
245            "expected_status {s:?}: codes and ranges like 200-399 or 200,204"
246        ))
247    };
248    let mut out = Vec::new();
249    for part in s.split(',').map(str::trim) {
250        let (a, b) = part.split_once('-').unwrap_or((part, part));
251        let a: u16 = a.trim().parse().map_err(|_| bad())?;
252        let b: u16 = b.trim().parse().map_err(|_| bad())?;
253        if !(100..=599).contains(&a) || !(100..=599).contains(&b) || a > b {
254            return Err(bad());
255        }
256        out.push((a, b));
257    }
258    if out.is_empty() || out.len() > 10 {
259        return Err(bad());
260    }
261    Ok(out)
262}
263
264/// Headers isb sets itself, or that would change how the request is framed.
265const RESERVED_HEADERS: &[&str] = &[
266    "host",
267    "content-length",
268    "transfer-encoding",
269    "connection",
270    "upgrade",
271];
272
273fn validate_headers(hs: &[Header]) -> Result<()> {
274    if hs.len() > 10 {
275        return Err(Error::invalid("at most 10 headers"));
276    }
277    for h in hs {
278        let n = &h.name;
279        let token = !n.is_empty()
280            && n.len() <= 64
281            && n.chars()
282                .all(|c| c.is_ascii_alphanumeric() || "!#$%&'*+-.^_`|~".contains(c));
283        if !token || RESERVED_HEADERS.contains(&n.to_ascii_lowercase().as_str()) {
284            return Err(Error::invalid(format!("header name {n:?}")));
285        }
286        match (&h.value, &h.secret) {
287            (Some(v), None) if v.len() <= 1024 && !v.contains(['\r', '\n']) => {}
288            (None, Some(s)) => crate::secrets::validate_name(s)?,
289            _ => {
290                return Err(Error::invalid(format!(
291                    "header {n}: a value (without line breaks, at most 1024) or a secret, not both"
292                )));
293            }
294        }
295    }
296    Ok(())
297}
298
299fn validate_tcp_host(h: &str) -> Result<()> {
300    let ok = !h.is_empty()
301        && h.len() <= 253
302        && h.chars()
303            .all(|c| c.is_ascii_alphanumeric() || ".-:[]".contains(c));
304    if ok {
305        Ok(())
306    } else {
307        Err(Error::invalid(format!(
308            "host {h:?} is not a host name or address"
309        )))
310    }
311}
312
313impl Monitor {
314    /// A bare monitor of `kind` with every default.
315    pub fn new(name: &str, kind: Kind) -> Monitor {
316        serde_json::from_value(serde_json::json!({"name": name, "type": kind}))
317            .expect("defaults deserialize")
318    }
319
320    /// Check everything but whether an app or secret exists.
321    pub fn validate(&self) -> Result<()> {
322        validate_name(&self.name)?;
323        self.validate_target()?;
324        if !(MIN_INTERVAL..=86_400).contains(&self.interval) {
325            return Err(Error::invalid(format!(
326                "interval: {MIN_INTERVAL} to 86400 seconds"
327            )));
328        }
329        if !(1..=60).contains(&self.timeout) || self.timeout >= self.interval {
330            return Err(Error::invalid(
331                "timeout: 1 to 60 seconds, under the interval",
332            ));
333        }
334        for (what, n) in [
335            ("failure_threshold", self.failure_threshold),
336            ("recovery_threshold", self.recovery_threshold),
337        ] {
338            if !(1..=10).contains(&n) {
339                return Err(Error::invalid(format!("{what}: 1 to 10 checks")));
340            }
341        }
342        if self.cert_expiry_days > 365 {
343            return Err(Error::invalid("cert_expiry_days: 0 to 365"));
344        }
345        Ok(())
346    }
347
348    fn validate_target(&self) -> Result<()> {
349        let http_only = self.keyword.is_some()
350            || self.keyword_absent.is_some()
351            || !self.headers.is_empty()
352            || self.follow_redirects;
353        match self.kind {
354            Kind::Http => {
355                let url = self
356                    .url
357                    .as_deref()
358                    .ok_or_else(|| Error::invalid("an http monitor needs url"))?;
359                crate::net::parse_url(url).map_err(|e| Error::invalid(format!("url: {e}")))?;
360                self.refuse(&[
361                    ("host", self.host.is_some()),
362                    ("port", self.port.is_some()),
363                    ("app", self.app.is_some()),
364                    (
365                        "stack or service",
366                        self.stack.is_some() || self.service.is_some(),
367                    ),
368                    ("domain", self.domain.is_some()),
369                    ("path", self.path.is_some()),
370                ])?;
371            }
372            Kind::Tcp => {
373                validate_tcp_host(self.host.as_deref().unwrap_or_default())?;
374                if self.port.is_none_or(|p| p == 0) {
375                    return Err(Error::invalid("a tcp monitor needs host and port"));
376                }
377                self.refuse(&[
378                    ("url", self.url.is_some()),
379                    ("app", self.app.is_some()),
380                    (
381                        "stack or service",
382                        self.stack.is_some() || self.service.is_some(),
383                    ),
384                    ("domain", self.domain.is_some()),
385                    ("path", self.path.is_some()),
386                    ("keyword, headers or follow_redirects", http_only),
387                ])?;
388            }
389            Kind::App => {
390                let app = self.app.as_deref().unwrap_or_default();
391                crate::app::validate_app_name(app)
392                    .map_err(|_| Error::invalid(format!("app {app:?}: an app's name")))?;
393                self.validate_path()?;
394                self.refuse(&[
395                    ("url", self.url.is_some()),
396                    ("host", self.host.is_some()),
397                    ("port", self.port.is_some()),
398                    (
399                        "stack or service",
400                        self.stack.is_some() || self.service.is_some(),
401                    ),
402                ])?;
403            }
404            Kind::Service => {
405                let stack = self.stack.as_deref().unwrap_or_default();
406                crate::stack::validate_stack_name(stack)
407                    .map_err(|_| Error::invalid(format!("stack {stack:?}: a stack's name")))?;
408                validate_service_name(self.service.as_deref().unwrap_or_default())?;
409                self.validate_path()?;
410                self.refuse(&[
411                    ("url", self.url.is_some()),
412                    ("host", self.host.is_some()),
413                    ("port", self.port.is_some()),
414                    ("app", self.app.is_some()),
415                ])?;
416            }
417        }
418        if self.kind != Kind::Tcp {
419            if !matches!(self.method.as_str(), "GET" | "HEAD") {
420                return Err(Error::invalid("method: GET or HEAD"));
421            }
422            parse_status(&self.expected_status)?;
423            for k in [&self.keyword, &self.keyword_absent].into_iter().flatten() {
424                if k.is_empty() || k.len() > 200 {
425                    return Err(Error::invalid("keywords: 1 to 200 characters"));
426                }
427            }
428            validate_headers(&self.headers)?;
429        }
430        Ok(())
431    }
432
433    fn validate_path(&self) -> Result<()> {
434        match &self.path {
435            Some(p) if !p.starts_with('/') || p.len() > 1024 || p.contains(char::is_whitespace) => {
436                Err(Error::invalid("path: starts with /, no spaces"))
437            }
438            _ => Ok(()),
439        }
440    }
441
442    /// Does it follow something by reference (an app or a stack service)?
443    pub fn follows(&self) -> bool {
444        matches!(self.kind, Kind::App | Kind::Service)
445    }
446
447    fn refuse(&self, fields: &[(&str, bool)]) -> Result<()> {
448        match fields.iter().find(|(_, set)| *set) {
449            Some((f, _)) => Err(Error::invalid(format!(
450                "{f} does not apply to a {} monitor",
451                self.kind.as_str()
452            ))),
453            None => Ok(()),
454        }
455    }
456
457    /// The secrets the monitor reads.
458    pub fn secrets(&self) -> Vec<&str> {
459        self.headers
460            .iter()
461            .filter_map(|h| h.secret.as_deref())
462            .collect()
463    }
464
465    /// What it checks, in a few words.
466    pub fn target(&self) -> String {
467        match self.kind {
468            Kind::Http => probe::display_url(self.url.as_deref().unwrap_or_default()),
469            Kind::Tcp => format!(
470                "{}:{}",
471                self.host.as_deref().unwrap_or_default(),
472                self.port.unwrap_or_default()
473            ),
474            Kind::App => format!("app {}", self.app.as_deref().unwrap_or_default()),
475            Kind::Service => format!(
476                "service {}/{}",
477                self.stack.as_deref().unwrap_or_default(),
478                self.service.as_deref().unwrap_or_default()
479            ),
480        }
481    }
482}
483
484/// Where an org's monitors live.
485pub fn dir(state: &Path, org: &OrgId) -> PathBuf {
486    org.dir(state).join("monitors")
487}
488
489pub(crate) fn read_json<T: for<'de> Deserialize<'de> + Default>(p: &Path) -> Result<T> {
490    match std::fs::read(p) {
491        Ok(b) => {
492            serde_json::from_slice(&b).map_err(|e| Error::invalid(format!("{}: {e}", p.display())))
493        }
494        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(T::default()),
495        Err(e) => Err(e.into()),
496    }
497}
498
499pub(crate) fn write_json<T: Serialize>(p: &Path, v: &T) -> Result<()> {
500    if let Some(d) = p.parent() {
501        std::fs::create_dir_all(d)?;
502    }
503    let tmp = p.with_extension("tmp");
504    std::fs::write(&tmp, serde_json::to_vec_pretty(v)?)?;
505    std::fs::rename(&tmp, p)?;
506    Ok(())
507}
508
509/// A duration in words: `4m 12s`, `2h 5m`, `3d 1h`.
510pub fn human(ms: u64) -> String {
511    let s = ms / 1000;
512    match s {
513        0..60 => format!("{s}s"),
514        60..3600 => format!("{}m {}s", s / 60, s % 60),
515        3600..86_400 => format!("{}h {}m", s / 3600, s % 3600 / 60),
516        _ => format!("{}d {}h", s / 86_400, s % 86_400 / 3600),
517    }
518}
519
520#[cfg(test)]
521mod tests {
522    use super::*;
523    use serde_json::json;
524
525    fn m(v: serde_json::Value) -> Result<Monitor> {
526        let m: Monitor = serde_json::from_value(v).map_err(|e| Error::invalid(e.to_string()))?;
527        m.validate()?;
528        Ok(m)
529    }
530
531    #[test]
532    fn defaults_and_validation() {
533        let h =
534            m(json!({"name": "shop", "type": "http", "url": "https://shop.example.com/"})).unwrap();
535        assert_eq!(
536            (
537                h.interval,
538                h.timeout,
539                h.failure_threshold,
540                h.recovery_threshold
541            ),
542            (60, 10, 2, 2)
543        );
544        assert_eq!(
545            (
546                h.method.as_str(),
547                h.expected_status.as_str(),
548                h.cert_expiry_days
549            ),
550            ("GET", "200-399", 14)
551        );
552        assert!(m(json!({"name": "x", "type": "http"})).is_err());
553        assert!(m(json!({"name": "x", "type": "http", "url": "ftp://a"})).is_err());
554        assert!(
555            m(json!({"name": "x", "type": "http", "url": "https://a/", "interval": 10})).is_err()
556        );
557        assert!(m(json!({"name": "x", "type": "http", "url": "https://a/", "interval": 30, "timeout": 30})).is_err());
558        assert!(m(json!({"name": "x", "type": "http", "url": "https://a/", "port": 80})).is_err());
559        assert!(
560            m(json!({"name": "x", "type": "http", "url": "https://a/", "method": "POST"})).is_err()
561        );
562        assert!(
563            m(json!({"name": "x", "type": "http", "url": "https://a/", "failure_threshold": 0}))
564                .is_err()
565        );
566        assert!(m(json!({"name": "X", "type": "http", "url": "https://a/"})).is_err());
567        assert!(m(json!({"name": "x", "type": "http", "url": "https://a/", "bogus": 1})).is_err());
568        m(json!({"name": "db", "type": "tcp", "host": "db.example.com", "port": 5432})).unwrap();
569        assert!(m(json!({"name": "db", "type": "tcp", "host": "db.example.com"})).is_err());
570        assert!(m(json!({"name": "db", "type": "tcp", "host": "a b", "port": 1})).is_err());
571        assert!(
572            m(json!({"name": "db", "type": "tcp", "host": "a", "port": 1, "keyword": "x"}))
573                .is_err()
574        );
575        m(json!({"name": "web", "type": "app", "app": "web", "path": "/healthz"})).unwrap();
576        assert!(m(json!({"name": "web", "type": "app", "app": "web", "path": "healthz"})).is_err());
577        assert!(
578            m(json!({"name": "web", "type": "app", "app": "web", "url": "https://a/"})).is_err()
579        );
580    }
581
582    #[test]
583    fn headers_and_status_ranges() {
584        let ok = json!({"name": "a", "type": "http", "url": "https://a/", "headers": [
585            {"name": "CF-Access-Client-Id", "secret": "CF_ID"}, {"name": "Accept", "value": "text/html"}]});
586        assert_eq!(m(ok).unwrap().secrets(), vec!["CF_ID"]);
587        for bad in [
588            json!([{"name": "Host", "value": "x"}]),
589            json!([{"name": "X", "value": "a\r\nb: c"}]),
590            json!([{"name": "X", "value": "a", "secret": "S"}]),
591            json!([{"name": "X"}]),
592            json!([{"name": "Bad Name", "value": "a"}]),
593        ] {
594            assert!(
595                m(json!({"name": "a", "type": "http", "url": "https://a/", "headers": bad}))
596                    .is_err(),
597                "{bad}"
598            );
599        }
600        assert_eq!(parse_status("200-399").unwrap(), vec![(200, 399)]);
601        assert_eq!(
602            parse_status("200, 204,300-301").unwrap(),
603            vec![(200, 200), (204, 204), (300, 301)]
604        );
605        for bad in ["", "abc", "399-200", "99", "200-700", "200,"] {
606            assert!(parse_status(bad).is_err(), "{bad}");
607        }
608    }
609
610    #[test]
611    fn words() {
612        assert_eq!(human(4_000), "4s");
613        assert_eq!(human(252_000), "4m 12s");
614        assert_eq!(human(7_500_000), "2h 5m");
615        assert_eq!(human(266_400_000), "3d 2h");
616        assert_eq!(Monitor::new("x", Kind::Tcp).interval, 60);
617    }
618}
isb_apps/monitor/mod.rs

isb_apps/monitor/
mod.rs
