diff --git a/CLAUDE.md b/CLAUDE.md index 6768cdf5..41104e5f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -195,9 +195,7 @@ hand-maintained per-file tree drifts out of sync with the code. verb that talks to a service rather than acting locally: it POSTs `/api/agents` to swarm-controller over the daemon's unix socket and prints the queued job id without waiting (the DAG settling would not - mean the agent is up). With `--swarm`, the controller verbs go - through the swarm domain's gateway instead, as the operator's own SSO - login (`swarmctl login`). Does not link `swarm-controller` — its wire + mean the agent is up). Does not link `swarm-controller` — its wire shapes are a local mirror kept in this crate, not a shared crate, so the two binaries can be deployed independently. ⚠️ Its user store is authelia's own **`users.yml`, read and diff --git a/Cargo.lock b/Cargo.lock index 6d7f6459..6b2e88f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4949,25 +4949,19 @@ name = "swarmctl" version = "0.1.0" dependencies = [ "anyhow", - "base64", "bytes", "chrono", "clap", "clap-markdown", "clap_complete", - "getrandom 0.3.4", "hive-types", "http-body-util", "hyper", "hyper-util", - "reqwest", - "rustix", "serde", "serde_json", "serde_norway", - "sha2 0.11.0", "tokio", - "url", ] [[package]] diff --git a/docs/swarm/sso.md b/docs/swarm/sso.md index 7e3107eb..47474846 100644 --- a/docs/swarm/sso.md +++ b/docs/swarm/sso.md @@ -281,17 +281,6 @@ All three matter. A change that silently deleted the browser page would pass a deny-only check, and one that quietly stopped denying would pass a page-only check. The person who introduced the split verified it this way. -## Operator CLI login - -`swarmctl login` logs an operator in as their own account, from any -machine, through the public `swarmctl` client authelia always registers: no -secret, authorization code with PKCE, a loopback redirect, and a refresh -token. Its access token carries `authelia.bearer.authz` with the audience -`https:///api/`, so the gateway admits it at the controller's API and -nowhere else, under the same access rule as a browser session there. -Revocation is per account: once authelia stops accepting the account, the -gateway refuses its token too. See [`swarmctl/README.md`](../../swarmctl/README.md#remote-use-swarmctl-login). - ## What this doesn't do - **It doesn't disable local login — except on Grafana.** Forgejo and diff --git a/docs/tools/swarmctl-cli.md b/docs/tools/swarmctl-cli.md index aad42920..ea9b3588 100644 --- a/docs/tools/swarmctl-cli.md +++ b/docs/tools/swarmctl-cli.md @@ -27,7 +27,6 @@ This document contains the help content for the `swarmctl` command-line program. * [`swarmctl user list`↴](#swarmctl-user-list) * [`swarmctl forge`↴](#swarmctl-forge) * [`swarmctl forge make-admin`↴](#swarmctl-forge-make-admin) -* [`swarmctl login`↴](#swarmctl-login) * [`swarmctl completions`↴](#swarmctl-completions) ## `swarmctl` @@ -42,7 +41,6 @@ swarm-level operator CLI * `schedule` — Manage scheduled prompts * `user` — Manage subjects in the swarm's SSO provider * `forge` — Manage human accounts on the swarm's forge -* `login` — Log in to the swarm named by `--swarm` as your own SSO account * `completions` — Generate a shell completion script for `swarmctl` and print it to stdout ###### **Options:** @@ -51,8 +49,6 @@ swarm-level operator CLI * `--users-file ` — Host-side path of authelia's users database — that is, the path inside the container, prefixed with the container's root. This is the only user store: it's read before every change and written in place, and `swarm-authelia-bridge` writes the same file. -* `--swarm ` — The swarm's domain, as `https://`. 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` -* `--ca-file ` — 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` @@ -79,7 +75,7 @@ Asks the swarm-controller to insert its agent-creation job graph — SSO identit **This returns as soon as it queues the work.** It doesn't wait, and a finished graph would not mean the agent is up either: the last node publishes a deploy, after which the hive converges on its own clock. Watch the swarm UI's job view, or the hive itself, for the rest. -No approval gate guards this: the caller is root on the controller's host, or logged in through the swarm domain's SSO gate. +No approval gate guards this: running this binary already means being root on the controller's host. **Usage:** `swarmctl agent create [OPTIONS] --hive ` @@ -469,16 +465,6 @@ The forge creates a human's account on their first SSO login, and this fails unt -## `swarmctl login` - -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. - -**Usage:** `swarmctl login` - - - ## `swarmctl completions` Generate a shell completion script for `swarmctl` and print it to stdout. diff --git a/nix/host-modules/swarm-authelia-service.nix b/nix/host-modules/swarm-authelia-service.nix index eb9b1006..ec2c0c41 100644 --- a/nix/host-modules/swarm-authelia-service.nix +++ b/nix/host-modules/swarm-authelia-service.nix @@ -171,7 +171,6 @@ in type = lib.types.enum [ "interactive" "machine" - "native" ]; default = "interactive"; example = "machine"; @@ -186,12 +185,6 @@ in client's own, so it receives an access token and no id-token. - `native` is a public client a person runs on their own - machine (RFC 8252): no secret, authorization code with - PKCE through a loopback `redirectUris` entry, and a refresh - token. Its access token carries `authelia.bearer.authz` - for its `audience`, so the gateway admits it as that person. - This is declared rather than inferred from an empty `redirectUris`, because authelia permits only the grants a client names — omitting `grant_types` yields diff --git a/nix/host-modules/swarm-authelia.nix b/nix/host-modules/swarm-authelia.nix index 2c465c9b..52324885 100644 --- a/nix/host-modules/swarm-authelia.nix +++ b/nix/host-modules/swarm-authelia.nix @@ -221,18 +221,6 @@ let redirectUris = [ ]; }; - # `swarmctl login`: an operator's own account, from wherever they run the - # CLI. The id is the one `swarmctl/src/remote.rs` sends. The audience is - # the controller's API on the swarm apex, which authelia matches as a URL - # prefix; the redirect matches a loopback callback on any port. - swarmctlClient = { - id = "swarmctl"; - description = "swarmctl (swarm operator CLI)"; - kind = "native"; - redirectUris = [ "http://127.0.0.1/callback" ]; - audience = [ "https://${swarmDomain}/api/" ]; - }; - # Derived from the client list rather than carrying its own `enable`; # `docs/swarm/sso.md` says why that is one fact instead of two. # @@ -302,19 +290,7 @@ let '' printf -- ' - client_id: %s\n' ${lib.escapeShellArg c.id} printf -- ' client_name: %s\n' ${lib.escapeShellArg c.description} - '' - # A native client is public: it has no secret, so none is minted for it. - + ( - if c.kind == "native" then - '' - printf -- ' public: true\n' - '' - else - '' - printf -- " client_secret: '%s'\n" "$(cat ${lib.escapeShellArg "${clientsDir}/${c.id}.digest"})" - '' - ) - + '' + printf -- " client_secret: '%s'\n" "$(cat ${lib.escapeShellArg "${clientsDir}/${c.id}.digest"})" printf -- ' authorization_policy: one_factor\n' '' # Read plainly, and that is a property of the list rather than of @@ -322,7 +298,7 @@ let # `oidc.clients`, so the module system has applied the submodule and # each option's default is present. A derived entry that names only # the fields it cares about still arrives with the rest filled in. - + lib.optionalString (c.kind != "native" && c.tokenEndpointAuthMethod != null) '' + + lib.optionalString (c.tokenEndpointAuthMethod != null) '' printf -- ' token_endpoint_auth_method: %s\n' ${lib.escapeShellArg c.tokenEndpointAuthMethod} '' # Flow-style YAML, matching `scopes` below. The values are client ids @@ -340,25 +316,6 @@ let printf -- ' grant_types: ["client_credentials"]\n' printf -- ' scopes: [${lib.optionalString c.bearerAuthz "authelia.bearer.authz"}]\n' '' - # What authelia 4.39.20 requires of a public client holding - # `authelia.bearer.authz`; its validator refuses the client without - # any one of these lines. - else if c.kind == "native" then - '' - printf -- ' token_endpoint_auth_method: none\n' - printf -- ' require_pkce: true\n' - printf -- ' pkce_challenge_method: S256\n' - printf -- ' require_pushed_authorization_requests: true\n' - printf -- ' consent_mode: explicit\n' - printf -- ' response_types: [code]\n' - printf -- ' response_modes: [form_post]\n' - printf -- ' grant_types: [authorization_code, refresh_token]\n' - printf -- ' scopes: [offline_access, authelia.bearer.authz]\n' - printf -- ' redirect_uris:\n' - ${lib.concatMapStrings (u: '' - printf -- ' - %s\n' ${lib.escapeShellArg u} - '') c.redirectUris} - '' else '' printf -- ' scopes: [openid, profile, email, groups]\n' @@ -425,7 +382,7 @@ let ${lib.concatMapStrings (c: '' mint ${lib.escapeShellArg c.id} - '') (lib.filter (c: c.kind != "native") cfg.oidc.clients)} + '') cfg.oidc.clients} # Re-rendered every boot, deliberately: the secret is minted once, # but the metadata around it (a new redirect URI, a renamed client) @@ -550,8 +507,7 @@ in # — see `oidcEnabled`'s comment above. services.hyperhive.swarm.authelia.oidc.clients = lib.optionals cfg.oidc.hiveIdentities (hiveClients ++ agentClients) - ++ [ bridgeClient ] - ++ lib.optional (swarmDomain != null) swarmctlClient; + ++ [ bridgeClient ]; # A redirect URI on a machine client is not harmless-but-unused: it # means whoever wrote it believes a browser is involved. Failing here @@ -626,17 +582,13 @@ in # Asserting here moves the same failure to evaluation, where a # wrong value costs a build and nothing else. { - assertion = lib.all ( - c: !(c.bearerAuthz || c.kind == "native") || c.audience != [ ] - ) cfg.oidc.clients; + assertion = lib.all (c: !c.bearerAuthz || c.audience != [ ]) cfg.oidc.clients; message = "services.hyperhive.swarm.authelia.oidc.clients: " + lib.concatStringsSep ", " ( - map (c: "client '${c.id}'") ( - lib.filter (c: (c.bearerAuthz || c.kind == "native") && c.audience == [ ]) cfg.oidc.clients - ) + map (c: "client '${c.id}'") (lib.filter (c: c.bearerAuthz && c.audience == [ ]) cfg.oidc.clients) ) - + " sets bearerAuthz or kind = \"native\" but declares no audience. The audience is " + + " sets bearerAuthz but declares no audience. The audience is " + "what authorises a bearer token at a given URL, so without " + "one the scope grants access to nothing and authelia refuses " + "the configuration outright."; diff --git a/nix/host-modules/swarm-ui.nix b/nix/host-modules/swarm-ui.nix index 7cba8a8d..c168e409 100644 --- a/nix/host-modules/swarm-ui.nix +++ b/nix/host-modules/swarm-ui.nix @@ -38,23 +38,18 @@ let inherit (stylixTheme) stylixThemeColors themedColorsCss; # Repeated verbatim by every location that should be operator-gated - # (`/`, `/api/docs/`; `/api/` takes `authRequest` with its own 401 - # handler) rather than set once on the server: + # (`/`, `/api/`, `/api/docs/`) rather than set once on the server: # nginx's `auth_request` does NOT inherit across sibling locations, so # setting it only on the page would leave this UI's own API and API # docs reachable without a session. - authRequest = '' + swarmAuthRequest = '' auth_request /__hive_authelia; # Captured BEFORE the error_page jump: inside the 401 handler # `$request_uri` is the internal one, so building the return # link there sends the operator back to the auth subrequest # instead of the page they asked for. auth_request_set $target_url $scheme://$http_host$request_uri; - ''; - loginRedirect = "https://${autheliaCfg.domain}/?rd=$target_url"; - swarmAuthRequest = '' - ${authRequest} - error_page 401 =302 ${loginRedirect}; + error_page 401 =302 https://${autheliaCfg.domain}/?rd=$target_url; ''; swarmCfg = config.services.hyperhive.swarm; hiveDomain = config.services.hyperhive.domain; @@ -223,24 +218,13 @@ in # `Remote-User` from authelia's answer, set here so it replaces # any the client sent: swarm-controller takes it as the sender of # the text the operator types to an agent (`agent_input.rs`). - # A refused caller that sent an `Authorization` header (`swarmctl - # --swarm`) gets the 401 itself; a browser still gets the login. - # The bearer token stops here and never reaches the controller. extraConfig = '' - ${authRequest} - error_page 401 = @swarm_api_unauthorized; + ${swarmAuthRequest} auth_request_set $authelia_user $upstream_http_remote_user; proxy_set_header Remote-User $authelia_user; - proxy_set_header Authorization ""; ''; }; - "@swarm_api_unauthorized".extraConfig = '' - if ($http_authorization) { - return 401; - } - return 302 ${loginRedirect}; - ''; - # ⚠️ THE ONE PROXIED LOCATION ON THIS VHOST WITH NO `auth_request`, + # ⚠️ THE ONE LOCATION ON THIS VHOST WITH NO `swarmAuthRequest`, # and that is deliberate rather than an omission. # # A forge webhook is a machine POST carrying an HMAC signature diff --git a/nix/module-eval/core-toggle.nix b/nix/module-eval/core-toggle.nix index d50ab957..e7de4502 100644 --- a/nix/module-eval/core-toggle.nix +++ b/nix/module-eval/core-toggle.nix @@ -234,52 +234,6 @@ let in v.forceSSL && !(v.addSSL or false); } - { - # `swarmctl --swarm` reads a 401 as "log in again"; a 302 would hand it - # the login page instead. The `/` clause is the control: the browser - # route still redirects, so a pass means the two routes differ. - name = "the controller API answers a refused bearer caller 401 and a browser the login"; - ok = - let - l = allLocal.services.nginx.virtualHosts."t.local".locations; - api = l."/api/".extraConfig; - refused = l."@swarm_api_unauthorized".extraConfig; - in - lib.hasInfix "error_page 401 = @swarm_api_unauthorized;" api - && !(lib.hasInfix "=302" api) - && lib.hasInfix ''proxy_set_header Authorization "";'' api - && lib.hasInfix "if ($http_authorization) {\n return 401;\n}" refused - && lib.hasInfix "return 302 https://auth.t.local/?rd=$target_url;" refused - && lib.hasInfix "error_page 401 =302 https://auth.t.local/" l."/".extraConfig; - } - { - # authelia 4.39.20 refuses a bearer-scoped public client missing any of - # these at startup, after a green deploy; and a public client must not - # be handed a secret. - name = "the swarmctl login client is public, bearer-scoped to the controller API, and gets no secret"; - ok = - let - s = allLocal.containers.swarm-authelia.config.systemd.services.authelia-swarm-secrets.script; - entry = lib.last (lib.splitString "client_id: %s\\n' swarmctl" s); - client = lib.head (lib.splitString "client_id: %s" entry); - in - lib.all (line: lib.hasInfix line client) [ - "public: true" - "token_endpoint_auth_method: none" - "require_pkce: true" - "pkce_challenge_method: S256" - "require_pushed_authorization_requests: true" - "consent_mode: explicit" - "response_modes: [form_post]" - "grant_types: [authorization_code, refresh_token]" - "scopes: [offline_access, authelia.bearer.authz]" - "audience: [%s]\\n' https://t.local/api/" - "' - %s\\n' http://127.0.0.1/callback" - ] - && !(lib.hasInfix "client_secret" client) - && !(lib.hasInfix "mint swarmctl" s) - && lib.hasInfix "mint swarm-authelia-bridge" s; - } { name = "a hive with matrix off serves no matrix discovery endpoint"; ok = diff --git a/swarmctl/Cargo.toml b/swarmctl/Cargo.toml index 3fb5ab0f..25098920 100644 --- a/swarmctl/Cargo.toml +++ b/swarmctl/Cargo.toml @@ -10,8 +10,6 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true -# PKCE verifier, state and S256 challenge for `swarmctl login`. -base64.workspace = true chrono.workspace = true # `agent create` — bare HTTP/1.1 client over swarm-controller's unix # socket. No new workspace dependency: `bytes`, `http-body-util`, @@ -20,22 +18,14 @@ bytes = "1" clap.workspace = true clap-markdown = "0.1" clap_complete.workspace = true -getrandom.workspace = true hive-types.workspace = true http-body-util.workspace = true -# `server`: the loopback listener `swarmctl login` receives its code on. -hyper = { workspace = true, features = ["server"] } +hyper.workspace = true hyper-util.workspace = true -# HTTPS to the swarm domain and its SSO provider, for `--swarm`. -reqwest.workspace = true -# `flock` on the login cache, which a refresh must not race. -rustix.workspace = true serde.workspace = true serde_json.workspace = true serde_norway = "0.9.42" -sha2.workspace = true tokio.workspace = true -url.workspace = true [lints] workspace = true diff --git a/swarmctl/README.md b/swarmctl/README.md index 88c2a547..5907ce03 100644 --- a/swarmctl/README.md +++ b/swarmctl/README.md @@ -1,9 +1,7 @@ # swarmctl Swarm-level operator CLI. Runs as **root on the host running -`swarm-controller`**, and acts on that host directly; or, for the verbs -that talk to the controller, from anywhere as your own SSO account (see -[Remote use](#remote-use-swarmctl-login)). +`swarm-controller`**, and acts on that host directly. Distinct from `hivectl`, which drives one hive's `hive-c0re` over its admin socket. This crate does not link `swarm-controller`, for the same @@ -21,11 +19,15 @@ allocated inside it, at activation — see the uid-assignment issue) or world-readable password hashes. Both are worse than root. So this binary **serves** no socket, publishes no HTTP route and has no -privileged helper. +privileged helper. When a verb 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. That rules out a way in, not a way out: `agent create` is a _client_ of -the controller, because the work it asks for is a job graph only the -controller can queue (see below). Nothing here becomes reachable by that. +the controller's socket, because the work it asks for is a job graph only +the controller can queue (see below). Nothing here becomes reachable by +that. ## One file, two writers @@ -51,8 +53,6 @@ an error. | `SWARMCTL_AUTHELIA_MACHINE` | container name, for `systemctl -M` | | `SWARMCTL_AUTHELIA_UNIT` | authelia's unit inside that container | | `SWARM_CONTROLLER_SOCKET` | the controller's unix socket, for `agent` and `forge` verbs | -| `SWARMCTL_SWARM` | `https://`: reach the controller there, not the socket | -| `SWARMCTL_CA_FILE` | a PEM CA to trust beside the system roots, for `SWARMCTL_SWARM` | Every verb and flag: [`docs/tools/swarmctl-cli.md`](../docs/tools/swarmctl-cli.md). The sections below cover why each verb behaves as it does. @@ -102,9 +102,8 @@ the swarm UI's job view for the rest. The new agent starts `paused`: it doesn't drive turns until you set it `up` in the swarm UI. -No approval gate, for the same reason nothing else here has one: the -caller is root on the controller's host, or logged in through the swarm -domain's SSO gate. +No approval gate, for the same reason nothing else here has one: running +this binary already means being root on the controller's host. `--hive` is required. It is an _address_ — where the deploy message goes — not an attribute of the agent, so there is no sensible default. The @@ -170,28 +169,3 @@ same socket as `agent create`; each verb waits for the stored result. What a schedule does once stored is in [`docs/swarm/schedules.md`](../docs/swarm/schedules.md). The shapes are mirrored in `src/schedule.rs`, for the same reason as `agent create`'s. - -## Remote use: `swarmctl login` - -The `agent`, `schedule` and `forge` verbs also work from any machine, as -your own SSO account, through the swarm domain's gateway: - -```console -$ export SWARMCTL_SWARM=https://swarm.example.com -$ swarmctl login -Log in to https://swarm.example.com/ in your browser: -… -logged in to https://swarm.example.com/; login kept in ~/.local/state/swarmctl/swarm.example.com.json -$ swarmctl schedule list -``` - -`login` opens the swarm's SSO login in a browser and receives the result on -a loopback port, so run it on the machine whose browser you use. swarmctl -keeps the login, mode 0600, under `$XDG_STATE_HOME/swarmctl/` (default -`~/.local/state`) and refreshes it as it's used, and once more if the gateway -refuses it. When it can't be refreshed, any verb says to run `swarmctl login` -again. To log out, delete the file. - -A swarm whose gateway serves a certificate from the swarm's own CA needs -that CA: `--ca-file` or `SWARMCTL_CA_FILE`. The `user` verbs edit the local -users file and have no remote form. diff --git a/swarmctl/src/agent.rs b/swarmctl/src/agent.rs index 315c34bf..700ac7e1 100644 --- a/swarmctl/src/agent.rs +++ b/swarmctl/src/agent.rs @@ -29,8 +29,6 @@ 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)] @@ -72,7 +70,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(target: &Target, name: &str, hive: &str) -> Result<()> { +pub(crate) fn create(socket: &Path, 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 @@ -81,10 +79,10 @@ pub(crate) fn create(target: &Target, name: &str, hive: &str) -> Result<()> { let hive = parse_ident(hive, "hive")?; let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request")?; - let resp = rt.block_on(post_create(target, &name, &hive))?; + let resp = rt.block_on(post_create(socket, &name, &hive))?; println!("queued: job node {}", resp.node_id); println!( @@ -112,17 +110,17 @@ pub(crate) fn parse_ident(value: &str, what: &str) -> Result { /// /// Synchronous for the same reason [`create`] is, and built on the same /// round trip. -pub(crate) fn mint_identity(target: &Target, name: &str) -> Result<()> { +pub(crate) fn mint_identity(socket: &Path, 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_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request")?; let resp: MintIdentityResponse = rt.block_on(post( - target, + socket, &format!("/api/agents/{name}/identity"), &serde_json::json!({}), "mint-identity", @@ -141,15 +139,15 @@ pub(crate) fn mint_identity(target: &Target, 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(target: &Target, name: &str) -> Result<()> { +pub(crate) fn mint_forge_token(socket: &Path, name: &str) -> Result<()> { let name = parse_ident(name, "agent name")?; let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request")?; let resp: MintForgeTokenResponse = rt.block_on(post( - target, + socket, &format!("/api/agents/{name}/forge-token"), &serde_json::json!({}), "mint-forge-token", @@ -164,10 +162,10 @@ pub(crate) fn mint_forge_token(target: &Target, name: &str) -> Result<()> { Ok(()) } -/// One `POST /api/agents` round trip at the controller. -async fn post_create(target: &Target, name: &str, hive: &str) -> Result { +/// One `POST /api/agents` round trip over the controller's unix socket. +async fn post_create(socket: &Path, name: &str, hive: &str) -> Result { post( - target, + socket, "/api/agents", &CreateAgentRequest { name, hive }, "create-agent", @@ -175,25 +173,25 @@ async fn post_create(target: &Target, name: &str, hive: &str) -> Result( - target: &Target, + socket: &Path, uri: &str, request: &Req, what: &str, ) -> Result { - send(target, hyper::Method::POST, uri, Some(request), what).await + send(socket, hyper::Method::POST, uri, Some(request), what).await } -/// 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 `()`. +/// 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 `()`. pub(crate) async fn send( - target: &Target, + socket: &Path, method: hyper::Method, uri: &str, request: Option<&Req>, @@ -204,34 +202,6 @@ pub(crate) async fn send( .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>, - what: &str, -) -> Result<(hyper::StatusCode, bytes::Bytes)> { let stream = UnixStream::connect(socket) .await .with_context(|| connect_hint(socket))?; @@ -267,7 +237,21 @@ async fn over_socket( .await .context("reading swarm-controller's response")? .to_bytes(); - Ok((status, body)) + + 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() + ) + }) } /// Turn a failed response into one line the operator can act on. diff --git a/swarmctl/src/forge.rs b/swarmctl/src/forge.rs index 564e7ef0..c7f3f1f6 100644 --- a/swarmctl/src/forge.rs +++ b/swarmctl/src/forge.rs @@ -6,11 +6,12 @@ //! 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)] @@ -28,15 +29,15 @@ enum AdminChange { /// Run `swarmctl forge make-admin`. /// /// Synchronous for the same reason `agent create` is. -pub(crate) fn make_admin(target: &Target, name: &str) -> Result<()> { +pub(crate) fn make_admin(socket: &Path, name: &str) -> Result<()> { let name = parse_ident(name, "user name")?; let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request")?; let resp: MakeForgeAdminResponse = rt.block_on(post( - target, + socket, &format!("/api/forge/users/{name}/admin"), &serde_json::json!({}), "forge-make-admin", diff --git a/swarmctl/src/inference.rs b/swarmctl/src/inference.rs index 840e1508..8e3f866b 100644 --- a/swarmctl/src/inference.rs +++ b/swarmctl/src/inference.rs @@ -9,11 +9,12 @@ //! 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)] @@ -65,7 +66,7 @@ impl SetFlags { fn runtime() -> Result { tokio::runtime::Builder::new_current_thread() - .enable_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request") } @@ -78,10 +79,10 @@ fn member(agent: &str) -> Result { } /// Run `swarmctl agent inference get`. -pub(crate) fn get(target: &Target, agent: &str) -> Result<()> { +pub(crate) fn get(socket: &Path, agent: &str) -> Result<()> { let uri = member(agent)?; let view: View = runtime()?.block_on(send( - target, + socket, hyper::Method::GET, &uri, None::<&()>, @@ -97,11 +98,11 @@ pub(crate) fn get(target: &Target, 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(target: &Target, agent: &str, flags: SetFlags) -> Result<()> { +pub(crate) fn set(socket: &Path, agent: &str, flags: SetFlags) -> Result<()> { let uri = member(agent)?; let rt = runtime()?; let current: View = rt.block_on(send( - target, + socket, hyper::Method::GET, &uri, None::<&()>, @@ -109,7 +110,7 @@ pub(crate) fn set(target: &Target, agent: &str, flags: SetFlags) -> Result<()> { ))?; let wanted = flags.merge(current.configured.unwrap_or_default())?; let stored: Settings = rt.block_on(send( - target, + socket, hyper::Method::PUT, &uri, Some(&wanted), @@ -120,10 +121,10 @@ pub(crate) fn set(target: &Target, agent: &str, flags: SetFlags) -> Result<()> { } /// Run `swarmctl agent inference clear`. -pub(crate) fn clear(target: &Target, agent: &str) -> Result<()> { +pub(crate) fn clear(socket: &Path, agent: &str) -> Result<()> { let uri = member(agent)?; runtime()?.block_on(send::<(), ()>( - target, + socket, hyper::Method::DELETE, &uri, None, diff --git a/swarmctl/src/login.rs b/swarmctl/src/login.rs deleted file mode 100644 index ac05e031..00000000 --- a/swarmctl/src/login.rs +++ /dev/null @@ -1,450 +0,0 @@ -//! `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; -use std::time::Duration; - -use anyhow::{Context as _, Result, bail}; -use base64::Engine as _; -use http_body_util::{BodyExt as _, Full, Limited}; -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"; -/// A `form_post` answer is a few hundred bytes. -const MAX_CALLBACK_BODY: usize = 16 * 1024; - -/// 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 { - 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 { - 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. Each -/// connection runs on its own task, so one that never sends a request can't -/// hold up the one carrying the answer; the first final answer wins. -async fn receive_code(listener: &TcpListener, state: &str, auth: &str) -> Result { - let expected = Arc::new((state.to_owned(), auth.to_owned())); - let (answers, mut answer) = tokio::sync::mpsc::unbounded_channel(); - loop { - tokio::select! { - accepted = listener.accept() => { - let (stream, _) = accepted.context("accepting the callback")?; - tokio::spawn(serve_callback(stream, Arc::clone(&expected), answers.clone())); - } - Some(outcome) = answer.recv() => return outcome.map_err(anyhow::Error::msg), - } - } -} - -/// Answer one loopback connection, sending a code or a refusal to `answers`. -async fn serve_callback( - stream: tokio::net::TcpStream, - expected: Arc<(String, String)>, - answers: tokio::sync::mpsc::UnboundedSender>, -) { - let service = hyper::service::service_fn(move |req: hyper::Request| { - let expected = Arc::clone(&expected); - let answers = answers.clone(); - async move { - let (parts, body) = req.into_parts(); - // Over the cap reads as an empty body, which carries no state. - let body = Limited::new(body, MAX_CALLBACK_BODY) - .collect() - .await - .map(http_body_util::Collected::to_bytes) - .unwrap_or_default(); - let (state, auth) = &*expected; - let (status, text) = match callback(&parts.method, parts.uri.path(), &body, state, auth) - { - Callback::Ignored(status, text) => (status, text), - Callback::Code(code) => { - let _ = answers.send(Ok(code)); - ( - hyper::StatusCode::OK, - "swarmctl is logged in. You can close this tab.", - ) - } - Callback::Refused(why) => { - let _ = answers.send(Err(why)); - ( - hyper::StatusCode::BAD_REQUEST, - "The login failed; swarmctl prints why.", - ) - } - }; - Ok::<_, std::convert::Infallible>( - 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; -} - -#[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 }); - - assert!( - post_to(addr, "code=stale&state=s0") - .await - .starts_with("HTTP/1.1 400") - ); - assert!( - post_to(addr, "code=fresh&state=s1") - .await - .starts_with("HTTP/1.1 200") - ); - assert_eq!(server.await.expect("join").expect("code"), "fresh"); - } - - /// One form POST to the loopback listener; the whole answer as text. - async fn post_to(addr: std::net::SocketAddr, body: &str) -> String { - 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 - } - - /// A connection that never sends a request (a browser's preconnect, any - /// local process) must not hold up the one carrying the answer. - #[tokio::test] - async fn an_idle_connection_does_not_block_the_real_callback() { - 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 _idle = tokio::net::TcpStream::connect(addr).await.expect("connect"); - tokio::time::sleep(Duration::from_millis(100)).await; - let answer = - tokio::time::timeout(Duration::from_secs(5), post_to(addr, "code=c1&state=s1")) - .await - .expect("the callback is answered while another connection idles"); - assert!(answer.starts_with("HTTP/1.1 200"), "{answer}"); - assert_eq!(server.await.expect("join").expect("code"), "c1"); - } - - /// A body past the cap is not read, so it can't complete the login; the - /// wait goes on for a real answer. - #[tokio::test] - async fn an_oversized_callback_body_is_refused_without_ending_the_wait() { - 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 pad = "a".repeat(MAX_CALLBACK_BODY); - let big = post_to(addr, &format!("code=big&state=s1&pad={pad}")).await; - assert!(big.starts_with("HTTP/1.1 400"), "{big}"); - assert!( - post_to(addr, "code=c1&state=s1") - .await - .starts_with("HTTP/1.1 200") - ); - assert_eq!(server.await.expect("join").expect("code"), "c1"); - } -} diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index 84edc540..0f9337e1 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -12,14 +12,16 @@ //! root. //! //! So this binary **serves** no socket, publishes no HTTP route and has -//! no privileged helper. +//! 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. //! -//! 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. +//! 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. //! //! Distinct from `hivectl`, which drives one hive's `hive-c0re` over its //! admin socket. This crate deliberately does not link `swarm-controller`, @@ -28,8 +30,6 @@ mod agent; mod forge; mod inference; -mod login; -mod remote; mod schedule; mod users; @@ -42,7 +42,6 @@ use std::process::Command; use anyhow::{Context, Result, bail}; use clap::{Args, Parser, Subcommand}; -use remote::Target; use users::{User, UserStore}; #[derive(Parser)] @@ -50,52 +49,10 @@ 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://`. 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, - /// 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, -} - -impl RemoteArgs { - fn swarm(&self) -> Result> { - 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, remote::state_dir()?)) - .transpose() - } - - /// The swarm when one is named, the controller's socket otherwise. - fn target(&self, socket: Option) -> Result { - 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 @@ -184,14 +141,6 @@ 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 @@ -234,8 +183,8 @@ enum AgentVerb { /// own clock. Watch the swarm UI's job view, or the hive itself, for /// the rest. /// - /// No approval gate guards this: the caller is root on the controller's - /// host, or logged in through the swarm domain's SSO gate. + /// No approval gate guards this: running this binary already means + /// being root on the controller's host. Create(AgentCreateArgs), /// Queue a re-mint of an existing agent's identity at the swarm's /// secret store. @@ -497,8 +446,8 @@ struct ControllerSocketArg { } impl ControllerSocketArg { - fn resolve(self, remote: &RemoteArgs) -> Result { - remote.target(self.controller_socket) + fn resolve(self) -> Result { + path_from(self.controller_socket, "SWARM_CONTROLLER_SOCKET") } } @@ -572,11 +521,7 @@ struct UpdateArgs { } fn main() -> Result<()> { - let Cli { - paths, - remote, - command, - } = Cli::parse(); + let Cli { paths, command } = Cli::parse(); match command { // Resolves its own socket path, not `PathArgs`: this verb needs // none of the `SWARMCTL_AUTHELIA_*` values, and requiring them @@ -585,40 +530,35 @@ fn main() -> Result<()> { Verb::Agent { command: AgentVerb::Create(args), } => { - let target = remote.target(args.controller_socket)?; - agent::create(&target, &args.name, &args.hive) + let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?; + agent::create(&socket, &args.name, &args.hive) } // Same socket-resolution reasoning as `Create` above. Verb::Agent { command: AgentVerb::MintIdentity(args), } => { - let target = remote.target(args.controller_socket)?; - agent::mint_identity(&target, &args.name) + let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?; + agent::mint_identity(&socket, &args.name) } // Same socket-resolution reasoning as `Create` above. Verb::Agent { command: AgentVerb::MintForgeToken(args), } => { - let target = remote.target(args.controller_socket)?; - agent::mint_forge_token(&target, &args.name) + let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?; + agent::mint_forge_token(&socket, &args.name) } Verb::Agent { command: AgentVerb::Inference { command }, - } => inference(&remote, command), + } => inference(command), // Same socket-resolution reasoning as `Create` above. Verb::Forge { command: ForgeVerb::MakeAdmin(args), } => { - let target = remote.target(args.controller_socket)?; - forge::make_admin(&target, &args.name) + let socket = path_from(args.controller_socket, "SWARM_CONTROLLER_SOCKET")?; + forge::make_admin(&socket, &args.name) } // Same socket-resolution reasoning as `Create` above. - Verb::Schedule { command } => schedule(&remote, command), - Verb::Login => login::login( - &remote - .swarm()? - .context("`swarmctl login` needs `--swarm` or `SWARMCTL_SWARM`")?, - ), + Verb::Schedule { command } => schedule(command), // 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. @@ -658,11 +598,11 @@ fn main() -> Result<()> { } } -fn inference(remote: &RemoteArgs, command: InferenceVerb) -> Result<()> { +fn inference(command: InferenceVerb) -> Result<()> { match command { - InferenceVerb::Get(a) => inference::get(&a.socket.resolve(remote)?, &a.agent), + InferenceVerb::Get(a) => inference::get(&a.socket.resolve()?, &a.agent), InferenceVerb::Set(a) => inference::set( - &a.socket.resolve(remote)?, + &a.socket.resolve()?, &a.agent, inference::SetFlags { model: a.model, @@ -670,14 +610,14 @@ fn inference(remote: &RemoteArgs, command: InferenceVerb) -> Result<()> { subagent_models: a.subagent_models, }, ), - InferenceVerb::Clear(a) => inference::clear(&a.socket.resolve(remote)?, &a.agent), + InferenceVerb::Clear(a) => inference::clear(&a.socket.resolve()?, &a.agent), } } -fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> { +fn schedule(command: ScheduleVerb) -> Result<()> { match command { ScheduleVerb::Add(a) => schedule::add( - &a.socket.resolve(remote)?, + &a.socket.resolve()?, &schedule::AddFlags { name: &a.name, agents: &a.agents, @@ -687,7 +627,7 @@ fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> { paused: a.paused, }, ), - ScheduleVerb::List(a) => schedule::list(&a.socket.resolve(remote)?, a.agent.as_deref()), + ScheduleVerb::List(a) => schedule::list(&a.socket.resolve()?, a.agent.as_deref()), ScheduleVerb::Edit(a) => { let edit = schedule::EditRequest::new(schedule::EditFlags { name: a.name, @@ -698,11 +638,11 @@ fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> { add_agents: a.add_agents, remove_agents: a.remove_agents, })?; - schedule::edit(&a.socket.resolve(remote)?, &a.id, &edit) + schedule::edit(&a.socket.resolve()?, &a.id, &edit) } - 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), + 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), } } @@ -992,75 +932,6 @@ fn write_atomic(path: &Path, contents: &str) -> Result<()> { mod tests { use super::*; - /// Root on the controller host, with no `--swarm` and no login: the verbs - /// still go over the unix socket, unchanged by the remote path. - #[test] - fn with_no_swarm_named_the_verbs_use_the_controller_socket() { - assert!( - std::env::var_os("SWARMCTL_SWARM").is_none(), - "this test needs SWARMCTL_SWARM unset" - ); - let dir = std::env::temp_dir().join(format!("swarmctl-socket-{}", std::process::id())); - let _ = fs::remove_dir_all(&dir); - fs::create_dir_all(&dir).expect("dir"); - let socket = dir.join("controller.sock"); - let listener = std::os::unix::net::UnixListener::bind(&socket).expect("bind"); - listener.set_nonblocking(true).expect("nonblocking"); - let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); - let server_seen = std::sync::Arc::clone(&seen); - std::thread::spawn(move || { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("runtime"); - rt.block_on(async move { - let listener = tokio::net::UnixListener::from_std(listener).expect("listener"); - let (stream, _) = listener.accept().await.expect("accept"); - let service = hyper::service::service_fn( - move |req: hyper::Request| { - server_seen.lock().expect("seen").push(format!( - "{} {} auth={}", - req.method(), - req.uri(), - req.headers().contains_key(hyper::header::AUTHORIZATION) - )); - async { - Ok::<_, std::convert::Infallible>(hyper::Response::new( - http_body_util::Full::new(bytes::Bytes::from_static(b"[]")), - )) - } - }, - ); - let _ = hyper::server::conn::http1::Builder::new() - .serve_connection(hyper_util::rt::TokioIo::new(stream), service) - .await; - }); - }); - - let cli = Cli::try_parse_from([ - "swarmctl", - "schedule", - "list", - "--controller-socket", - socket.to_str().expect("utf-8"), - ]) - .expect("parses"); - let Verb::Schedule { - command: ScheduleVerb::List(args), - } = cli.command - else { - panic!("parsed as another verb"); - }; - let target = args.socket.resolve(&cli.remote).expect("resolves"); - assert!(matches!(&target, Target::Socket(p) if *p == socket)); - schedule::list(&target, None).expect("lists over the socket"); - assert_eq!( - *seen.lock().expect("seen"), - ["GET /api/schedules auth=false"] - ); - fs::remove_dir_all(&dir).expect("cleanup"); - } - /// clap's own consistency checks over the whole tree — a duplicated /// long flag or a malformed `value_name` panics at parse time in /// production and is otherwise only found by running the binary. diff --git a/swarmctl/src/remote.rs b/swarmctl/src/remote.rs deleted file mode 100644 index 44a31614..00000000 --- a/swarmctl/src/remote.rs +++ /dev/null @@ -1,607 +0,0 @@ -//! 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://`, with no path. - pub origin: url::Url, - /// A PEM CA certificate to trust beside the system roots. - pub ca_file: Option, - /// Where the login cache lives: `$XDG_STATE_HOME/swarmctl` by default. - pub state_dir: PathBuf, -} - -impl Swarm { - pub(crate) fn parse(url: &str, ca_file: Option, state_dir: PathBuf) -> Result { - 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, - state_dir, - }) - } - - /// 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 { - 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) -> PathBuf { - 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"), - }; - self.state_dir.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. - /// A 401 is answered by one refresh and one retry; a second 401 means the - /// login itself is gone. - pub(crate) async fn request( - &self, - method: hyper::Method, - uri: &str, - body: Option>, - what: &str, - ) -> Result<(hyper::StatusCode, bytes::Bytes)> { - self.request_with(&self.http()?, method, uri, body, what) - .await - } - - async fn request_with( - &self, - http: &reqwest::Client, - method: hyper::Method, - uri: &str, - body: Option>, - what: &str, - ) -> Result<(hyper::StatusCode, bytes::Bytes)> { - let url = self.origin.join(uri).context("joining the request path")?; - let send = |token: String| { - let mut req = http.request(method.clone(), url.clone()).bearer_auth(token); - if let Some(body) = &body { - req = req - .header(reqwest::header::CONTENT_TYPE, "application/json") - .body(body.clone()); - } - req.send() - }; - let context = || format!("sending the {what} request to {}", self.origin); - let token = self.access_token(http, None).await?; - let mut resp = send(token.clone()).await.with_context(context)?; - if resp.status() == reqwest::StatusCode::UNAUTHORIZED { - let fresh = self.access_token(http, Some(&token)).await?; - resp = send(fresh).await.with_context(context)?; - } - 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 or is - /// `rejected`, the one the gateway just refused. - async fn access_token(&self, http: &reqwest::Client, rejected: Option<&str>) -> Result { - let path = self.cache_path(); - let now = chrono::Utc::now().timestamp(); - let due = |c: &Cached| needs_refresh(c, now) || rejected == Some(c.access_token.as_str()); - let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?; - if !due(&cached) { - return Ok(cached.access_token); - } - let _lock = lock(&path)?; - // Another process may have refreshed while this one waited for the lock. - let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?; - if !due(&cached) { - 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) - } -} - -/// `$XDG_STATE_HOME/swarmctl`, else `~/.local/state/swarmctl`. -pub(crate) fn state_dir() -> Result { - 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"))) - .map(|state| state.join("swarmctl")) - .context("neither XDG_STATE_HOME nor HOME is set, so there is nowhere to keep the login") -} - -/// 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, -} - -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 { - 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( - resp: reqwest::Response, -) -> Result { - 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> { - 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 { - 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 std::sync::{Arc, Mutex}; - - 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, "/s".into()) - .expect("parses"); - assert_eq!(swarm.audience(), "https://swarm.example.test/api/"); - assert!(Swarm::parse("http://swarm.example.test", None, "/s".into()).is_err()); - } - - /// authelia's token endpoint and the gateway's `/api/`, as far as these - /// tests need them: a refresh rotates both tokens, a used refresh token - /// revokes the login (as authelia does), and `/api/` admits only the - /// current access token. - struct Mock { - refresh: Option, - access: Option, - refreshes: u32, - api_calls: u32, - api_refuses_all: bool, - } - - fn mock() -> Arc> { - Arc::new(Mutex::new(Mock { - refresh: Some("rt-0".into()), - access: Some("at-0".into()), - refreshes: 0, - api_calls: 0, - api_refuses_all: false, - })) - } - - /// Serve `mock` on a loopback port from its own thread, so callers on - /// other threads can block on the cache lock without stalling it. - fn serve(mock: &Arc>) -> url::Url { - let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); - let addr = listener.local_addr().expect("addr"); - listener.set_nonblocking(true).expect("nonblocking"); - let mock = Arc::clone(mock); - std::thread::spawn(move || { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("runtime"); - rt.block_on(async move { - let listener = tokio::net::TcpListener::from_std(listener).expect("listener"); - loop { - let (stream, _) = listener.accept().await.expect("accept"); - let mock = Arc::clone(&mock); - tokio::spawn(async move { - let service = hyper::service::service_fn(move |req| { - let mock = Arc::clone(&mock); - async move { Ok::<_, std::convert::Infallible>(answer(&mock, req).await) } - }); - let _ = hyper::server::conn::http1::Builder::new() - .serve_connection(hyper_util::rt::TokioIo::new(stream), service) - .await; - }); - } - }); - }); - url::Url::parse(&format!("http://{addr}/")).expect("url") - } - - async fn answer( - mock: &Mutex, - req: hyper::Request, - ) -> hyper::Response> { - use http_body_util::BodyExt as _; - let reply = |status: u16, body: serde_json::Value| { - hyper::Response::builder() - .status(status) - .body(http_body_util::Full::new(body.to_string().into())) - .expect("response") - }; - let path = req.uri().path().to_owned(); - let bearer = req - .headers() - .get(hyper::header::AUTHORIZATION) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.strip_prefix("Bearer ")) - .map(str::to_owned); - let body = req.into_body().collect().await.expect("body").to_bytes(); - if path == "/api/oidc/token" { - // Wide enough that two unlocked refreshers are both in flight. - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - let presented = url::form_urlencoded::parse(&body) - .find(|(k, _)| k == "refresh_token") - .map(|(_, v)| v.into_owned()); - let mut m = mock.lock().expect("mock"); - if presented.is_none() || presented != m.refresh { - m.refresh = None; - m.access = None; - return reply(400, serde_json::json!({ "error": "invalid_grant" })); - } - m.refreshes += 1; - m.access = Some(format!("at-{}", m.refreshes)); - m.refresh = Some(format!("rt-{}", m.refreshes)); - return reply( - 200, - serde_json::json!({ - "access_token": m.access, "refresh_token": m.refresh, "expires_in": 3600, - }), - ); - } - let mut m = mock.lock().expect("mock"); - m.api_calls += 1; - if !m.api_refuses_all && bearer.is_some() && bearer == m.access { - reply(200, serde_json::json!({ "ok": true })) - } else { - reply(401, serde_json::json!({})) - } - } - - fn scratch(name: &str) -> PathBuf { - let dir = std::env::temp_dir().join(format!("swarmctl-{name}-{}", std::process::id())); - let _ = fs::remove_dir_all(&dir); - dir - } - - /// A swarm whose API and authelia are both `origin`, with a cached login. - fn logged_in(origin: &url::Url, dir: &Path, access_token: &str, expires_at: i64) -> Swarm { - let swarm = Swarm { - origin: origin.clone(), - ca_file: None, - state_dir: dir.to_owned(), - }; - let cache = Cached { - auth: origin.as_str().trim_end_matches('/').to_owned(), - access_token: access_token.into(), - refresh_token: "rt-0".into(), - expires_at, - }; - write_cache(&swarm.cache_path(), &cache).expect("cache"); - swarm - } - - /// Plain HTTP only: no system roots, which the build sandbox doesn't have. - fn client() -> reqwest::Client { - reqwest::Client::builder() - .tls_certs_only([]) - .build() - .expect("client") - } - - fn block_on(fut: impl std::future::Future) -> T { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("runtime") - .block_on(fut) - } - - /// Two refreshers on one expired cache, as two `swarmctl` processes would - /// be: one refresh, and both get the token it issued. Without the lock - /// and the reread under it, the second presents the used refresh token - /// and authelia revokes the login. - #[test] - fn concurrent_refreshers_refresh_once_and_share_the_token() { - let mock = mock(); - let origin = serve(&mock); - let dir = scratch("concurrent"); - logged_in(&origin, &dir, "at-0", 0); - - let start = Arc::new(std::sync::Barrier::new(2)); - let refreshers: Vec<_> = (0..2) - .map(|_| { - let swarm = Swarm { - origin: origin.clone(), - ca_file: None, - state_dir: dir.clone(), - }; - let start = Arc::clone(&start); - std::thread::spawn(move || { - let http = client(); - start.wait(); - block_on(swarm.access_token(&http, None)) - }) - }) - .collect(); - let tokens: Vec = refreshers - .into_iter() - .map(|r| r.join().expect("thread").expect("refreshed")) - .collect(); - - let m = mock.lock().expect("mock"); - assert_eq!(m.refreshes, 1); - assert_eq!(tokens, ["at-1", "at-1"]); - assert_eq!(m.access.as_deref(), Some("at-1"), "the login is still live"); - fs::remove_dir_all(&dir).expect("cleanup"); - } - - /// The clock says the token is good and the gateway refuses it: refresh - /// once and retry. - #[test] - fn a_refused_token_is_refreshed_once_and_the_request_retried() { - let mock = mock(); - let origin = serve(&mock); - let dir = scratch("retry"); - let swarm = logged_in(&origin, &dir, "at-revoked", i64::MAX / 2); - - let (status, _) = - block_on(swarm.request_with(&client(), hyper::Method::GET, "/api/x", None, "x")) - .expect("the retry succeeds"); - assert_eq!(status, hyper::StatusCode::OK); - let m = mock.lock().expect("mock"); - assert_eq!((m.refreshes, m.api_calls), (1, 2)); - drop(m); - let cached = read_cache(&swarm.cache_path()) - .expect("read") - .expect("cache"); - assert_eq!(cached.access_token, "at-1"); - fs::remove_dir_all(&dir).expect("cleanup"); - } - - /// A 401 after the refresh means the login itself is gone: say so, once. - #[test] - fn a_second_401_says_to_log_in_and_does_not_retry_again() { - let mock = mock(); - mock.lock().expect("mock").api_refuses_all = true; - let origin = serve(&mock); - let dir = scratch("refused"); - let swarm = logged_in(&origin, &dir, "at-0", i64::MAX / 2); - - let err = block_on(swarm.request_with(&client(), hyper::Method::GET, "/api/x", None, "x")) - .expect_err("still refused") - .to_string(); - assert!(err.contains("(401)") && err.contains("login`"), "{err}"); - let m = mock.lock().expect("mock"); - assert_eq!((m.refreshes, m.api_calls), (1, 2)); - fs::remove_dir_all(&dir).expect("cleanup"); - } -} diff --git a/swarmctl/src/schedule.rs b/swarmctl/src/schedule.rs index 83adaa28..b7becc07 100644 --- a/swarmctl/src/schedule.rs +++ b/swarmctl/src/schedule.rs @@ -8,11 +8,12 @@ //! 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)] @@ -141,7 +142,7 @@ fn agents(names: &[String]) -> Result> { fn runtime() -> Result { tokio::runtime::Builder::new_current_thread() - .enable_all() + .enable_io() .build() .context("starting a tokio runtime for the controller request") } @@ -166,7 +167,7 @@ pub(crate) struct AddFlags<'a> { } /// Run `swarmctl schedule add`. -pub(crate) fn add(target: &Target, flags: &AddFlags<'_>) -> Result<()> { +pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> { let request = AddRequest { name: flags.name, agents: agents(flags.agents)?, @@ -176,7 +177,7 @@ pub(crate) fn add(target: &Target, flags: &AddFlags<'_>) -> Result<()> { paused: flags.paused, }; let added: ScheduleView = runtime()?.block_on(send( - target, + socket, hyper::Method::POST, COLLECTION, Some(&request), @@ -188,10 +189,10 @@ pub(crate) fn add(target: &Target, flags: &AddFlags<'_>) -> Result<()> { } /// Run `swarmctl schedule list`, optionally only the schedules naming `agent`. -pub(crate) fn list(target: &Target, agent: Option<&str>) -> Result<()> { +pub(crate) fn list(socket: &Path, agent: Option<&str>) -> Result<()> { let agent = agent.map(|a| parse_ident(a, "agent name")).transpose()?; let schedules: Vec = runtime()?.block_on(send( - target, + socket, hyper::Method::GET, COLLECTION, None::<&()>, @@ -211,9 +212,9 @@ pub(crate) fn list(target: &Target, agent: Option<&str>) -> Result<()> { } /// Run `swarmctl schedule edit`. -pub(crate) fn edit(target: &Target, id: &str, edit: &EditRequest) -> Result<()> { +pub(crate) fn edit(socket: &Path, id: &str, edit: &EditRequest) -> Result<()> { let edited: ScheduleView = runtime()?.block_on(send( - target, + socket, hyper::Method::PATCH, &member(id)?, Some(edit), @@ -224,15 +225,15 @@ pub(crate) fn edit(target: &Target, id: &str, edit: &EditRequest) -> Result<()> } /// Run `swarmctl schedule pause` or `resume`. -pub(crate) fn set_paused(target: &Target, id: &str, paused: bool) -> Result<()> { - edit(target, id, &EditRequest::paused(paused)) +pub(crate) fn set_paused(socket: &Path, id: &str, paused: bool) -> Result<()> { + edit(socket, id, &EditRequest::paused(paused)) } /// Run `swarmctl schedule rm`. -pub(crate) fn remove(target: &Target, id: &str) -> Result<()> { +pub(crate) fn remove(socket: &Path, id: &str) -> Result<()> { let uri = member(id)?; runtime()?.block_on(send::<(), ()>( - target, + socket, hyper::Method::DELETE, &uri, None,