feat: implement rotation handling

internal/config/case.go

@@ -119,6 +119,13 @@
 	// can't decode columnar objects, and the s3 receiver's destructive drain
 	// would corrupt the read. See VerifierConfig.
 	Verifier *VerifierConfig `yaml:"verifier"`
+
+	// Rotation, when set, parameterizes the
+	// director_agent_tls_cert_rotation_correctness driver: which director TLS
+	// cert/CA rotation to perform mid-run and how the enrolled agent is expected
+	// to respond. Required by (and only meaningful for) that type. See
+	// RotationConfig and runDirectorAgentCertRotation.
+	Rotation *RotationConfig `yaml:"rotation"`
 }
 
 // VerifierConfig configures the post-drain DuckDB verifier container (see
@@ -425,6 +432,65 @@
 	return "secret"
 }
 
+// Rotation parameterizes the director_agent_tls_cert_rotation_correctness
+// driver (see TestCase.Rotation, runDirectorAgentCertRotation). The director
+// deploys an agent that streams back over the proxy_tls listener; mid-run the
+// director's serving cert/CA is rotated on disk and the director is bounced so
+// the enrolled agent must re-handshake. Mode selects which rotation and the
+// expected agent response.
+type RotationConfig struct {
+	// Mode selects the mid-run rotation:
+	//   "same_ca"        — re-sign the director leaf under the SAME CA
+	//                      (RotateServerCert). The agent reconnects transparently
+	//                      because the chain still validates. Verdict: delivery
+	//                      resumes (count grows after the bounce).
+	//   "new_ca_recover" — rotate to a BRAND-NEW CA written to ca.crt and served
+	//                      at /dl/cert.pem (RotateServerCertNewCA). A bootstrap
+	//                      agent (no operator-pinned CA) must re-fetch the new CA
+	//                      and reconnect. Verdict: delivery resumes.
+	//   "new_ca_reject"  — TWO PHASE. Phase 1 re-signs the leaf under an UNTRUSTED
+	//                      CA the director never serves (RotateServerCertWrongCA):
+	//                      the agent MUST fail validation, so delivery STALLS
+	//                      (a missing stall is a SECURITY failure — validation is
+	//                      disabled). Phase 2 restores a trusted leaf
+	//                      (RotateServerCert) and delivery must resume.
+	Mode string `yaml:"mode"`
+
+	// SettleSeconds is the pause after a rotation+bounce before the driver samples
+	// the receiver, giving the director time to rebind its listener and the agent
+	// time to detect the dropped session and reconnect (default 25s).
+	SettleSeconds int `yaml:"settle_seconds"`
+
+	// StallSeconds (new_ca_reject only) is how long the receiver count must hold
+	// flat after the untrusted rotation for the bad cert to count as rejected
+	// (default 20s). The case's endpoint seed loop MUST still be appending fresh
+	// records during this window, else the stall is vacuous — see the case NOTES.
+	StallSeconds int `yaml:"stall_seconds"`
+}
+
+// Rotation mode values for RotationConfig.Mode.
+const (
+	RotationSameCA       = "same_ca"
+	RotationNewCARecover = "new_ca_recover"
+	RotationNewCAReject  = "new_ca_reject"
+)
+
+// SettleSecondsOrDefault / StallSecondsOrDefault centralize the rotation timing
+// defaults so the driver and any caller agree.
+func (rc *RotationConfig) SettleSecondsOrDefault() int {
+	if rc != nil && rc.SettleSeconds > 0 {
+		return rc.SettleSeconds
+	}
+	return 25
+}
+
+func (rc *RotationConfig) StallSecondsOrDefault() int {
+	if rc != nil && rc.StallSeconds > 0 {
+		return rc.StallSeconds
+	}
+	return 20
+}
+
 // AgentConfig configures an external agent container in the test topology
 // (see TestCase.Agent). Unlike endpoints (which the subject connects out to),
 // the agent connects INTO the subject over the bench network — it starts after
@@ -448,6 +514,65 @@
 	MountsSharedData bool `yaml:"mounts_shared_data"`
 }
 
