From e142ebef75ff2fa6bc42098d75a75b1d8337c69d Mon Sep 17 00:00:00 2001
From: Arian van Putten <arian@arianvp.me>
Date: Mon, 4 May 2026 14:31:39 +0200
Subject: [PATCH] nixos/spire: make Workload API socket reachable by non-root
 workloads
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The unit's UMask=0027 was masking spire-agent's own
os.MkdirAll("/run/spire/agent/public", 0755) down to mode 0750, so any
process not in the spire-agent group got EACCES on connect() — defeating
the point of workload attestation, which is supposed to identify
arbitrary callers regardless of their unix identity.

Pre-create the directory via RuntimeDirectory so systemd applies
RuntimeDirectoryMode (0755) independent of umask; spire-agent's MkdirAll
then becomes a no-op.

Also exercises the path in the NixOS test by fetching an SVID as a
normal user.
---
 .../modules/services/security/spire/agent.nix |  2 +-
 nixos/tests/spire.nix                         | 29 +++++++++++++++----
 2 files changed, 25 insertions(+), 6 deletions(-)

diff --git a/nixos/modules/services/security/spire/agent.nix b/nixos/modules/services/security/spire/agent.nix
index dc5cfa7a992e..5a5d1112b448 100644
--- a/nixos/modules/services/security/spire/agent.nix
+++ b/nixos/modules/services/security/spire/agent.nix
@@ -150,7 +150,7 @@ in
         Restart = "on-failure";
         StateDirectory = "spire/agent";
         StateDirectoryMode = "0700";
-        RuntimeDirectory = "spire/agent";
+        RuntimeDirectory = "spire/agent/public";
 
         # TODO: Switch to DynamicUser once https://github.com/NixOS/nixpkgs/issues/299476 lands
         # Without it, the systemd plugin can not talk to dbus
diff --git a/nixos/tests/spire.nix b/nixos/tests/spire.nix
index dfdeba26e635..dec1d0748df1 100644
--- a/nixos/tests/spire.nix
+++ b/nixos/tests/spire.nix
@@ -45,6 +45,14 @@ let
         config.services.spire.agent.settings.agent.socket_path;
       virtualisation.credentials."spire.trust_bundle".source = "./trust_bundle";
       systemd.services.spire-agent.serviceConfig.ImportCredential = [ "spire.trust_bundle" ];
+
+      # A non-root, non-spire-agent user used to verify that arbitrary
+      # workloads can reach the Workload API socket.
+      users.users.workload = {
+        isNormalUser = true;
+        group = "workload";
+      };
+      users.groups.workload = { };
     };
 in
 {
@@ -180,25 +188,36 @@ in
         join_token = server.succeed("spire-server token generate -socketPath ${adminSocket}").split()[1]
         with open(agent.state_dir / "join_token", "w") as f:
           f.write(join_token)
-        parent_id = f"spiffe://${trustDomain}/spire/agent/join_token/{join_token}"
-        server.succeed(f"spire-server entry create -socketPath ${adminSocket} -selector 'systemd:id:backdoor.service' -parentID {parent_id} -spiffeID {spiffe_id}")
+        return f"spiffe://${trustDomain}/spire/agent/join_token/{join_token}"
 
 
       def provision_tpm(agent):
         agent.wait_for_unit("tpm2.target")
         ek_hash = agent.succeed("get_tpm_pubhash").strip()
-        parent_id = f"spiffe://${trustDomain}/spire/agent/tpm/{ek_hash}"
-        server.succeed(f"spire-server entry create -socketPath ${adminSocket} -selector 'systemd:id:backdoor.service' -parentID '{parent_id}' -spiffeID {spiffe_id}")
+        return f"spiffe://${trustDomain}/spire/agent/tpm/{ek_hash}"
+
+
+      def register_entry(parent_id, selector, spiffe_id):
+        server.succeed(f"spire-server entry create -socketPath ${adminSocket} -selector '{selector}' -parentID '{parent_id}' -spiffeID '{spiffe_id}'")
 
 
       def test_agent(name, agent_node, provision_fn):
         with subtest(f"Setup SPIRE agent with {name} attestation"):
           provision_trust_bundle(agent_node)
-          provision_fn(agent_node)
+          parent_id = provision_fn(agent_node)
+          register_entry(parent_id, "systemd:id:backdoor.service", spiffe_id)
+          register_entry(parent_id, "unix:user:workload", "spiffe://${trustDomain}/workload")
           agent_node.wait_for_unit("spire-agent.service")
           agent_node.wait_until_succeeds("spire-agent healthcheck -socketPath $SPIFFE_ENDPOINT_SOCKET", timeout=90)
         with subtest(f"Test certificate authentication from {name} agent"):
           agent_node.wait_until_succeeds("spire-agent api fetch x509 -socketPath $SPIFFE_ENDPOINT_SOCKET -write .")
+        with subtest(f"Workload running as non-root, non-spire-agent user can reach Workload API ({name})"):
+          # SPIRE's security model relies on workload attestation, not unix
+          # permissions on the socket. Verify an unrelated user can connect.
+          agent_node.wait_until_succeeds(
+            "su -s /bin/sh workload -c "
+            "'spire-agent api fetch x509 -socketPath \"$SPIFFE_ENDPOINT_SOCKET\"'"
+          )
         # TODO: Add something to communicate with