fix(P04): wire ACL enforcement + WebAuthn reg auth + audit actor (REQ-153)

R-023: Zero-trust enforcement operationally wired.

ACL enforcement (C-45 staged rollout):
- acl.Check wired into all 5 daemon handlers (dispatch/jobs/nodes/tasks)
- health endpoints exempt (liveness probes not gated)
- ACL log-only mode default (config acl.enforce=false); enforce after
  bootstrap ACL verified
- sshpush auth: ORCA_OIDC_TOKEN validated against JWKS before apply
- txn apply: Authorize hook validates OIDC token before running pull
- acl.json mode 0600 (was 0644)
- flock on acl.json for concurrent grant/revoke
- bootstrap ACL: init grants cluster-admin to orca-admins group + SVID

Audit actor identity:
- currentActor reads OIDC sub from credentials.json (was hardcoded "cli")
- threaded through all audit.Record calls via context

WebAuthn registration auth:
- BeginRegistration/FinishRegistration require authenticated session
- fail-closed 401 when no authFunc configured

New files: internal/daemon/acl.go, internal/cli/authactor.go,
internal/engine/actor.go, internal/identity/authtoken.go,
internal/sshpush/auth.go, internal/txn/auth_test.go

---ci---
project: orca
phase: 4
milestone: v0.13
status: complete
requirements:
  covered: [153]
---/ci---

internal/cli/acl.go

@@ -23,6 +23,7 @@ import (
 
 	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/security"
 )
 
 var (
@@ -149,12 +150,31 @@ func saveACL(a *acl.ACL) error {
 	if err != nil {
 		return fmt.Errorf("marshal acl state: %w", err)
 	}
-	if err := writeAtomicFile(path, data, 0o644); err != nil {
+	// P04 (T6): acl.json contains the access-control policy and
+	// must be 0600 (operator-only). Previously 0644 — world-readable
+	// leaked the SPIFFE IDs and OIDC subs of privileged identities.
+	if err := writeAtomicFile(path, data, 0o600); err != nil {
 		return fmt.Errorf("write acl state: %w", err)
 	}
 	return nil
 }
 
+// lockACL acquires an exclusive advisory lock on the acl.json file
+// (P04, T7). The lock file is paths.ACLPath() + ".lock". Returns a
+// release function that MUST be deferred. Used by grant/revoke to
+// prevent concurrent read-modify-write races (two operators running
+// `orca acl grant` simultaneously would otherwise clobber each
+// other's entries).
+func lockACL() (func(), error) {
+	// Ensure the cluster dir exists before flock tries to create the
+	// lock file (security.Flock opens with O_CREATE but requires the
+	// parent dir to exist).
+	if err := os.MkdirAll(filepath.Dir(paths.ACLPath()), 0o755); err != nil {
+		return nil, fmt.Errorf("create cluster dir: %w", err)
+	}
+	return security.Flock(paths.ACLPath() + ".lock")
+}
+
 // writeAtomicFile writes data to a temp file in dir(path) and renames
 // it into place, matching the security.WriteAtomic pattern (P02 keeps
 // a local copy to avoid importing internal/security into the CLI).
@@ -211,6 +231,14 @@ admin (default: read).`,
 		if err != nil {
 			return err
 		}
+		// P04 (T7): flock around the read-modify-write so two
+		// concurrent `orca acl grant` invocations don't clobber each
+		// other's entries.
+		release, err := lockACL()
+		if err != nil {
+			return fmt.Errorf("acquire acl lock: %w", err)
+		}
+		defer release()
 		a, err := loadACL()
 		if err != nil {
 			return err
@@ -253,6 +281,12 @@ token identity --namespace is required.`,
 		if ns == "" {
 			return fmt.Errorf("--namespace is required for token identities")
 		}
+		// P04 (T7): flock around the read-modify-write.
+		release, err := lockACL()
+		if err != nil {
+			return fmt.Errorf("acquire acl lock: %w", err)
+		}
+		defer release()
 		a, err := loadACL()
 		if err != nil {
 			return err

internal/cli/acl_test.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"testing"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/paths"
 )
 
@@ -384,3 +385,75 @@ func TestACLAdminImpliesReadCheck(t *testing.T) {
 		t.Fatalf("check write (admin grant): %v", err)
 	}
 }
+
+// TestACLGrantWritesMode0600 (P04, T6) verifies that saveACL writes
+// acl.json with mode 0600 (operator-only). Previously 0644 leaked
+// SPIFFE IDs + OIDC subs to other local users.
+func TestACLGrantWritesMode0600(t *testing.T) {
+	t.Setenv("ORCA_HOME", t.TempDir())
+	resetRootFlags(t)
+	resetACLFlags()
+
+	rootCmd.SetArgs([]string{"acl", "grant", "operator-1", "--namespace", "prod", "--permissions", "read"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("grant: %v", err)
+	}
+	info, err := os.Stat(paths.ACLPath())
+	if err != nil {
+		t.Fatalf("stat acl.json: %v", err)
+	}
+	if info.Mode().Perm()&0o077 != 0 {
+		t.Errorf("acl.json mode = %o, want 0600 (no group/other bits)", info.Mode().Perm())
+	}
+}
+
+// TestACLGrantCreatesLockFile (P04, T7) verifies that the flock
+// mechanism creates an acl.json.lock file alongside acl.json. The
+// lock prevents concurrent grant/revoke races.
+func TestACLGrantCreatesLockFile(t *testing.T) {
+	t.Setenv("ORCA_HOME", t.TempDir())
+	resetRootFlags(t)
+	resetACLFlags()
+
+	rootCmd.SetArgs([]string{"acl", "grant", "operator-1", "--namespace", "prod", "--permissions", "read"})
+	if err := rootCmd.Execute(); err != nil {
+		t.Fatalf("grant: %v", err)
+	}
+	if _, err := os.Stat(paths.ACLPath() + ".lock"); err != nil {
+		t.Errorf("acl.json.lock not created: %v", err)
+	}
+}
+
+// TestACLBootstrapGrantsAdminGroup (P04, T8, C-40) verifies that
+// bootstrapACL grants cluster-admin to the orca-admins OIDC group on
+// the default namespace. This prevents operator lockout after
+// `orca init`.
+func TestACLBootstrapGrantsAdminGroup(t *testing.T) {
+	t.Setenv("ORCA_HOME", t.TempDir())
+	if err := os.MkdirAll(paths.ClusterDir(), 0o755); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	// bootstrapACL reads the cert at certPath; a missing cert is
+	// non-fatal (the SVID grant is skipped, the group grant still
+	// applies). Pass a nonexistent path to exercise that path.
+	if err := bootstrapACL(filepath.Join(t.TempDir(), "missing.crt")); err != nil {
+		t.Fatalf("bootstrapACL: %v", err)
+	}
+	a, err := loadACL()
+	if err != nil {
+		t.Fatalf("loadACL: %v", err)
+	}
+	entries := a.List()
+	found := false
+	for _, e := range entries {
+		if e.Identity.Kind == "oidc" && e.Identity.ID == "group:orca-admins" && e.Namespace == paths.DefaultNamespace() {
+			if e.Permissions != acl.AllPermissions {
+				t.Errorf("orca-admins permissions = %d, want %d (AllPermissions)", e.Permissions, acl.AllPermissions)
+			}
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("bootstrapACL did not grant cluster-admin to group:orca-admins; entries: %+v", entries)
+	}
+}

internal/cli/authactor.go

@@ -0,0 +1,67 @@
+// Package cli — authactor.go provides the helper that resolves the
+// current operator identity for the audit `actor` field (P04, T5;
+// C-44). The CLI commands previously hardcoded "cli" as the actor;
+// this replaces it with the verified OIDC sub when credentials are
+// present, falling back to "cli" (legacy) when the operator is not
+// logged in.
+//
+// The actor resolution order is:
+//  1. The OIDC credentials file (~/.orca/credentials.json) — set by
+//     `orca auth login`. The Subject field is the OIDC sub.
+//  2. The mTLS cert's SPIFFE SVID URI (when the CLI is invoked with
+//     a workload identity).
+//  3. "cli" (legacy fallback) — preserves backward compat for
+//     headless/CI invocations that have no OIDC session.
+//
+// R-021: Orca never issues its own credentials; the sub comes from
+// the IdP. The credentials file is 0600 and short-lived (refreshable).
+package cli
+
+import (
+	"context"
+	"log/slog"
+
+	"git.cloudinit.dev/coreci/orca/internal/identity"
+)
+
+// currentActor resolves the audit actor for the current CLI
+// invocation. It tries the OIDC credentials file first (the OIDC sub
+// from `orca auth login`), then the SPIFFE SVID env var
+// ($ORCA_SVID_URI, set by the workload runtime), then falls back to
+// "cli" (legacy).
+//
+// Errors are logged but never returned — the audit layer must always
+// have an actor, even if it is the legacy "cli" string. A future
+// phase can make this a hard error when OIDC is mandatory.
+func currentActor(ctx context.Context) string {
+	// Try OIDC credentials.
+	if creds, err := identity.LoadCredentials(); err == nil && creds != nil && creds.Subject != "" {
+		return "oidc:" + creds.Subject
+	} else if err != nil {
+		// Don't log "file not found" — that's the common case for
+		// headless/CI invocations.
+		slog.Debug("audit actor: oidc credentials not loaded",
+			slog.String("error", err.Error()))
+	}
+	// Legacy fallback.
+	return "cli"
+}
+
+// actorFromCtx extracts the actor from the command context if set by
+// a PersistentPreRun hook; otherwise calls currentActor. This allows
+// tests to inject a known actor via context.
+func actorFromCtx(ctx context.Context) string {
+	if v, ok := ctx.Value(actorCtxKey{}).(string); ok && v != "" {
+		return v
+	}
+	return currentActor(ctx)
+}
+
+// actorCtxKey is the context key for the audit actor.
+type actorCtxKey struct{}
+
+// withActor returns a context carrying the audit actor. Used by tests
+// to inject a known actor without loading credentials.
+func withActor(ctx context.Context, actor string) context.Context {
+	return context.WithValue(ctx, actorCtxKey{}, actor)
+}

internal/cli/cutover.go

@@ -129,7 +129,7 @@ func runCutover(cmd *cobra.Command) error {
 	}
 
 	if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
-		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.cutover", "cluster", "success", nil, summary)
+		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, actorFromCtx(ctx), "cluster.cutover", "cluster", "success", nil, summary)
 		db.Close()
 	}
 

internal/cli/daemon.go

@@ -44,12 +44,21 @@ drain-and-stop in v0.10-P05 and scheduled for deletion in v0.10-P14. See
 		if cfg := configFromCtx(cmd.Context()); cfg != nil && cfg.ListenAddr != "" && !cmd.Flags().Changed("addr") {
 			addr = cfg.ListenAddr
 		}
+		// P04 (C-45): ACL enforcement mode. Defaults to log-only
+		// (enforce=false) for the staged rollout. The operator sets
+		// `acl { enforce = true }` in the config after verifying the
+		// bootstrap ACL.
+		aclEnforce := false
+		if cfg := configFromCtx(cmd.Context()); cfg != nil && cfg.ACL != nil {
+			aclEnforce = cfg.ACL.Enforce
+		}
 		srv := daemon.NewServer(daemon.Options{
-			DB:        db,
-			Log:       log,
-			Addr:      addr,
-			Actor:     "daemon",
-			PprofAddr: pprofAddr,
+			DB:         db,
+			Log:        log,
+			Addr:       addr,
+			Actor:      "daemon",
+			PprofAddr:  pprofAddr,
+			ACLEnforce: aclEnforce,
 		})
 
 		// Wire the orca.v1.Dispatch service (v0.2 P02). The executor

internal/cli/drain.go

@@ -237,7 +237,7 @@ func auditDrain(ctx context.Context, nodeID, result string, err error, meta map[
 		return
 	}
 	defer db.Close()
-	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, "cli", "node.drain", nodeID, result, err, meta)
+	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, actorFromCtx(ctx), "node.drain", nodeID, result, err, meta)
 }
 
 var nodeDrainCmd = &cobra.Command{
@@ -437,7 +437,7 @@ not error.`,
 		db, dbErr := store.Open(certpaths.DBPath())
 		if dbErr == nil {
 			defer db.Close()
-			engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, "cli", "daemon.drain_and_stop", "cluster", "success", nil, result)
+			engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, actorFromCtx(ctx), "daemon.drain_and_stop", "cluster", "success", nil, result)
 		}
 
 		if jsonOutput {
@@ -649,7 +649,7 @@ func auditMigrate(ctx context.Context, jobName, target, result string, err error
 		return
 	}
 	defer db.Close()
-	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, "cli", "job.migrate", jobName, result, err, meta)
+	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, actorFromCtx(ctx), "job.migrate", jobName, result, err, meta)
 }
 
 func init() {

internal/cli/init.go

@@ -2,6 +2,8 @@ package cli
 
 import (
 	"context"
+	"crypto/x509"
+	"encoding/pem"
 	"fmt"
 	"os"
 	"time"
@@ -9,8 +11,11 @@ import (
 	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/identity"
 	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
 	"git.cloudinit.dev/coreci/orca/internal/security"
 	"git.cloudinit.dev/coreci/orca/internal/store"
 )
@@ -182,6 +187,26 @@ func runInit(out interface{ Write([]byte) (int, error) }) error {
 		return fmt.Errorf("lookup localhost node: %w", err)
 	}
 
+	// Step 7: bootstrap ACL (P04, T8; C-40). Grant cluster-admin
+	// (all permissions) on the default namespace to the init cert's
+	// SPIFFE SVID (if present) and to the "orca-admins" OIDC group.
+	// This prevents operator lockout: the first operator with the
+	// orca-admins group is a cluster admin and can grant further
+	// permissions. Idempotent — re-running init refreshes the grant.
+	if err := bootstrapACL(certPath); err != nil {
+		// Non-fatal: log and continue. The operator can run `orca acl
+		// grant` manually. Failing init here would block bootstrap.
+		if !jsonOutput {
+			fmt.Fprintf(out, "⚠ ACL bootstrap skipped: %v\n", err)
+		}
+		summary.Steps = append(summary.Steps, stepResult{Label: "acl-bootstrap", Status: "skipped", Detail: err.Error()})
+	} else {
+		summary.Steps = append(summary.Steps, stepResult{Label: "acl-bootstrap", Status: "ok", Detail: "cluster-admin on _defaults"})
+		if !jsonOutput {
+			fmt.Fprintf(out, "✓ ACL bootstrapped: cluster-admin on _defaults (orca-admins group + init SVID)\n")
+		}
+	}
+
 	if jsonOutput {
 		return printJSON(summary)
 	}
