vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarmctl: log in with the operator's own SSO account and reach the controller through the swarm domain

Browse source 
`swarmctl --swarm https://<swarm> login` runs the OAuth 2.0
authorization-code flow with PKCE against the swarm's authelia as a new
public `swarmctl` client: a pushed authorization request, the browser login
and explicit consent, and the code delivered by form_post to a loopback
port. The access and refresh tokens are cached at mode 0600 under
$XDG_STATE_HOME/swarmctl/ and refreshed under a file lock, since authelia
rotates the refresh token and revokes the login when a used one is
presented again. With --swarm (or SWARMCTL_SWARM), the agent, schedule and
forge verbs send their requests to https://<swarm>/api/ with the bearer
token instead of over the controller's unix socket, which stays the default.

authelia registers the client on every authelia host with a swarm domain:
`kind = "native"`, public, no minted secret, scopes offline_access and
authelia.bearer.authz, audience https://<swarm>/api/ (matched as a URL
prefix), redirect http://127.0.0.1/callback (any loopback port). Every
field is one authelia 4.39.20's validator requires for a public client
holding the bearer scope.

The swarm apex's /api/ answers a refused caller that sent an
Authorization header with the 401 itself instead of the login redirect,
so the CLI can tell "log in again" apart from an answer; a browser still
gets the redirect. The bearer token is not forwarded to the controller.

Refs #4409 (mara: c94764, c94782).
This commit is contained in:
atlas 2026-10-11 23:43:29 +02:00 • committed by mara
commit 2df19e4899
17 changed files with 1118 additions and 122 deletions
Download patch file Download diff file Expand all files Collapse all files

84
swarmctl/src/agent.rs
View file