+// Rotation parameterizes the director_agent_tls_cert_rotation_correctness
+// driver (see TestCase.Rotation, runDirectorAgentCertRotation). The director
+// deploys an agent that streams back over the proxy_tls listener; mid-run the
+// director's serving cert/CA is rotated on disk and the director is bounced so
+// the enrolled agent must re-handshake. Mode selects which rotation and the
+// expected agent response.
+type RotationConfig struct {
+	// Mode selects the mid-run rotation:
+	//   "same_ca"        — re-sign the director leaf under the SAME CA
+	//                      (RotateServerCert). The agent reconnects transparently
+	//                      because the chain still validates. Verdict: delivery
+	//                      resumes (count grows after the bounce).
+	//   "new_ca_recover" — rotate to a BRAND-NEW CA written to ca.crt and served
+	//                      at /dl/cert.pem (RotateServerCertNewCA). A bootstrap
+	//                      agent (no operator-pinned CA) must re-fetch the new CA
+	//                      and reconnect. Verdict: delivery resumes.
+	//   "new_ca_reject"  — TWO PHASE. Phase 1 re-signs the leaf under an UNTRUSTED
+	//                      CA the director never serves (RotateServerCertWrongCA):
+	//                      the agent MUST fail validation, so delivery STALLS
+	//                      (a missing stall is a SECURITY failure — validation is
+	//                      disabled). Phase 2 restores a trusted leaf
+	//                      (RotateServerCert) and delivery must resume.
+	Mode string `yaml:"mode"`
+
+	// SettleSeconds is the pause after a rotation+bounce before the driver samples
+	// the receiver, giving the director time to rebind its listener and the agent
+	// time to detect the dropped session and reconnect (default 25s).
+	SettleSeconds int `yaml:"settle_seconds"`
+
+	// StallSeconds (new_ca_reject only) is how long the receiver count must hold
+	// flat after the untrusted rotation for the bad cert to count as rejected
+	// (default 20s). The case's endpoint seed loop MUST still be appending fresh
+	// records during this window, else the stall is vacuous — see the case NOTES.
+	StallSeconds int `yaml:"stall_seconds"`
+}
+
+// Rotation mode values for RotationConfig.Mode.
+const (
+	RotationSameCA       = "same_ca"
+	RotationNewCARecover = "new_ca_recover"
+	RotationNewCAReject  = "new_ca_reject"
+)
+
+// SettleSecondsOrDefault / StallSecondsOrDefault centralize the rotation timing
+// defaults so the driver and any caller agree.
+func (rc *RotationConfig) SettleSecondsOrDefault() int {
+	if rc != nil && rc.SettleSeconds > 0 {
+		return rc.SettleSeconds
+	}
+	return 25
+}
+
+func (rc *RotationConfig) StallSecondsOrDefault() int {
+	if rc != nil && rc.StallSeconds > 0 {
+		return rc.StallSeconds
+	}
+	return 20
+}
+
 // Endpoint is an auxiliary container in the test topology (see
 // TestCase.Endpoints). It's a host the subject reaches on the bench network —
 // not a generator or receiver.
@@ -593,10 +718,24 @@
 // DuckDB verifier container instead of a receiver.
 func (tc *TestCase) UsesVerifier() bool { return tc.Verifier != nil }
 
+// IsDirectorAgentRotationType reports whether the case is the director↔agent
+// TLS cert-rotation correctness flow, which has its own subject-driven (no
+// generator) driver — see runDirectorAgentCertRotation.
+func (tc *TestCase) IsDirectorAgentRotationType() bool {
+	return tc.Type == "director_agent_tls_cert_rotation_correctness"
+}
+
 // UsesAgent reports whether the case adds an external agent container to the
 // topology. The agent connects into the subject rather than being connected to.
 func (tc *TestCase) UsesAgent() bool { return tc.Agent != nil }
 
