Message Attestation: example with out-of-process policy server

internal/examples/agent/agent/agent.go

@@ -27,6 +27,7 @@ import (
 	"github.com/open-telemetry/opamp-go/client/types"
 	"github.com/open-telemetry/opamp-go/internal/examples/config"
 	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/open-telemetry/opamp-go/signing"
 )
 
 var localConfig = []byte(`
@@ -87,6 +88,15 @@ type Agent struct {
 	// lastConnectionSettingsHash stores the hash of the most recently received
 	// ConnectionSettingsOffers, used when reporting connection settings status.
 	lastConnectionSettingsHash []byte
+
+	// payloadVerifier, when non-nil, enables Message Attestation: every
+	// inbound ServerToAgent message must arrive in a SignedServerToAgent
+	// envelope whose signature chains to this verifier's trust anchor.
+	payloadVerifier signing.Verifier
+
+	// payloadTOFUStore, when non-nil, enables TOFU enrollment for the
+	// payload trust anchor. Mutually exclusive with payloadVerifier.
+	payloadTOFUStore signing.TOFUStore
 }
 
 type proxySettings struct {
@@ -138,6 +148,25 @@ func WithNoClientCertRequest() Option {
 	}
 }
 
+// WithPayloadVerifier enables Message Attestation. Every inbound ServerToAgent
+// message must arrive in a SignedServerToAgent envelope whose signature chains
+// to the trust anchor embedded in v.
+func WithPayloadVerifier(v signing.Verifier) Option {
+	return func(agent *Agent) {
+		agent.payloadVerifier = v
+	}
+}
+
+// WithPayloadTOFUStore enables TOFU enrollment for the payload trust anchor.
+// On first connection the agent accepts and persists the root CA delivered by
+// the server; on subsequent connections the persisted anchor is used directly.
+// Mutually exclusive with WithPayloadVerifier.
+func WithPayloadTOFUStore(s signing.TOFUStore) Option {
+	return func(agent *Agent) {
+		agent.payloadTOFUStore = s
+	}
+}
+
 func NewAgent(agentConfig *config.AgentConfig, options ...Option) *Agent {
 	agent := &Agent{
 		logger:          &Logger{Logger: log.Default()},
@@ -190,6 +219,8 @@ func (agent *Agent) connect(ops ...settingsOp) error {
 		OpAMPServerURL:    agent.agentConfig.Endpoint,
 		HeartbeatInterval: agent.agentConfig.HeartbeatInterval,
 		InstanceUid:       types.InstanceUid(agent.instanceId),
+		PayloadVerifier:   agent.payloadVerifier,
+		PayloadTOFUStore:  agent.payloadTOFUStore,
 		Callbacks: types.Callbacks{
 			OnConnect: func(ctx context.Context) {
 				agent.logger.Debugf(ctx, "Connected to the server.")
@@ -232,6 +263,12 @@ func (agent *Agent) connect(ops ...settingsOp) error {
 		protobufs.AgentCapabilities_AgentCapabilities_ReportsOwnMetrics |
 		protobufs.AgentCapabilities_AgentCapabilities_AcceptsOpAMPConnectionSettings |
 		protobufs.AgentCapabilities_AgentCapabilities_ReportsConnectionSettingsStatus
+	if agent.payloadVerifier != nil {
+		supportedCapabilities |= protobufs.AgentCapabilities_AgentCapabilities_RequiresPayloadTrustVerification
+	} else if agent.payloadTOFUStore != nil {
+		supportedCapabilities |= protobufs.AgentCapabilities_AgentCapabilities_RequiresPayloadTrustVerification |
+			protobufs.AgentCapabilities_AgentCapabilities_AcceptsPayloadTrustAnchorTOFU
+	}
 	err = agent.client.SetCapabilities(&supportedCapabilities)
 	if err != nil {
 		return err

internal/examples/agent/main.go

@@ -15,6 +15,7 @@ import (
 	opampinternal "github.com/open-telemetry/opamp-go/internal"
 	"github.com/open-telemetry/opamp-go/internal/examples/agent/agent"
 	"github.com/open-telemetry/opamp-go/internal/examples/config"
+	"github.com/open-telemetry/opamp-go/signing"
 
 	"github.com/google/uuid"
 	"go.opentelemetry.io/collector/config/configtls"
@@ -42,6 +43,18 @@ type flagConfig struct {
 	// scaleCount = 1 runs a normal agent
 	// scaleCount > 1 runs scale test agents (pre-assigned IDs, no initial cert request)
 	scaleCount uint64
+	// attestationCAFile, when non-empty, enables Message Attestation: every
+	// inbound ServerToAgent must carry a valid signature chaining to the CA
+	// certificate stored at this path. Run internal/examples/policysrv first
+	// to generate the CA and start the signing server.
+	attestationCAFile string
+
+	// attestationTOFUStoreFile, when non-empty, enables TOFU enrollment for
+	// Message Attestation. On the first connection the agent accepts and
+	// persists the root CA delivered by the server; on subsequent connections
+	// the persisted anchor is loaded from this file. Mutually exclusive with
+	// --attestation-ca.
+	attestationTOFUStoreFile string
 }
 
 func (cfg flagConfig) verifyArgs() error {
@@ -157,6 +170,14 @@ func loadEnv(cfg *flagConfig) {
 			cfg.scaleCount = count
 		}
 	}
+
+	if s, ok := os.LookupEnv("AGENT_ATTESTATION_CA"); ok {
+		cfg.attestationCAFile = s
+	}
+
+	if s, ok := os.LookupEnv("AGENT_ATTESTATION_TOFU_STORE"); ok {
+		cfg.attestationTOFUStoreFile = s
+	}
 }
 
 func main() {
@@ -172,6 +193,20 @@ func main() {
 	flag.DurationVar(&cfg.heartbeat, "heartbeat", time.Second*30, "Heartbeat duration (env var: AGENT_HEARTBEAT).")
 	flag.BoolVar(&cfg.quietAgent, "quite-agent", false, "Disable agent logger (env var: AGENT_QUIET).")
 	flag.Uint64Var(&cfg.scaleCount, "scale-count", 1, "The number of agents to start in scale mode (env var: AGENT_SCALE_COUNT).")
+	flag.StringVar(&cfg.attestationCAFile, "attestation-ca", "",
+		"Path to a PEM-encoded CA certificate used to verify signed ServerToAgent messages\n"+
+			"(Message Attestation). When set, the agent declares the\n"+
+			"RequiresPayloadTrustVerification capability and rejects any message whose\n"+
+			"signature does not chain to this CA. Obtain the CA from the policy server's\n"+
+			"/v1/ca endpoint or from /tmp/opamp-policy-ca.pem after running policysrv\n"+
+			"(env var: AGENT_ATTESTATION_CA).")
+	flag.StringVar(&cfg.attestationTOFUStoreFile, "attestation-tofu-store", "",
+		"Path to a file used as the TOFU trust anchor store for Message Attestation.\n"+
+			"On the first connection the agent accepts and persists the root CA delivered\n"+
+			"by the server; on subsequent connections the persisted anchor is loaded from\n"+
+			"this file. Disabled by default. Mutually exclusive with --attestation-ca.\n"+
+			"WARNING: TOFU provides no security on the first connection.\n"+
+			"(env var: AGENT_ATTESTATION_TOFU_STORE).")
 
 	flag.Parse()
 	loadEnv(&cfg)
@@ -211,6 +246,26 @@ func main() {
 // If an error is encountered when starting an agent, it is return along with all started agents.
 func runScale(ctx context.Context, cfg flagConfig) ([]*agent.Agent, error) {
 	nopLogger := &opampinternal.NopLogger{}
+
+	var verifier signing.Verifier
+	if cfg.attestationCAFile != "" {
+		v, err := signing.VerifierFromFile(cfg.attestationCAFile)
+		if err != nil {
+			return nil, fmt.Errorf("load attestation CA: %w", err)
+		}
+		verifier = v
+		log.Printf("Message Attestation enabled — trust anchor: %s", cfg.attestationCAFile)
+	}
+
+	var tofuStore signing.TOFUStore
+	if cfg.attestationTOFUStoreFile != "" {
+		if cfg.attestationCAFile != "" {
+			return nil, fmt.Errorf("--attestation-ca and --attestation-tofu-store are mutually exclusive")
+		}
+		tofuStore = signing.NewFileTOFUStore(cfg.attestationTOFUStoreFile)
+		log.Printf("Message Attestation TOFU enrollment enabled — store: %s", cfg.attestationTOFUStoreFile)
+	}
+
 	agentConfig := &config.AgentConfig{
 		Endpoint:          cfg.endpoint,
 		HeartbeatInterval: &cfg.heartbeat,
@@ -239,6 +294,12 @@ func runScale(ctx context.Context, cfg flagConfig) ([]*agent.Agent, error) {
 			agent.WithAgentType(cfg.agentType),
 			agent.WithAgentVersion(cfg.agentVersion),
 		}
+		if verifier != nil {
+			opts = append(opts, agent.WithPayloadVerifier(verifier))
+		}
+		if tofuStore != nil {
+			opts = append(opts, agent.WithPayloadTOFUStore(tofuStore))
+		}
 		if cfg.quietAgent {
 			opts = append(opts, agent.WithLogger(nopLogger))
 		}

internal/examples/policysrv/main.go

@@ -0,0 +1,164 @@
+// policysrv is a minimal example of an out-of-process OpAMP policy/signing
+// server, as described in supplementary-guidelines.md.
+//
+// # Architecture
+//
+// The OpAMP distribution server (internal/examples/server) holds no private
+// key material. Before delivering each ServerToAgent message it calls this
+// server's /v1/sign endpoint to obtain a signature over the payload bytes.
+// The structural isolation is the key security property: an attacker who
+// compromises the distribution server gains the ability to send messages,
+// but cannot produce valid signatures without also compromising this server.
+//
+//	Agent ──AgentToServer──► OpAMP Server ──sign request──► Policy Server
+//	                                       ◄──signature──────────────────
+//	     ◄──SignedServerToAgent───────────
+//
+// # Policy enforcement
+//
+// In this example the server signs every request unconditionally. A
+// production policy server would decode the ServerToAgent payload before
+// signing and reject messages that violate organizational constraints, for
+// example:
+//   - Deny ServerToAgentCommand messages to immutable agents.
+//   - Enforce per-team RemoteConfig ownership.
+//   - Require that only the latest approved component version may be installed.
+//
+// # Usage (three separate terminals, all run from internal/examples/)
+//
+//	# Terminal 1 – policy/signing server
+//	go run ./policysrv
+//
+//	# Terminal 2 – OpAMP distribution server (points at policy server)
+//	go run ./server --policy-server http://localhost:4322
+//
+//	# Terminal 3 – agent (pre-provisioned with the CA cert written above)
+//	go run ./agent --attestation-ca /tmp/opamp-policy-ca.pem
+//
+// In production the CA certificate would be distributed out-of-band via
+// configuration management tooling (Ansible, Chef, Puppet, or a secrets
+// manager), or compiled into the agent binary. The /v1/ca endpoint is
+// provided for demo convenience only and is not part of the OpAMP protocol.
+package main
+
+import (
+	"context"
+	"crypto/x509"
+	"encoding/pem"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/open-telemetry/opamp-go/signing"
+)
+
+const (
+	listenAddr = ":4322"
+	caOutPath  = "/tmp/opamp-policy-ca.pem"
+)
+
+func main() {
+	// Generate an ephemeral CA and signing leaf.
+	// In production: load the CA and leaf private key from an HSM or
+	// secrets manager; never write the private key to disk.
+	ca, caKey, err := signing.GenerateCA(signing.AlgorithmECDSAP256SHA256, signing.CertOptions{})
+	if err != nil {
+		log.Fatalf("generate CA: %v", err)
+	}
+	leaf, leafKey, err := signing.GenerateLeaf(signing.AlgorithmECDSAP256SHA256, ca, caKey, signing.CertOptions{
+		// SAN required by the spec: the leaf must match the OpAMP distribution server's hostname.
+		// The example server binds to 0.0.0.0:4320; agents may connect by hostname or IP, so
+		// include both. Production deployments set these to the actual hostname(s) or IP(s).
+		DNSNames:    []string{"localhost"},
+		IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
+	})
+	if err != nil {
+		log.Fatalf("generate leaf: %v", err)
+	}
+	localSigner, err := signing.NewLocalSigner(leafKey, []*x509.Certificate{leaf})
+	if err != nil {
+		log.Fatalf("new signer: %v", err)
+	}
+	signer := localSigner.WithRootCA(ca)
+
+	caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.Raw})
+	leafPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: leaf.Raw})
+
+	// Write the CA cert for agents to load as their payload trust anchor.
+	// In production this step is replaced by your configuration-management
+	// pipeline; do not derive trust anchors from the network at runtime.
+	if err := os.WriteFile(caOutPath, caPEM, 0o644); err != nil {
+		log.Fatalf("write CA cert: %v", err)
+	}
+	log.Printf("CA certificate → %s", caOutPath)
+
+	mux := http.NewServeMux()
+
+	// POST /v1/sign
+	// The OpAMP server sends the serialised ServerToAgent payload here.
+	// This handler signs it and returns the raw signature bytes.
+	//
+	// Production note: decode the payload (proto.Unmarshal into
+	// protobufs.ServerToAgent) here to apply policy before signing.
+	mux.HandleFunc("POST /v1/sign", func(w http.ResponseWriter, r *http.Request) {
+		payload, err := io.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+		sig, err := signer.Sign(r.Context(), payload)
+		if err != nil {
+			http.Error(w, "sign: "+err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/octet-stream")
+		_, _ = w.Write(sig)
+		log.Printf("[policy] signed %d-byte payload → %d-byte signature", len(payload), len(sig))
+	})
+
+	// GET /v1/chain
+	// Returns the PEM-encoded signing certificate chain (leaf only here;
+	// include any intermediates between the leaf and the root CA).
+	// The root CA is excluded — agents already possess it as their trust anchor.
+	mux.HandleFunc("GET /v1/chain", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/x-pem-file")
+		_, _ = w.Write(leafPEM)
+	})
+
+	// GET /v1/ca
+	// Returns the CA certificate in PEM.
+	// Demo convenience only — not part of the OpAMP protocol. In
+	// production, provision the CA cert out-of-band.
+	mux.HandleFunc("GET /v1/ca", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/x-pem-file")
+		_, _ = w.Write(caPEM)
+	})
+
+	ln, err := net.Listen("tcp", listenAddr)
+	if err != nil {
+		log.Fatalf("listen on %s: %v", listenAddr, err)
+	}
+	srv := &http.Server{Handler: mux}
+
+	log.Printf("Policy server listening on %s", listenAddr)
+	log.Println("Next steps (run from internal/examples/):")
+	log.Printf("  OpAMP server: go run ./server --policy-server http://localhost%s", listenAddr)
+	log.Printf("  Agent:        go run ./agent --attestation-ca %s", caOutPath)
+
+	stop := make(chan os.Signal, 1)
+	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {
+			log.Fatalf("serve: %v", err)
+		}
+	}()
+
+	<-stop
+	log.Println("Shutting down…")
+	_ = srv.Shutdown(context.Background())
+}

internal/examples/server/main.go

@@ -9,6 +9,7 @@ import (
 	"github.com/open-telemetry/opamp-go/internal/examples/server/data"
 	"github.com/open-telemetry/opamp-go/internal/examples/server/opampsrv"
 	"github.com/open-telemetry/opamp-go/internal/examples/server/uisrv"
+	"github.com/open-telemetry/opamp-go/signing"
 )
 
 var logger = log.New(log.Default().Writer(), "[MAIN] ", log.Default().Flags()|log.Lmsgprefix|log.Lmicroseconds)
@@ -17,17 +18,32 @@ func main() {
 	var emitMetrics bool
 	flag.BoolVar(&emitMetrics, "emit-metrics", false, "Emit metrics to stdout.")
 
+	var policyServerURL string
+	flag.StringVar(&policyServerURL, "policy-server", "",
+		"Base URL of the out-of-process policy/signing server (e.g. http://localhost:4322).\n"+
+			"When set, every outbound ServerToAgent message is signed via that server,\n"+
+			"demonstrating the isolated signing architecture from supplementary-guidelines.md.\n"+
+			"Run internal/examples/policysrv first to start a local policy server.")
+
 	flag.Parse()
 
 	curDir, err := os.Getwd()
 	if err != nil {
 		panic(err)
 	}
 
+	// If a policy server URL is provided, create a RemoteSigner that delegates
+	// all signing to it. The OpAMP server itself never touches the private key.
+	var payloadSigner signing.Signer
+	if policyServerURL != "" {
+		payloadSigner = signing.NewRemoteSigner(policyServerURL)
+		logger.Printf("Message Attestation enabled — signing via policy server at %s", policyServerURL)
+	}
+
 	logger.Println("OpAMP Server starting...")
 
 	uisrv.Start(curDir)
-	opampSrv := opampsrv.NewServer(&data.AllAgents, emitMetrics)
+	opampSrv := opampsrv.NewServer(&data.AllAgents, emitMetrics, payloadSigner)
 	opampSrv.Start()
 
 	logger.Println("OpAMP Server running...")