From e2d006de4e6f075a9eca64e419a6f9d0040033d7 Mon Sep 17 00:00:00 2001 From: Stanley Liu Date: Thu, 2 Jul 2026 17:31:15 -0400 Subject: [PATCH] Add Message Attestation example with out-of-process policy server Demonstrates the recommended production deployment pattern from supplementary-guidelines.md: the OpAMP distribution server holds no private key material. Signing is delegated to a separate policy server that can inspect the payload, enforce organisational policies, and delegate to an HSM before returning a signature. policysrv: standalone HTTP signing service exposing /v1/sign, /v1/chain, /v1/ca. Run it alongside the example server to see the full attestation flow end-to-end. Example server: --policy-server flag wires a RemoteSigner so the OpAMP server signs every outbound ServerToAgent through the policy server. Example agent: --attestation-ca configures a pre-provisioned trust anchor; --attestation-tofu-store enables TOFU enrollment for environments where out-of-band CA provisioning is impractical. --- internal/examples/agent/agent/agent.go | 37 ++++ internal/examples/agent/main.go | 61 +++++++ internal/examples/policysrv/main.go | 164 ++++++++++++++++++ internal/examples/server/main.go | 18 +- internal/examples/server/opampsrv/opampsrv.go | 25 ++- 5 files changed, 296 insertions(+), 9 deletions(-) create mode 100644 internal/examples/policysrv/main.go diff --git a/internal/examples/agent/agent/agent.go b/internal/examples/agent/agent/agent.go index 96fa5efd..5eac11b1 100644 --- a/internal/examples/agent/agent/agent.go +++ b/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 diff --git a/internal/examples/agent/main.go b/internal/examples/agent/main.go index 9a221a05..3b9088f8 100644 --- a/internal/examples/agent/main.go +++ b/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)) } diff --git a/internal/examples/policysrv/main.go b/internal/examples/policysrv/main.go new file mode 100644 index 00000000..9da5d941 --- /dev/null +++ b/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()) +} diff --git a/internal/examples/server/main.go b/internal/examples/server/main.go index 349190c1..b82a867a 100644 --- a/internal/examples/server/main.go +++ b/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,6 +18,13 @@ 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() @@ -24,10 +32,18 @@ func main() { 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...") diff --git a/internal/examples/server/opampsrv/opampsrv.go b/internal/examples/server/opampsrv/opampsrv.go index 8b7f702b..7f2de6e2 100644 --- a/internal/examples/server/opampsrv/opampsrv.go +++ b/internal/examples/server/opampsrv/opampsrv.go @@ -16,16 +16,23 @@ import ( "github.com/open-telemetry/opamp-go/protobufs" "github.com/open-telemetry/opamp-go/server" "github.com/open-telemetry/opamp-go/server/types" + "github.com/open-telemetry/opamp-go/signing" ) type Server struct { - opampSrv server.OpAMPServer - agents *data.Agents - logger *Logger - metrics *metricsTracker + opampSrv server.OpAMPServer + agents *data.Agents + logger *Logger + metrics *metricsTracker + payloadSigner signing.Signer } -func NewServer(agents *data.Agents, emitMetrics bool) *Server { +// NewServer creates a new OpAMP server. payloadSigner, when non-nil, enables +// Message Attestation: every outbound ServerToAgent message is wrapped in a +// SignedServerToAgent envelope signed by the given signer. Use +// signing.NewRemoteSigner to delegate signing to an out-of-process policy +// server as recommended in supplementary-guidelines.md. +func NewServer(agents *data.Agents, emitMetrics bool, payloadSigner signing.Signer) *Server { logger := &Logger{ log.New( log.Default().Writer(), @@ -40,9 +47,10 @@ func NewServer(agents *data.Agents, emitMetrics bool) *Server { } srv := &Server{ - agents: agents, - logger: logger, - metrics: metrics, + agents: agents, + logger: logger, + metrics: metrics, + payloadSigner: payloadSigner, } srv.opampSrv = server.New(logger) @@ -53,6 +61,7 @@ func NewServer(agents *data.Agents, emitMetrics bool) *Server { func (srv *Server) Start() { settings := server.StartSettings{ Settings: server.Settings{ + PayloadSigner: srv.payloadSigner, Callbacks: types.Callbacks{ OnConnecting: func(request *http.Request) types.ConnectionResponse { return types.ConnectionResponse{