+// IsDirectorAgentRotationType reports whether the case is the director↔agent
+// TLS cert-rotation correctness flow, which has its own subject-driven (no
+// generator) driver — see runDirectorAgentCertRotation.
+func (tc *TestCase) IsDirectorAgentRotationType() bool {
+	return tc.Type == "director_agent_tls_cert_rotation_correctness"
+}
+
 // IsPerformanceType reports whether the case is scored as a throughput test —
 // the plain `performance` type or the Kafka variant `kafka_performance`.
 func (tc *TestCase) IsPerformanceType() bool {
@@ -726,6 +865,53 @@
 	if err := tc.validateVerifier(); err != nil {
 		return err
 	}
+	if err := tc.validateRotation(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// validateRotation checks the optional `rotation:` block and the
+// director_agent_tls_cert_rotation_correctness type's structural requirements:
+// the block is required for (and only meaningful to) that type, the mode must be
+// known, and the case must be subject-driven (an endpoint the director deploys
+// onto, no generator) with a min_received floor for the verdict.
+func (tc *TestCase) validateRotation() error {
+	if !tc.IsDirectorAgentRotationType() {
+		if tc.Rotation != nil {
+			return fmt.Errorf("case %q: `rotation:` is only valid for type director_agent_tls_cert_rotation_correctness", tc.Name)
+		}
+		return nil
+	}
+	if tc.Rotation == nil {
+		return fmt.Errorf("case %q: type director_agent_tls_cert_rotation_correctness requires a `rotation:` block", tc.Name)
+	}
+	switch tc.Rotation.Mode {
+	case RotationSameCA, RotationNewCARecover, RotationNewCAReject:
+	default:
+		return fmt.Errorf("case %q: rotation.mode %q must be one of %s, %s, %s",
+			tc.Name, tc.Rotation.Mode, RotationSameCA, RotationNewCARecover, RotationNewCAReject)
+	}
+	// 0/unset defaults via SettleSecondsOrDefault/StallSecondsOrDefault; reject
+	// negatives so a typo like `settle_seconds: -5` can't be silently defaulted.
+	if tc.Rotation.SettleSeconds < 0 {
+		return fmt.Errorf("case %q: rotation.settle_seconds must be >= 0 (0/unset defaults to 25), got %d", tc.Name, tc.Rotation.SettleSeconds)
+	}
+	if tc.Rotation.StallSeconds < 0 {
+		return fmt.Errorf("case %q: rotation.stall_seconds must be >= 0 (0/unset defaults to 20), got %d", tc.Name, tc.Rotation.StallSeconds)
+	}
+	// The director drives data by collecting from an endpoint and forwarding —
+	// there is no generator, so the verdict rests on min_received plus the
+	// post-rotation count behaviour. Guard the two structural preconditions.
+	if tc.HasGenerator() {
+		return fmt.Errorf("case %q: director_agent_tls_cert_rotation_correctness is subject-driven and must not declare a generator", tc.Name)
+	}
+	if len(tc.Endpoints) == 0 {
+		return fmt.Errorf("case %q: director_agent_tls_cert_rotation_correctness requires an `endpoints:` block (the host the director deploys the agent onto)", tc.Name)
+	}
+	if tc.Correctness.MinReceived <= 0 {
+		return fmt.Errorf("case %q: director_agent_tls_cert_rotation_correctness requires correctness.min_received > 0", tc.Name)
+	}
 	return nil
 }
 

internal/orchestrator/tls.go

@@ -248,6 +248,84 @@ func RotateServerCertWrongCA(outDir string, serverHosts []string) error {
 	return writePEMKey(filepath.Join(abs, "server.key"), srvKey, 0o644)
 }
 
+// RotateServerCertNewCA rotates the entire trust root: it generates a BRAND-NEW
+// CA, overwrites ca.crt / ca.key in place, and re-signs server.crt / server.key
+// under the new CA (same SAN set). Unlike RotateServerCertWrongCA — which throws
+// the new CA away so nothing can ever trust the new leaf — this PERSISTS the new
+// CA, so a party that re-reads ca.crt (e.g. a director serving it at
+// /dl/cert.pem, or a bootstrap agent that re-fetches it) can recover trust and
+// reconnect. It models a full CA rollover with re-distribution. Used by the
+// director↔agent "new_ca_recover" rotation case.
+//
+// The client bundle (client.crt) is intentionally left untouched: the
+// director↔agent cases have no generator/client leaf, and rewriting it would
+// only matter to a client that pins the old CA, which this rollover deliberately
+// supersedes.
+func RotateServerCertNewCA(outDir string, serverHosts []string) error {
+	abs, err := filepath.Abs(outDir)
+	if err != nil {
+		return err
+	}
+
+	// Fresh CA — PERSISTED to ca.crt/ca.key (the difference from
+	// RotateServerCertWrongCA), so anything that re-reads the CA can recover.
+	caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return fmt.Errorf("generate new ca key: %w", err)
+	}
+	caTpl := &x509.Certificate{
+		SerialNumber:          bigSerial(),
+		Subject:               pkix.Name{CommonName: "PipeBench Rotated CA"},
+		NotBefore:             time.Now().Add(-1 * time.Hour),
+		NotAfter:              time.Now().Add(24 * time.Hour),
+		IsCA:                  true,
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+		BasicConstraintsValid: true,
+	}
+	caDER, err := x509.CreateCertificate(rand.Reader, caTpl, caTpl, &caKey.PublicKey, caKey)
+	if err != nil {
+		return fmt.Errorf("sign new ca: %w", err)
+	}
+	caCert, err := x509.ParseCertificate(caDER)
+	if err != nil {
+		return fmt.Errorf("parse new ca: %w", err)
+	}
+	if err := writePEMCert(filepath.Join(abs, "ca.crt"), caDER); err != nil {
+		return err
+	}
+	if err := writePEMKey(filepath.Join(abs, "ca.key"), caKey, 0o600); err != nil {
+		return err
+	}
+
+	if len(serverHosts) == 0 {
+		serverHosts = []string{"subject"}
+	}
+	serverDNS, serverIP := splitHosts(serverHosts)
+	srvKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		return fmt.Errorf("generate server key: %w", err)
+	}
+	srvTpl := &x509.Certificate{
+		SerialNumber: bigSerial(),
+		Subject:      pkix.Name{CommonName: serverHosts[0]},
+		NotBefore:    time.Now().Add(-1 * time.Hour),
+		NotAfter:     time.Now().Add(24 * time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		DNSNames:     serverDNS,
+		IPAddresses:  serverIP,
+	}
+	srvDER, err := x509.CreateCertificate(rand.Reader, srvTpl, caCert, &srvKey.PublicKey, caKey)
+	if err != nil {
+		return fmt.Errorf("re-sign server cert under new ca: %w", err)
+	}
+	if err := writePEMCert(filepath.Join(abs, "server.crt"), srvDER); err != nil {
+		return err
+	}
+	return writePEMKey(filepath.Join(abs, "server.key"), srvKey, 0o644)
+}
+
 // loadCA reads and parses the CA cert + key written by GenerateTLSCerts.
 func loadCA(dir string) (*x509.Certificate, *ecdsa.PrivateKey, error) {
 	certPEM, err := os.ReadFile(filepath.Join(dir, "ca.crt"))