netget/server/consul/
wire.rs

1//! Consul HTTP API shapes, shared by the server and the client: KV entries with their base64
2//! values, catalog and health entries built from the few fields a handler decides, and the
3//! agent's self description. Measured against Consul 1.20.2.
4use anyhow::{bail, ensure, Context, Result};
5use base64::Engine;
6use serde_json::{json, Map, Value};
7
8/// Largest KV value Consul accepts, and the largest request body here.
9pub const MAX_VALUE_BYTES: usize = 512 * 1024;
10/// Most entries in one KV listing or catalog answer.
11pub const MAX_ENTRIES: usize = 10_000;
12pub const DATACENTER: &str = "dc1";
13pub const NODE: &str = "netget";
14pub const VERSION: &str = "1.20.2";
15
16pub fn b64(bytes: &[u8]) -> String {
17    base64::engine::general_purpose::STANDARD.encode(bytes)
18}
19
20pub fn unb64(s: &str) -> Result<Vec<u8>> {
21    base64::engine::general_purpose::STANDARD
22        .decode(s)
23        .context("invalid base64 value")
24}
25
26/// A value as the handler sees it: text when it is UTF-8, otherwise hex, with which one.
27pub fn shown(bytes: &[u8]) -> (String, &'static str) {
28    match std::str::from_utf8(bytes) {
29        // Text is shown as text when its only control characters are line breaks and tabs.
30        Ok(s) if !crate::utils::sanitize::has_controls(&s.replace(['\n', '\t'], "")) => {
31            (s.to_string(), "utf8")
32        }
33        _ => (hex::encode(bytes), "hex"),
34    }
35}
36
37/// The bytes of an action's `value`, by its `encoding` (utf8 by default, or hex).
38pub fn value_bytes(v: &Value) -> Result<Vec<u8>> {
39    let value = match &v["value"] {
40        Value::Null => return Ok(Vec::new()),
41        Value::String(s) => s.clone(),
42        other => other.to_string(),
43    };
44    let bytes = match v["encoding"].as_str().unwrap_or("utf8") {
45        "utf8" => value.into_bytes(),
46        "hex" => hex::decode(value.trim()).context("invalid hex value")?,
47        e => bail!("encoding must be utf8 or hex, not {e}"),
48    };
49    ensure!(
50        bytes.len() <= MAX_VALUE_BYTES,
51        "value larger than {MAX_VALUE_BYTES} bytes"
52    );
53    Ok(bytes)
54}
55
56pub fn check_key(key: &str) -> Result<()> {
57    ensure!(key.len() <= 2048, "key longer than 2048 bytes");
58    ensure!(
59        !key.contains(['\0', '\r', '\n']),
60        "key contains a control character"
61    );
62    Ok(())
63}
64
65/// One KV entry as the API returns it.
66pub fn kv_entry(e: &Value, index: u64) -> Result<Value> {
67    let key = e["key"].as_str().context("each entry needs a key")?;
68    check_key(key)?;
69    let value = value_bytes(e)?;
70    let modify = e["modify_index"].as_u64().unwrap_or(index);
71    Ok(
72        json!({"LockIndex": 0, "Key": key, "Flags": e["flags"].as_u64().unwrap_or(0),
73              "Value": if e["value"].is_null() { Value::Null } else { json!(b64(&value)) },
74              "CreateIndex": e["create_index"].as_u64().unwrap_or(modify), "ModifyIndex": modify}),
75    )
76}
77
78fn node() -> Value {
79    json!({"ID": "00000000-0000-0000-0000-00000000c0de", "Node": NODE, "Address": "127.0.0.1",
80           "Datacenter": DATACENTER, "TaggedAddresses": {"lan": "127.0.0.1", "wan": "127.0.0.1"},
81           "Meta": {}, "CreateIndex": 1, "ModifyIndex": 1})
82}
83
84/// A service instance the handler describes: {id?, name, address?, port?, tags?, meta?}.
85pub struct Instance {
86    pub id: String,
87    pub name: String,
88    pub address: String,
89    pub port: u64,
90    pub tags: Vec<String>,
91    pub meta: Map<String, Value>,
92}
93
94pub fn instance(v: &Value) -> Result<Instance> {
95    let name = v["name"]
96        .as_str()
97        .context("each service needs a name")?
98        .to_string();
99    ensure!(
100        !name.is_empty() && name.len() <= 256,
101        "service name must be 1..=256 bytes"
102    );
103    let port = v["port"].as_u64().unwrap_or(0);
104    ensure!(port <= 65535, "port out of range");
105    let tags = v["tags"]
106        .as_array()
107        .map(|a| {
108            a.iter()
109                .filter_map(Value::as_str)
110                .map(str::to_string)
111                .collect()
112        })
113        .unwrap_or_default();
114    Ok(Instance {
115        id: v["id"].as_str().unwrap_or(&name).to_string(),
116        address: v["address"].as_str().unwrap_or_default().to_string(),
117        port,
118        tags,
119        meta: v["meta"].as_object().cloned().unwrap_or_default(),
120        name,
121    })
122}
123
124fn service(i: &Instance, index: u64) -> Value {
125    json!({"ID": i.id, "Service": i.name, "Tags": i.tags, "Address": i.address, "Port": i.port,
126           "Meta": i.meta, "Weights": {"Passing": 1, "Warning": 1}, "EnableTagOverride": false,
127           "Datacenter": DATACENTER, "CreateIndex": index, "ModifyIndex": index})
128}
129
130/// `/v1/catalog/service/<name>` entries.
131pub fn catalog_entry(i: &Instance, index: u64) -> Value {
132    let mut out = node();
133    let fields = json!({"ServiceKind": "", "ServiceID": i.id, "ServiceName": i.name, "ServiceTags": i.tags,
134        "ServiceAddress": i.address, "ServiceWeights": {"Passing": 1, "Warning": 1}, "ServiceMeta": i.meta,
135        "ServicePort": i.port, "ServiceEnableTagOverride": false, "ServiceTaggedAddresses": {},
136        "CreateIndex": index, "ModifyIndex": index});
137    for (k, v) in fields.as_object().unwrap() {
138        out[k] = v.clone();
139    }
140    out
141}
142
143/// `/v1/health/service/<name>` entries, each with a passing serf check.
144pub fn health_entry(i: &Instance, index: u64) -> Value {
145    json!({"Node": node(), "Service": service(i, index), "Checks": [{
146        "Node": NODE, "CheckID": "serfHealth", "Name": "Serf Health Status", "Status": "passing",
147        "Notes": "", "Output": "Agent alive and reachable", "ServiceID": "", "ServiceName": "",
148        "ServiceTags": [], "Type": "", "CreateIndex": index, "ModifyIndex": index}]})
149}
150
151/// `/v1/agent/services`: a map of ID to service.
152pub fn agent_service(i: &Instance) -> Value {
153    let mut s = service(i, 0);
154    let m = s.as_object_mut().unwrap();
155    m.remove("CreateIndex");
156    m.remove("ModifyIndex");
157    s
158}
159
160/// `/v1/agent/self`: enough of a description for clients that read it.
161pub fn agent_self() -> Value {
162    json!({"Config": {"Datacenter": DATACENTER, "PrimaryDatacenter": DATACENTER, "NodeName": NODE,
163                      "NodeID": "00000000-0000-0000-0000-00000000c0de", "Server": true,
164                      "Version": VERSION, "Revision": "netget"},
165           "Member": {"Name": NODE, "Addr": "127.0.0.1", "Port": 8301, "Tags": {"dc": DATACENTER}, "Status": 1},
166           "Meta": {}})
167}
168
169/// A field of a request body, matched without regard to case: Consul decodes JSON with Go's
170/// encoding/json, which does, and clients rely on it (py-consul sends lower-case keys).
171fn field<'a>(body: &'a Value, name: &str) -> &'a Value {
172    body.as_object()
173        .and_then(|m| {
174            m.get(name).or_else(|| {
175                m.iter()
176                    .find(|(k, _)| k.eq_ignore_ascii_case(name))
177                    .map(|(_, v)| v)
178            })
179        })
180        .unwrap_or(&Value::Null)
181}
182
183/// A service registration body as the handler sees it.
184pub fn registration(body: &Value) -> Result<Value> {
185    let name = field(body, "Name").as_str().context("Name is required")?;
186    ensure!(
187        !name.is_empty() && name.len() <= 256,
188        "Name must be 1..=256 bytes"
189    );
190    Ok(
191        json!({"id": field(body, "ID").as_str().unwrap_or(name), "name": name,
192              "address": field(body, "Address").as_str().unwrap_or_default(),
193              "port": field(body, "Port").as_u64().unwrap_or(0), "tags": field(body, "Tags").clone(),
194              "meta": field(body, "Meta").clone()}),
195    )
196}
netget/server/consul/wire.rs

netget/server/consul/
wire.rs
