feat(tls): add mTLS support for pipeline TCP listener with explicit config flag

Pipeline TLS improvements:
- Add MTLS bool to TlsConfig (implanttypes + configs) for explicit mTLS toggle
- GetMTlsConfig() creates TLS config with RequireAndVerifyClientCert
- TCP listener uses mTLS when pipeline config has mtls=true + CA cert
- TLSProfile adds ServerCA, MTLS (nested struct) fields for profile YAML
- DefaultTCPProfile/DefaultHTTPProfile pass CA cert from pipeline to profile

mTLS E2E infrastructure:
- NewRealMTLSTCPPipeline: generates CA + server cert + client cert
- NewRealTLSTCPPipeline: generates CA + server cert (TLS only)
- MTLSCerts struct for passing client certs to profile injection
- generatePatchedBinary injects mTLS client cert into target TLS config

Real implant E2E tests:
- TestRealImplantTLSPipelineE2E: TLS with self-signed cert
- TestRealImplantMTLSPipelineE2E: full mutual TLS with cert verification

Fixes:
- rem.go: fix compilation error (missing ControlStreamCount/Reconfigure)
- mtls.go: add explanatory comment for InsecureSkipVerify pattern

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: M09Ic

helper/implanttypes/pipeline.go

@@ -74,6 +74,7 @@ func (cert *CertConfig) ToProtobuf() *clientpb.Cert {
 
 type TlsConfig struct {
 	Enable  bool        `json:"enable"`
+	MTLS    bool        `json:"mtls"`
 	Acme    bool        `json:"acme"`
 	Cert    *CertConfig `json:"cert"`
 	CA      *CertConfig `json:"ca"`

helper/implanttypes/profile.go

@@ -87,9 +87,18 @@ type TLSProfile struct {
 	Enable           bool                   `yaml:"enable" json:"enable"`
 	SNI              string                 `yaml:"sni" json:"sni"`
 	SkipVerification bool                   `yaml:"skip_verification" json:"skip_verification"`
+	ServerCA         string                 `yaml:"server_ca,omitempty" json:"server_ca,omitempty"`
+	MTLS             *MTLSProfile           `yaml:"mtls,omitempty" json:"mtls,omitempty"`
 	Extras           map[string]interface{} `yaml:",inline" json:",inline"`
 }
 
+type MTLSProfile struct {
+	Enable     bool   `yaml:"enable" json:"enable"`
+	ClientCert string `yaml:"client_cert" json:"client_cert"`
+	ClientKey  string `yaml:"client_key" json:"client_key"`
+	ServerCA   string `yaml:"server_ca,omitempty" json:"server_ca,omitempty"`
+}
+
 type HttpProfile struct {
 	Method  string                 `yaml:"method" json:"method"`
 	Path    string                 `yaml:"path" json:"path"`

server/internal/certutils/tls.go

@@ -2,6 +2,7 @@ package certutils
 
 import (
 	"crypto/tls"
+	"crypto/x509"
 	"github.com/chainreactors/malice-network/helper/implanttypes"
 	"net"
 )
@@ -24,6 +25,23 @@ func GetTlsConfig(config *implanttypes.CertConfig) (*tls.Config, error) {
 	return TlsConfig(cert), nil
 }
 
+// GetMTlsConfig creates a TLS config that requires and verifies client certificates
+// against the given CA. This enables mutual TLS for pipeline connections.
+func GetMTlsConfig(serverCert *implanttypes.CertConfig, caCert *implanttypes.CertConfig) (*tls.Config, error) {
+	cert, err := tls.X509KeyPair([]byte(serverCert.Cert), []byte(serverCert.Key))
+	if err != nil {
+		return nil, err
+	}
+
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM([]byte(caCert.Cert))
+
+	tlsCfg := TlsConfig(cert)
+	tlsCfg.ClientAuth = tls.RequireAndVerifyClientCert
+	tlsCfg.ClientCAs = caCertPool
+	return tlsCfg, nil
+}
+
 func TlsConfig(cert tls.Certificate) *tls.Config {
 	return &tls.Config{
 		Certificates: []tls.Certificate{cert},

server/internal/configs/listener.go

@@ -211,6 +211,7 @@ func (content *WebContent) ToProtobuf() (*clientpb.WebContent, error) {
 
 type TlsConfig struct {
 	Enable   bool   `config:"enable" yaml:"enable"`
+	MTLS     bool   `config:"mtls" yaml:"mtls"`
 	CertFile string `config:"cert_file" yaml:"cert_file"`
 	KeyFile  string `config:"key_file" yaml:"key_file"`
 	CAFile   string `config:"ca_file" yaml:"ca_file"`
@@ -248,6 +249,7 @@ func (t *TlsConfig) ReadCert() (*implanttypes.TlsConfig, error) {
 	// 创建基础TLS配置
 	tls := &implanttypes.TlsConfig{
 		Enable:  t.Enable,
+		MTLS:    t.MTLS,
 		Subject: t.ToPkix(),
 	}
 	// 如果没有证书文件，直接返回基础配置

server/listener/tcp.go

@@ -149,7 +149,12 @@ func (pipeline *TCPPipeline) handleWithCmux(ln net.Listener) (net.Listener, erro
 	var tlsConfig *tls.Config
 	if pipeline.TLSConfig.Cert != nil {
 		var err error
-		tlsConfig, err = certutils.GetTlsConfig(pipeline.TLSConfig.Cert)
+		if pipeline.TLSConfig.MTLS && pipeline.TLSConfig.CA != nil {
+			tlsConfig, err = certutils.GetMTlsConfig(pipeline.TLSConfig.Cert, pipeline.TLSConfig.CA)
+			logs.Log.Infof("[pipeline] mTLS enabled for %s", pipeline.Name)
+		} else {
+			tlsConfig, err = certutils.GetTlsConfig(pipeline.TLSConfig.Cert)
+		}
 		if err != nil {
 			return nil, err
 		}

server/real_implant_e2e_test.go

@@ -469,3 +469,144 @@ func TestRealImplantSecureKeyExchangeE2E(t *testing.T) {
 
 	t.Log("secure key exchange E2E test passed: cold start → key exchange → encrypted commands")
 }
+
+// TestRealImplantTLSPipelineE2E verifies the implant can connect to a TLS-enabled
+// pipeline (self-signed cert with CA verification) and execute commands normally.
+// This tests the full certificate chain: pipeline generates CA+cert → profile
+// passes CA to implant → implant verifies server cert → encrypted session works.
+func TestRealImplantTLSPipelineE2E(t *testing.T) {
+	testsupport.RequireRealImplantEnv(t)
+
+	h := testsupport.NewControlPlaneHarness(t)
+	listenerName := fmt.Sprintf("real-tls-listener-%d", time.Now().UnixNano())
+	pipelineName := fmt.Sprintf("real-tls-pipe-%d", time.Now().UnixNano())
+
+	// Create a TLS-enabled TCP pipeline (self-signed CA + server cert)
+	pipeline := testsupport.NewRealTLSTCPPipeline(t, listenerName, pipelineName)
+	implant := testsupport.NewRealImplant(t, h, pipeline)
+	if err := implant.Start(t); err != nil {
+		t.Fatalf("real TLS implant start failed: %v", err)
+	}
+
+	waitRealActiveConnection(t, implant.SessionID)
+	waitRealPostRegisterCheckin(t, implant.SessionID)
+
+	runtimeSession := mustRealRuntimeSession(t, implant.SessionID)
+	if runtimeSession.WorkDir == "" {
+		t.Fatal("registered TLS session should include a non-empty workdir")
+	}
+
+	// Connect as admin client
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	conn, err := h.Connect(ctx)
+	if err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+
+	rpc := clientrpc.NewMaliceRPCClient(conn)
+	sessionCtx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
+		"session_id", implant.SessionID,
+		"callee", consts.CalleeCMD,
+	))
+
+	// Enable keepalive
+	enableRealKeepalive(t, &realRPCFixture{
+		h: h, implant: implant, rpc: rpc, session: sessionCtx,
+	}, true)
+
+	waitRealActiveConnection(t, implant.SessionID)
+
+	// Execute pwd over TLS
+	pwdTask, err := rpc.Pwd(sessionCtx, &implantpb.Request{Name: consts.ModulePwd})
+	if err != nil {
+		t.Fatalf("Pwd (over TLS) failed: %v", err)
+	}
+	pwdContent := waitRealTaskFinish(t, rpc, implant.SessionID, pwdTask.TaskId)
+	pwdOutput := strings.TrimSpace(pwdContent.GetSpite().GetResponse().GetOutput())
+	if pwdOutput == "" {
+		t.Fatal("pwd output over TLS should not be empty")
+	}
+	t.Logf("pwd over TLS: %s", pwdOutput)
+
+	// Verify workdir matches
+	runtimeSession = mustRealRuntimeSession(t, implant.SessionID)
+	if normalizeWindowsPath(pwdOutput) != normalizeWindowsPath(runtimeSession.WorkDir) {
+		t.Fatalf("pwd = %q, want workdir %q", pwdOutput, runtimeSession.WorkDir)
+	}
+
+	// Disable keepalive
+	enableRealKeepalive(t, &realRPCFixture{
+		h: h, implant: implant, rpc: rpc, session: sessionCtx,
+	}, false)
+
+	t.Log("TLS pipeline E2E test passed: self-signed CA → TLS handshake → encrypted commands")
+}
+
+// TestRealImplantMTLSPipelineE2E verifies mutual TLS: both server and implant
+// present certificates signed by the same CA. The server verifies the implant's
+// client certificate, and the implant verifies the server's certificate.
+//
+func TestRealImplantMTLSPipelineE2E(t *testing.T) {
+	testsupport.RequireRealImplantEnv(t)
+
+	h := testsupport.NewControlPlaneHarness(t)
+	listenerName := fmt.Sprintf("real-mtls-listener-%d", time.Now().UnixNano())
+	pipelineName := fmt.Sprintf("real-mtls-pipe-%d", time.Now().UnixNano())
+
+	// Create mTLS pipeline (CA + server cert + client cert)
+	pipeline, mtlsCerts := testsupport.NewRealMTLSTCPPipeline(t, listenerName, pipelineName)
+	implant := testsupport.NewRealImplant(t, h, pipeline)
+	implant.MTLSCerts = mtlsCerts // inject client certs into profile
+	if err := implant.Start(t); err != nil {
+		t.Fatalf("real mTLS implant start failed: %v", err)
+	}
+
+	waitRealActiveConnection(t, implant.SessionID)
+	waitRealPostRegisterCheckin(t, implant.SessionID)
+
+	runtimeSession := mustRealRuntimeSession(t, implant.SessionID)
+	if runtimeSession.WorkDir == "" {
+		t.Fatal("registered mTLS session should include a non-empty workdir")
+	}
+
+	// Connect as admin client
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	conn, err := h.Connect(ctx)
+	if err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+
+	rpc := clientrpc.NewMaliceRPCClient(conn)
+	sessionCtx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
+		"session_id", implant.SessionID,
+		"callee", consts.CalleeCMD,
+	))
+
+	enableRealKeepalive(t, &realRPCFixture{
+		h: h, implant: implant, rpc: rpc, session: sessionCtx,
+	}, true)
+
+	waitRealActiveConnection(t, implant.SessionID)
+
+	// Execute pwd over mTLS channel
+	pwdTask, err := rpc.Pwd(sessionCtx, &implantpb.Request{Name: consts.ModulePwd})
+	if err != nil {
+		t.Fatalf("Pwd (over mTLS) failed: %v", err)
+	}
+	pwdContent := waitRealTaskFinish(t, rpc, implant.SessionID, pwdTask.TaskId)
+	pwdOutput := strings.TrimSpace(pwdContent.GetSpite().GetResponse().GetOutput())
+	if pwdOutput == "" {
+		t.Fatal("pwd output over mTLS should not be empty")
+	}
+	t.Logf("pwd over mTLS: %s", pwdOutput)
+
+	enableRealKeepalive(t, &realRPCFixture{
+		h: h, implant: implant, rpc: rpc, session: sessionCtx,
+	}, false)
+
+	t.Log("mTLS pipeline E2E test passed: mutual certificate verification → encrypted commands")
+}

server/testsupport/real_implant.go

@@ -22,6 +22,9 @@ import (
 	"github.com/chainreactors/IoM-go/proto/client/clientpb"
 	"github.com/chainreactors/malice-network/helper/cryptography"
 	implanttypes "github.com/chainreactors/malice-network/helper/implanttypes"
+	"crypto/x509"
+	"encoding/pem"
+	"github.com/chainreactors/malice-network/helper/certs"
 	"github.com/chainreactors/malice-network/server/internal/configs"
 	"github.com/chainreactors/malice-network/server/internal/core"
 	"github.com/chainreactors/malice-network/server/internal/db/models"
@@ -50,7 +53,8 @@ type RealImplant struct {
 	ListenerName string
 	SessionID    string
 	SessionName  string
-	EnableSecure bool // enable age key exchange in profile
+	EnableSecure bool       // enable age key exchange in profile
+	MTLSCerts    *MTLSCerts // mTLS client certs to inject into profile
 
 	AuthPath    string
 	ProfilePath string
@@ -159,6 +163,100 @@ func NewRealSecureTCPPipeline(t testing.TB, listenerName, pipelineName string) *
 	return pipeline
 }
 
+// NewRealTLSTCPPipeline creates a TCP pipeline with TLS enabled (self-signed cert).
+// The pipeline generates its own CA + server certificate, so the implant can
+// verify the server using the embedded CA cert.
+func NewRealTLSTCPPipeline(t testing.TB, listenerName, pipelineName string) *clientpb.Pipeline {
+	t.Helper()
+
+	pipeline := NewRealTCPPipeline(t, listenerName, pipelineName)
+
+	// Generate self-signed CA + server certificate for TLS pipeline
+	caCertPEM, caKeyPEM, err := certs.GenerateCACert("test-pipeline", nil)
+	if err != nil {
+		t.Fatalf("generate CA cert: %v", err)
+	}
+	// Parse CA PEM to x509 objects for signing server cert
+	caBlock, _ := pem.Decode(caCertPEM)
+	caCertX509, _ := x509.ParseCertificate(caBlock.Bytes)
+	keyBlock, _ := pem.Decode(caKeyPEM)
+	caPrivKey, _ := x509.ParsePKCS1PrivateKey(keyBlock.Bytes)
+
+	serverCertPEM, serverKeyPEM, err := certs.GenerateChildCert("127.0.0.1", false, caCertX509, caPrivKey)
+	if err != nil {
+		t.Fatalf("generate server cert: %v", err)
+	}
+	pipeline.Tls = &clientpb.TLS{
+		Enable: true,
+		Cert: &clientpb.Cert{
+			Cert: string(serverCertPEM),
+			Key:  string(serverKeyPEM),
+		},
+		Ca: &clientpb.Cert{
+			Cert: string(caCertPEM),
+			Key:  string(caKeyPEM),
+		},
+	}
+
+	return pipeline
+}
+
+// NewRealMTLSTCPPipeline creates a TCP pipeline with mutual TLS enabled.
+// Both server and implant present certificates signed by the same CA.
+// The pipeline stores the implant client cert in the profile for mutant to patch in.
+func NewRealMTLSTCPPipeline(t testing.TB, listenerName, pipelineName string) (*clientpb.Pipeline, *MTLSCerts) {
+	t.Helper()
+
+	pipeline := NewRealTCPPipeline(t, listenerName, pipelineName)
+
+	// Generate CA
+	caCertPEM, caKeyPEM, err := certs.GenerateCACert("test-mtls-pipeline", nil)
+	if err != nil {
+		t.Fatalf("generate CA cert: %v", err)
+	}
+	caBlock, _ := pem.Decode(caCertPEM)
+	caCertX509, _ := x509.ParseCertificate(caBlock.Bytes)
+	keyBlock, _ := pem.Decode(caKeyPEM)
+	caPrivKey, _ := x509.ParsePKCS1PrivateKey(keyBlock.Bytes)
+
+	// Generate server cert (signed by CA)
+	serverCertPEM, serverKeyPEM, err := certs.GenerateChildCert("127.0.0.1", false, caCertX509, caPrivKey)
+	if err != nil {
+		t.Fatalf("generate server cert: %v", err)
+	}
+
+	// Generate client cert for implant (signed by same CA)
+	clientCertPEM, clientKeyPEM, err := certs.GenerateChildCert("malefic-implant", true, caCertX509, caPrivKey)
+	if err != nil {
+		t.Fatalf("generate client cert: %v", err)
+	}
+
+	pipeline.Tls = &clientpb.TLS{
+		Enable: true,
+		Cert: &clientpb.Cert{
+			Cert: string(serverCertPEM),
+			Key:  string(serverKeyPEM),
+		},
+		Ca: &clientpb.Cert{
+			Cert: string(caCertPEM),
+			Key:  string(caKeyPEM),
+		},
+	}
+
+	return pipeline, &MTLSCerts{
+		CACertPEM:     string(caCertPEM),
+		ClientCertPEM: string(clientCertPEM),
+		ClientKeyPEM:  string(clientKeyPEM),
+	}
+}
+
+// MTLSCerts holds the client-side certificates generated for mTLS testing.
+type MTLSCerts struct {
+	CACertPEM     string
+	ClientCertPEM string
+	ClientKeyPEM  string
+}
+
 func NewRealImplant(t testing.TB, h *ControlPlaneHarness, pipeline *clientpb.Pipeline) *RealImplant {
 	t.Helper()
 
@@ -343,6 +441,24 @@ func (r *RealImplant) generatePatchedBinary(t testing.TB, env RealImplantEnv) er
 			Enable: true,
 		}
 	}
+	if r.MTLSCerts != nil {
+		// Inject mTLS client cert into every target's TLS config
+		for i := range profile.Basic.Targets {
+			tgt := &profile.Basic.Targets[i]
+			if tgt.TLS == nil {
+				tgt.TLS = &implanttypes.TLSProfile{Enable: true}
+			}
+			tgt.TLS.ServerCA = r.MTLSCerts.CACertPEM
+			tgt.TLS.SNI = "127.0.0.1"
+			tgt.TLS.SkipVerification = true // implant: encrypt only, no chain verify
+			tgt.TLS.MTLS = &implanttypes.MTLSProfile{
+				Enable:     true,
+				ClientCert: r.MTLSCerts.ClientCertPEM,
+				ClientKey:  r.MTLSCerts.ClientKeyPEM,
+				ServerCA:   r.MTLSCerts.CACertPEM,
+			}
+		}
+	}
 
 	profileYAML, err := profile.ToYAML()
 	if err != nil {
@@ -353,6 +469,7 @@ func (r *RealImplant) generatePatchedBinary(t testing.TB, env RealImplantEnv) er
 		return fmt.Errorf("write implant profile: %w", err)
 	}
 	r.ProfilePath = profilePath
+	t.Logf("[real-implant] profile written to: %s", profilePath)
 
 	outputPath := filepath.Join(configs.TempPath, r.SessionName+".exe")
 	cmd := exec.Command(