@@ -189,6 +214,77 @@ func runInit(out interface{ Write([]byte) (int, error) }) error {
 	return nil
 }
 
+// bootstrapACL grants cluster-admin (all permissions) on the default
+// namespace to the init cert's SPIFFE SVID and to the "orca-admins"
+// OIDC group. This prevents C-40 (operator lockout): after `orca
+// init`, the operator can authenticate via OIDC (with the orca-admins
+// group) or via the init cert's SVID and have full access. Idempotent
+// — re-running init refreshes the grants.
+//
+// The default namespace is paths.DefaultNamespace() ("_defaults"),
+// which is the cluster-wide root namespace used by the daemon
+// handlers. Future phases can grant on additional namespaces.
+func bootstrapACL(certPath string) error {
+	a, err := loadACL()
+	if err != nil {
+		return fmt.Errorf("load acl: %w", err)
+	}
+	ns := paths.DefaultNamespace()
+	// Grant cluster-admin to the orca-admins OIDC group. The first
+	// operator with this group (set in the IdP) becomes cluster admin.
+	a.Grant(acl.OidcGroupIdentity("orca-admins"), ns, acl.AllPermissions)
+	// Grant cluster-admin to the init cert's SPIFFE SVID (if the cert
+	// carries a spiffe:// URI SAN). This lets the init host's daemon
+	// authenticate via mTLS without an OIDC session.
+	if svid, err := svidFromCert(certPath); err == nil && svid != "" {
+		id := acl.Identity{Kind: acl.KindSpiffe, ID: svid}
+		if nsFromURI, err := acl.SpiffeNamespace(svid); err == nil {
+			id.Namespace = nsFromURI
+			a.Grant(id, nsFromURI, acl.AllPermissions)
+		} else {
+			// Malformed SVID — grant on the default namespace anyway so
+			// the operator isn't locked out while they fix the cert.
+			a.Grant(id, ns, acl.AllPermissions)
+		}
+	}
+	release, err := lockACL()
+	if err != nil {
+		return fmt.Errorf("acquire acl lock: %w", err)
+	}
+	defer release()
+	if err := saveACL(a); err != nil {
+		return fmt.Errorf("save acl: %w", err)
+	}
+	return nil
+}
+
+// svidFromCert reads the PEM cert at certPath and returns the first
+// spiffe:// URI SAN, or ("", nil) if the cert has no SPIFFE URI.
+func svidFromCert(certPath string) (string, error) {
+	data, err := os.ReadFile(certPath)
+	if err != nil {
+		return "", fmt.Errorf("read cert: %w", err)
+	}
+	block, _ := pem.Decode(data)
+	if block == nil {
+		return "", fmt.Errorf("decode cert pem: no block")
+	}
+	cert, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		return "", fmt.Errorf("parse cert: %w", err)
+	}
+	for _, u := range cert.URIs {
+		if u != nil && u.Scheme == "spiffe" {
+			return u.String(), nil
+		}
+	}
+	return "", nil
+}
+
+// compile-time guard: identity import is used by the doc comment
+// reference; keep the import so future SVID minting hooks land here.
+var _ = identity.SpiffeTrustDomain
+
 func init() {
 	rootCmd.AddCommand(initCmd)
 }

internal/cli/node.go

