Add rootful runtime qualification packet

This commit is contained in:
rcourtman
2026-09-01 22:38:39 +01:00
parent 813c2b6837
commit 584cef81a1
9 changed files with 2579 additions and 2 deletions
+3
View File
@@ -271,6 +271,9 @@ scripts/release_control/*
!scripts/release_control/secure_runtime_attestation_v6_test.py
!scripts/release_control/secure_runtime_attestation_v7.py
!scripts/release_control/secure_runtime_attestation_v7_test.py
!scripts/release_control/secure_runtime_rootful_attestation_v1.py
!scripts/release_control/secure_runtime_rootful_attestation_v1_test.py
!scripts/release_control/secure_runtime_rootful_source_manifest_v1.json
!scripts/release_control/secure_runtime_rootless_attestation_v1.py
!scripts/release_control/secure_runtime_rootless_attestation_v1_test.py
!scripts/release_control/secure_runtime_rootless_source_manifest_v1.json
@@ -1760,7 +1760,9 @@ lifecycle conditions, but compact empty-state spacing, icon treatment, and text
hierarchy must compose frontend-primitives' `EmptyState` `variant="panel"`
instead of lifecycle-local centered icon/text shells.
32. `scripts/run-secure-runtime-rootless-qualification.sh` shared with `deployment-installability`: the rootless runtime qualification wrapper is both an agent lifecycle provider-boundary proof entry point and a deployment installability proof harness.
32. `scripts/run-secure-runtime-rootful-qualification.sh` shared with `deployment-installability`: the rootful runtime qualification wrapper is both an agent lifecycle typed-helper provider-boundary proof entry point and a deployment installability proof harness.
33. `scripts/run-secure-runtime-rootless-qualification.sh` shared with `deployment-installability`: the rootless runtime qualification wrapper is both an agent lifecycle provider-boundary proof entry point and a deployment installability proof harness.
## Extension Points
@@ -7392,6 +7394,29 @@ published-release provenance, default-profile authorization, rootless
action/update qualification, or independent security review and cannot change
the product default.
### Rootful container qualification has its own typed-helper packet
`scripts/run-secure-runtime-rootful-qualification.sh` is the destructive,
explicit entrypoint for real rootful Docker and Podman proof on two distinct
disposable Ubuntu/systemd hosts. It never mounts a host daemon socket and runs
the outer containers without a default route. The packet binds exact clean Go
artifacts and governed source hashes, then records fresh install, legacy
migration with authority reduction, collector and helper restart continuity,
helper loss without an authoritative empty replacement, exact recovery,
bounded helper-operation failure, ordinary collector-update preservation,
authority isolation, and cleanup for both runtimes. Rootful telemetry is
intentionally summary-only: the collector remains unable to open the
root-owned daemon socket and gains neither container actions nor updates.
`scripts/release_control/secure_runtime_rootful_attestation_v1.py` accepts only
the exact Docker-then-Podman ten-scenario contract, distinct host and daemon
identities, canonical rootful socket paths, stable semantic inventory, causal
report streams, bounded failure and recovery, and immutable artifact/source
bindings. Adding this packet does not itself qualify the runtime. Only a
retained passing receipt and attestation may support a local opt-in claim, and
neither can establish published-release provenance, default-profile
authorization, appliance support, or independent security review.
### Command and durable typed dispatch are context-honest and canceled when abandoned
The agent command transport now refuses to dispatch work its caller has
@@ -322,6 +322,20 @@ Failed runtime tests must retain the disposable systemd journal and container
log before nonce-bound cleanup; absence of a receipt is not a replacement for
the causal failure record.
Live rootful-runtime proof is also separated from the schema-v7 systemd
packet. The explicit
`scripts/run-secure-runtime-rootful-qualification.sh` wrapper creates distinct
network-isolated Ubuntu/systemd hosts for real rootful Docker and Podman and
never mounts a host daemon socket. Its standalone schema-v1 receipt binds the
exact qualification, collector, helper, installer, source manifest, canonical
root-owned socket, summary inventory, migration, restart, loss, recovery,
bounded operation, update-preservation, authority-isolation, and cleanup
evidence before the independent validator can emit a local artifact-bound
self-attestation. All Go artifacts require exact clean VCS metadata and all
retained outputs use a private mode-`0700` host directory. Checking in the
tooling or passing its contract tests is not live qualification, published
release proof, default-profile authorization, or independent review.
Release builds and archives carry both helper and runner binaries for the five
Linux targets (`amd64`, `arm64`, `armv7`, `armv6`, and `386`) with checksum,
Ed25519, and SSH signature sidecars. Exact archive/container-context validation
@@ -899,7 +913,9 @@ artifact-selection behaviour.
`NoNewPrivileges=false` exception above: it restores only the privilege
`lxc-attach` needs and leaves the rest of the sandbox intact.
9. `scripts/run-secure-runtime-rootless-qualification.sh` shared with `agent-lifecycle`: the rootless runtime qualification wrapper is both an agent lifecycle provider-boundary proof entry point and a deployment installability proof harness.
9. `scripts/run-secure-runtime-rootful-qualification.sh` shared with `agent-lifecycle`: the rootful runtime qualification wrapper is both an agent lifecycle typed-helper provider-boundary proof entry point and a deployment installability proof harness.
10. `scripts/run-secure-runtime-rootless-qualification.sh` shared with `agent-lifecycle`: the rootless runtime qualification wrapper is both an agent lifecycle provider-boundary proof entry point and a deployment installability proof harness.
## Extension Points
@@ -1287,6 +1287,14 @@
"deployment-installability"
]
},
{
"path": "scripts/run-secure-runtime-rootful-qualification.sh",
"rationale": "the rootful runtime qualification wrapper is both an agent lifecycle typed-helper provider-boundary proof entry point and a deployment installability proof harness",
"subsystems": [
"agent-lifecycle",
"deployment-installability"
]
},
{
"path": "scripts/run-secure-runtime-rootless-qualification.sh",
"rationale": "the rootless runtime qualification wrapper is both an agent lifecycle provider-boundary proof entry point and a deployment installability proof harness",
@@ -1397,6 +1405,7 @@
"scripts/installtests/windows_agent_lifecycle.ps1",
"scripts/installtests/windowslifecycleserver/main.go",
"scripts/run-native-pve-action-qualification.sh",
"scripts/run-secure-runtime-rootful-qualification.sh",
"scripts/run-secure-runtime-rootless-qualification.sh",
"tests/integration/tests/84-docker-restart-real-lab-artifact.spec.ts"
],
@@ -1481,6 +1490,20 @@
"scripts/installtests/native_pve_action_qualification_test.go"
]
},
{
"id": "rootful-container-runtime-qualification",
"label": "disposable rootful Docker and Podman typed-helper lifecycle qualification proof",
"match_prefixes": [],
"match_files": [
"scripts/run-secure-runtime-rootful-qualification.sh"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"scripts/installtests/secure_runtime_rootful_qualification_test.go",
"scripts/release_control/secure_runtime_rootful_attestation_v1_test.py"
]
},
{
"id": "rootless-container-runtime-qualification",
"label": "disposable rootless Docker and Podman lifecycle qualification proof",
@@ -4513,6 +4536,8 @@
"scripts/release_control/resolve_release_promotion.py",
"scripts/release_control/secure_runtime_attestation_v6.py",
"scripts/release_control/secure_runtime_attestation_v7.py",
"scripts/release_control/secure_runtime_rootful_attestation_v1.py",
"scripts/release_control/secure_runtime_rootful_source_manifest_v1.json",
"scripts/release_control/secure_runtime_rootless_attestation_v1.py",
"scripts/release_control/secure_runtime_rootless_source_manifest_v1.json",
"scripts/release_control/secure_runtime_source_manifest_v6.json",
@@ -4522,6 +4547,7 @@
"scripts/release_update_key.go",
"scripts/run-release-backend-tests.sh",
"scripts/run-release-preflight.sh",
"scripts/run-secure-runtime-rootful-qualification.sh",
"scripts/run-secure-runtime-rootless-qualification.sh",
"scripts/run_cloud_public_signup_smoke.sh",
"scripts/run_demo_public_browser_smoke.sh",
@@ -4757,6 +4783,22 @@
"scripts/release_control/secure_runtime_attestation_v7_test.py"
]
},
{
"id": "rootful-runtime-qualification-tooling",
"label": "standalone rootful Docker and Podman typed-helper qualification packet proof",
"match_prefixes": [],
"match_files": [
"scripts/release_control/secure_runtime_rootful_attestation_v1.py",
"scripts/release_control/secure_runtime_rootful_source_manifest_v1.json",
"scripts/run-secure-runtime-rootful-qualification.sh"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"scripts/installtests/secure_runtime_rootful_qualification_test.go",
"scripts/release_control/secure_runtime_rootful_attestation_v1_test.py"
]
},
{
"id": "rootless-runtime-qualification-tooling",
"label": "standalone rootless Docker and Podman qualification packet proof",
@@ -0,0 +1,988 @@
//go:build !windows
package installtests
// This file is a standalone rootful-container qualification packet. The live
// test is opt-in and must run only inside the disposable Ubuntu/systemd hosts
// created by scripts/run-secure-runtime-rootful-qualification.sh.
import (
"context"
"debug/buildinfo"
"encoding/json"
"errors"
"fmt"
"net"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"runtime"
"slices"
"strconv"
"strings"
"syscall"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/agenthelper"
agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker"
agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
)
const (
rootfulQualOptIn = "PULSE_SECURE_RUNTIME_ROOTFUL_QUALIFICATION"
rootfulQualOptInValue = "disposable-v1"
rootfulQualMarker = "/etc/pulse-secure-runtime-rootful-qualification"
rootfulQualReceiptPath = "/opt/pulse/result/rootful-receipt.json"
rootfulQualResultDir = "/opt/pulse/result"
rootfulQualFixture = "pulse-rootful-qualification-fixture:v1"
rootfulQualRunningName = "pulse-rootful-running"
rootfulQualExitedName = "pulse-rootful-exited"
rootfulQualHelperSock = "/run/pulse-agent/helper.sock"
)
var rootfulQualScenarioOrder = []string{
"fresh_install",
"legacy_migration",
"collector_restart",
"helper_restart",
"helper_loss",
"helper_recovery",
"operation_bounds",
"update_preservation",
"authority_isolation",
"cleanup",
}
type rootfulQualReceipt struct {
SchemaVersion int `json:"schema_version"`
Kind string `json:"kind"`
Result string `json:"result"`
SourceCommit string `json:"source_commit"`
StartedAt string `json:"started_at"`
CompletedAt string `json:"completed_at"`
SourceHashes map[string]string `json:"source_hashes"`
Artifacts rootlessQualArtifacts `json:"artifacts"`
Runs []rootfulQualRun `json:"runs"`
}
type rootfulQualRun struct {
Host rootlessQualHost `json:"host"`
Runtime rootfulQualRuntime `json:"runtime"`
Scenarios []rootlessQualScenario `json:"scenarios"`
}
type rootfulQualRuntime struct {
Runtime string `json:"runtime"`
RuntimeVersion string `json:"runtime_version"`
DaemonID string `json:"daemon_id"`
DaemonRootless bool `json:"daemon_rootless"`
SocketPath string `json:"socket_path"`
SocketUID int `json:"socket_uid"`
SocketGID int `json:"socket_gid"`
SocketMode string `json:"socket_mode"`
SocketType string `json:"socket_type"`
SocketSymlink bool `json:"socket_symlink"`
}
type rootfulQualDaemon struct {
runtime string
unit string
socket string
dataRoot string
fixture string
}
func TestSecureRuntimeRootfulQualification(t *testing.T) {
if os.Getenv(rootfulQualOptIn) != rootfulQualOptInValue {
t.Skip("run through scripts/run-secure-runtime-rootful-qualification.sh inside its disposable systemd container")
}
runtimeKind := strings.TrimSpace(os.Getenv("PULSE_ROOTFUL_RUNTIME"))
receiptPath := strings.TrimSpace(os.Getenv("PULSE_ROOTFUL_RECEIPT"))
rootfulQualRequireDisposableHost(t, runtimeKind, receiptPath)
collector := secureRuntimeReadArtifact(t, "PULSE_SECURE_RUNTIME_COLLECTOR")
collectorSignature := secureRuntimeReadSignature(t, "PULSE_SECURE_RUNTIME_COLLECTOR_SIGNATURE")
helper := secureRuntimeReadArtifact(t, "PULSE_SECURE_RUNTIME_HELPER")
collectorVersion := secureRuntimeArtifactVersion(t, "PULSE_SECURE_RUNTIME_COLLECTOR")
installerPath := strings.TrimSpace(os.Getenv("PULSE_SECURE_RUNTIME_INSTALLER"))
if !filepath.IsAbs(installerPath) {
t.Fatalf("PULSE_SECURE_RUNTIME_INSTALLER must be absolute: %q", installerPath)
}
started := time.Now().UTC()
fixture := newSecureRuntimeLabFixture(collector, collectorSignature, helper, nil, collectorVersion)
defer fixture.actionServer.Shutdown()
server := httptestNewServer(t, fixture)
defer server.Close()
collectorCredential := secureRuntimeLabToken
daemon := rootfulQualDaemonFor(runtimeKind)
defer rootlessQualBestEffortStop(daemon.unit, rootfulQualHungUnit(runtimeKind))
rootfulQualPrepareFixture(t, daemon)
rootfulQualStartDaemon(t, daemon)
rootfulQualCreateFixtures(t, daemon)
baseline := rootfulQualRuntimeBaseline(t, daemon)
if baseline.Count != 2 {
t.Fatalf("rootful %s baseline count = %d, want 2", runtimeKind, baseline.Count)
}
runtimeVersion := rootfulQualRuntimeVersion(t, daemon)
daemonID := rootfulQualDaemonID(t, daemon)
socketUID, socketGID, socketMode := rootfulQualSocketIdentity(t, daemon.socket)
if socketUID != 0 {
t.Fatalf("rootful runtime socket UID = %d, want 0", socketUID)
}
var scenarios []rootlessQualScenario
appendScenario := func(name string, began time.Time, report *agentsdocker.Report, evidence map[string]any) {
scenario := rootlessQualScenario{
Name: name, Result: "passed", StartedAt: began.Format(time.RFC3339Nano),
CompletedAt: time.Now().UTC().Format(time.RFC3339Nano), Evidence: evidence,
}
if report != nil {
stream, sequence, ok := agentshost.ParseReportSequenceID(report.SequenceID)
if !ok {
t.Fatalf("scenario %s received invalid sequence ID %q", name, report.SequenceID)
}
scenario.ReportStreamID = &stream
scenario.ReportSequence = &sequence
}
scenarios = append(scenarios, scenario)
}
freshStarted := time.Now().UTC()
secureRuntimeRunInstaller(t, installerPath, server.URL,
"--least-privilege", "--enable-privileged-helper", "--enable-docker")
fresh := rootfulQualWaitSummary(t, fixture, freshStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
secureRuntimeAssertSafeProfile(t)
secureRuntimeAssertHelperProtocol(t)
rootfulQualAssertCollectorSocketDenied(t, daemon.socket)
freshPID := secureRuntimeCollectorMainPID(t)
freshHelperPID, _ := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
appendScenario("fresh_install", freshStarted, &fresh.Report,
rootfulQualSummaryEvidence(freshPID, freshHelperPID, daemonID, fresh.Report))
rootlessQualUninstallPulse(t, installerPath, server.URL, collectorCredential)
rootlessQualAssertPulseRemoved(t)
if registered, revoked, uninstalls := fixture.collectorLifecycleSnapshot(); registered || !revoked || uninstalls != 1 {
t.Fatalf("fresh collector uninstall was not durably modeled: registered=%t revoked=%t uninstalls=%d", registered, revoked, uninstalls)
}
fixture.replaceCollectorCredential(secureRuntimeLabTokenV2, secureRuntimeCollectorBindingV2)
collectorCredential = secureRuntimeLabTokenV2
migrationStarted := time.Now().UTC()
secureRuntimeRunInstallerWithCollectorCredential(t, installerPath, server.URL, collectorCredential,
"--enable-commands", "--command-authority", "command-capable", "--enable-docker")
legacy := rootlessQualWaitReport(t, fixture, migrationStarted, 75*time.Second, func(report agentsdocker.Report) bool {
return rootlessQualComplete(report) && report.Host.CollectionMode == "" && report.Host.Runtime == runtimeKind && rootlessQualSemanticDigest(report) == baseline.SemanticDigest
})
secureRuntimeAssertRootCommandProfile(t)
legacyPID := secureRuntimeCollectorMainPID(t)
applyStarted := time.Now().UTC()
secureRuntimeRunInstallerWithCollectorCredential(t, installerPath, server.URL, collectorCredential, "--safe-profile-apply")
migrated := rootfulQualWaitSummary(t, fixture, applyStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
secureRuntimeAssertSafeProfile(t)
secureRuntimeAssertHelperProtocol(t)
rootfulQualAssertCollectorSocketDenied(t, daemon.socket)
migratedPID := secureRuntimeCollectorMainPID(t)
if legacyPID == migratedPID || fixture.authorityReductionCount() < 1 || rootlessQualSemanticDigest(legacy.Report) != rootlessQualSemanticDigest(migrated.Report) {
t.Fatalf("rootful migration did not replace/reduce the collector with summary parity")
}
migratedHelperPID, _ := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
migrationEvidence := rootfulQualSummaryEvidence(migratedPID, migratedHelperPID, daemonID, migrated.Report)
migrationEvidence["legacy_profile"] = "root-command-capable"
migrationEvidence["target_profile"] = "typed-helper-monitoring-only"
migrationEvidence["authority_reduced"] = true
migrationEvidence["legacy_collector_pid"] = legacyPID
appendScenario("legacy_migration", migrationStarted, &migrated.Report, migrationEvidence)
collectorRestartStarted := time.Now().UTC()
previousStream, _, _ := agentshost.ParseReportSequenceID(migrated.Report.SequenceID)
secureRuntimeCommand(t, 20*time.Second, "systemctl", "restart", "pulse-agent.service")
collectorPID := secureRuntimeCollectorMainPID(t)
collectorRestart := rootfulQualWaitSummary(t, fixture, collectorRestartStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
collectorStream, _, _ := agentshost.ParseReportSequenceID(collectorRestart.Report.SequenceID)
if collectorPID == migratedPID || collectorStream == previousStream {
t.Fatalf("collector restart did not replace PID/report stream: pid=%d/%d stream=%s/%s", migratedPID, collectorPID, previousStream, collectorStream)
}
helperPID, _ := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
collectorRestartEvidence := rootfulQualSummaryEvidence(collectorPID, helperPID, daemonID, collectorRestart.Report)
collectorRestartEvidence["previous_collector_pid"] = migratedPID
collectorRestartEvidence["previous_report_stream_id"] = previousStream
appendScenario("collector_restart", collectorRestartStarted, &collectorRestart.Report, collectorRestartEvidence)
helperPIDBefore, helperInvocationBefore := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
secureRuntimeCommand(t, 20*time.Second, "systemctl", "restart", "pulse-agent-helper.service")
helperRestartStarted := time.Now().UTC()
helperPIDAfter, helperInvocationAfter := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
if helperPIDBefore == helperPIDAfter || helperInvocationBefore == helperInvocationAfter {
t.Fatalf("helper restart did not replace exact service identity")
}
helperRestart := rootfulQualWaitSummary(t, fixture, helperRestartStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
helperRestartEvidence := rootfulQualSummaryEvidence(collectorPID, helperPIDAfter, daemonID, helperRestart.Report)
helperRestartEvidence["previous_helper_pid"] = helperPIDBefore
helperRestartEvidence["previous_helper_invocation_id"] = helperInvocationBefore
helperRestartEvidence["helper_invocation_id"] = helperInvocationAfter
appendScenario("helper_restart", helperRestartStarted, &helperRestart.Report, helperRestartEvidence)
lossStarted := time.Now().UTC()
secureRuntimeCommand(t, 20*time.Second, "systemctl", "stop", "pulse-agent-helper.socket", "pulse-agent-helper.service")
loss := rootfulQualWaitStatusOnly(t, fixture, lossStarted, runtimeKind, 75*time.Second)
lossStream, lossSequence, _ := agentshost.ParseReportSequenceID(loss.Report.SequenceID)
if len(loss.Report.Containers) != 0 || secureRuntimeCollectorMainPID(t) != collectorPID {
t.Fatal("helper loss emitted an authoritative empty inventory or replaced the collector")
}
appendScenario("helper_loss", lossStarted, &loss.Report, map[string]any{
"collector_pid": collectorPID, "previous_helper_pid": helperPIDAfter,
"collection_mode": "typed-helper-unavailable-status-only", "helper_available": false,
"status_only": true, "inventory_complete": false, "inventory_present": false,
"authoritative_inventory_replacement": false,
"previous_authoritative_inventory_count": baseline.Count,
"previous_authoritative_semantic_sha256": baseline.SemanticDigest,
"operation_status": "degraded", "operation": agenthelper.OperationContainerInventory,
"container_updates_enabled": false, "container_actions_enabled": false, "direct_socket_access": false,
})
recoveryStarted := time.Now().UTC()
secureRuntimeCommand(t, 20*time.Second, "systemctl", "start", "pulse-agent-helper.socket")
recovered := rootfulQualWaitSummary(t, fixture, recoveryStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
secureRuntimeAssertHelperProtocol(t)
helperRecoveryPID, _ := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
recoveryStream, recoverySequence, _ := agentshost.ParseReportSequenceID(recovered.Report.SequenceID)
if recoveryStream != lossStream || recoverySequence <= lossSequence {
t.Fatalf("helper recovery did not advance the same report stream")
}
recoveryEvidence := rootfulQualSummaryEvidence(collectorPID, helperRecoveryPID, daemonID, recovered.Report)
recoveryEvidence["previous_helper_pid"] = helperPIDAfter
recoveryEvidence["previous_status_report_sequence"] = lossSequence
appendScenario("helper_recovery", recoveryStarted, &recovered.Report, recoveryEvidence)
boundStarted := time.Now().UTC()
rootfulQualStopDaemon(t, daemon)
rootfulQualStartHungDaemon(t, daemon)
probeElapsed := rootfulQualRunBoundProbe(t, 2*time.Second)
probeCompletedAt := time.Now().UTC()
boundStatus := rootfulQualWaitStatusOnly(t, fixture, probeCompletedAt, runtimeKind, 45*time.Second)
boundStatusStream, boundStatusSequence, _ := agentshost.ParseReportSequenceID(boundStatus.Report.SequenceID)
rootlessQualBestEffortStop(rootfulQualHungUnit(runtimeKind))
_ = os.Remove(daemon.socket)
rootfulQualStartDaemon(t, daemon)
boundRecoveryStarted := time.Now().UTC()
boundRecovery := rootfulQualWaitSummary(t, fixture, boundRecoveryStarted, runtimeKind, baseline.SemanticDigest, 75*time.Second)
boundRecoveryStream, boundRecoverySequence, _ := agentshost.ParseReportSequenceID(boundRecovery.Report.SequenceID)
if boundRecoveryStream != boundStatusStream || boundRecoverySequence <= boundStatusSequence {
t.Fatalf("bounded operation recovery did not advance the same report stream")
}
appendScenario("operation_bounds", boundStarted, &boundRecovery.Report, map[string]any{
"collector_pid": collectorPID, "helper_pid": helperRecoveryPID,
"operation": agenthelper.OperationContainerInventory, "failure_class": "bounded-timeout",
"collection_mode": agentsdocker.CollectionModeTypedHelperSummary,
"inventory_complete": true, "full_fields_present": false, "stats_present": false,
"secondary_structure_sha256": "",
"deadline_ms": 2000, "elapsed_ms": probeElapsed.Milliseconds(), "bounded_failure_observed": true,
"status_only_report_sequence": boundStatusSequence, "recovery_report_sequence": boundRecoverySequence,
"previous_authoritative_inventory_count": baseline.Count,
"previous_authoritative_semantic_sha256": baseline.SemanticDigest,
"recovery_inventory_count": baseline.Count, "recovery_semantic_sha256": rootlessQualSemanticDigest(boundRecovery.Report),
"authoritative_empty_replacement": false, "collector_alive": true, "helper_alive": true,
"container_updates_enabled": false, "container_actions_enabled": false, "direct_socket_access": false,
})
updateStarted := time.Now().UTC()
preUpdatePID := secureRuntimeCollectorMainPID(t)
preUpdateStream, _, _ := agentshost.ParseReportSequenceID(boundRecovery.Report.SequenceID)
secureRuntimeRunInstallerWithCollectorCredential(t, installerPath, server.URL, collectorCredential, "--update")
postUpdatePID := secureRuntimeCollectorMainPID(t)
updated := rootfulQualWaitSummary(t, fixture, updateStarted, runtimeKind, baseline.SemanticDigest, 90*time.Second)
postUpdateStream, _, _ := agentshost.ParseReportSequenceID(updated.Report.SequenceID)
if postUpdatePID == preUpdatePID || postUpdateStream == preUpdateStream {
t.Fatalf("ordinary update did not restart the safe collector: pid=%d/%d stream=%s/%s", preUpdatePID, postUpdatePID, preUpdateStream, postUpdateStream)
}
postUpdateHelperPID, _ := rootfulQualUnitIdentity(t, "pulse-agent-helper.service")
if postUpdateHelperPID != helperRecoveryPID {
t.Fatalf("ordinary collector update replaced the independent helper process: pid=%d/%d", helperRecoveryPID, postUpdateHelperPID)
}
updateEvidence := rootfulQualSummaryEvidence(postUpdatePID, postUpdateHelperPID, daemonID, updated.Report)
updateEvidence["previous_collector_pid"] = preUpdatePID
updateEvidence["previous_helper_pid"] = helperRecoveryPID
updateEvidence["previous_report_stream_id"] = preUpdateStream
updateEvidence["update_applied"] = true
updateEvidence["collector_binary_sha256"] = secureRuntimeHash(secureRuntimeReadFile(t, "/usr/local/bin/pulse-agent"))
updateEvidence["helper_binary_sha256"] = secureRuntimeHash(secureRuntimeReadFile(t, "/usr/local/lib/pulse-agent/pulse-agent-helper"))
appendScenario("update_preservation", updateStarted, &updated.Report, updateEvidence)
authorityStarted := time.Now().UTC()
collectorUID := rootlessQualUID(t, "pulse-agent")
groups := strings.Fields(rootlessQualCommand(t, 10*time.Second, "id", "-nG", "pulse-agent"))
if slices.Contains(groups, "docker") || slices.Contains(groups, "podman") {
t.Fatalf("safe collector retained a rootful daemon group: %v", groups)
}
rootfulQualAssertCollectorSocketDenied(t, daemon.socket)
helperNetworkDenied := rootlessQualAssertHelperNetworkDenied(t)
commandSessionPresent := fixture.actionServer.IsAgentConnectedForOrganization(secureRuntimeLabOrgID, secureRuntimeLabAgentID)
if !helperNetworkDenied || commandSessionPresent || secureRuntimeCollectorHasArgument("--enable-commands") || secureRuntimeCollectorProcessUID(t) != collectorUID {
t.Fatal("rootful authority isolation did not remain exact")
}
appendScenario("authority_isolation", authorityStarted, nil, map[string]any{
"collector_pid": postUpdatePID, "collector_uid": collectorUID, "effective_uid": collectorUID,
"effective_root": false, "safe_profile_enabled": true, "commands_enabled": false,
"privileged_helper_enabled": true, "reduction_request_observed": true,
"collector_command_transport_present": false, "collector_command_session_present": false,
"container_actions_enabled": false, "container_updates_enabled": false,
"rootful_socket_access": false, "direct_socket_access": false, "helper_network_access": false,
})
cleanupStarted := time.Now().UTC()
rootlessQualUninstallPulse(t, installerPath, server.URL, collectorCredential)
rootfulQualRemoveFixtures(t, daemon)
rootfulQualStopDaemon(t, daemon)
rootfulQualRemoveRuntimeState(t, daemon)
rootlessQualAssertPulseRemoved(t)
if registered, revoked, uninstalls := fixture.collectorLifecycleSnapshot(); registered || !revoked || uninstalls != 2 {
t.Fatalf("final collector uninstall was not durably modeled: registered=%t revoked=%t uninstalls=%d", registered, revoked, uninstalls)
}
if _, err := os.Lstat(daemon.socket); !errors.Is(err, os.ErrNotExist) {
t.Fatalf("cleanup left rootful runtime socket %s: %v", daemon.socket, err)
}
stateClean := rootfulQualRuntimeStateClean(daemon)
if !stateClean {
t.Fatal("cleanup left rootful runtime state")
}
appendScenario("cleanup", cleanupStarted, nil, map[string]any{
"collector_stopped": true, "helper_stopped": true, "runtime_stopped": true,
"socket_absent": true, "fixtures_removed": true, "state_clean": true,
})
receipt := rootfulQualReceipt{
SchemaVersion: 1, Kind: "pulse-secure-runtime-rootful-qualification", Result: "passed",
SourceCommit: strings.TrimSpace(os.Getenv("PULSE_ROOTFUL_SOURCE_COMMIT")),
StartedAt: started.Format(time.RFC3339Nano), CompletedAt: time.Now().UTC().Format(time.RFC3339Nano),
SourceHashes: rootfulQualSourceHashes(t), Artifacts: rootfulQualArtifactIdentities(t, installerPath),
Runs: []rootfulQualRun{{
Host: rootlessQualHost{
MachineID: strings.TrimSpace(string(rootlessQualReadFile(t, "/etc/machine-id"))), Architecture: runtime.GOARCH,
Kernel: rootlessQualCommand(t, 10*time.Second, "uname", "-srvmo"),
SystemdVersion: strings.SplitN(rootlessQualCommand(t, 10*time.Second, "systemctl", "--version"), "\n", 2)[0],
},
Runtime: rootfulQualRuntime{
Runtime: runtimeKind, RuntimeVersion: runtimeVersion, DaemonID: daemonID, DaemonRootless: false,
SocketPath: daemon.socket, SocketUID: socketUID, SocketGID: socketGID, SocketMode: socketMode,
SocketType: "unix", SocketSymlink: false,
},
Scenarios: scenarios,
}},
}
if err := rootfulQualValidateReceipt(receipt, 1); err != nil {
t.Fatalf("generated rootful receipt failed validation: %v", err)
}
rootlessQualWriteJSON(t, receiptPath, receipt)
}
// TestSecureRuntimeRootfulBoundProbe is invoked as the installed collector UID
// by the live qualification test. It proves the real helper and provider honor
// a caller-supplied bounded deadline against an accepted but unresponsive
// rootful daemon connection.
func TestSecureRuntimeRootfulBoundProbe(t *testing.T) {
if os.Getenv("PULSE_ROOTFUL_BOUND_PROBE") != "1" {
t.Skip("internal rootful qualification subprocess")
}
deadlineMillis, err := strconv.Atoi(os.Getenv("PULSE_ROOTFUL_BOUND_DEADLINE_MS"))
if err != nil || deadlineMillis < 1 {
t.Fatalf("invalid bound-probe deadline: %v", err)
}
deadline := time.Duration(deadlineMillis) * time.Millisecond
client, err := agenthelper.NewClient(agenthelper.ClientConfig{SocketPath: rootfulQualHelperSock, MaxDeadline: deadline})
if err != nil {
t.Fatal(err)
}
var response agenthelper.ContainerInventoryResult
_, err = client.Call(context.Background(), agenthelper.OperationContainerInventory, agenthelper.OperationVersion1, deadline, struct{}{}, &response)
var remote *agenthelper.RemoteError
var networkError net.Error
typedDeadline := errors.As(err, &remote) && remote.Code == agenthelper.ErrorDeadlineExceeded
localDeadline := errors.As(err, &networkError) && networkError.Timeout()
if !typedDeadline && !localDeadline {
t.Fatalf("bounded helper operation error = %T %v", err, err)
}
fmt.Println("ROOTFUL_BOUND_RESULT=deadline_exceeded")
}
func httptestNewServer(t *testing.T, handler *secureRuntimeLabFixture) *httptest.Server {
t.Helper()
return httptest.NewServer(handler)
}
func rootfulQualRequireDisposableHost(t *testing.T, runtimeKind, receiptPath string) {
t.Helper()
if os.Geteuid() != 0 || (runtimeKind != "docker" && runtimeKind != "podman") {
t.Fatalf("qualification requires root and PULSE_ROOTFUL_RUNTIME=docker|podman")
}
marker, err := os.ReadFile(rootfulQualMarker)
if err != nil || strings.TrimSpace(string(marker)) != rootfulQualOptInValue {
t.Fatalf("disposable marker is absent or invalid: %v", err)
}
if receiptPath != rootfulQualReceiptPath {
t.Fatalf("PULSE_ROOTFUL_RECEIPT must use %q: %q", rootfulQualReceiptPath, receiptPath)
}
info, err := os.Lstat(rootfulQualResultDir)
if err != nil {
t.Fatal(err)
}
stat, ok := info.Sys().(*syscall.Stat_t)
if !ok || !info.IsDir() || info.Mode()&os.ModeSymlink != 0 || info.Mode().Perm() != 0o700 || stat.Uid != 0 {
t.Fatalf("rootful result directory must be root-owned mode 0700: mode=%s stat=%#v", info.Mode(), info.Sys())
}
osRelease := string(rootlessQualReadFile(t, "/etc/os-release"))
if !strings.Contains(osRelease, "VERSION_ID=\"24.04\"") && !strings.Contains(osRelease, "VERSION_ID=24.04") {
t.Fatal("qualification host is not Ubuntu 24.04")
}
if _, err := os.Stat("/run/systemd/system"); err != nil {
t.Fatalf("qualification host is not booted under systemd: %v", err)
}
if rootlessQualHasDefaultRoute(string(rootlessQualReadFile(t, "/proc/net/route"))) {
t.Fatal("qualification workload must run with outer-container networking disabled")
}
}
func rootfulQualDaemonFor(runtimeKind string) rootfulQualDaemon {
if runtimeKind == "docker" {
return rootfulQualDaemon{runtime: runtimeKind, unit: "pulse-rootful-docker", socket: "/var/run/docker.sock", dataRoot: "/var/lib/pulse-rootful-docker", fixture: "/opt/pulse/rootful-fixture"}
}
return rootfulQualDaemon{runtime: runtimeKind, unit: "pulse-rootful-podman", socket: "/run/podman/podman.sock", dataRoot: "/var/lib/containers", fixture: "/opt/pulse/rootful-fixture"}
}
func rootfulQualPrepareFixture(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
rootlessQualCommand(t, 10*time.Second, "install", "-d", "-o", "root", "-g", "root", "-m", "0700", daemon.fixture)
containerfile := "FROM scratch\nCOPY busybox /busybox\nENTRYPOINT [\"/busybox\"]\n"
if err := os.WriteFile(filepath.Join(daemon.fixture, "Containerfile"), []byte(containerfile), 0o600); err != nil {
t.Fatal(err)
}
busybox, err := os.ReadFile("/bin/busybox")
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(daemon.fixture, "busybox"), busybox, 0o700); err != nil {
t.Fatal(err)
}
}
func rootfulQualStartDaemon(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
rootlessQualBestEffortStop(daemon.unit)
_ = os.Remove(daemon.socket)
rootlessQualCommand(t, 10*time.Second, "install", "-d", "-o", "root", "-g", "root", "-m", "0755", filepath.Dir(daemon.socket))
if daemon.runtime == "docker" {
rootlessQualCommand(t, 20*time.Second, "systemd-run", "--quiet", "--collect", "--unit", daemon.unit, "--property=Type=exec", "--",
"/usr/bin/dockerd", "--host=unix://"+daemon.socket, "--data-root="+daemon.dataRoot,
"--exec-root=/run/pulse-rootful-docker", "--pidfile=/run/pulse-rootful-docker.pid", "--storage-driver=vfs", "--iptables=false", "--bridge=none")
} else {
rootlessQualCommand(t, 20*time.Second, "systemd-run", "--quiet", "--collect", "--unit", daemon.unit, "--property=Type=exec", "--",
"/usr/bin/podman", "system", "service", "--time=0", "unix://"+daemon.socket)
}
rootlessQualWaitSocket(t, daemon.socket)
rootlessQualCommand(t, 10*time.Second, "chmod", "0660", daemon.socket)
rootfulQualRuntimeCommand(t, daemon, 30*time.Second, "info")
}
func rootfulQualStopDaemon(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
rootlessQualStopUnit(t, daemon.unit)
_ = os.Remove(daemon.socket)
}
func rootfulQualRuntimeCommand(t *testing.T, daemon rootfulQualDaemon, timeout time.Duration, args ...string) string {
t.Helper()
if daemon.runtime == "docker" {
return rootlessQualCommand(t, timeout, "docker", append([]string{"--host", "unix://" + daemon.socket}, args...)...)
}
return rootlessQualCommand(t, timeout, "podman", append([]string{"--url", "unix://" + daemon.socket}, args...)...)
}
func rootfulQualCreateFixtures(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
rootfulQualRuntimeCommand(t, daemon, 2*time.Minute, "build", "--network=none", "-t", rootfulQualFixture, "-f", filepath.Join(daemon.fixture, "Containerfile"), daemon.fixture)
rootfulQualRuntimeCommand(t, daemon, 30*time.Second, "run", "-d", "--restart=always", "--name", rootfulQualRunningName, rootfulQualFixture, "sleep", "3600")
if out, err := rootfulQualRuntimeCommandError(daemon, 30*time.Second, "run", "--name", rootfulQualExitedName, rootfulQualFixture, "true"); err != nil {
t.Fatalf("create exited fixture: %v\n%s", err, out)
}
}
func rootfulQualRemoveFixtures(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
for _, name := range []string{rootfulQualRunningName, rootfulQualExitedName} {
_, _ = rootfulQualRuntimeCommandError(daemon, 30*time.Second, "rm", "-f", name)
}
_, _ = rootfulQualRuntimeCommandError(daemon, 30*time.Second, "rmi", "-f", rootfulQualFixture)
}
func rootfulQualRuntimeCommandError(daemon rootfulQualDaemon, timeout time.Duration, args ...string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
name := "podman"
prefix := []string{"--url", "unix://" + daemon.socket}
if daemon.runtime == "docker" {
name = "docker"
prefix = []string{"--host", "unix://" + daemon.socket}
}
output, err := exec.CommandContext(ctx, name, append(prefix, args...)...).CombinedOutput()
return strings.TrimSpace(string(output)), err
}
func rootfulQualRuntimeBaseline(t *testing.T, daemon rootfulQualDaemon) rootlessQualBaseline {
t.Helper()
return rootlessQualBaselineFromPSOutput(rootfulQualRuntimeCommand(t, daemon, 30*time.Second, "ps", "-a", "--format", "{{.Names}}|{{.Image}}|{{.State}}"))
}
func rootfulQualRuntimeVersion(t *testing.T, daemon rootfulQualDaemon) string {
t.Helper()
return rootfulQualRuntimeCommand(t, daemon, 30*time.Second, "version", "--format", "{{.Server.Version}}")
}
func rootfulQualDaemonID(t *testing.T, daemon rootfulQualDaemon) string {
t.Helper()
d := rootlessQualDaemon{runtime: daemon.runtime, rootfulSock: daemon.socket}
return rootlessQualDaemonID(t, d, false)
}
func rootfulQualSocketIdentity(t *testing.T, path string) (int, int, string) {
t.Helper()
info, err := os.Lstat(path)
if err != nil {
t.Fatal(err)
}
if info.Mode()&os.ModeSocket == 0 || info.Mode()&os.ModeSymlink != 0 || info.Mode().Perm() != 0o660 {
t.Fatalf("unsafe rootful runtime socket %s mode=%s", path, info.Mode())
}
stat, ok := info.Sys().(*syscall.Stat_t)
if !ok {
t.Fatalf("socket %s lacks Unix stat identity", path)
}
return int(stat.Uid), int(stat.Gid), fmt.Sprintf("%04o", info.Mode().Perm())
}
func rootfulQualWaitSummary(t *testing.T, fixture *secureRuntimeLabFixture, after time.Time, runtimeKind, digest string, timeout time.Duration) secureRuntimeDockerReport {
t.Helper()
report := rootlessQualWaitReport(t, fixture, after, timeout, func(report agentsdocker.Report) bool {
return rootlessQualComplete(report) && report.InventoryComplete != nil && *report.InventoryComplete &&
report.Host.CollectionMode == agentsdocker.CollectionModeTypedHelperSummary && report.Host.Runtime == runtimeKind &&
rootlessQualSemanticDigest(report) == digest && len(report.Containers) > 0
})
rootlessQualAssertHelperSummaryOnly(t, report.Report, runtimeKind)
return report
}
func rootfulQualWaitStatusOnly(t *testing.T, fixture *secureRuntimeLabFixture, after time.Time, runtimeKind string, timeout time.Duration) secureRuntimeDockerReport {
t.Helper()
return rootlessQualWaitReport(t, fixture, after, timeout, func(report agentsdocker.Report) bool {
return report.InventoryComplete != nil && !*report.InventoryComplete && report.Host.Runtime == runtimeKind &&
report.Host.CollectionMode == agentsdocker.CollectionModeTypedHelperSummary && len(report.Containers) == 0 &&
secureRuntimeDockerHelperModuleState(report) == "degraded"
})
}
func rootfulQualSummaryEvidence(collectorPID, helperPID int, daemonID string, report agentsdocker.Report) map[string]any {
digest := rootlessQualDigestReport(report)
return map[string]any{
"collector_pid": collectorPID, "helper_pid": helperPID,
"collection_mode": agentsdocker.CollectionModeTypedHelperSummary,
"inventory_complete": true, "inventory_count": digest.Count,
"semantic_sha256": digest.SemanticDigest, "full_fields_present": false,
"stats_present": false, "secondary_structure_sha256": "",
"container_updates_enabled": false, "container_actions_enabled": false,
"direct_socket_access": false, "daemon_id": daemonID, "daemon_rootless": false,
}
}
func rootfulQualUnitIdentity(t *testing.T, unit string) (int, string) {
t.Helper()
pidText := rootlessQualCommand(t, 10*time.Second, "systemctl", "show", unit, "--property=MainPID", "--value")
pid, err := strconv.Atoi(pidText)
if err != nil || pid <= 0 {
t.Fatalf("invalid %s MainPID %q", unit, pidText)
}
invocation := rootlessQualCommand(t, 10*time.Second, "systemctl", "show", unit, "--property=InvocationID", "--value")
if len(invocation) != 32 {
t.Fatalf("invalid %s InvocationID %q", unit, invocation)
}
return pid, invocation
}
func rootfulQualAssertCollectorSocketDenied(t *testing.T, socket string) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
command := exec.CommandContext(ctx, "runuser", "-u", "pulse-agent", "--", "curl", "-fsS", "--max-time", "2", "--unix-socket", socket, "http://runtime/_ping")
if output, err := command.CombinedOutput(); err == nil {
t.Fatalf("safe collector unexpectedly reached rootful socket %s: %s", socket, strings.TrimSpace(string(output)))
}
}
func rootfulQualHungUnit(runtimeKind string) string { return "pulse-rootful-" + runtimeKind + "-hung" }
func rootfulQualStartHungDaemon(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
scriptPath := filepath.Join(daemon.fixture, "hung-runtime.py")
script := `import os, socket, threading, time, sys
path = sys.argv[1]
try: os.unlink(path)
except FileNotFoundError: pass
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.bind(path)
os.chmod(path, 0o660)
s.listen(64)
def hold(c):
try: time.sleep(120)
finally: c.close()
while True:
c, _ = s.accept()
threading.Thread(target=hold, args=(c,), daemon=True).start()
`
if err := os.WriteFile(scriptPath, []byte(script), 0o700); err != nil {
t.Fatal(err)
}
rootlessQualCommand(t, 20*time.Second, "systemd-run", "--quiet", "--collect", "--unit", rootfulQualHungUnit(daemon.runtime), "--property=Type=exec", "--",
"/usr/bin/python3", scriptPath, daemon.socket)
rootlessQualWaitSocket(t, daemon.socket)
}
func rootfulQualRunBoundProbe(t *testing.T, deadline time.Duration) time.Duration {
t.Helper()
executable, err := os.Executable()
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), deadline+3*time.Second)
defer cancel()
started := time.Now()
cmd := exec.CommandContext(ctx, "runuser", "-u", "pulse-agent", "--", "env",
"PULSE_ROOTFUL_BOUND_PROBE=1", fmt.Sprintf("PULSE_ROOTFUL_BOUND_DEADLINE_MS=%d", deadline.Milliseconds()),
executable, "-test.run", "^TestSecureRuntimeRootfulBoundProbe$", "-test.count=1", "-test.v", "-test.timeout=10s")
output, err := cmd.CombinedOutput()
elapsed := time.Since(started)
if err != nil || !strings.Contains(string(output), "ROOTFUL_BOUND_RESULT=deadline_exceeded") {
t.Fatalf("bounded helper probe failed after %s: %v\n%s", elapsed, err, output)
}
if elapsed < deadline/2 || elapsed > deadline+time.Second {
t.Fatalf("bounded helper probe elapsed %s outside expected interval", elapsed)
}
return elapsed
}
func rootfulQualRemoveRuntimeState(t *testing.T, daemon rootfulQualDaemon) {
t.Helper()
roots := []string{daemon.dataRoot, "/run/pulse-rootful-docker"}
deadline := time.Now().Add(30 * time.Second)
for {
mountInfo := string(rootlessQualReadFile(t, "/proc/self/mountinfo"))
remaining, err := rootlessQualMountPointsBelow(mountInfo, roots)
if err != nil {
t.Fatalf("inspect disposable runtime mounts: %v", err)
}
if len(remaining) == 0 {
break
}
if time.Now().After(deadline) {
t.Fatalf("runtime mounts remain after service shutdown: %q", remaining)
}
time.Sleep(100 * time.Millisecond)
}
for _, path := range []string{daemon.dataRoot, "/run/pulse-rootful-docker", daemon.fixture} {
if err := os.RemoveAll(path); err != nil {
t.Fatalf("remove disposable runtime path %s: %v", path, err)
}
}
}
func rootfulQualRuntimeStateClean(daemon rootfulQualDaemon) bool {
for _, path := range []string{daemon.socket, daemon.dataRoot, "/run/pulse-rootful-docker", daemon.fixture} {
if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) {
return false
}
}
return true
}
func rootfulQualArtifactIdentities(t *testing.T, installerPath string) rootlessQualArtifacts {
t.Helper()
executable, err := os.Executable()
if err != nil {
t.Fatal(err)
}
return rootlessQualArtifacts{
QualificationTest: rootfulQualGoArtifact(t, executable, "dockeragent.test"),
Collector: rootfulQualGoArtifact(t, strings.TrimSpace(os.Getenv("PULSE_SECURE_RUNTIME_COLLECTOR")), "pulse-agent"),
Helper: rootfulQualGoArtifact(t, strings.TrimSpace(os.Getenv("PULSE_SECURE_RUNTIME_HELPER")), "pulse-agent-helper"),
Installer: rootlessQualInstallerArtifact{PathBasename: filepath.Base(installerPath), SHA256: secureRuntimeHash(rootlessQualReadFile(t, installerPath))},
}
}
func rootfulQualGoArtifact(t *testing.T, path, basename string) rootlessQualArtifact {
t.Helper()
info, err := buildinfo.ReadFile(path)
if err != nil {
t.Fatalf("read Go build metadata for %s: %v", path, err)
}
artifact := rootlessQualArtifact{PathBasename: filepath.Base(path), SHA256: secureRuntimeHash(rootlessQualReadFile(t, path)), Package: info.Path, GoVersion: info.GoVersion}
for _, setting := range info.Settings {
switch setting.Key {
case "vcs.revision":
artifact.VCSRevision = setting.Value
case "vcs.modified":
artifact.VCSModified = setting.Value == "true"
}
}
wantCommit := strings.TrimSpace(os.Getenv("PULSE_ROOTFUL_SOURCE_COMMIT"))
if artifact.PathBasename != basename || artifact.Package == "" || artifact.VCSRevision != wantCommit || artifact.VCSModified {
t.Fatalf("rootful qualification artifact is not an exact clean source build: %+v", artifact)
}
return artifact
}
func rootfulQualSourceHashes(t *testing.T) map[string]string {
t.Helper()
path := strings.TrimSpace(os.Getenv("PULSE_ROOTFUL_SOURCE_HASHES"))
if !filepath.IsAbs(path) {
t.Fatalf("PULSE_ROOTFUL_SOURCE_HASHES must be absolute: %q", path)
}
var hashes map[string]string
if err := json.Unmarshal(rootlessQualReadFile(t, path), &hashes); err != nil {
t.Fatal(err)
}
if len(hashes) == 0 {
t.Fatal("rootful source hash map is empty")
}
return hashes
}
func rootfulQualValidateReceipt(receipt rootfulQualReceipt, expectedRuns int) error {
if receipt.SchemaVersion != 1 || receipt.Kind != "pulse-secure-runtime-rootful-qualification" || receipt.Result != "passed" {
return errors.New("invalid rootful qualification identity")
}
if len(receipt.SourceCommit) != 40 || receipt.StartedAt == "" || receipt.CompletedAt == "" || len(receipt.SourceHashes) == 0 || len(receipt.Runs) != expectedRuns {
return errors.New("incomplete rootful qualification envelope")
}
for _, run := range receipt.Runs {
if run.Runtime.Runtime != "docker" && run.Runtime.Runtime != "podman" {
return fmt.Errorf("unsupported runtime %q", run.Runtime.Runtime)
}
if run.Runtime.DaemonRootless || run.Runtime.DaemonID == "" || run.Runtime.SocketUID != 0 || run.Runtime.SocketPath == "" || run.Runtime.SocketMode != "0660" || run.Runtime.SocketType != "unix" || run.Runtime.SocketSymlink {
return errors.New("invalid rootful runtime identity")
}
if run.Host.MachineID == "" || len(run.Scenarios) != len(rootfulQualScenarioOrder) {
return errors.New("incomplete rootful host/scenario evidence")
}
for index, scenario := range run.Scenarios {
if scenario.Name != rootfulQualScenarioOrder[index] || scenario.Result != "passed" || scenario.StartedAt == "" || scenario.CompletedAt == "" || scenario.Evidence == nil {
return fmt.Errorf("invalid scenario %d", index)
}
isReporting := scenario.Name != "authority_isolation" && scenario.Name != "cleanup"
if isReporting != (scenario.ReportSequence != nil && scenario.ReportStreamID != nil) {
return fmt.Errorf("scenario %s report binding mismatch", scenario.Name)
}
}
}
return nil
}
func TestRootfulQualificationReceiptContract(t *testing.T) {
stream := "stream"
sequence := uint64(1)
scenarios := make([]rootlessQualScenario, 0, len(rootfulQualScenarioOrder))
for _, name := range rootfulQualScenarioOrder {
scenario := rootlessQualScenario{Name: name, Result: "passed", StartedAt: time.Now().UTC().Format(time.RFC3339Nano), CompletedAt: time.Now().UTC().Format(time.RFC3339Nano), Evidence: map[string]any{"observed": true}}
if name != "authority_isolation" && name != "cleanup" {
scenario.ReportStreamID = &stream
scenario.ReportSequence = &sequence
}
scenarios = append(scenarios, scenario)
}
receipt := rootfulQualReceipt{
SchemaVersion: 1, Kind: "pulse-secure-runtime-rootful-qualification", Result: "passed",
SourceCommit: strings.Repeat("a", 40), StartedAt: time.Now().UTC().Format(time.RFC3339Nano), CompletedAt: time.Now().UTC().Format(time.RFC3339Nano),
SourceHashes: map[string]string{"go.mod": strings.Repeat("b", 64)},
Runs: []rootfulQualRun{{Host: rootlessQualHost{MachineID: strings.Repeat("1", 32)}, Runtime: rootfulQualRuntime{Runtime: "docker", RuntimeVersion: "1", DaemonID: "daemon", SocketPath: "/var/run/docker.sock", SocketUID: 0, SocketGID: 999, SocketMode: "0660", SocketType: "unix"}, Scenarios: scenarios}},
}
if err := rootfulQualValidateReceipt(receipt, 1); err != nil {
t.Fatal(err)
}
receipt.Runs[0].Scenarios[3], receipt.Runs[0].Scenarios[4] = receipt.Runs[0].Scenarios[4], receipt.Runs[0].Scenarios[3]
if err := rootfulQualValidateReceipt(receipt, 1); err == nil {
t.Fatal("validator accepted reordered rootful scenarios")
}
}
func TestRootfulQualificationGoSchemaPassesPythonValidator(t *testing.T) {
commit := strings.Repeat("a", 40)
digest := strings.Repeat("b", 64)
started := time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)
receipt := rootfulQualReceipt{
SchemaVersion: 1, Kind: "pulse-secure-runtime-rootful-qualification", Result: "passed",
SourceCommit: commit, StartedAt: started.Format(time.RFC3339Nano),
CompletedAt: started.Add(2 * time.Minute).Format(time.RFC3339Nano),
SourceHashes: map[string]string{"internal/agenthelper/container_inventory.go": digest, "scripts/install.sh": digest},
Artifacts: rootlessQualArtifacts{
QualificationTest: rootlessQualArtifact{PathBasename: "dockeragent.test", SHA256: digest, Package: "github.com/rcourtman/pulse-go-rewrite/scripts/installtests.test", GoVersion: "go1.25.0", VCSRevision: commit},
Collector: rootlessQualArtifact{PathBasename: "pulse-agent", SHA256: digest, Package: "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent", GoVersion: "go1.25.0", VCSRevision: commit},
Helper: rootlessQualArtifact{PathBasename: "pulse-agent-helper", SHA256: digest, Package: "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent-helper", GoVersion: "go1.25.0", VCSRevision: commit},
Installer: rootlessQualInstallerArtifact{PathBasename: "install.sh", SHA256: digest},
},
}
for index, runtimeKind := range []string{"docker", "podman"} {
receipt.Runs = append(receipt.Runs, rootfulQualValidatorFixtureRun(runtimeKind, index, started.Add(time.Duration(index)*30*time.Second), digest))
}
path := filepath.Join(t.TempDir(), "receipt.json")
rootlessQualWriteJSON(t, path, receipt)
validator := repoFile("scripts", "release_control", "secure_runtime_rootful_attestation_v1.py")
program := `import importlib.util, pathlib, sys
path=pathlib.Path(sys.argv[1]).resolve()
sys.path.insert(0, str(path.parent))
spec=importlib.util.spec_from_file_location("validator", path)
module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
module.parse_receipt_bytes(pathlib.Path(sys.argv[2]).read_bytes())
`
cmd := exec.Command("python3", "-I", "-c", program, validator, path)
if output, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("Go rootful receipt schema failed the real Python validator: %v\n%s", err, output)
}
}
func rootfulQualValidatorFixtureRun(runtimeKind string, index int, began time.Time, digest string) rootfulQualRun {
base := index * 100
daemonID := runtimeKind + "-daemon"
socketPath := "/var/run/docker.sock"
socketGID := 999
if runtimeKind == "podman" {
socketPath = "/run/podman/podman.sock"
socketGID = 0
}
stream := func(value string) *string { return &value }
sequence := func(value uint64) *uint64 { return &value }
makeScenario := func(offset int, name string, streamID *string, seq *uint64, evidence map[string]any) rootlessQualScenario {
start := began.Add(time.Duration(offset) * time.Second)
return rootlessQualScenario{
Name: name, Result: "passed", StartedAt: start.Format(time.RFC3339Nano),
CompletedAt: start.Add(time.Second).Format(time.RFC3339Nano),
ReportStreamID: streamID, ReportSequence: seq, Evidence: evidence,
}
}
summary := func(collectorPID, helperPID int) map[string]any {
return map[string]any{
"collector_pid": collectorPID, "helper_pid": helperPID,
"collection_mode": "typed-helper-summary", "inventory_complete": true,
"inventory_count": 2, "semantic_sha256": digest, "full_fields_present": false,
"stats_present": false, "secondary_structure_sha256": "",
"container_updates_enabled": false, "container_actions_enabled": false,
"direct_socket_access": false, "daemon_id": daemonID, "daemon_rootless": false,
}
}
fresh := summary(base+100, base+200)
migration := summary(base+110, base+210)
migration["legacy_profile"] = "root-command-capable"
migration["target_profile"] = "typed-helper-monitoring-only"
migration["authority_reduced"] = true
migration["legacy_collector_pid"] = base + 90
collectorRestart := summary(base+120, base+210)
collectorRestart["previous_collector_pid"] = base + 110
collectorRestart["previous_report_stream_id"] = runtimeKind + "-migration"
helperRestart := summary(base+120, base+220)
helperRestart["previous_helper_pid"] = base + 210
helperRestart["previous_helper_invocation_id"] = runtimeKind + "-helper-old"
helperRestart["helper_invocation_id"] = runtimeKind + "-helper-new"
loss := map[string]any{
"collector_pid": base + 120, "previous_helper_pid": base + 220,
"collection_mode": "typed-helper-unavailable-status-only", "helper_available": false,
"status_only": true, "inventory_complete": false, "inventory_present": false,
"authoritative_inventory_replacement": false,
"previous_authoritative_inventory_count": 2, "previous_authoritative_semantic_sha256": digest,
"operation_status": "degraded", "operation": "container.inventory",
"container_updates_enabled": false, "container_actions_enabled": false, "direct_socket_access": false,
}
recovery := summary(base+120, base+230)
recovery["previous_helper_pid"] = base + 220
recovery["previous_status_report_sequence"] = uint64(3)
bounds := map[string]any{
"collector_pid": base + 120, "helper_pid": base + 230,
"operation": "container.inventory", "failure_class": "bounded-timeout",
"deadline_ms": 2000, "elapsed_ms": 2000, "bounded_failure_observed": true,
"status_only_report_sequence": uint64(5), "recovery_report_sequence": uint64(6),
"collection_mode": "typed-helper-summary", "inventory_complete": true,
"previous_authoritative_inventory_count": 2, "previous_authoritative_semantic_sha256": digest,
"recovery_inventory_count": 2, "recovery_semantic_sha256": digest,
"full_fields_present": false, "stats_present": false, "secondary_structure_sha256": "",
"authoritative_empty_replacement": false, "collector_alive": true, "helper_alive": true,
"container_updates_enabled": false, "container_actions_enabled": false, "direct_socket_access": false,
}
update := summary(base+130, base+230)
update["previous_collector_pid"] = base + 120
update["previous_helper_pid"] = base + 230
update["previous_report_stream_id"] = runtimeKind + "-steady"
update["update_applied"] = true
update["collector_binary_sha256"] = digest
update["helper_binary_sha256"] = digest
authority := map[string]any{
"collector_pid": base + 130, "collector_uid": 1000 + index, "effective_uid": 1000 + index,
"effective_root": false, "safe_profile_enabled": true, "commands_enabled": false,
"privileged_helper_enabled": true, "reduction_request_observed": true,
"collector_command_transport_present": false, "collector_command_session_present": false,
"container_actions_enabled": false, "container_updates_enabled": false,
"rootful_socket_access": false, "direct_socket_access": false, "helper_network_access": false,
}
return rootfulQualRun{
Host: rootlessQualHost{
MachineID: "machine-" + runtimeKind, Architecture: "amd64",
Kernel: "Linux fixture", SystemdVersion: "systemd 255",
},
Runtime: rootfulQualRuntime{
Runtime: runtimeKind, RuntimeVersion: "1.0.0", DaemonID: daemonID,
DaemonRootless: false, SocketPath: socketPath, SocketUID: 0, SocketGID: socketGID,
SocketMode: "0660", SocketType: "unix", SocketSymlink: false,
},
Scenarios: []rootlessQualScenario{
makeScenario(0, "fresh_install", stream(runtimeKind+"-fresh"), sequence(1), fresh),
makeScenario(2, "legacy_migration", stream(runtimeKind+"-migration"), sequence(1), migration),
makeScenario(4, "collector_restart", stream(runtimeKind+"-steady"), sequence(1), collectorRestart),
makeScenario(6, "helper_restart", stream(runtimeKind+"-steady"), sequence(2), helperRestart),
makeScenario(8, "helper_loss", stream(runtimeKind+"-steady"), sequence(3), loss),
makeScenario(10, "helper_recovery", stream(runtimeKind+"-steady"), sequence(4), recovery),
makeScenario(12, "operation_bounds", stream(runtimeKind+"-steady"), sequence(6), bounds),
makeScenario(14, "update_preservation", stream(runtimeKind+"-update"), sequence(1), update),
makeScenario(16, "authority_isolation", nil, nil, authority),
makeScenario(18, "cleanup", nil, nil, map[string]any{
"collector_stopped": true, "helper_stopped": true, "runtime_stopped": true,
"socket_absent": true, "fixtures_removed": true, "state_clean": true,
}),
},
}
}
func TestRootfulQualificationWrapperInvariants(t *testing.T) {
raw, err := os.ReadFile(repoFile("scripts", "run-secure-runtime-rootful-qualification.sh"))
if err != nil {
t.Fatal(err)
}
script := string(raw)
for _, required := range []string{
"pulse-secure-runtime-rootful-qualification", "PULSE_ROOTFUL_QUALIFICATION_CONFIRM",
"--network none", "--cgroupns=private", "docker-receipt.json", "podman-receipt.json",
"secure_runtime_rootful_attestation_v1.py", "qualification output directory must have exact mode 0700",
"capture_qualification_container_diagnostics", "journalctl --no-pager -n 2000",
"org.pulse.rootful-qualification.run", "-buildvcs=true",
"github.com/rcourtman/pulse-go-rewrite/scripts/installtests.test",
} {
if !strings.Contains(script, required) {
t.Fatalf("rootful qualification wrapper missing %q", required)
}
}
if count := strings.Count(script, "-buildvcs=true"); count != 3 {
t.Fatalf("rootful wrapper must require VCS metadata for exactly three Go artifacts: got %d", count)
}
for _, forbidden := range []string{"/var/run/docker.sock:/", "/run/docker.sock:/", "/run/podman/podman.sock:/", "--pid=host", "--cgroupns=host"} {
if strings.Contains(script, forbidden) {
t.Fatalf("rootful qualification wrapper contains forbidden host boundary %q", forbidden)
}
}
}
@@ -0,0 +1,568 @@
#!/usr/bin/env python3
"""Validate the secret-free opt-in rootful runtime qualification receipt.
This validator qualifies only local, artifact-bound Docker and Podman
typed-helper summary collection. It does not qualify a published release,
authorize the safe profile as the default, or substitute for external review.
"""
from __future__ import annotations
import argparse
import contextlib
import hashlib
import json
import re
import stat
import sys
from datetime import datetime
from pathlib import Path
from typing import Any
import secure_runtime_rootless_attestation_v1 as hardened
RECEIPT_SCHEMA_VERSION = 1
ATTESTATION_SCHEMA_VERSION = 1
RECEIPT_KIND = "pulse-secure-runtime-rootful-qualification"
ATTESTATION_KIND = "pulse-secure-runtime-rootful-receipt-attestation"
CLASSIFICATION = "local-opt-in-rootful-runtime-artifact-bound-self-attestation"
SOURCE_MANIFEST_SCHEMA_VERSION = 1
SOURCE_MANIFEST_ID = "secure-runtime-rootful-v1"
SOURCE_MANIFEST_PATH = "scripts/release_control/secure_runtime_rootful_source_manifest_v1.json"
MAX_RECEIPT_BYTES = hardened.MAX_RECEIPT_BYTES
REQUIRED_RUNTIMES = ("docker", "podman")
REQUIRED_SCENARIOS = (
"fresh_install",
"legacy_migration",
"collector_restart",
"helper_restart",
"helper_loss",
"helper_recovery",
"operation_bounds",
"update_preservation",
"authority_isolation",
"cleanup",
)
ValidationError = hardened.ValidationError
fail = hardened.fail
require = hardened.require
require_object = hardened.require_object
require_bool = hardened.require_bool
require_int = hardened.require_int
require_text = hardened.require_text
require_digest = hardened.require_digest
parse_timestamp = hardened.parse_timestamp
reject_sensitive_evidence = hardened.reject_sensitive_evidence
read_immutable_receipt = hardened.read_immutable_receipt
immutable_artifact_snapshot = hardened.immutable_artifact_snapshot
inspect_go_build_identity = hardened.inspect_go_build_identity
verify_source_commit = hardened.verify_source_commit
IDENTITY_RE = hardened.IDENTITY_RE
VERSION_RE = hardened.VERSION_RE
GO_VERSION_RE = hardened.GO_VERSION_RE
COMMIT_RE = hardened.COMMIT_RE
SUMMARY_KEYS = {
"collector_pid",
"helper_pid",
"collection_mode",
"inventory_complete",
"inventory_count",
"semantic_sha256",
"full_fields_present",
"stats_present",
"secondary_structure_sha256",
"container_updates_enabled",
"container_actions_enabled",
"direct_socket_access",
"daemon_id",
"daemon_rootless",
}
def expected_socket_path(runtime: str) -> str:
return "/var/run/docker.sock" if runtime == "docker" else "/run/podman/podman.sock"
def validate_socket_mode(value: Any, label: str) -> str:
mode = require_text(value, label, pattern=re.compile(r"^0[0-7]{3}$"), maximum=4)
numeric = int(mode, 8)
require(numeric & 0o600 == 0o600, f"{label} must grant owner read and write")
require(numeric & 0o006 == 0, f"{label} must not grant world read or write")
return mode
def validate_summary(
value: Any,
label: str,
runtime: dict[str, Any],
*,
extra_keys: set[str] | None = None,
expected_collector_pid: int | None = None,
expected_helper_pid: int | None = None,
) -> dict[str, Any]:
item = require_object(value, label, SUMMARY_KEYS | (extra_keys or set()))
collector_pid = require_int(item["collector_pid"], f"{label}.collector_pid", minimum=2)
helper_pid = require_int(item["helper_pid"], f"{label}.helper_pid", minimum=2)
if expected_collector_pid is not None:
require(collector_pid == expected_collector_pid, f"{label} unexpectedly changed collector PID")
if expected_helper_pid is not None:
require(helper_pid == expected_helper_pid, f"{label} unexpectedly changed helper PID")
require(item["collection_mode"] == "typed-helper-summary", f"{label}.collection_mode differs")
require_bool(item["inventory_complete"], True, f"{label}.inventory_complete")
require_int(item["inventory_count"], f"{label}.inventory_count", minimum=1)
require_digest(item["semantic_sha256"], f"{label}.semantic_sha256")
require_bool(item["full_fields_present"], False, f"{label}.full_fields_present")
require_bool(item["stats_present"], False, f"{label}.stats_present")
require(item["secondary_structure_sha256"] == "", f"{label}.secondary_structure_sha256 must be empty")
for name in ("container_updates_enabled", "container_actions_enabled", "direct_socket_access"):
require_bool(item[name], False, f"{label}.{name}")
require(item["daemon_id"] == runtime["daemon_id"], f"{label}.daemon_id differs")
require_bool(item["daemon_rootless"], False, f"{label}.daemon_rootless")
return item
def require_summary_continuity(baseline: dict[str, Any], current: dict[str, Any], label: str) -> None:
require(current["inventory_count"] == baseline["inventory_count"], f"{label} inventory count differs")
require(current["semantic_sha256"] == baseline["semantic_sha256"], f"{label} semantic digest differs")
def validate_scenario(
value: Any,
expected_name: str,
label: str,
receipt_start: datetime,
receipt_end: datetime,
) -> tuple[dict[str, Any], datetime, datetime]:
item = require_object(
value,
label,
{"name", "result", "started_at", "completed_at", "report_stream_id", "report_sequence", "evidence"},
)
require(item["name"] == expected_name, f"{label}.name must be {expected_name}")
require(item["result"] == "passed", f"{label}.result must be passed")
started = parse_timestamp(item["started_at"], f"{label}.started_at")
completed = parse_timestamp(item["completed_at"], f"{label}.completed_at")
require(receipt_start <= started < completed <= receipt_end, f"{label} chronology is outside the receipt window")
if expected_name in {"authority_isolation", "cleanup"}:
require(item["report_stream_id"] is None, f"{label}.report_stream_id must be null")
require(item["report_sequence"] is None, f"{label}.report_sequence must be null")
else:
require_text(item["report_stream_id"], f"{label}.report_stream_id", pattern=IDENTITY_RE)
require_int(item["report_sequence"], f"{label}.report_sequence", minimum=1)
return item, started, completed
def validate_run(run_value: Any, expected_runtime: str, index: int, receipt_start: datetime, receipt_end: datetime) -> dict[str, Any]:
label = f"runs[{index}]"
run = require_object(run_value, label, {"host", "runtime", "scenarios"})
host = require_object(run["host"], f"{label}.host", {"machine_id", "architecture", "kernel", "systemd_version"})
require_text(host["machine_id"], f"{label}.host.machine_id", pattern=IDENTITY_RE)
require(host["architecture"] in {"amd64", "arm64"}, f"{label}.host.architecture is unsupported")
require_text(host["kernel"], f"{label}.host.kernel", maximum=128)
require_text(host["systemd_version"], f"{label}.host.systemd_version", maximum=128)
runtime = require_object(
run["runtime"],
f"{label}.runtime",
{"runtime", "runtime_version", "daemon_id", "daemon_rootless", "socket_path", "socket_uid", "socket_gid", "socket_mode", "socket_type", "socket_symlink"},
)
require(runtime["runtime"] == expected_runtime, f"{label}.runtime.runtime must be {expected_runtime}")
require_text(runtime["runtime_version"], f"{label}.runtime.runtime_version", pattern=VERSION_RE, maximum=128)
require_text(runtime["daemon_id"], f"{label}.runtime.daemon_id", pattern=IDENTITY_RE)
require_bool(runtime["daemon_rootless"], False, f"{label}.runtime.daemon_rootless")
require(runtime["socket_path"] == expected_socket_path(expected_runtime), f"{label}.runtime.socket_path differs")
require_int(runtime["socket_uid"], f"{label}.runtime.socket_uid")
require(runtime["socket_uid"] == 0, f"{label}.runtime.socket_uid must be root")
require_int(runtime["socket_gid"], f"{label}.runtime.socket_gid")
validate_socket_mode(runtime["socket_mode"], f"{label}.runtime.socket_mode")
require(runtime["socket_type"] == "unix", f"{label}.runtime.socket_type must be unix")
require_bool(runtime["socket_symlink"], False, f"{label}.runtime.socket_symlink")
scenarios = run["scenarios"]
require(isinstance(scenarios, list) and len(scenarios) == len(REQUIRED_SCENARIOS), f"{label}.scenarios must contain exactly ten scenarios")
parsed: dict[str, dict[str, Any]] = {}
previous_completed: datetime | None = None
for scenario_index, name in enumerate(REQUIRED_SCENARIOS):
scenario, started, completed = validate_scenario(
scenarios[scenario_index], name, f"{label}.scenarios[{scenario_index}]", receipt_start, receipt_end
)
if previous_completed is not None:
require(previous_completed <= started, f"{label}.scenarios are not chronological")
previous_completed = completed
parsed[name] = scenario
fresh = validate_summary(parsed["fresh_install"]["evidence"], f"{label}.fresh_install", runtime)
migration = validate_summary(
parsed["legacy_migration"]["evidence"],
f"{label}.legacy_migration",
runtime,
extra_keys={"legacy_profile", "target_profile", "authority_reduced", "legacy_collector_pid"},
)
require(migration["legacy_profile"] == "root-command-capable", f"{label}.legacy_migration legacy profile differs")
require(migration["target_profile"] == "typed-helper-monitoring-only", f"{label}.legacy_migration target profile differs")
require_bool(migration["authority_reduced"], True, f"{label}.legacy_migration.authority_reduced")
legacy_pid = require_int(migration["legacy_collector_pid"], f"{label}.legacy_migration.legacy_collector_pid", minimum=2)
require(migration["collector_pid"] != legacy_pid, f"{label}.legacy_migration retained legacy collector PID")
require_summary_continuity(fresh, migration, f"{label}.legacy_migration")
collector_restart = validate_summary(
parsed["collector_restart"]["evidence"],
f"{label}.collector_restart",
runtime,
extra_keys={"previous_collector_pid", "previous_report_stream_id"},
)
require(collector_restart["previous_collector_pid"] == migration["collector_pid"], f"{label}.collector_restart predecessor PID differs")
require(collector_restart["collector_pid"] != migration["collector_pid"], f"{label}.collector_restart did not change PID")
require(collector_restart["previous_report_stream_id"] == parsed["legacy_migration"]["report_stream_id"], f"{label}.collector_restart predecessor stream differs")
require_summary_continuity(fresh, collector_restart, f"{label}.collector_restart")
install_streams = [parsed[name]["report_stream_id"] for name in ("fresh_install", "legacy_migration", "collector_restart")]
require(len(set(install_streams)) == 3, f"{label} install and collector restart report streams must be distinct")
helper_restart = validate_summary(
parsed["helper_restart"]["evidence"],
f"{label}.helper_restart",
runtime,
extra_keys={"previous_helper_pid", "previous_helper_invocation_id", "helper_invocation_id"},
expected_collector_pid=collector_restart["collector_pid"],
)
require(helper_restart["previous_helper_pid"] == collector_restart["helper_pid"], f"{label}.helper_restart predecessor PID differs")
require(helper_restart["helper_pid"] != collector_restart["helper_pid"], f"{label}.helper_restart did not change helper PID")
previous_invocation = require_text(helper_restart["previous_helper_invocation_id"], f"{label}.helper_restart.previous_helper_invocation_id", pattern=IDENTITY_RE)
invocation = require_text(helper_restart["helper_invocation_id"], f"{label}.helper_restart.helper_invocation_id", pattern=IDENTITY_RE)
require(previous_invocation != invocation, f"{label}.helper_restart did not change InvocationID")
require_summary_continuity(fresh, helper_restart, f"{label}.helper_restart")
steady_stream = parsed["collector_restart"]["report_stream_id"]
require(parsed["helper_restart"]["report_stream_id"] == steady_stream, f"{label}.helper_restart changed report stream")
require(parsed["helper_restart"]["report_sequence"] > parsed["collector_restart"]["report_sequence"], f"{label}.helper_restart report did not advance")
loss_label = f"{label}.helper_loss"
loss = require_object(
parsed["helper_loss"]["evidence"],
loss_label,
{"collector_pid", "previous_helper_pid", "collection_mode", "helper_available", "status_only", "inventory_complete", "inventory_present", "authoritative_inventory_replacement", "previous_authoritative_inventory_count", "previous_authoritative_semantic_sha256", "operation_status", "operation", "container_updates_enabled", "container_actions_enabled", "direct_socket_access"},
)
require(loss["collector_pid"] == helper_restart["collector_pid"], f"{loss_label} changed collector PID")
require(loss["previous_helper_pid"] == helper_restart["helper_pid"], f"{loss_label} previous helper PID differs")
require(loss["collection_mode"] == "typed-helper-unavailable-status-only", f"{loss_label}.collection_mode differs")
for name, expected in (("helper_available", False), ("status_only", True), ("inventory_complete", False), ("inventory_present", False), ("authoritative_inventory_replacement", False)):
require_bool(loss[name], expected, f"{loss_label}.{name}")
require(loss["previous_authoritative_inventory_count"] == fresh["inventory_count"], f"{loss_label} prior count differs")
require(loss["previous_authoritative_semantic_sha256"] == fresh["semantic_sha256"], f"{loss_label} prior digest differs")
require(loss["operation_status"] == "degraded", f"{loss_label}.operation_status differs")
require(loss["operation"] == "container.inventory", f"{loss_label}.operation differs")
for name in ("container_updates_enabled", "container_actions_enabled", "direct_socket_access"):
require_bool(loss[name], False, f"{loss_label}.{name}")
require(parsed["helper_loss"]["report_stream_id"] == steady_stream, f"{loss_label} changed report stream")
require(parsed["helper_loss"]["report_sequence"] > parsed["helper_restart"]["report_sequence"], f"{loss_label} report did not advance")
recovery = validate_summary(
parsed["helper_recovery"]["evidence"],
f"{label}.helper_recovery",
runtime,
extra_keys={"previous_helper_pid", "previous_status_report_sequence"},
expected_collector_pid=helper_restart["collector_pid"],
)
require(recovery["previous_helper_pid"] == helper_restart["helper_pid"], f"{label}.helper_recovery previous helper PID differs")
require(recovery["helper_pid"] != helper_restart["helper_pid"], f"{label}.helper_recovery did not replace helper PID")
require(recovery["previous_status_report_sequence"] == parsed["helper_loss"]["report_sequence"], f"{label}.helper_recovery status sequence differs")
require(parsed["helper_recovery"]["report_stream_id"] == steady_stream, f"{label}.helper_recovery changed report stream")
require(parsed["helper_recovery"]["report_sequence"] > parsed["helper_loss"]["report_sequence"], f"{label}.helper_recovery report did not advance")
require_summary_continuity(fresh, recovery, f"{label}.helper_recovery")
bounds_label = f"{label}.operation_bounds"
bounds = require_object(
parsed["operation_bounds"]["evidence"],
bounds_label,
{"collector_pid", "helper_pid", "operation", "failure_class", "deadline_ms", "elapsed_ms", "bounded_failure_observed", "status_only_report_sequence", "recovery_report_sequence", "collection_mode", "inventory_complete", "previous_authoritative_inventory_count", "previous_authoritative_semantic_sha256", "recovery_inventory_count", "recovery_semantic_sha256", "full_fields_present", "stats_present", "secondary_structure_sha256", "authoritative_empty_replacement", "collector_alive", "helper_alive", "container_updates_enabled", "container_actions_enabled", "direct_socket_access"},
)
require(bounds["collector_pid"] == recovery["collector_pid"], f"{bounds_label} changed collector PID")
require(bounds["helper_pid"] == recovery["helper_pid"], f"{bounds_label} changed helper PID")
require(bounds["operation"] == "container.inventory", f"{bounds_label}.operation differs")
require(bounds["failure_class"] == "bounded-timeout", f"{bounds_label}.failure_class differs")
deadline = require_int(bounds["deadline_ms"], f"{bounds_label}.deadline_ms", minimum=1)
elapsed = require_int(bounds["elapsed_ms"], f"{bounds_label}.elapsed_ms", minimum=1)
require(deadline <= 30_000, f"{bounds_label}.deadline_ms exceeds the helper operation ceiling")
require(elapsed <= deadline + 1000, f"{bounds_label} exceeded bounded deadline allowance")
require_bool(bounds["bounded_failure_observed"], True, f"{bounds_label}.bounded_failure_observed")
status_sequence = require_int(bounds["status_only_report_sequence"], f"{bounds_label}.status_only_report_sequence", minimum=1)
recovery_sequence = require_int(bounds["recovery_report_sequence"], f"{bounds_label}.recovery_report_sequence", minimum=1)
require(parsed["operation_bounds"]["report_sequence"] == recovery_sequence, f"{bounds_label} wrapper sequence differs")
require(parsed["operation_bounds"]["report_stream_id"] == steady_stream, f"{bounds_label} changed report stream")
require(parsed["helper_recovery"]["report_sequence"] < status_sequence < recovery_sequence, f"{bounds_label} report sequence is not causal")
require(bounds["collection_mode"] == "typed-helper-summary", f"{bounds_label}.collection_mode differs")
require_bool(bounds["inventory_complete"], True, f"{bounds_label}.inventory_complete")
require(bounds["previous_authoritative_inventory_count"] == fresh["inventory_count"], f"{bounds_label} prior count differs")
require(bounds["previous_authoritative_semantic_sha256"] == fresh["semantic_sha256"], f"{bounds_label} prior digest differs")
require(bounds["recovery_inventory_count"] == fresh["inventory_count"], f"{bounds_label} recovery count differs")
require(bounds["recovery_semantic_sha256"] == fresh["semantic_sha256"], f"{bounds_label} recovery digest differs")
require_bool(bounds["full_fields_present"], False, f"{bounds_label}.full_fields_present")
require_bool(bounds["stats_present"], False, f"{bounds_label}.stats_present")
require(bounds["secondary_structure_sha256"] == "", f"{bounds_label}.secondary_structure_sha256 must be empty")
require_bool(bounds["authoritative_empty_replacement"], False, f"{bounds_label}.authoritative_empty_replacement")
for name in ("collector_alive", "helper_alive"):
require_bool(bounds[name], True, f"{bounds_label}.{name}")
for name in ("container_updates_enabled", "container_actions_enabled", "direct_socket_access"):
require_bool(bounds[name], False, f"{bounds_label}.{name}")
update = validate_summary(
parsed["update_preservation"]["evidence"],
f"{label}.update_preservation",
runtime,
extra_keys={"previous_collector_pid", "previous_helper_pid", "previous_report_stream_id", "update_applied", "collector_binary_sha256", "helper_binary_sha256"},
)
require(update["previous_collector_pid"] == recovery["collector_pid"], f"{label}.update_preservation predecessor PID differs")
require(update["collector_pid"] != recovery["collector_pid"], f"{label}.update_preservation did not replace collector PID")
require(update["previous_helper_pid"] == recovery["helper_pid"], f"{label}.update_preservation predecessor helper PID differs")
require(update["helper_pid"] == recovery["helper_pid"], f"{label}.update_preservation unexpectedly replaced helper PID")
require(update["previous_report_stream_id"] == steady_stream, f"{label}.update_preservation predecessor stream differs")
require(parsed["update_preservation"]["report_stream_id"] != steady_stream, f"{label}.update_preservation did not create a new stream")
require_bool(update["update_applied"], True, f"{label}.update_preservation.update_applied")
require_digest(update["collector_binary_sha256"], f"{label}.update_preservation.collector_binary_sha256")
require_digest(update["helper_binary_sha256"], f"{label}.update_preservation.helper_binary_sha256")
require_summary_continuity(fresh, update, f"{label}.update_preservation")
authority_label = f"{label}.authority_isolation"
authority = require_object(
parsed["authority_isolation"]["evidence"],
authority_label,
{"collector_pid", "collector_uid", "effective_uid", "effective_root", "safe_profile_enabled", "commands_enabled", "privileged_helper_enabled", "reduction_request_observed", "collector_command_transport_present", "collector_command_session_present", "container_actions_enabled", "container_updates_enabled", "rootful_socket_access", "direct_socket_access", "helper_network_access"},
)
require(authority["collector_pid"] == update["collector_pid"], f"{authority_label} collector PID differs")
uid = require_int(authority["collector_uid"], f"{authority_label}.collector_uid", minimum=1)
require(authority["effective_uid"] == uid, f"{authority_label}.effective_uid differs")
require_bool(authority["effective_root"], False, f"{authority_label}.effective_root")
require_bool(authority["safe_profile_enabled"], True, f"{authority_label}.safe_profile_enabled")
require_bool(authority["privileged_helper_enabled"], True, f"{authority_label}.privileged_helper_enabled")
require_bool(authority["reduction_request_observed"], True, f"{authority_label}.reduction_request_observed")
for name in ("commands_enabled", "collector_command_transport_present", "collector_command_session_present", "container_actions_enabled", "container_updates_enabled", "rootful_socket_access", "direct_socket_access", "helper_network_access"):
require_bool(authority[name], False, f"{authority_label}.{name}")
cleanup = require_object(
parsed["cleanup"]["evidence"],
f"{label}.cleanup",
{"collector_stopped", "helper_stopped", "runtime_stopped", "socket_absent", "fixtures_removed", "state_clean"},
)
for name in cleanup:
require_bool(cleanup[name], True, f"{label}.cleanup.{name}")
return run
def validate_receipt(receipt: Any) -> dict[str, Any]:
reject_sensitive_evidence(receipt)
root = require_object(receipt, "receipt", {"schema_version", "kind", "result", "source_commit", "started_at", "completed_at", "artifacts", "source_hashes", "runs"})
require(type(root["schema_version"]) is int and root["schema_version"] == RECEIPT_SCHEMA_VERSION, "receipt.schema_version must be 1")
require(root["kind"] == RECEIPT_KIND, f"receipt.kind must be {RECEIPT_KIND}")
require(root["result"] == "passed", "receipt.result must be passed")
source_commit = require_text(root["source_commit"], "receipt.source_commit", pattern=COMMIT_RE, maximum=40)
started = parse_timestamp(root["started_at"], "receipt.started_at")
completed = parse_timestamp(root["completed_at"], "receipt.completed_at")
require(started < completed, "receipt chronology is invalid")
artifacts = require_object(root["artifacts"], "receipt.artifacts", {"qualification_test", "collector", "helper", "installer"})
expected = {
"qualification_test": ("dockeragent.test", "github.com/rcourtman/pulse-go-rewrite/scripts/installtests.test"),
"collector": ("pulse-agent", "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent"),
"helper": ("pulse-agent-helper", "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent-helper"),
}
for name, (basename, package) in expected.items():
artifact = require_object(artifacts[name], f"receipt.artifacts.{name}", {"path_basename", "sha256", "package", "go_version", "vcs_revision", "vcs_modified"})
require(artifact["path_basename"] == basename, f"receipt.artifacts.{name}.path_basename differs")
require_digest(artifact["sha256"], f"receipt.artifacts.{name}.sha256")
require(artifact["package"] == package, f"receipt.artifacts.{name}.package differs")
require_text(artifact["go_version"], f"receipt.artifacts.{name}.go_version", pattern=GO_VERSION_RE, maximum=32)
require(artifact["vcs_revision"] == source_commit, f"receipt.artifacts.{name}.vcs_revision differs")
require_bool(artifact["vcs_modified"], False, f"receipt.artifacts.{name}.vcs_modified")
installer = require_object(artifacts["installer"], "receipt.artifacts.installer", {"path_basename", "sha256"})
require(installer["path_basename"] == "install.sh", "receipt.artifacts.installer.path_basename differs")
require_digest(installer["sha256"], "receipt.artifacts.installer.sha256")
source_hashes = root["source_hashes"]
require(isinstance(source_hashes, dict) and source_hashes, "receipt.source_hashes must be a non-empty object")
for source_path, digest in source_hashes.items():
require_text(source_path, "receipt.source_hashes path", maximum=512)
candidate = Path(source_path)
require(source_path == candidate.as_posix() and not candidate.is_absolute() and ".." not in candidate.parts, "receipt.source_hashes contains a non-canonical path")
require_digest(digest, f"receipt.source_hashes[{source_path}]")
require(source_hashes.get("scripts/install.sh") == installer["sha256"], "installer digest must equal governed scripts/install.sh")
runs = root["runs"]
require(isinstance(runs, list) and len(runs) == 2, "receipt.runs must contain exactly two runtime runs")
validated = [validate_run(runs[i], runtime, i, started, completed) for i, runtime in enumerate(REQUIRED_RUNTIMES)]
require(validated[0]["host"]["machine_id"] != validated[1]["host"]["machine_id"], "Docker and Podman runs must use distinct hosts")
require(validated[0]["runtime"]["daemon_id"] != validated[1]["runtime"]["daemon_id"], "Docker and Podman daemon identities must differ")
docker_stream = validated[0]["scenarios"][2]["report_stream_id"]
podman_stream = validated[1]["scenarios"][2]["report_stream_id"]
require(docker_stream != podman_stream, "Docker and Podman report streams must differ")
for index, run in enumerate(validated):
update = run["scenarios"][REQUIRED_SCENARIOS.index("update_preservation")]["evidence"]
require(update["collector_binary_sha256"] == artifacts["collector"]["sha256"], f"runs[{index}].update_preservation collector artifact digest differs")
require(update["helper_binary_sha256"] == artifacts["helper"]["sha256"], f"runs[{index}].update_preservation helper artifact digest differs")
return root
def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
fail(f"receipt contains duplicate JSON key {key!r}")
result[key] = value
return result
def parse_receipt_bytes(data: bytes) -> dict[str, Any]:
require(len(data) <= MAX_RECEIPT_BYTES, "receipt exceeds the maximum size")
try:
parsed = json.loads(data.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys, parse_constant=lambda value: fail(f"receipt contains invalid number {value}"))
except UnicodeDecodeError as exc:
raise ValidationError("receipt is not UTF-8") from exc
except json.JSONDecodeError as exc:
raise ValidationError(f"receipt is not valid JSON: {exc}") from exc
return validate_receipt(parsed)
def _canonical_relative_path(value: Any, label: str) -> str:
text = require_text(value, label, maximum=512)
candidate = Path(text)
require(not candidate.is_absolute() and candidate.as_posix() == text and text not in {"", "."} and ".." not in candidate.parts, f"{label} is not a canonical repository-relative path")
return text
def load_source_manifest(checkout: Path, manifest_path: Path) -> tuple[bytes, dict[str, str]]:
manifest_bytes = read_immutable_receipt(manifest_path)
try:
manifest = json.loads(manifest_bytes.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValidationError(f"source manifest is invalid JSON: {exc}") from exc
manifest = require_object(manifest, "source manifest", {"schema_version", "manifest_id", "target_os", "description", "exact_paths", "recursive_roots", "include_suffixes", "exclude_suffixes"})
require(type(manifest["schema_version"]) is int and manifest["schema_version"] == SOURCE_MANIFEST_SCHEMA_VERSION, "source manifest schema differs")
require(manifest["manifest_id"] == SOURCE_MANIFEST_ID, "source manifest ID differs")
require(manifest["target_os"] == "linux", "source manifest target_os must be linux")
require_text(manifest["description"], "source manifest description", maximum=1024)
for field in ("exact_paths", "recursive_roots", "include_suffixes", "exclude_suffixes"):
require(isinstance(manifest[field], list) and all(isinstance(item, str) for item in manifest[field]), f"source manifest {field} must be a text array")
require(len(manifest[field]) == len(set(manifest[field])), f"source manifest {field} contains duplicates")
require(manifest["exact_paths"], "source manifest exact_paths must not be empty")
include_suffixes = tuple(manifest["include_suffixes"])
exclude_suffixes = tuple(manifest["exclude_suffixes"])
require(include_suffixes and all(item.startswith(".") for item in include_suffixes), "source manifest include_suffixes are invalid")
require(all(item.startswith("_") or item.startswith(".") for item in exclude_suffixes), "source manifest exclude_suffixes are invalid")
paths = {_canonical_relative_path(item, "source manifest exact path") for item in manifest["exact_paths"]}
for raw_root in manifest["recursive_roots"]:
root_text = _canonical_relative_path(raw_root, "source manifest recursive root")
root = checkout / root_text
require(root.is_dir() and not root.is_symlink(), f"source manifest recursive root is not a regular directory: {root_text}")
for candidate in root.rglob("*"):
if candidate.is_dir() or candidate.name.endswith(exclude_suffixes) or not candidate.name.endswith(include_suffixes):
continue
paths.add(candidate.relative_to(checkout).as_posix())
hashes: dict[str, str] = {}
checkout_resolved = checkout.resolve(strict=True)
for relative in sorted(paths):
candidate = checkout / relative
try:
metadata = candidate.lstat()
except OSError as exc:
raise ValidationError(f"unable to inspect governed source {relative}: {exc}") from exc
require(stat.S_ISREG(metadata.st_mode) and not stat.S_ISLNK(metadata.st_mode), f"governed source is not a regular file: {relative}")
require(candidate.resolve(strict=True).is_relative_to(checkout_resolved), f"governed source escapes checkout: {relative}")
hashes[relative] = hashlib.sha256(read_immutable_receipt(candidate)).hexdigest()
require(hashes, "source manifest expands to no source files")
return manifest_bytes, hashes
def create_attestation(
receipt_path: Path,
qualification_test_path: Path,
collector_path: Path,
helper_path: Path,
installer_path: Path,
*,
checkout: Path | None = None,
manifest_path: Path | None = None,
verify_git_commit: bool = True,
) -> dict[str, Any]:
data = read_immutable_receipt(receipt_path)
receipt = parse_receipt_bytes(data)
checkout = (checkout or Path(__file__).resolve().parents[2]).resolve(strict=True)
manifest_path = manifest_path or checkout / SOURCE_MANIFEST_PATH
manifest_bytes, source_hashes = load_source_manifest(checkout, manifest_path)
require(receipt["source_hashes"] == source_hashes, "receipt source_hashes do not match the canonical source manifest")
if verify_git_commit:
verify_source_commit(checkout, receipt["source_commit"], source_hashes)
supplied = {
"qualification_test": (qualification_test_path, "dockeragent.test", True),
"collector": (collector_path, "pulse-agent", True),
"helper": (helper_path, "pulse-agent-helper", True),
"installer": (installer_path, "install.sh", False),
}
bindings: dict[str, dict[str, Any]] = {}
with contextlib.ExitStack() as stack:
snapshots: dict[str, Path] = {}
for name, (path, basename, executable) in supplied.items():
snapshot, digest = stack.enter_context(immutable_artifact_snapshot(path, basename, executable=executable))
require(digest == receipt["artifacts"][name]["sha256"], f"{name} bytes do not match receipt SHA-256")
snapshots[name] = snapshot
bindings[name] = {"path_basename": basename, "sha256": digest}
for name in ("qualification_test", "collector", "helper"):
build = inspect_go_build_identity(snapshots[name])
claimed = receipt["artifacts"][name]
require(build["package"] == claimed["package"], f"{name} Go package differs from receipt")
require(build["go_version"] == claimed["go_version"], f"{name} Go version differs from receipt")
require(build["vcs_revision"] == claimed["vcs_revision"] == receipt["source_commit"], f"{name} VCS revision differs")
require_bool(build["vcs_modified"], False, f"{name} vcs.modified")
bindings[name].update(build)
validator_bytes = read_immutable_receipt(Path(__file__).absolute())
return {
"schema_version": ATTESTATION_SCHEMA_VERSION,
"kind": ATTESTATION_KIND,
"classification": CLASSIFICATION,
"receipt_sha256": hashlib.sha256(data).hexdigest(),
"validator_source_sha256": hashlib.sha256(validator_bytes).hexdigest(),
"source_manifest_schema_version": SOURCE_MANIFEST_SCHEMA_VERSION,
"source_manifest_id": SOURCE_MANIFEST_ID,
"source_manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
"source_hash_count": len(source_hashes),
"source_commit": receipt["source_commit"],
"source_commit_verified": verify_git_commit,
"artifact_bindings": bindings,
"validated_runtimes": list(REQUIRED_RUNTIMES),
"limitations": [
"not-published-release-provenance",
"not-default-profile-authorization",
"not-independent-security-review",
"production-exact-scope-proof-is-external-prior",
],
}
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("receipt", type=Path, help="schema-v1 rootful qualification receipt")
parser.add_argument("--qualification-test", required=True, type=Path, help="exact dockeragent.test qualification binary")
parser.add_argument("--collector", required=True, type=Path, help="exact pulse-agent binary")
parser.add_argument("--helper", required=True, type=Path, help="exact pulse-agent-helper binary")
parser.add_argument("--installer", required=True, type=Path, help="exact install.sh blob")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
arguments = parse_args(argv)
try:
attestation = create_attestation(arguments.receipt, arguments.qualification_test, arguments.collector, arguments.helper, arguments.installer)
except ValidationError as exc:
print(f"rootful receipt validation failed: {exc}", file=sys.stderr)
return 1
print(json.dumps(attestation, sort_keys=True, separators=(",", ":")))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,475 @@
#!/usr/bin/env python3
"""Adversarial tests for the schema-v1 rootful qualification validator."""
from __future__ import annotations
import copy
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent))
import secure_runtime_rootful_attestation_v1 as attester
def sha(value: str | bytes) -> str:
return hashlib.sha256(value if isinstance(value, bytes) else value.encode()).hexdigest()
class RootfulAttestationV1Test(unittest.TestCase):
def setUp(self) -> None:
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.commit = "a" * 40
self.artifacts: dict[str, Path] = {}
for key, basename in {
"qualification_test": "dockeragent.test",
"collector": "pulse-agent",
"helper": "pulse-agent-helper",
"installer": "install.sh",
}.items():
path = self.root / basename
path.write_bytes(f"{key}-bytes".encode())
path.chmod(0o700 if key != "installer" else 0o600)
self.artifacts[key] = path
self.sources = {
"runtime.go": b"package runtime\n",
"internal/api/api_tokens.go": b"package api\n",
"scripts/install.sh": self.artifacts["installer"].read_bytes(),
}
for relative, data in self.sources.items():
target = self.root / relative
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
self.manifest = self.root / "manifest.json"
self.manifest.write_text(
json.dumps(
{
"schema_version": 1,
"manifest_id": attester.SOURCE_MANIFEST_ID,
"target_os": "linux",
"description": "test boundary",
"exact_paths": sorted(self.sources),
"recursive_roots": [],
"include_suffixes": [".go"],
"exclude_suffixes": ["_test.go"],
}
),
encoding="utf-8",
)
self.receipt = self.make_receipt()
def tearDown(self) -> None:
self.temp.cleanup()
@staticmethod
def ts(index: int) -> str:
value = datetime(2026, 9, 1, 10, tzinfo=timezone.utc) + timedelta(minutes=index)
return value.isoformat().replace("+00:00", "Z")
def summary(self, runtime: str, collector_pid: int, helper_pid: int, daemon_id: str) -> dict:
return {
"collector_pid": collector_pid,
"helper_pid": helper_pid,
"collection_mode": "typed-helper-summary",
"inventory_complete": True,
"inventory_count": 2,
"semantic_sha256": sha(runtime + "-summary"),
"full_fields_present": False,
"stats_present": False,
"secondary_structure_sha256": "",
"container_updates_enabled": False,
"container_actions_enabled": False,
"direct_socket_access": False,
"daemon_id": daemon_id,
"daemon_rootless": False,
}
def make_run(self, runtime: str, index: int) -> dict:
daemon = runtime + "-rootful-daemon"
base_collector = 4100 + index * 100
base_helper = 5100 + index * 100
fresh = self.summary(runtime, base_collector, base_helper, daemon)
migration = {
**self.summary(runtime, base_collector + 1, base_helper + 1, daemon),
"legacy_profile": "root-command-capable",
"target_profile": "typed-helper-monitoring-only",
"authority_reduced": True,
"legacy_collector_pid": base_collector + 50,
}
collector_restart = {
**self.summary(runtime, base_collector + 2, base_helper + 1, daemon),
"previous_collector_pid": migration["collector_pid"],
"previous_report_stream_id": runtime + "-migration-stream",
}
helper_restart = {
**self.summary(runtime, collector_restart["collector_pid"], base_helper + 2, daemon),
"previous_helper_pid": collector_restart["helper_pid"],
"previous_helper_invocation_id": runtime + "-helper-before",
"helper_invocation_id": runtime + "-helper-after",
}
loss = {
"collector_pid": helper_restart["collector_pid"],
"previous_helper_pid": helper_restart["helper_pid"],
"collection_mode": "typed-helper-unavailable-status-only",
"helper_available": False,
"status_only": True,
"inventory_complete": False,
"inventory_present": False,
"authoritative_inventory_replacement": False,
"previous_authoritative_inventory_count": fresh["inventory_count"],
"previous_authoritative_semantic_sha256": fresh["semantic_sha256"],
"operation_status": "degraded",
"operation": "container.inventory",
"container_updates_enabled": False,
"container_actions_enabled": False,
"direct_socket_access": False,
}
recovery = {
**self.summary(runtime, helper_restart["collector_pid"], base_helper + 3, daemon),
"previous_helper_pid": helper_restart["helper_pid"],
"previous_status_report_sequence": 3,
}
bounds = {
"collector_pid": recovery["collector_pid"],
"helper_pid": recovery["helper_pid"],
"operation": "container.inventory",
"failure_class": "bounded-timeout",
"deadline_ms": 2000,
"elapsed_ms": 2050,
"bounded_failure_observed": True,
"status_only_report_sequence": 5,
"recovery_report_sequence": 6,
"collection_mode": "typed-helper-summary",
"inventory_complete": True,
"previous_authoritative_inventory_count": fresh["inventory_count"],
"previous_authoritative_semantic_sha256": fresh["semantic_sha256"],
"recovery_inventory_count": fresh["inventory_count"],
"recovery_semantic_sha256": fresh["semantic_sha256"],
"full_fields_present": False,
"stats_present": False,
"secondary_structure_sha256": "",
"authoritative_empty_replacement": False,
"collector_alive": True,
"helper_alive": True,
"container_updates_enabled": False,
"container_actions_enabled": False,
"direct_socket_access": False,
}
update = {
**self.summary(runtime, base_collector + 3, recovery["helper_pid"], daemon),
"previous_collector_pid": recovery["collector_pid"],
"previous_helper_pid": recovery["helper_pid"],
"previous_report_stream_id": runtime + "-restart-stream",
"update_applied": True,
"collector_binary_sha256": sha(self.artifacts["collector"].read_bytes()),
"helper_binary_sha256": sha(self.artifacts["helper"].read_bytes()),
}
authority = {
"collector_pid": update["collector_pid"],
"collector_uid": 1200 + index,
"effective_uid": 1200 + index,
"effective_root": False,
"safe_profile_enabled": True,
"commands_enabled": False,
"privileged_helper_enabled": True,
"reduction_request_observed": True,
"collector_command_transport_present": False,
"collector_command_session_present": False,
"container_actions_enabled": False,
"container_updates_enabled": False,
"rootful_socket_access": False,
"direct_socket_access": False,
"helper_network_access": False,
}
cleanup = {
"collector_stopped": True,
"helper_stopped": True,
"runtime_stopped": True,
"socket_absent": True,
"fixtures_removed": True,
"state_clean": True,
}
evidences = [fresh, migration, collector_restart, helper_restart, loss, recovery, bounds, update, authority, cleanup]
reports = {
"fresh_install": (runtime + "-fresh-stream", 1),
"legacy_migration": (runtime + "-migration-stream", 1),
"collector_restart": (runtime + "-restart-stream", 1),
"helper_restart": (runtime + "-restart-stream", 2),
"helper_loss": (runtime + "-restart-stream", 3),
"helper_recovery": (runtime + "-restart-stream", 4),
"operation_bounds": (runtime + "-restart-stream", 6),
"update_preservation": (runtime + "-update-stream", 1),
}
scenarios = []
for scenario_index, name in enumerate(attester.REQUIRED_SCENARIOS):
stream, sequence = reports.get(name, (None, None))
scenarios.append(
{
"name": name,
"result": "passed",
"started_at": self.ts(index * 25 + scenario_index * 2 + 1),
"completed_at": self.ts(index * 25 + scenario_index * 2 + 2),
"report_stream_id": stream,
"report_sequence": sequence,
"evidence": evidences[scenario_index],
}
)
return {
"host": {
"machine_id": f"{index + 1:032x}",
"architecture": "amd64",
"kernel": "6.12.0-test",
"systemd_version": "systemd-257",
},
"runtime": {
"runtime": runtime,
"runtime_version": "27.3.1" if runtime == "docker" else "5.2.2",
"daemon_id": daemon,
"daemon_rootless": False,
"socket_path": attester.expected_socket_path(runtime),
"socket_uid": 0,
"socket_gid": 0,
"socket_mode": "0660",
"socket_type": "unix",
"socket_symlink": False,
},
"scenarios": scenarios,
}
def make_receipt(self) -> dict:
packages = {
"qualification_test": "github.com/rcourtman/pulse-go-rewrite/scripts/installtests.test",
"collector": "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent",
"helper": "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent-helper",
}
artifacts = {
"installer": {
"path_basename": "install.sh",
"sha256": sha(self.artifacts["installer"].read_bytes()),
}
}
for name, package in packages.items():
artifacts[name] = {
"path_basename": self.artifacts[name].name,
"sha256": sha(self.artifacts[name].read_bytes()),
"package": package,
"go_version": "go1.25.1",
"vcs_revision": self.commit,
"vcs_modified": False,
}
return {
"schema_version": 1,
"kind": attester.RECEIPT_KIND,
"result": "passed",
"source_commit": self.commit,
"started_at": self.ts(0),
"completed_at": self.ts(50),
"source_hashes": {name: sha(data) for name, data in self.sources.items()},
"artifacts": artifacts,
"runs": [self.make_run("docker", 0), self.make_run("podman", 1)],
}
def scenario(self, receipt: dict, run: int, name: str) -> dict:
return receipt["runs"][run]["scenarios"][attester.REQUIRED_SCENARIOS.index(name)]
def invalid(self, mutation) -> None:
receipt = copy.deepcopy(self.receipt)
mutation(receipt)
with self.assertRaises(attester.ValidationError):
attester.validate_receipt(receipt)
def write_receipt(self, receipt: dict | None = None) -> Path:
path = self.root / "receipt.json"
path.write_text(json.dumps(receipt or self.receipt, sort_keys=True), encoding="utf-8")
return path
def valid_build(self, package: str) -> dict:
return {"package": package, "go_version": "go1.25.1", "vcs_revision": self.commit, "vcs_modified": False}
def attest(self, receipt: dict | None = None) -> dict:
packages = {name: self.receipt["artifacts"][name]["package"] for name in ("qualification_test", "collector", "helper")}
by_basename = {path.name: key for key, path in self.artifacts.items()}
with mock.patch.object(attester, "inspect_go_build_identity", side_effect=lambda path: self.valid_build(packages[by_basename[path.name]])):
return attester.create_attestation(
self.write_receipt(receipt),
self.artifacts["qualification_test"],
self.artifacts["collector"],
self.artifacts["helper"],
self.artifacts["installer"],
checkout=self.root,
manifest_path=self.manifest,
verify_git_commit=False,
)
def test_valid_receipt_and_local_attestation(self) -> None:
attester.validate_receipt(self.receipt)
result = self.attest()
self.assertEqual(result["validated_runtimes"], ["docker", "podman"])
self.assertEqual(result["classification"], attester.CLASSIFICATION)
self.assertEqual(set(result["artifact_bindings"]), set(self.artifacts))
self.assertEqual(
result["limitations"],
[
"not-published-release-provenance",
"not-default-profile-authorization",
"not-independent-security-review",
"production-exact-scope-proof-is-external-prior",
],
)
def test_exact_runtime_topology_and_order_fail_closed(self) -> None:
mutations = [
lambda r: r.update(schema_version=True),
lambda r: r["runs"].pop(),
lambda r: r["runs"].reverse(),
lambda r: r["runs"][1]["host"].update(machine_id=r["runs"][0]["host"]["machine_id"]),
lambda r: r["runs"][1]["runtime"].update(daemon_id=r["runs"][0]["runtime"]["daemon_id"]),
lambda r: r["runs"][0]["runtime"].update(daemon_rootless=True),
lambda r: r["runs"][0]["runtime"].update(socket_path="/run/docker.sock"),
lambda r: r["runs"][0]["runtime"].update(socket_uid=1000),
lambda r: r["runs"][0]["runtime"].update(socket_mode="0666"),
lambda r: r["runs"][0]["runtime"].update(socket_symlink=True),
lambda r: r["runs"][0]["scenarios"].reverse(),
]
for mutation in mutations:
with self.subTest(mutation=mutation):
self.invalid(mutation)
def test_summary_is_exactly_helper_only_and_continuous(self) -> None:
mutations = [
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(collection_mode="direct-rootful-socket"),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(full_fields_present=True),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(stats_present=True),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(secondary_structure_sha256=sha("secondary")),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(container_updates_enabled=True),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(container_actions_enabled=True),
lambda r: self.scenario(r, 0, "fresh_install")["evidence"].update(direct_socket_access=True),
lambda r: self.scenario(r, 0, "helper_recovery")["evidence"].update(inventory_count=3),
lambda r: self.scenario(r, 0, "update_preservation")["evidence"].update(semantic_sha256=sha("wrong")),
]
for mutation in mutations:
with self.subTest(mutation=mutation):
self.invalid(mutation)
def test_pid_stream_loss_recovery_and_bounds_are_causal(self) -> None:
mutations = [
lambda r: self.scenario(r, 0, "legacy_migration")["evidence"].update(authority_reduced=False),
lambda r: self.scenario(r, 0, "collector_restart")["evidence"].update(previous_collector_pid=9999),
lambda r: self.scenario(r, 0, "helper_restart")["evidence"].update(previous_helper_pid=9999),
lambda r: self.scenario(r, 0, "helper_restart")["evidence"].update(helper_invocation_id="docker-helper-before"),
lambda r: self.scenario(r, 0, "helper_loss")["evidence"].update(status_only=False),
lambda r: self.scenario(r, 0, "helper_loss")["evidence"].update(inventory_present=True),
lambda r: self.scenario(r, 0, "helper_loss")["evidence"].update(authoritative_inventory_replacement=True),
lambda r: self.scenario(r, 0, "helper_recovery")["evidence"].update(previous_status_report_sequence=2),
lambda r: self.scenario(r, 0, "operation_bounds")["evidence"].update(elapsed_ms=5000),
lambda r: self.scenario(r, 0, "operation_bounds")["evidence"].update(deadline_ms=60000, elapsed_ms=100),
lambda r: self.scenario(r, 0, "operation_bounds")["evidence"].update(authoritative_empty_replacement=True),
lambda r: self.scenario(r, 0, "operation_bounds")["evidence"].update(stats_present=True),
lambda r: self.scenario(r, 0, "operation_bounds")["evidence"].update(recovery_report_sequence=5),
lambda r: self.scenario(r, 0, "update_preservation")["evidence"].update(previous_collector_pid=9999),
lambda r: self.scenario(r, 0, "update_preservation")["evidence"].update(previous_helper_pid=9999),
lambda r: self.scenario(r, 0, "update_preservation")["evidence"].update(helper_pid=9999),
lambda r: self.scenario(r, 0, "update_preservation")["evidence"].update(collector_binary_sha256=sha("wrong")),
]
for mutation in mutations:
with self.subTest(mutation=mutation):
self.invalid(mutation)
def test_authority_cleanup_secret_and_extra_fields_fail_closed(self) -> None:
mutations = [
lambda r: self.scenario(r, 0, "authority_isolation")["evidence"].update(commands_enabled=True),
lambda r: self.scenario(r, 0, "authority_isolation")["evidence"].update(rootful_socket_access=True),
lambda r: self.scenario(r, 0, "authority_isolation")["evidence"].update(helper_network_access=True),
lambda r: self.scenario(r, 0, "cleanup")["evidence"].update(socket_absent=False),
lambda r: self.scenario(r, 0, "cleanup")["evidence"].update(user_state_clean=True),
lambda r: r["runs"][0]["host"].update(kernel="Bearer abcdefghijklmnop"),
]
for mutation in mutations:
with self.subTest(mutation=mutation):
self.invalid(mutation)
def test_artifact_source_and_duplicate_json_binding(self) -> None:
self.artifacts["collector"].write_bytes(b"different")
self.artifacts["collector"].chmod(0o700)
with self.assertRaises(attester.ValidationError):
self.attest()
self.artifacts["collector"].write_bytes(b"collector-bytes")
self.artifacts["collector"].chmod(0o700)
helper = self.artifacts["helper"]
real_helper = self.root / "helper-real"
helper.rename(real_helper)
helper.symlink_to(real_helper)
with self.assertRaises(attester.ValidationError):
self.attest()
helper.unlink()
real_helper.rename(helper)
bad = copy.deepcopy(self.receipt)
bad["source_hashes"]["runtime.go"] = sha("wrong")
with self.assertRaises(attester.ValidationError):
self.attest(bad)
with self.assertRaises(attester.ValidationError):
attester.parse_receipt_bytes(b'{"schema_version":1,"schema_version":1}')
with mock.patch.object(attester, "inspect_go_build_identity", return_value=self.valid_build("github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent")):
with self.assertRaises(attester.ValidationError):
attester.create_attestation(
self.write_receipt(), self.artifacts["qualification_test"], self.artifacts["collector"],
self.artifacts["helper"], self.artifacts["installer"], checkout=self.root,
manifest_path=self.manifest, verify_git_commit=False,
)
args = attester.parse_args([
"receipt.json", "--qualification-test", "dockeragent.test", "--collector", "pulse-agent",
"--helper", "pulse-agent-helper", "--installer", "install.sh",
])
self.assertEqual(args.qualification_test.name, "dockeragent.test")
@unittest.skipUnless(hasattr(os, "mkfifo"), "FIFO validation requires Unix mkfifo")
def test_fifo_receipt_and_artifact_paths_fail_without_blocking(self) -> None:
module_dir = Path(attester.__file__).resolve().parent
receipt_fifo = self.root / "receipt.fifo"
os.mkfifo(receipt_fifo, 0o600)
receipt_probe = (
"import sys; from pathlib import Path; sys.path.insert(0, sys.argv[1]); "
"import secure_runtime_rootful_attestation_v1 as a; a.read_immutable_receipt(Path(sys.argv[2]))"
)
result = subprocess.run([sys.executable, "-I", "-c", receipt_probe, str(module_dir), str(receipt_fifo)], capture_output=True, timeout=2)
self.assertNotEqual(result.returncode, 0)
self.assertIn(b"receipt must be a regular file", result.stderr)
artifact_fifo = self.root / "pulse-agent"
artifact_fifo.unlink()
os.mkfifo(artifact_fifo, 0o700)
artifact_probe = (
"import sys; from pathlib import Path; sys.path.insert(0, sys.argv[1]); "
"import secure_runtime_rootful_attestation_v1 as a; "
"a.immutable_artifact_snapshot(Path(sys.argv[2]), 'pulse-agent', executable=True).__enter__()"
)
result = subprocess.run([sys.executable, "-I", "-c", artifact_probe, str(module_dir), str(artifact_fifo)], capture_output=True, timeout=2)
self.assertNotEqual(result.returncode, 0)
self.assertIn(b"artifact pulse-agent must be a regular file", result.stderr)
def test_manifest_contract_binds_transitive_harness_and_production_boundary(self) -> None:
manifest = json.loads((Path(__file__).with_name("secure_runtime_rootful_source_manifest_v1.json")).read_text(encoding="utf-8"))
required = {
"scripts/installtests/secure_runtime_rootful_qualification_test.go",
"scripts/installtests/secure_runtime_rootless_qualification_test.go",
"scripts/installtests/secure_runtime_systemd_lab_test.go",
"scripts/release_control/secure_runtime_rootful_attestation_v1.py",
"scripts/release_control/secure_runtime_rootful_source_manifest_v1.json",
"scripts/release_control/secure_runtime_rootless_attestation_v1.py",
"scripts/run-secure-runtime-rootful-qualification.sh",
}
self.assertTrue(required.issubset(manifest["exact_paths"]))
self.assertIn("internal/agenthelper", manifest["recursive_roots"])
self.assertIn("pkg/auth", manifest["recursive_roots"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,51 @@
{
"schema_version": 1,
"manifest_id": "secure-runtime-rootful-v1",
"target_os": "linux",
"description": "Production source boundary for rootful Docker and Podman typed-helper summary inventory, loss and recovery, operation bounds, update preservation, installer migration, and the standalone rootful qualification packet.",
"exact_paths": [
"VERSION",
"go.mod",
"go.sum",
"pkg/agents/docker/report.go",
"pkg/agents/docker/report_limits.go",
"scripts/install.sh",
"scripts/release_ldflags.sh",
"scripts/installtests/secure_runtime_rootful_qualification_test.go",
"scripts/installtests/secure_runtime_rootless_qualification_test.go",
"scripts/installtests/secure_runtime_systemd_lab_test.go",
"scripts/installtests/testdata/secure_runtime_docker_fixture.go",
"scripts/release_control/secure_runtime_rootful_attestation_v1.py",
"scripts/release_control/secure_runtime_rootful_source_manifest_v1.json",
"scripts/release_control/secure_runtime_rootless_attestation_v1.py",
"scripts/run-secure-runtime-rootful-qualification.sh"
],
"recursive_roots": [
"cmd/pulse-agent",
"cmd/pulse-agent-helper",
"internal/actionrunner",
"internal/agentexec",
"internal/agenthelper",
"internal/agenttls",
"internal/agentupdate",
"internal/api",
"internal/collectorlifecycle",
"internal/config",
"internal/dockeragent",
"internal/hostagent",
"internal/operationreceipt",
"internal/securityutil",
"internal/updatesignature",
"pkg/auth",
"pkg/agents/host",
"pkg/securityutil",
"pkg/tlsutil"
],
"include_suffixes": [
".go",
".tmpl"
],
"exclude_suffixes": [
"_test.go"
]
}
+409
View File
@@ -0,0 +1,409 @@
#!/usr/bin/env bash
set -euo pipefail
readonly REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
readonly UBUNTU_IMAGE="${PULSE_ROOTFUL_UBUNTU_IMAGE:?set PULSE_ROOTFUL_UBUNTU_IMAGE to an immutable ubuntu@sha256:... Ubuntu 24.04 image}"
readonly OUTPUT_PARENT="${PULSE_ROOTFUL_QUALIFICATION_OUTPUT_DIR:?set PULSE_ROOTFUL_QUALIFICATION_OUTPUT_DIR to an existing absolute private directory}"
readonly CONFIRM="${PULSE_ROOTFUL_QUALIFICATION_CONFIRM:-}"
portable_mode() {
stat -c '%a' "$1" 2>/dev/null || stat -f '%Lp' "$1"
}
portable_uid() {
stat -c '%u' "$1" 2>/dev/null || stat -f '%u' "$1"
}
if [[ ! "${UBUNTU_IMAGE}" =~ ^ubuntu@sha256:[0-9a-f]{64}$ ]]; then
echo "ERROR: PULSE_ROOTFUL_UBUNTU_IMAGE must be an exact ubuntu@sha256 digest" >&2
exit 2
fi
if [[ "${OUTPUT_PARENT}" != /* || ! -d "${OUTPUT_PARENT}" || -L "${OUTPUT_PARENT}" ]]; then
echo "ERROR: PULSE_ROOTFUL_QUALIFICATION_OUTPUT_DIR must be an existing absolute non-symlink directory" >&2
exit 2
fi
if [[ "$(portable_mode "${OUTPUT_PARENT}")" != "700" ]]; then
echo "ERROR: qualification output directory must have exact mode 0700" >&2
exit 2
fi
if [[ "$(portable_uid "${OUTPUT_PARENT}")" != "$(id -u)" ]]; then
echo "ERROR: qualification output directory must be owned by the invoking user" >&2
exit 2
fi
if ! command -v docker >/dev/null 2>&1 || ! docker info >/dev/null 2>&1; then
echo "ERROR: a working Docker CLI/daemon is required to create disposable qualification containers" >&2
exit 2
fi
if [[ "$(git -C "${REPO_ROOT}" branch --show-current)" != "main" ]]; then
echo "ERROR: qualification builds are allowed only from main" >&2
exit 2
fi
if [[ -n "$(git -C "${REPO_ROOT}" status --porcelain)" ]]; then
echo "ERROR: qualification requires a clean exact source checkout" >&2
exit 2
fi
readonly SOURCE_COMMIT="$(git -C "${REPO_ROOT}" rev-parse HEAD)"
readonly EXPECTED_CONFIRM="I_HAVE_VERIFIED_THESE_ARE_DISPOSABLE_ROOTFUL_SYSTEMD_CONTAINERS_COMMIT_${SOURCE_COMMIT}"
if [[ "${CONFIRM}" != "${EXPECTED_CONFIRM}" ]]; then
echo "ERROR: exact destructive opt-in required:" >&2
echo "export PULSE_ROOTFUL_QUALIFICATION_CONFIRM=${EXPECTED_CONFIRM}" >&2
exit 2
fi
readonly RUN_STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
readonly OUTPUT_DIR="${OUTPUT_PARENT}/${RUN_STAMP}-${SOURCE_COMMIT:0:12}"
readonly IMAGE_TAG="pulse-rootful-qualification:${SOURCE_COMMIT:0:12}"
readonly CONTAINER_RUN_LABEL="org.pulse.rootful-qualification.run"
readonly CONTAINER_RUN_NONCE="$(openssl rand -hex 16)"
if docker image inspect "${IMAGE_TAG}" >/dev/null 2>&1; then
echo "ERROR: qualification image tag already exists; refusing to overwrite ${IMAGE_TAG}" >&2
exit 2
fi
PACKET_DIR="$(mktemp -d /tmp/pulse-rootful-packet.XXXXXX)"
CONTAINER_IDS=()
IMAGE_CREATED=false
inspect_container_nonce() {
docker inspect --format '{{ index .Config.Labels "org.pulse.rootful-qualification.run" }}' "$1"
}
remove_container_strict() {
local container_id="$1"
local observed_nonce remaining
observed_nonce="$(inspect_container_nonce "${container_id}")" || {
echo "ERROR: unable to verify ownership label for qualification container ${container_id}" >&2
return 1
}
if [[ "${observed_nonce}" != "${CONTAINER_RUN_NONCE}" ]]; then
echo "ERROR: qualification container ${container_id} ownership label changed" >&2
return 1
fi
docker rm -f "${container_id}" >/dev/null || return 1
remaining="$(docker ps -aq --no-trunc --filter "id=${container_id}")" || return 1
if printf '%s\n' "${remaining}" | grep -Fxq "${container_id}"; then
echo "ERROR: qualification container ${container_id} remains after removal" >&2
return 1
fi
}
forget_container() {
local removed_id="$1" candidate
local retained=()
for candidate in "${CONTAINER_IDS[@]}"; do
[[ "${candidate}" == "${removed_id}" ]] || retained+=("${candidate}")
done
CONTAINER_IDS=("${retained[@]}")
}
remove_container_best_effort() {
local container_id="$1" observed_nonce
observed_nonce="$(inspect_container_nonce "${container_id}" 2>/dev/null)" || {
echo "WARNING: unable to inspect qualification container ${container_id}; manual cleanup may be required" >&2
return
}
if [[ "${observed_nonce}" != "${CONTAINER_RUN_NONCE}" ]]; then
echo "WARNING: refusing to remove qualification container after ownership-label mismatch: ${container_id}" >&2
return
fi
docker rm -f "${container_id}" >/dev/null 2>&1 || \
echo "WARNING: failed to remove qualification container ${container_id}" >&2
}
remove_image_strict() {
local observed_nonce
[[ "${IMAGE_CREATED}" == true ]] || return 0
observed_nonce="$(docker image inspect --format '{{ index .Config.Labels "org.pulse.rootful-qualification.run" }}' "${IMAGE_TAG}")" || return 1
if [[ "${observed_nonce}" != "${CONTAINER_RUN_NONCE}" ]]; then
echo "ERROR: qualification image ownership label changed" >&2
return 1
fi
docker image rm "${IMAGE_TAG}" >/dev/null || return 1
IMAGE_CREATED=false
}
cleanup() {
local container_id discovered discovered_ids=""
discovered_ids="$(docker ps -aq --no-trunc --filter "label=${CONTAINER_RUN_LABEL}=${CONTAINER_RUN_NONCE}" 2>/dev/null)" || true
for container_id in "${CONTAINER_IDS[@]}"; do
remove_container_best_effort "${container_id}"
done
while IFS= read -r discovered; do
[[ -n "${discovered}" ]] || continue
if [[ " ${CONTAINER_IDS[*]} " != *" ${discovered} "* ]]; then
remove_container_best_effort "${discovered}"
fi
done <<<"${discovered_ids}"
if [[ "${IMAGE_CREATED}" == true ]]; then
remove_image_strict >/dev/null 2>&1 || echo "WARNING: failed to remove qualification image ${IMAGE_TAG}" >&2
fi
if [[ "${PACKET_DIR}" == /tmp/pulse-rootful-packet.* && -d "${PACKET_DIR}" ]]; then
find "${PACKET_DIR}" -type f -exec chmod u+w {} + 2>/dev/null || true
rm -rf -- "${PACKET_DIR}"
fi
}
trap cleanup EXIT INT TERM
sha256_files() {
if command -v sha256sum >/dev/null 2>&1; then
sha256sum "$@"
else
shasum -a 256 "$@"
fi
}
mkdir "${OUTPUT_DIR}"
chmod 0700 "${OUTPUT_DIR}"
openssl genpkey -algorithm ED25519 -out "${PACKET_DIR}/update-private.pem"
chmod 0600 "${PACKET_DIR}/update-private.pem"
openssl pkey -in "${PACKET_DIR}/update-private.pem" -pubout -outform DER -out "${PACKET_DIR}/update-public.der"
chmod 0600 "${PACKET_DIR}/update-public.der"
update_public_key="$(python3 -I - "${PACKET_DIR}/update-public.der" <<'PY'
import base64
import pathlib
import sys
spki = pathlib.Path(sys.argv[1]).read_bytes()
prefix = bytes.fromhex("302a300506032b6570032100")
if len(spki) != len(prefix) + 32 or not spki.startswith(prefix):
raise SystemExit("unexpected Ed25519 SubjectPublicKeyInfo encoding")
print(base64.b64encode(spki[len(prefix):]).decode("ascii"), end="")
PY
)"
qualification_version="rootful-v1.${SOURCE_COMMIT:0:12}"
agent_ldflags="$(cd "${REPO_ROOT}" && ./scripts/release_ldflags.sh agent --version "${qualification_version}" --update-public-keys "${update_public_key}")"
helper_ldflags="$(cd "${REPO_ROOT}" && ./scripts/release_ldflags.sh agent --version "${qualification_version}" --update-public-keys "${update_public_key}")"
(
cd "${REPO_ROOT}"
CGO_ENABLED=0 GOOS=linux GOARCH="$(go env GOARCH)" GOFLAGS= GOWORK=off go build -trimpath -buildvcs=true -ldflags "${agent_ldflags}" -o "${PACKET_DIR}/pulse-agent" ./cmd/pulse-agent
CGO_ENABLED=0 GOOS=linux GOARCH="$(go env GOARCH)" GOFLAGS= GOWORK=off go build -trimpath -buildvcs=true -ldflags "${helper_ldflags}" -o "${PACKET_DIR}/pulse-agent-helper" ./cmd/pulse-agent-helper
CGO_ENABLED=0 GOOS=linux GOARCH="$(go env GOARCH)" GOFLAGS= GOWORK=off go test -c -trimpath -buildvcs=true -o "${PACKET_DIR}/dockeragent.test" ./scripts/installtests
)
openssl pkeyutl -sign -rawin -inkey "${PACKET_DIR}/update-private.pem" -in "${PACKET_DIR}/pulse-agent" | openssl base64 -A >"${PACKET_DIR}/pulse-agent.sig"
printf '\n' >>"${PACKET_DIR}/pulse-agent.sig"
rm -f "${PACKET_DIR}/update-private.pem" "${PACKET_DIR}/update-public.der"
unset update_public_key agent_ldflags helper_ldflags
install -m 0700 "${REPO_ROOT}/scripts/install.sh" "${PACKET_DIR}/install.sh"
chmod 0700 "${PACKET_DIR}/pulse-agent" "${PACKET_DIR}/pulse-agent-helper" "${PACKET_DIR}/dockeragent.test"
verify_vcs_artifact() {
local artifact="$1" expected_package="$2" metadata
metadata="$(go version -m "${artifact}")" || {
echo "ERROR: unable to inspect Go build metadata for ${artifact}" >&2
return 1
}
grep -Fq $'path\t'"${expected_package}" <<<"${metadata}" || {
echo "ERROR: artifact package identity mismatch for ${artifact}" >&2
return 1
}
grep -Fq "vcs.revision=${SOURCE_COMMIT}" <<<"${metadata}" || {
echo "ERROR: artifact vcs.revision does not match source commit: ${artifact}" >&2
return 1
}
grep -Fq 'vcs.modified=false' <<<"${metadata}" || {
echo "ERROR: artifact vcs.modified identity is not clean: ${artifact}" >&2
return 1
}
}
verify_vcs_artifact "${PACKET_DIR}/pulse-agent" "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent"
verify_vcs_artifact "${PACKET_DIR}/pulse-agent-helper" "github.com/rcourtman/pulse-go-rewrite/cmd/pulse-agent-helper"
verify_vcs_artifact "${PACKET_DIR}/dockeragent.test" "github.com/rcourtman/pulse-go-rewrite/scripts/installtests.test"
python3 -I - "${REPO_ROOT}" "${PACKET_DIR}/source-hashes.json" <<'PY'
import importlib.util
import json
import pathlib
import sys
checkout = pathlib.Path(sys.argv[1]).resolve(strict=True)
destination = pathlib.Path(sys.argv[2])
validator_path = checkout / "scripts/release_control/secure_runtime_rootful_attestation_v1.py"
spec = importlib.util.spec_from_file_location("rootful_validator", validator_path)
if spec is None or spec.loader is None:
raise SystemExit("unable to load rootful receipt validator")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
_, hashes = module.load_source_manifest(checkout, checkout / module.SOURCE_MANIFEST_PATH)
destination.write_text(json.dumps(hashes, sort_keys=True, separators=(",", ":")) + "\n")
PY
chmod 0600 "${PACKET_DIR}/source-hashes.json" "${PACKET_DIR}/pulse-agent.sig"
cat >"${PACKET_DIR}/Dockerfile" <<EOF
FROM ${UBUNTU_IMAGE}
ENV container=docker DEBIAN_FRONTEND=noninteractive
RUN apt-get update && apt-get install -y --no-install-recommends \
bash busybox-static ca-certificates curl dbus gnupg iproute2 jq kmod openssl passwd procps python3 \
software-properties-common systemd systemd-sysv podman && \
install -d -m 0755 /etc/apt/keyrings && \
curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc && \
chmod 0644 /etc/apt/keyrings/docker.asc && \
printf 'deb [arch=%s signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/ubuntu noble stable\n' "\$(dpkg --print-architecture)" >/etc/apt/sources.list.d/docker.list && \
apt-get update && apt-get install -y --no-install-recommends docker-ce docker-ce-cli containerd.io && \
apt-get clean && rm -rf /var/lib/apt/lists/* && \
ln -sf /dev/null /etc/systemd/system/docker.service && \
ln -sf /dev/null /etc/systemd/system/docker.socket && \
ln -sf /dev/null /etc/systemd/system/podman.service && \
ln -sf /dev/null /etc/systemd/system/podman.socket && \
install -d -m 0700 /opt/pulse/packet /opt/pulse/result && \
printf '%s\n' disposable-v1 >/etc/pulse-secure-runtime-rootful-qualification && \
rm -f /etc/machine-id && touch /etc/machine-id && \
systemctl set-default multi-user.target
STOPSIGNAL SIGRTMIN+3
CMD ["/sbin/init"]
EOF
docker build --pull --no-cache --network default \
--label "${CONTAINER_RUN_LABEL}=${CONTAINER_RUN_NONCE}" \
-t "${IMAGE_TAG}" -f "${PACKET_DIR}/Dockerfile" "${PACKET_DIR}" | tee "${OUTPUT_DIR}/image-build.log"
IMAGE_CREATED=true
docker image inspect "${IMAGE_TAG}" >"${OUTPUT_DIR}/qualification-image-inspect.json"
chmod 0600 "${OUTPUT_DIR}/image-build.log" "${OUTPUT_DIR}/qualification-image-inspect.json"
capture_qualification_container_diagnostics() {
local runtime_name="$1" container_id="$2"
docker logs "${container_id}" >"${OUTPUT_DIR}/${runtime_name}-container.log" 2>&1 || true
docker exec "${container_id}" journalctl --no-pager -n 2000 >"${OUTPUT_DIR}/${runtime_name}-journal.log" 2>&1 || true
chmod 0600 "${OUTPUT_DIR}/${runtime_name}-container.log" "${OUTPUT_DIR}/${runtime_name}-journal.log"
}
run_runtime() {
local runtime_name="$1"
local container_name="pulse-rootful-qual-${runtime_name}-${SOURCE_COMMIT:0:8}-$$"
local container_id local_receipt machine_id_file machine_id deadline mounts
local_receipt="${OUTPUT_DIR}/${runtime_name}-receipt.json"
machine_id_file="${PACKET_DIR}/.machine-id-${runtime_name}"
machine_id="$(openssl rand -hex 16)"
if [[ ! "${machine_id}" =~ ^[0-9a-f]{32}$ || "${machine_id}" == "00000000000000000000000000000000" ]]; then
echo "ERROR: unable to generate a valid machine ID for ${runtime_name}" >&2
return 1
fi
printf '%s\n' "${machine_id}" >"${machine_id_file}"
chmod 0444 "${machine_id_file}"
container_id="$(docker create --name "${container_name}" --hostname "pulse-rootful-${runtime_name}" \
--label "${CONTAINER_RUN_LABEL}=${CONTAINER_RUN_NONCE}" \
--privileged --network none --cgroupns=private \
--tmpfs /run:rw,nosuid,nodev,mode=755 --tmpfs /run/lock:rw,nosuid,nodev,mode=755 \
"${IMAGE_TAG}")"
CONTAINER_IDS+=("${container_id}")
docker cp "${machine_id_file}" "${container_id}:/etc/machine-id"
rm -f -- "${machine_id_file}"
docker cp "${PACKET_DIR}/." "${container_id}:/opt/pulse/packet"
docker start "${container_id}" >/dev/null
deadline=$((SECONDS + 60))
until docker exec "${container_id}" systemctl is-system-running --wait >/dev/null 2>&1; do
if (( SECONDS >= deadline )); then
capture_qualification_container_diagnostics "${runtime_name}" "${container_id}"
echo "ERROR: ${runtime_name} disposable systemd container did not become ready" >&2
return 1
fi
sleep 1
done
if docker exec "${container_id}" sh -c 'ip route | grep -q "^default "'; then
echo "ERROR: ${runtime_name} qualification container unexpectedly has a default route" >&2
return 1
fi
mounts="$(docker inspect "${container_id}" --format '{{range .Mounts}}{{println .Source "->" .Destination}}{{end}}')"
if grep -E '/(var/)?run/(docker|podman)(\.sock)?' <<<"${mounts}"; then
echo "ERROR: host runtime socket was mounted into ${runtime_name} qualification container" >&2
return 1
fi
if ! docker exec \
-e PULSE_SECURE_RUNTIME_ROOTFUL_QUALIFICATION=disposable-v1 \
-e "PULSE_ROOTFUL_RUNTIME=${runtime_name}" \
-e PULSE_ROOTFUL_RECEIPT=/opt/pulse/result/rootful-receipt.json \
-e PULSE_ROOTFUL_SOURCE_HASHES=/opt/pulse/packet/source-hashes.json \
-e "PULSE_ROOTFUL_SOURCE_COMMIT=${SOURCE_COMMIT}" \
-e PULSE_SECURE_RUNTIME_COLLECTOR=/opt/pulse/packet/pulse-agent \
-e PULSE_SECURE_RUNTIME_COLLECTOR_SIGNATURE=/opt/pulse/packet/pulse-agent.sig \
-e PULSE_SECURE_RUNTIME_HELPER=/opt/pulse/packet/pulse-agent-helper \
-e PULSE_SECURE_RUNTIME_INSTALLER=/opt/pulse/packet/install.sh \
"${container_id}" /opt/pulse/packet/dockeragent.test \
-test.run '^TestSecureRuntimeRootfulQualification$' -test.count=1 -test.v -test.timeout=45m \
| tee "${OUTPUT_DIR}/${runtime_name}-test.log"; then
capture_qualification_container_diagnostics "${runtime_name}" "${container_id}"
chmod 0600 "${OUTPUT_DIR}/${runtime_name}-test.log"
return 1
fi
docker exec "${container_id}" test -f /opt/pulse/result/rootful-receipt.json || {
capture_qualification_container_diagnostics "${runtime_name}" "${container_id}"
echo "ERROR: ${runtime_name} qualification did not retain its receipt" >&2
return 1
}
docker cp "${container_id}:/opt/pulse/result/rootful-receipt.json" "${local_receipt}"
capture_qualification_container_diagnostics "${runtime_name}" "${container_id}"
chmod 0600 "${local_receipt}" "${OUTPUT_DIR}/${runtime_name}-test.log"
remove_container_strict "${container_id}"
forget_container "${container_id}"
}
run_runtime docker
run_runtime podman
python3 -I - "${OUTPUT_DIR}/docker-receipt.json" "${OUTPUT_DIR}/podman-receipt.json" "${OUTPUT_DIR}/receipt.json" <<'PY'
import json
import pathlib
import sys
docker_path, podman_path, output_path = map(pathlib.Path, sys.argv[1:])
docker = json.loads(docker_path.read_text())
podman = json.loads(podman_path.read_text())
if docker.get("result") != "passed" or podman.get("result") != "passed":
raise SystemExit('per-runtime qualification result != "passed"')
for field in ("schema_version", "kind", "source_commit", "source_hashes", "artifacts"):
if docker.get(field) != podman.get(field):
raise SystemExit(f"per-runtime qualification field differs: {field}")
runs = docker.get("runs", []) + podman.get("runs", [])
if [run.get("runtime", {}).get("runtime") for run in runs] != ["docker", "podman"]:
raise SystemExit("per-runtime receipts are not exact Docker then Podman runs")
machine_ids = [run.get("host", {}).get("machine_id") for run in runs]
daemon_ids = [run.get("runtime", {}).get("daemon_id") for run in runs]
if len(set(machine_ids)) != 2 or len(set(daemon_ids)) != 2:
raise SystemExit("Docker and Podman qualification hosts/daemons must have distinct identities")
combined = {
"schema_version": docker["schema_version"], "kind": docker["kind"], "result": "passed",
"source_commit": docker["source_commit"],
"started_at": min(docker["started_at"], podman["started_at"]),
"completed_at": max(docker["completed_at"], podman["completed_at"]),
"source_hashes": docker["source_hashes"], "artifacts": docker["artifacts"], "runs": runs,
}
output_path.write_text(json.dumps(combined, indent=2, sort_keys=True) + "\n")
PY
chmod 0600 "${OUTPUT_DIR}/receipt.json"
python3 -I "${REPO_ROOT}/scripts/release_control/secure_runtime_rootful_attestation_v1.py" \
"${OUTPUT_DIR}/receipt.json" \
--qualification-test "${PACKET_DIR}/dockeragent.test" \
--collector "${PACKET_DIR}/pulse-agent" \
--helper "${PACKET_DIR}/pulse-agent-helper" \
--installer "${PACKET_DIR}/install.sh" \
>"${OUTPUT_DIR}/attestation.json"
chmod 0600 "${OUTPUT_DIR}/attestation.json"
sha256_files "${OUTPUT_DIR}/receipt.json" "${OUTPUT_DIR}/attestation.json" >"${OUTPUT_DIR}/sha256.txt"
chmod 0600 "${OUTPUT_DIR}/sha256.txt"
if [[ -n "$(docker ps -aq --no-trunc --filter "label=${CONTAINER_RUN_LABEL}=${CONTAINER_RUN_NONCE}")" ]]; then
echo "ERROR: labeled qualification containers remain after strict cleanup" >&2
exit 1
fi
remove_image_strict
if [[ -n "$(docker images -q --filter "label=${CONTAINER_RUN_LABEL}=${CONTAINER_RUN_NONCE}")" ]]; then
echo "ERROR: labeled qualification images remain after strict cleanup" >&2
exit 1
fi
while IFS= read -r retained; do
[[ -z "${retained}" ]] && continue
if [[ "$(portable_uid "${retained}")" != "$(id -u)" ]]; then
echo "ERROR: retained output is not owned by the invoking user: ${retained}" >&2
exit 1
fi
chmod go-rwx "${retained}"
done < <(find "${OUTPUT_DIR}" -mindepth 1 -maxdepth 1 -type f -print)
echo "Rootful qualification passed: ${OUTPUT_DIR}"