isb_core/ingress/
cloudflare.rs

1//! The Cloudflare Tunnel provider: an org's hostnames reach its services
2//! through the org's own cloudflared instead of public listeners.
3//!
4//! - The org stores a remotely-managed tunnel's token as the secret
5//!   `cloudflare-tunnel-token`. isb runs cloudflared as a stack in the org
6//!   ([`tunnel_stack`]): an OCI image pinned by digest, unprivileged, on the
7//!   org's bridge behind the org's ACL. It never runs on the host, where a
8//!   tunnel configured from the Cloudflare dashboard could reach the host's
9//!   loopback services and sockets.
10//! - cloudflared sends the org's hostnames to the org's Caddy listener on
11//!   the bridge address (`http://<gateway>:<tunnel port>`), which routes
12//!   them like the public edge does.
13//! - With the secret `cloudflare-api-token` as well, isb manages the
14//!   tunnel's ingress rules and the hostnames' CNAMEs through the API
15//!   ([`sync`]); without it, the org points the hostnames at the origin URL
16//!   in the dashboard itself.
17
18use std::time::Duration;
19
20use base64::Engine;
21use serde::{Deserialize, Serialize};
22use serde_json::{Value, json};
23
24use crate::error::{Error, Result};
25
26/// The org secret holding the tunnel token.
27pub const TOKEN_SECRET: &str = "cloudflare-tunnel-token";
28/// The org secret holding an API token (Tunnel edit, DNS edit).
29pub const API_TOKEN_SECRET: &str = "cloudflare-api-token";
30/// The stack isb runs cloudflared in, in each tunnel org.
31pub const TUNNEL_STACK: &str = "isb-tunnel";
32/// cloudflared 2026.9.3, by the digest of its multi-arch index.
33pub const CLOUDFLARED_IMAGE: &str = "docker:cloudflare/cloudflared@sha256:072c067d25ccbe61d46e18f0d0723255f2bb5304f7317caa95b27031520ff92c";
34pub const API_BASE: &str = "https://api.cloudflare.com/client/v4";
35/// Marks the DNS records isb made, so it only ever deletes its own.
36pub const RECORD_COMMENT: &str = "managed by isb ingress";
37
38/// What a tunnel token says: account and tunnel ids (and a secret we never
39/// look at).
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct TunnelToken {
42    pub account: String,
43    pub tunnel: String,
44}
45
46/// Decode a tunnel token: base64 of `{"a": account, "t": tunnel, "s": ...}`.
47pub fn parse_token(token: &str) -> Result<TunnelToken> {
48    let t = token.trim();
49    let raw = base64::engine::general_purpose::STANDARD
50        .decode(t)
51        .or_else(|_| base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(t))
52        .map_err(|_| Error::invalid("the tunnel token is not base64"))?;
53    #[derive(Deserialize)]
54    struct T {
55        a: String,
56        t: String,
57    }
58    let v: T = serde_json::from_slice(&raw)
59        .map_err(|_| Error::invalid("the tunnel token is not a Cloudflare tunnel token"))?;
60    Ok(TunnelToken {
61        account: v.a,
62        tunnel: v.t,
63    })
64}
65
66/// The compose file of an org's cloudflared stack.
67pub fn tunnel_stack() -> crate::spec::ComposeFile {
68    let y = format!(
69        r#"
70secrets:
71  tunnel_token: {{external: true, name: {TOKEN_SECRET}}}
72services:
73  cloudflared:
74    image: "{CLOUDFLARED_IMAGE}"
75    command: [cloudflared, --no-autoupdate, tunnel, run]
76    environment:
77      TUNNEL_TOKEN: {{secret: tunnel_token}}
78    labels: {{isb.ingress: tunnel}}
79"#
80    );
81    serde_yaml_ng::from_str(&y).expect("the tunnel stack parses")
82}
83
84/// A Cloudflare API client.
85pub struct Api {
86    base: String,
87    token: String,
88    agent: ureq::Agent,
89}
90
91/// What [`sync`] changed.
92#[derive(Debug, Clone, Default, Serialize)]
93pub struct SyncReport {
94    pub ingress_rules: usize,
95    pub created: Vec<String>,
96    pub updated: Vec<String>,
97    pub deleted: Vec<String>,
98    /// Hostnames left alone, and why (a record isb did not make, no zone).
99    pub skipped: Vec<String>,
100}
101
102impl Api {
103    pub fn new(base: &str, token: &str) -> Api {
104        let agent: ureq::Agent = ureq::Agent::config_builder()
105            .timeout_global(Some(Duration::from_secs(20)))
106            .http_status_as_error(false)
107            .user_agent(concat!("isb/", env!("CARGO_PKG_VERSION")))
108            .build()
109            .into();
110        Api {
111            base: base.trim_end_matches('/').to_string(),
112            token: token.trim().to_string(),
113            agent,
114        }
115    }
116
117    /// One call; `result` of a `success: true` envelope.
118    fn call(&self, method: &str, path: &str, body: Option<&Value>) -> Result<Value> {
119        let url = format!("{}{path}", self.base);
120        let step = format!("cloudflare {method} {path}");
121        let fail = |m: String| Error::OperationFailed {
122            step: step.clone(),
123            message: m,
124        };
125        let auth = format!("Bearer {}", self.token);
126        let payload = match body {
127            Some(b) => serde_json::to_vec(b)?,
128            None => Vec::new(),
129        };
130        let resp = match method {
131            "GET" => self.agent.get(&url).header("Authorization", &auth).call(),
132            "DELETE" => self
133                .agent
134                .delete(&url)
135                .header("Authorization", &auth)
136                .call(),
137            "PUT" => self
138                .agent
139                .put(&url)
140                .header("Authorization", &auth)
141                .header("Content-Type", "application/json")
142                .send(&payload[..]),
143            "POST" => self
144                .agent
145                .post(&url)
146                .header("Authorization", &auth)
147                .header("Content-Type", "application/json")
148                .send(&payload[..]),
149            _ => return Err(fail("bad request".into())),
150        };
151        let mut resp = resp.map_err(|e| fail(e.to_string()))?;
152        let status = resp.status().as_u16();
153        let text = resp
154            .body_mut()
155            .with_config()
156            .limit(8 << 20)
157            .read_to_string()
158            .map_err(|e| fail(format!("HTTP {status}: {e}")))?;
159        let v: Value = serde_json::from_str(&text)
160            .map_err(|e| fail(format!("HTTP {status}: not JSON ({e})")))?;
161        if v["success"].as_bool() != Some(true) {
162            let errs: Vec<String> = v["errors"]
163                .as_array()
164                .into_iter()
165                .flatten()
166                .map(|e| format!("{} ({})", e["message"].as_str().unwrap_or("?"), e["code"]))
167                .collect();
168            return Err(fail(format!("HTTP {status}: {}", errs.join("; "))));
169        }
170        Ok(v["result"].clone())
171    }
172
173    /// Replace the tunnel's ingress rules: each hostname to `origin`, then
174    /// a 404 for anything else.
175    pub fn put_ingress(
176        &self,
177        account: &str,
178        tunnel: &str,
179        hosts: &[String],
180        origin: &str,
181    ) -> Result<()> {
182        let mut ingress: Vec<Value> = hosts
183            .iter()
184            .map(|h| json!({"hostname": h, "service": origin, "originRequest": {}}))
185            .collect();
186        ingress.push(json!({"service": "http_status:404"}));
187        self.call(
188            "PUT",
189            &format!(
190                "/accounts/{}/cfd_tunnel/{}/configurations",
191                seg(account),
192                seg(tunnel)
193            ),
194            Some(&json!({"config": {"ingress": ingress}})),
195        )?;
196        Ok(())
197    }
198
199    /// The zone a hostname is in: the longest suffix Cloudflare has a zone
200    /// for.
201    pub fn zone_for(&self, host: &str) -> Result<Option<String>> {
202        let h = host.trim_start_matches("*.");
203        let labels: Vec<&str> = h.split('.').collect();
204        for i in 0..labels.len().saturating_sub(1) {
205            let name = labels[i..].join(".");
206            let r = self.call("GET", &format!("/zones?name={}", seg(&name)), None)?;
207            if let Some(id) = r
208                .as_array()
209                .and_then(|a| a.first())
210                .and_then(|z| z["id"].as_str())
211            {
212                return Ok(Some(id.to_string()));
213            }
214        }
215        Ok(None)
216    }
217
218    fn records(&self, zone: &str, query: &str) -> Result<Vec<Value>> {
219        let r = self.call(
220            "GET",
221            &format!("/zones/{}/dns_records?per_page=100&{query}", seg(zone)),
222            None,
223        )?;
224        Ok(r.as_array().cloned().unwrap_or_default())
225    }
226}
227
228/// Percent-encode a path or query segment.
229fn seg(s: &str) -> String {
230    let mut out = String::new();
231    for b in s.bytes() {
232        if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'.' | b'_' | b'~') {
233            out.push(b as char);
234        } else {
235            out.push_str(&format!("%{b:02X}"));
236        }
237    }
238    out
239}
240
241/// What an org asks of the API.
242#[derive(Debug, Clone, Default)]
243pub struct SyncPlan {
244    pub account: String,
245    pub tunnel: String,
246    /// The zone all hostnames are in; looked up per hostname when unset.
247    pub zone: Option<String>,
248    pub hosts: Vec<String>,
249    /// `http://<gateway>:<port>`.
250    pub origin: String,
251}
252
253/// Point the tunnel's ingress at the org's listener and keep a proxied
254/// CNAME to the tunnel for each hostname. Records isb did not make are
255/// never changed; isb's own are deleted when their hostname goes.
256pub fn sync(api: &Api, plan: &SyncPlan) -> Result<SyncReport> {
257    let mut report = SyncReport::default();
258    let mut hosts = plan.hosts.clone();
259    hosts.sort();
260    hosts.dedup();
261    api.put_ingress(&plan.account, &plan.tunnel, &hosts, &plan.origin)?;
262    report.ingress_rules = hosts.len();
263    let target = format!("{}.cfargotunnel.com", plan.tunnel);
264    let mut zones: Vec<String> = Vec::new();
265    for h in &hosts {
266        let zone = match &plan.zone {
267            Some(z) => z.clone(),
268            None => match api.zone_for(h)? {
269                Some(z) => z,
270                None => {
271                    report
272                        .skipped
273                        .push(format!("{h}: no Cloudflare zone for it"));
274                    continue;
275                }
276            },
277        };
278        if !zones.contains(&zone) {
279            zones.push(zone.clone());
280        }
281        let existing = api.records(&zone, &format!("name={}", seg(h)))?;
282        let body = json!({
283            "type": "CNAME",
284            "name": h,
285            "content": target,
286            "proxied": true,
287            "ttl": 1,
288            "comment": RECORD_COMMENT,
289        });
290        match existing.first() {
291            None => {
292                api.call(
293                    "POST",
294                    &format!("/zones/{}/dns_records", seg(&zone)),
295                    Some(&body),
296                )?;
297                report.created.push(h.clone());
298            }
299            Some(r) if r["type"] == "CNAME" && r["content"] == json!(target) => {}
300            Some(r) if r["comment"] == json!(RECORD_COMMENT) => {
301                let id = r["id"].as_str().unwrap_or_default();
302                api.call(
303                    "PUT",
304                    &format!("/zones/{}/dns_records/{}", seg(&zone), seg(id)),
305                    Some(&body),
306                )?;
307                report.updated.push(h.clone());
308            }
309            Some(r) => report.skipped.push(format!(
310                "{h}: a {} record isb did not make exists; point it at {target} yourself",
311                r["type"].as_str().unwrap_or("DNS")
312            )),
313        }
314    }
315    // isb's records for this tunnel whose hostname went away.
316    if let Some(z) = &plan.zone {
317        if !zones.contains(z) {
318            zones.push(z.clone());
319        }
320    }
321    for zone in zones {
322        let ours = api.records(&zone, &format!("type=CNAME&content={}", seg(&target)))?;
323        for r in ours {
324            let name = r["name"].as_str().unwrap_or_default().to_string();
325            if r["comment"] == json!(RECORD_COMMENT) && !hosts.contains(&name) {
326                let id = r["id"].as_str().unwrap_or_default();
327                api.call(
328                    "DELETE",
329                    &format!("/zones/{}/dns_records/{}", seg(&zone), seg(id)),
330                    None,
331                )?;
332                report.deleted.push(name);
333            }
334        }
335    }
336    Ok(report)
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use std::io::{Read, Write};
343    use std::net::TcpListener;
344    use std::sync::{Arc, Mutex};
345
346    /// A fake Cloudflare API: records requests, keeps DNS records in memory.
347    struct Fake {
348        log: Arc<Mutex<Vec<(String, String, Value)>>>,
349        records: Arc<Mutex<Vec<Value>>>,
350        base: String,
351    }
352
353    #[expect(
354        clippy::too_many_lines,
355        clippy::excessive_nesting,
356        reason = "predates the lint ratchet; split it when next changed"
357    )]
358    fn fake() -> Fake {
359        let l = TcpListener::bind("127.0.0.1:0").unwrap();
360        let base = format!("http://{}/client/v4", l.local_addr().unwrap());
361        let log: Arc<Mutex<Vec<(String, String, Value)>>> = Arc::default();
362        let records: Arc<Mutex<Vec<Value>>> = Arc::new(Mutex::new(vec![
363            // Someone's own record: never touched.
364            json!({"id": "r-user", "type": "A", "name": "keep.example.com", "content": "192.0.2.1", "comment": null}),
365            // isb's, for a hostname no longer served.
366            json!({"id": "r-old", "type": "CNAME", "name": "old.example.com", "content": "tun-1.cfargotunnel.com", "comment": RECORD_COMMENT}),
367        ]));
368        let (lg, rs) = (log.clone(), records.clone());
369        std::thread::spawn(move || {
370            let mut next = 0;
371            for s in l.incoming() {
372                let Ok(mut s) = s else { break };
373                let mut buf = Vec::new();
374                let mut chunk = [0u8; 8192];
375                // Read the head and a Content-Length body.
376                let (head, body) = loop {
377                    let n = s.read(&mut chunk).unwrap();
378                    buf.extend_from_slice(&chunk[..n]);
379                    if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
380                        let head = String::from_utf8_lossy(&buf[..i]).to_string();
381                        let len: usize = head
382                            .lines()
383                            .find_map(|l| {
384                                l.to_ascii_lowercase()
385                                    .strip_prefix("content-length:")
386                                    .map(|v| v.trim().parse().unwrap())
387                            })
388                            .unwrap_or(0);
389                        while buf.len() < i + 4 + len {
390                            let n = s.read(&mut chunk).unwrap();
391                            buf.extend_from_slice(&chunk[..n]);
392                        }
393                        break (head, buf[i + 4..i + 4 + len].to_vec());
394                    }
395                };
396                let line = head.lines().next().unwrap().to_string();
397                let mut parts = line.split(' ');
398                let method = parts.next().unwrap().to_string();
399                let path = parts
400                    .next()
401                    .unwrap()
402                    .trim_start_matches("/client/v4")
403                    .to_string();
404                assert!(head.contains("Bearer api-tok"), "{head}");
405                let body: Value = serde_json::from_slice(&body).unwrap_or(Value::Null);
406                lg.lock()
407                    .unwrap()
408                    .push((method.clone(), path.clone(), body.clone()));
409                let (p, q) = path.split_once('?').unwrap_or((&path, ""));
410                let param = |k: &str| {
411                    q.split('&')
412                        .find_map(|kv| kv.strip_prefix(&format!("{k}=")))
413                        .map(|v| v.replace("%2A", "*"))
414                };
415                let result = match (method.as_str(), p) {
416                    ("GET", "/zones") => match param("name").as_deref() {
417                        Some("example.com") => json!([{"id": "zone-1"}]),
418                        _ => json!([]),
419                    },
420                    ("GET", "/zones/zone-1/dns_records") => {
421                        let rs = rs.lock().unwrap();
422                        let v: Vec<Value> = rs
423                            .iter()
424                            .filter(|r| param("name").is_none_or(|n| r["name"] == json!(n)))
425                            .filter(|r| param("type").is_none_or(|t| r["type"] == json!(t)))
426                            .filter(|r| param("content").is_none_or(|c| r["content"] == json!(c)))
427                            .cloned()
428                            .collect();
429                        json!(v)
430                    }
431                    ("POST", "/zones/zone-1/dns_records") => {
432                        next += 1;
433                        let mut r = body.clone();
434                        r["id"] = json!(format!("r-new-{next}"));
435                        rs.lock().unwrap().push(r.clone());
436                        r
437                    }
438                    ("DELETE", p) if p.starts_with("/zones/zone-1/dns_records/") => {
439                        let id = p.rsplit('/').next().unwrap().to_string();
440                        rs.lock().unwrap().retain(|r| r["id"] != json!(id));
441                        json!({"id": id})
442                    }
443                    ("PUT", "/accounts/acc-1/cfd_tunnel/tun-1/configurations") => json!({}),
444                    _ => {
445                        let resp = json!({"success": false, "errors": [{"code": 7003, "message": "no route"}]});
446                        let b = resp.to_string();
447                        let _ = write!(
448                            s,
449                            "HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}",
450                            b.len()
451                        );
452                        continue;
453                    }
454                };
455                let b = json!({"success": true, "errors": [], "result": result}).to_string();
456                let _ = write!(
457                    s,
458                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}",
459                    b.len()
460                );
461            }
462        });
463        Fake { log, records, base }
464    }
465
466    #[test]
467    fn tokens() {
468        let raw = r#"{"a":"acc-1","t":"tun-1","s":"c2VjcmV0"}"#;
469        let tok = base64::engine::general_purpose::STANDARD.encode(raw);
470        assert_eq!(
471            parse_token(&format!(" {tok}\n")).unwrap(),
472            TunnelToken {
473                account: "acc-1".into(),
474                tunnel: "tun-1".into()
475            }
476        );
477        assert!(parse_token("nope!").is_err());
478        assert!(parse_token(&base64::engine::general_purpose::STANDARD.encode("{}")).is_err());
479    }
480
481    #[test]
482    fn the_tunnel_stack() {
483        let f = tunnel_stack();
484        let s = &f.services["cloudflared"];
485        assert_eq!(s.image, CLOUDFLARED_IMAGE);
486        assert_eq!(s.env.secrets["TUNNEL_TOKEN"], "tunnel_token");
487        assert!(f.secrets["tunnel_token"].external);
488        assert!(s.domains.is_empty());
489    }
490
491    #[test]
492    fn sync_against_a_fake_api() {
493        let f = fake();
494        let api = Api::new(&f.base, "api-tok");
495        let plan = SyncPlan {
496            account: "acc-1".into(),
497            tunnel: "tun-1".into(),
498            zone: None,
499            hosts: vec![
500                "app.example.com".into(),
501                "keep.example.com".into(),
502                "app.example.com".into(),
503                "other.example.net".into(),
504            ],
505            origin: "http://10.70.1.1:8480".into(),
506        };
507        let r = sync(&api, &plan).unwrap();
508        assert_eq!(r.ingress_rules, 3);
509        assert_eq!(r.created, vec!["app.example.com"]);
510        assert_eq!(r.deleted, vec!["old.example.com"]);
511        assert_eq!(r.skipped.len(), 2, "{:?}", r.skipped);
512        assert!(
513            r.skipped
514                .iter()
515                .any(|s| s.contains("keep.example.com: a A record"))
516        );
517        assert!(
518            r.skipped
519                .iter()
520                .any(|s| s.contains("other.example.net: no Cloudflare zone"))
521        );
522
523        let log = f.log.lock().unwrap().clone();
524        let (m, p, body) = &log[0];
525        assert_eq!(
526            (m.as_str(), p.as_str()),
527            ("PUT", "/accounts/acc-1/cfd_tunnel/tun-1/configurations")
528        );
529        assert_eq!(
530            body,
531            &json!({"config": {"ingress": [
532                {"hostname": "app.example.com", "service": "http://10.70.1.1:8480", "originRequest": {}},
533                {"hostname": "keep.example.com", "service": "http://10.70.1.1:8480", "originRequest": {}},
534                {"hostname": "other.example.net", "service": "http://10.70.1.1:8480", "originRequest": {}},
535                {"service": "http_status:404"},
536            ]}})
537        );
538        let created = log.iter().find(|(m, _, _)| m == "POST").unwrap();
539        assert_eq!(created.2["content"], "tun-1.cfargotunnel.com");
540        assert_eq!(created.2["proxied"], true);
541        assert_eq!(created.2["comment"], RECORD_COMMENT);
542        let recs = f.records.lock().unwrap().clone();
543        assert!(
544            recs.iter().any(|r| r["id"] == "r-user"),
545            "the user's record stays"
546        );
547        assert!(!recs.iter().any(|r| r["id"] == "r-old"));
548
549        // Second run: nothing to do.
550        let r = sync(&api, &plan).unwrap();
551        assert!(r.created.is_empty() && r.updated.is_empty() && r.deleted.is_empty());
552    }
553
554    #[test]
555    fn api_errors_name_the_step() {
556        let f = fake();
557        let api = Api::new(&f.base, "api-tok");
558        let e = api
559            .put_ingress("acc-x", "tun-1", &[], "http://x")
560            .unwrap_err();
561        let s = e.to_string();
562        assert!(s.contains("cloudflare PUT /accounts/acc-x"), "{s}");
563        assert!(s.contains("no route (7003)"), "{s}");
564    }
565}
isb_core/ingress/cloudflare.rs

isb_core/ingress/
cloudflare.rs