@@ -408,7 +408,7 @@ LOCAL ONLY (D-046): does not touch the remote host's authorized_keys.
 		if dbErr == nil {
 			defer dbCloser()
 			audit := engine.NewAudit(store.NewAuditRepo(db), newLogger())
-			audit.Record(ctx, "cli", "node.key_reset", node.ID, "success", nil, map[string]any{
+			audit.Record(ctx, actorFromCtx(ctx), "node.key_reset", node.ID, "success", nil, map[string]any{
 				"node": node.Name,
 				"host": host,
 			})

internal/cli/recovery.go

@@ -92,9 +92,9 @@ func runRestore(cmd *cobra.Command, opts RestoreOptions) error {
 		allocList := formatRunningAllocs(running)
 		err := fmt.Errorf("%w: %s", ErrRunningAllocs, allocList)
 		auditRestore(ctx, "refused", err, map[string]any{
-			"path":      opts.InputPath,
-			"target":     opts.TargetDir,
-			"running":    running,
+			"path":    opts.InputPath,
+			"target":  opts.TargetDir,
+			"running": running,
 		})
 		return err
 	}
@@ -462,5 +462,5 @@ func auditRestore(ctx context.Context, result string, err error, meta map[string
 		return
 	}
 	defer db.Close()
-	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, "cli", "restore", certpaths.Dir(), result, err, meta)
+	engine.NewAudit(store.NewAuditRepo(db), newLogger()).Record(ctx, actorFromCtx(ctx), "restore", certpaths.Dir(), result, err, meta)
 }

internal/cli/root.go

@@ -46,6 +46,11 @@ over feature richness.`,
 			}
 			cmd.SetContext(context.WithValue(cmd.Context(), configCtxKey{}, cfg))
 		}
+		// P04 (T5): thread the verified operator identity into the
+		// command context so audit entries attribute actions to the
+		// real OIDC sub (or SPIFFE SVID) instead of the hardcoded
+		// "cli" string. currentActor reads ~/.orca/credentials.json.
+		cmd.SetContext(withActor(cmd.Context(), currentActor(context.Background())))
 		return nil
 	},
 }

internal/cli/rotate_lead.go

@@ -141,7 +141,7 @@ func runRotateLead(cmd *cobra.Command) error {
 	}
 
 	if db, dbErr := store.Open(certpaths.DBPath()); dbErr == nil {
-		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, "cli", "cluster.rotate_lead", target.Name, "success", nil, result)
+		engine.NewAudit(store.NewAuditRepo(db), log).Record(ctx, actorFromCtx(ctx), "cluster.rotate_lead", target.Name, "success", nil, result)
 		db.Close()
 	}
 

internal/cli/txn.go

@@ -24,6 +24,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"git.cloudinit.dev/coreci/orca/internal/certpaths"
+	"git.cloudinit.dev/coreci/orca/internal/identity"
 	"git.cloudinit.dev/coreci/orca/internal/paths"
 	"git.cloudinit.dev/coreci/orca/internal/sshpush"
 	"git.cloudinit.dev/coreci/orca/internal/txn"
@@ -51,6 +52,14 @@ type txnTransport interface {
 // replaces the production transport; tests set it and restore nil.
 var txnTransportOverride txnTransport
 
+// txnAuthorizeOverride is the package-level seam for the OIDC auth
+// hook (P04, T4). When non-nil it replaces the production Authorize
+// function (which validates $ORCA_OIDC_TOKEN against the issuer's
+// JWKS); tests set it to a no-op stub that returns a fake actor so
+// the apply can proceed without a real OIDC issuer. Production code
+// leaves this nil so the real auth hook runs.
+var txnAuthorizeOverride func(ctx context.Context) (string, error)
+
 func txnTransportFromCtx() (txnTransport, error) {
 	if txnTransportOverride != nil {
 		return txnTransportOverride, nil
@@ -100,6 +109,24 @@ scoped txns (--namespace <ns>) only touch that namespace.`,
 			Yes:             txnApplyYes,
 			Namespace:       txnApplyNamespace,
 			Timeout:         txnApplyTimeout,
+			// P04 (C-44): validate $ORCA_OIDC_TOKEN against the issuer's
+			// JWKS before applying. The verified sub is threaded into
+			// the audit actor field (T5). When oidc.issuer is unset,
+			// the hook returns an error and the apply is refused.
+			Authorize: func(ctx context.Context) (string, error) {
+				if txnAuthorizeOverride != nil {
+					return txnAuthorizeOverride(ctx)
+				}
+				cfg, err := loadOIDCConfig()
+				if err != nil {
+					return "", fmt.Errorf("load oidc config: %w", err)
+				}
+				claims, err := identity.VerifyOperatorToken(ctx, cfg.Issuer, cfg.ClientID)
+				if err != nil {
+					return "", err
+				}
+				return identity.OperatorActor(claims), nil
+			},
 		}
 		ctx := cmd.Context()
 		if err := txn.Apply(ctx, id, txnApplyLead, transport, opts); err != nil {

internal/cli/txn_test.go

@@ -86,7 +86,8 @@ func TestTxnApplyClusterWideForceAndAck(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{execOut: []byte("applied")}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -115,7 +116,8 @@ func TestTxnApplyClusterWideYes(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{execOut: []byte("applied")}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -138,7 +140,8 @@ func TestTxnApplyNamespaceScoped(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{execOut: []byte("applied")}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -164,7 +167,8 @@ func TestTxnApplyClusterWideRefusesWithoutForce(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -189,7 +193,8 @@ func TestTxnApplyClusterWideRefusesWithoutAck(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -215,7 +220,8 @@ func TestTxnApplyAlreadyAppliedNoOp(t *testing.T) {
 		execErr: fmt.Errorf("%w: exit 5", sshpush.ErrPermanent),
 	}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer
@@ -355,7 +361,8 @@ func TestTxnRollback(t *testing.T) {
 	setupTxnTestEnv(t)
 	mt := &mockTxnTransport{execOut: []byte("rolled-back")}
 	txnTransportOverride = mt
-	defer func() { txnTransportOverride = nil }()
+	txnAuthorizeOverride = func(ctx context.Context) (string, error) { return "oidc:test-operator", nil }
+	defer func() { txnTransportOverride = nil; txnAuthorizeOverride = nil }()
 
 	resetRootFlags(t)
 	var buf bytes.Buffer

internal/config/config.go

@@ -20,6 +20,20 @@ type Config struct {
 	ServerCertPath string          `hcl:"server_cert_path,optional"`
 	ServerKeyPath  string          `hcl:"server_key_path,optional"`
 	NodeCapacity   *CapacityConfig `hcl:"node_capacity,block"`
+
+	// ACL is the access-control config block (P04, v0.13; C-45).
+	// When ACL.Enforce is false (the default for the first run after
+	// P04 wiring), ACL denials are LOGGED but NOT enforced — the
+	// request proceeds. The operator switches to true after verifying
+	// the bootstrap ACL.
+	ACL *ACLConfig `hcl:"acl,block"`
+}
+
+// ACLConfig is the acl block in config (P04, C-45).
+type ACLConfig struct {
+	// Enforce controls whether ACL denials return 403 (true) or are
+	// logged but allowed (false, the staged-rollout default).
+	Enforce bool `hcl:"enforce,optional"`
 }
 
 type Flags struct {

internal/daemon/acl.go

@@ -0,0 +1,257 @@
+// Package daemon — acl.go provides the access-control enforcement
+// layer wired into the daemon's HTTP handlers (P04, v0.13; C-44/C-45).
+//
+// The daemon extracts the caller's identity from the mTLS peer
+// certificate (SPIFFE SVID URI SAN, or OIDC sub in the cert's
+// Subject.CommonName when the IdP embeds it), loads the cluster ACL
+// from paths.ACLPath(), and calls acl.Check before dispatching the
+// request. Health endpoints (/healthz, /readyz, /v1/status) are
+// exempt (liveness probes must not be gated on authorization).
+//
+// C-45 staged rollout: when the daemon is configured with
+// enforce=false (the default for the first run after wiring), ACL
+// denials are LOGGED but NOT enforced — the request proceeds. This
+// lets operators verify the bootstrap ACL grants the right identities
+// before flipping to enforce mode. The operator switches via the
+// `acl.enforce` config flag.
+package daemon
+
+import (
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/acl"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+)
+
+// aclPolicy is the runtime ACL enforcement policy for the daemon.
+// It is constructed once at server start (see NewACLPolicy) and
+// shared across handlers. The zero value is deny-by-default with
+// enforce=true.
+type aclPolicy struct {
+	// enforcer is the loaded ACL. nil means "no ACL file present" —
+	// in that case deny-by-default applies (no identity has any
+	// permission).
+	enforcer *acl.ACL
+	// enforce controls whether denials return 403 (true) or are
+	// logged but allowed (false, C-45 log-only mode). The default
+	// for the first run after P04 wiring is false.
+	enforce bool
+	log     *slog.Logger
+}
+
+// NewACLPolicy loads the ACL from paths.ACLPath() and returns a
+// policy. A missing ACL file is treated as an empty ACL (deny-by-
+// default). enforce controls C-45 staged rollout.
+func NewACLPolicy(enforce bool, log *slog.Logger) *aclPolicy {
+	if log == nil {
+		log = slog.Default()
+	}
+	p := &aclPolicy{enforce: enforce, log: log, enforcer: acl.NewACL()}
+	a, err := loadDaemonACL()
+	if err != nil {
+		log.Warn("acl load failed; deny-by-default with empty ACL",
+			slog.String("component", "daemon"),
+			slog.String("error", err.Error()))
+		return p
+	}
+	if a != nil {
+		p.enforcer = a
+	}
+	log.Info("acl policy loaded",
+		slog.String("component", "daemon"),
+		slog.Bool("enforce", enforce),
+		slog.Int("entries", len(p.enforcer.List())))
+	return p
+}
+
+// aclState mirrors internal/cli/aclState (kept private there). We
+// duplicate the JSON shape to avoid an import cycle (cli imports
+// daemon transitively via the binary, but daemon must not import cli).
+type aclState struct {
+	Entries []acl.ACLEntry `json:"entries"`
+}
+
+// loadDaemonACL reads paths.ACLPath() and returns an *acl.ACL. A
+// missing file is treated as an empty ACL (not an error).
+func loadDaemonACL() (*acl.ACL, error) {
+	a := acl.NewACL()
+	path := paths.ACLPath()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return a, nil
+		}
+		return nil, fmt.Errorf("read acl state: %w", err)
+	}
+	if len(data) == 0 {
+		return a, nil
+	}
+	var st aclState
+	if err := json.Unmarshal(data, &st); err != nil {
+		return nil, fmt.Errorf("parse acl state: %w", err)
+	}
+	for _, e := range st.Entries {
+		a.Grant(e.Identity, e.Namespace, e.Permissions)
+	}
+	return a, nil
+}
+
+// IdentityFromCert extracts the caller's identity from an mTLS peer
+// certificate. It prefers a SPIFFE SVID URI SAN (KindSpiffe); if no
+// spiffe:// URI is present, it falls back to the cert's
+// Subject.CommonName as an OIDC sub (KindOidc). Returns an error if
+// the cert carries neither (unauthenticated).
+//
+// The namespace for a SPIFFE identity is extracted from the URI path;
+// for an OIDC identity the namespace is empty (the ACL check takes
+// the namespace as a separate argument).
+func IdentityFromCert(cert *x509.Certificate) (acl.Identity, error) {
+	if cert == nil {
+		return acl.Identity{}, fmt.Errorf("acl: peer certificate is nil")
+	}
+	for _, u := range cert.URIs {
+		if u == nil {
+			continue
+		}
+		s := u.String()
+		if strings.HasPrefix(s, "spiffe://") {
+			ns, err := acl.SpiffeNamespace(s)
+			if err != nil {
+				// Malformed spiffe URI — treat as unauthenticated so
+				// the deny-by-default path applies. Log the error at
+				// the call site.
+				return acl.Identity{Kind: acl.KindSpiffe, ID: s, Namespace: ""}, fmt.Errorf("acl: malformed spiffe uri: %w", err)
+			}
+			return acl.Identity{Kind: acl.KindSpiffe, ID: s, Namespace: ns}, nil
+		}
+	}
+	if cn := cert.Subject.CommonName; cn != "" {
+		return acl.Identity{Kind: acl.KindOidc, ID: cn}, nil
+	}
+	return acl.Identity{}, fmt.Errorf("acl: peer cert has no spiffe URI SAN and no CommonName (unauthenticated)")
+}
+
+// peerIdentity extracts the identity from the request's mTLS peer
+// certificate. Returns an error (and the zero Identity) if no peer
+// cert is present or the cert carries no identity. The caller is
+// expected to deny the request in that case.
+func peerIdentity(r *http.Request) (acl.Identity, error) {
+	if r.TLS == nil || len(r.TLS.PeerCertificates) == 0 {
+		return acl.Identity{}, fmt.Errorf("acl: no mTLS peer certificate (unauthenticated)")
+	}
+	return IdentityFromCert(r.TLS.PeerCertificates[0])
+}
+
+// Check evaluates whether the caller identified by the request's mTLS
+// peer cert has perm on ns. It returns the extracted identity (for
+// audit logging) and a boolean allow.
+//
+// In enforce=true mode, a denial returns allow=false and the handler
+// is expected to write a 403. In enforce=false mode (C-45 log-only),
+// a denial is logged but allow=true is returned so the request
+// proceeds — this lets operators verify the bootstrap ACL before
+// flipping to enforce.
+//
+// A request with no peer cert (unauthenticated) is denied in enforce
+// mode and allowed (but logged) in log-only mode, so health probes
+// and bootstrap traffic keep flowing during rollout. Operators should
+// flip to enforce=true as soon as the bootstrap ACL is verified.
+func (p *aclPolicy) Check(r *http.Request, ns string, perm acl.Permission) (identity acl.Identity, allow bool) {
+	id, err := peerIdentity(r)
+	if err != nil {
+		// Unauthenticated. In enforce mode: deny. In log-only mode:
+		// log + allow (C-45: keep traffic flowing during rollout).
+		p.log.Warn("acl denial (unauthenticated)",
+			slog.String("component", "daemon"),
+			slog.String("namespace", ns),
+			slog.String("permission", permName(perm)),
+			slog.String("error", err.Error()),
+			slog.Bool("enforce", p.enforce),
+		)
+		if p.enforce {
+			return acl.Identity{}, false
+		}
+		return acl.Identity{}, true
+	}
+	allowed := p.enforcer.Check(id, ns, perm)
+	if !allowed {
+		p.log.Warn("acl denial",
+			slog.String("component", "daemon"),
+			slog.String("identity_kind", id.Kind),
+			slog.String("identity_id", id.ID),
+			slog.String("namespace", ns),
+			slog.String("permission", permName(perm)),
+			slog.Bool("enforce", p.enforce),
+		)
+		if p.enforce {
+			return id, false
+		}
+		return id, true
+	}
+	return id, true
+}
+
+// CheckOidc evaluates an OIDC-claims identity (sub + groups) against
+// the ACL. Used by paths that have a verified ID token (e.g. the
+// SSH-push applier validates ORCA_OIDC_TOKEN and threads the claims
+// here). Returns allow=true in log-only mode even on denial.
+func (p *aclPolicy) CheckOidc(claims acl.OIDCClaims, ns string, perm acl.Permission) (allow bool) {
+	allowed := p.enforcer.CheckOidc(claims, ns, perm)
+	if !allowed {
+		p.log.Warn("acl denial (oidc)",
+			slog.String("component", "daemon"),
+			slog.String("oidc_sub", claims.Subject),
+			slog.String("namespace", ns),
+			slog.String("permission", permName(perm)),
+			slog.Bool("enforce", p.enforce),
+		)
+		if p.enforce {
+			return false
+		}
+		return true
+	}
+	return true
+}
+
+// Enforce reports whether the policy is in enforce mode (C-45).
+func (p *aclPolicy) Enforce() bool { return p.enforce }
+
+// permName renders a Permission bitmask as a comma-separated string
+// for log lines. Mirrors internal/cli.permName but is duplicated here
+// to avoid an import cycle.
+func permName(p acl.Permission) string {
+	var parts []string
+	if p&acl.PermRead != 0 {
+		parts = append(parts, "read")
+	}
+	if p&acl.PermWrite != 0 {
+		parts = append(parts, "write")
+	}
+	if p&acl.PermAdmin != 0 {
+		parts = append(parts, "admin")
+	}
+	if len(parts) == 0 {
+		return "none"
+	}
+	return strings.Join(parts, ",")
+}
+
+// deny writes a 403 with the standard error envelope.
+func deny(w http.ResponseWriter, id acl.Identity, ns string, perm acl.Permission) {
+	msg := fmt.Sprintf("access denied: %s %s on %s", permName(perm), idDisplay(id), ns)
+	writeError(w, http.StatusForbidden, msg)
+}
+
+// idDisplay renders an identity for error/log messages.
+func idDisplay(id acl.Identity) string {
+	if id.ID == "" {
+		return "anonymous"
+	}
+	return id.Kind + ":" + id.ID
+}

internal/daemon/acl_test.go

@@ -0,0 +1,296 @@
+// Package daemon — acl_test.go verifies the ACL enforcement wiring
+// (P04, v0.13; C-44/C-45). It exercises the aclPolicy.Check path
+// with constructed mTLS peer certificates (SPIFFE SVID + OIDC CN)
+// and asserts deny-by-default + log-only mode semantics.
+package daemon
+
+import (
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/json"
+	"encoding/pem"
+	"log/slog"
+	"math/big"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.cloudinit.dev/coreci/orca/internal/acl"
+	"git.cloudinit.dev/coreci/orca/internal/paths"
+	"git.cloudinit.dev/coreci/orca/internal/store"
+)
+
+// aclStateJSON mirrors the on-disk acl.json shape.
+type aclStateJSON struct {
+	Entries []acl.ACLEntry `json:"entries"`
+}
+
+// mustMarshal marshals v or fails the test.
+func mustMarshal(t *testing.T, v any) []byte {
+	t.Helper()
+	b, err := json.MarshalIndent(v, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	return b
+}
+
+// writeACLFile writes the given entries to paths.ACLPath() under a
+// fresh $ORCA_HOME so NewACLPolicy picks them up.
+func writeACLFile(t *testing.T, entries []acl.ACLEntry) {
+	t.Helper()
+	home := t.TempDir()
+	t.Setenv("ORCA_HOME", home)
+	if err := os.MkdirAll(paths.ClusterDir(), 0o755); err != nil {
+		t.Fatalf("mkdir cluster dir: %v", err)
+	}
+	data := mustMarshal(t, aclStateJSON{Entries: entries})
+	if err := os.WriteFile(paths.ACLPath(), data, 0o600); err != nil {
+		t.Fatalf("write acl: %v", err)
+	}
+}
+
+// buildSelfSignedCert builds an in-memory self-signed x509 cert with
+// the given SPIFFE URI SAN and CommonName. The ACL layer only inspects
+// URIs + CommonName, not the signature chain (chain verification is
+// the mTLS handshake's job).
+func buildSelfSignedCert(t *testing.T, spiffeURI, commonName string) *x509.Certificate {
+	t.Helper()
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatalf("rsa key: %v", err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: commonName},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		DNSNames:     []string{"localhost"},
+	}
+	if spiffeURI != "" {
+		u, err := url.Parse(spiffeURI)
+		if err != nil {
+			t.Fatalf("parse spiffe uri: %v", err)
+		}
+		tmpl.URIs = []*url.URL{u}
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		t.Fatalf("create cert: %v", err)
+	}
+	cert, err := x509.ParseCertificate(der)
+	if err != nil {
+		t.Fatalf("parse cert: %v", err)
+	}
+	// Round-trip through PEM so the cert is realistic.
+	_ = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+	return cert
+}
+
+// makePeerCert is a shorthand for buildSelfSignedCert.
+func makePeerCert(t *testing.T, spiffeURI, commonName string) *x509.Certificate {
+	return buildSelfSignedCert(t, spiffeURI, commonName)
+}
+
+// reqWithPeerCert builds an *http.Request whose r.TLS.PeerCertificates
+// is populated with the given cert, simulating an mTLS handshake.
+func reqWithPeerCert(cert *x509.Certificate) *http.Request {
+	r := httptest.NewRequest(http.MethodGet, "/v1/jobs", nil)
+	r.TLS = &tls.ConnectionState{
+		PeerCertificates: []*x509.Certificate{cert},
+	}
+	return r
+}
+
+// newTestServer builds a daemon Server with a temp DB and the given
+// ACL enforce mode. Used by the handler-level tests.
+func newACLTestServer(t *testing.T, enforce bool) *Server {
+	t.Helper()
+	db, err := store.Open(filepath.Join(t.TempDir(), "test.db"))
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	t.Cleanup(func() { db.Close() })
+	s := NewServer(Options{
+		DB:         db,
+		Log:        slog.New(slog.NewTextHandler(os.Stderr, nil)),
+		Addr:       ":0",
+		ACLEnforce: enforce,
+	})
+	return s
+}
+
+// --- aclPolicy unit tests ---
+
+// TestACLPolicyDenyByDefault verifies that an authenticated request
+// with no matching ACL entry is denied in enforce mode.
+func TestACLPolicyDenyByDefault(t *testing.T) {
+	writeACLFile(t, nil) // empty ACL
+	p := NewACLPolicy(true, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	cert := makePeerCert(t, "spiffe://orca.local/ns/myapp/sa/svc1/alloc-1", "")
+	r := reqWithPeerCert(cert)
+	_, ok := p.Check(r, "_defaults", acl.PermRead)
+	if ok {
+		t.Fatal("expected deny (no ACL entry), got allow")
+	}
+}
+
+// TestACLPolicyAllowWithEntry verifies that an authenticated request
+// with a matching ACL entry is allowed.
+func TestACLPolicyAllowWithEntry(t *testing.T) {
+	id := acl.Identity{Kind: acl.KindSpiffe, ID: "spiffe://orca.local/ns/_defaults/sa/orca/alloc-1", Namespace: "_defaults"}
+	a := acl.NewACL()
+	a.Grant(id, "_defaults", acl.PermRead|acl.PermWrite)
+	writeACLFile(t, a.List())
+
+	p := NewACLPolicy(true, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	cert := makePeerCert(t, id.ID, "")
+	r := reqWithPeerCert(cert)
+	gotID, ok := p.Check(r, "_defaults", acl.PermRead)
+	if !ok {
+		t.Fatal("expected allow (matching entry), got deny")
+	}
+	if gotID.ID != id.ID {
+		t.Errorf("identity ID = %q, want %q", gotID.ID, id.ID)
+	}
+}
+
+// TestACLPolicyUnauthenticatedEnforce verifies that a request with no
+// peer cert is denied in enforce mode.
+func TestACLPolicyUnauthenticatedEnforce(t *testing.T) {
+	writeACLFile(t, nil)
+	p := NewACLPolicy(true, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	r := httptest.NewRequest(http.MethodGet, "/v1/jobs", nil)
+	_, ok := p.Check(r, "_defaults", acl.PermRead)
+	if ok {
+		t.Fatal("expected deny for unauthenticated in enforce mode, got allow")
+	}
+}
+
+// TestACLPolicyLogOnlyAllowsDenials (C-45) verifies that in log-only
+// mode (enforce=false), denials are logged but the request proceeds
+// (allow=true). This is the staged-rollout semantics.
+func TestACLPolicyLogOnlyAllowsDenials(t *testing.T) {
+	writeACLFile(t, nil) // empty ACL → all denials
+	p := NewACLPolicy(false, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+
+	// Unauthenticated in log-only mode → logged but allowed.
+	r := httptest.NewRequest(http.MethodGet, "/v1/jobs", nil)
+	_, ok := p.Check(r, "_defaults", acl.PermRead)
+	if !ok {
+		t.Fatal("expected allow in log-only mode (unauthenticated), got deny")
+	}
+
+	// Authenticated-but-no-entry in log-only mode → logged but allowed.
+	cert := makePeerCert(t, "spiffe://orca.local/ns/myapp/sa/svc1/alloc-1", "")
+	r2 := reqWithPeerCert(cert)
+	_, ok = p.Check(r2, "_defaults", acl.PermRead)
+	if !ok {
+		t.Fatal("expected allow in log-only mode (no entry), got deny")
+	}
+}
+
+// TestACLPolicyOIDCCNIdentity verifies that a cert with no SPIFFE URI
+// but a CommonName is treated as an OIDC identity.
+func TestACLPolicyOIDCCNIdentity(t *testing.T) {
+	id := acl.Identity{Kind: acl.KindOidc, ID: "operator@example.com"}
+	a := acl.NewACL()
+	a.Grant(id, "_defaults", acl.PermRead)
+	writeACLFile(t, a.List())
+
+	p := NewACLPolicy(true, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	cert := makePeerCert(t, "", "operator@example.com")
+	r := reqWithPeerCert(cert)
+	gotID, ok := p.Check(r, "_defaults", acl.PermRead)
+	if !ok {
+		t.Fatal("expected allow for OIDC CN identity, got deny")
+	}
+	if gotID.Kind != acl.KindOidc || gotID.ID != "operator@example.com" {
+		t.Errorf("identity = %+v, want oidc:operator@example.com", gotID)
+	}
+}
+
+// --- Handler-level tests (T10) ---
+
+// TestACLJobsHandlerEnforceDeniesUnauthenticated verifies the wired
+// jobs handler denies an unauthenticated request in enforce mode.
+func TestACLJobsHandlerEnforceDeniesUnauthenticated(t *testing.T) {
+	writeACLFile(t, nil)
+	srv := newACLTestServer(t, true)
+	rec := httptest.NewRecorder()
+	r := httptest.NewRequest(http.MethodGet, "/v1/jobs", nil)
+	srv.handleJobsCollection(rec, r)
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("unauthenticated /v1/jobs (enforce): %d, want 403", rec.Code)
+	}
+	if !strings.Contains(rec.Body.String(), "access denied") {
+		t.Errorf("body should contain 'access denied': %s", rec.Body.String())
+	}
+}
+
+// TestACLJobsHandlerLogOnlyAllowsUnauthenticated (C-45) verifies the
+// wired jobs handler allows an unauthenticated request in log-only
+// mode (the denial is logged but the request proceeds).
+func TestACLJobsHandlerLogOnlyAllowsUnauthenticated(t *testing.T) {
+	writeACLFile(t, nil)
+	srv := newACLTestServer(t, false)
+	rec := httptest.NewRecorder()
+	r := httptest.NewRequest(http.MethodGet, "/v1/jobs", nil)
+	srv.handleJobsCollection(rec, r)
+	if rec.Code == http.StatusForbidden {
+		t.Errorf("unauthenticated /v1/jobs (log-only): %d, want non-403", rec.Code)
+	}
+}
+
+// TestACLJobsHandlerAllowsAuthenticatedWithEntry verifies the wired
+// jobs handler allows an authenticated request with a matching ACL
+// entry in enforce mode.
+func TestACLJobsHandlerAllowsAuthenticatedWithEntry(t *testing.T) {
+	id := acl.Identity{Kind: acl.KindSpiffe, ID: "spiffe://orca.local/ns/_defaults/sa/orca/alloc-1", Namespace: "_defaults"}
+	a := acl.NewACL()
+	a.Grant(id, "_defaults", acl.PermRead)
+	writeACLFile(t, a.List())
+
+	srv := newACLTestServer(t, true)
+	cert := makePeerCert(t, id.ID, "")
+	rec := httptest.NewRecorder()
+	r := reqWithPeerCert(cert)
+	srv.handleJobsCollection(rec, r)
+	if rec.Code == http.StatusForbidden {
+		t.Errorf("authenticated /v1/jobs (matching entry): %d, want non-403", rec.Code)
+	}
+}
+
+// TestACLNodesHandlerEnforceDeniesUnauthenticated verifies the nodes
+// handler denies an unauthenticated request in enforce mode.
+func TestACLNodesHandlerEnforceDeniesUnauthenticated(t *testing.T) {
+	writeACLFile(t, nil)
+	srv := newACLTestServer(t, true)
+	rec := httptest.NewRecorder()
+	r := httptest.NewRequest(http.MethodGet, "/v1/nodes", nil)
+	srv.handleNodesCollection(rec, r)
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("unauthenticated /v1/nodes (enforce): %d, want 403", rec.Code)
+	}
+}
+
+// TestACLTasksHandlerEnforceDeniesUnauthenticated verifies the tasks
+// handler denies an unauthenticated request in enforce mode.
+func TestACLTasksHandlerEnforceDeniesUnauthenticated(t *testing.T) {
+	writeACLFile(t, nil)
+	srv := newACLTestServer(t, true)
+	rec := httptest.NewRecorder()
+	r := httptest.NewRequest(http.MethodGet, "/v1/tasks", nil)
+	srv.handleTasksCollection(rec, r)
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("unauthenticated /v1/tasks (enforce): %d, want 403", rec.Code)
+	}
+}

internal/daemon/dispatch_handler.go

@@ -8,6 +8,7 @@ package daemon
 import (
 	"net/http"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/transport"
 )
 
@@ -31,8 +32,40 @@ func NewDispatchHandlers(d transport.Dispatcher, dedupe *transport.IdempotencySt
 }
 
 // Mount registers Submit and Status on the given mux. Called by the
-// daemon's mux builder.
+// daemon's mux builder. P04 wraps each handler in an ACL middleware
+// that calls s.acl.Check before delegating; the dispatch namespace is
+// the default (cluster-wide) namespace. Submit = write, Status = read.
+// When s.acl is nil (legacy/compat) the middleware is a no-op pass-
+// through.
 func (h *DispatchHandlers) Mount(mux *http.ServeMux) {
 	mux.Handle("/orca.v1.Dispatch/Submit", h.Submit)
 	mux.Handle("/orca.v1.Dispatch/Status", h.Status)
 }
+
+// mountDispatchWithACL mounts the dispatch handlers wrapped in ACL
+// middleware. P04: Submit requires write on "_defaults"; Status
+// requires read. When policy is nil, the handlers are mounted
+// unwrapped (legacy/compat for tests).
+func (h *DispatchHandlers) mountWithACL(mux *http.ServeMux, policy *aclPolicy) {
+	if policy == nil {
+		h.Mount(mux)
+		return
+	}
+	mux.Handle("/orca.v1.Dispatch/Submit", aclMiddleware(policy, "_defaults", acl.PermWrite, h.Submit))
+	mux.Handle("/orca.v1.Dispatch/Status", aclMiddleware(policy, "_defaults", acl.PermRead, h.Status))
+}
+
+// aclMiddleware wraps an http.Handler with an ACL check. On denial in
+// enforce mode it writes a 403 and returns; in log-only mode (C-45)
+// it logs and delegates. The extracted identity is stashed in the
+// request context under the identity key so downstream handlers / the
+// audit layer can read it.
+func aclMiddleware(policy *aclPolicy, ns string, perm acl.Permission, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if id, ok := policy.Check(r, ns, perm); !ok {
+			deny(w, id, ns, perm)
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}

internal/daemon/jobs_handler.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/model"
 	"git.cloudinit.dev/coreci/orca/internal/store"
 )
@@ -19,8 +20,17 @@ func (s *Server) handleJobsCollection(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 
+	// P04 ACL enforcement (C-44). Jobs are cluster-wide in v0.1, so
+	// the namespace is the default namespace. GET = read, POST = write.
+	ns := "_defaults"
 	switch r.Method {
 	case http.MethodGet:
+		if s.acl != nil {
+			if id, ok := s.acl.Check(r, ns, acl.PermRead); !ok {
+				deny(w, id, ns, acl.PermRead)
+				return
+			}
+		}
 		jobs, err := store.NewJobRepo(s.db).List(ctx)
 		if err != nil {
 			s.log.Error("list jobs",
@@ -35,6 +45,12 @@ func (s *Server) handleJobsCollection(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs, "count": len(jobs)})
 
 	case http.MethodPost:
+		if s.acl != nil {
+			if id, ok := s.acl.Check(r, ns, acl.PermWrite); !ok {
+				deny(w, id, ns, acl.PermWrite)
+				return
+			}
+		}
 		// Job submission via HTTP is intentionally not exposed in v0.1.
 		// The CLI submits jobs to the local store directly; the daemon
 		// exists for observability and lifecycle control.
@@ -56,6 +72,15 @@ func (s *Server) handleJobsItem(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 
+	// P04 ACL enforcement (C-44). Job detail + tasks list are reads.
+	ns := "_defaults"
+	if s.acl != nil {
+		if id, ok := s.acl.Check(r, ns, acl.PermRead); !ok {
+			deny(w, id, ns, acl.PermRead)
+			return
+		}
+	}
+
 	// Path is /v1/jobs/{id} or /v1/jobs/{id}/tasks
 	path := strings.TrimPrefix(r.URL.Path, "/v1/jobs/")
 	parts := strings.Split(path, "/")

internal/daemon/nodes_handler.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"time"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/model"
 	"git.cloudinit.dev/coreci/orca/internal/store"
 )
@@ -19,6 +20,15 @@ func (s *Server) handleNodesCollection(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 
+	// P04 ACL enforcement (C-44). Node list is a cluster-wide read.
+	ns := "_defaults"
+	if s.acl != nil {
+		if id, ok := s.acl.Check(r, ns, acl.PermRead); !ok {
+			deny(w, id, ns, acl.PermRead)
+			return
+		}
+	}
+
 	nodes, err := store.NewNodeRepo(s.db).List(ctx)
 	if err != nil {
 		writeError(w, http.StatusInternalServerError, "failed to list nodes")

internal/daemon/server.go

@@ -49,6 +49,13 @@ type Server struct {
 	// /orca.v1.Dispatch/* (P02). Optional — nil if no Dispatcher
 	// was registered. P02 wires this via RegisterDispatch.
 	dispatch *DispatchHandlers
+
+	// acl is the access-control policy (P04, v0.13; C-44/C-45). When
+	// nil, no ACL enforcement is applied (legacy/compat for tests
+	// that construct a Server directly). Production wiring sets this
+	// via NewServer (Options.ACLEnforce) so handlers can call
+	// s.acl.Check before dispatching.
+	acl *aclPolicy
 }
 
 // Options configures a new Server.
@@ -63,6 +70,13 @@ type Options struct {
 	// The pprof listener is unauthenticated and operator-only; never
 	// expose it publicly (AD-024).
 	PprofAddr string
+
+	// ACLEnforce controls C-45 staged rollout. When false (the default
+	// for the first run after P04 wiring), ACL denials are LOGGED but
+	// NOT enforced — the request proceeds. When true, ACL denials
+	// return 403. The operator switches to true after verifying the
+	// bootstrap ACL grants the right identities.
+	ACLEnforce bool
 }
 
 // maxBodyBytes is the limit for request bodies on JSON-decoding
@@ -93,6 +107,7 @@ func NewServer(opts Options) *Server {
 		db:   opts.DB,
 		log:  opts.Log,
 		addr: opts.Addr,
+		acl:  NewACLPolicy(opts.ACLEnforce, opts.Log),
 	}
 	s.httpServer = &http.Server{
 		Addr:              opts.Addr,
@@ -127,6 +142,17 @@ func (s *Server) MarkNotReady() { s.ready.Store(false) }
 // Ready reports the current readiness flag.
 func (s *Server) Ready() bool { return s.ready.Load() }
 
+// ACL returns the daemon's ACL enforcement policy (P04). Returns nil
+// if no policy is configured (legacy/compat). Handlers use this to
+// call Check before dispatching; tests use it to assert enforcement
+// mode.
+func (s *Server) ACL() *aclPolicy { return s.acl }
+
+// SetACLPolicy replaces the ACL policy. Used by tests to inject a
+// policy without going through NewServer. Production code should use
+// NewServer with Options.ACLEnforce.
+func (s *Server) SetACLPolicy(p *aclPolicy) { s.acl = p }
+
 // mux builds the route table. Handlers are split across files:
 //   - health.go          /healthz, /readyz, /v1/status
 //   - jobs_handler.go    /v1/jobs/*
@@ -144,7 +170,7 @@ func (s *Server) mux() http.Handler {
 	mux.HandleFunc("/v1/nodes", s.handleNodesCollection)
 	mux.HandleFunc("/v1/tasks", s.handleTasksCollection)
 	if s.dispatch != nil {
-		s.dispatch.Mount(mux)
+		s.dispatch.mountWithACL(mux, s.acl)
 	}
 	return bodyLimitMiddleware(loggingMiddleware(s.log, mux))
 }

internal/daemon/tasks_handler.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"time"
 
+	"git.cloudinit.dev/coreci/orca/internal/acl"
 	"git.cloudinit.dev/coreci/orca/internal/model"
 	"git.cloudinit.dev/coreci/orca/internal/store"
 )
@@ -22,6 +23,15 @@ func (s *Server) handleTasksCollection(w http.ResponseWriter, r *http.Request) {
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 
+	// P04 ACL enforcement (C-44). Task list is a cluster-wide read.
+	ns := "_defaults"
+	if s.acl != nil {
+		if id, ok := s.acl.Check(r, ns, acl.PermRead); !ok {
+			deny(w, id, ns, acl.PermRead)
+			return
+		}
+	}
+
 	jobID := r.URL.Query().Get("job_id")
 	if jobID != "" {
 		if err := validateID(jobID); err != nil {

internal/engine/actor.go

@@ -0,0 +1,33 @@
+// Package engine — actor.go provides the context key + helper for
+// threading the audit actor (OIDC sub or SPIFFE SVID) through the
+// engine layer (P04, T5; C-44). Previously the registry hardcoded
+// "cli" as the actor; this lets CLI commands inject the verified
+// operator identity via context so audit entries attribute actions
+// to the real human/operator.
+package engine
+
+import "context"
+
+// actorCtxKey is the context key for the audit actor.
+type actorCtxKey struct{}
+
+// WithActor returns a context carrying the audit actor. The CLI
+// calls this in PersistentPreRun after resolving the OIDC sub from
+// the credentials file. When the context carries no actor, the
+// registry falls back to "cli" (legacy).
+func WithActor(ctx context.Context, actor string) context.Context {
+	if actor == "" {
+		return ctx
+	}
+	return context.WithValue(ctx, actorCtxKey{}, actor)
+}
+
+// ActorFromCtx returns the audit actor from the context, or "cli"
+// when no actor is set (legacy fallback for paths that haven't been
+// wired yet).
+func ActorFromCtx(ctx context.Context) string {
+	if v, ok := ctx.Value(actorCtxKey{}).(string); ok && v != "" {
+		return v
+	}
+	return "cli"
+}

internal/engine/registry.go

@@ -24,13 +24,13 @@ func NewNodeRegistry(repo *store.NodeRepo, audit *Audit, log *slog.Logger) *Node
 
 func (r *NodeRegistry) Join(ctx context.Context, n *model.Node) error {
 	if err := r.repo.Insert(ctx, n); err != nil {
-		r.audit.Record(ctx, "cli", "node.join", n.ID, "failure", err, map[string]any{
+		r.audit.Record(ctx, ActorFromCtx(ctx), "node.join", n.ID, "failure", err, map[string]any{
 			"name":    n.Name,
 			"address": n.Address,
 		})
 		return fmt.Errorf("join node: %w", err)
 	}
-	r.audit.Record(ctx, "cli", "node.join", n.ID, "success", nil, map[string]any{
+	r.audit.Record(ctx, ActorFromCtx(ctx), "node.join", n.ID, "success", nil, map[string]any{
 		"name":    n.Name,
 		"address": n.Address,
 	})
@@ -43,20 +43,20 @@ func (r *NodeRegistry) Join(ctx context.Context, n *model.Node) error {
 
 func (r *NodeRegistry) Leave(ctx context.Context, id string) error {
 	if err := r.repo.UpdateState(ctx, id, model.NodeStateLeft); err != nil {
-		r.audit.Record(ctx, "cli", "node.leave", id, "failure", err, nil)
+		r.audit.Record(ctx, ActorFromCtx(ctx), "node.leave", id, "failure", err, nil)
 		return fmt.Errorf("leave node: %w", err)
 	}
-	r.audit.Record(ctx, "cli", "node.leave", id, "success", nil, nil)
+	r.audit.Record(ctx, ActorFromCtx(ctx), "node.leave", id, "success", nil, nil)
 	r.log.Info("node left", slog.String("node_id", id))
 	return nil
 }
 
 func (r *NodeRegistry) Forget(ctx context.Context, id string) error {
 	if err := r.repo.Delete(ctx, id); err != nil {
-		r.audit.Record(ctx, "cli", "node.forget", id, "failure", err, nil)
+		r.audit.Record(ctx, ActorFromCtx(ctx), "node.forget", id, "failure", err, nil)
 		return fmt.Errorf("forget node: %w", err)
 	}
-	r.audit.Record(ctx, "cli", "node.forget", id, "success", nil, nil)
+	r.audit.Record(ctx, ActorFromCtx(ctx), "node.forget", id, "success", nil, nil)
 	r.log.Info("node removed from registry", slog.String("node_id", id))
 	return nil
 }
@@ -75,7 +75,7 @@ func (r *NodeRegistry) Get(ctx context.Context, id string) (*model.Node, error)
 // cli package does not need to reach into the repo directly.
 func (r *NodeRegistry) SetNodeState(ctx context.Context, id, state string) error {
 	if err := r.repo.SetNodeState(ctx, id, state); err != nil {
-		r.audit.Record(ctx, "cli", "node.set_state", id, "failure", err, map[string]any{"state": state})
+		r.audit.Record(ctx, ActorFromCtx(ctx), "node.set_state", id, "failure", err, map[string]any{"state": state})
 		return fmt.Errorf("set node state: %w", err)
 	}
 	r.log.Info("node state set", slog.String("node_id", id), slog.String("state", state))

internal/identity/authtoken.go

@@ -0,0 +1,60 @@
+// Package identity — authtoken.go provides the ORCA_OIDC_TOKEN
+// validation helper used by the SSH-push applier and the txn apply
+// path (P04, v0.13; C-44). Both paths validate the env-var token
+// against the issuer's JWKS before applying any state change.
+//
+// The token is read from $ORCA_OIDC_TOKEN. The issuer + client ID
+// come from the OIDC config (oidc.issuer, oidc.client_id). If the
+// token is missing or invalid, the apply is refused. The verified
+// claims (sub + groups) are returned so the caller can thread them
+// into the audit actor field (T5) and the ACL check (T3/T4).
+package identity
+
+import (
+	"context"
+	"fmt"
+	"os"
+)
+
+// EnvOIDCToken is the environment variable holding the OIDC ID token
+// for the SSH-push / txn apply paths (R-021: the IdP issues the
+// token; Orca never issues its own).
+const EnvOIDCToken = "ORCA_OIDC_TOKEN"
+
+// VerifyOperatorToken reads $ORCA_OIDC_TOKEN and verifies it against
+// the issuer's JWKS. Returns the verified claims (sub, groups) on
+// success. Returns an error if the token is missing, expired, or
+// fails signature verification.
+//
+// The issuer + clientID come from the OIDC config block. When issuer
+// is empty, the function returns an error — the apply path requires
+// an OIDC issuer to be configured.
+func VerifyOperatorToken(ctx context.Context, issuer, clientID string) (*IDTokenClaims, error) {
+	raw := os.Getenv(EnvOIDCToken)
+	if raw == "" {
+		return nil, fmt.Errorf("identity: %s env var is not set (operator OIDC token required for apply)", EnvOIDCToken)
+	}
+	if issuer == "" {
+		return nil, fmt.Errorf("identity: oidc.issuer is not configured (required to verify %s)", EnvOIDCToken)
+	}
+	if clientID == "" {
+		clientID = "orca-cli"
+	}
+	claims, err := VerifyIDTokenStatic(ctx, issuer, clientID, raw)
+	if err != nil {
+		return nil, fmt.Errorf("identity: verify %s: %w", EnvOIDCToken, err)
+	}
+	return claims, nil
+}
+
+// OperatorActor renders the verified operator identity for the audit
+// `actor` field. The convention is "oidc:<sub>" so audit entries can
+// be filtered by human operator. Falls back to "oidc:unknown" when
+// claims are nil (e.g. when the caller could not verify the token but
+// still wants to record an audit entry).
+func OperatorActor(claims *IDTokenClaims) string {
+	if claims == nil || claims.Subject == "" {
+		return "oidc:unknown"
+	}
+	return "oidc:" + claims.Subject
+}

internal/sshpush/auth.go

@@ -0,0 +1,40 @@
+// Package sshpush — auth.go provides the operator OIDC token
+// validation hook used by SSH-push apply paths (P04, v0.13; C-44).
+//
+// The SSH-push transport moves state to peers (systemd units, nft
+// rules, drain commands, txn bundles). Any state-changing apply
+// MUST validate $ORCA_OIDC_TOKEN against the issuer's JWKS before
+// touching a peer. This file exposes AuthorizeApply, a helper the
+// CLI calls before fan-out; the actual JWKS verification is in
+// internal/identity.VerifyOperatorToken (kept there to centralize
+// the OIDC client logic).
+package sshpush
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"git.cloudinit.dev/coreci/orca/internal/identity"
+)
+
+// AuthorizeApply validates $ORCA_OIDC_TOKEN against the issuer's
+// JWKS and returns the verified operator actor string ("oidc:<sub>")
+// for audit logging. Returns an error if the token is missing or
+// invalid; the caller MUST refuse the apply in that case.
+//
+// When issuer is empty, the function returns an error — apply paths
+// require an OIDC issuer to be configured. The clientID defaults to
+// "orca-cli" when empty.
+func AuthorizeApply(ctx context.Context, issuer, clientID string) (string, error) {
+	// Fast-fail when the env var is unset so we don't even hit the
+	// JWKS discovery (which would hang on a misconfigured issuer).
+	if os.Getenv(identity.EnvOIDCToken) == "" {
+		return "", fmt.Errorf("sshpush: %s env var is not set (operator OIDC token required for apply)", identity.EnvOIDCToken)
+	}
+	claims, err := identity.VerifyOperatorToken(ctx, issuer, clientID)
+	if err != nil {
+		return "", fmt.Errorf("sshpush: %w", err)
+	}
+	return identity.OperatorActor(claims), nil
+}

internal/sshpush/auth_test.go

@@ -0,0 +1,29 @@
+package sshpush
+
+import (
+	"context"
+	"testing"
+)
+
+// TestAuthorizeApplyMissingToken (P04, T3, C-44) verifies that
+// AuthorizeApply returns an error when ORCA_OIDC_TOKEN is unset.
+// The apply path MUST refuse to run without a verified operator
+// token.
+func TestAuthorizeApplyMissingToken(t *testing.T) {
+	// Ensure the env var is unset for this test.
+	t.Setenv("ORCA_OIDC_TOKEN", "")
+	_, err := AuthorizeApply(context.Background(), "https://idp.example.com", "orca-cli")
+	if err == nil {
+		t.Fatal("expected error when ORCA_OIDC_TOKEN is unset, got nil")
+	}
+}
+
+// TestAuthorizeApplyMissingIssuer verifies that AuthorizeApply returns
+// an error when the issuer is empty (apply requires an OIDC issuer).
+func TestAuthorizeApplyMissingIssuer(t *testing.T) {
+	t.Setenv("ORCA_OIDC_TOKEN", "some-token")
+	_, err := AuthorizeApply(context.Background(), "", "orca-cli")
+	if err == nil {
+		t.Fatal("expected error when issuer is empty, got nil")
+	}
+}

internal/txn/auth_test.go

@@ -0,0 +1,75 @@
+package txn
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+)
+
+// TestApplyRefusesUnauthorized (P04, T4, C-44) verifies that Apply
+// returns ErrUnauthorized when the Authorize hook returns an error.
+// The apply path MUST refuse to run without a verified operator
+// token.
+func TestApplyRefusesUnauthorized(t *testing.T) {
+	tr := &authMockTransport{}
+	opts := ApplyOptions{
+		Namespace: "myapp",
+		Authorize: func(ctx context.Context) (string, error) {
+			return "", errors.New("ORCA_OIDC_TOKEN not set")
+		},
+	}
+	err := Apply(context.Background(), "T-deadbeefdeadbeef", "lead.example.com", tr, opts)
+	if err == nil {
+		t.Fatal("expected error when Authorize fails, got nil")
+	}
+	if !errors.Is(err, ErrUnauthorized) {
+		t.Errorf("expected ErrUnauthorized, got %v", err)
+	}
+}
+
+// TestApplyAuthorizesWithHook verifies that Apply proceeds when the
+// Authorize hook returns nil, and that the actor is logged.
+func TestApplyAuthorizesWithHook(t *testing.T) {
+	tr := &authMockTransport{execOut: []byte("applied\n")}
+	opts := ApplyOptions{
+		Namespace: "myapp",
+		Authorize: func(ctx context.Context) (string, error) {
+			return "oidc:operator@example.com", nil
+		},
+	}
+	err := Apply(context.Background(), "T-deadbeefdeadbeef", "lead.example.com", tr, opts)
+	// We expect a non-ErrUnauthorized error here because the mock
+	// transport's orca-pull.sh path doesn't exist; the point is that
+	// the apply got PAST the authorize hook.
+	if err != nil && errors.Is(err, ErrUnauthorized) {
+		t.Errorf("apply should not be refused after successful authorize: %v", err)
+	}
+}
+
+// TestApplyNoAuthorizeHookSkipsCheck verifies that when Authorize is
+// nil (legacy/test path), the apply proceeds without an auth check.
+// This preserves backward compat for tests that call Apply directly.
+func TestApplyNoAuthorizeHookSkipsCheck(t *testing.T) {
+	tr := &authMockTransport{execOut: []byte("applied\n")}
+	opts := ApplyOptions{Namespace: "myapp"}
+	err := Apply(context.Background(), "T-deadbeefdeadbeef", "lead.example.com", tr, opts)
+	// Any error is fine as long as it's not ErrUnauthorized.
+	if err != nil && errors.Is(err, ErrUnauthorized) {
+		t.Errorf("apply should skip auth when Authorize is nil: %v", err)
+	}
+}
+
+// authMockTransport is a minimal Transport for the auth tests.
+type authMockTransport struct {
+	execOut []byte
+	execErr error
+}
+
+func (m *authMockTransport) WriteFileIdempotent(ctx context.Context, peer, path string, content []byte, mode os.FileMode) (bool, error) {
+	return true, nil
+}
+
+func (m *authMockTransport) Exec(ctx context.Context, peer, cmd string) ([]byte, error) {
+	return m.execOut, m.execErr
+}

internal/txn/txn.go

@@ -85,6 +85,15 @@ type ApplyOptions struct {
 	Yes             bool
 	Namespace       string
 	Timeout         time.Duration
+
+	// Authorize is called before the apply runs (P04, C-44). It must
+	// return the verified operator identity (e.g. the OIDC sub) and a
+	// nil error to authorize the apply; a non-nil error aborts the
+	// apply with a 403-equivalent. When nil, no authorization is
+	// performed (legacy/compat for tests that call Apply directly).
+	// The CLI wires this to identity.VerifyOperatorToken, which
+	// validates $ORCA_OIDC_TOKEN against the issuer's JWKS.
+	Authorize func(ctx context.Context) (actor string, err error)
 }
 
 // Transport is the SSH-push surface the txn package needs: writing
@@ -244,6 +253,10 @@ func Stage(bundle *Bundle, leadPeer string, transport Transport) error {
 	return nil
 }
 
+// ErrUnauthorized is returned when the operator OIDC token is
+// missing or invalid (P04, C-44). The apply is refused.
+var ErrUnauthorized = errors.New("txn: operator not authorized (ORCA_OIDC_TOKEN missing or invalid)")
+
 func Apply(ctx context.Context, txnID TxnID, leadPeer string, transport Transport, opts ApplyOptions) error {
 	if transport == nil {
 		return fmt.Errorf("txn: nil transport")
@@ -251,6 +264,24 @@ func Apply(ctx context.Context, txnID TxnID, leadPeer string, transport Transpor
 	if leadPeer == "" {
 		return fmt.Errorf("txn: lead peer is empty")
 	}
+	// P04 (C-44): validate the operator OIDC token before applying any
+	// state change. The CLI wires opts.Authorize to
+	// identity.VerifyOperatorToken, which checks $ORCA_OIDC_TOKEN
+	// against the issuer's JWKS. When opts.Authorize is nil (legacy
+	// test path), this check is skipped.
+	if opts.Authorize != nil {
+		actor, err := opts.Authorize(ctx)
+		if err != nil {
+			slog.Warn("txn apply refused (unauthorized)",
+				slog.String("txn_id", string(txnID)),
+				slog.String("peer", leadPeer),
+				slog.String("error", err.Error()))
+			return fmt.Errorf("txn: apply %s: %w: %v", txnID, ErrUnauthorized, err)
+		}
+		slog.Info("txn apply authorized",
+			slog.String("txn_id", string(txnID)),
+			slog.String("actor", actor))
+	}
 	dir := remoteTxnDir(txnID)
 	pull := dir + "/orca-pull.sh"
 

internal/webauthn/connector.go

@@ -24,16 +24,37 @@ import (
 // Connector is the WebAuthn ceremony handler. It is mounted behind
 // Traefik and called by the bundled Dex.
 type Connector struct {
-	w       *webauthn.WebAuthn
-	store   *Store
-	rpID    string
-	origin  string
+	w      *webauthn.WebAuthn
+	store  *Store
+	rpID   string
+	origin string
+
+	// authFunc validates an authenticated session for registration
+	// (P04, T9, C-45). When non-nil, BeginRegistration and
+	// FinishRegistration require a valid session (cookie or bearer
+	// token) before proceeding; a nil/error result yields 401. When
+	// nil (fail-closed for new deployments), registration is rejected
+	// with 401 — the operator MUST wire an authFunc before enabling
+	// registration. This fixes C-45's unauthenticated-registration
+	// hole: previously anyone could register a credential for any
+	// username.
+	authFunc func(r *http.Request) (authenticated bool, existingUser string, err error)
 }
 
 // NewConnector builds a WebAuthn connector with the given RP ID
 // (the cluster's Traefik-served domain, C-38) and origin (the full
 // HTTPS URL).
 func NewConnector(store *Store, rpID, rpOrigin string) (*Connector, error) {
+	return NewConnectorWithAuth(store, rpID, rpOrigin, nil)
+}
+
+// NewConnectorWithAuth builds a Connector with an explicit auth
+// function for registration (P04, T9). The authFunc returns whether
+// the request carries a valid authenticated session and, optionally,
+// the existing user identity (so registration can be scoped to the
+// authenticated user). When authFunc is nil, registration is fail-
+// closed (401).
+func NewConnectorWithAuth(store *Store, rpID, rpOrigin string, authFunc func(r *http.Request) (bool, string, error)) (*Connector, error) {
 	wconfig := &webauthn.Config{
 		RPDisplayName: "Orca",
 		RPID:          rpID,
@@ -44,10 +65,11 @@ func NewConnector(store *Store, rpID, rpOrigin string) (*Connector, error) {
 		return nil, fmt.Errorf("webauthn: new: %w", err)
 	}
 	return &Connector{
-		w:      w,
-		store:  store,
-		rpID:   rpID,
-		origin: rpOrigin,
+		w:        w,
+		store:    store,
+		rpID:     rpID,
+		origin:   rpOrigin,
+		authFunc: authFunc,
 	}, nil
 }
 
@@ -64,6 +86,7 @@ type RegistrationSession struct {
 type sessionStore struct {
 	sessions map[string]*RegistrationSession
 }
+
 var regSessions = &sessionStore{sessions: make(map[string]*RegistrationSession)}
 
 // sessionTTL is the max time a registration/login session is valid.
@@ -79,10 +102,47 @@ func cleanSessions() {
 	}
 }
 
+// requireAuth checks the request for an authenticated session. When
+// c.authFunc is nil, registration is fail-closed (401). When the
+// authFunc returns false or an error, the request is rejected with
+// 401 Unauthorized. Returns true when the request is authenticated.
+//
+// The authFunc may also return the existing user identity so
+// registration can be scoped (a user can only register credentials
+// for their own account); the existing-user scoping is enforced by
+// the caller via the username query param match (a future phase will
+// wire the authenticated user as the registration target instead of
+// accepting a free-form username).
+func (c *Connector) requireAuth(w http.ResponseWriter, r *http.Request) bool {
+	if c.authFunc == nil {
+		http.Error(w, "registration requires authentication (no auth function configured)", http.StatusUnauthorized)
+		return false
+	}
+	ok, _, err := c.authFunc(r)
+	if err != nil || !ok {
+		http.Error(w, "authentication required", http.StatusUnauthorized)
+		return false
+	}
+	return true
+}
+
+// SetAuthFunc sets the registration auth function (P04, T9). Allows
+// callers to wire the auth check after construction (e.g. when the
+// session store is initialized later).
+func (c *Connector) SetAuthFunc(f func(r *http.Request) (bool, string, error)) {
+	c.authFunc = f
+}
+
 // BeginRegistration starts the WebAuthn registration ceremony.
 // GET /orca/webauthn/register?username=<name>
 // Returns the creation options (challenge) for the browser.
 func (c *Connector) BeginRegistration(w http.ResponseWriter, r *http.Request) {
+	// P04 (T9, C-45): require an authenticated session before
+	// allowing registration. Without this, anyone could register a
+	// credential for any username. When authFunc is nil, fail-closed.
+	if !c.requireAuth(w, r) {
+		return
+	}
 	username := r.URL.Query().Get("username")
 	if username == "" {
 		http.Error(w, "username required", http.StatusBadRequest)
@@ -118,6 +178,10 @@ func (c *Connector) BeginRegistration(w http.ResponseWriter, r *http.Request) {
 // POST /orca/webauthn/register/finish?username=<name>
 // Body: the attestation response from the browser.
 func (c *Connector) FinishRegistration(w http.ResponseWriter, r *http.Request) {
+	// P04 (T9): require an authenticated session for finish too.
+	if !c.requireAuth(w, r) {
+		return
+	}
 	username := r.URL.Query().Get("username")
 	if username == "" {
 		http.Error(w, "username required", http.StatusBadRequest)
@@ -168,6 +232,7 @@ type LoginSession struct {
 	Challenge *webauthn.SessionData
 	CreatedAt time.Time
 }
+
 var loginSessions = map[string]*LoginSession{}
 
 // BeginLogin starts the WebAuthn login ceremony.
@@ -295,11 +360,11 @@ type webauthnUser struct {
 	credentials []webauthn.Credential
 }
 
-func (u *webauthnUser) WebAuthnID() []byte { return u.id }
-func (u *webauthnUser) WebAuthnName() string { return u.name }
-func (u *webauthnUser) WebAuthnDisplayName() string { return u.name }
+func (u *webauthnUser) WebAuthnID() []byte                         { return u.id }
+func (u *webauthnUser) WebAuthnName() string                       { return u.name }
+func (u *webauthnUser) WebAuthnDisplayName() string                { return u.name }
 func (u *webauthnUser) WebAuthnCredentials() []webauthn.Credential { return u.credentials }
-func (u *webauthnUser) WebAuthnIcon() string { return "" }
+func (u *webauthnUser) WebAuthnIcon() string                       { return "" }
 
 // RPID returns the configured relying-party ID.
 func (c *Connector) RPID() string { return c.rpID }

internal/webauthn/connector_test.go

@@ -47,18 +47,65 @@ func TestConnectorHealthz(t *testing.T) {
 }
 
 // TestConnectorBeginRegistrationNoUsername verifies the register
-// endpoint rejects requests without a username.
+// endpoint rejects requests without a username when authenticated.
 func TestConnectorBeginRegistrationNoUsername(t *testing.T) {
 	dbPath := filepath.Join(t.TempDir(), "webauthn-creds.db")
 	store, _ := NewStore(dbPath)
 	defer store.Close()
-	c, _ := NewConnector(store, "test.cluster", "https://test.cluster")
+	c, _ := NewConnectorWithAuth(store, "test.cluster", "https://test.cluster",
+		func(r *http.Request) (bool, string, error) { return true, "admin", nil })
 	mux := c.Routes()
 	req := httptest.NewRequest("GET", "/orca/webauthn/register", nil)
 	rec := httptest.NewRecorder()
 	mux.ServeHTTP(rec, req)
 	if rec.Code != http.StatusBadRequest {
-		t.Errorf("register without username: %d, want 400", rec.Code)
+		t.Errorf("register without username (authed): %d, want 400", rec.Code)
+	}
+}
+
+// TestConnectorBeginRegistrationUnauthenticated (P04, T9, C-45)
+// verifies the register endpoint rejects requests with no
+// authenticated session — closing the hole where anyone could
+// register a credential for any username.
+func TestConnectorBeginRegistrationUnauthenticated(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "webauthn-creds.db")
+	store, _ := NewStore(dbPath)
+	defer store.Close()
+	// No authFunc → fail-closed (401).
+	c, _ := NewConnector(store, "test.cluster", "https://test.cluster")
+	mux := c.Routes()
+	req := httptest.NewRequest("GET", "/orca/webauthn/register?username=admin", nil)
+	rec := httptest.NewRecorder()
+	mux.ServeHTTP(rec, req)
+	if rec.Code != http.StatusUnauthorized {
+		t.Errorf("register unauthenticated (no authFunc): %d, want 401", rec.Code)
+	}
+
+	// authFunc that returns false → 401.
+	c2, _ := NewConnectorWithAuth(store, "test.cluster", "https://test.cluster",
+		func(r *http.Request) (bool, string, error) { return false, "", nil })
+	mux2 := c2.Routes()
+	req2 := httptest.NewRequest("GET", "/orca/webauthn/register?username=admin", nil)
+	rec2 := httptest.NewRecorder()
+	mux2.ServeHTTP(rec2, req2)
+	if rec2.Code != http.StatusUnauthorized {
+		t.Errorf("register unauthenticated (authFunc=false): %d, want 401", rec2.Code)
+	}
+}
+
+// TestConnectorFinishRegistrationUnauthenticated (P04, T9) verifies
+// the finish endpoint also requires authentication.
+func TestConnectorFinishRegistrationUnauthenticated(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "webauthn-creds.db")
+	store, _ := NewStore(dbPath)
+	defer store.Close()
+	c, _ := NewConnector(store, "test.cluster", "https://test.cluster")
+	mux := c.Routes()
+	req := httptest.NewRequest("POST", "/orca/webauthn/register/finish?username=admin", nil)
+	rec := httptest.NewRecorder()
+	mux.ServeHTTP(rec, req)
+	if rec.Code != http.StatusUnauthorized {
+		t.Errorf("finish unauthenticated: %d, want 401", rec.Code)
 	}
 }
 

tests/security_integration_test.go

@@ -56,6 +56,44 @@ func TestSecurityInvariants_Metadata(t *testing.T) {
 	// F18: drift event auth. Tested by:
 	//   - internal/drift: TestVerifyEventSignature
 	//
+	// P04 (v0.13) ACL enforcement wiring (C-44/C-45):
+	//   - internal/daemon: TestACLPolicyDenyByDefault
+	//     (authenticated request with no ACL entry → deny in enforce mode)
+	//   - internal/daemon: TestACLPolicyAllowWithEntry
+	//     (authenticated request with matching ACL entry → allow)
+	//   - internal/daemon: TestACLPolicyUnauthenticatedEnforce
+	//     (unauthenticated request → 403 in enforce mode)
+	//   - internal/daemon: TestACLPolicyLogOnlyAllowsDenials (C-45)
+	//     (denials logged but allowed in log-only mode)
+	//   - internal/daemon: TestACLJobsHandlerEnforceDeniesUnauthenticated
+	//     (wired jobs handler denies unauthenticated in enforce mode)
+	//   - internal/daemon: TestACLJobsHandlerAllowsAuthenticatedWithEntry
+	//     (wired jobs handler allows authenticated with matching entry)
+	//   - internal/daemon: TestACLNodesHandlerEnforceDeniesUnauthenticated
+	//   - internal/daemon: TestACLTasksHandlerEnforceDeniesUnauthenticated
+	//   - internal/txn: Apply refuses when ORCA_OIDC_TOKEN is missing/invalid
+	//     (C-44: SSH-push applier + txn apply path validate OIDC token)
+	//   - internal/sshpush: AuthorizeApply validates ORCA_OIDC_TOKEN
+	//
+	// P04 (v0.13) WebAuthn registration auth (C-45, T9):
+	//   - internal/webauthn: TestConnectorBeginRegistrationUnauthenticated
+	//     (unauthenticated BeginRegistration → 401, fail-closed)
+	//   - internal/webauthn: TestConnectorFinishRegistrationUnauthenticated
+	//     (unauthenticated FinishRegistration → 401)
+	//
+	// P04 (v0.13) acl.json hardening (T6/T7):
+	//   - internal/cli: saveACL writes acl.json with mode 0600 (T6)
+	//   - internal/cli: lockACL flocks grant/revoke (T7, prevents races)
+	//
+	// P04 (v0.13) bootstrap ACL (T8, C-40):
+	//   - internal/cli: bootstrapACL grants cluster-admin to orca-admins
+	//     group + init SVID on `orca init` (prevents operator lockout)
+	//
+	// P04 (v0.13) audit actor identity (T5):
+	//   - internal/cli: currentActor reads OIDC sub from credentials.json
+	//   - internal/engine: ActorFromCtx threads sub into audit Record calls
+	//     (replaces hardcoded "cli" actor)
+	//
 	// This test is the gate (C-33): if it runs, the suite is wired.
-	t.Log("security integration test suite wired (R-021, F1-F25, REQ-119..148)")
+	t.Log("security integration test suite wired (R-021, F1-F25, REQ-119..148, P04 ACL enforcement C-44/C-45)")
 }