@ -29,6 +29,8 @@ use hyper_util::rt::TokioIo;
use serde::{Deserialize, Serialize};
use tokio::net::UnixStream;
use crate::remote::Target;
/// Body of `POST /api/agents`. Mirrors the controller's own
/// `CreateAgentRequest` — see this module's doc comment.
#[derive(Serialize)]
@ -70,7 +72,7 @@ struct MintForgeTokenResponse {
/// built here keeps that cost inside the one arm that incurs it, the same
/// way `PathArgs::resolve` is called only in the arms that need the
/// deployment's env vars.
pub(crate) fn create(socket: &Path, name: &str, hive: &str) -> Result<()> {
pub(crate) fn create(target: &Target, name: &str, hive: &str) -> Result<()> {
// Client-side before the round trip, so a typo is an immediate local
// error rather than a 400 the operator has to wait for. The controller
// validates both again — that is the gate that matters, this is the
@ -79,10 +81,10 @@ pub(crate) fn create(socket: &Path, name: &str, hive: &str) -> Result<()> {
let hive = parse_ident(hive, "hive")?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")?;
let resp = rt.block_on(post_create(socket, &name, &hive))?;
let resp = rt.block_on(post_create(target, &name, &hive))?;
println!("queued: job node {}", resp.node_id);
println!(
@ -110,17 +112,17 @@ pub(crate) fn parse_ident(value: &str, what: &str) -> Result<String> {
///
/// Synchronous for the same reason [`create`] is, and built on the same
/// round trip.
pub(crate) fn mint_identity(socket: &Path, name: &str) -> Result<()> {
pub(crate) fn mint_identity(target: &Target, name: &str) -> Result<()> {
// Client-side first, so a typo is a local error rather than a 400 the
// operator waits for. The controller validates it again.
let name = parse_ident(name, "agent name")?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")?;
let resp: MintIdentityResponse = rt.block_on(post(
socket,
target,
&format!("/api/agents/{name}/identity"),
&serde_json::json!({}),
"mint-identity",
@ -139,15 +141,15 @@ pub(crate) fn mint_identity(socket: &Path, name: &str) -> Result<()> {
///
/// Synchronous for the same reason [`create`] is, and built on the same
/// round trip. No `--hive`: the token's store path has no hive in it.
pub(crate) fn mint_forge_token(socket: &Path, name: &str) -> Result<()> {
pub(crate) fn mint_forge_token(target: &Target, name: &str) -> Result<()> {
let name = parse_ident(name, "agent name")?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")?;
let resp: MintForgeTokenResponse = rt.block_on(post(
socket,
target,
&format!("/api/agents/{name}/forge-token"),
&serde_json::json!({}),
"mint-forge-token",
@ -162,10 +164,10 @@ pub(crate) fn mint_forge_token(socket: &Path, name: &str) -> Result<()> {
Ok(())
}
/// One `POST /api/agents` round trip over the controller's unix socket.
async fn post_create(socket: &Path, name: &str, hive: &str) -> Result<CreateAgentResponse> {
/// One `POST /api/agents` round trip at the controller.
async fn post_create(target: &Target, name: &str, hive: &str) -> Result<CreateAgentResponse> {
post(
socket,
target,
"/api/agents",
&CreateAgentRequest { name, hive },
"create-agent",
@ -173,25 +175,25 @@ async fn post_create(socket: &Path, name: &str, hive: &str) -> Result<CreateAgen
.await
}
/// One JSON `POST` to `uri` over the controller's unix socket.
/// One JSON `POST` to `uri` at the controller.
///
/// `what` names the request in error messages — the operator needs to know
/// which call failed, and every other part of this function is identical
/// between the two verbs.
pub(crate) async fn post<Req: Serialize, Resp: serde::de::DeserializeOwned>(
socket: &Path,
target: &Target,
uri: &str,
request: &Req,
what: &str,
) -> Result<Resp> {
send(socket, hyper::Method::POST, uri, Some(request), what).await
send(target, hyper::Method::POST, uri, Some(request), what).await
}
/// One request to `uri` over the controller's unix socket, with `request` as
/// its JSON body when there is one. A `204` decodes as JSON `null`, so a
/// caller expecting no body asks for `()`.
/// One request to `uri` at the controller, with `request` as its JSON body
/// when there is one. A `204` decodes as JSON `null`, so a caller expecting no
/// body asks for `()`.
pub(crate) async fn send<Req: Serialize, Resp: serde::de::DeserializeOwned>(
socket: &Path,
target: &Target,
method: hyper::Method,
uri: &str,
request: Option<&Req>,
@ -202,6 +204,34 @@ pub(crate) async fn send<Req: Serialize, Resp: serde::de::DeserializeOwned>(
.transpose()
.with_context(|| format!("serialising the {what} request"))?;
let (status, body) = match target {
Target::Socket(socket) => over_socket(socket, method, uri, body, what).await?,
Target::Swarm(swarm) => swarm.request(method, uri, body, what).await?,
};
if !status.is_success() {
bail!("{}", describe_error(status, &body));
}
let body = if status == hyper::StatusCode::NO_CONTENT {
bytes::Bytes::from_static(b"null")
} else {
body
};
serde_json::from_slice(&body).with_context(|| {
format!(
"swarm-controller answered {status} but its body is not a {what} response: {}",
String::from_utf8_lossy(&body).trim()
)
})
}
async fn over_socket(
socket: &Path,
method: hyper::Method,
uri: &str,
body: Option<Vec<u8>>,
what: &str,
) -> Result<(hyper::StatusCode, bytes::Bytes)> {
let stream = UnixStream::connect(socket)
.await
.with_context(|| connect_hint(socket))?;
@ -237,21 +267,7 @@ pub(crate) async fn send<Req: Serialize, Resp: serde::de::DeserializeOwned>(
.await
.context("reading swarm-controller's response")?
.to_bytes();
if !status.is_success() {
bail!("{}", describe_error(status, &body));
}
let body = if status == hyper::StatusCode::NO_CONTENT {
bytes::Bytes::from_static(b"null")
} else {
body
};
serde_json::from_slice(&body).with_context(|| {
format!(
"swarm-controller answered {status} but its body is not a {what} response: {}",
String::from_utf8_lossy(&body).trim()
)
})
Ok((status, body))
}
/// Turn a failed response into one line the operator can act on.

9
swarmctl/src/forge.rs
View file

@ -6,12 +6,11 @@
//! socket; the shape below mirrors the controller's own private
//! `MakeForgeAdminResponse` for the reason that module's doc comment gives.
use std::path::Path;
use anyhow::{Context as _, Result};
use serde::Deserialize;
use crate::agent::{parse_ident, post};
use crate::remote::Target;
/// Success body of `POST /api/forge/users/{name}/admin`.
#[derive(Deserialize)]
@ -29,15 +28,15 @@ enum AdminChange {
/// Run `swarmctl forge make-admin`.
///
/// Synchronous for the same reason `agent create` is.
pub(crate) fn make_admin(socket: &Path, name: &str) -> Result<()> {
pub(crate) fn make_admin(target: &Target, name: &str) -> Result<()> {
let name = parse_ident(name, "user name")?;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")?;
let resp: MakeForgeAdminResponse = rt.block_on(post(
socket,
target,
&format!("/api/forge/users/{name}/admin"),
&serde_json::json!({}),
"forge-make-admin",

19
swarmctl/src/inference.rs
View file

@ -9,12 +9,11 @@
//! The shapes below mirror the controller's `InferenceSettings` and
//! `AgentInferenceView`, for the reason [`crate::agent`]'s module doc gives.
use std::path::Path;
use anyhow::{Context as _, Result, bail};
use serde::{Deserialize, Serialize};
use crate::agent::{parse_ident, send};
use crate::remote::Target;
/// One agent's settings; an absent field is not set at swarm level.
#[derive(Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
@ -66,7 +65,7 @@ impl SetFlags {
fn runtime() -> Result<tokio::runtime::Runtime> {
tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")
}
@ -79,10 +78,10 @@ fn member(agent: &str) -> Result<String> {
}
/// Run `swarmctl agent inference get`.
pub(crate) fn get(socket: &Path, agent: &str) -> Result<()> {
pub(crate) fn get(target: &Target, agent: &str) -> Result<()> {
let uri = member(agent)?;
let view: View = runtime()?.block_on(send(
socket,
target,
hyper::Method::GET,
&uri,
None::<&()>,
@ -98,11 +97,11 @@ pub(crate) fn get(socket: &Path, agent: &str) -> Result<()> {
/// Run `swarmctl agent inference set`: read the agent's settings, write the
/// flags over them, and store the result.
pub(crate) fn set(socket: &Path, agent: &str, flags: SetFlags) -> Result<()> {
pub(crate) fn set(target: &Target, agent: &str, flags: SetFlags) -> Result<()> {
let uri = member(agent)?;
let rt = runtime()?;
let current: View = rt.block_on(send(
socket,
target,
hyper::Method::GET,
&uri,
None::<&()>,
@ -110,7 +109,7 @@ pub(crate) fn set(socket: &Path, agent: &str, flags: SetFlags) -> Result<()> {
))?;
let wanted = flags.merge(current.configured.unwrap_or_default())?;
let stored: Settings = rt.block_on(send(
socket,
target,
hyper::Method::PUT,
&uri,
Some(&wanted),
@ -121,10 +120,10 @@ pub(crate) fn set(socket: &Path, agent: &str, flags: SetFlags) -> Result<()> {
}
/// Run `swarmctl agent inference clear`.
pub(crate) fn clear(socket: &Path, agent: &str) -> Result<()> {
pub(crate) fn clear(target: &Target, agent: &str) -> Result<()> {
let uri = member(agent)?;
runtime()?.block_on(send::<(), ()>(
socket,
target,
hyper::Method::DELETE,
&uri,
None,

387
swarmctl/src/login.rs Normal file
View file

@ -0,0 +1,387 @@
//! `swarmctl login` — the OAuth 2.0 authorization-code flow with PKCE against
//! the swarm's authelia, as the public `swarmctl` client.
//!
//! authelia only lets a client holding `authelia.bearer.authz` use a pushed
//! authorization request (PAR), explicit consent, and a `form_post` response:
//! the code arrives as a browser form POST to a loopback port this process
//! listens on. The registered redirect URI carries no port; authelia matches a
//! loopback redirect on any port (RFC 8252 §7.3).
use std::sync::{Arc, Mutex};
use std::time::Duration;
use anyhow::{Context as _, Result, bail};
use base64::Engine as _;
use http_body_util::{BodyExt as _, Full};
use hyper_util::rt::TokioIo;
use serde::Deserialize;
use sha2::Digest as _;
use tokio::net::TcpListener;
use crate::remote::{CLIENT_ID, Cached, Swarm, lock, oauth_json, token_request, write_cache};
/// authelia's PAR `request_uri` lives this long, so waiting longer is futile.
const WAIT: Duration = Duration::from_mins(5);
const CALLBACK_PATH: &str = "/callback";
/// Run `swarmctl login`.
pub(crate) fn login(swarm: &Swarm) -> Result<()> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("starting a tokio runtime for the login")?;
let path = swarm.cache_path()?;
rt.block_on(async {
let http = swarm.http()?;
let auth = discover_auth(&http, swarm).await?;
let listener = TcpListener::bind("127.0.0.1:0")
.await
.context("opening a loopback port for the login callback")?;
let redirect_uri = format!(
"http://127.0.0.1:{}{CALLBACK_PATH}",
listener.local_addr()?.port()
);
let verifier = random_token()?;
let state = random_token()?;
let audience = swarm.audience();
let par: ParResponse = oauth_json(
http.post(format!("{auth}/api/oidc/pushed-authorization-request"))
.form(&[
("client_id", CLIENT_ID),
("response_type", "code"),
("response_mode", "form_post"),
("redirect_uri", &redirect_uri),
("scope", "offline_access authelia.bearer.authz"),
("audience", &audience),
("state", &state),
("code_challenge", &challenge(&verifier)),
("code_challenge_method", "S256"),
])
.send()
.await
.with_context(|| format!("pushing the authorization request to {auth}"))?,
)
.await
.context("pushing the authorization request")?;
let mut url = url::Url::parse(&format!("{auth}/api/oidc/authorization"))?;
url.query_pairs_mut()
.append_pair("client_id", CLIENT_ID)
.append_pair("request_uri", &par.request_uri);
eprintln!("Log in to {} in your browser:\n\n {url}\n", swarm.origin);
open_browser(url.as_str());
let code = tokio::time::timeout(WAIT, receive_code(&listener, &state, &auth))
.await
.context("the login was not completed within 5 minutes")??;
let token = token_request(
&http,
&auth,
&[
("grant_type", "authorization_code"),
("client_id", CLIENT_ID),
("code", &code),
("redirect_uri", &redirect_uri),
("code_verifier", &verifier),
],
)
.await
.context("exchanging the authorization code")?;
let refresh_token = token
.refresh_token
.context("authelia issued no refresh token")?;
let _lock = lock(&path)?;
write_cache(
&path,
&Cached {
auth,
access_token: token.access_token,
refresh_token,
expires_at: chrono::Utc::now().timestamp() + token.expires_in,
},
)
})?;
println!(
"logged in to {}; login kept in {}",
swarm.origin,
path.display()
);
Ok(())
}
#[derive(Deserialize)]
struct ParResponse {
request_uri: String,
}
/// authelia's origin, read off the login redirect the gateway gives a
/// request that carries no credential.
async fn discover_auth(http: &reqwest::Client, swarm: &Swarm) -> Result<String> {
let url = swarm.origin.join("api/")?;
let resp = http
.get(url.clone())
.send()
.await
.with_context(|| format!("reaching {url}"))?;
let location = resp
.headers()
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| url::Url::parse(v).ok())
.filter(|l| resp.status().is_redirection() && l.scheme() == "https");
let Some(location) = location else {
bail!(
"{url} answered {} without an https redirect to the swarm's login, so its SSO \
provider is unknown",
resp.status()
);
};
Ok(location.origin().ascii_serialization())
}
/// 32 bytes from the kernel CSPRNG, base64url: a PKCE verifier (RFC 7636 §4.1
/// asks for 43–128 characters) or an unguessable `state`.
fn random_token() -> Result<String> {
let mut bytes = [0u8; 32];
getrandom::fill(&mut bytes)
.map_err(|e| anyhow::anyhow!("drawing randomness from the kernel: {e}"))?;
Ok(base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes))
}
/// The S256 PKCE challenge for `verifier`.
fn challenge(verifier: &str) -> String {
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(sha2::Sha256::digest(verifier))
}
/// Best effort: the URL is printed first, for a host with no browser.
fn open_browser(url: &str) {
let opener = if cfg!(target_os = "macos") {
"open"
} else {
"xdg-open"
};
let _ = std::process::Command::new(opener)
.arg(url)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn();
}
/// What one request to the loopback listener means.
#[derive(Debug, PartialEq, Eq)]
enum Callback {
/// Not the callback, or not this login's: answer it and keep waiting.
Ignored(hyper::StatusCode, &'static str),
Code(String),
/// authelia answered this login with an error.
Refused(String),
}
/// Judge one request. A `state` other than this login's is refused without
/// ending the wait: a stale tab from an earlier attempt must not be able to
/// complete or abort this one.
fn callback(method: &hyper::Method, path: &str, body: &[u8], state: &str, auth: &str) -> Callback {
if method != hyper::Method::POST || path != CALLBACK_PATH {
return Callback::Ignored(hyper::StatusCode::NOT_FOUND, "not found");
}
let field = |name: &str| {
url::form_urlencoded::parse(body)
.find(|(k, _)| k == name)
.map(|(_, v)| v.into_owned())
};
if field("state").as_deref() != Some(state) {
return Callback::Ignored(
hyper::StatusCode::BAD_REQUEST,
"This login answer is not for the swarmctl login that is waiting.",
);
}
if let Some(iss) = field("iss")
&& iss.trim_end_matches('/') != auth
{
return Callback::Refused(format!("the login answer came from {iss}, not {auth}"));
}
if let Some(error) = field("error") {
let description = field("error_description").unwrap_or_default();
return Callback::Refused(format!("authelia refused the login: {error} {description}"));
}
match field("code") {
Some(code) => Callback::Code(code),
None => Callback::Refused("the login answer carried no code".into()),
}
}
/// Serve the loopback listener until this login's answer arrives.
async fn receive_code(listener: &TcpListener, state: &str, auth: &str) -> Result<String> {
loop {
let (stream, _) = listener.accept().await.context("accepting the callback")?;
let outcome: Arc<Mutex<Option<Callback>>> = Arc::default();
let service = hyper::service::service_fn(|req: hyper::Request<hyper::body::Incoming>| {
let outcome = Arc::clone(&outcome);
async move {
let (parts, body) = req.into_parts();
let body = body.collect().await?.to_bytes();
let judged = callback(&parts.method, parts.uri.path(), &body, state, auth);
let (status, text) = match &judged {
Callback::Ignored(status, text) => (*status, *text),
Callback::Code(_) => (
hyper::StatusCode::OK,
"swarmctl is logged in. You can close this tab.",
),
Callback::Refused(_) => (
hyper::StatusCode::BAD_REQUEST,
"The login failed; swarmctl prints why.",
),
};
if !matches!(judged, Callback::Ignored(..)) {
*outcome.lock().expect("callback lock") = Some(judged);
}
Ok::<_, hyper::Error>(
hyper::Response::builder()
.status(status)
.header(hyper::header::CONTENT_TYPE, "text/plain; charset=utf-8")
.body(Full::new(bytes::Bytes::from_static(text.as_bytes())))
.expect("static response"),
)
}
});
let _ = hyper::server::conn::http1::Builder::new()
.keep_alive(false)
.serve_connection(TokioIo::new(stream), service)
.await;
match outcome.lock().expect("callback lock").take() {
Some(Callback::Code(code)) => return Ok(code),
Some(Callback::Refused(why)) => bail!("{why}"),
Some(Callback::Ignored(..)) | None => {}
}
}
}
#[cfg(test)]
mod tests {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use super::*;
const AUTH: &str = "https://auth.example.test";
/// RFC 7636 appendix B.
#[test]
fn the_challenge_matches_the_rfc_vector() {
assert_eq!(
challenge("dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"),
"E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
);
}
#[test]
fn verifiers_are_long_enough_url_safe_and_fresh() {
let a = random_token().expect("random");
let b = random_token().expect("random");
assert_eq!(a.len(), 43);
assert!(
a.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
);
assert_ne!(a, b);
}
fn post(body: &str) -> Callback {
callback(
&hyper::Method::POST,
CALLBACK_PATH,
body.as_bytes(),
"s1",
AUTH,
)
}
#[test]
fn the_matching_state_yields_the_code() {
let body = format!(
"code=c%2B1&iss={}&scope=x&state=s1",
AUTH.replace(':', "%3A")
);
assert_eq!(post(&body), Callback::Code("c+1".into()));
}
#[test]
fn a_wrong_or_missing_state_is_refused_without_ending_the_wait() {
for body in ["code=c1&state=s2", "code=c1"] {
assert!(
matches!(
post(body),
Callback::Ignored(hyper::StatusCode::BAD_REQUEST, _)
),
"{body}"
);
}
}
#[test]
fn an_error_or_a_foreign_issuer_ends_the_login() {
assert!(matches!(
post("error=access_denied&state=s1"),
Callback::Refused(_)
));
assert!(matches!(
post("code=c1&iss=https%3A%2F%2Fevil.test&state=s1"),
Callback::Refused(_)
));
}
#[test]
fn other_paths_and_methods_are_ignored() {
let get = callback(&hyper::Method::GET, CALLBACK_PATH, b"state=s1", "s1", AUTH);
let favicon = callback(
&hyper::Method::POST,
"/favicon.ico",
b"state=s1",
"s1",
AUTH,
);
assert!(matches!(
get,
Callback::Ignored(hyper::StatusCode::NOT_FOUND, _)
));
assert!(matches!(
favicon,
Callback::Ignored(hyper::StatusCode::NOT_FOUND, _)
));
}
/// Over a real socket: a stale tab's POST is answered 400 and the
/// listener keeps waiting for the one carrying this login's state.
#[tokio::test]
async fn the_listener_skips_a_wrong_state_and_returns_the_right_code() {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
let addr = listener.local_addr().expect("addr");
let server = tokio::spawn(async move { receive_code(&listener, "s1", AUTH).await });
let send = |body: &'static str| async move {
let mut s = tokio::net::TcpStream::connect(addr).await.expect("connect");
let req = format!(
"POST /callback HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: \
application/x-www-form-urlencoded\r\nContent-Length: {}\r\n\r\n{body}",
body.len()
);
s.write_all(req.as_bytes()).await.expect("write");
let mut answer = String::new();
s.read_to_string(&mut answer).await.expect("read");
answer
};
assert!(
send("code=stale&state=s0")
.await
.starts_with("HTTP/1.1 400")
);
assert!(
send("code=fresh&state=s1")
.await
.starts_with("HTTP/1.1 200")
);
assert_eq!(server.await.expect("join").expect("code"), "fresh");
}
}

129
swarmctl/src/main.rs
View file

@ -12,16 +12,14 @@
//! root.
//!
//! So this binary **serves** no socket, publishes no HTTP route and has
//! no privileged helper. When a verb eventually has to run as a non-root
//! user or from another host, the answer is a **group-gated admin
//! socket** — separate from the controller's `0666` gateway-facing one —
//! not a widening of what root does here.
//! no privileged helper.
//!
//! Acting directly is not the same as acting *alone*, though: `agent
//! create` is a client of the controller's own unix socket, because the
//! work it asks for is a job graph only the controller can queue (see
//! [`agent`]). That is the opposite direction from the socket the
//! paragraph above rules out — nothing here becomes reachable by it.
//! Acting directly is not the same as acting *alone*, though: the `agent`,
//! `schedule` and `forge` verbs are clients of the controller, because the
//! work they ask for is only the controller's to do (see [`agent`]). They
//! reach it over its unix socket on its host, or, with `--swarm`, through
//! the swarm domain's gateway as the operator's own SSO account
//! ([`remote`]). The `user` verbs act on the local users file only.
//!
//! Distinct from `hivectl`, which drives one hive's `hive-c0re` over its
//! admin socket. This crate deliberately does not link `swarm-controller`,
@ -30,6 +28,8 @@
mod agent;
mod forge;
mod inference;
mod login;
mod remote;
mod schedule;
mod users;
@ -42,6 +42,7 @@ use std::process::Command;
use anyhow::{Context, Result, bail};
use clap::{Args, Parser, Subcommand};
use remote::Target;
use users::{User, UserStore};
#[derive(Parser)]
@ -49,10 +50,51 @@ use users::{User, UserStore};
struct Cli {
#[command(flatten)]
paths: PathArgs,
#[command(flatten)]
remote: RemoteArgs,
#[command(subcommand)]
command: Verb,
}
/// Which swarm to reach through its domain rather than the local socket.
#[derive(Args)]
struct RemoteArgs {
/// The swarm's domain, as `https://<domain>`. Given, the `agent`,
/// `schedule` and `forge` verbs go through its gateway as the account
/// `swarmctl login` logged in, instead of over the controller's socket.
/// Falls back to `SWARMCTL_SWARM`.
#[arg(long, value_name = "URL")]
swarm: Option<String>,
/// A PEM CA certificate to trust beside the system roots, for a swarm
/// whose gateway serves a certificate from the swarm's own CA. Falls back
/// to `SWARMCTL_CA_FILE`.
#[arg(long, value_name = "PATH")]
ca_file: Option<PathBuf>,
}
impl RemoteArgs {
fn swarm(&self) -> Result<Option<remote::Swarm>> {
let url = self
.swarm
.clone()
.or_else(|| std::env::var("SWARMCTL_SWARM").ok())
.filter(|u| !u.is_empty());
let ca_file = self
.ca_file
.clone()
.or_else(|| std::env::var_os("SWARMCTL_CA_FILE").map(PathBuf::from));
url.map(|u| remote::Swarm::parse(&u, ca_file)).transpose()
}
/// The swarm when one is named, the controller's socket otherwise.
fn target(&self, socket: Option<PathBuf>) -> Result<Target> {
Ok(match self.swarm()? {
Some(swarm) => Target::Swarm(swarm),
None => Target::Socket(path_from(socket, "SWARM_CONTROLLER_SOCKET")?),
})
}
}
/// Where the deployment put the things this CLI has to touch.
///
/// Every one of these is supplied by the nix module that installs
@ -141,6 +183,14 @@ enum Verb {
#[command(subcommand)]
command: ForgeVerb,
},
/// Log in to the swarm named by `--swarm` as your own SSO account.
///
/// Opens the swarm's SSO login in your browser (and prints the URL) and
/// receives the answer on a loopback port, so run it where that browser
/// runs. swarmctl keeps the login, mode 0600, under
/// `$XDG_STATE_HOME/swarmctl/` and refreshes it as it's used. To log out,
/// delete that file.
Login,
/// Emit the full CLI reference as `CommonMark` to stdout.
///
/// Hidden tooling command used by the docs build to keep the published
@ -183,8 +233,8 @@ enum AgentVerb {
/// own clock. Watch the swarm UI's job view, or the hive itself, for
/// the rest.
///
/// No approval gate guards this: running this binary already means
/// being root on the controller's host.
/// No approval gate guards this: the caller is root on the controller's
/// host, or logged in through the swarm domain's SSO gate.
Create(AgentCreateArgs),
/// Queue a re-mint of an existing agent's identity at the swarm's
/// secret store.
@ -446,8 +496,8 @@ struct ControllerSocketArg {
}
impl ControllerSocketArg {
fn resolve(self) -> Result<PathBuf> {
path_from(self.controller_socket, "SWARM_CONTROLLER_SOCKET")
fn resolve(self, remote: &RemoteArgs) -> Result<Target> {
remote.target(self.controller_socket)
}
}
@ -521,7 +571,11 @@ struct UpdateArgs {
}
fn main() -> Result<()> {
let Cli { paths, command } = Cli::parse();
let Cli {
paths,
remote,
command,
} = Cli::parse();
match command {
// Resolves its own socket path, not `PathArgs`: this verb needs
// none of the `SWARMCTL_AUTHELIA_*` values, and requiring them
@ -530,35 +584,40 @@ fn main() -> Result<()> {
Verb::Agent {
command: AgentVerb::Create(args),
} => {
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
agent::create(&socket, &args.name, &args.hive)
let target = remote.target(args.controller_socket)?;
agent::create(&target, &args.name, &args.hive)
}
// Same socket-resolution reasoning as `Create` above.
Verb::Agent {
command: AgentVerb::MintIdentity(args),
} => {
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
agent::mint_identity(&socket, &args.name)
let target = remote.target(args.controller_socket)?;
agent::mint_identity(&target, &args.name)
}
// Same socket-resolution reasoning as `Create` above.
Verb::Agent {
command: AgentVerb::MintForgeToken(args),
} => {
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
agent::mint_forge_token(&socket, &args.name)
let target = remote.target(args.controller_socket)?;
agent::mint_forge_token(&target, &args.name)
}
Verb::Agent {
command: AgentVerb::Inference { command },
} => inference(command),
} => inference(&remote, command),
// Same socket-resolution reasoning as `Create` above.
Verb::Forge {
command: ForgeVerb::MakeAdmin(args),
} => {
let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?;
forge::make_admin(&socket, &args.name)
let target = remote.target(args.controller_socket)?;
forge::make_admin(&target, &args.name)
}
// Same socket-resolution reasoning as `Create` above.
Verb::Schedule { command } => schedule(command),
Verb::Schedule { command } => schedule(&remote, command),
Verb::Login => login::login(
&remote
.swarm()?
.context("`swarmctl login` needs `--swarm` or `SWARMCTL_SWARM`")?,
),
// Resolved lazily, inside the one arm that actually touches the
// deployment env vars — see the `MarkdownDocs` doc comment above
// for why an unconditional resolve up front would be wrong.
@ -598,11 +657,11 @@ fn main() -> Result<()> {
}
}
fn inference(command: InferenceVerb) -> Result<()> {
fn inference(remote: &RemoteArgs, command: InferenceVerb) -> Result<()> {
match command {
InferenceVerb::Get(a) => inference::get(&a.socket.resolve()?, &a.agent),
InferenceVerb::Get(a) => inference::get(&a.socket.resolve(remote)?, &a.agent),
InferenceVerb::Set(a) => inference::set(
&a.socket.resolve()?,
&a.socket.resolve(remote)?,
&a.agent,
inference::SetFlags {
model: a.model,
@ -610,14 +669,14 @@ fn inference(command: InferenceVerb) -> Result<()> {
subagent_models: a.subagent_models,
},
),
InferenceVerb::Clear(a) => inference::clear(&a.socket.resolve()?, &a.agent),
InferenceVerb::Clear(a) => inference::clear(&a.socket.resolve(remote)?, &a.agent),
}
}
fn schedule(command: ScheduleVerb) -> Result<()> {
fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> {
match command {
ScheduleVerb::Add(a) => schedule::add(
&a.socket.resolve()?,
&a.socket.resolve(remote)?,
&schedule::AddFlags {
name: &a.name,
agents: &a.agents,
@ -627,7 +686,7 @@ fn schedule(command: ScheduleVerb) -> Result<()> {
paused: a.paused,
},
),
ScheduleVerb::List(a) => schedule::list(&a.socket.resolve()?, a.agent.as_deref()),
ScheduleVerb::List(a) => schedule::list(&a.socket.resolve(remote)?, a.agent.as_deref()),
ScheduleVerb::Edit(a) => {
let edit = schedule::EditRequest::new(schedule::EditFlags {
name: a.name,
@ -638,11 +697,11 @@ fn schedule(command: ScheduleVerb) -> Result<()> {
add_agents: a.add_agents,
remove_agents: a.remove_agents,
})?;
schedule::edit(&a.socket.resolve()?, &a.id, &edit)
schedule::edit(&a.socket.resolve(remote)?, &a.id, &edit)
}
ScheduleVerb::Pause(a) => schedule::set_paused(&a.socket.resolve()?, &a.id, true),
ScheduleVerb::Resume(a) => schedule::set_paused(&a.socket.resolve()?, &a.id, false),
ScheduleVerb::Rm(a) => schedule::remove(&a.socket.resolve()?, &a.id),
ScheduleVerb::Pause(a) => schedule::set_paused(&a.socket.resolve(remote)?, &a.id, true),
ScheduleVerb::Resume(a) => schedule::set_paused(&a.socket.resolve(remote)?, &a.id, false),
ScheduleVerb::Rm(a) => schedule::remove(&a.socket.resolve(remote)?, &a.id),
}
}

352
swarmctl/src/remote.rs Normal file
View file

@ -0,0 +1,352 @@
//! Reaching swarm-controller through the swarm domain's gateway instead of its
//! unix socket, as the operator's own SSO account.
//!
//! The gateway's `/api/` location runs authelia's `auth_request` and accepts an
//! OAuth 2.0 bearer token in place of a session cookie. `swarmctl login`
//! ([`crate::login`]) obtains one for the public `swarmctl` client and this
//! module caches it, refreshes it and presents it.
//!
//! ⚠️ authelia **rotates** the refresh token on every use, and presenting an
//! already-used one revokes the whole login. Every refresh therefore runs under
//! an exclusive lock on the cache, and rereads the cache once it holds it.
use std::fs::{self, DirBuilder, File, OpenOptions};
use std::io::Write as _;
use std::os::unix::fs::{DirBuilderExt as _, OpenOptionsExt as _};
use std::path::{Path, PathBuf};
use anyhow::{Context as _, Result, anyhow, bail};
use serde::{Deserialize, Serialize};
/// Registered under the same id in `swarm-authelia.nix`.
pub(crate) const CLIENT_ID: &str = "swarmctl";
/// Refresh this long before the access token's stated expiry, so a token
/// doesn't lapse between the check and the gateway's look at it.
const EXPIRY_MARGIN_SECS: i64 = 60;
/// Where a controller request goes.
pub(crate) enum Target {
/// swarm-controller's unix socket on this host.
Socket(PathBuf),
/// The swarm domain's gateway, with the cached login.
Swarm(Swarm),
}
/// A swarm reached by its domain.
pub(crate) struct Swarm {
/// `https://<swarm domain>`, with no path.
pub origin: url::Url,
/// A PEM CA certificate to trust beside the system roots.
pub ca_file: Option<PathBuf>,
}
impl Swarm {
pub(crate) fn parse(url: &str, ca_file: Option<PathBuf>) -> Result<Self> {
let parsed =
url::Url::parse(url).with_context(|| format!("--swarm {url:?} is not a URL"))?;
if parsed.scheme() != "https" || parsed.host_str().is_none() {
bail!("--swarm {url:?} must be an https:// URL naming the swarm's domain");
}
let origin = url::Url::parse(&parsed.origin().ascii_serialization())
.context("re-parsing the swarm's origin")?;
Ok(Self { origin, ca_file })
}
/// What the token must name for the gateway to admit it at `/api/`.
/// authelia matches it as a URL prefix.
pub(crate) fn audience(&self) -> String {
format!("{}api/", self.origin)
}
/// An HTTP client that never follows a redirect: from the gateway one
/// means "log in", and following it would land on the login page.
pub(crate) fn http(&self) -> Result<reqwest::Client> {
let mut builder = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none());
if let Some(path) = &self.ca_file {
let pem = fs::read(path)
.with_context(|| format!("reading the CA certificate {}", path.display()))?;
let cert = reqwest::Certificate::from_pem(&pem)
.with_context(|| format!("parsing {} as a PEM certificate", path.display()))?;
builder = builder.add_root_certificate(cert);
}
builder.build().context("building the HTTPS client")
}
pub(crate) fn cache_path(&self) -> Result<PathBuf> {
let state = std::env::var_os("XDG_STATE_HOME")
.map(PathBuf::from)
.or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
.context(
"neither XDG_STATE_HOME nor HOME is set, so there is nowhere to keep the login",
)?;
let host = self.origin.host_str().unwrap_or_default();
let name = match self.origin.port() {
Some(port) => format!("{host}_{port}.json"),
None => format!("{host}.json"),
};
Ok(state.join("swarmctl").join(name))
}
fn not_logged_in(&self) -> String {
format!(
"not logged in to {} — run `swarmctl --swarm {} login`",
self.origin, self.origin
)
}
/// One request to `uri` (a path under the origin) with the cached login.
pub(crate) async fn request(
&self,
method: hyper::Method,
uri: &str,
body: Option<Vec<u8>>,
what: &str,
) -> Result<(hyper::StatusCode, bytes::Bytes)> {
let http = self.http()?;
let token = self.access_token(&http).await?;
let url = self.origin.join(uri).context("joining the request path")?;
let mut req = http.request(method, url).bearer_auth(token);
if let Some(body) = body {
req = req
.header(reqwest::header::CONTENT_TYPE, "application/json")
.body(body);
}
let resp = req
.send()
.await
.with_context(|| format!("sending the {what} request to {}", self.origin))?;
let status = resp.status();
if status == reqwest::StatusCode::UNAUTHORIZED {
bail!(
"{} refused the cached login (401). {}",
self.origin,
self.not_logged_in()
);
}
if status.is_redirection() {
bail!(
"{} answered {status} instead of the controller's response; its `/api/` \
did not accept a bearer token",
self.origin
);
}
let body = resp.bytes().await.context("reading the response")?;
Ok((status, body))
}
/// The cached access token, refreshed first if it is about to expire.
async fn access_token(&self, http: &reqwest::Client) -> Result<String> {
let path = self.cache_path()?;
let now = chrono::Utc::now().timestamp();
let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?;
if !needs_refresh(&cached, now) {
return Ok(cached.access_token);
}
let _lock = lock(&path)?;
let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?;
if !needs_refresh(&cached, now) {
return Ok(cached.access_token);
}
let token = token_request(
http,
&cached.auth,
&[
("grant_type", "refresh_token"),
("client_id", CLIENT_ID),
("refresh_token", &cached.refresh_token),
],
)
.await
.with_context(|| format!("refreshing the login. {}", self.not_logged_in()))?;
let fresh = refreshed(cached, token, now);
write_cache(&path, &fresh)?;
Ok(fresh.access_token)
}
}
/// The token cache: one file per swarm, mode 0600, in a 0700 directory.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Cached {
/// authelia's origin, where the refresh goes.
pub auth: String,
pub access_token: String,
pub refresh_token: String,
/// Unix seconds.
pub expires_at: i64,
}
/// authelia's token endpoint answer.
#[derive(Deserialize)]
pub(crate) struct TokenResponse {
pub access_token: String,
pub expires_in: i64,
pub refresh_token: Option<String>,
}
pub(crate) fn needs_refresh(cached: &Cached, now: i64) -> bool {
cached.expires_at - EXPIRY_MARGIN_SECS <= now
}
/// The cache after a refresh. The new refresh token replaces the old one,
/// which is dead from the moment it was used.
pub(crate) fn refreshed(cached: Cached, token: TokenResponse, now: i64) -> Cached {
Cached {
refresh_token: token.refresh_token.unwrap_or(cached.refresh_token),
access_token: token.access_token,
expires_at: now + token.expires_in,
auth: cached.auth,
}
}
/// POST a form to authelia's token endpoint.
pub(crate) async fn token_request(
http: &reqwest::Client,
auth: &str,
form: &[(&str, &str)],
) -> Result<TokenResponse> {
let resp = http
.post(format!("{auth}/api/oidc/token"))
.form(form)
.send()
.await
.with_context(|| format!("calling {auth}/api/oidc/token"))?;
oauth_json(resp).await
}
/// Decode an OAuth 2.0 JSON answer, or turn its `error` into the message.
pub(crate) async fn oauth_json<T: serde::de::DeserializeOwned>(
resp: reqwest::Response,
) -> Result<T> {
let status = resp.status();
let body = resp.bytes().await.context("reading authelia's response")?;
if !status.is_success() {
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
let field = |k| v.get(k).and_then(serde_json::Value::as_str).unwrap_or("");
return Err(anyhow!(
"authelia answered {status}: {} {}",
field("error"),
field("error_description")
));
}
serde_json::from_slice(&body).context("decoding authelia's response")
}
pub(crate) fn read_cache(path: &Path) -> Result<Option<Cached>> {
match fs::read(path) {
Ok(bytes) => serde_json::from_slice(&bytes)
.map(Some)
.with_context(|| format!("decoding the login cache {}", path.display())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e).with_context(|| format!("reading the login cache {}", path.display())),
}
}
/// Write through a 0600 temp file and rename, so the tokens are never
/// readable by anyone else, not even for the moment between create and chmod.
pub(crate) fn write_cache(path: &Path, cached: &Cached) -> Result<()> {
let dir = path
.parent()
.context("the login cache path has no directory")?;
DirBuilder::new()
.recursive(true)
.mode(0o700)
.create(dir)
.with_context(|| format!("creating {}", dir.display()))?;
let tmp = path.with_extension("json.tmp");
let _ = fs::remove_file(&tmp);
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&tmp)
.with_context(|| format!("creating {}", tmp.display()))?;
file.write_all(&serde_json::to_vec(cached)?)
.and_then(|()| file.sync_all())
.with_context(|| format!("writing {}", tmp.display()))?;
fs::rename(&tmp, path).with_context(|| format!("replacing {}", path.display()))
}
/// Hold an exclusive lock beside the cache until the returned file drops.
pub(crate) fn lock(path: &Path) -> Result<File> {
if let Some(dir) = path.parent() {
DirBuilder::new().recursive(true).mode(0o700).create(dir)?;
}
let lock_path = path.with_extension("lock");
let file = OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.mode(0o600)
.open(&lock_path)
.with_context(|| format!("opening {}", lock_path.display()))?;
rustix::fs::flock(&file, rustix::fs::FlockOperation::LockExclusive)
.with_context(|| format!("locking {}", lock_path.display()))?;
Ok(file)
}
#[cfg(test)]
mod tests {
use std::os::unix::fs::PermissionsExt as _;
use super::*;
fn cached(expires_at: i64) -> Cached {
Cached {
auth: "https://auth.example.test".into(),
access_token: "at-old".into(),
refresh_token: "rt-old".into(),
expires_at,
}
}
#[test]
fn a_token_is_refreshed_only_inside_the_margin() {
assert!(!needs_refresh(
&cached(1_000),
1_000 - EXPIRY_MARGIN_SECS - 1
));
assert!(needs_refresh(&cached(1_000), 1_000 - EXPIRY_MARGIN_SECS));
assert!(needs_refresh(&cached(1_000), 2_000));
}
/// authelia rotates the refresh token; keeping the used one would make
/// the next refresh revoke the whole login.
#[test]
fn a_refresh_replaces_both_tokens_and_the_expiry() {
let token = TokenResponse {
access_token: "at-new".into(),
expires_in: 3599,
refresh_token: Some("rt-new".into()),
};
let fresh = refreshed(cached(0), token, 5_000);
assert_eq!(fresh.access_token, "at-new");
assert_eq!(fresh.refresh_token, "rt-new");
assert_eq!(fresh.expires_at, 8_599);
assert_eq!(fresh.auth, "https://auth.example.test");
}
#[test]
fn the_cache_is_written_owner_only_and_reads_back() {
let dir = std::env::temp_dir().join(format!("swarmctl-cache-{}", std::process::id()));
let path = dir.join("swarmctl").join("swarm.example.test.json");
let _ = fs::remove_dir_all(&dir);
assert_eq!(read_cache(&path).expect("missing is not an error"), None);
write_cache(&path, &cached(42)).expect("writes");
write_cache(&path, &cached(43)).expect("overwrites");
let mode = |p: &Path| fs::metadata(p).expect("stat").permissions().mode() & 0o777;
assert_eq!(mode(&path), 0o600);
assert_eq!(mode(path.parent().expect("dir")), 0o700);
assert_eq!(read_cache(&path).expect("reads"), Some(cached(43)));
fs::remove_dir_all(&dir).expect("cleanup");
}
#[test]
fn the_audience_is_the_api_prefix_of_the_origin() {
let swarm = Swarm::parse("https://swarm.example.test/some/page", None).expect("parses");
assert_eq!(swarm.audience(), "https://swarm.example.test/api/");
assert!(Swarm::parse("http://swarm.example.test", None).is_err());
}
}

25
swarmctl/src/schedule.rs
View file

@ -8,12 +8,11 @@
//! The shapes below mirror the controller's own `ScheduleView` and request
//! bodies, for the reason [`crate::agent`]'s module doc gives.
use std::path::Path;
use anyhow::{Context as _, Result, bail};
use serde::{Deserialize, Serialize};
use crate::agent::{parse_ident, send};
use crate::remote::Target;
/// One schedule as the controller shows it.
#[derive(Debug, Deserialize)]
@ -142,7 +141,7 @@ fn agents(names: &[String]) -> Result<Vec<String>> {
fn runtime() -> Result<tokio::runtime::Runtime> {
tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_all()
.build()
.context("starting a tokio runtime for the controller request")
}
@ -167,7 +166,7 @@ pub(crate) struct AddFlags<'a> {
}
/// Run `swarmctl schedule add`.
pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> {
pub(crate) fn add(target: &Target, flags: &AddFlags<'_>) -> Result<()> {
let request = AddRequest {
name: flags.name,
agents: agents(flags.agents)?,
@ -177,7 +176,7 @@ pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> {
paused: flags.paused,
};
let added: ScheduleView = runtime()?.block_on(send(
socket,
target,
hyper::Method::POST,
COLLECTION,
Some(&request),
@ -189,10 +188,10 @@ pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> {
}
/// Run `swarmctl schedule list`, optionally only the schedules naming `agent`.
pub(crate) fn list(socket: &Path, agent: Option<&str>) -> Result<()> {
pub(crate) fn list(target: &Target, agent: Option<&str>) -> Result<()> {
let agent = agent.map(|a| parse_ident(a, "agent name")).transpose()?;
let schedules: Vec<ScheduleView> = runtime()?.block_on(send(
socket,
target,
hyper::Method::GET,
COLLECTION,
None::<&()>,
@ -212,9 +211,9 @@ pub(crate) fn list(socket: &Path, agent: Option<&str>) -> Result<()> {
}
/// Run `swarmctl schedule edit`.
pub(crate) fn edit(socket: &Path, id: &str, edit: &EditRequest) -> Result<()> {
pub(crate) fn edit(target: &Target, id: &str, edit: &EditRequest) -> Result<()> {
let edited: ScheduleView = runtime()?.block_on(send(
socket,
target,
hyper::Method::PATCH,
&member(id)?,
Some(edit),
@ -225,15 +224,15 @@ pub(crate) fn edit(socket: &Path, id: &str, edit: &EditRequest) -> Result<()> {
}
/// Run `swarmctl schedule pause` or `resume`.
pub(crate) fn set_paused(socket: &Path, id: &str, paused: bool) -> Result<()> {
edit(socket, id, &EditRequest::paused(paused))
pub(crate) fn set_paused(target: &Target, id: &str, paused: bool) -> Result<()> {
edit(target, id, &EditRequest::paused(paused))
}
/// Run `swarmctl schedule rm`.
pub(crate) fn remove(socket: &Path, id: &str) -> Result<()> {
pub(crate) fn remove(target: &Target, id: &str) -> Result<()> {
let uri = member(id)?;
runtime()?.block_on(send::<(), ()>(
socket,
target,
hyper::Method::DELETE,
&uri,
None,